#!/usr/bin/env python3 """Route definitions for the intrusion-detection web UI.""" from __future__ import annotations import os from flask import Flask, abort, jsonify, render_template, request from ..utils.constants import ( APP_REPORT_DATA_DIR, BLOCKLIST_NAME, MIKROTIK_DATA_DIR, WEB_HOST, WEB_PORT, WHITELIST_NAMES, ) from ..utils.ip_utils import ip_version def register_routes(app: Flask, report_sources: dict, scan_kinds: tuple, web_scan, helpers) -> None: """Register all web UI routes on the Flask application.""" _count_lines = helpers["_count_lines"] _read_lines = helpers["_read_lines"] _read_json = helpers["_read_json"] _params_from_request = helpers["_params_from_request"] _with_router = helpers["_with_router"] _sync_firewall = helpers["_sync_firewall"] # --- pages ------------------------------------------------------------- @app.get("/") def index(): return render_template("dashboard.html") @app.get("/scan") def scan_page(): return render_template("scan.html") @app.get("/reports") def reports_page(): return render_template("reports.html") @app.get("/blocklist") def blocklist_page(): return render_template("blocklist.html") @app.get("/firewall") def firewall_page(): return render_template("firewall.html") @app.get("/whitelist") def whitelist_page(): return render_template("whitelist.html", whitelist_names=WHITELIST_NAMES) @app.get("/graph") def graph_page(): return render_template("graph.html") # --- status ------------------------------------------------------------ @app.get("/api/status") def status(): v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt") v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt") return jsonify({ "host": WEB_HOST, "port": WEB_PORT, "scan_running": web_scan.is_scan_running(), "latest_scans": web_scan.latest_scans(5), "blocklist_ipv4_count": _count_lines(v4_path), "blocklist_ipv6_count": _count_lines(v6_path), "firewall_snapshot": { "exists": os.path.isdir(MIKROTIK_DATA_DIR), "path": MIKROTIK_DATA_DIR, }, }) # --- scans ------------------------------------------------------------- @app.post("/api/scan/") def start_scan(kind): if kind not in scan_kinds: abort(404, f"unknown scan kind: {kind}") job = web_scan.start_scan(kind, _params_from_request()) if job.get("error"): return jsonify(job), 409 return jsonify(job), 202 @app.get("/api/scan/status") def scan_status(): return jsonify({ "running": web_scan.is_scan_running(), "latest_scans": web_scan.latest_scans(10), }) @app.get("/api/scan/") def scan_job(job_id): job = web_scan.get_scan(job_id) if job is None: abort(404, f"unknown job: {job_id}") return jsonify(job) # --- reports ----------------------------------------------------------- @app.get("/api/reports/") def list_reports(source): base = report_sources.get(source) if base is None: abort(404, f"unknown report source: {source}") names = sorted( n for n in os.listdir(base) if os.path.isfile(os.path.join(base, n))) return jsonify({"source": source, "files": names}) @app.get("/api/reports//") def get_report(source, name): base = report_sources.get(source) if base is None: abort(404, f"unknown report source: {source}") full = os.path.realpath(os.path.join(base, name)) if not full.startswith(os.path.realpath(base)): abort(403, "invalid report path") if not os.path.isfile(full): abort(404, "report not found") with open(full, "r", encoding="utf-8", errors="replace") as f: content = f.read() return jsonify({"source": source, "name": name, "content": content}) # --- blocklist --------------------------------------------------------- @app.get("/api/blocklist") def blocklist(): return jsonify({ "ipv4": _read_lines( os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")), "ipv6": _read_lines( os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")), }) @app.post("/api/blocklist/add") def blocklist_add(): data = request.get_json(silent=True) or {} ip = (data.get("ip") or "").strip() version = ip_version(ip) if not ip or version not in (4, 6): return jsonify({"error": "missing or invalid 'ip' field"}), 400 def _add(fw): if version == 4: fw.write_address_list_ipv4(BLOCKLIST_NAME, [ip]) else: fw.write_address_list_ipv6(BLOCKLIST_NAME, [ip]) return jsonify({"ok": True, "action": "add", "ip": ip}) return _with_router(_add) @app.post("/api/blocklist/remove") def blocklist_remove(): data = request.get_json(silent=True) or {} ip = (data.get("ip") or "").strip() version = ip_version(ip) if not ip or version not in (4, 6): return jsonify({"error": "missing or invalid 'ip' field"}), 400 def _remove(fw): path = "/ip" if version == 4 else "/ipv6" fw.execute_command( f"{path} firewall address-list remove " f"[find list={BLOCKLIST_NAME} address={ip}]") return jsonify({"ok": True, "action": "remove", "ip": ip}) return _with_router(_remove) # --- graph ------------------------------------------------------------- @app.get("/api/graph/ipv4") def graph_ipv4(): blocked = _read_lines( os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")) addr_v4 = _read_json( os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv4.json")) whitelisted = sorted({e.get("address") for e in addr_v4 if e.get("list") in WHITELIST_NAMES and e.get("address")}) return jsonify({ "blocked": blocked, "whitelisted": whitelisted, "counts": { "blocked": len(blocked), "whitelisted": len(whitelisted), }, "blocklist_name": BLOCKLIST_NAME, "whitelist_names": WHITELIST_NAMES, }) # --- firewall ---------------------------------------------------------- @app.get("/api/firewall/state") def firewall_state(): return jsonify({ "address_lists": { "ipv4": _read_json( os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv4.json")), "ipv6": _read_json( os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv6.json")), }, "firewall_rules": { "ipv4": _read_json( os.path.join(MIKROTIK_DATA_DIR, "firewall_ipv4.json")), "ipv6": _read_json( os.path.join(MIKROTIK_DATA_DIR, "firewall_ipv6.json")), }, "snapshot_dir": MIKROTIK_DATA_DIR, }) @app.post("/api/firewall/sync") def firewall_sync(): return _sync_firewall(dry_run=True) @app.post("/api/firewall/apply") def firewall_apply(): return _sync_firewall(dry_run=False) # --- whitelist --------------------------------------------------------- @app.get("/api/whitelist") def whitelist(): addr_v4 = _read_json( os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv4.json")) addr_v6 = _read_json( os.path.join(MIKROTIK_DATA_DIR, "address_list_ipv6.json")) result: dict[str, dict[str, list[str]]] = {} for name in WHITELIST_NAMES: result[name] = { "ipv4": sorted(e.get("address") for e in addr_v4 if e.get("list") == name), "ipv6": sorted(e.get("address") for e in addr_v6 if e.get("list") == name), } return jsonify({"names": WHITELIST_NAMES, "whitelists": result}) @app.post("/api/whitelist/update") def whitelist_update(): data = request.get_json(silent=True) or {} name = (data.get("name") or "").strip() address = (data.get("address") or "").strip() action = (data.get("action") or "add").strip().lower() version = ip_version(address) if name not in WHITELIST_NAMES: return jsonify({"error": f"unknown whitelist: {name}"}), 400 if not address or version not in (4, 6): return jsonify({"error": "missing or invalid 'address' field"}), 400 if action not in ("add", "remove"): return jsonify({"error": "action must be 'add' or 'remove'"}), 400 def _update(fw): path = "/ip" if version == 4 else "/ipv6" if action == "add": if version == 4: fw.write_address_list_ipv4(name, [address]) else: fw.write_address_list_ipv6(name, [address]) else: fw.execute_command( f"{path} firewall address-list remove " f"[find list={name} address={address}]") return jsonify({"ok": True, "action": action, "name": name, "address": address}) return _with_router(_update)