intrusion-detection/app/web/__init__.py

#!/usr/bin/env python3
"""Flask web UI for the intrusion-detection system.

A localhost-only web server (``127.0.0.1:<high port>``) 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)


# ---------------------------------------------------------------------------
# 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,
    }
    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()