#!/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, jsonify, request from ..utils.constants import ( APP_REPORT_DATA_DIR, APP_REPORT_DIR, COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR, DEFAULT_ERROR_RATIO, DEFAULT_MIN_DENIED, DEFAULT_MIN_ERRORS, DEFAULT_TOP, MIKROTIK_REPORT_DIR, NPM_LOG_PATH, NPM_REPORT_DIR, REPORT_DIR, WEB_HOST, WEB_PORT, ) from ..firewall import MikrotikFirewall 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) def _analysis_helpers(): """SQLite-backed log-analysis query helpers (None when DB unavailable). Functions take the shared DB connection each call so routes stay stateless across the background scan thread. """ try: from ..db import get_connection from ..db import repositories as repo except Exception: return None def top_ips(since: str | None = None, limit: int = 50) -> list: return repo.top_ips(get_connection(), since=since, limit=limit) def ip_timeline(ip: str) -> list: return repo.ip_timeline(get_connection(), ip=ip) def cross_source(limit: int = 100) -> list: return repo.correlate_across_sources(get_connection(), limit=limit) def conn(): return get_connection() def request_count(conn) -> int: return repo.count_requests(conn) def scan_count(conn) -> int: return repo.count_scans(conn) return {"top_ips": top_ips, "ip_timeline": ip_timeline, "cross_source": cross_source, "conn": conn, "request_count": request_count, "scan_count": scan_count} # --------------------------------------------------------------------------- # app factory # --------------------------------------------------------------------------- def create_app() -> Flask: """Build and configure the Flask application.""" _app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) app = Flask(__name__, template_folder=os.path.join(_app_dir, "templates"), static_folder=os.path.join(_app_dir, "static")) # Seed the in-memory job store from disk so the dashboard keeps showing # recent scans across server restarts. web_scan.load_persisted() from .routes import register_routes helpers = { "_count_lines": _count_lines, "_read_lines": _read_lines, "_read_json": _read_json, "_params_from_request": _params_from_request, "_with_router": _with_router, "_sync_firewall": _sync_firewall, "_analysis": _analysis_helpers(), } register_routes(app, REPORT_SOURCES, SCAN_KINDS, web_scan, helpers) 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()