#!/usr/bin/env python3 """Flask web UI for the intrusion-detection system. A localhost-only web server (``127.0.0.1:``) for running scans, browsing reports, and managing the MikroTik blocklist / firewall. Run directly: ``python -m app.web`` Or via the entry point: ``python app.py --web`` (see ``serve()``). """ from __future__ import annotations import json import os from flask import Flask, abort, jsonify, render_template, request from .constants import ( APP_REPORT_DATA_DIR, APP_REPORT_DIR, BLOCKLIST_NAME, COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR, DEFAULT_ERROR_RATIO, DEFAULT_MIN_DENIED, DEFAULT_MIN_ERRORS, DEFAULT_TOP, MIKROTIK_DATA_DIR, MIKROTIK_REPORT_DIR, NPM_LOG_PATH, NPM_REPORT_DIR, REPORT_DIR, WEB_HOST, WEB_PORT, WHITELIST_NAMES, ) from .firewall import MikrotikFirewall from .ip_utils import ip_version from .scan import ScanParams from . import web_scan #: report sources exposed by the API -> report directory REPORT_SOURCES = { "copyparty": COPYPARTY_REPORT_DIR, "npm": NPM_REPORT_DIR, "app": APP_REPORT_DIR, "mikrotik": MIKROTIK_REPORT_DIR, } SCAN_KINDS = ("copyparty", "npm", "consolidated") # --------------------------------------------------------------------------- # helpers # --------------------------------------------------------------------------- def _count_lines(path: str) -> int | None: """Return the line count of a file, or None if it cannot be read.""" try: with open(path, "r", encoding="utf-8") as f: return sum(1 for _ in f) except OSError: return None def _read_lines(path: str) -> list[str]: """Return non-empty lines of a file, or [] if it cannot be read.""" try: with open(path, "r", encoding="utf-8") as f: return [line.strip() for line in f if line.strip()] except OSError: return [] def _read_json(path: str, default: list | dict | None = None) -> list | dict: """Read a JSON file, returning ``default`` (or []) on failure.""" try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except (OSError, ValueError): return default if default is not None else [] def _params_from_request() -> ScanParams: """Build ScanParams from the JSON request body, falling back to defaults.""" data = request.get_json(silent=True) or {} def _pick(key: str, default): value = data.get(key) return default if value is None or value == "" else value return ScanParams( copyparty_log_path=str( _pick("copyparty_log_path", COPYPARTY_LOG_PATH)), copyparty_report_dir=str( _pick("copyparty_report_dir", COPYPARTY_REPORT_DIR)), min_denied=int(_pick("min_denied", DEFAULT_MIN_DENIED)), npm_log_path=str(_pick("npm_log_path", NPM_LOG_PATH)), npm_report_dir=str(_pick("npm_report_dir", NPM_REPORT_DIR)), min_errors=int(_pick("min_errors", DEFAULT_MIN_ERRORS)), error_ratio=float(_pick("error_ratio", DEFAULT_ERROR_RATIO)), report_dir=str(_pick("report_dir", REPORT_DIR)), top=int(_pick("top", DEFAULT_TOP)), quiet=bool(data.get("quiet", False)), full=bool(data.get("full", False)), update_firewall=bool(data.get("update_firewall", False)), dry_run=bool(data.get("dry_run", False)), ) def _with_router(fn): """Connect to the router, run ``fn(fw)``, always disconnect. Returns a JSON error response (502) if the router is unreachable. """ fw = MikrotikFirewall() try: fw.connect() return fn(fw) except Exception as e: return jsonify({"error": str(e)}), 502 finally: fw.disconnect() def _sync_firewall(dry_run: bool): """Sync the latest consolidated blocklist to the router (preview or apply).""" v4 = _read_lines(os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")) v6 = _read_lines(os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")) def _sync(fw): changes = fw.sync_blocklist(v4, v6, dry_run=dry_run) return jsonify({**changes, "dry_run": dry_run}) return _with_router(_sync) # --------------------------------------------------------------------------- # app factory # --------------------------------------------------------------------------- def create_app() -> Flask: """Build and configure the Flask application.""" app = Flask(__name__) @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) # --- 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()) # A busy-lock rejection is a bare {"error": ...}; a started job has # "error" == None, so test for a truthy value, not key presence. 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) # --- 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(): """Preview the blocklist sync (dry-run, no changes applied).""" return _sync_firewall(dry_run=True) @app.post("/api/firewall/apply") def firewall_apply(): """Apply the blocklist sync (add new / remove stale entries).""" 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) return app def serve(host: str = WEB_HOST, port: int = WEB_PORT) -> None: """Run the web server on localhost at the configured high port.""" app = create_app() print(f"Intrusion Detection web UI running at http://{host}:{port}") app.run(host=host, port=port, use_reloader=False) if __name__ == "__main__": serve()