zwitschi
0e2c69579a
- Deleted the existing MikroTik block list script (mikrotik-block.rsc) which contained whitelisted and blocked IP addresses. - Updated scan.py to include options for synchronizing the MikroTik blocked list after scanning. - Added functionality to preview firewall changes without applying them. - Integrated the MikrotikFirewall class to manage the synchronization of blocked IPs with the MikroTik device.
320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""
|
|
firewall.py -
|
|
Module for interacting with the Mikrotik routeros shell to manage firewall rules and IP address lists.
|
|
Provides functions to read the current firewall configuration, add and remove IPs from the blocklist.
|
|
Reads the latest report from the IDS scan, reads the current firewall configuration, and updates the router's blocklist accordingly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import paramiko
|
|
from datetime import datetime
|
|
import shlex
|
|
from typing import Iterable
|
|
from utils import is_whitelisted, parse_ip, split_versions
|
|
|
|
DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "mikrotik")
|
|
REPORT_DIR = os.path.join(os.path.dirname(__file__), "reports")
|
|
MIKROTIK_DATA_DIR = os.path.join(REPORT_DIR, "mikrotik", "data")
|
|
|
|
SSH_TIMEOUT = 10 # seconds
|
|
SSH_PORT = 22
|
|
ROUTER_IP = os.environ.get("ROUTER_IP", "192.168.88.1")
|
|
ROUTER_USER = os.environ.get("ROUTER_USER", "admin")
|
|
ROUTER_PASSWORD = os.environ.get("ROUTER_PASSWORD", "M11ffpgm")
|
|
BLOCKLIST_NAME = os.environ.get("BLOCKLIST_NAME", "Blocked")
|
|
WHITELIST_NAMES = os.environ.get(
|
|
"WHITELIST_NAMES", "Cloudflare,Openfactory").split(",")
|
|
|
|
|
|
class MikrotikFirewall:
|
|
"""
|
|
Class to manage Mikrotik firewall rules and IP address lists.
|
|
Provides methods to read the current blocklist, add/remove IPs, and update the blocklist based on IDS scan reports.
|
|
"""
|
|
|
|
def __init__(self, router_ip: str = ROUTER_IP, username: str = ROUTER_USER,
|
|
password: str = ROUTER_PASSWORD):
|
|
self.router_ip = router_ip
|
|
self.username = username
|
|
self.password = password
|
|
self.client: paramiko.SSHClient | None = None
|
|
|
|
def connect(self) -> None:
|
|
"""Establish an SSH connection to the Mikrotik router."""
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
try:
|
|
# RouterOS 7.20.x exposes an RSA host key and classic SSH ciphers.
|
|
client.connect(
|
|
self.router_ip,
|
|
port=SSH_PORT,
|
|
username=self.username,
|
|
password=self.password,
|
|
timeout=SSH_TIMEOUT,
|
|
banner_timeout=SSH_TIMEOUT,
|
|
auth_timeout=SSH_TIMEOUT,
|
|
allow_agent=False,
|
|
look_for_keys=False,
|
|
compress=False,
|
|
disabled_algorithms={"pubkeys": ["rsa-sha2-512"]},
|
|
)
|
|
except Exception:
|
|
client.close()
|
|
raise
|
|
self.client = client
|
|
|
|
def disconnect(self) -> None:
|
|
"""Close the SSH connection."""
|
|
if self.client:
|
|
self.client.close()
|
|
self.client = None
|
|
|
|
def execute_command(self, command: str) -> str:
|
|
"""Execute a command on the Mikrotik router via SSH and return the output."""
|
|
if not self.client:
|
|
raise RuntimeError("SSH client is not connected.")
|
|
stdin, stdout, stderr = self.client.exec_command(command)
|
|
output = stdout.read().decode("utf-8")
|
|
error = stderr.read().decode("utf-8")
|
|
if error:
|
|
raise RuntimeError(f"Error executing command '{command}': {error}")
|
|
return output
|
|
|
|
def get_address_list_ipv4(self, name: str) -> set[str]:
|
|
"""Retrieve the current IPv4 address list from the Mikrotik router."""
|
|
output = self.execute_command("/ip firewall address-list print")
|
|
ips = set()
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 3 and parts[1] == "list" and parts[2] == name:
|
|
ip = parts[0]
|
|
ips.add(ip)
|
|
return ips
|
|
|
|
def get_address_list_ipv6(self, name: str) -> set[str]:
|
|
"""Retrieve the current IPv6 address list from the Mikrotik router."""
|
|
output = self.execute_command("/ipv6 firewall address-list print")
|
|
ips = set()
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) >= 3 and parts[1] == "list" and parts[2] == name:
|
|
ip = parts[0]
|
|
ips.add(ip)
|
|
return ips
|
|
|
|
@staticmethod
|
|
def _parse_terse_output(output: str) -> list[dict[str, str]]:
|
|
"""Parse RouterOS `print terse` records into key/value dictionaries."""
|
|
records = []
|
|
for line in output.splitlines():
|
|
fields = {}
|
|
for field in shlex.split(line):
|
|
key, separator, value = field.partition("=")
|
|
if separator:
|
|
fields[key] = value
|
|
if fields:
|
|
records.append(fields)
|
|
return records
|
|
|
|
def get_address_lists(self, ipv6: bool = False) -> list[dict[str, str]]:
|
|
"""Retrieve all address-list entries in structured form."""
|
|
path = "/ipv6" if ipv6 else "/ip"
|
|
output = self.execute_command(
|
|
f"{path} firewall address-list print terse")
|
|
return self._parse_terse_output(output)
|
|
|
|
def get_firewall_rules(self, ipv6: bool = False) -> list[dict[str, str]]:
|
|
"""Retrieve all firewall filter rules in structured form."""
|
|
path = "/ipv6" if ipv6 else "/ip"
|
|
output = self.execute_command(
|
|
f"{path} firewall filter print terse")
|
|
return self._parse_terse_output(output)
|
|
|
|
def save_state_report(self, report_dir: str = REPORT_DIR,
|
|
changes: dict | None = None) -> str:
|
|
"""Save current address lists and firewall rules for auditing."""
|
|
state = {
|
|
"router": self.router_ip,
|
|
"collected_at": datetime.now().astimezone().isoformat(),
|
|
"address_lists": {
|
|
"ipv4": self.get_address_lists(),
|
|
"ipv6": self.get_address_lists(ipv6=True),
|
|
},
|
|
"firewall_rules": {
|
|
"ipv4": self.get_firewall_rules(),
|
|
"ipv6": self.get_firewall_rules(ipv6=True),
|
|
},
|
|
}
|
|
if changes is not None:
|
|
state["changes"] = changes
|
|
data_dir = os.path.join(report_dir, "mikrotik", "data")
|
|
os.makedirs(data_dir, exist_ok=True)
|
|
for version in ("ipv4", "ipv6"):
|
|
with open(os.path.join(data_dir, f"address_list_{version}.json"), "w", encoding="utf-8") as file:
|
|
json.dump(state["address_lists"][version],
|
|
file, indent=2, sort_keys=True)
|
|
with open(os.path.join(data_dir, f"firewall_{version}.json"), "w", encoding="utf-8") as file:
|
|
json.dump(state["firewall_rules"][version],
|
|
file, indent=2, sort_keys=True)
|
|
timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
|
|
report_path = os.path.join(report_dir, "mikrotik", f"{timestamp}.json")
|
|
with open(report_path, "w", encoding="utf-8") as file:
|
|
json.dump(state, file, indent=2, sort_keys=True)
|
|
return report_path
|
|
|
|
def sync_blocklist(self, blocked_ipv4: Iterable[str],
|
|
blocked_ipv6: Iterable[str],
|
|
dry_run: bool = False,
|
|
report_dir: str = REPORT_DIR) -> dict[str, list[str]]:
|
|
"""Synchronize the Blocked list with scan targets and audit the result."""
|
|
desired_ipv4 = set(blocked_ipv4)
|
|
desired_ipv6 = set(blocked_ipv6)
|
|
current_ipv4 = self.get_address_list_ipv4(BLOCKLIST_NAME)
|
|
current_ipv6 = self.get_address_list_ipv6(BLOCKLIST_NAME)
|
|
whitelisted_ipv4 = set()
|
|
whitelisted_ipv6 = set()
|
|
for name in WHITELIST_NAMES:
|
|
whitelisted_ipv4.update(self.get_address_list_ipv4(name))
|
|
whitelisted_ipv6.update(self.get_address_list_ipv6(name))
|
|
|
|
changes = {
|
|
"added_ipv4": sorted(desired_ipv4 - current_ipv4 - whitelisted_ipv4),
|
|
"added_ipv6": sorted(desired_ipv6 - current_ipv6 - whitelisted_ipv6),
|
|
"removed_ipv4": sorted(current_ipv4 - desired_ipv4),
|
|
"removed_ipv6": sorted(current_ipv6 - desired_ipv6),
|
|
"dry_run": dry_run,
|
|
}
|
|
if not dry_run:
|
|
for ip in changes["removed_ipv4"]:
|
|
self.execute_command(
|
|
f"/ip firewall address-list remove [find list={BLOCKLIST_NAME} address={ip}]")
|
|
for ip in changes["removed_ipv6"]:
|
|
self.execute_command(
|
|
f"/ipv6 firewall address-list remove [find list={BLOCKLIST_NAME} address={ip}]")
|
|
self.write_address_list_ipv4(BLOCKLIST_NAME, changes["added_ipv4"])
|
|
self.write_address_list_ipv6(BLOCKLIST_NAME, changes["added_ipv6"])
|
|
self.save_state_report(report_dir, changes)
|
|
return changes
|
|
|
|
def write_address_list_ipv4(self, name: str, ips: Iterable[str]) -> None:
|
|
"""Write a list of IPv4 addresses to the Mikrotik router's address list."""
|
|
for ip in ips:
|
|
if is_whitelisted(ip):
|
|
continue # Skip whitelisted IPs
|
|
self.execute_command(
|
|
f"/ip firewall address-list add list={name} address={ip}")
|
|
|
|
def write_address_list_ipv6(self, name: str, ips: Iterable[str]) -> None:
|
|
"""Write a list of IPv6 addresses to the Mikrotik router's address list."""
|
|
for ip in ips:
|
|
if is_whitelisted(ip):
|
|
continue # Skip whitelisted IPs
|
|
self.execute_command(
|
|
f"/ipv6 firewall address-list add list={name} address={ip}")
|
|
|
|
|
|
def get_scan_report(report_dir: str = REPORT_DIR) -> dict[str, list[str]]:
|
|
"""Read the latest scan report `mikrotik-block.rsc` from the report directory and return the list of blocked IPs."""
|
|
report_path = os.path.join(report_dir, "mikrotik-block.rsc")
|
|
if not os.path.exists(report_path):
|
|
raise FileNotFoundError(f"Scan report not found: {report_path}")
|
|
|
|
# add sections, they start with `/ip firewall address-list` or `/ipv6 firewall address-list` and end with a blank line or EOF
|
|
sections = {
|
|
"/ip firewall address-list": [],
|
|
"/ipv6 firewall address-list": [],
|
|
}
|
|
# add list=Cloudflare or list=Openfactory to whitelist sections, list=Blocked to blocked section
|
|
with open(report_path, "r", encoding="utf-8") as f:
|
|
current_section = None
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
current_section = None
|
|
continue
|
|
if line.startswith("/ip firewall address-list"):
|
|
current_section = "/ip firewall address-list"
|
|
continue
|
|
if line.startswith("/ipv6 firewall address-list"):
|
|
current_section = "/ipv6 firewall address-list"
|
|
continue
|
|
if current_section:
|
|
sections[current_section].append(line)
|
|
|
|
return sections
|
|
|
|
|
|
def main():
|
|
"""Legacy entry point for applying the latest generated RouterOS report."""
|
|
# Read the latest scan report
|
|
scan_report = get_scan_report(REPORT_DIR)
|
|
|
|
# Connect to the Mikrotik router
|
|
firewall = MikrotikFirewall()
|
|
firewall.connect()
|
|
|
|
# read current blocked IPs from the router
|
|
current_blocked_ipv4 = firewall.get_address_list_ipv4(BLOCKLIST_NAME)
|
|
current_blocked_ipv6 = firewall.get_address_list_ipv6(BLOCKLIST_NAME)
|
|
|
|
# read whitelisted IPs from the router
|
|
whitelisted_ipv4 = set()
|
|
whitelisted_ipv6 = set()
|
|
for wl_name in WHITELIST_NAMES:
|
|
whitelisted_ipv4.update(firewall.get_address_list_ipv4(wl_name))
|
|
whitelisted_ipv6.update(firewall.get_address_list_ipv6(wl_name))
|
|
|
|
# read blocked IPs from the scan report
|
|
blocked_ipv4 = set()
|
|
blocked_ipv6 = set()
|
|
for line in scan_report["/ip firewall address-list"]:
|
|
if "list=Blocked" in line:
|
|
parts = line.split()
|
|
for part in parts:
|
|
if part.startswith("address="):
|
|
ip = part.split("=")[1]
|
|
blocked_ipv4.add(ip)
|
|
for line in scan_report["/ipv6 firewall address-list"]:
|
|
if "list=Blocked" in line:
|
|
parts = line.split()
|
|
for part in parts:
|
|
if part.startswith("address="):
|
|
ip = part.split("=")[1]
|
|
blocked_ipv6.add(ip)
|
|
|
|
# Determine which IPs to add and remove
|
|
to_add_ipv4 = blocked_ipv4 - current_blocked_ipv4 - whitelisted_ipv4
|
|
to_add_ipv6 = blocked_ipv6 - current_blocked_ipv6 - whitelisted_ipv6
|
|
to_remove_ipv4 = current_blocked_ipv4 - blocked_ipv4
|
|
to_remove_ipv6 = current_blocked_ipv6 - blocked_ipv6
|
|
|
|
# Remove old blocked IPs
|
|
for ip in to_remove_ipv4:
|
|
firewall.execute_command(
|
|
f"/ip firewall address-list remove [find list={BLOCKLIST_NAME} address={ip}]")
|
|
for ip in to_remove_ipv6:
|
|
firewall.execute_command(
|
|
f"/ipv6 firewall address-list remove [find list={BLOCKLIST_NAME} address={ip}]")
|
|
|
|
# Add new blocked IPs
|
|
firewall.write_address_list_ipv4(BLOCKLIST_NAME, to_add_ipv4)
|
|
firewall.write_address_list_ipv6(BLOCKLIST_NAME, to_add_ipv6)
|
|
|
|
firewall.save_state_report(REPORT_DIR)
|
|
|
|
# Disconnect from the router
|
|
firewall.disconnect()
|
|
|
|
# Print summary
|
|
print(
|
|
f"Removed {len(to_remove_ipv4)} IPv4 and {len(to_remove_ipv6)} IPv6 addresses from the blocklist.")
|
|
print(
|
|
f"Added {len(to_add_ipv4)} IPv4 and {len(to_add_ipv6)} IPv6 addresses to the blocklist.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|