diff --git a/app/db/backfill.py b/app/db/backfill.py new file mode 100644 index 0000000..61bdad7 --- /dev/null +++ b/app/db/backfill.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""R001 — one-time backfill of legacy runtime data into the SQLite store. + +Imports existing JSON/CSV/git-history data so the DB becomes the authoritative +source without losing anything already accumulated on a deployed instance: + +- cumulative blocklist from the git-history reconstruction (old source of + truth) -> ``blocklist_entries`` (grow-only ledger) +- firewall snapshots from ``data/mikrotik/*.json`` -> ``firewall_snapshots`` +- per-source requests from ``reports//data/requests.csv`` (best effort) + +Idempotent: skips rows already present. Safe to run repeatedly. + +Usage: + python -m app.db.backfill [--dry-run] +""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import sys +from datetime import datetime, timezone + +from ..utils.constants import ( + APP_REPORT_DATA_DIR, + DATA_DIR, + MIKROTIK_DATA_DIR, + NPM_REPORT_DIR, + COPYPARTY_REPORT_DIR, +) + + +def utcnow() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _read_cidr_lines(path: str) -> set[str]: + try: + with open(path, "r", encoding="utf-8") as f: + return {line.strip() for line in f if line.strip()} + except OSError: + return set() + + +def backfill_blocklist(conn, dry_run: bool = False) -> dict: + """Import the legacy cumulative blocklist into the ledger.""" + from . import repositories as repo + from ..utils.blocklist_history import ( + committed_lines, current_file_lines, + ) + + v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt") + v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt") + # Legacy source of truth: union of all git revisions + working file + + # (there is no earlier scan target to add here). + complete_v4 = committed_lines(v4_path) | current_file_lines(v4_path) + complete_v6 = committed_lines(v6_path) | current_file_lines(v6_path) + cidrs = sorted(complete_v4 | complete_v6) + if not cidrs: + return {"blocklist": {"cidrs": 0, "added": 0}} + if dry_run: + return {"blocklist": {"cidrs": len(cidrs), "added": None, "dry_run": True}} + result = repo.record_blocklist(conn, cidrs, added_by="backfill") + return {"blocklist": result} + + +def backfill_firewall(conn, dry_run: bool = False) -> dict: + """Import the latest firewall snapshot from ``data/mikrotik/*.json``.""" + from . import repositories as repo + + def _load(name: str) -> list: + try: + with open(os.path.join(MIKROTIK_DATA_DIR, name), "r", + encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, list) else [] + except (OSError, ValueError): + return [] + + addr_v4 = _load("address_list_ipv4.json") + addr_v6 = _load("address_list_ipv6.json") + fw_v4 = _load("firewall_ipv4.json") + fw_v6 = _load("firewall_ipv6.json") + if not (addr_v4 or addr_v6 or fw_v4 or fw_v6): + return {"firewall": {"dry_run": dry_run, "snapshot": None}} + # Idempotent: if a firewall snapshot already exists (previous backfill), + # don't create another copy of the same latest state. + n = conn.execute( + "SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0] + if n: + return {"firewall": {"dry_run": dry_run, "snapshot": "exists", + "skipped": True}} + if dry_run: + return {"firewall": {"dry_run": True, "snapshot": None}} + sid = repo.save_firewall_snapshot( + conn, address_lists_v4=addr_v4, address_lists_v6=addr_v6, + rules_v4=fw_v4, rules_v6=fw_v6) + return {"firewall": {"snapshot": sid}} + + +def backfill_requests(conn, dry_run: bool = False) -> dict: + """Best-effort import of per-source ``requests.csv`` files. + + Only ingests rows that don't already exist (keyed on source+log_file+ts+ip). + Legacy CSV columns vary by source; we keep a conservative mapping and never + fail the whole backfill over a single bad file. + """ + from . import repositories as repo + + seen = set() + for r in conn.execute( + "SELECT source, log_file, ts, ip FROM requests"): + seen.add((r["source"], r["log_file"], r["ts"], r["ip"])) + + candidates = [ + (os.path.join(COPYPARTY_REPORT_DIR, "data", "requests.csv"), "copyparty"), + (os.path.join(NPM_REPORT_DIR, "data", "requests.csv"), "npm"), + ] + total = 0 + for path, source in candidates: + if not os.path.exists(path): + continue + try: + with open(path, "r", encoding="utf-8", errors="replace") as f: + reader = csv.DictReader(f) + rows = [] + for row in reader: + ip = (row.get("ip") or "").strip() + ts = row.get("ts") or row.get("timestamp") or "" + log_file = row.get("log_file") or row.get("file") or "" + if not ip: + continue + key = (source, log_file, ts, ip) + if key in seen: + continue + seen.add(key) + rows.append({ + "source": source, "scan_id": None, "log_file": log_file, + "ts": ts, "ip": ip, + "version": 4 if ":" not in ip else 6, + "method": row.get("method"), "path": row.get("path"), + "status": _int_or_none(row.get("status")), + "upstream_status": _int_or_none( + row.get("upstream_status")), + "host": row.get("host"), "user": row.get("user"), + "ua": row.get("ua") or row.get("user_agent"), + "referer": row.get("referer"), + "denied": _bool_str(row.get("denied")), + "banned": _bool_str(row.get("banned")), + "is_error": _bool_str(row.get("is_error")), + "extra": None, + }) + total += 1 + if not dry_run and total: + repo.insert_requests(conn, rows) + except Exception as exc: + print(f" WARN: requests.csv backfill {source}: {exc}") + return {"requests": {"imported": total, "dry_run": dry_run}} + + +def _int_or_none(v): + try: + if v in (None, "", "-"): + return None + return int(v) + except (TypeError, ValueError): + return None + + +def _bool_str(v): + if isinstance(v, str): + return int(v.strip().lower() in ("1", "true", "yes", "t")) + return int(bool(v)) + + +def main(argv: list[str] | None = None) -> None: + p = argparse.ArgumentParser(prog="python -m app.db.backfill", + description="Backfill legacy runtime data into SQLite.") + p.add_argument("--dry-run", action="store_true", + help="report what would be imported without writing") + opts = p.parse_args(argv) + + from . import get_connection + conn = get_connection() + print(f"== backfill (dry_run={opts.dry_run}) ==") + print("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run)) + print("firewall :", backfill_firewall(conn, dry_run=opts.dry_run)) + print("requests :", backfill_requests(conn, dry_run=opts.dry_run)) + print("== done ==") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/templates/firewall.html b/app/templates/firewall.html index 6a66afd..18c7b23 100644 --- a/app/templates/firewall.html +++ b/app/templates/firewall.html @@ -23,6 +23,11 @@ endblock %} {% block content %}

Firewall rules

+ +
+

Snapshot history (SQLite)

+
+
{% endblock %} {% block scripts %} {% endblock %} diff --git a/app/web/routes.py b/app/web/routes.py index 14f06b8..e12df06 100644 --- a/app/web/routes.py +++ b/app/web/routes.py @@ -265,6 +265,19 @@ def register_routes(app: Flask, "snapshot_dir": MIKROTIK_DATA_DIR, }) + @app.get("/api/firewall/snapshots") + def firewall_snapshots(): + """Recent firewall snapshots persisted in the SQLite store (F001).""" + try: + from ..db import get_connection + from ..db import repositories as repo + conn = get_connection() + limit = request.args.get("limit", 10, type=int) + return jsonify({"snapshots": repo.recent_firewall_snapshots( + conn, limit=limit)}) + except Exception as exc: + return jsonify({"error": str(exc), "snapshots": []}), 503 + @app.post("/api/firewall/sync") def firewall_sync(): return _sync_firewall(dry_run=True) diff --git a/app/web/web_scan.py b/app/web/web_scan.py index eb63ffb..b989b47 100644 --- a/app/web/web_scan.py +++ b/app/web/web_scan.py @@ -258,23 +258,71 @@ def latest_file_scan() -> dict[str, Any] | None: } +def _db_scans(limit: int = 10) -> list[dict[str, Any]]: + """Recent scan records from the SQLite ``scans`` table (S001). + + Converts DB rows into the web job-dict shape so the dashboard can surface + scans that were persisted by ``run_scan`` / analyzers even if they were + never run through the in-memory job store (e.g. CLI scans or restarts). + Returns [] when the DB is unavailable. + """ + try: + from ..db import get_connection + from ..db import repositories as repo + conn = get_connection() + rows = repo.latest_scans(conn, limit=limit) + except Exception: + return [] + out = [] + for r in rows: + try: + result_json = r.get("result_json") + result = json.loads(result_json) if result_json else {} + except (TypeError, ValueError): + result = {} + out.append({ + "job_id": f"db-scan-{r['id']}", + "kind": r.get("kind") or "consolidated", + "state": r.get("status") or "done", + "started": r.get("started_at"), + "finished": r.get("finished_at"), + "error": r.get("error"), + "busy": False, + "details": {}, + "result": result or None, + }) + return out + + def latest_scans(limit: int = 10) -> list[dict[str, Any]]: """Return the most recently started jobs, newest first. - Appends a file-derived "latest scan" entry when the newest report files on - disk reflect a scan newer than any known job (e.g. after a restart or a - CLI run), so the dashboard always shows the latest completed scan. + Merges three sources, newest-first: + 1. the in-memory job store (currently running / last-run jobs), + 2. the SQLite ``scans`` table (persisted by run_scan/analyzers — S001), + 3. a file-derived "latest scan" fallback for pre-DB runs. + Any DB-backed scan not already represented in memory is appended so the + dashboard shows scans from CLI runs and across restarts. """ load_persisted() with _jobs_lock: jobs = sorted(_jobs.values(), key=lambda j: j.started or "", reverse=True) entries = [j.to_dict() for j in jobs] + + # Add DB scan records that aren't already in the in-memory store. Key on + # (kind, started_at) to avoid duplicating a job that run_scan also logged. + known = {(e.get("kind"), e.get("started")) for e in entries} + for db_row in _db_scans(limit=limit * 2): + if (db_row.get("kind"), db_row.get("started")) not in known: + entries.append(db_row) + file_entry = latest_file_scan() if file_entry is not None: - newest = jobs[0] if jobs else None - newest_ts = _iso_key( - (newest.finished or newest.started) if newest else None) + newest = (entries or [None])[0] + newest_key = ( + (newest.get("finished") or newest.get("started")) if newest else None) + newest_ts = _iso_key(newest_key) # Only surface the file-derived entry when it is genuinely newer than # every known job (tolerance avoids duplicating the run that produced # it, where collected_at lands a moment after the job finished).