Compare commits

..
6 Commits
Author SHA1 Message Date
zwitschi 6f2e479958 feat(reports): store and browse report artifacts 2026-11-06 12:27:00 +01:00
zwitschi 5967a43eb5 test(firewall): add tests for evidence-first firewall sync behavior 2026-11-06 12:09:26 +01:00
zwitschi a9a03c1347 docs: define report storage and API design 2026-11-06 12:09:09 +01:00
zwitschi 03f3c65a58 docs: inventory report artifacts and consumers 2026-11-06 12:08:05 +01:00
zwitschi 5ece4cd113 feat(prune): add functionality to remove old single-host IP blocks while keeping CIDR ranges intact 2026-11-06 11:33:12 +01:00
zwitschi 714aaf901e fix(scan): _db_complete_blocklist must return sets, not lists
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.
2026-10-24 20:00:59 +02:00
14 changed files with 1590 additions and 514 deletions

No files matched your search

+142
View File
@@ -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.
+27
View File
@@ -7,6 +7,7 @@
from __future__ import annotations from __future__ import annotations
import hashlib
import os import os
import re import re
from collections import Counter from collections import Counter
@@ -336,6 +337,7 @@ class Analyzer:
self.write_data() self.write_data()
report_path = self.write_report() report_path = self.write_report()
self.persist_report_artifact(report_path, db_scan_id)
print(f"data files: {self.data_dir}") print(f"data files: {self.data_dir}")
print(f"summary report: {report_path}") print(f"summary report: {report_path}")
@@ -349,3 +351,28 @@ class Analyzer:
result.update(rsc=rsc, blocked=n_blocked, result.update(rsc=rsc, blocked=n_blocked,
whitelisted=n_whitelisted) whitelisted=n_whitelisted)
return result 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})")
+166 -1
View File
@@ -19,19 +19,30 @@ from __future__ import annotations
import argparse import argparse
import csv import csv
import hashlib
import json import json
import os import os
import re
import sys import sys
from datetime import datetime, timezone from datetime import datetime, timezone
from ..utils.constants import ( from ..utils.constants import (
APP_REPORT_DIR,
APP_REPORT_DATA_DIR, APP_REPORT_DATA_DIR,
DATA_DIR, DATA_DIR,
MIKROTIK_DATA_DIR, MIKROTIK_DATA_DIR,
MIKROTIK_REPORT_DIR,
NPM_REPORT_DIR, NPM_REPORT_DIR,
COPYPARTY_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: def utcnow() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds") 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] "SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
if n: if n:
return {"firewall": {"dry_run": dry_run, "snapshot": "exists", return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
"skipped": True}} "skipped": True}}
if dry_run: if dry_run:
return {"firewall": {"dry_run": True, "snapshot": None}} return {"firewall": {"dry_run": True, "snapshot": None}}
sid = repo.save_firewall_snapshot( 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}} 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): def _int_or_none(v):
try: try:
if v in (None, "", "-"): if v in (None, "", "-"):
@@ -181,14 +338,22 @@ def main(argv: list[str] | None = None) -> None:
description="Backfill legacy runtime data into SQLite.") description="Backfill legacy runtime data into SQLite.")
p.add_argument("--dry-run", action="store_true", p.add_argument("--dry-run", action="store_true",
help="report what would be imported without writing") 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) opts = p.parse_args(argv)
from . import get_connection from . import get_connection
conn = get_connection() conn = get_connection()
print(f"== backfill (dry_run={opts.dry_run}) ==") 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("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run))
print("firewall :", backfill_firewall(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("requests :", backfill_requests(conn, dry_run=opts.dry_run))
print("reports :", backfill_report_artifacts(conn, dry_run=opts.dry_run))
print("== done ==") print("== done ==")
+175 -3
View File
@@ -10,8 +10,10 @@ from __future__ import annotations
import json import json
import sqlite3 import sqlite3
import base64
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime, timezone from datetime import datetime, timedelta, timezone
import ipaddress
from typing import Any, Iterator, Sequence from typing import Any, Iterator, Sequence
from ..utils.models import IPStats from ..utils.models import IPStats
@@ -104,7 +106,8 @@ def _parse_ts(ts: str) -> tuple[str, float | None]:
cleaned = ts.strip() cleaned = ts.strip()
# Accept a trailing timezone like +HH:MM or Z; fall back to naive parse. # Accept a trailing timezone like +HH:MM or Z; fall back to naive parse.
try: try:
epoch = datetime.fromisoformat(cleaned.replace("Z", "+00:00")).timestamp() epoch = datetime.fromisoformat(
cleaned.replace("Z", "+00:00")).timestamp()
except ValueError: except ValueError:
try: try:
epoch = datetime.fromisoformat(cleaned).timestamp() 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]) 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, def correlate_across_sources(conn: sqlite3.Connection,
limit: int = 100) -> list[dict[str, Any]]: limit: int = 100) -> list[dict[str, Any]]:
"""IPs seen in more than one log source, with per-source counts.""" """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): with transaction(conn):
for cidr in to_add: for cidr in to_add:
_block_entry_inner(conn, cidr, added_by=added_by, last_seen=now, _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 # Refresh last_seen on entries that are still desired so the ledger
# reflects recency for existing rows too. # reflects recency for existing rows too.
conn.executemany( 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)} 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], *, def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *,
added_by: str | None = None) -> dict[str, Any]: added_by: str | None = None) -> dict[str, Any]:
"""Reconcile active blocklist to ``desired_cidrs``. """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 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 # Whitelists
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+24 -8
View File
@@ -14,7 +14,7 @@ import sqlite3
from typing import Iterable from typing import Iterable
#: Latest schema version. Bump by one whenever a new migration is appended. #: 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 #: 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; #: 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, log_path TEXT,
enabled INTEGER NOT NULL DEFAULT 1 enabled INTEGER NOT NULL DEFAULT 1
) )
""",
""" """
INSERT OR IGNORE INTO log_sources (name, log_path) ],
VALUES ('copyparty', NULL), ('npm', NULL)
2: [
"""
CREATE TABLE IF NOT EXISTS report_artifacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
scan_id INTEGER,
report_type TEXT NOT NULL,
generated_at TEXT NOT NULL,
schema_version INTEGER NOT NULL DEFAULT 1,
payload_json TEXT NOT NULL,
original_path TEXT NOT NULL,
content_sha256 TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE (original_path, content_sha256)
)
""", """,
# ------------------------------------------------------------------ "CREATE INDEX IF NOT EXISTS idx_report_artifacts_source_time "
# requests — one row per parsed request; insert-only "ON report_artifacts (source, generated_at DESC, id DESC)",
# ------------------------------------------------------------------ "CREATE INDEX IF NOT EXISTS idx_report_artifacts_type_time "
"ON report_artifacts (report_type, generated_at DESC, id DESC)",
""" """
CREATE TABLE IF NOT EXISTS requests ( CREATE TABLE IF NOT EXISTS requests (
id INTEGER PRIMARY KEY AUTOINCREMENT, 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): for version in range(current + 1, SCHEMA_VERSION + 1):
statements = MIGRATIONS.get(version) statements = MIGRATIONS.get(version)
if statements is None: if statements is None:
raise RuntimeError(f"missing migration for schema version {version}") raise RuntimeError(
f"missing migration for schema version {version}")
with conn: with conn:
for ddl in statements: for ddl in statements:
conn.execute(ddl) conn.execute(ddl)
+431 -423
View File
@@ -1,423 +1,431 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Consolidated intrusion detection log scanner. """Consolidated intrusion detection log scanner.
Runs both copyparty and Nginx Proxy Manager (NPM) log analyzers, Runs both copyparty and Nginx Proxy Manager (NPM) log analyzers,
and outputs a single consolidated MikroTik RouterOS blocklist. and outputs a single consolidated MikroTik RouterOS blocklist.
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import os import os
from dataclasses import dataclass, field from dataclasses import dataclass, field
from ..utils.blocklist_history import ( from ..utils.blocklist_history import (
committed_lines, committed_lines,
cumulative_blocklist, cumulative_blocklist,
current_file_lines, current_file_lines,
write_blocklist_report, write_blocklist_report,
) )
from ..utils.constants import ( from ..utils.constants import (
COPYPARTY_LOG_PATH, COPYPARTY_LOG_PATH,
COPYPARTY_REPORT_DIR, COPYPARTY_REPORT_DIR,
DEFAULT_ERROR_RATIO, DEFAULT_ERROR_RATIO,
DEFAULT_MIN_DENIED, DEFAULT_MIN_DENIED,
DEFAULT_MIN_ERRORS, DEFAULT_MIN_ERRORS,
DEFAULT_TOP, DEFAULT_TOP,
NPM_LOG_PATH, NPM_LOG_PATH,
NPM_REPORT_DIR, NPM_REPORT_DIR,
REPORT_DIR, REPORT_DIR,
REPORT_RETENTION_DAYS, REPORT_RETENTION_DAYS,
REPORT_RETENTION_MAX_FILES, REPORT_RETENTION_MAX_FILES,
SUBNET_THRESHOLD, SUBNET_THRESHOLD,
) )
from ..firewall import MikrotikFirewall, write_blocklist_files from ..firewall import MikrotikFirewall, write_blocklist_files
from ..utils.retention import prune_report_dirs from ..utils.retention import prune_report_dirs
from ..copyparty import CopypartyAnalyzer from ..copyparty import CopypartyAnalyzer
from ..npm import NpmAnalyzer from ..npm import NpmAnalyzer
from ..utils.ip_utils import is_whitelisted from ..utils.ip_utils import is_whitelisted
from ..mikrotik import split_cidr_versions from ..mikrotik import split_cidr_versions
from ..utils.whois import consolidate_by_subnet from ..utils.whois import consolidate_by_subnet
def _sync_blocklist_to_db(connected: bool, desired_v4: list[str], def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
desired_v6: list[str], added_by: str | None = None) -> None: desired_v6: list[str],
"""Best-effort grow-only sync of block targets into the SQLite ledger. 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 Uses ``record_blocklist`` (add-only) so the ledger is the authoritative
leaves the scan window, matching the router's cumulative model. Returns the cumulative record: once an IP is ever blocked it stays active even after it
added delta or None if the DB is unavailable. 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 try:
from ..db import repositories as repo from ..db import get_connection
conn = get_connection() from ..db import repositories as repo
return repo.record_blocklist( conn = get_connection()
conn, list(desired_v4) + list(desired_v6), added_by=added_by) result = repo.record_blocklist(
except Exception as exc: # never break the scan for a DB issue conn, list(desired_v4) + list(desired_v6), added_by=added_by)
print(f"WARN: could not sync blocklist to db ({exc})") # Retention policy: expire stale single-host entries after 30 days
return None # while keeping CIDR ranges/subnets intact.
pruned = repo.prune_old_single_ip_blocks(conn, older_than_days=30)
result["pruned_single_ips"] = pruned
def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]): return result
"""Complete cumulative blocklist from the DB ledger (B001). except Exception as exc: # never break the scan for a DB issue
print(f"WARN: could not sync blocklist to db ({exc})")
Returns ``(complete_v4, complete_v6)`` derived from the active blocklist return None
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. def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]):
""" """Complete cumulative blocklist from the DB ledger (B001).
try:
from ..db import get_connection Returns ``(complete_v4, complete_v6)`` as **sets** (same contract as
from ..db import repositories as repo ``cumulative_blocklist``) derived from the active blocklist ledger plus
conn = get_connection() this scan's newly desired targets (in case the DB write for this scan has
active = [e["cidr"] for e in repo.active_blocklist(conn)] not run yet). Falls back to the git-history reconstruction when the DB is
except Exception: unavailable, so nothing regresses.
active = [] """
union = set(active) | set(desired_v4) | set(desired_v6) try:
v4, v6 = split_cidr_versions(union) from ..db import get_connection
return v4, v6 from ..db import repositories as repo
conn = get_connection()
active = [e["cidr"] for e in repo.active_blocklist(conn)]
@dataclass except Exception:
class ScanParams: active = []
"""Configuration for a consolidated scan. union = set(active) | set(desired_v4) | set(desired_v6)
v4, v6 = split_cidr_versions(union)
Shared by the CLI entry point and the web interface so both run the return set(v4), set(v6)
exact same pipeline.
"""
@dataclass
copyparty_log_path: str = COPYPARTY_LOG_PATH class ScanParams:
copyparty_report_dir: str = COPYPARTY_REPORT_DIR """Configuration for a consolidated scan.
min_denied: int = DEFAULT_MIN_DENIED
npm_log_path: str = NPM_LOG_PATH Shared by the CLI entry point and the web interface so both run the
npm_report_dir: str = NPM_REPORT_DIR exact same pipeline.
min_errors: int = DEFAULT_MIN_ERRORS """
error_ratio: float = DEFAULT_ERROR_RATIO
report_dir: str = REPORT_DIR copyparty_log_path: str = COPYPARTY_LOG_PATH
top: int = DEFAULT_TOP copyparty_report_dir: str = COPYPARTY_REPORT_DIR
quiet: bool = False min_denied: int = DEFAULT_MIN_DENIED
full: bool = False npm_log_path: str = NPM_LOG_PATH
update_firewall: bool = False npm_report_dir: str = NPM_REPORT_DIR
dry_run: bool = False min_errors: int = DEFAULT_MIN_ERRORS
error_ratio: float = DEFAULT_ERROR_RATIO
report_dir: str = REPORT_DIR
@dataclass top: int = DEFAULT_TOP
class ScanResult: quiet: bool = False
"""Structured outcome of a consolidated scan.""" full: bool = False
update_firewall: bool = False
ok: bool dry_run: bool = False
cp_exists: bool = False
npm_exists: bool = False
cp_candidates: int = 0 @dataclass
npm_candidates: int = 0 class ScanResult:
blocked_ips: int = 0 """Structured outcome of a consolidated scan."""
blocked_targets: int = 0
blocked_ipv4: list[str] = field(default_factory=list) ok: bool
blocked_ipv6: list[str] = field(default_factory=list) cp_exists: bool = False
blocklist_v4_path: str | None = None npm_exists: bool = False
blocklist_v6_path: str | None = None cp_candidates: int = 0
firewall_connected: bool = False npm_candidates: int = 0
firewall_changes: dict | None = None blocked_ips: int = 0
blocklist_report_path: str | None = None blocked_targets: int = 0
errors: list[str] = field(default_factory=list) blocked_ipv4: list[str] = field(default_factory=list)
blocked_ipv6: list[str] = field(default_factory=list)
blocklist_v4_path: str | None = None
def run_scan(params: ScanParams) -> ScanResult: blocklist_v6_path: str | None = None
"""Run the full consolidated scan pipeline and return structured results. firewall_connected: bool = False
firewall_changes: dict | None = None
Steps: connect to MikroTik (optional) -> snapshot state -> run copyparty blocklist_report_path: str | None = None
+ NPM analyzers -> consolidate candidates -> write blocklists -> compare errors: list[str] = field(default_factory=list)
and apply firewall changes (optional).
"""
errors: list[str] = [] def run_scan(params: ScanParams) -> ScanResult:
"""Run the full consolidated scan pipeline and return structured results.
# --- Step 0: open the SQLite store for this consolidated run -------------
db_scan_id: int | None = None Steps: connect to MikroTik (optional) -> snapshot state -> run copyparty
try: + NPM analyzers -> consolidate candidates -> write blocklists -> compare
from ..db import get_connection and apply firewall changes (optional).
from ..db import repositories as repo """
from ..utils.whitelists import sync_whitelists_to_db errors: list[str] = []
conn = get_connection()
sync_whitelists_to_db() # --- Step 0: open the SQLite store for this consolidated run -------------
params_map = { db_scan_id: int | None = None
"full": params.full, "dry_run": params.dry_run, try:
"update_firewall": params.update_firewall, from ..db import get_connection
"quiet": params.quiet, "top": params.top, from ..db import repositories as repo
} from ..utils.whitelists import sync_whitelists_to_db
db_scan_id = repo.start_scan(conn, "ids", params=params_map, conn = get_connection()
full=params.full, dry_run=params.dry_run) sync_whitelists_to_db()
except Exception as exc: params_map = {
print(f"WARN: sqlite store unavailable ({exc})") "full": params.full, "dry_run": params.dry_run,
"update_firewall": params.update_firewall,
# --- Step 1-3: connect to MikroTik RouterOS and capture current state --- "quiet": params.quiet, "top": params.top,
# If the router is unreachable we fall back to scan-only mode: still }
# generate reports and blocklists, but skip the firewall sync. db_scan_id = repo.start_scan(conn, "ids", params=params_map,
firewall: MikrotikFirewall | None = None full=params.full, dry_run=params.dry_run)
if params.update_firewall: except Exception as exc:
firewall = MikrotikFirewall() print(f"WARN: sqlite store unavailable ({exc})")
try:
firewall.connect() # --- Step 1-3: connect to MikroTik RouterOS and capture current state ---
# Read current firewall rules + address lists and snapshot them to # If the router is unreachable we fall back to scan-only mode: still
# reports/mikrotik/data/ and the git-diffable data/mikrotik/. # generate reports and blocklists, but skip the firewall sync.
firewall.save_state_report(params.report_dir) firewall: MikrotikFirewall | None = None
except Exception as e: if params.update_firewall:
message = f"could not read MikroTik router ({firewall.router_ip}): {e}" firewall = MikrotikFirewall()
errors.append(message) try:
print(f"\nWARNING: {message}") firewall.connect()
firewall.disconnect() # Read current firewall rules + address lists and snapshot them to
firewall = None # reports/mikrotik/data/ and the git-diffable data/mikrotik/.
firewall.save_state_report(params.report_dir)
# --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------ except Exception as e:
cp_analyzer = CopypartyAnalyzer( message = f"could not read MikroTik router ({firewall.router_ip}): {e}"
log_path=params.copyparty_log_path, errors.append(message)
report_dir=params.copyparty_report_dir, print(f"\nWARNING: {message}")
top=params.top, firewall.disconnect()
quiet=params.quiet, firewall = None
full=params.full,
min_denied=params.min_denied, # --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------
) cp_analyzer = CopypartyAnalyzer(
log_path=params.copyparty_log_path,
npm_analyzer = NpmAnalyzer( report_dir=params.copyparty_report_dir,
log_path=params.npm_log_path, top=params.top,
report_dir=params.npm_report_dir, quiet=params.quiet,
top=params.top, full=params.full,
quiet=params.quiet, min_denied=params.min_denied,
full=params.full, )
min_errors=params.min_errors,
error_ratio=params.error_ratio, npm_analyzer = NpmAnalyzer(
) log_path=params.npm_log_path,
report_dir=params.npm_report_dir,
print("=== Running copyparty scan ===") top=params.top,
cp_exists = True quiet=params.quiet,
try: full=params.full,
cp_analyzer.run(db_scan_id=db_scan_id) min_errors=params.min_errors,
except Exception as e: error_ratio=params.error_ratio,
errors.append(f"copyparty scan failed: {e}") )
print(f"Error running copyparty scan: {e}")
cp_exists = False print("=== Running copyparty scan ===")
cp_exists = True
print("\n=== Running NPM scan ===") try:
npm_exists = True cp_analyzer.run(db_scan_id=db_scan_id)
try: except Exception as e:
npm_analyzer.run(db_scan_id=db_scan_id) errors.append(f"copyparty scan failed: {e}")
except Exception as e: print(f"Error running copyparty scan: {e}")
errors.append(f"NPM scan failed: {e}") cp_exists = False
print(f"Error running NPM scan: {e}")
npm_exists = False print("\n=== Running NPM scan ===")
npm_exists = True
if not cp_exists and not npm_exists: try:
message = "Both scans failed. Unable to generate consolidated blocklist." npm_analyzer.run(db_scan_id=db_scan_id)
errors.append(message) except Exception as e:
print(f"\n{message}") errors.append(f"NPM scan failed: {e}")
if firewall is not None: print(f"Error running NPM scan: {e}")
firewall.disconnect() npm_exists = False
if db_scan_id is not None:
try: if not cp_exists and not npm_exists:
from ..db import get_connection message = "Both scans failed. Unable to generate consolidated blocklist."
from ..db import repositories as repo errors.append(message)
repo.finish_scan(get_connection(), db_scan_id, status="error", print(f"\n{message}")
error="; ".join(errors)) if firewall is not None:
except Exception: firewall.disconnect()
pass if db_scan_id is not None:
return ScanResult(ok=False, errors=errors) try:
from ..db import get_connection
# Consolidate block candidates from ..db import repositories as repo
consolidated_candidates: set[str] = set() repo.finish_scan(get_connection(), db_scan_id, status="error",
if cp_exists: error="; ".join(errors))
consolidated_candidates.update(cp_analyzer.block_candidates()) except Exception:
if npm_exists: pass
consolidated_candidates.update(npm_analyzer.block_candidates()) return ScanResult(ok=False, errors=errors)
# Exclude whitelisted IPs # Consolidate block candidates
final_blocked_ips = { consolidated_candidates: set[str] = set()
ip for ip in consolidated_candidates if not is_whitelisted(ip)} if cp_exists:
consolidated_candidates.update(cp_analyzer.block_candidates())
# Group into subnets via WHOIS if multiple malicious IPs in same subnet if npm_exists:
print("\nGrouping blocked IPs into subnets using WHOIS...") consolidated_candidates.update(npm_analyzer.block_candidates())
final_blocked_targets = consolidate_by_subnet(
final_blocked_ips, threshold=SUBNET_THRESHOLD) # Exclude whitelisted IPs
blocked_ipv4, blocked_ipv6 = split_cidr_versions(final_blocked_targets) final_blocked_ips = {
ip for ip in consolidated_candidates if not is_whitelisted(ip)}
cp_candidates = len(cp_analyzer.block_candidates()) if cp_exists else 0
npm_candidates = len(npm_analyzer.block_candidates()) if npm_exists else 0 # Group into subnets via WHOIS if multiple malicious IPs in same subnet
print("\nGrouping blocked IPs into subnets using WHOIS...")
print("\n=== Consolidated Summary ===") final_blocked_targets = consolidate_by_subnet(
print(f"copyparty block candidates: {cp_candidates}") final_blocked_ips, threshold=SUBNET_THRESHOLD)
print(f"NPM block candidates: {npm_candidates}") blocked_ipv4, blocked_ipv6 = split_cidr_versions(final_blocked_targets)
print(f"Total unified blocked IPs: {len(final_blocked_ips)}")
print(f"Total target blocks/subnets: {len(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
# --- Step 6: write consolidated blocklists to reports/app/data ----------
# Capture the previous blocklist state BEFORE the files are overwritten so print("\n=== Consolidated Summary ===")
# the change report can compute the complete-set delta. print(f"copyparty block candidates: {cp_candidates}")
app_data_dir = os.path.join(params.report_dir, "app", "data") print(f"NPM block candidates: {npm_candidates}")
blocklist_v4_path = os.path.join(app_data_dir, "blocklist_ipv4.txt") print(f"Total unified blocked IPs: {len(final_blocked_ips)}")
blocklist_v6_path = os.path.join(app_data_dir, "blocklist_ipv6.txt") print(f"Total target blocks/subnets: {len(final_blocked_targets)}")
prev_v4 = current_file_lines(blocklist_v4_path)
prev_v6 = current_file_lines(blocklist_v6_path) # --- Step 6: write consolidated blocklists to reports/app/data ----------
history_v4 = committed_lines(blocklist_v4_path) # Capture the previous blocklist state BEFORE the files are overwritten so
history_v6 = committed_lines(blocklist_v6_path) # the change report can compute the complete-set delta.
app_data_dir = os.path.join(params.report_dir, "app", "data")
write_blocklist_files(blocked_ipv4, blocked_ipv6, params.report_dir) blocklist_v4_path = os.path.join(app_data_dir, "blocklist_ipv4.txt")
print(f"blocklist files:\n {blocklist_v4_path}\n {blocklist_v6_path}") blocklist_v6_path = os.path.join(app_data_dir, "blocklist_ipv6.txt")
prev_v4 = current_file_lines(blocklist_v4_path)
# --- Steps 5+7: compare current state with candidates and apply --------- prev_v6 = current_file_lines(blocklist_v6_path)
firewall_changes: dict | None = None history_v4 = committed_lines(blocklist_v4_path)
blocklist_report_path: str | None = None history_v6 = committed_lines(blocklist_v6_path)
if firewall is not None:
try: write_blocklist_files(blocked_ipv4, blocked_ipv6, params.report_dir)
firewall_changes = firewall.sync_blocklist( print(f"blocklist files:\n {blocklist_v4_path}\n {blocklist_v6_path}")
blocked_ipv4, blocked_ipv6, dry_run=params.dry_run,
report_dir=params.report_dir) # --- Steps 5+7: compare current state with candidates and apply ---------
finally: firewall_changes: dict | None = None
firewall.disconnect() blocklist_report_path: str | None = None
action = "Would" if params.dry_run else "" if firewall is not None:
print(f"{action} add {len(firewall_changes['added_ipv4'])} IPv4 and " try:
f"{len(firewall_changes['added_ipv6'])} IPv6 addresses; " firewall_changes = firewall.sync_blocklist(
f"remove {len(firewall_changes['removed_ipv4'])} IPv4 and " blocked_ipv4, blocked_ipv6, dry_run=params.dry_run,
f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.") report_dir=params.report_dir)
finally:
# Timestamped change report — only when the router was reachable. firewall.disconnect()
# Source of truth for the cumulative set: the SQLite |ledger| first action = "Would" if params.dry_run else ""
# (B001), falling back to the git-history reconstruction only if the print(f"{action} add {len(firewall_changes['added_ipv4'])} IPv4 and "
# DB is unavailable. f"{len(firewall_changes['added_ipv6'])} IPv6 addresses; "
complete_v4, complete_v6 = _db_complete_blocklist( f"remove {len(firewall_changes['removed_ipv4'])} IPv4 and "
blocked_ipv4, blocked_ipv6) f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
if not complete_v4 and not complete_v6:
complete_v4, complete_v6 = cumulative_blocklist( # Timestamped change report — only when the router was reachable.
blocklist_v4_path, blocklist_v6_path, # Source of truth for the cumulative set: the SQLite |ledger| first
blocked_ipv4, blocked_ipv6) # (B001), falling back to the git-history reconstruction only if the
blocklist_report_path = write_blocklist_report( # DB is unavailable.
params.report_dir, complete_v4, complete_v6 = _db_complete_blocklist(
history_v4=history_v4, history_v6=history_v6, blocked_ipv4, blocked_ipv6)
prev_v4=prev_v4, prev_v6=prev_v6, if not complete_v4 and not complete_v6:
complete_v4=complete_v4, complete_v6=complete_v6, complete_v4, complete_v6 = cumulative_blocklist(
firewall_changes=firewall_changes, blocklist_v4_path, blocklist_v6_path,
dry_run=params.dry_run, blocked_ipv4, blocked_ipv6)
errors=errors) blocklist_report_path = write_blocklist_report(
print(f"blocklist report: {blocklist_report_path}") params.report_dir,
history_v4=history_v4, history_v6=history_v6,
# --- Step 8: prune old reports (retention) ------------------------------ prev_v4=prev_v4, prev_v6=prev_v6,
# Each report subdirectory keeps at most REPORT_RETENTION_MAX_FILES files, complete_v4=complete_v4, complete_v6=complete_v6,
# all newer than REPORT_RETENTION_DAYS days. The blocklist files in firewall_changes=firewall_changes,
# ``app/data`` are always protected. ``--dry-run`` only previews. dry_run=params.dry_run,
pruned = prune_report_dirs( errors=errors,
params.report_dir, params.copyparty_report_dir, params.npm_report_dir, scan_id=db_scan_id)
dry_run=params.dry_run) print(f"blocklist report: {blocklist_report_path}")
total_pruned = sum(len(paths) for paths in pruned.values())
action = "Would prune" if params.dry_run else "Pruned" # --- Step 8: prune old reports (retention) ------------------------------
print(f"{action} {total_pruned} report file(s) " # Each report subdirectory keeps at most REPORT_RETENTION_MAX_FILES files,
f"(keep {REPORT_RETENTION_DAYS}d / " # all newer than REPORT_RETENTION_DAYS days. The blocklist files in
f"max {REPORT_RETENTION_MAX_FILES} per dir)") # ``app/data`` are always protected. ``--dry-run`` only previews.
pruned = prune_report_dirs(
# --- Step 9: persist consolidated results to SQLite --------------------- params.report_dir, params.copyparty_report_dir, params.npm_report_dir,
# Sync the blocklist ledger and close out the scan record regardless of dry_run=params.dry_run)
# whether the router sync was attempted. total_pruned = sum(len(paths) for paths in pruned.values())
_sync_blocklist_to_db( action = "Would prune" if params.dry_run else "Pruned"
firewall is not None, blocked_ipv4, blocked_ipv6, print(f"{action} {total_pruned} report file(s) "
added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}")) f"(keep {REPORT_RETENTION_DAYS}d / "
if db_scan_id is not None: f"max {REPORT_RETENTION_MAX_FILES} per dir)")
try:
from ..db import get_connection # --- Step 9: persist consolidated results to SQLite ---------------------
from ..db import repositories as repo # Sync the blocklist ledger and close out the scan record regardless of
conn = get_connection() # whether the router sync was attempted.
result_payload = { _sync_blocklist_to_db(
"ok": True, firewall is not None, blocked_ipv4, blocked_ipv6,
"cp_candidates": cp_candidates, added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}"))
"npm_candidates": npm_candidates, if db_scan_id is not None:
"blocked_ips": len(final_blocked_ips), try:
"blocked_targets": len(final_blocked_targets), from ..db import get_connection
"firewall_connected": firewall is not None, from ..db import repositories as repo
} conn = get_connection()
repo.finish_scan( result_payload = {
conn, db_scan_id, status="done", "ok": True,
result=result_payload, log_files=0, "cp_candidates": cp_candidates,
request_count=(cp_analyzer and len(cp_analyzer.requests) or 0) "npm_candidates": npm_candidates,
+ (npm_analyzer and len(npm_analyzer.requests) or 0), "blocked_ips": len(final_blocked_ips),
error="; ".join(errors) or None, "blocked_targets": len(final_blocked_targets),
) "firewall_connected": firewall is not None,
except Exception as exc: }
print(f"WARN: could not finalize scan in db ({exc})") repo.finish_scan(
conn, db_scan_id, status="done",
return ScanResult( result=result_payload, log_files=0,
ok=True, request_count=(cp_analyzer and len(cp_analyzer.requests) or 0)
cp_exists=cp_exists, + (npm_analyzer and len(npm_analyzer.requests) or 0),
npm_exists=npm_exists, error="; ".join(errors) or None,
cp_candidates=cp_candidates, )
npm_candidates=npm_candidates, except Exception as exc:
blocked_ips=len(final_blocked_ips), print(f"WARN: could not finalize scan in db ({exc})")
blocked_targets=len(final_blocked_targets),
blocked_ipv4=blocked_ipv4, return ScanResult(
blocked_ipv6=blocked_ipv6, ok=True,
blocklist_v4_path=blocklist_v4_path, cp_exists=cp_exists,
blocklist_v6_path=blocklist_v6_path, npm_exists=npm_exists,
firewall_connected=firewall is not None, cp_candidates=cp_candidates,
firewall_changes=firewall_changes, npm_candidates=npm_candidates,
blocklist_report_path=blocklist_report_path, blocked_ips=len(final_blocked_ips),
errors=errors, blocked_targets=len(final_blocked_targets),
) blocked_ipv4=blocked_ipv4,
blocked_ipv6=blocked_ipv6,
blocklist_v4_path=blocklist_v4_path,
def _build_parser() -> argparse.ArgumentParser: blocklist_v6_path=blocklist_v6_path,
parser = argparse.ArgumentParser( firewall_connected=firewall is not None,
description="Consolidated IDS Scanner: run copyparty + NPM analyzers and generate unified blocklist." firewall_changes=firewall_changes,
) blocklist_report_path=blocklist_report_path,
errors=errors,
# 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, def _build_parser() -> argparse.ArgumentParser:
help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})") parser = argparse.ArgumentParser(
parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED, description="Consolidated IDS Scanner: run copyparty + NPM analyzers and generate unified blocklist."
help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})") )
# NPM arguments # copyparty arguments
parser.add_argument("--npm-log-path", default=NPM_LOG_PATH, parser.add_argument("--copyparty-log-path", default=COPYPARTY_LOG_PATH,
help=f"NPM log directory (default: {NPM_LOG_PATH})") help=f"copyparty log directory (default: {COPYPARTY_LOG_PATH})")
parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR, parser.add_argument("--copyparty-report-dir", default=COPYPARTY_REPORT_DIR,
help=f"NPM report directory (default: {NPM_REPORT_DIR})") help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})")
parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS, parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED,
help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})") help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})")
parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO,
help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})") # NPM arguments
parser.add_argument("--npm-log-path", default=NPM_LOG_PATH,
# Shared arguments help=f"NPM log directory (default: {NPM_LOG_PATH})")
parser.add_argument("--report-dir", default=REPORT_DIR, parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR,
help=f"consolidated reports directory (default: {REPORT_DIR})") help=f"NPM report directory (default: {NPM_REPORT_DIR})")
parser.add_argument("--top", type=int, default=DEFAULT_TOP, parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS,
help=f"rows in the top-IP tables (default: {DEFAULT_TOP})") help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})")
parser.add_argument("--quiet", action="store_true", parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO,
help="write files, suppress console reports") help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})")
parser.add_argument("--full", action="store_true",
help="rescan all log files from the start (ignore incremental state)") # Shared arguments
parser.add_argument("--update-firewall", action="store_true", parser.add_argument("--report-dir", default=REPORT_DIR,
help="synchronize the MikroTik Blocked list after scanning") help=f"consolidated reports directory (default: {REPORT_DIR})")
parser.add_argument("--dry-run", action="store_true", parser.add_argument("--top", type=int, default=DEFAULT_TOP,
help="preview firewall changes without applying them") help=f"rows in the top-IP tables (default: {DEFAULT_TOP})")
return parser parser.add_argument("--quiet", action="store_true",
help="write files, suppress console reports")
parser.add_argument("--full", action="store_true",
def main(argv: list[str] | None = None) -> None: help="rescan all log files from the start (ignore incremental state)")
args = _build_parser().parse_args(argv) parser.add_argument("--update-firewall", action="store_true",
params = ScanParams( help="synchronize the MikroTik Blocked list after scanning")
copyparty_log_path=args.copyparty_log_path, parser.add_argument("--dry-run", action="store_true",
copyparty_report_dir=args.copyparty_report_dir, help="preview firewall changes without applying them")
min_denied=args.min_denied, return parser
npm_log_path=args.npm_log_path,
npm_report_dir=args.npm_report_dir,
min_errors=args.min_errors, def main(argv: list[str] | None = None) -> None:
error_ratio=args.error_ratio, args = _build_parser().parse_args(argv)
report_dir=args.report_dir, params = ScanParams(
top=args.top, copyparty_log_path=args.copyparty_log_path,
quiet=args.quiet, copyparty_report_dir=args.copyparty_report_dir,
full=args.full, min_denied=args.min_denied,
update_firewall=args.update_firewall, npm_log_path=args.npm_log_path,
dry_run=args.dry_run, npm_report_dir=args.npm_report_dir,
) min_errors=args.min_errors,
run_scan(params) error_ratio=args.error_ratio,
report_dir=args.report_dir,
top=args.top,
if __name__ == "__main__": quiet=args.quiet,
main() full=args.full,
update_firewall=args.update_firewall,
dry_run=args.dry_run,
)
run_scan(params)
if __name__ == "__main__":
main()
+124 -76
View File
@@ -1,76 +1,124 @@
{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% {% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% endblock %} {% block content %}
endblock %} {% block content %} <h1>Reports</h1>
<h1>Reports</h1> <p class="muted">Browse the timestamped reports written by each scanner.</p>
<p class="muted">Browse the timestamped reports written by each scanner.</p>
<div class="panel">
<div class="panel"> <div class="panel-actions" style="margin-top: 0">
<div class="panel-actions" style="margin-top: 0"> <label style="flex-direction: row; align-items: center; gap: 0.5rem"
<label style="flex-direction: row; align-items: center; gap: 0.5rem" >Source
>Source <select id="source">
<select id="source"> <option value="copyparty">copyparty</option>
<option value="copyparty">copyparty</option> <option value="npm">npm</option>
<option value="npm">npm</option> <option value="app">app</option>
<option value="app">app</option> <option value="mikrotik">mikrotik</option>
<option value="mikrotik">mikrotik</option> </select>
</select> </label>
</label> <button class="btn" id="refresh-btn">Refresh</button>
<button class="btn" id="refresh-btn">Refresh</button> </div>
</div> <div class="list" id="files"></div>
<div class="list" id="files"></div> </div>
</div>
<div class="panel">
<div class="panel"> <div class="panel-head"><h2 id="content-title">Report content</h2></div>
<div class="panel-head"><h2 id="content-title">Report content</h2></div> <pre class="content" id="content">Select a report on the left to view it.</pre>
<pre class="content" id="content"> </div>
Select a report on the left to view it.</pre {% endblock %} {% block scripts %}
> <script>
</div> let nextCursor = null;
{% endblock %} {% block scripts %}
<script> function addReportLink(list, label, onClick) {
async function loadFiles() { const link = document.createElement("a");
const source = document.getElementById("source").value; link.href = "#";
const list = document.getElementById("files"); link.textContent = label;
list.innerHTML = '<div class="muted">Loading…</div>'; link.addEventListener("click", (event) => {
try { event.preventDefault();
const data = await api(`/api/reports/${source}`); onClick();
list.innerHTML = ""; });
const files = data.files || []; list.append(link);
if (!files.length) { }
list.innerHTML = '<div class="muted">No reports in this source.</div>';
return; async function loadFiles(append = false) {
} const source = document.getElementById("source").value;
for (const name of files) { const list = document.getElementById("files");
list.append( if (!append) {
el("a", { nextCursor = null;
href: "#", list.innerHTML = '<div class="muted">Loading…</div>';
text: name, } else {
onclick: (ev) => { document.getElementById("load-more")?.remove();
ev.preventDefault(); }
openReport(name);
}, let importedPaths = new Set();
}), try {
); const query = new URLSearchParams({ source, limit: "50" });
} if (append && nextCursor) query.set("cursor", nextCursor);
} catch (e) { const data = await api(`/api/reports?${query}`);
list.innerHTML = `<div class="muted">${escapeHtml(e.message)}</div>`; if (!append) list.innerHTML = "";
} for (const artifact of data.items || []) {
} importedPaths.add(artifact.original_path);
addReportLink(list, `${artifact.generated_at} | ${artifact.report_type} | ${artifact.original_path}`, () =>
async function openReport(name) { openArtifact(artifact.id),
const source = document.getElementById("source").value; );
document.getElementById("content-title").textContent = name; }
const pre = document.getElementById("content"); nextCursor = data.next_cursor;
pre.textContent = "Loading…"; } catch (e) {
try { if (append) {
const data = await api(`/api/reports/${source}/${name}`); list.append(el("div", { className: "muted", text: e.message }));
pre.textContent = data.content; return;
} catch (e) { }
pre.textContent = `Error: ${e.message}`; }
}
} if (!append) {
try {
document.getElementById("refresh-btn").addEventListener("click", loadFiles); const data = await api(`/api/reports/${source}`);
document.getElementById("source").addEventListener("change", loadFiles); for (const name of data.files || []) {
loadFiles(); if (importedPaths.has(`${source}/${name}`)) continue;
</script> addReportLink(list, `${name} | legacy file`, () => openReport(name));
{% endblock %} }
} catch (e) {
if (!list.childElementCount) list.append(el("div", { className: "muted", text: e.message }));
}
}
if (nextCursor) {
const button = document.createElement("button");
button.id = "load-more";
button.className = "btn";
button.textContent = "Load more";
button.addEventListener("click", () => loadFiles(true));
list.append(button);
}
if (!list.childElementCount) list.innerHTML = '<div class="muted">No reports in this source.</div>';
}
async function openArtifact(id) {
document.getElementById("content-title").textContent = "Database report";
const pre = document.getElementById("content");
pre.textContent = "Loading…";
try {
const data = await api(`/api/report-artifacts/${id}`);
document.getElementById("content-title").textContent = data.original_path;
const payload = data.payload || {};
pre.textContent = payload.format === "text" ? payload.content || "" : JSON.stringify(payload, null, 2);
} catch (e) {
pre.textContent = `Error: ${e.message}`;
}
}
async function openReport(name) {
const source = document.getElementById("source").value;
document.getElementById("content-title").textContent = name;
const pre = document.getElementById("content");
pre.textContent = "Loading…";
try {
const data = await api(`/api/reports/${source}/${name}`);
pre.textContent = data.content;
} catch (e) {
pre.textContent = `Error: ${e.message}`;
}
}
document.getElementById("refresh-btn").addEventListener("click", () => loadFiles());
document.getElementById("source").addEventListener("change", () => loadFiles());
loadFiles();
</script>
{% endblock %}
+16
View File
@@ -20,6 +20,7 @@ import re
import subprocess import subprocess
from datetime import datetime from datetime import datetime
from typing import Any, Iterable from typing import Any, Iterable
import hashlib
from .files import write_json from .files import write_json
from .reporting import timestamp_name from .reporting import timestamp_name
@@ -151,6 +152,7 @@ def write_blocklist_report(report_dir: str,
firewall_changes: dict | None = None, firewall_changes: dict | None = None,
dry_run: bool = False, dry_run: bool = False,
errors: Iterable[str] = (), errors: Iterable[str] = (),
scan_id: int | None = None,
) -> str: ) -> str:
"""Write the timestamped blocklist report to ``<report_dir>/app/``. """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) os.makedirs(app_dir, exist_ok=True)
path = os.path.join(app_dir, f"{timestamp_name()}.json") path = os.path.join(app_dir, f"{timestamp_name()}.json")
write_json(path, payload) 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 return path
+83 -1
View File
@@ -3,9 +3,13 @@
from __future__ import annotations from __future__ import annotations
import csv
import io
import json
import os 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 ( from ..utils.constants import (
APP_REPORT_DATA_DIR, APP_REPORT_DATA_DIR,
@@ -154,6 +158,84 @@ def register_routes(app: Flask,
# --- reports ----------------------------------------------------------- # --- 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>") @app.get("/api/reports/<source>")
def list_reports(source): def list_reports(source):
base = report_sources.get(source) base = report_sources.get(source)
+65
View File
@@ -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()
+114
View File
@@ -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()
+67
View File
@@ -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()
+71
View File
@@ -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()
+85 -2
View File
@@ -13,6 +13,7 @@ import os
import sqlite3 import sqlite3
import tempfile import tempfile
import unittest import unittest
from datetime import datetime, timedelta, timezone
from app.db import connect from app.db import connect
from app.db import repositories as repo from app.db import repositories as repo
@@ -46,7 +47,7 @@ class SchemaTest(unittest.TestCase):
"blocklist_reasons", "whitelist_entries", "blocklist_reasons", "whitelist_entries",
"whitelist_sources", "firewall_snapshots", "whitelist_sources", "firewall_snapshots",
"firewall_address_list_entries", "firewall_rule_entries", "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}") self.assertIn(t, tables, f"missing table {t}")
conn.close() conn.close()
@@ -110,7 +111,8 @@ class BlocklistTest(unittest.TestCase):
added_by="scan:1") added_by="scan:1")
self.assertEqual(r1["active"], 2) self.assertEqual(r1["active"], 2)
# a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative). # 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)] active = [e["cidr"] for e in repo.active_blocklist(self.conn)]
self.assertIn("6.6.6.6/32", active) self.assertIn("6.6.6.6/32", active)
self.assertEqual(r2["active"], 2) self.assertEqual(r2["active"], 2)
@@ -122,6 +124,38 @@ class BlocklistTest(unittest.TestCase):
[e["cidr"] for e in repo.active_blocklist(self.conn)]) [e["cidr"] for e in repo.active_blocklist(self.conn)])
self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32")) 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): class WhitelistTest(unittest.TestCase):
def setUp(self): def setUp(self):
@@ -161,5 +195,54 @@ class FirewallTest(unittest.TestCase):
"SELECT COUNT(*) FROM firewall_rule_entries").fetchone()[0], 1) "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__": if __name__ == "__main__":
unittest.main() unittest.main()