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.

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/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

@@ -19,19 +19,30 @@ 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")
@@ -92,7 +103,7 @@ def backfill_firewall(conn, dry_run: bool = False) -> dict:
         "SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
     if n:
         return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
-                              "skipped": True}}
+                             "skipped": True}}
     if dry_run:
         return {"firewall": {"dry_run": True, "snapshot": None}}
     sid = repo.save_firewall_snapshot(
@@ -161,6 +172,152 @@ def backfill_requests(conn, dry_run: bool = False) -> dict:
     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, "", "-"):
@@ -181,14 +338,22 @@ def main(argv: list[str] | None = None) -> None:
                                 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 ==")
 
 

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."""
@@ -339,7 +445,7 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
     with transaction(conn):
         for cidr in to_add:
             _block_entry_inner(conn, cidr, added_by=added_by, last_seen=now,
-                              reasons=reasons)
+                               reasons=reasons)
         # Refresh last_seen on entries that are still desired so the ledger
         # reflects recency for existing rows too.
         conn.executemany(
@@ -348,6 +454,52 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
     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``.
@@ -402,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
 # ---------------------------------------------------------------------------

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

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

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

@@ -13,6 +13,7 @@ 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
@@ -46,7 +47,7 @@ class SchemaTest(unittest.TestCase):
                   "blocklist_reasons", "whitelist_entries",
                   "whitelist_sources", "firewall_snapshots",
                   "firewall_address_list_entries", "firewall_rule_entries",
-                  "whois_info", "log_sources"):
+                  "whois_info", "log_sources", "report_artifacts"):
             self.assertIn(t, tables, f"missing table {t}")
         conn.close()
 
@@ -110,7 +111,8 @@ class BlocklistTest(unittest.TestCase):
                                    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")
+        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)
@@ -122,6 +124,38 @@ class BlocklistTest(unittest.TestCase):
                          [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):
@@ -161,5 +195,54 @@ class FirewallTest(unittest.TestCase):
             "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()