feat: write consolidated blocklists and mirror git-diffable router state
- Add `write_blocklist_files()` to emit final IPv4/IPv6 blocklists to `reports/app/data/` - Add `APP_REPORT_DIR` / `APP_REPORT_DATA_DIR` constants for consolidated outputs - Write firewall state snapshots to `data/mikrotik/` for git diff auditing - Capture router state before scanning; fall back to scan-only if router is unreachable - Use `split_cidr_versions` for CIDR-aware subnet splitting - Update README with output layout and git-diffable state notes
This commit is contained in:
1 parent
6005340ecb
commit
28c77691bb
4 files changed
+91
-26
No files matched your search
+44
-11
@@ -15,6 +15,7 @@ import shlex
|
||||
from typing import Iterable
|
||||
from .constants import (
|
||||
BLOCKLIST_NAME,
|
||||
MIKROTIK_DATA_DIR,
|
||||
REPORT_DIR,
|
||||
ROUTER_IP,
|
||||
ROUTER_PASSWORD,
|
||||
@@ -23,6 +24,7 @@ from .constants import (
|
||||
SSH_TIMEOUT,
|
||||
WHITELIST_NAMES,
|
||||
)
|
||||
from .files import write_lines
|
||||
from .ip_utils import is_whitelisted
|
||||
|
||||
|
||||
@@ -130,9 +132,20 @@ class MikrotikFirewall:
|
||||
f"{path} firewall filter print terse")
|
||||
return self._parse_terse_output(output)
|
||||
|
||||
@staticmethod
|
||||
def _write_json(path: str, obj) -> None:
|
||||
with open(path, "w", encoding="utf-8") as file:
|
||||
json.dump(obj, file, indent=2, sort_keys=True)
|
||||
|
||||
def save_state_report(self, report_dir: str = REPORT_DIR,
|
||||
changes: dict | None = None) -> str:
|
||||
"""Save current address lists and firewall rules for auditing."""
|
||||
"""Save current address lists and firewall rules for auditing.
|
||||
|
||||
Writes structured snapshots to ``<report_dir>/mikrotik/data/`` and to
|
||||
the common ``data/mikrotik/`` directory (overwritten every run so git
|
||||
can diff changes to the firewall state over time), plus a timestamped
|
||||
JSON report under ``<report_dir>/mikrotik/``.
|
||||
"""
|
||||
state = {
|
||||
"router": self.router_ip,
|
||||
"collected_at": datetime.now().astimezone().isoformat(),
|
||||
@@ -147,19 +160,22 @@ class MikrotikFirewall:
|
||||
}
|
||||
if changes is not None:
|
||||
state["changes"] = changes
|
||||
data_dir = os.path.join(report_dir, "mikrotik", "data")
|
||||
os.makedirs(data_dir, exist_ok=True)
|
||||
|
||||
report_data_dir = os.path.join(report_dir, "mikrotik", "data")
|
||||
os.makedirs(report_data_dir, exist_ok=True)
|
||||
os.makedirs(MIKROTIK_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)
|
||||
for data_dir in (report_data_dir, MIKROTIK_DATA_DIR):
|
||||
self._write_json(
|
||||
os.path.join(data_dir, f"address_list_{version}.json"),
|
||||
state["address_lists"][version])
|
||||
self._write_json(
|
||||
os.path.join(data_dir, f"firewall_{version}.json"),
|
||||
state["firewall_rules"][version])
|
||||
|
||||
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)
|
||||
self._write_json(report_path, state)
|
||||
return report_path
|
||||
|
||||
def sync_blocklist(self, blocked_ipv4: Iterable[str],
|
||||
@@ -213,6 +229,23 @@ class MikrotikFirewall:
|
||||
f"/ipv6 firewall address-list add list={name} address={ip}")
|
||||
|
||||
|
||||
def write_blocklist_files(blocked_ipv4: Iterable[str],
|
||||
blocked_ipv6: Iterable[str],
|
||||
report_dir: str = REPORT_DIR) -> tuple[str, str]:
|
||||
"""Write the final consolidated blocklist to ``<report_dir>/app/data/``.
|
||||
|
||||
Returns the paths of the IPv4 and IPv6 blocklist files
|
||||
(``blocklist_ipv4.txt`` / ``blocklist_ipv6.txt``).
|
||||
"""
|
||||
data_dir = os.path.join(report_dir, "app", "data")
|
||||
os.makedirs(data_dir, exist_ok=True)
|
||||
v4_path = os.path.join(data_dir, "blocklist_ipv4.txt")
|
||||
v6_path = os.path.join(data_dir, "blocklist_ipv6.txt")
|
||||
write_lines(v4_path, sorted(blocked_ipv4))
|
||||
write_lines(v6_path, sorted(blocked_ipv6))
|
||||
return v4_path, v6_path
|
||||
|
||||
|
||||
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")
|
||||
|
||||
Reference in new issue
Block a user