Compare commits

..
4 Commits
Author SHA1 Message Date
zwitschi 6051d2b548 test(db): repository + migration tests (T001); document analyze in README
tests/test_repositories.py covers schema migrations (idempotent), scan
lifecycle, request insert/query/cross-source, grow-only blocklist ledger,
unblock, whitelist lookup/replace, and firewall snapshots. 10 tests total.
README usage documents the 'python app.py analyze' command.
2026-10-24 19:56:22 +02:00
zwitschi 2fb4032766 feat(db): backfill, DB-backed web scans, firewall snapshot view (R001/S001/F001)
- R001: app/db/backfill.py imports git-history blocklist -> ledger, firewall
  snapshots, and per-source requests.csv (idempotent; ran live migrating 705
  historical blocklist entries).
- S001: web_scan.latest_scans() now merges SQLite 'scans' records so the
  dashboard surfaces CLI/restart scans (_db_scans).
- F001: /api/firewall/snapshots + 'Snapshot history' panel; added
  recent_firewall_snapshots() query with entry counts.
2026-10-24 19:56:20 +02:00
zwitschi ba48054293 feat(blocklist): make SQLite ledger authoritative for the cumulative set (B001)
- Add repo.record_blocklist(): grow-only, never auto-removes (mirrors router
  cumulative model; preserves entries that leave the NPM log window).
- scan.py uses it on the scan path and derives the report's complete set from
  the ledger (_db_complete_blocklist), falling back to git history only if the
  DB is unavailable.
2026-10-24 19:56:17 +02:00
zwitschi 4a30f6b582 feat: add 'python app.py analyze' CLI report (L002)
Reads the persisted SQLite store to print a detailed analysis: latest scans,
top IPs, cross-source IPs, active blocklist ledger, and a per-IP request
history drill-down (--ip). Supports --since, --top, --json. No scan is run.
2026-10-24 19:56:15 +02:00
10 changed files with 710 additions and 13 deletions

No files matched your search

+5
View File
@@ -172,6 +172,11 @@ python app.py scan --quiet
python app.py scan --update-firewall --dry-run
python app.py scan --update-firewall
# detailed log-analysis report from the SQLite store (needs scans run first)
python app.py analyze # top IPs + cross-source + blocklist
python app.py analyze --ip 1.2.3.4 # drill into one IP's request history
python app.py analyze --json # machine-readable output
# individual scanners (still runnable directly)
python -m app.copyparty
python -m app.npm
+10
View File
@@ -40,12 +40,22 @@ def _run_scan(argv: list[str]) -> None:
scan_main(argv)
def _run_analyze(argv: list[str]) -> None:
from app.analyze import main as analyze_main
analyze_main(argv)
def main() -> None:
# `python app.py scan ...` runs the consolidated CLI scan.
if len(sys.argv) > 1 and sys.argv[1] == "scan":
_run_scan(sys.argv[2:])
return
# `python app.py analyze ...` prints a detailed report from the SQLite store.
if len(sys.argv) > 1 and sys.argv[1] == "analyze":
_run_analyze(sys.argv[2:])
return
# Default: start the web UI (parse_known_args tolerates stray flags).
args, _ = _web_parser().parse_known_args()
_run_web(args.web_host, args.web_port)
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python3
"""CLI detailed log-analysis report (L002).
``python app.py analyze [--ip IP] [--since ISO] [--top N] [--json]`` reads the
persisted SQLite store (request history, per-IP aggregates, blocklist ledger)
and prints a detailed, interactive analysis. It complements the web UI's
``/analysis`` page for use on a headless server or in scripts.
No scan is run here: this only queries what ``scan`` has already persisted.
"""
from __future__ import annotations
import argparse
import json
import sys
from datetime import datetime
def _fmt_ts(epoch: float | None, default: str = "-") -> str:
if not epoch:
return default
try:
return datetime.fromtimestamp(epoch).isoformat(timespec="seconds")
except (ValueError, OverflowError, OSError):
return default
def _build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="python app.py analyze",
description="Detailed log-analysis report from the SQLite store.")
p.add_argument("--ip", metavar="IP", default=None,
help="focus the report on a single IP (drill-down)")
p.add_argument("--since", metavar="ISO", default=None,
help="only analyse requests at/after this ISO timestamp")
p.add_argument("--top", type=int, default=20,
help="rows in the top-IP / threat tables (default: 20)")
p.add_argument("--json", action="store_true",
help="emit raw JSON instead of a human-readable report")
return p
def _report(conn, opts) -> dict:
from .db import repositories as repo
data: dict = {}
data["scans"] = repo.latest_scans(conn, limit=10)
data["totals"] = {
"requests": repo.count_requests(conn),
"scans": repo.count_scans(conn),
}
if opts.ip:
data["ip"] = {
"ip": opts.ip,
"timeline": repo.ip_timeline(conn, ip=opts.ip),
"active_blocklist": [e for e in repo.active_blocklist(conn)
if e["cidr"].startswith(opts.ip)],
}
else:
data["top_ips"] = repo.top_ips(conn, since=opts.since, limit=opts.top)
data["cross_source"] = repo.correlate_across_sources(
conn, limit=opts.top)
data["blocklist"] = repo.active_blocklist(conn)
data["recent_errors"] = repo.query_requests(
conn, since=opts.since, limit=opts.top)
return data
def _render_text(data: dict, opts) -> str:
lines = ["Detailed log analysis", "=" * 40, ""]
totals = data["totals"]
lines.append(f"SQLite store: {totals['requests']} requests across "
f"{totals['scans']} scan(s)")
if data["scans"]:
lines.append("Latest scans:")
for s in data["scans"]:
lines.append(f" #{s['id']:<4} {s['kind']:<8} {s['status']:<7} "
f"{s.get('started') or s.get('started_at') or '-'}")
lines.append("")
if "ip" in data:
ip = data["ip"]
lines.append(f"== Request history for {ip['ip']} ==")
bl = ip["active_blocklist"]
if bl:
lines.append(f" (source: active blocklist entries "
f"{', '.join(e['cidr'] for e in bl)})")
lines.append(f"{'time':<24} {'src':<9} {'method':<8} {'status':<6} path")
for r in ip["timeline"][:opts.top]:
lines.append(
f"{r.get('ts') or '-':<24} {str(r.get('source')):<9} "
f"{str(r.get('method') or '-'):<8} "
f"{str(r.get('status') or '-'):<6} {r.get('path') or '-'}")
else:
lines.append("== Top IPs ==")
lines.append(
f"{'ip':<18} {'ver':<3} {'reqs':<6} {'err':<5} {'mal':<5} "
f"{'first':<22} last")
for r in data["top_ips"]:
lines.append(
f"{str(r['ip']):<18} {r.get('version') or '-':<3} "
f"{r['total']:<6} {r['errors']:<5} {r['malicious_hits']:<5} "
f"{_fmt_ts(r['first_epoch']):<22} {_fmt_ts(r['last_epoch'])}")
lines.append("")
lines.append("== Cross-source IPs ==")
if data["cross_source"]:
for r in data["cross_source"]:
lines.append(
f" {r['ip']:<18} sources={r['sources']} "
f"({r['source_list']}) total={r['total']}")
else:
lines.append(" (none — no IP seen in more than one log source)")
lines.append("")
lines.append("== Active blocklist (ledger) ==")
if data["blocklist"]:
for e in data["blocklist"]:
lines.append(f" {e['cidr']:<22} added={e['added_at']}")
else:
lines.append(" (empty)")
return "\n".join(lines)
def main(argv: list[str] | None = None) -> None:
opts = _build_parser().parse_args(argv)
try:
from .db import get_connection
conn = get_connection()
data = _report(conn, opts)
except Exception as exc:
print(f"error: could not read SQLite store ({exc}); "
f"run a scan first so data exists.", file=sys.stderr)
sys.exit(1)
if opts.json:
print(json.dumps(data, indent=2, sort_keys=True, default=str))
else:
print(_render_text(data, opts))
if __name__ == "__main__":
main()
+196
View File
@@ -0,0 +1,196 @@
#!/usr/bin/env python3
"""R001 — one-time backfill of legacy runtime data into the SQLite store.
Imports existing JSON/CSV/git-history data so the DB becomes the authoritative
source without losing anything already accumulated on a deployed instance:
- cumulative blocklist from the git-history reconstruction (old source of
truth) -> ``blocklist_entries`` (grow-only ledger)
- firewall snapshots from ``data/mikrotik/*.json`` -> ``firewall_snapshots``
- per-source requests from ``reports/<source>/data/requests.csv`` (best effort)
Idempotent: skips rows already present. Safe to run repeatedly.
Usage:
python -m app.db.backfill [--dry-run]
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import sys
from datetime import datetime, timezone
from ..utils.constants import (
APP_REPORT_DATA_DIR,
DATA_DIR,
MIKROTIK_DATA_DIR,
NPM_REPORT_DIR,
COPYPARTY_REPORT_DIR,
)
def utcnow() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def _read_cidr_lines(path: str) -> set[str]:
try:
with open(path, "r", encoding="utf-8") as f:
return {line.strip() for line in f if line.strip()}
except OSError:
return set()
def backfill_blocklist(conn, dry_run: bool = False) -> dict:
"""Import the legacy cumulative blocklist into the ledger."""
from . import repositories as repo
from ..utils.blocklist_history import (
committed_lines, current_file_lines,
)
v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")
v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")
# Legacy source of truth: union of all git revisions + working file +
# (there is no earlier scan target to add here).
complete_v4 = committed_lines(v4_path) | current_file_lines(v4_path)
complete_v6 = committed_lines(v6_path) | current_file_lines(v6_path)
cidrs = sorted(complete_v4 | complete_v6)
if not cidrs:
return {"blocklist": {"cidrs": 0, "added": 0}}
if dry_run:
return {"blocklist": {"cidrs": len(cidrs), "added": None, "dry_run": True}}
result = repo.record_blocklist(conn, cidrs, added_by="backfill")
return {"blocklist": result}
def backfill_firewall(conn, dry_run: bool = False) -> dict:
"""Import the latest firewall snapshot from ``data/mikrotik/*.json``."""
from . import repositories as repo
def _load(name: str) -> list:
try:
with open(os.path.join(MIKROTIK_DATA_DIR, name), "r",
encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, list) else []
except (OSError, ValueError):
return []
addr_v4 = _load("address_list_ipv4.json")
addr_v6 = _load("address_list_ipv6.json")
fw_v4 = _load("firewall_ipv4.json")
fw_v6 = _load("firewall_ipv6.json")
if not (addr_v4 or addr_v6 or fw_v4 or fw_v6):
return {"firewall": {"dry_run": dry_run, "snapshot": None}}
# Idempotent: if a firewall snapshot already exists (previous backfill),
# don't create another copy of the same latest state.
n = conn.execute(
"SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
if n:
return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
"skipped": True}}
if dry_run:
return {"firewall": {"dry_run": True, "snapshot": None}}
sid = repo.save_firewall_snapshot(
conn, address_lists_v4=addr_v4, address_lists_v6=addr_v6,
rules_v4=fw_v4, rules_v6=fw_v6)
return {"firewall": {"snapshot": sid}}
def backfill_requests(conn, dry_run: bool = False) -> dict:
"""Best-effort import of per-source ``requests.csv`` files.
Only ingests rows that don't already exist (keyed on source+log_file+ts+ip).
Legacy CSV columns vary by source; we keep a conservative mapping and never
fail the whole backfill over a single bad file.
"""
from . import repositories as repo
seen = set()
for r in conn.execute(
"SELECT source, log_file, ts, ip FROM requests"):
seen.add((r["source"], r["log_file"], r["ts"], r["ip"]))
candidates = [
(os.path.join(COPYPARTY_REPORT_DIR, "data", "requests.csv"), "copyparty"),
(os.path.join(NPM_REPORT_DIR, "data", "requests.csv"), "npm"),
]
total = 0
for path, source in candidates:
if not os.path.exists(path):
continue
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
reader = csv.DictReader(f)
rows = []
for row in reader:
ip = (row.get("ip") or "").strip()
ts = row.get("ts") or row.get("timestamp") or ""
log_file = row.get("log_file") or row.get("file") or ""
if not ip:
continue
key = (source, log_file, ts, ip)
if key in seen:
continue
seen.add(key)
rows.append({
"source": source, "scan_id": None, "log_file": log_file,
"ts": ts, "ip": ip,
"version": 4 if ":" not in ip else 6,
"method": row.get("method"), "path": row.get("path"),
"status": _int_or_none(row.get("status")),
"upstream_status": _int_or_none(
row.get("upstream_status")),
"host": row.get("host"), "user": row.get("user"),
"ua": row.get("ua") or row.get("user_agent"),
"referer": row.get("referer"),
"denied": _bool_str(row.get("denied")),
"banned": _bool_str(row.get("banned")),
"is_error": _bool_str(row.get("is_error")),
"extra": None,
})
total += 1
if not dry_run and total:
repo.insert_requests(conn, rows)
except Exception as exc:
print(f" WARN: requests.csv backfill {source}: {exc}")
return {"requests": {"imported": total, "dry_run": dry_run}}
def _int_or_none(v):
try:
if v in (None, "", "-"):
return None
return int(v)
except (TypeError, ValueError):
return None
def _bool_str(v):
if isinstance(v, str):
return int(v.strip().lower() in ("1", "true", "yes", "t"))
return int(bool(v))
def main(argv: list[str] | None = None) -> None:
p = argparse.ArgumentParser(prog="python -m app.db.backfill",
description="Backfill legacy runtime data into SQLite.")
p.add_argument("--dry-run", action="store_true",
help="report what would be imported without writing")
opts = p.parse_args(argv)
from . import get_connection
conn = get_connection()
print(f"== backfill (dry_run={opts.dry_run}) ==")
print("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run))
print("firewall :", backfill_firewall(conn, dry_run=opts.dry_run))
print("requests :", backfill_requests(conn, dry_run=opts.dry_run))
print("== done ==")
if __name__ == "__main__":
sys.exit(main())
+40
View File
@@ -322,6 +322,32 @@ def unblock_entry(conn: sqlite3.Connection, cidr: str, *,
return removed
def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
added_by: str | None = None, reasons=None) -> dict[str, Any]:
"""Grow-only record: add any ``cidrs`` not already active to the ledger.
Unlike :func:`sync_blocklist`, this **never removes** entries. It mirrors
the router's cumulative model: once an IP/subnet has ever been blocked it
stays active even after it leaves the scan window (e.g. NPM ~30-day log
retention), exactly like the old git-history backfill. Returns the added
delta and the resulting active count.
"""
cidr_set = set(cidrs)
before = {r["cidr"] for r in active_blocklist(conn)}
to_add = sorted(cidr_set - before)
now = utcnow()
with transaction(conn):
for cidr in to_add:
_block_entry_inner(conn, cidr, added_by=added_by, last_seen=now,
reasons=reasons)
# Refresh last_seen on entries that are still desired so the ledger
# reflects recency for existing rows too.
conn.executemany(
"UPDATE blocklist_entries SET last_seen=? WHERE cidr=? AND status='active'",
[(now, cidr) for cidr in cidr_set.intersection(before)])
return {"added": to_add, "active": len(before) + len(to_add)}
def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *,
added_by: str | None = None) -> dict[str, Any]:
"""Reconcile active blocklist to ``desired_cidrs``.
@@ -465,6 +491,20 @@ def save_firewall_snapshot(conn: sqlite3.Connection, *,
return snapshot_id
def recent_firewall_snapshots(conn: sqlite3.Connection,
limit: int = 10) -> list[dict[str, Any]]:
"""Most recent firewall snapshots with entry counts (F001)."""
rows = conn.execute(
"SELECT fs.id, fs.captured_at, fs.kind, fs.trigger, "
"(SELECT COUNT(*) FROM firewall_address_list_entries e "
" WHERE e.snapshot_id = fs.id) AS addr_entries, "
"(SELECT COUNT(*) FROM firewall_rule_entries r "
" WHERE r.snapshot_id = fs.id) AS rule_entries "
"FROM firewall_snapshots fs ORDER BY fs.id DESC LIMIT ?",
(limit,)).fetchall()
return [dict(r) for r in rows]
# ---------------------------------------------------------------------------
# WHOIS cache
# ---------------------------------------------------------------------------
+37 -7
View File
@@ -42,18 +42,42 @@ from ..utils.whois import consolidate_by_subnet
def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
desired_v6: list[str], added_by: str | None = None) -> None:
"""Best-effort sync of the consolidated blocklist targets into SQLite.
"""Best-effort grow-only sync of block targets into the SQLite ledger.
Keep the DB ledger in sync regardless of whether a router sync happened so
analysis always sees the current policy."""
Uses ``record_blocklist`` (add-only) so the ledger is the authoritative
cumulative record: once an IP is ever blocked it stays active even after it
leaves the scan window, matching the router's cumulative model. Returns the
added delta or None if the DB is unavailable.
"""
try:
from ..db import get_connection
from ..db import repositories as repo
conn = get_connection()
repo.sync_blocklist(conn, list(desired_v4) + list(desired_v6),
added_by=added_by)
return repo.record_blocklist(
conn, list(desired_v4) + list(desired_v6), added_by=added_by)
except Exception as exc: # never break the scan for a DB issue
print(f"WARN: could not sync blocklist to db ({exc})")
return None
def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]):
"""Complete cumulative blocklist from the DB ledger (B001).
Returns ``(complete_v4, complete_v6)`` derived from the active blocklist
ledger plus this scan's newly desired targets (in case the DB write for
this scan has not run yet). Falls back to the git-history reconstruction
when the DB is unavailable, so nothing regresses.
"""
try:
from ..db import get_connection
from ..db import repositories as repo
conn = get_connection()
active = [e["cidr"] for e in repo.active_blocklist(conn)]
except Exception:
active = []
union = set(active) | set(desired_v4) | set(desired_v6)
v4, v6 = split_cidr_versions(union)
return v4, v6
@dataclass
@@ -256,9 +280,15 @@ def run_scan(params: ScanParams) -> ScanResult:
f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
# Timestamped change report — only when the router was reachable.
complete_v4, complete_v6 = cumulative_blocklist(
blocklist_v4_path, blocklist_v6_path,
# Source of truth for the cumulative set: the SQLite |ledger| first
# (B001), falling back to the git-history reconstruction only if the
# DB is unavailable.
complete_v4, complete_v6 = _db_complete_blocklist(
blocked_ipv4, blocked_ipv6)
if not complete_v4 and not complete_v6:
complete_v4, complete_v6 = cumulative_blocklist(
blocklist_v4_path, blocklist_v6_path,
blocked_ipv4, blocked_ipv6)
blocklist_report_path = write_blocklist_report(
params.report_dir,
history_v4=history_v4, history_v6=history_v6,
+45
View File
@@ -23,6 +23,11 @@ endblock %} {% block content %}
<div class="panel-head"><h2>Firewall rules</h2></div>
<div id="firewall-rules"></div>
</div>
<div class="panel">
<div class="panel-head"><h2>Snapshot history (SQLite)</h2></div>
<div id="fw-snapshots"></div>
</div>
{% endblock %} {% block scripts %}
<script>
async function loadState() {
@@ -35,6 +40,45 @@ endblock %} {% block content %}
}
}
async function loadSnapshots() {
const box = document.getElementById("fw-snapshots");
try {
const d = await api("/api/firewall/snapshots?limit=10");
const snaps = d.snapshots || [];
if (!snaps.length) {
box.innerHTML =
'<p class="muted">No SQLite snapshots yet — they are recorded on each <code>--update-firewall</code> run.</p>';
return;
}
box.innerHTML = "";
const rows = snaps.map((s) =>
el(
"tr",
el("td", { class: "mono", text: `#${s.id}` }),
el("td", { text: fmtTime(s.captured_at) }),
el("td", { text: s.kind }),
el("td", { text: s.trigger }),
el("td", { class: "num", text: s.addr_entries }),
el("td", { class: "num", text: s.rule_entries }),
),
);
box.append(
el(
"table",
{ class: "table" },
el(
"thead",
el("tr", {}, el("th", { text: "ID" }), el("th", { text: "Captured" }), el("th", { text: "Kind" }), el("th", { text: "Trigger" }), el("th", { text: "Addr" }), el("th", { text: "Rules" })),
),
el("tbody", {}, ...rows),
),
);
} catch (e) {
box.innerHTML=
'<p class="muted">Snapshot history unavailable (SQLite).</p>';
}
}
function renderAddressLists(addr) {
const box = document.getElementById("address-lists");
box.innerHTML = "";
@@ -172,5 +216,6 @@ endblock %} {% block content %}
.getElementById("apply-btn")
.addEventListener("click", () => sync(false));
loadState();
loadSnapshots();
</script>
{% endblock %}
+13
View File
@@ -265,6 +265,19 @@ def register_routes(app: Flask,
"snapshot_dir": MIKROTIK_DATA_DIR,
})
@app.get("/api/firewall/snapshots")
def firewall_snapshots():
"""Recent firewall snapshots persisted in the SQLite store (F001)."""
try:
from ..db import get_connection
from ..db import repositories as repo
conn = get_connection()
limit = request.args.get("limit", 10, type=int)
return jsonify({"snapshots": repo.recent_firewall_snapshots(
conn, limit=limit)})
except Exception as exc:
return jsonify({"error": str(exc), "snapshots": []}), 503
@app.post("/api/firewall/sync")
def firewall_sync():
return _sync_firewall(dry_run=True)
+54 -6
View File
@@ -258,23 +258,71 @@ def latest_file_scan() -> dict[str, Any] | None:
}
def _db_scans(limit: int = 10) -> list[dict[str, Any]]:
"""Recent scan records from the SQLite ``scans`` table (S001).
Converts DB rows into the web job-dict shape so the dashboard can surface
scans that were persisted by ``run_scan`` / analyzers even if they were
never run through the in-memory job store (e.g. CLI scans or restarts).
Returns [] when the DB is unavailable.
"""
try:
from ..db import get_connection
from ..db import repositories as repo
conn = get_connection()
rows = repo.latest_scans(conn, limit=limit)
except Exception:
return []
out = []
for r in rows:
try:
result_json = r.get("result_json")
result = json.loads(result_json) if result_json else {}
except (TypeError, ValueError):
result = {}
out.append({
"job_id": f"db-scan-{r['id']}",
"kind": r.get("kind") or "consolidated",
"state": r.get("status") or "done",
"started": r.get("started_at"),
"finished": r.get("finished_at"),
"error": r.get("error"),
"busy": False,
"details": {},
"result": result or None,
})
return out
def latest_scans(limit: int = 10) -> list[dict[str, Any]]:
"""Return the most recently started jobs, newest first.
Appends a file-derived "latest scan" entry when the newest report files on
disk reflect a scan newer than any known job (e.g. after a restart or a
CLI run), so the dashboard always shows the latest completed scan.
Merges three sources, newest-first:
1. the in-memory job store (currently running / last-run jobs),
2. the SQLite ``scans`` table (persisted by run_scan/analyzers — S001),
3. a file-derived "latest scan" fallback for pre-DB runs.
Any DB-backed scan not already represented in memory is appended so the
dashboard shows scans from CLI runs and across restarts.
"""
load_persisted()
with _jobs_lock:
jobs = sorted(_jobs.values(),
key=lambda j: j.started or "", reverse=True)
entries = [j.to_dict() for j in jobs]
# Add DB scan records that aren't already in the in-memory store. Key on
# (kind, started_at) to avoid duplicating a job that run_scan also logged.
known = {(e.get("kind"), e.get("started")) for e in entries}
for db_row in _db_scans(limit=limit * 2):
if (db_row.get("kind"), db_row.get("started")) not in known:
entries.append(db_row)
file_entry = latest_file_scan()
if file_entry is not None:
newest = jobs[0] if jobs else None
newest_ts = _iso_key(
(newest.finished or newest.started) if newest else None)
newest = (entries or [None])[0]
newest_key = (
(newest.get("finished") or newest.get("started")) if newest else None)
newest_ts = _iso_key(newest_key)
# Only surface the file-derived entry when it is genuinely newer than
# every known job (tolerance avoids duplicating the run that produced
# it, where collected_at lands a moment after the job finished).
+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python3
"""Tests for the SQLite repositories + schema migrations (T001).
Run with:
python -m unittest discover -s tests -p 'test_*.py'
or (if pytest is installed):
python -m pytest tests/
"""
from __future__ import annotations
import os
import sqlite3
import tempfile
import unittest
from app.db import connect
from app.db import repositories as repo
from app.db.schema import SCHEMA_VERSION, apply_migrations
def _new_conn() -> sqlite3.Connection:
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
os.unlink(path)
return connect(path)
class SchemaTest(unittest.TestCase):
def test_migrations_apply_and_are_idempotent(self):
conn = _new_conn()
self.assertEqual(conn.execute("PRAGMA user_version").fetchone()[0],
SCHEMA_VERSION)
# Re-applying should be a no-op (no exception, version unchanged).
apply_migrations(conn)
self.assertEqual(conn.execute("PRAGMA user_version").fetchone()[0],
SCHEMA_VERSION)
conn.close()
def test_core_tables_exist(self):
conn = _new_conn()
tables = {r["name"] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'")}
for t in ("requests", "ip_stats", "scans", "scans_sources",
"blocklist_entries", "blocklist_events",
"blocklist_reasons", "whitelist_entries",
"whitelist_sources", "firewall_snapshots",
"firewall_address_list_entries", "firewall_rule_entries",
"whois_info", "log_sources"):
self.assertIn(t, tables, f"missing table {t}")
conn.close()
class ScanAndRequestTest(unittest.TestCase):
def setUp(self):
self.conn = _new_conn()
def tearDown(self):
self.conn.close()
def test_scan_lifecycle(self):
sid = repo.start_scan(self.conn, "ids", params={"full": True})
self.assertEqual(repo.get_scan(self.conn, sid)["status"], "running")
repo.finish_scan(self.conn, sid, status="done",
result={"ok": True}, request_count=3)
s = repo.get_scan(self.conn, sid)
self.assertEqual(s["status"], "done")
self.assertEqual(s["request_count"], 3)
self.assertEqual([x["id"] for x in repo.latest_scans(self.conn)],
[sid])
def test_insert_and_query_requests(self):
sid = repo.start_scan(self.conn, "ids")
rows = [
{"source": "npm", "scan_id": sid, "log_file": "a.log",
"ts": "2026-09-01 10:00:00", "ip": "1.2.3.4", "version": 4,
"method": "GET", "path": "/.env", "status": 404,
"host": "h", "user": None, "ua": "curl", "referer": None,
"denied": False, "banned": False, "is_error": True, "extra": None},
{"source": "copyparty", "scan_id": sid, "log_file": "f.txt",
"ts": "2026-09-01 10:00:01", "ip": "1.2.3.4", "version": 4,
"method": "GET", "path": "/admin", "status": 403,
"host": None, "user": None, "ua": None, "referer": None,
"denied": True, "banned": False, "is_error": True, "extra": None},
]
self.assertEqual(repo.insert_requests(self.conn, rows), 2)
self.assertEqual(repo.count_requests(self.conn), 2)
# cross-source
cs = repo.correlate_across_sources(self.conn)
self.assertEqual(cs[0]["ip"], "1.2.3.4")
self.assertEqual(cs[0]["sources"], 2)
# timeline
tl = repo.ip_timeline(self.conn, "1.2.3.4")
self.assertEqual(len(tl), 2)
# top ips
top = repo.top_ips(self.conn)
self.assertEqual(top[0]["total"], 2)
repo.finish_scan(self.conn, sid, status="done")
class BlocklistTest(unittest.TestCase):
def setUp(self):
self.conn = _new_conn()
def tearDown(self):
self.conn.close()
def test_grow_only_record(self):
r1 = repo.record_blocklist(self.conn, ["5.5.5.5/32", "6.6.6.6/32"],
added_by="scan:1")
self.assertEqual(r1["active"], 2)
# a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative).
r2 = repo.record_blocklist(self.conn, ["5.5.5.5/32"], added_by="scan:2")
active = [e["cidr"] for e in repo.active_blocklist(self.conn)]
self.assertIn("6.6.6.6/32", active)
self.assertEqual(r2["active"], 2)
def test_unblock(self):
repo.record_blocklist(self.conn, ["7.7.7.7/32"])
self.assertTrue(repo.unblock_entry(self.conn, "7.7.7.7/32"))
self.assertNotIn("7.7.7.7/32",
[e["cidr"] for e in repo.active_blocklist(self.conn)])
self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32"))
class WhitelistTest(unittest.TestCase):
def setUp(self):
self.conn = _new_conn()
def tearDown(self):
self.conn.close()
def test_whitelist_lookup(self):
repo.set_whitelist(self.conn, "Cloudflare", ["173.245.48.0/20"],
source="test")
self.assertTrue(repo.is_whitelisted_db(self.conn, "173.245.48.5"))
self.assertFalse(repo.is_whitelisted_db(self.conn, "8.8.8.8"))
# provider replace deactivates old ranges
repo.set_whitelist(self.conn, "Cloudflare", ["104.16.0.0/13"])
self.assertFalse(repo.is_whitelisted_db(self.conn, "173.245.48.5"))
class FirewallTest(unittest.TestCase):
def setUp(self):
self.conn = _new_conn()
def tearDown(self):
self.conn.close()
def test_firewall_snapshot(self):
sid = repo.save_firewall_snapshot(
self.conn,
address_lists_v4=[{"list": "Blocked", "address": "5.5.5.5/32"}],
address_lists_v6=[{"list": "Init7", "address": "2001:1620::/30"}],
rules_v4=[{"chain": "input", "action": "drop", "comment": "x"}],
rules_v6=[])
self.assertEqual(sid, 1)
self.assertEqual(self.conn.execute(
"SELECT COUNT(*) FROM firewall_address_list_entries").fetchone()[0], 2)
self.assertEqual(self.conn.execute(
"SELECT COUNT(*) FROM firewall_rule_entries").fetchone()[0], 1)
if __name__ == "__main__":
unittest.main()