112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""MikroTik RouterOS script generation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from datetime import datetime
|
|
from typing import Iterable
|
|
|
|
from .ip_utils import is_ipv4, is_ipv6, is_whitelisted, parse_ip
|
|
from .whitelists import WHITELISTS
|
|
from .constants import BLOCKLIST_NAME
|
|
|
|
|
|
def split_cidr_versions(targets: Iterable[str]) -> tuple[list[str], list[str]]:
|
|
"""Split target CIDRs/IPs into (sorted IPv4 list, sorted IPv6 list)."""
|
|
v4 = []
|
|
v6 = []
|
|
for t in targets:
|
|
ip = t.split("/")[0]
|
|
if is_ipv4(ip):
|
|
v4.append(t)
|
|
elif is_ipv6(ip):
|
|
v6.append(t)
|
|
|
|
def key_cidr(cidr: str):
|
|
ip = cidr.split("/")[0]
|
|
addr = parse_ip(ip)
|
|
if addr is None:
|
|
return (9, b"", cidr)
|
|
return (addr.version, addr.packed, cidr)
|
|
|
|
return sorted(v4, key=key_cidr), sorted(v6, key=key_cidr)
|
|
|
|
|
|
def mikrotik_script(tag: str, blocked_ips_or_subnets: Iterable[str],
|
|
whitelists: dict[str, list[str]] | None = None,
|
|
generated: str | None = None) -> tuple[str, int, int]:
|
|
"""Build a RouterOS .rsc script.
|
|
|
|
Whitelisted ranges become `accept` rules placed above the `drop` rules.
|
|
Blocked targets can be individual IPs (e.g. 1.2.3.4/32) or CIDR blocks
|
|
(e.g. 1.2.3.0/24).
|
|
|
|
Returns (script_text, blocked_count, whitelisted_cidr_count).
|
|
"""
|
|
whitelists = WHITELISTS if whitelists is None else whitelists
|
|
generated = generated or datetime.now().isoformat(timespec="seconds")
|
|
|
|
def clean_blocked(targets: Iterable[str]) -> list[str]:
|
|
cleaned = []
|
|
for t in targets:
|
|
ip = t.split("/")[0]
|
|
if not is_whitelisted(ip):
|
|
cleaned.append(t)
|
|
return cleaned
|
|
|
|
blocked = clean_blocked(set(blocked_ips_or_subnets))
|
|
block_v4, block_v6 = split_cidr_versions(blocked)
|
|
|
|
lines: list[str] = [f"# {tag} firewall export - generated {generated}", ""]
|
|
wl_count = 0
|
|
|
|
for name, cidrs in whitelists.items():
|
|
nets = [ipaddress.ip_network(c) for c in cidrs]
|
|
wl_count += len(nets)
|
|
for version, cmd in ((4, "/ip firewall address-list"),
|
|
(6, "/ipv6 firewall address-list")):
|
|
chunk = [n for n in nets if n.version == version]
|
|
if not chunk:
|
|
continue
|
|
lines.append(f"# {name} IPv{version} ranges (whitelisted)")
|
|
lines.append(cmd)
|
|
lines += [f"add list={name} address={n}" for n in chunk]
|
|
lines.append("")
|
|
|
|
if block_v4:
|
|
lines.append(f"# Blocked IPv4 ({len(block_v4)}, non-whitelisted)")
|
|
lines.append("/ip firewall address-list")
|
|
lines += [f"add list={BLOCKLIST_NAME} address={t}" for t in block_v4]
|
|
lines.append("")
|
|
if block_v6:
|
|
lines.append(f"# Blocked IPv6 ({len(block_v6)}, non-whitelisted)")
|
|
lines.append("/ipv6 firewall address-list")
|
|
lines += [f"add list={BLOCKLIST_NAME} address={t}" for t in block_v6]
|
|
lines.append("")
|
|
|
|
for version, cmd, blocks in ((4, "/ip firewall filter", block_v4),
|
|
(6, "/ipv6 firewall filter", block_v6)):
|
|
lines.append(cmd)
|
|
for name in whitelists:
|
|
lines.append(
|
|
f"add chain=input action=accept src-address-list={name} "
|
|
f'comment="{tag} {name.lower()} whitelist (v{version})"')
|
|
if blocks:
|
|
for chain in ("input", "forward"):
|
|
lines.append(
|
|
f"add chain={chain} action=drop src-address-list={BLOCKLIST_NAME} "
|
|
f'comment="{tag} blocked (v{version})"')
|
|
lines.append("")
|
|
|
|
return "\n".join(lines), len(blocked), wl_count
|
|
|
|
|
|
def write_mikrotik(path: str, tag: str,
|
|
blocked_ips_or_subnets: Iterable[str]) -> tuple[str, int, int]:
|
|
text, n_blocked, n_whitelisted = mikrotik_script(
|
|
tag, blocked_ips_or_subnets)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
return path, n_blocked, n_whitelisted
|