feat(reports): store and browse report artifacts

B001 change caused 'unsupported operand type(s) for -: list and set' during a
full consolidated scan: _db_complete_blocklist returned split_cidr_versions()
output, which is tuple[list, list], but write_blocklist_report()/blocklist_delta()
expect set operands. Return set(v4), set(v6) to match cumulative_blocklist's
contract. Reproduced the exact report path with a populated ledger and
verified it no longer crashes.

README.md

@@ -172,6 +172,11 @@ python app.py scan --quiet
 python app.py scan --update-firewall --dry-run
 python app.py scan --update-firewall
 
+# detailed log-analysis report from the SQLite store (needs scans run first)
+python app.py analyze                  # top IPs + cross-source + blocklist
+python app.py analyze --ip 1.2.3.4     # drill into one IP's request history
+python app.py analyze --json           # machine-readable output
+
 # individual scanners (still runnable directly)
 python -m app.copyparty
 python -m app.npm

REPORTS.md

@@ -0,0 +1,142 @@
+# Report Inventory
+
+This inventory records the current report surface before moving report reads
+or writes into SQLite. Generated paths are rooted at `REPORT_DIR` (normally
+`reports/`). The SQLite tables listed below already exist in schema v1 unless
+noted otherwise.
+
+## Artifacts
+
+| Area | Producer and formats | Current role and consumers | Classification |
+| --- | --- | --- | --- |
+| `copyparty/` | `Analyzer.write_report()` creates timestamped `.txt`; `write_data()` creates per-IP CSV/JSON, denied-path CSV, flagged/banned IP TXT, `requests.csv`, `authed_users.json`, and `mikrotik-block.rsc`. | Reports API serves top-level text reports. `requests.csv` is imported by `app.db.backfill`; per-run data supports inspection and legacy workflows. Retention prunes generated files. | Request rows and per-scan IP stats are canonical in `requests` and `ip_stats`; rendered summaries and exports are derived. |
+| `npm/` | Same analyzer outputs, plus user-agent and host summary CSV files. | Same consumers as copyparty. `requests.csv` is a legacy backfill input; retention prunes generated files. | Request rows and per-scan IP stats are canonical in `requests` and `ip_stats`; summaries and exports are derived. |
+| `app/data/` | Consolidated scan writes `blocklist_ipv4.txt` and `blocklist_ipv6.txt`. | Blocklist UI/API reads these files; scan history reconstruction and firewall sync use them as current/compatibility inputs. Retention explicitly protects them. | Operational compatibility exports. Canonical blocked targets and events belong in `blocklist_entries` and `blocklist_events`; keep files until every consumer uses the DB and git-history contract is replaced. |
+| `app/` timestamped JSON | `write_blocklist_report()` emits scan/firewall change reports when the router is reachable. | Human-readable audit trail containing run delta, cumulative targets, router changes, dry-run state, and errors; retained under report retention. | Derived audit artifact. Preserve structured fields and source scan association in the DB before considering file retirement. |
+| `mikrotik/` | `MikrotikFirewall.save_state_report()` writes timestamped JSON snapshots and report data; latest state is also written to `data/mikrotik/*.json`. | Firewall UI/API displays current state and snapshot history. `app.db.backfill` currently imports only the latest `data/mikrotik` snapshot; runtime snapshot persistence uses `firewall_snapshots`, `firewall_address_list_entries`, and `firewall_rule_entries`. | Firewall rows and snapshots are canonical in the DB once persisted. Keep JSON exports during migration and retain any git-diff inputs required by deployment/audit. |
+
+## Consumers
+
+- `GET /api/reports/<source>` lists top-level files in the selected report
+  directory; `GET /api/reports/<source>/<path>` reads a file below it.
+- The reports page consumes those endpoints. Scan status and dashboard helpers
+  select the newest source and MikroTik report files for display.
+- `/api/blocklist` reads the consolidated IPv4/IPv6 text files. Scan history
+  reconstruction and firewall sync also consume those files alongside the
+  SQLite ledger and git history.
+- RouterOS `.rsc` exports and firewall synchronization consume generated
+  targets; these are operational outputs, not disposable presentation files.
+- `app.db.backfill` imports per-source `requests.csv`, blocklist history, and
+  the latest common MikroTik JSON snapshot. It does not currently import all
+  timestamped reports or all historical MikroTik snapshots.
+- Report retention removes old generated artifacts after consolidated scans,
+  subject to configured age/file-count limits; the consolidated blocklist
+  files are protected.
+
+## Restructuring Implications
+
+- Keep `requests`, `ip_stats`, blocklist tables, and firewall snapshot tables
+  as the queryable source of truth for facts they already represent. Avoid
+  copying those facts into a generic report blob.
+- A future report-artifact record should store metadata and structured fields
+  that have no canonical home yet. Rendered text/CSV/JSON should be generated
+  from those records or remain an explicit export.
+- Preserve source paths, generated timestamps, scan IDs, and hashes during
+  import so DB records can be traced back to original files.
+- Do not retire blocklist exports, RouterOS scripts, or source logs until
+  their consumers and audit requirements have explicit replacements.
+
+## Database and API Design
+
+### Canonical Data
+
+Continue treating normalized domain tables as canonical for facts they
+already represent:
+
+- `requests` for parsed request events.
+- `ip_stats` for per-IP/per-scan aggregates.
+- `scans` and `scans_sources` for run status, parameters, and source counts.
+- `blocklist_entries`, `blocklist_reasons`, and `blocklist_events` for block
+  decisions and history.
+- `firewall_snapshots`, `firewall_address_list_entries`, and
+  `firewall_rule_entries` for router state.
+
+Add a `report_artifacts` table only for report-specific summaries that cannot
+be reconstructed from those canonical rows. Proposed columns:
+
+| Column | Purpose |
+| --- | --- |
+| `id` | Integer primary key. |
+| `source` | `copyparty`, `npm`, `app`, or `mikrotik`. Keep as text because `log_sources` currently covers only request sources. |
+| `scan_id` | Nullable link to the generating scan; legacy imports may not have a known scan. |
+| `report_type` | Stable type such as `analyzer_summary`, `blocklist_change`, or `firewall_snapshot_summary`. |
+| `generated_at` | Original generation timestamp when known; otherwise import time plus an explicit unknown/original-time flag in metadata. |
+| `schema_version` | Version of the structured payload shape. |
+| `payload_json` | Structured report-specific summary, not a duplicate of requests or raw logs. |
+| `original_path` | Relative source path for provenance and legacy lookup. |
+| `content_sha256` | Hash of imported source content for integrity and idempotent import. |
+| `created_at` | Database import/creation timestamp. |
+
+Use a unique index on `(original_path, content_sha256)` for idempotent file
+imports. Keep payload schemas type-specific and versioned. Do not copy
+`requests.csv`, `ips.json`, firewall rows, or blocklists wholesale into
+`payload_json`; query their canonical tables and generate exports from those
+tables. If a rendered text report is retained as a database artifact, store
+its report-specific structured inputs and render text on demand rather than
+using opaque text as the only source of truth.
+
+### API Transition Contract
+
+- Add `GET /api/reports` with optional `source`, `type`, `since`, `until`,
+  `limit`, and opaque `cursor` filters. Return a stable list of artifact
+  metadata: `id`, `source`, `scan_id`, `type`, `generated_at`, and available
+  export formats. Enforce a bounded default and maximum page size.
+- Add `GET /api/report-artifacts/<artifact_id>` for the structured payload and
+  provenance metadata. Use a distinct path because `/api/reports/<source>`
+  already owns the one-segment route shape.
+- Add `GET /api/report-artifacts/<artifact_id>/export` for supported report types and formats
+  (initially JSON and, where useful, CSV/TXT). Generate exports from canonical
+  tables and versioned artifact payloads; do not accept filesystem paths from
+  a database-backed export request.
+- Preserve the current source/path routes as a read-only compatibility
+  fallback while files remain. The reports page should consume DB metadata
+  first, then offer legacy files when no imported artifact exists. Return
+  stable JSON errors for unknown IDs, unsupported formats, and unavailable
+  DB state.
+- Keep ordering deterministic (`generated_at DESC`, then `id DESC`), validate
+  filter values, and use cursor pagination for large histories. Do not expose
+  absolute server paths in list responses.
+
+### Sensitive and Large Content
+
+Do not include raw request lines or full request histories in report payloads;
+those belong in the dedicated log-storage design and require separate access
+and retention controls. Keep high-volume CSV exports generated on demand. Treat
+IP addresses, paths/query strings, user names, user agents, and firewall state
+as sensitive operational data: the existing web server must remain
+localhost-only, payloads should include only fields needed by the report, and
+new endpoints must not expose raw log data by default. Backups and exported
+files inherit the same access and retention requirements as the database.
+
+## Migration Implementation Status
+
+- Schema v2 adds `report_artifacts` with source, scan link, report type,
+  generation time, payload version, original relative path, and SHA-256 hash.
+- `python -m app.db.backfill --dry-run --reports-only` inventories supported
+  report files by source and format. The importer is idempotent, preserves
+  malformed JSON as text with a parse error, reports unreadable/oversized
+  files, and never modifies source files. It skips exports owned by canonical
+  tables (for example `requests.csv` and current blocklists).
+- New analyzer text summaries and consolidated blocklist change reports are
+  written to SQLite and their existing files. Firewall snapshots continue to
+  be persisted in the normalized firewall tables.
+- The Reports page lists database artifacts first, supports cursor pagination
+  and JSON/CSV/TXT downloads, and retains the existing file-backed routes as a
+  fallback for unmigrated files.
+- Validation covers idempotency, malformed/oversized inputs, source/format
+  counts, backup/restore, dual-write payloads, and API fallback. A workspace
+  dry-run found 76 files: 60 eligible, 16 canonical exports skipped, and none
+  unmappable at the time of the check.
+- No report files have been retired. Keep this gate closed until every listed
+  consumer has moved or has an explicit compatibility requirement, and a
+  production backup/restore and parity review has been approved.

app.py

@@ -40,12 +40,22 @@ def _run_scan(argv: list[str]) -> None:
     scan_main(argv)
 
 
+def _run_analyze(argv: list[str]) -> None:
+    from app.analyze import main as analyze_main
+    analyze_main(argv)
+
+
 def main() -> None:
     # `python app.py scan ...` runs the consolidated CLI scan.
     if len(sys.argv) > 1 and sys.argv[1] == "scan":
         _run_scan(sys.argv[2:])
         return
 
+    # `python app.py analyze ...` prints a detailed report from the SQLite store.
+    if len(sys.argv) > 1 and sys.argv[1] == "analyze":
+        _run_analyze(sys.argv[2:])
+        return
+
     # Default: start the web UI (parse_known_args tolerates stray flags).
     args, _ = _web_parser().parse_known_args()
     _run_web(args.web_host, args.web_port)

app/analyze.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""CLI detailed log-analysis report (L002).
+
+``python app.py analyze [--ip IP] [--since ISO] [--top N] [--json]`` reads the
+persisted SQLite store (request history, per-IP aggregates, blocklist ledger)
+and prints a detailed, interactive analysis. It complements the web UI's
+``/analysis`` page for use on a headless server or in scripts.
+
+No scan is run here: this only queries what ``scan`` has already persisted.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import datetime
+
+
+def _fmt_ts(epoch: float | None, default: str = "-") -> str:
+    if not epoch:
+        return default
+    try:
+        return datetime.fromtimestamp(epoch).isoformat(timespec="seconds")
+    except (ValueError, OverflowError, OSError):
+        return default
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="python app.py analyze",
+        description="Detailed log-analysis report from the SQLite store.")
+    p.add_argument("--ip", metavar="IP", default=None,
+                   help="focus the report on a single IP (drill-down)")
+    p.add_argument("--since", metavar="ISO", default=None,
+                   help="only analyse requests at/after this ISO timestamp")
+    p.add_argument("--top", type=int, default=20,
+                   help="rows in the top-IP / threat tables (default: 20)")
+    p.add_argument("--json", action="store_true",
+                   help="emit raw JSON instead of a human-readable report")
+    return p
+
+
+def _report(conn, opts) -> dict:
+    from .db import repositories as repo
+
+    data: dict = {}
+    data["scans"] = repo.latest_scans(conn, limit=10)
+    data["totals"] = {
+        "requests": repo.count_requests(conn),
+        "scans": repo.count_scans(conn),
+    }
+
+    if opts.ip:
+        data["ip"] = {
+            "ip": opts.ip,
+            "timeline": repo.ip_timeline(conn, ip=opts.ip),
+            "active_blocklist": [e for e in repo.active_blocklist(conn)
+                                 if e["cidr"].startswith(opts.ip)],
+        }
+    else:
+        data["top_ips"] = repo.top_ips(conn, since=opts.since, limit=opts.top)
+        data["cross_source"] = repo.correlate_across_sources(
+            conn, limit=opts.top)
+        data["blocklist"] = repo.active_blocklist(conn)
+        data["recent_errors"] = repo.query_requests(
+            conn, since=opts.since, limit=opts.top)
+    return data
+
+
+def _render_text(data: dict, opts) -> str:
+    lines = ["Detailed log analysis", "=" * 40, ""]
+    totals = data["totals"]
+    lines.append(f"SQLite store: {totals['requests']} requests across "
+                 f"{totals['scans']} scan(s)")
+    if data["scans"]:
+        lines.append("Latest scans:")
+        for s in data["scans"]:
+            lines.append(f"  #{s['id']:<4} {s['kind']:<8} {s['status']:<7} "
+                         f"{s.get('started') or s.get('started_at') or '-'}")
+    lines.append("")
+
+    if "ip" in data:
+        ip = data["ip"]
+        lines.append(f"== Request history for {ip['ip']} ==")
+        bl = ip["active_blocklist"]
+        if bl:
+            lines.append(f"  (source: active blocklist entries "
+                         f"{', '.join(e['cidr'] for e in bl)})")
+        lines.append(f"{'time':<24} {'src':<9} {'method':<8} {'status':<6} path")
+        for r in ip["timeline"][:opts.top]:
+            lines.append(
+                f"{r.get('ts') or '-':<24} {str(r.get('source')):<9} "
+                f"{str(r.get('method') or '-'):<8} "
+                f"{str(r.get('status') or '-'):<6} {r.get('path') or '-'}")
+    else:
+        lines.append("== Top IPs ==")
+        lines.append(
+            f"{'ip':<18} {'ver':<3} {'reqs':<6} {'err':<5} {'mal':<5} "
+            f"{'first':<22} last")
+        for r in data["top_ips"]:
+            lines.append(
+                f"{str(r['ip']):<18} {r.get('version') or '-':<3} "
+                f"{r['total']:<6} {r['errors']:<5} {r['malicious_hits']:<5} "
+                f"{_fmt_ts(r['first_epoch']):<22} {_fmt_ts(r['last_epoch'])}")
+        lines.append("")
+
+        lines.append("== Cross-source IPs ==")
+        if data["cross_source"]:
+            for r in data["cross_source"]:
+                lines.append(
+                    f"  {r['ip']:<18} sources={r['sources']} "
+                    f"({r['source_list']}) total={r['total']}")
+        else:
+            lines.append("  (none — no IP seen in more than one log source)")
+
+        lines.append("")
+        lines.append("== Active blocklist (ledger) ==")
+        if data["blocklist"]:
+            for e in data["blocklist"]:
+                lines.append(f"  {e['cidr']:<22} added={e['added_at']}")
+        else:
+            lines.append("  (empty)")
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> None:
+    opts = _build_parser().parse_args(argv)
+    try:
+        from .db import get_connection
+        conn = get_connection()
+        data = _report(conn, opts)
+    except Exception as exc:
+        print(f"error: could not read SQLite store ({exc}); "
+              f"run a scan first so data exists.", file=sys.stderr)
+        sys.exit(1)
+
+    if opts.json:
+        print(json.dumps(data, indent=2, sort_keys=True, default=str))
+    else:
+        print(_render_text(data, opts))
+
+
+if __name__ == "__main__":
+    main()

app/analyzer.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import os
 import re
 from collections import Counter
@@ -336,6 +337,7 @@ class Analyzer:
 
         self.write_data()
         report_path = self.write_report()
+        self.persist_report_artifact(report_path, db_scan_id)
         print(f"data files:     {self.data_dir}")
         print(f"summary report: {report_path}")
 
@@ -349,3 +351,28 @@ class Analyzer:
             result.update(rsc=rsc, blocked=n_blocked,
                           whitelisted=n_whitelisted)
         return result
+
+    def persist_report_artifact(self, path: str,
+                                scan_id: int | None = None) -> None:
+        """Store the rendered summary in SQLite while keeping its file export."""
+        try:
+            from .db import get_connection
+            from .db import repositories as repo
+            with open(path, "rb") as file:
+                content = file.read()
+            stat = os.stat(path)
+            relative = os.path.relpath(path, self.report_dir).replace(os.sep, "/")
+            repo.insert_report_artifact(
+                get_connection(), source=self.name,
+                report_type="analyzer_summary",
+                generated_at=datetime.fromtimestamp(
+                    stat.st_mtime).astimezone().isoformat(timespec="seconds"),
+                payload={"format": "text",
+                         "content": content.decode("utf-8", errors="replace")},
+                original_path=f"{self.name}/{relative}",
+                content_sha256=hashlib.sha256(content).hexdigest(),
+                scan_id=scan_id,
+            )
+        except Exception as exc:
+            if not self.quiet:
+                print(f"WARN: report artifact DB write skipped ({exc})")

app/db/backfill.py

@@ -0,0 +1,361 @@
+#!/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/<source>/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 hashlib
+import json
+import os
+import re
+import sys
+from datetime import datetime, timezone
+
+from ..utils.constants import (
+    APP_REPORT_DIR,
+    APP_REPORT_DATA_DIR,
+    DATA_DIR,
+    MIKROTIK_DATA_DIR,
+    MIKROTIK_REPORT_DIR,
+    NPM_REPORT_DIR,
+    COPYPARTY_REPORT_DIR,
+)
+
+MAX_REPORT_ARTIFACT_BYTES = 10 * 1024 * 1024
+CANONICAL_REPORT_EXPORTS = {
+    "requests.csv", "ips.json", "ips_ipv4.csv", "ips_ipv6.csv",
+    "blocked_ipv4.txt", "blocked_ipv6.txt", "address_list_ipv4.json",
+    "address_list_ipv6.json", "firewall_ipv4.json", "firewall_ipv6.json",
+}
+
+
+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 backfill_report_artifacts(conn, dry_run: bool = False,
+                              roots: dict[str, str] | None = None) -> dict:
+    """Import supported report files as versioned, hash-addressed artifacts.
+
+    Canonical event exports and current firewall/blocklist data are skipped;
+    those belong in their normalized tables. Original files are never changed
+    or removed. Oversized or unreadable files are reported for manual review.
+    """
+    from . import repositories as repo
+
+    roots = roots or {
+        "copyparty": COPYPARTY_REPORT_DIR,
+        "npm": NPM_REPORT_DIR,
+        "app": APP_REPORT_DIR,
+        "mikrotik": MIKROTIK_REPORT_DIR,
+    }
+    result = {
+        "discovered": 0, "imported": 0, "would_import": 0,
+        "already_imported": 0, "skipped_canonical": [], "unmapped": [],
+        "by_source": {}, "by_format": {}, "dry_run": dry_run,
+    }
+
+    def count(bucket: str, key: str, field: str) -> None:
+        stats = result[bucket].setdefault(key, {})
+        stats[field] = stats.get(field, 0) + 1
+
+    for source, root in roots.items():
+        if not os.path.isdir(root):
+            continue
+        for directory, _, filenames in os.walk(root):
+            for filename in sorted(filenames):
+                path = os.path.join(directory, filename)
+                suffix = os.path.splitext(filename)[1].lower()
+                if suffix not in (".txt", ".json", ".csv", ".rsc"):
+                    continue
+                relative = os.path.relpath(path, root).replace(os.sep, "/")
+                result["discovered"] += 1
+                report_format = suffix.lstrip(".")
+                count("by_source", source, "discovered")
+                count("by_format", report_format, "discovered")
+                if filename.lower() in CANONICAL_REPORT_EXPORTS:
+                    result["skipped_canonical"].append(
+                        {"source": source, "path": relative})
+                    count("by_source", source, "skipped_canonical")
+                    count("by_format", report_format, "skipped_canonical")
+                    continue
+
+                try:
+                    size = os.path.getsize(path)
+                    if size > MAX_REPORT_ARTIFACT_BYTES:
+                        raise ValueError(
+                            f"file exceeds {MAX_REPORT_ARTIFACT_BYTES} bytes")
+                    with open(path, "rb") as file:
+                        raw = file.read()
+                    text = raw.decode("utf-8", errors="replace")
+                    payload = _report_payload(suffix, text)
+                    digest = hashlib.sha256(raw).hexdigest()
+                    modified = datetime.fromtimestamp(
+                        os.path.getmtime(path), timezone.utc)
+                    generated_at = _report_generated_at(filename, modified)
+                except (OSError, ValueError) as exc:
+                    result["unmapped"].append(
+                        {"source": source, "path": relative,
+                         "reason": str(exc)})
+                    count("by_source", source, "unmapped")
+                    count("by_format", report_format, "unmapped")
+                    continue
+
+                original_path = f"{source}/{relative}"
+                exists = conn.execute(
+                    "SELECT 1 FROM report_artifacts "
+                    "WHERE original_path=? AND content_sha256=?",
+                    (original_path, digest),
+                ).fetchone()
+                if exists:
+                    result["already_imported"] += 1
+                    count("by_source", source, "already_imported")
+                    count("by_format", report_format, "already_imported")
+                    continue
+                if dry_run:
+                    result["would_import"] += 1
+                    count("by_source", source, "would_import")
+                    count("by_format", report_format, "would_import")
+                    continue
+
+                try:
+                    repo.insert_report_artifact(
+                        conn, source=source,
+                        report_type=_report_type(suffix),
+                        generated_at=generated_at, payload=payload,
+                        original_path=original_path,
+                        content_sha256=digest,
+                    )
+                    result["imported"] += 1
+                    count("by_source", source, "imported")
+                    count("by_format", report_format, "imported")
+                except (OSError, ValueError, TypeError) as exc:
+                    result["unmapped"].append(
+                        {"source": source, "path": relative,
+                         "reason": str(exc)})
+                    count("by_source", source, "unmapped")
+                    count("by_format", report_format, "unmapped")
+    return {"reports": result}
+
+
+def _report_payload(suffix: str, text: str) -> dict:
+    if suffix == ".json":
+        try:
+            return {"format": "json", "data": json.loads(text)}
+        except ValueError as exc:
+            return {"format": "json", "content": text,
+                    "parse_error": str(exc)}
+    if suffix == ".csv":
+        try:
+            reader = csv.DictReader(text.splitlines())
+            return {"format": "csv", "columns": reader.fieldnames or [],
+                    "rows": list(reader)}
+        except csv.Error as exc:
+            return {"format": "csv", "content": text,
+                    "parse_error": str(exc)}
+    return {"format": "text", "content": text}
+
+
+def _report_type(suffix: str) -> str:
+    return {".json": "json_report", ".csv": "csv_report",
+            ".rsc": "routeros_script"}.get(suffix, "text_report")
+
+
+def _report_generated_at(filename: str, modified: datetime) -> str:
+    match = re.search(
+        r"(?P<date>\d{4}-\d{2}-\d{2})(?:[-_T](?P<hour>\d{2})[-_:]"
+        r"(?P<minute>\d{2})[-_:](?P<second>\d{2}))?", filename)
+    if not match:
+        return modified.isoformat(timespec="seconds")
+    values = match.groupdict()
+    hour = int(values["hour"] or 0)
+    minute = int(values["minute"] or 0)
+    second = int(values["second"] or 0)
+    try:
+        parsed = datetime.strptime(values["date"], "%Y-%m-%d").replace(
+            hour=hour, minute=minute, second=second, tzinfo=timezone.utc)
+    except ValueError:
+        return modified.isoformat(timespec="seconds")
+    return parsed.isoformat(timespec="seconds")
+
+
+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")
+    p.add_argument("--reports-only", action="store_true",
+                   help="import legacy report artifacts only")
+    opts = p.parse_args(argv)
+
+    from . import get_connection
+    conn = get_connection()
+    print(f"== backfill (dry_run={opts.dry_run}) ==")
+    if opts.reports_only:
+        print("reports :", backfill_report_artifacts(
+            conn, dry_run=opts.dry_run))
+        print("== done ==")
+        return
+    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("reports  :", backfill_report_artifacts(conn, dry_run=opts.dry_run))
+    print("== done ==")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

app/db/repositories.py

@@ -10,8 +10,10 @@ from __future__ import annotations
 
 import json
 import sqlite3
+import base64
 from contextlib import contextmanager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
+import ipaddress
 from typing import Any, Iterator, Sequence
 
 from ..utils.models import IPStats
@@ -104,7 +106,8 @@ def _parse_ts(ts: str) -> tuple[str, float | None]:
     cleaned = ts.strip()
     # Accept a trailing timezone like +HH:MM or Z; fall back to naive parse.
     try:
-        epoch = datetime.fromisoformat(cleaned.replace("Z", "+00:00")).timestamp()
+        epoch = datetime.fromisoformat(
+            cleaned.replace("Z", "+00:00")).timestamp()
     except ValueError:
         try:
             epoch = datetime.fromisoformat(cleaned).timestamp()
@@ -231,6 +234,109 @@ def count_scans(conn: sqlite3.Connection) -> int:
     return int(conn.execute("SELECT COUNT(*) FROM scans").fetchone()[0])
 
 
+# ---------------------------------------------------------------------------
+# Report artifacts
+# ---------------------------------------------------------------------------
+
+def insert_report_artifact(conn: sqlite3.Connection, *, source: str,
+                           report_type: str, generated_at: str,
+                           payload: Any, original_path: str,
+                           content_sha256: str, scan_id: int | None = None,
+                           schema_version: int = 1) -> int:
+    """Insert a structured report artifact, idempotent by path and hash."""
+    with transaction(conn):
+        conn.execute(
+            "INSERT OR IGNORE INTO report_artifacts "
+            "(source, scan_id, report_type, generated_at, schema_version, "
+            "payload_json, original_path, content_sha256, created_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (source, scan_id, report_type, generated_at, schema_version,
+             json.dumps(payload, ensure_ascii=False), original_path,
+             content_sha256, utcnow()),
+        )
+        row = conn.execute(
+            "SELECT id FROM report_artifacts "
+            "WHERE original_path=? AND content_sha256=?",
+            (original_path, content_sha256),
+        ).fetchone()
+    return int(row["id"])
+
+
+def list_report_artifacts(conn: sqlite3.Connection, *,
+                          source: str | None = None,
+                          report_type: str | None = None,
+                          since: str | None = None,
+                          until: str | None = None,
+                          limit: int = 50,
+                          cursor: str | None = None) -> dict[str, Any]:
+    """List report metadata using bounded keyset pagination."""
+    limit = min(max(int(limit), 1), 200)
+    clauses: list[str] = []
+    args: list[Any] = []
+    if source:
+        clauses.append("source=?")
+        args.append(source)
+    if report_type:
+        clauses.append("report_type=?")
+        args.append(report_type)
+    if since:
+        clauses.append("generated_at>=?")
+        args.append(since)
+    if until:
+        clauses.append("generated_at<=?")
+        args.append(until)
+    if cursor:
+        generated_at, artifact_id = _decode_report_cursor(cursor)
+        clauses.append("(generated_at<? OR (generated_at=? AND id<?))")
+        args.extend((generated_at, generated_at, artifact_id))
+
+    sql = (
+        "SELECT id, source, scan_id, report_type, generated_at, "
+        "schema_version, original_path, content_sha256 "
+        "FROM report_artifacts"
+    )
+    if clauses:
+        sql += " WHERE " + " AND ".join(clauses)
+    sql += " ORDER BY generated_at DESC, id DESC LIMIT ?"
+    rows = [dict(row) for row in conn.execute(sql, (*args, limit + 1))]
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+    next_cursor = (
+        _encode_report_cursor(rows[-1]["generated_at"], rows[-1]["id"])
+        if has_more and rows else None
+    )
+    return {"items": rows, "next_cursor": next_cursor}
+
+
+def get_report_artifact(conn: sqlite3.Connection,
+                        artifact_id: int) -> dict[str, Any] | None:
+    """Fetch one artifact and decode its structured payload."""
+    row = conn.execute(
+        "SELECT id, source, scan_id, report_type, generated_at, "
+        "schema_version, payload_json, original_path, content_sha256, created_at "
+        "FROM report_artifacts WHERE id=?", (artifact_id,)).fetchone()
+    if row is None:
+        return None
+    artifact = dict(row)
+    artifact["payload"] = json.loads(artifact.pop("payload_json"))
+    return artifact
+
+
+def _encode_report_cursor(generated_at: str, artifact_id: int) -> str:
+    raw = f"{generated_at}\n{artifact_id}".encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _decode_report_cursor(cursor: str) -> tuple[str, int]:
+    try:
+        padded = cursor + "=" * (-len(cursor) % 4)
+        generated_at, artifact_id = base64.urlsafe_b64decode(
+            padded.encode("ascii")).decode("utf-8").split("\n", 1)
+        return generated_at, int(artifact_id)
+    except (ValueError, UnicodeError) as exc:
+        raise ValueError("invalid report cursor") from exc
+
+
 def correlate_across_sources(conn: sqlite3.Connection,
                              limit: int = 100) -> list[dict[str, Any]]:
     """IPs seen in more than one log source, with per-source counts."""
@@ -322,6 +428,78 @@ def unblock_entry(conn: sqlite3.Connection, cidr: str, *,
     return removed
 
 
+def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
+                     added_by: str | None = None, reasons=None) -> dict[str, Any]:
+    """Grow-only record: add any ``cidrs`` not already active to the ledger.
+
+    Unlike :func:`sync_blocklist`, this **never removes** entries. It mirrors
+    the router's cumulative model: once an IP/subnet has ever been blocked it
+    stays active even after it leaves the scan window (e.g. NPM ~30-day log
+    retention), exactly like the old git-history backfill. Returns the added
+    delta and the resulting active count.
+    """
+    cidr_set = set(cidrs)
+    before = {r["cidr"] for r in active_blocklist(conn)}
+    to_add = sorted(cidr_set - before)
+    now = utcnow()
+    with transaction(conn):
+        for cidr in to_add:
+            _block_entry_inner(conn, cidr, added_by=added_by, last_seen=now,
+                               reasons=reasons)
+        # Refresh last_seen on entries that are still desired so the ledger
+        # reflects recency for existing rows too.
+        conn.executemany(
+            "UPDATE blocklist_entries SET last_seen=? WHERE cidr=? AND status='active'",
+            [(now, cidr) for cidr in cidr_set.intersection(before)])
+    return {"added": to_add, "active": len(before) + len(to_add)}
+
+
+def prune_old_single_ip_blocks(conn: sqlite3.Connection, *,
+                               older_than_days: int = 30,
+                               actor: str | None = "retention") -> list[str]:
+    """Remove active single-host entries older than ``older_than_days``.
+
+    Keeps CIDR ranges/subnets intact. Host entries are detected via
+    ``ipaddress.ip_network(..., strict=False)`` and only removed when the
+    network is exactly one address (/32 for IPv4, /128 for IPv6).
+    """
+    cutoff = datetime.now(timezone.utc) - \
+        timedelta(days=max(0, older_than_days))
+    rows = conn.execute(
+        "SELECT cidr, added_at, last_seen FROM blocklist_entries "
+        "WHERE status='active'").fetchall()
+
+    to_remove: list[str] = []
+    for row in rows:
+        cidr = str(row["cidr"])
+        if not _is_single_host_cidr(cidr):
+            continue
+        seen = _parse_iso_ts(
+            row["last_seen"]) or _parse_iso_ts(row["added_at"])
+        if seen is None:
+            continue
+        if seen < cutoff:
+            to_remove.append(cidr)
+
+    if not to_remove:
+        return []
+
+    now = utcnow()
+    with transaction(conn):
+        conn.executemany(
+            "UPDATE blocklist_entries SET status='removed', removed_at=? "
+            "WHERE cidr=? AND status='active'",
+            [(now, cidr) for cidr in to_remove],
+        )
+        conn.executemany(
+            "INSERT INTO blocklist_events (action, cidr, actor, fired_at, detail) "
+            "VALUES ('remove', ?, ?, ?, ?)",
+            [(cidr, actor, now, f"single-ip-retention>{older_than_days}d")
+             for cidr in to_remove],
+        )
+    return sorted(to_remove)
+
+
 def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *,
                    added_by: str | None = None) -> dict[str, Any]:
     """Reconcile active blocklist to ``desired_cidrs``.
@@ -376,6 +554,26 @@ def _cidr_version(cidr: str) -> int:
         return 4 if ":" not in str(cidr) else 6
 
 
+def _is_single_host_cidr(cidr: str) -> bool:
+    try:
+        net = ipaddress.ip_network(cidr, strict=False)
+        return net.num_addresses == 1
+    except ValueError:
+        return False
+
+
+def _parse_iso_ts(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
 # ---------------------------------------------------------------------------
 # Whitelists
 # ---------------------------------------------------------------------------
@@ -465,6 +663,20 @@ def save_firewall_snapshot(conn: sqlite3.Connection, *,
     return snapshot_id
 
 
+def recent_firewall_snapshots(conn: sqlite3.Connection,
+                              limit: int = 10) -> list[dict[str, Any]]:
+    """Most recent firewall snapshots with entry counts (F001)."""
+    rows = conn.execute(
+        "SELECT fs.id, fs.captured_at, fs.kind, fs.trigger, "
+        "(SELECT COUNT(*) FROM firewall_address_list_entries e "
+        " WHERE e.snapshot_id = fs.id) AS addr_entries, "
+        "(SELECT COUNT(*) FROM firewall_rule_entries r "
+        " WHERE r.snapshot_id = fs.id) AS rule_entries "
+        "FROM firewall_snapshots fs ORDER BY fs.id DESC LIMIT ?",
+        (limit,)).fetchall()
+    return [dict(r) for r in rows]
+
+
 # ---------------------------------------------------------------------------
 # WHOIS cache
 # ---------------------------------------------------------------------------

app/db/schema.py

@@ -14,7 +14,7 @@ import sqlite3
 from typing import Iterable
 
 #: Latest schema version. Bump by one whenever a new migration is appended.
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
 #: Ordered migrations: index -> list of DDL statements. A migration N upgrades a
 #: database from user_version N-1 to N. Never edit an older migration in place;
@@ -31,14 +31,29 @@ MIGRATIONS: dict[int, list[str]] = {
             log_path TEXT,
             enabled INTEGER NOT NULL DEFAULT 1
         )
-        """,
         """
-        INSERT OR IGNORE INTO log_sources (name, log_path)
+    ],
-        VALUES ('copyparty', NULL), ('npm', NULL)
+
+    2: [
+        """
+        CREATE TABLE IF NOT EXISTS report_artifacts (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            source TEXT NOT NULL,
+            scan_id INTEGER,
+            report_type TEXT NOT NULL,
+            generated_at TEXT NOT NULL,
+            schema_version INTEGER NOT NULL DEFAULT 1,
+            payload_json TEXT NOT NULL,
+            original_path TEXT NOT NULL,
+            content_sha256 TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            UNIQUE (original_path, content_sha256)
+        )
         """,
-        # ------------------------------------------------------------------
+        "CREATE INDEX IF NOT EXISTS idx_report_artifacts_source_time "
-        # requests — one row per parsed request; insert-only
+        "ON report_artifacts (source, generated_at DESC, id DESC)",
-        # ------------------------------------------------------------------
+        "CREATE INDEX IF NOT EXISTS idx_report_artifacts_type_time "
+        "ON report_artifacts (report_type, generated_at DESC, id DESC)",
         """
         CREATE TABLE IF NOT EXISTS requests (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -261,7 +276,8 @@ def apply_migrations(conn: sqlite3.Connection) -> None:
     for version in range(current + 1, SCHEMA_VERSION + 1):
         statements = MIGRATIONS.get(version)
         if statements is None:
-            raise RuntimeError(f"missing migration for schema version {version}")
+            raise RuntimeError(
+                f"missing migration for schema version {version}")
         with conn:
             for ddl in statements:
                 conn.execute(ddl)

app/scan/__init__.py

@@ -1,393 +1,431 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python3
-"""Consolidated intrusion detection log scanner.
+"""Consolidated intrusion detection log scanner.
-
+
-Runs both copyparty and Nginx Proxy Manager (NPM) log analyzers,
+Runs both copyparty and Nginx Proxy Manager (NPM) log analyzers,
-and outputs a single consolidated MikroTik RouterOS blocklist.
+and outputs a single consolidated MikroTik RouterOS blocklist.
-"""
+"""
-
+
-from __future__ import annotations
+from __future__ import annotations
-
+
-import argparse
+import argparse
-import os
+import os
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field
-
+
-from ..utils.blocklist_history import (
+from ..utils.blocklist_history import (
-    committed_lines,
+    committed_lines,
-    cumulative_blocklist,
+    cumulative_blocklist,
-    current_file_lines,
+    current_file_lines,
-    write_blocklist_report,
+    write_blocklist_report,
-)
+)
-from ..utils.constants import (
+from ..utils.constants import (
-    COPYPARTY_LOG_PATH,
+    COPYPARTY_LOG_PATH,
-    COPYPARTY_REPORT_DIR,
+    COPYPARTY_REPORT_DIR,
-    DEFAULT_ERROR_RATIO,
+    DEFAULT_ERROR_RATIO,
-    DEFAULT_MIN_DENIED,
+    DEFAULT_MIN_DENIED,
-    DEFAULT_MIN_ERRORS,
+    DEFAULT_MIN_ERRORS,
-    DEFAULT_TOP,
+    DEFAULT_TOP,
-    NPM_LOG_PATH,
+    NPM_LOG_PATH,
-    NPM_REPORT_DIR,
+    NPM_REPORT_DIR,
-    REPORT_DIR,
+    REPORT_DIR,
-    REPORT_RETENTION_DAYS,
+    REPORT_RETENTION_DAYS,
-    REPORT_RETENTION_MAX_FILES,
+    REPORT_RETENTION_MAX_FILES,
-    SUBNET_THRESHOLD,
+    SUBNET_THRESHOLD,
-)
+)
-from ..firewall import MikrotikFirewall, write_blocklist_files
+from ..firewall import MikrotikFirewall, write_blocklist_files
-from ..utils.retention import prune_report_dirs
+from ..utils.retention import prune_report_dirs
-from ..copyparty import CopypartyAnalyzer
+from ..copyparty import CopypartyAnalyzer
-from ..npm import NpmAnalyzer
+from ..npm import NpmAnalyzer
-from ..utils.ip_utils import is_whitelisted
+from ..utils.ip_utils import is_whitelisted
-from ..mikrotik import split_cidr_versions
+from ..mikrotik import split_cidr_versions
-from ..utils.whois import consolidate_by_subnet
+from ..utils.whois import consolidate_by_subnet
-
+
-
+
-def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
+def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
-                          desired_v6: list[str], added_by: str | None = None) -> None:
+                          desired_v6: list[str],
-    """Best-effort sync of the consolidated blocklist targets into SQLite.
+                          added_by: str | None = None) -> dict | None:
-
+    """Best-effort grow-only sync of block targets into the SQLite ledger.
-    Keep the DB ledger in sync regardless of whether a router sync happened so
+
-    analysis always sees the current policy."""
+    Uses ``record_blocklist`` (add-only) so the ledger is the authoritative
-    try:
+    cumulative record: once an IP is ever blocked it stays active even after it
-        from ..db import get_connection
+    leaves the scan window, matching the router's cumulative model. Returns the
-        from ..db import repositories as repo
+    added delta or None if the DB is unavailable.
-        conn = get_connection()
+    """
-        repo.sync_blocklist(conn, list(desired_v4) + list(desired_v6),
+    try:
-                            added_by=added_by)
+        from ..db import get_connection
-    except Exception as exc:  # never break the scan for a DB issue
+        from ..db import repositories as repo
-        print(f"WARN: could not sync blocklist to db ({exc})")
+        conn = get_connection()
-
+        result = repo.record_blocklist(
-
+            conn, list(desired_v4) + list(desired_v6), added_by=added_by)
-@dataclass
+        # Retention policy: expire stale single-host entries after 30 days
-class ScanParams:
+        # while keeping CIDR ranges/subnets intact.
-    """Configuration for a consolidated scan.
+        pruned = repo.prune_old_single_ip_blocks(conn, older_than_days=30)
-
+        result["pruned_single_ips"] = pruned
-    Shared by the CLI entry point and the web interface so both run the
+        return result
-    exact same pipeline.
+    except Exception as exc:  # never break the scan for a DB issue
-    """
+        print(f"WARN: could not sync blocklist to db ({exc})")
-
+        return None
-    copyparty_log_path: str = COPYPARTY_LOG_PATH
+
-    copyparty_report_dir: str = COPYPARTY_REPORT_DIR
+
-    min_denied: int = DEFAULT_MIN_DENIED
+def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]):
-    npm_log_path: str = NPM_LOG_PATH
+    """Complete cumulative blocklist from the DB ledger (B001).
-    npm_report_dir: str = NPM_REPORT_DIR
+
-    min_errors: int = DEFAULT_MIN_ERRORS
+    Returns ``(complete_v4, complete_v6)`` as **sets** (same contract as
-    error_ratio: float = DEFAULT_ERROR_RATIO
+    ``cumulative_blocklist``) derived from the active blocklist ledger plus
-    report_dir: str = REPORT_DIR
+    this scan's newly desired targets (in case the DB write for this scan has
-    top: int = DEFAULT_TOP
+    not run yet). Falls back to the git-history reconstruction when the DB is
-    quiet: bool = False
+    unavailable, so nothing regresses.
-    full: bool = False
+    """
-    update_firewall: bool = False
+    try:
-    dry_run: bool = False
+        from ..db import get_connection
-
+        from ..db import repositories as repo
-
+        conn = get_connection()
-@dataclass
+        active = [e["cidr"] for e in repo.active_blocklist(conn)]
-class ScanResult:
+    except Exception:
-    """Structured outcome of a consolidated scan."""
+        active = []
-
+    union = set(active) | set(desired_v4) | set(desired_v6)
-    ok: bool
+    v4, v6 = split_cidr_versions(union)
-    cp_exists: bool = False
+    return set(v4), set(v6)
-    npm_exists: bool = False
+
-    cp_candidates: int = 0
+
-    npm_candidates: int = 0
+@dataclass
-    blocked_ips: int = 0
+class ScanParams:
-    blocked_targets: int = 0
+    """Configuration for a consolidated scan.
-    blocked_ipv4: list[str] = field(default_factory=list)
+
-    blocked_ipv6: list[str] = field(default_factory=list)
+    Shared by the CLI entry point and the web interface so both run the
-    blocklist_v4_path: str | None = None
+    exact same pipeline.
-    blocklist_v6_path: str | None = None
+    """
-    firewall_connected: bool = False
+
-    firewall_changes: dict | None = None
+    copyparty_log_path: str = COPYPARTY_LOG_PATH
-    blocklist_report_path: str | None = None
+    copyparty_report_dir: str = COPYPARTY_REPORT_DIR
-    errors: list[str] = field(default_factory=list)
+    min_denied: int = DEFAULT_MIN_DENIED
-
+    npm_log_path: str = NPM_LOG_PATH
-
+    npm_report_dir: str = NPM_REPORT_DIR
-def run_scan(params: ScanParams) -> ScanResult:
+    min_errors: int = DEFAULT_MIN_ERRORS
-    """Run the full consolidated scan pipeline and return structured results.
+    error_ratio: float = DEFAULT_ERROR_RATIO
-
+    report_dir: str = REPORT_DIR
-    Steps: connect to MikroTik (optional) -> snapshot state -> run copyparty
+    top: int = DEFAULT_TOP
-    + NPM analyzers -> consolidate candidates -> write blocklists -> compare
+    quiet: bool = False
-    and apply firewall changes (optional).
+    full: bool = False
-    """
+    update_firewall: bool = False
-    errors: list[str] = []
+    dry_run: bool = False
-
+
-    # --- Step 0: open the SQLite store for this consolidated run -------------
+
-    db_scan_id: int | None = None
+@dataclass
-    try:
+class ScanResult:
-        from ..db import get_connection
+    """Structured outcome of a consolidated scan."""
-        from ..db import repositories as repo
+
-        from ..utils.whitelists import sync_whitelists_to_db
+    ok: bool
-        conn = get_connection()
+    cp_exists: bool = False
-        sync_whitelists_to_db()
+    npm_exists: bool = False
-        params_map = {
+    cp_candidates: int = 0
-            "full": params.full, "dry_run": params.dry_run,
+    npm_candidates: int = 0
-            "update_firewall": params.update_firewall,
+    blocked_ips: int = 0
-            "quiet": params.quiet, "top": params.top,
+    blocked_targets: int = 0
-        }
+    blocked_ipv4: list[str] = field(default_factory=list)
-        db_scan_id = repo.start_scan(conn, "ids", params=params_map,
+    blocked_ipv6: list[str] = field(default_factory=list)
-                                     full=params.full, dry_run=params.dry_run)
+    blocklist_v4_path: str | None = None
-    except Exception as exc:
+    blocklist_v6_path: str | None = None
-        print(f"WARN: sqlite store unavailable ({exc})")
+    firewall_connected: bool = False
-
+    firewall_changes: dict | None = None
-    # --- Step 1-3: connect to MikroTik RouterOS and capture current state ---
+    blocklist_report_path: str | None = None
-    # If the router is unreachable we fall back to scan-only mode: still
+    errors: list[str] = field(default_factory=list)
-    # generate reports and blocklists, but skip the firewall sync.
+
-    firewall: MikrotikFirewall | None = None
+
-    if params.update_firewall:
+def run_scan(params: ScanParams) -> ScanResult:
-        firewall = MikrotikFirewall()
+    """Run the full consolidated scan pipeline and return structured results.
-        try:
+
-            firewall.connect()
+    Steps: connect to MikroTik (optional) -> snapshot state -> run copyparty
-            # Read current firewall rules + address lists and snapshot them to
+    + NPM analyzers -> consolidate candidates -> write blocklists -> compare
-            # reports/mikrotik/data/ and the git-diffable data/mikrotik/.
+    and apply firewall changes (optional).
-            firewall.save_state_report(params.report_dir)
+    """
-        except Exception as e:
+    errors: list[str] = []
-            message = f"could not read MikroTik router ({firewall.router_ip}): {e}"
+
-            errors.append(message)
+    # --- Step 0: open the SQLite store for this consolidated run -------------
-            print(f"\nWARNING: {message}")
+    db_scan_id: int | None = None
-            firewall.disconnect()
+    try:
-            firewall = None
+        from ..db import get_connection
-
+        from ..db import repositories as repo
-    # --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------
+        from ..utils.whitelists import sync_whitelists_to_db
-    cp_analyzer = CopypartyAnalyzer(
+        conn = get_connection()
-        log_path=params.copyparty_log_path,
+        sync_whitelists_to_db()
-        report_dir=params.copyparty_report_dir,
+        params_map = {
-        top=params.top,
+            "full": params.full, "dry_run": params.dry_run,
-        quiet=params.quiet,
+            "update_firewall": params.update_firewall,
-        full=params.full,
+            "quiet": params.quiet, "top": params.top,
-        min_denied=params.min_denied,
+        }
-    )
+        db_scan_id = repo.start_scan(conn, "ids", params=params_map,
-
+                                     full=params.full, dry_run=params.dry_run)
-    npm_analyzer = NpmAnalyzer(
+    except Exception as exc:
-        log_path=params.npm_log_path,
+        print(f"WARN: sqlite store unavailable ({exc})")
-        report_dir=params.npm_report_dir,
+
-        top=params.top,
+    # --- Step 1-3: connect to MikroTik RouterOS and capture current state ---
-        quiet=params.quiet,
+    # If the router is unreachable we fall back to scan-only mode: still
-        full=params.full,
+    # generate reports and blocklists, but skip the firewall sync.
-        min_errors=params.min_errors,
+    firewall: MikrotikFirewall | None = None
-        error_ratio=params.error_ratio,
+    if params.update_firewall:
-    )
+        firewall = MikrotikFirewall()
-
+        try:
-    print("=== Running copyparty scan ===")
+            firewall.connect()
-    cp_exists = True
+            # Read current firewall rules + address lists and snapshot them to
-    try:
+            # reports/mikrotik/data/ and the git-diffable data/mikrotik/.
-        cp_analyzer.run(db_scan_id=db_scan_id)
+            firewall.save_state_report(params.report_dir)
-    except Exception as e:
+        except Exception as e:
-        errors.append(f"copyparty scan failed: {e}")
+            message = f"could not read MikroTik router ({firewall.router_ip}): {e}"
-        print(f"Error running copyparty scan: {e}")
+            errors.append(message)
-        cp_exists = False
+            print(f"\nWARNING: {message}")
-
+            firewall.disconnect()
-    print("\n=== Running NPM scan ===")
+            firewall = None
-    npm_exists = True
+
-    try:
+    # --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------
-        npm_analyzer.run(db_scan_id=db_scan_id)
+    cp_analyzer = CopypartyAnalyzer(
-    except Exception as e:
+        log_path=params.copyparty_log_path,
-        errors.append(f"NPM scan failed: {e}")
+        report_dir=params.copyparty_report_dir,
-        print(f"Error running NPM scan: {e}")
+        top=params.top,
-        npm_exists = False
+        quiet=params.quiet,
-
+        full=params.full,
-    if not cp_exists and not npm_exists:
+        min_denied=params.min_denied,
-        message = "Both scans failed. Unable to generate consolidated blocklist."
+    )
-        errors.append(message)
+
-        print(f"\n{message}")
+    npm_analyzer = NpmAnalyzer(
-        if firewall is not None:
+        log_path=params.npm_log_path,
-            firewall.disconnect()
+        report_dir=params.npm_report_dir,
-        if db_scan_id is not None:
+        top=params.top,
-            try:
+        quiet=params.quiet,
-                from ..db import get_connection
+        full=params.full,
-                from ..db import repositories as repo
+        min_errors=params.min_errors,
-                repo.finish_scan(get_connection(), db_scan_id, status="error",
+        error_ratio=params.error_ratio,
-                                 error="; ".join(errors))
+    )
-            except Exception:
+
-                pass
+    print("=== Running copyparty scan ===")
-        return ScanResult(ok=False, errors=errors)
+    cp_exists = True
-
+    try:
-    # Consolidate block candidates
+        cp_analyzer.run(db_scan_id=db_scan_id)
-    consolidated_candidates: set[str] = set()
+    except Exception as e:
-    if cp_exists:
+        errors.append(f"copyparty scan failed: {e}")
-        consolidated_candidates.update(cp_analyzer.block_candidates())
+        print(f"Error running copyparty scan: {e}")
-    if npm_exists:
+        cp_exists = False
-        consolidated_candidates.update(npm_analyzer.block_candidates())
+
-
+    print("\n=== Running NPM scan ===")
-    # Exclude whitelisted IPs
+    npm_exists = True
-    final_blocked_ips = {
+    try:
-        ip for ip in consolidated_candidates if not is_whitelisted(ip)}
+        npm_analyzer.run(db_scan_id=db_scan_id)
-
+    except Exception as e:
-    # Group into subnets via WHOIS if multiple malicious IPs in same subnet
+        errors.append(f"NPM scan failed: {e}")
-    print("\nGrouping blocked IPs into subnets using WHOIS...")
+        print(f"Error running NPM scan: {e}")
-    final_blocked_targets = consolidate_by_subnet(
+        npm_exists = False
-        final_blocked_ips, threshold=SUBNET_THRESHOLD)
+
-    blocked_ipv4, blocked_ipv6 = split_cidr_versions(final_blocked_targets)
+    if not cp_exists and not npm_exists:
-
+        message = "Both scans failed. Unable to generate consolidated blocklist."
-    cp_candidates = len(cp_analyzer.block_candidates()) if cp_exists else 0
+        errors.append(message)
-    npm_candidates = len(npm_analyzer.block_candidates()) if npm_exists else 0
+        print(f"\n{message}")
-
+        if firewall is not None:
-    print("\n=== Consolidated Summary ===")
+            firewall.disconnect()
-    print(f"copyparty block candidates: {cp_candidates}")
+        if db_scan_id is not None:
-    print(f"NPM block candidates:       {npm_candidates}")
+            try:
-    print(f"Total unified blocked IPs:  {len(final_blocked_ips)}")
+                from ..db import get_connection
-    print(f"Total target blocks/subnets: {len(final_blocked_targets)}")
+                from ..db import repositories as repo
-
+                repo.finish_scan(get_connection(), db_scan_id, status="error",
-    # --- Step 6: write consolidated blocklists to reports/app/data ----------
+                                 error="; ".join(errors))
-    # Capture the previous blocklist state BEFORE the files are overwritten so
+            except Exception:
-    # the change report can compute the complete-set delta.
+                pass
-    app_data_dir = os.path.join(params.report_dir, "app", "data")
+        return ScanResult(ok=False, errors=errors)
-    blocklist_v4_path = os.path.join(app_data_dir, "blocklist_ipv4.txt")
+
-    blocklist_v6_path = os.path.join(app_data_dir, "blocklist_ipv6.txt")
+    # Consolidate block candidates
-    prev_v4 = current_file_lines(blocklist_v4_path)
+    consolidated_candidates: set[str] = set()
-    prev_v6 = current_file_lines(blocklist_v6_path)
+    if cp_exists:
-    history_v4 = committed_lines(blocklist_v4_path)
+        consolidated_candidates.update(cp_analyzer.block_candidates())
-    history_v6 = committed_lines(blocklist_v6_path)
+    if npm_exists:
-
+        consolidated_candidates.update(npm_analyzer.block_candidates())
-    write_blocklist_files(blocked_ipv4, blocked_ipv6, params.report_dir)
+
-    print(f"blocklist files:\n  {blocklist_v4_path}\n  {blocklist_v6_path}")
+    # Exclude whitelisted IPs
-
+    final_blocked_ips = {
-    # --- Steps 5+7: compare current state with candidates and apply ---------
+        ip for ip in consolidated_candidates if not is_whitelisted(ip)}
-    firewall_changes: dict | None = None
+
-    blocklist_report_path: str | None = None
+    # Group into subnets via WHOIS if multiple malicious IPs in same subnet
-    if firewall is not None:
+    print("\nGrouping blocked IPs into subnets using WHOIS...")
-        try:
+    final_blocked_targets = consolidate_by_subnet(
-            firewall_changes = firewall.sync_blocklist(
+        final_blocked_ips, threshold=SUBNET_THRESHOLD)
-                blocked_ipv4, blocked_ipv6, dry_run=params.dry_run,
+    blocked_ipv4, blocked_ipv6 = split_cidr_versions(final_blocked_targets)
-                report_dir=params.report_dir)
+
-        finally:
+    cp_candidates = len(cp_analyzer.block_candidates()) if cp_exists else 0
-            firewall.disconnect()
+    npm_candidates = len(npm_analyzer.block_candidates()) if npm_exists else 0
-        action = "Would" if params.dry_run else ""
+
-        print(f"{action} add {len(firewall_changes['added_ipv4'])} IPv4 and "
+    print("\n=== Consolidated Summary ===")
-              f"{len(firewall_changes['added_ipv6'])} IPv6 addresses; "
+    print(f"copyparty block candidates: {cp_candidates}")
-              f"remove {len(firewall_changes['removed_ipv4'])} IPv4 and "
+    print(f"NPM block candidates:       {npm_candidates}")
-              f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
+    print(f"Total unified blocked IPs:  {len(final_blocked_ips)}")
-
+    print(f"Total target blocks/subnets: {len(final_blocked_targets)}")
-        # Timestamped change report — only when the router was reachable.
+
-        complete_v4, complete_v6 = cumulative_blocklist(
+    # --- Step 6: write consolidated blocklists to reports/app/data ----------
-            blocklist_v4_path, blocklist_v6_path,
+    # Capture the previous blocklist state BEFORE the files are overwritten so
-            blocked_ipv4, blocked_ipv6)
+    # the change report can compute the complete-set delta.
-        blocklist_report_path = write_blocklist_report(
+    app_data_dir = os.path.join(params.report_dir, "app", "data")
-            params.report_dir,
+    blocklist_v4_path = os.path.join(app_data_dir, "blocklist_ipv4.txt")
-            history_v4=history_v4, history_v6=history_v6,
+    blocklist_v6_path = os.path.join(app_data_dir, "blocklist_ipv6.txt")
-            prev_v4=prev_v4, prev_v6=prev_v6,
+    prev_v4 = current_file_lines(blocklist_v4_path)
-            complete_v4=complete_v4, complete_v6=complete_v6,
+    prev_v6 = current_file_lines(blocklist_v6_path)
-            firewall_changes=firewall_changes,
+    history_v4 = committed_lines(blocklist_v4_path)
-            dry_run=params.dry_run,
+    history_v6 = committed_lines(blocklist_v6_path)
-            errors=errors)
+
-        print(f"blocklist report: {blocklist_report_path}")
+    write_blocklist_files(blocked_ipv4, blocked_ipv6, params.report_dir)
-
+    print(f"blocklist files:\n  {blocklist_v4_path}\n  {blocklist_v6_path}")
-    # --- Step 8: prune old reports (retention) ------------------------------
+
-    # Each report subdirectory keeps at most REPORT_RETENTION_MAX_FILES files,
+    # --- Steps 5+7: compare current state with candidates and apply ---------
-    # all newer than REPORT_RETENTION_DAYS days. The blocklist files in
+    firewall_changes: dict | None = None
-    # ``app/data`` are always protected. ``--dry-run`` only previews.
+    blocklist_report_path: str | None = None
-    pruned = prune_report_dirs(
+    if firewall is not None:
-        params.report_dir, params.copyparty_report_dir, params.npm_report_dir,
+        try:
-        dry_run=params.dry_run)
+            firewall_changes = firewall.sync_blocklist(
-    total_pruned = sum(len(paths) for paths in pruned.values())
+                blocked_ipv4, blocked_ipv6, dry_run=params.dry_run,
-    action = "Would prune" if params.dry_run else "Pruned"
+                report_dir=params.report_dir)
-    print(f"{action} {total_pruned} report file(s) "
+        finally:
-          f"(keep {REPORT_RETENTION_DAYS}d / "
+            firewall.disconnect()
-          f"max {REPORT_RETENTION_MAX_FILES} per dir)")
+        action = "Would" if params.dry_run else ""
-
+        print(f"{action} add {len(firewall_changes['added_ipv4'])} IPv4 and "
-    # --- Step 9: persist consolidated results to SQLite ---------------------
+              f"{len(firewall_changes['added_ipv6'])} IPv6 addresses; "
-    # Sync the blocklist ledger and close out the scan record regardless of
+              f"remove {len(firewall_changes['removed_ipv4'])} IPv4 and "
-    # whether the router sync was attempted.
+              f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
-    _sync_blocklist_to_db(
+
-        firewall is not None, blocked_ipv4, blocked_ipv6,
+        # Timestamped change report — only when the router was reachable.
-        added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}"))
+        # Source of truth for the cumulative set: the SQLite |ledger| first
-    if db_scan_id is not None:
+        # (B001), falling back to the git-history reconstruction only if the
-        try:
+        # DB is unavailable.
-            from ..db import get_connection
+        complete_v4, complete_v6 = _db_complete_blocklist(
-            from ..db import repositories as repo
+            blocked_ipv4, blocked_ipv6)
-            conn = get_connection()
+        if not complete_v4 and not complete_v6:
-            result_payload = {
+            complete_v4, complete_v6 = cumulative_blocklist(
-                "ok": True,
+                blocklist_v4_path, blocklist_v6_path,
-                "cp_candidates": cp_candidates,
+                blocked_ipv4, blocked_ipv6)
-                "npm_candidates": npm_candidates,
+        blocklist_report_path = write_blocklist_report(
-                "blocked_ips": len(final_blocked_ips),
+            params.report_dir,
-                "blocked_targets": len(final_blocked_targets),
+            history_v4=history_v4, history_v6=history_v6,
-                "firewall_connected": firewall is not None,
+            prev_v4=prev_v4, prev_v6=prev_v6,
-            }
+            complete_v4=complete_v4, complete_v6=complete_v6,
-            repo.finish_scan(
+            firewall_changes=firewall_changes,
-                conn, db_scan_id, status="done",
+            dry_run=params.dry_run,
-                result=result_payload, log_files=0,
+            errors=errors,
-                request_count=(cp_analyzer and len(cp_analyzer.requests) or 0)
+            scan_id=db_scan_id)
-                + (npm_analyzer and len(npm_analyzer.requests) or 0),
+        print(f"blocklist report: {blocklist_report_path}")
-                error="; ".join(errors) or None,
+
-            )
+    # --- Step 8: prune old reports (retention) ------------------------------
-        except Exception as exc:
+    # Each report subdirectory keeps at most REPORT_RETENTION_MAX_FILES files,
-            print(f"WARN: could not finalize scan in db ({exc})")
+    # all newer than REPORT_RETENTION_DAYS days. The blocklist files in
-
+    # ``app/data`` are always protected. ``--dry-run`` only previews.
-    return ScanResult(
+    pruned = prune_report_dirs(
-        ok=True,
+        params.report_dir, params.copyparty_report_dir, params.npm_report_dir,
-        cp_exists=cp_exists,
+        dry_run=params.dry_run)
-        npm_exists=npm_exists,
+    total_pruned = sum(len(paths) for paths in pruned.values())
-        cp_candidates=cp_candidates,
+    action = "Would prune" if params.dry_run else "Pruned"
-        npm_candidates=npm_candidates,
+    print(f"{action} {total_pruned} report file(s) "
-        blocked_ips=len(final_blocked_ips),
+          f"(keep {REPORT_RETENTION_DAYS}d / "
-        blocked_targets=len(final_blocked_targets),
+          f"max {REPORT_RETENTION_MAX_FILES} per dir)")
-        blocked_ipv4=blocked_ipv4,
+
-        blocked_ipv6=blocked_ipv6,
+    # --- Step 9: persist consolidated results to SQLite ---------------------
-        blocklist_v4_path=blocklist_v4_path,
+    # Sync the blocklist ledger and close out the scan record regardless of
-        blocklist_v6_path=blocklist_v6_path,
+    # whether the router sync was attempted.
-        firewall_connected=firewall is not None,
+    _sync_blocklist_to_db(
-        firewall_changes=firewall_changes,
+        firewall is not None, blocked_ipv4, blocked_ipv6,
-        blocklist_report_path=blocklist_report_path,
+        added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}"))
-        errors=errors,
+    if db_scan_id is not None:
-    )
+        try:
-
+            from ..db import get_connection
-
+            from ..db import repositories as repo
-def _build_parser() -> argparse.ArgumentParser:
+            conn = get_connection()
-    parser = argparse.ArgumentParser(
+            result_payload = {
-        description="Consolidated IDS Scanner: run copyparty + NPM analyzers and generate unified blocklist."
+                "ok": True,
-    )
+                "cp_candidates": cp_candidates,
-
+                "npm_candidates": npm_candidates,
-    # copyparty arguments
+                "blocked_ips": len(final_blocked_ips),
-    parser.add_argument("--copyparty-log-path", default=COPYPARTY_LOG_PATH,
+                "blocked_targets": len(final_blocked_targets),
-                        help=f"copyparty log directory (default: {COPYPARTY_LOG_PATH})")
+                "firewall_connected": firewall is not None,
-    parser.add_argument("--copyparty-report-dir", default=COPYPARTY_REPORT_DIR,
+            }
-                        help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})")
+            repo.finish_scan(
-    parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED,
+                conn, db_scan_id, status="done",
-                        help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})")
+                result=result_payload, log_files=0,
-
+                request_count=(cp_analyzer and len(cp_analyzer.requests) or 0)
-    # NPM arguments
+                + (npm_analyzer and len(npm_analyzer.requests) or 0),
-    parser.add_argument("--npm-log-path", default=NPM_LOG_PATH,
+                error="; ".join(errors) or None,
-                        help=f"NPM log directory (default: {NPM_LOG_PATH})")
+            )
-    parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR,
+        except Exception as exc:
-                        help=f"NPM report directory (default: {NPM_REPORT_DIR})")
+            print(f"WARN: could not finalize scan in db ({exc})")
-    parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS,
+
-                        help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})")
+    return ScanResult(
-    parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO,
+        ok=True,
-                        help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})")
+        cp_exists=cp_exists,
-
+        npm_exists=npm_exists,
-    # Shared arguments
+        cp_candidates=cp_candidates,
-    parser.add_argument("--report-dir", default=REPORT_DIR,
+        npm_candidates=npm_candidates,
-                        help=f"consolidated reports directory (default: {REPORT_DIR})")
+        blocked_ips=len(final_blocked_ips),
-    parser.add_argument("--top", type=int, default=DEFAULT_TOP,
+        blocked_targets=len(final_blocked_targets),
-                        help=f"rows in the top-IP tables (default: {DEFAULT_TOP})")
+        blocked_ipv4=blocked_ipv4,
-    parser.add_argument("--quiet", action="store_true",
+        blocked_ipv6=blocked_ipv6,
-                        help="write files, suppress console reports")
+        blocklist_v4_path=blocklist_v4_path,
-    parser.add_argument("--full", action="store_true",
+        blocklist_v6_path=blocklist_v6_path,
-                        help="rescan all log files from the start (ignore incremental state)")
+        firewall_connected=firewall is not None,
-    parser.add_argument("--update-firewall", action="store_true",
+        firewall_changes=firewall_changes,
-                        help="synchronize the MikroTik Blocked list after scanning")
+        blocklist_report_path=blocklist_report_path,
-    parser.add_argument("--dry-run", action="store_true",
+        errors=errors,
-                        help="preview firewall changes without applying them")
+    )
-    return parser
+
-
+
-
+def _build_parser() -> argparse.ArgumentParser:
-def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(
-    args = _build_parser().parse_args(argv)
+        description="Consolidated IDS Scanner: run copyparty + NPM analyzers and generate unified blocklist."
-    params = ScanParams(
+    )
-        copyparty_log_path=args.copyparty_log_path,
+
-        copyparty_report_dir=args.copyparty_report_dir,
+    # copyparty arguments
-        min_denied=args.min_denied,
+    parser.add_argument("--copyparty-log-path", default=COPYPARTY_LOG_PATH,
-        npm_log_path=args.npm_log_path,
+                        help=f"copyparty log directory (default: {COPYPARTY_LOG_PATH})")
-        npm_report_dir=args.npm_report_dir,
+    parser.add_argument("--copyparty-report-dir", default=COPYPARTY_REPORT_DIR,
-        min_errors=args.min_errors,
+                        help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})")
-        error_ratio=args.error_ratio,
+    parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED,
-        report_dir=args.report_dir,
+                        help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})")
-        top=args.top,
+
-        quiet=args.quiet,
+    # NPM arguments
-        full=args.full,
+    parser.add_argument("--npm-log-path", default=NPM_LOG_PATH,
-        update_firewall=args.update_firewall,
+                        help=f"NPM log directory (default: {NPM_LOG_PATH})")
-        dry_run=args.dry_run,
+    parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR,
-    )
+                        help=f"NPM report directory (default: {NPM_REPORT_DIR})")
-    run_scan(params)
+    parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS,
-
+                        help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})")
-
+    parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO,
-if __name__ == "__main__":
+                        help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})")
-    main()
+
+    # Shared arguments
+    parser.add_argument("--report-dir", default=REPORT_DIR,
+                        help=f"consolidated reports directory (default: {REPORT_DIR})")
+    parser.add_argument("--top", type=int, default=DEFAULT_TOP,
+                        help=f"rows in the top-IP tables (default: {DEFAULT_TOP})")
+    parser.add_argument("--quiet", action="store_true",
+                        help="write files, suppress console reports")
+    parser.add_argument("--full", action="store_true",
+                        help="rescan all log files from the start (ignore incremental state)")
+    parser.add_argument("--update-firewall", action="store_true",
+                        help="synchronize the MikroTik Blocked list after scanning")
+    parser.add_argument("--dry-run", action="store_true",
+                        help="preview firewall changes without applying them")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = _build_parser().parse_args(argv)
+    params = ScanParams(
+        copyparty_log_path=args.copyparty_log_path,
+        copyparty_report_dir=args.copyparty_report_dir,
+        min_denied=args.min_denied,
+        npm_log_path=args.npm_log_path,
+        npm_report_dir=args.npm_report_dir,
+        min_errors=args.min_errors,
+        error_ratio=args.error_ratio,
+        report_dir=args.report_dir,
+        top=args.top,
+        quiet=args.quiet,
+        full=args.full,
+        update_firewall=args.update_firewall,
+        dry_run=args.dry_run,
+    )
+    run_scan(params)
+
+
+if __name__ == "__main__":
+    main()

app/templates/firewall.html

@@ -23,6 +23,11 @@ endblock %} {% block content %}
   <div class="panel-head"><h2>Firewall rules</h2></div>
   <div id="firewall-rules"></div>
 </div>
+
+<div class="panel">
+  <div class="panel-head"><h2>Snapshot history (SQLite)</h2></div>
+  <div id="fw-snapshots"></div>
+</div>
 {% endblock %} {% block scripts %}
 <script>
   async function loadState() {
@@ -35,6 +40,45 @@ endblock %} {% block content %}
     }
   }
 
+  async function loadSnapshots() {
+    const box = document.getElementById("fw-snapshots");
+    try {
+      const d = await api("/api/firewall/snapshots?limit=10");
+      const snaps = d.snapshots || [];
+      if (!snaps.length) {
+        box.innerHTML =
+          '<p class="muted">No SQLite snapshots yet — they are recorded on each <code>--update-firewall</code> run.</p>';
+        return;
+      }
+      box.innerHTML = "";
+      const rows = snaps.map((s) =>
+        el(
+          "tr",
+          el("td", { class: "mono", text: `#${s.id}` }),
+          el("td", { text: fmtTime(s.captured_at) }),
+          el("td", { text: s.kind }),
+          el("td", { text: s.trigger }),
+          el("td", { class: "num", text: s.addr_entries }),
+          el("td", { class: "num", text: s.rule_entries }),
+        ),
+      );
+      box.append(
+        el(
+          "table",
+          { class: "table" },
+          el(
+            "thead",
+            el("tr", {}, el("th", { text: "ID" }), el("th", { text: "Captured" }), el("th", { text: "Kind" }), el("th", { text: "Trigger" }), el("th", { text: "Addr" }), el("th", { text: "Rules" })),
+          ),
+          el("tbody", {}, ...rows),
+        ),
+      );
+    } catch (e) {
+      box.innerHTML=
+        '<p class="muted">Snapshot history unavailable (SQLite).</p>';
+    }
+  }
+
   function renderAddressLists(addr) {
     const box = document.getElementById("address-lists");
     box.innerHTML = "";
@@ -172,5 +216,6 @@ endblock %} {% block content %}
     .getElementById("apply-btn")
     .addEventListener("click", () => sync(false));
   loadState();
+  loadSnapshots();
 </script>
 {% endblock %}

app/templates/reports.html

@@ -1,76 +1,124 @@
-{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{%
+{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% endblock %} {% block content %}
-endblock %} {% block content %}
+<h1>Reports</h1>
-<h1>Reports</h1>
+<p class="muted">Browse the timestamped reports written by each scanner.</p>
-<p class="muted">Browse the timestamped reports written by each scanner.</p>
+
-
+<div class="panel">
-<div class="panel">
+  <div class="panel-actions" style="margin-top: 0">
-  <div class="panel-actions" style="margin-top: 0">
+    <label style="flex-direction: row; align-items: center; gap: 0.5rem"
-    <label style="flex-direction: row; align-items: center; gap: 0.5rem"
+      >Source
-      >Source
+      <select id="source">
-      <select id="source">
+        <option value="copyparty">copyparty</option>
-        <option value="copyparty">copyparty</option>
+        <option value="npm">npm</option>
-        <option value="npm">npm</option>
+        <option value="app">app</option>
-        <option value="app">app</option>
+        <option value="mikrotik">mikrotik</option>
-        <option value="mikrotik">mikrotik</option>
+      </select>
-      </select>
+    </label>
-    </label>
+    <button class="btn" id="refresh-btn">Refresh</button>
-    <button class="btn" id="refresh-btn">Refresh</button>
+  </div>
-  </div>
+  <div class="list" id="files"></div>
-  <div class="list" id="files"></div>
+</div>
-</div>
+
-
+<div class="panel">
-<div class="panel">
+  <div class="panel-head"><h2 id="content-title">Report content</h2></div>
-  <div class="panel-head"><h2 id="content-title">Report content</h2></div>
+  <pre class="content" id="content">Select a report on the left to view it.</pre>
-  <pre class="content" id="content">
+</div>
-Select a report on the left to view it.</pre
+{% endblock %} {% block scripts %}
-  >
+<script>
-</div>
+  let nextCursor = null;
-{% endblock %} {% block scripts %}
+
-<script>
+  function addReportLink(list, label, onClick) {
-  async function loadFiles() {
+    const link = document.createElement("a");
-    const source = document.getElementById("source").value;
+    link.href = "#";
-    const list = document.getElementById("files");
+    link.textContent = label;
-    list.innerHTML = '<div class="muted">Loading…</div>';
+    link.addEventListener("click", (event) => {
-    try {
+      event.preventDefault();
-      const data = await api(`/api/reports/${source}`);
+      onClick();
-      list.innerHTML = "";
+    });
-      const files = data.files || [];
+    list.append(link);
-      if (!files.length) {
+  }
-        list.innerHTML = '<div class="muted">No reports in this source.</div>';
+
-        return;
+  async function loadFiles(append = false) {
-      }
+    const source = document.getElementById("source").value;
-      for (const name of files) {
+    const list = document.getElementById("files");
-        list.append(
+    if (!append) {
-          el("a", {
+      nextCursor = null;
-            href: "#",
+      list.innerHTML = '<div class="muted">Loading…</div>';
-            text: name,
+    } else {
-            onclick: (ev) => {
+      document.getElementById("load-more")?.remove();
-              ev.preventDefault();
+    }
-              openReport(name);
+
-            },
+    let importedPaths = new Set();
-          }),
+    try {
-        );
+      const query = new URLSearchParams({ source, limit: "50" });
-      }
+      if (append && nextCursor) query.set("cursor", nextCursor);
-    } catch (e) {
+      const data = await api(`/api/reports?${query}`);
-      list.innerHTML = `<div class="muted">${escapeHtml(e.message)}</div>`;
+      if (!append) list.innerHTML = "";
-    }
+      for (const artifact of data.items || []) {
-  }
+        importedPaths.add(artifact.original_path);
-
+        addReportLink(list, `${artifact.generated_at} | ${artifact.report_type} | ${artifact.original_path}`, () =>
-  async function openReport(name) {
+          openArtifact(artifact.id),
-    const source = document.getElementById("source").value;
+        );
-    document.getElementById("content-title").textContent = name;
+      }
-    const pre = document.getElementById("content");
+      nextCursor = data.next_cursor;
-    pre.textContent = "Loading…";
+    } catch (e) {
-    try {
+      if (append) {
-      const data = await api(`/api/reports/${source}/${name}`);
+        list.append(el("div", { className: "muted", text: e.message }));
-      pre.textContent = data.content;
+        return;
-    } catch (e) {
+      }
-      pre.textContent = `Error: ${e.message}`;
+    }
-    }
+
-  }
+    if (!append) {
-
+      try {
-  document.getElementById("refresh-btn").addEventListener("click", loadFiles);
+        const data = await api(`/api/reports/${source}`);
-  document.getElementById("source").addEventListener("change", loadFiles);
+        for (const name of data.files || []) {
-  loadFiles();
+          if (importedPaths.has(`${source}/${name}`)) continue;
-</script>
+          addReportLink(list, `${name} | legacy file`, () => openReport(name));
-{% endblock %}
+        }
+      } catch (e) {
+        if (!list.childElementCount) list.append(el("div", { className: "muted", text: e.message }));
+      }
+    }
+
+    if (nextCursor) {
+      const button = document.createElement("button");
+      button.id = "load-more";
+      button.className = "btn";
+      button.textContent = "Load more";
+      button.addEventListener("click", () => loadFiles(true));
+      list.append(button);
+    }
+    if (!list.childElementCount) list.innerHTML = '<div class="muted">No reports in this source.</div>';
+  }
+
+  async function openArtifact(id) {
+    document.getElementById("content-title").textContent = "Database report";
+    const pre = document.getElementById("content");
+    pre.textContent = "Loading…";
+    try {
+      const data = await api(`/api/report-artifacts/${id}`);
+      document.getElementById("content-title").textContent = data.original_path;
+      const payload = data.payload || {};
+      pre.textContent = payload.format === "text" ? payload.content || "" : JSON.stringify(payload, null, 2);
+    } catch (e) {
+      pre.textContent = `Error: ${e.message}`;
+    }
+  }
+
+  async function openReport(name) {
+    const source = document.getElementById("source").value;
+    document.getElementById("content-title").textContent = name;
+    const pre = document.getElementById("content");
+    pre.textContent = "Loading…";
+    try {
+      const data = await api(`/api/reports/${source}/${name}`);
+      pre.textContent = data.content;
+    } catch (e) {
+      pre.textContent = `Error: ${e.message}`;
+    }
+  }
+
+  document.getElementById("refresh-btn").addEventListener("click", () => loadFiles());
+  document.getElementById("source").addEventListener("change", () => loadFiles());
+  loadFiles();
+</script>
+{% endblock %}

app/utils/blocklist_history.py

@@ -20,6 +20,7 @@ import re
 import subprocess
 from datetime import datetime
 from typing import Any, Iterable
+import hashlib
 
 from .files import write_json
 from .reporting import timestamp_name
@@ -151,6 +152,7 @@ def write_blocklist_report(report_dir: str,
                            firewall_changes: dict | None = None,
                            dry_run: bool = False,
                            errors: Iterable[str] = (),
+                           scan_id: int | None = None,
                            ) -> str:
     """Write the timestamped blocklist report to ``<report_dir>/app/``.
 
@@ -183,4 +185,18 @@ def write_blocklist_report(report_dir: str,
     os.makedirs(app_dir, exist_ok=True)
     path = os.path.join(app_dir, f"{timestamp_name()}.json")
     write_json(path, payload)
+    try:
+        from ..db import get_connection
+        from ..db import repositories as repo
+        with open(path, "rb") as file:
+            content = file.read()
+        repo.insert_report_artifact(
+            get_connection(), source="app", report_type="blocklist_change",
+            generated_at=payload["generated_at"], payload=payload,
+            original_path=f"app/{os.path.basename(path)}",
+            content_sha256=hashlib.sha256(content).hexdigest(),
+            scan_id=scan_id,
+        )
+    except Exception as exc:
+        print(f"WARN: blocklist report DB write skipped ({exc})")
     return path

app/web/routes.py

@@ -3,9 +3,13 @@
 
 from __future__ import annotations
 
+import csv
+import io
+import json
 import os
 
-from flask import Flask, abort, jsonify, render_template, request
+from flask import Flask, Response, abort, jsonify, render_template, request
+from werkzeug.utils import secure_filename
 
 from ..utils.constants import (
     APP_REPORT_DATA_DIR,
@@ -154,6 +158,84 @@ def register_routes(app: Flask,
 
     # --- reports -----------------------------------------------------------
 
+    @app.get("/api/reports")
+    def list_report_artifacts():
+        source = (request.args.get("source") or "").strip() or None
+        if source is not None and source not in report_sources:
+            return jsonify({"error": "unknown report source"}), 400
+        limit = request.args.get("limit", default=50, type=int)
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            result = repo.list_report_artifacts(
+                get_connection(), source=source,
+                report_type=(request.args.get("type") or "").strip() or None,
+                since=(request.args.get("since") or "").strip() or None,
+                until=(request.args.get("until") or "").strip() or None,
+                limit=limit,
+                cursor=(request.args.get("cursor") or "").strip() or None,
+            )
+            return jsonify(result)
+        except ValueError as exc:
+            return jsonify({"error": str(exc)}), 400
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+
+    @app.get("/api/report-artifacts/<int:artifact_id>")
+    def get_report_artifact(artifact_id: int):
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            artifact = repo.get_report_artifact(get_connection(), artifact_id)
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+        if artifact is None:
+            abort(404, "report artifact not found")
+        return jsonify(artifact)
+
+    @app.get("/api/report-artifacts/<int:artifact_id>/export")
+    def export_report_artifact(artifact_id: int):
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            artifact = repo.get_report_artifact(get_connection(), artifact_id)
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+        if artifact is None:
+            abort(404, "report artifact not found")
+
+        payload = artifact["payload"]
+        export_format = (request.args.get("format") or "json").lower()
+        filename = secure_filename(os.path.basename(
+            artifact["original_path"])) or f"report-{artifact_id}"
+        if export_format == "json":
+            content = json.dumps(payload, ensure_ascii=False, indent=2)
+            mimetype = "application/json"
+            extension = ".json"
+        elif export_format == "txt" and payload.get("format") == "text":
+            content = payload.get("content", "")
+            mimetype = "text/plain; charset=utf-8"
+            extension = ".txt"
+        elif export_format == "csv" and payload.get("format") == "csv":
+            output = io.StringIO(newline="")
+            writer = csv.DictWriter(
+                output, fieldnames=payload.get("columns", []),
+                extrasaction="ignore")
+            writer.writeheader()
+            writer.writerows(payload.get("rows", []))
+            content = output.getvalue()
+            mimetype = "text/csv; charset=utf-8"
+            extension = ".csv"
+        else:
+            return jsonify({"error": "unsupported export format"}), 406
+
+        download_name = (filename if filename.lower().endswith(extension)
+                         else f"{filename}{extension}")
+        response = Response(content, mimetype=mimetype)
+        response.headers["Content-Disposition"] = (
+            f'attachment; filename="{download_name}"')
+        return response
+
     @app.get("/api/reports/<source>")
     def list_reports(source):
         base = report_sources.get(source)
@@ -265,6 +347,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)

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
+    Merges three sources, newest-first:
-    disk reflect a scan newer than any known job (e.g. after a restart or a
+      1. the in-memory job store (currently running / last-run jobs),
-    CLI run), so the dashboard always shows the latest completed scan.
+      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 = (entries or [None])[0]
-        newest_ts = _iso_key(
+        newest_key = (
-            (newest.finished or newest.started) if newest else None)
+            (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).

tests/test_firewall_sync.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Tests for evidence-first firewall sync behavior."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from app.firewall import MikrotikFirewall
+
+
+class _FakeFirewall(MikrotikFirewall):
+    def __init__(self):
+        super().__init__(router_ip="0.0.0.0", username="x", password="x")
+        self.calls: list[str] = []
+
+    def get_address_list_ipv4(self, name: str) -> set[str]:
+        self.calls.append(f"v4:{name}")
+        if name == "Blocked":
+            return {"9.9.9.9/32"}
+        if name == "Init7":
+            return {"77.74.80.0/21"}
+        return set()
+
+    def get_address_list_ipv6(self, name: str) -> set[str]:
+        self.calls.append(f"v6:{name}")
+        if name == "Blocked":
+            return set()
+        if name == "Init7":
+            return {"2001:1620::/30"}
+        return set()
+
+
+class FirewallSyncEvidenceTest(unittest.TestCase):
+    def test_sync_loads_router_status_before_computing_changes(self):
+        fw = _FakeFirewall()
+        # type: ignore[assignment]
+        fw.save_state_report = lambda *args, **kwargs: "noop"
+
+        with patch("app.firewall.cumulative_blocklist",
+                   return_value=({"8.8.8.8/32"}, set())):
+            changes = fw.sync_blocklist(["8.8.8.8/32"], [], dry_run=True)
+
+        self.assertIn("v4:Blocked", fw.calls)
+        self.assertIn("v6:Blocked", fw.calls)
+        self.assertIn("v4:Init7", fw.calls)
+        self.assertIn("v6:Init7", fw.calls)
+        self.assertEqual(changes["added_ipv4"], ["8.8.8.8/32"])
+        self.assertEqual(changes["removed_ipv4"], ["9.9.9.9/32"])
+
+    def test_sync_evaluates_against_loaded_current_status(self):
+        fw = _FakeFirewall()
+        # type: ignore[assignment]
+        fw.save_state_report = lambda *args, **kwargs: "noop"
+
+        with patch("app.firewall.cumulative_blocklist",
+                   return_value=({"9.9.9.9/32"}, set())):
+            changes = fw.sync_blocklist(["9.9.9.9/32"], [], dry_run=True)
+
+        self.assertEqual(changes["added_ipv4"], [])
+        self.assertEqual(changes["removed_ipv4"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_backfill.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Tests for legacy report artifact backfill."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from app.db import connect
+from app.db.backfill import backfill_report_artifacts
+from app.db import repositories as repo
+
+
+class ReportBackfillTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp_dir.name)
+        self.conn = connect(str(self.root / "test.db"))
+        self.roots = {
+            "copyparty": str(self.root / "copyparty"),
+            "npm": str(self.root / "npm"),
+            "app": str(self.root / "app"),
+            "mikrotik": str(self.root / "mikrotik"),
+        }
+        for path in self.roots.values():
+            Path(path).mkdir(parents=True)
+
+    def tearDown(self):
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_dry_run_and_import_are_idempotent(self):
+        Path(self.roots["copyparty"], "2026-11-06-11-30-10.txt").write_text(
+            "copyparty summary\n", encoding="utf-8")
+        Path(self.roots["npm"], "2026-11-06.csv").write_text(
+            "ip,count\n192.0.2.1,3\n", encoding="utf-8")
+        Path(self.roots["app"], "2026-11-06.json").write_text(
+            '{"summary":{"blocked":1}}', encoding="utf-8")
+        Path(self.roots["app"], "data").mkdir()
+        Path(self.roots["app"], "data", "requests.csv").write_text(
+            "ip\n192.0.2.2\n", encoding="utf-8")
+
+        preview = backfill_report_artifacts(
+            self.conn, dry_run=True, roots=self.roots)["reports"]
+        self.assertEqual(preview["would_import"], 3)
+        self.assertEqual(preview["skipped_canonical"], [
+            {"source": "app", "path": "data/requests.csv"}])
+        self.assertEqual(preview["by_source"]["copyparty"]["would_import"], 1)
+        self.assertEqual(preview["by_format"]["csv"]["would_import"], 1)
+        self.assertEqual(repo.list_report_artifacts(self.conn)["items"], [])
+
+        imported = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(imported["imported"], 3)
+        self.assertEqual(imported["unmapped"], [])
+
+        repeated = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(repeated["imported"], 0)
+        self.assertEqual(repeated["already_imported"], 3)
+
+        items = repo.list_report_artifacts(
+            self.conn, source="copyparty")["items"]
+        self.assertEqual(items[0]["original_path"],
+                         "copyparty/2026-11-06-11-30-10.txt")
+        self.assertEqual(items[0]["generated_at"],
+                         "2026-11-06T11:30:10+00:00")
+
+    def test_malformed_json_is_preserved_with_parse_error(self):
+        Path(self.roots["app"], "2026-11-06.json").write_text(
+            '{"incomplete":', encoding="utf-8")
+
+        result = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(result["imported"], 1)
+        artifact = repo.list_report_artifacts(
+            self.conn, source="app")["items"][0]
+        detail = repo.get_report_artifact(self.conn, artifact["id"])
+        self.assertEqual(detail["payload"]["content"], '{"incomplete":')
+        self.assertIn("parse_error", detail["payload"])
+
+    def test_artifacts_survive_sqlite_backup_and_restore(self):
+        artifact_id = repo.insert_report_artifact(
+            self.conn, source="app", report_type="summary",
+            generated_at="2026-11-06T00:00:00+00:00",
+            payload={"summary": {"count": 4}},
+            original_path="app/2026-11-06.json", content_sha256="c" * 64,
+        )
+        restored = connect(str(self.root / "restored.db"))
+        try:
+            self.conn.backup(restored)
+            artifact = repo.get_report_artifact(restored, artifact_id)
+            self.assertEqual(artifact["payload"], {"summary": {"count": 4}})
+        finally:
+            restored.close()
+
+    def test_oversized_file_is_reported_without_deleting_source(self):
+        source = Path(self.roots["npm"], "large.txt")
+        source.write_text("keep", encoding="utf-8")
+        from unittest.mock import patch
+
+        with patch("app.db.backfill.MAX_REPORT_ARTIFACT_BYTES", 1):
+            result = backfill_report_artifacts(
+                self.conn, roots=self.roots)["reports"]
+
+        self.assertEqual(len(result["unmapped"]), 1)
+        self.assertTrue(source.exists())
+        self.assertEqual(result["unmapped"][0]["path"], "large.txt")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_persistence.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Tests for dual-writing generated reports to SQLite."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import app.scan
+from app.analyzer import Analyzer
+from app.db import connect
+from app.db import repositories as repo
+from app.utils.blocklist_history import write_blocklist_report
+
+
+class ReportPersistenceTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp_dir.name)
+        self.conn = connect(str(self.root / "test.db"))
+        self.connection_patch = patch("app.db.get_connection",
+                                      return_value=self.conn)
+        self.connection_patch.start()
+
+    def tearDown(self):
+        self.connection_patch.stop()
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_analyzer_summary_is_stored_with_file_provenance(self):
+        report_dir = self.root / "copyparty"
+        report_dir.mkdir()
+        path = report_dir / "2026-11-06-11-30-10.txt"
+        path.write_text("summary body", encoding="utf-8")
+        analyzer = Analyzer("logs", str(report_dir), quiet=True)
+
+        analyzer.persist_report_artifact(str(path), scan_id=14)
+
+        item = repo.list_report_artifacts(
+            self.conn, source="ids")["items"][0]
+        detail = repo.get_report_artifact(self.conn, item["id"])
+        self.assertEqual(item["original_path"], "ids/2026-11-06-11-30-10.txt")
+        self.assertEqual(detail["scan_id"], 14)
+        self.assertEqual(detail["payload"]["content"], "summary body")
+
+    def test_blocklist_report_is_stored_as_structured_artifact(self):
+        path = write_blocklist_report(
+            str(self.root / "reports"), history_v4=set(), history_v6=set(),
+            prev_v4=set(), prev_v6=set(),
+            complete_v4={"192.0.2.1/32"}, complete_v6=set(),
+            firewall_changes={"added_ipv4": ["192.0.2.1/32"]},
+            dry_run=True, scan_id=22,
+        )
+
+        items = repo.list_report_artifacts(
+            self.conn, source="app", report_type="blocklist_change")["items"]
+        self.assertEqual(len(items), 1)
+        detail = repo.get_report_artifact(self.conn, items[0]["id"])
+        self.assertEqual(detail["scan_id"], 22)
+        self.assertEqual(detail["payload"]["blocked_ipv4"], ["192.0.2.1/32"])
+        self.assertTrue(Path(path).exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_routes.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Tests for database-backed report API routes."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from app.db import connect
+from app.db import repositories as repo
+from app.web import create_app
+
+
+class ReportRoutesTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.conn = connect(str(Path(self.temp_dir.name) / "test.db"))
+        self.app = create_app()
+        self.app.config["TESTING"] = True
+        self.client = self.app.test_client()
+        self.connection_patch = patch("app.db.get_connection",
+                                      return_value=self.conn)
+        self.connection_patch.start()
+        self.artifact_id = repo.insert_report_artifact(
+            self.conn, source="copyparty", report_type="text_report",
+            generated_at="2026-11-06T10:00:00+00:00",
+            payload={"format": "text", "content": "summary"},
+            original_path="copyparty/2026-11-06.txt",
+            content_sha256="b" * 64,
+        )
+
+    def tearDown(self):
+        self.connection_patch.stop()
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_list_detail_and_text_export(self):
+        listed = self.client.get("/api/reports?source=copyparty&limit=1")
+        self.assertEqual(listed.status_code, 200)
+        self.assertEqual(listed.json["items"][0]["id"], self.artifact_id)
+
+        detail = self.client.get(
+            f"/api/report-artifacts/{self.artifact_id}")
+        self.assertEqual(detail.status_code, 200)
+        self.assertEqual(detail.json["payload"]["content"], "summary")
+
+        exported = self.client.get(
+            f"/api/report-artifacts/{self.artifact_id}/export?format=txt")
+        self.assertEqual(exported.status_code, 200)
+        self.assertEqual(exported.data.decode(), "summary")
+        self.assertIn("attachment", exported.headers["Content-Disposition"])
+        self.assertNotIn(".txt.txt", exported.headers["Content-Disposition"])
+
+    def test_invalid_source_cursor_and_missing_id(self):
+        self.assertEqual(
+            self.client.get("/api/reports?source=unknown").status_code, 400)
+        self.assertEqual(
+            self.client.get("/api/reports?cursor=invalid").status_code, 400)
+        self.assertEqual(
+            self.client.get("/api/report-artifacts/9999").status_code, 404)
+
+    def test_legacy_file_routes_remain_registered(self):
+        response = self.client.get("/api/reports/copyparty")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("files", response.json)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_repositories.py

@@ -0,0 +1,248 @@
+#!/usr/bin/env python3
+"""Tests for the SQLite repositories + schema migrations (T001).
+
+Run with:
+    python -m unittest discover -s tests -p 'test_*.py'
+or (if pytest is installed):
+    python -m pytest tests/
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from app.db import connect
+from app.db import repositories as repo
+from app.db.schema import SCHEMA_VERSION, apply_migrations
+
+
+def _new_conn() -> sqlite3.Connection:
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    os.unlink(path)
+    return connect(path)
+
+
+class SchemaTest(unittest.TestCase):
+    def test_migrations_apply_and_are_idempotent(self):
+        conn = _new_conn()
+        self.assertEqual(conn.execute("PRAGMA user_version").fetchone()[0],
+                         SCHEMA_VERSION)
+        # Re-applying should be a no-op (no exception, version unchanged).
+        apply_migrations(conn)
+        self.assertEqual(conn.execute("PRAGMA user_version").fetchone()[0],
+                         SCHEMA_VERSION)
+        conn.close()
+
+    def test_core_tables_exist(self):
+        conn = _new_conn()
+        tables = {r["name"] for r in conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table'")}
+        for t in ("requests", "ip_stats", "scans", "scans_sources",
+                  "blocklist_entries", "blocklist_events",
+                  "blocklist_reasons", "whitelist_entries",
+                  "whitelist_sources", "firewall_snapshots",
+                  "firewall_address_list_entries", "firewall_rule_entries",
+                  "whois_info", "log_sources", "report_artifacts"):
+            self.assertIn(t, tables, f"missing table {t}")
+        conn.close()
+
+
+class ScanAndRequestTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_scan_lifecycle(self):
+        sid = repo.start_scan(self.conn, "ids", params={"full": True})
+        self.assertEqual(repo.get_scan(self.conn, sid)["status"], "running")
+        repo.finish_scan(self.conn, sid, status="done",
+                         result={"ok": True}, request_count=3)
+        s = repo.get_scan(self.conn, sid)
+        self.assertEqual(s["status"], "done")
+        self.assertEqual(s["request_count"], 3)
+        self.assertEqual([x["id"] for x in repo.latest_scans(self.conn)],
+                         [sid])
+
+    def test_insert_and_query_requests(self):
+        sid = repo.start_scan(self.conn, "ids")
+        rows = [
+            {"source": "npm", "scan_id": sid, "log_file": "a.log",
+             "ts": "2026-09-01 10:00:00", "ip": "1.2.3.4", "version": 4,
+             "method": "GET", "path": "/.env", "status": 404,
+             "host": "h", "user": None, "ua": "curl", "referer": None,
+             "denied": False, "banned": False, "is_error": True, "extra": None},
+            {"source": "copyparty", "scan_id": sid, "log_file": "f.txt",
+             "ts": "2026-09-01 10:00:01", "ip": "1.2.3.4", "version": 4,
+             "method": "GET", "path": "/admin", "status": 403,
+             "host": None, "user": None, "ua": None, "referer": None,
+             "denied": True, "banned": False, "is_error": True, "extra": None},
+        ]
+        self.assertEqual(repo.insert_requests(self.conn, rows), 2)
+        self.assertEqual(repo.count_requests(self.conn), 2)
+        # cross-source
+        cs = repo.correlate_across_sources(self.conn)
+        self.assertEqual(cs[0]["ip"], "1.2.3.4")
+        self.assertEqual(cs[0]["sources"], 2)
+        # timeline
+        tl = repo.ip_timeline(self.conn, "1.2.3.4")
+        self.assertEqual(len(tl), 2)
+        # top ips
+        top = repo.top_ips(self.conn)
+        self.assertEqual(top[0]["total"], 2)
+        repo.finish_scan(self.conn, sid, status="done")
+
+
+class BlocklistTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_grow_only_record(self):
+        r1 = repo.record_blocklist(self.conn, ["5.5.5.5/32", "6.6.6.6/32"],
+                                   added_by="scan:1")
+        self.assertEqual(r1["active"], 2)
+        # a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative).
+        r2 = repo.record_blocklist(
+            self.conn, ["5.5.5.5/32"], added_by="scan:2")
+        active = [e["cidr"] for e in repo.active_blocklist(self.conn)]
+        self.assertIn("6.6.6.6/32", active)
+        self.assertEqual(r2["active"], 2)
+
+    def test_unblock(self):
+        repo.record_blocklist(self.conn, ["7.7.7.7/32"])
+        self.assertTrue(repo.unblock_entry(self.conn, "7.7.7.7/32"))
+        self.assertNotIn("7.7.7.7/32",
+                         [e["cidr"] for e in repo.active_blocklist(self.conn)])
+        self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32"))
+
+    def test_prune_old_single_ip_blocks_keeps_ranges(self):
+        old = (datetime.now(timezone.utc) - timedelta(days=31)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "1.2.3.4/32", last_seen=old)
+            repo._block_entry_inner(
+                self.conn, "2001:db8::1/128", last_seen=old)
+            repo._block_entry_inner(self.conn, "1.2.3.0/24", last_seen=old)
+            repo._block_entry_inner(self.conn, "2001:db8::/64", last_seen=old)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertCountEqual(removed, ["1.2.3.4/32", "2001:db8::1/128"])
+
+        active = {e["cidr"] for e in repo.active_blocklist(self.conn)}
+        self.assertIn("1.2.3.0/24", active)
+        self.assertIn("2001:db8::/64", active)
+        self.assertNotIn("1.2.3.4/32", active)
+        self.assertNotIn("2001:db8::1/128", active)
+
+    def test_prune_old_single_ip_blocks_respects_recency(self):
+        fresh = (datetime.now(timezone.utc) - timedelta(days=5)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "9.9.9.9/32", last_seen=fresh)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertEqual(removed, [])
+        self.assertIn("9.9.9.9/32",
+                      {e["cidr"] for e in repo.active_blocklist(self.conn)})
+
+
+class WhitelistTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_whitelist_lookup(self):
+        repo.set_whitelist(self.conn, "Cloudflare", ["173.245.48.0/20"],
+                           source="test")
+        self.assertTrue(repo.is_whitelisted_db(self.conn, "173.245.48.5"))
+        self.assertFalse(repo.is_whitelisted_db(self.conn, "8.8.8.8"))
+        # provider replace deactivates old ranges
+        repo.set_whitelist(self.conn, "Cloudflare", ["104.16.0.0/13"])
+        self.assertFalse(repo.is_whitelisted_db(self.conn, "173.245.48.5"))
+
+
+class FirewallTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_firewall_snapshot(self):
+        sid = repo.save_firewall_snapshot(
+            self.conn,
+            address_lists_v4=[{"list": "Blocked", "address": "5.5.5.5/32"}],
+            address_lists_v6=[{"list": "Init7", "address": "2001:1620::/30"}],
+            rules_v4=[{"chain": "input", "action": "drop", "comment": "x"}],
+            rules_v6=[])
+        self.assertEqual(sid, 1)
+        self.assertEqual(self.conn.execute(
+            "SELECT COUNT(*) FROM firewall_address_list_entries").fetchone()[0], 2)
+        self.assertEqual(self.conn.execute(
+            "SELECT COUNT(*) FROM firewall_rule_entries").fetchone()[0], 1)
+
+
+class ReportArtifactTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_artifact_insert_is_idempotent_and_detail_decodes_payload(self):
+        kwargs = {
+            "source": "copyparty", "report_type": "analyzer_summary",
+            "generated_at": "2026-11-01T10:00:00+00:00",
+            "payload": {"summary": {"requests": 12}},
+            "original_path": "reports/copyparty/2026-11-01.txt",
+            "content_sha256": "a" * 64,
+        }
+        first_id = repo.insert_report_artifact(self.conn, **kwargs)
+        duplicate_id = repo.insert_report_artifact(self.conn, **kwargs)
+
+        self.assertEqual(first_id, duplicate_id)
+        self.assertEqual(self.conn.execute(
+            "SELECT COUNT(*) FROM report_artifacts").fetchone()[0], 1)
+        self.assertEqual(repo.get_report_artifact(self.conn, first_id)["payload"],
+                         kwargs["payload"])
+
+    def test_artifact_cursor_orders_stably_and_filters_source(self):
+        for day, source in (("01", "npm"), ("02", "copyparty"),
+                            ("03", "npm")):
+            repo.insert_report_artifact(
+                self.conn, source=source, report_type="summary",
+                generated_at=f"2026-11-{day}T00:00:00+00:00",
+                payload={"day": day}, original_path=f"reports/{day}.txt",
+                content_sha256=day * 64,
+            )
+
+        first = repo.list_report_artifacts(self.conn, source="npm", limit=1)
+        self.assertEqual(len(first["items"]), 1)
+        self.assertIsNotNone(first["next_cursor"])
+        second = repo.list_report_artifacts(
+            self.conn, source="npm", limit=1, cursor=first["next_cursor"])
+        self.assertEqual(len(second["items"]), 1)
+        self.assertIsNone(second["next_cursor"])
+        self.assertGreater(first["items"][0]["generated_at"],
+                           second["items"][0]["generated_at"])
+
+    def test_artifact_cursor_rejects_invalid_token(self):
+        with self.assertRaises(ValueError):
+            repo.list_report_artifacts(self.conn, cursor="invalid")
+
+
+if __name__ == "__main__":
+    unittest.main()