96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BlockResult:
|
|
success: bool
|
|
message: str
|
|
|
|
|
|
class RouterOSClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
username: str,
|
|
password: str,
|
|
verify_tls: bool,
|
|
address_list: str,
|
|
timeout: int = 5,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.username = username
|
|
self.password = password
|
|
self.verify_tls = verify_tls
|
|
self.address_list = address_list
|
|
self.timeout = timeout
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(
|
|
self.base_url
|
|
and self.username
|
|
and self.password
|
|
and self.password != "CHANGE_ME"
|
|
)
|
|
|
|
def block_ip(self, address: str, timeout_value: str, comment: str) -> BlockResult:
|
|
if not self.configured:
|
|
return BlockResult(False, "RouterOS credentials are not configured")
|
|
try:
|
|
existing = self._request(
|
|
"GET",
|
|
"/rest/ip/firewall/address-list",
|
|
query={"list": self.address_list, "address": address},
|
|
)
|
|
if isinstance(existing, list) and existing:
|
|
return BlockResult(True, "address already present in RouterOS address-list")
|
|
|
|
body = {
|
|
"list": self.address_list,
|
|
"address": address,
|
|
"timeout": timeout_value,
|
|
"comment": comment[:220],
|
|
}
|
|
self._request("PUT", "/rest/ip/firewall/address-list", body=body)
|
|
return BlockResult(True, "address added to RouterOS address-list")
|
|
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, ValueError) as exc:
|
|
return BlockResult(False, f"RouterOS REST error: {exc}")
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
body: dict | None = None,
|
|
query: dict | None = None,
|
|
):
|
|
url = self.base_url + path
|
|
if query:
|
|
url += "?" + urllib.parse.urlencode(query)
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
token = base64.b64encode(f"{self.username}:{self.password}".encode("utf-8")).decode("ascii")
|
|
headers["Authorization"] = f"Basic {token}"
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
|
|
context = None
|
|
if url.lower().startswith("https://") and not self.verify_tls:
|
|
context = ssl._create_unverified_context()
|
|
|
|
with urllib.request.urlopen(request, timeout=self.timeout, context=context) as response:
|
|
raw = response.read()
|
|
if not raw:
|
|
return None
|
|
return json.loads(raw.decode("utf-8"))
|