From 6f2e47995859c0f6b16e8702664ae452cc6aa558 Mon Sep 17 00:00:00 2001 From: Zwitschi Date: Fri, 6 Nov 2026 12:27:00 +0100 Subject: [PATCH] feat(reports): store and browse report artifacts --- REPORTS.md | 32 ++++- app/analyzer.py | 27 +++++ app/db/backfill.py | 167 +++++++++++++++++++++++++- app/db/repositories.py | 104 ++++++++++++++++ app/db/schema.py | 32 +++-- app/scan/__init__.py | 3 +- app/templates/reports.html | 200 +++++++++++++++++++------------ app/utils/blocklist_history.py | 16 +++ app/web/routes.py | 84 ++++++++++++- tests/test_report_backfill.py | 114 ++++++++++++++++++ tests/test_report_persistence.py | 67 +++++++++++ tests/test_report_routes.py | 71 +++++++++++ tests/test_repositories.py | 51 +++++++- 13 files changed, 875 insertions(+), 93 deletions(-) create mode 100644 tests/test_report_backfill.py create mode 100644 tests/test_report_persistence.py create mode 100644 tests/test_report_routes.py diff --git a/REPORTS.md b/REPORTS.md index be36d17..ce53a05 100644 --- a/REPORTS.md +++ b/REPORTS.md @@ -91,11 +91,10 @@ using opaque text as the only source of truth. `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/reports/` for the structured payload and - provenance metadata. Keep it separate from the existing - `GET /api/reports//` route to avoid changing that route's - meaning during migration. -- Add an explicit export route for supported report types and formats +- Add `GET /api/report-artifacts/` for the structured payload and + provenance metadata. Use a distinct path because `/api/reports/` + already owns the one-segment route shape. +- Add `GET /api/report-artifacts//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. @@ -118,3 +117,26 @@ 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. diff --git a/app/analyzer.py b/app/analyzer.py index 907db42..ecd43f0 100644 --- a/app/analyzer.py +++ b/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})") diff --git a/app/db/backfill.py b/app/db/backfill.py index 61bdad7..2b214f8 100644 --- a/app/db/backfill.py +++ b/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\d{4}-\d{2}-\d{2})(?:[-_T](?P\d{2})[-_:]" + r"(?P\d{2})[-_:](?P\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 ==") diff --git a/app/db/repositories.py b/app/db/repositories.py index 21375f2..a12551f 100644 --- a/app/db/repositories.py +++ b/app/db/repositories.py @@ -10,6 +10,7 @@ from __future__ import annotations import json import sqlite3 +import base64 from contextlib import contextmanager from datetime import datetime, timedelta, timezone import ipaddress @@ -233,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 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.""" diff --git a/app/db/schema.py b/app/db/schema.py index e3a5f5f..2b472f2 100644 --- a/app/db/schema.py +++ b/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) diff --git a/app/scan/__init__.py b/app/scan/__init__.py index d8b022c..aa50f24 100644 --- a/app/scan/__init__.py +++ b/app/scan/__init__.py @@ -303,7 +303,8 @@ def run_scan(params: ScanParams) -> ScanResult: complete_v4=complete_v4, complete_v6=complete_v6, firewall_changes=firewall_changes, dry_run=params.dry_run, - errors=errors) + errors=errors, + scan_id=db_scan_id) print(f"blocklist report: {blocklist_report_path}") # --- Step 8: prune old reports (retention) ------------------------------ diff --git a/app/templates/reports.html b/app/templates/reports.html index 791ea67..f445201 100644 --- a/app/templates/reports.html +++ b/app/templates/reports.html @@ -1,76 +1,124 @@ -{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% -endblock %} {% block content %} -

Reports

-

Browse the timestamped reports written by each scanner.

- -
-
- - -
-
-
- -
-

Report content

- 
-Select a report on the left to view it.
-
-{% endblock %} {% block scripts %} - -{% endblock %} +{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% endblock %} {% block content %} +

Reports

+

Browse the timestamped reports written by each scanner.

+ +
+
+ + +
+
+
+ +
+

Report content

+ 
Select a report on the left to view it.
+
+{% endblock %} {% block scripts %} + +{% endblock %} diff --git a/app/utils/blocklist_history.py b/app/utils/blocklist_history.py index 88a99da..576fc26 100644 --- a/app/utils/blocklist_history.py +++ b/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 ``/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 diff --git a/app/web/routes.py b/app/web/routes.py index e12df06..9606d2e 100644 --- a/app/web/routes.py +++ b/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/") + 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//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/") def list_reports(source): base = report_sources.get(source) diff --git a/tests/test_report_backfill.py b/tests/test_report_backfill.py new file mode 100644 index 0000000..68e36cf --- /dev/null +++ b/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() diff --git a/tests/test_report_persistence.py b/tests/test_report_persistence.py new file mode 100644 index 0000000..3a0b79d --- /dev/null +++ b/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() diff --git a/tests/test_report_routes.py b/tests/test_report_routes.py new file mode 100644 index 0000000..e879c21 --- /dev/null +++ b/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() diff --git a/tests/test_repositories.py b/tests/test_repositories.py index e4dc7a9..86443ee 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -47,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() @@ -195,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()