Compare commits
4
Commits
1e80be746e
...
6051d2b548
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6051d2b548 | ||
|
|
2fb4032766 | ||
|
|
ba48054293 | ||
|
|
4a30f6b582 |
No files matched your search
@@ -172,6 +172,11 @@ python app.py scan --quiet
|
|||||||
python app.py scan --update-firewall --dry-run
|
python app.py scan --update-firewall --dry-run
|
||||||
python app.py scan --update-firewall
|
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)
|
# individual scanners (still runnable directly)
|
||||||
python -m app.copyparty
|
python -m app.copyparty
|
||||||
python -m app.npm
|
python -m app.npm
|
||||||
|
|||||||
@@ -40,12 +40,22 @@ def _run_scan(argv: list[str]) -> None:
|
|||||||
scan_main(argv)
|
scan_main(argv)
|
||||||
|
|
||||||
|
|
||||||
|
def _run_analyze(argv: list[str]) -> None:
|
||||||
|
from app.analyze import main as analyze_main
|
||||||
|
analyze_main(argv)
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
# `python app.py scan ...` runs the consolidated CLI scan.
|
# `python app.py scan ...` runs the consolidated CLI scan.
|
||||||
if len(sys.argv) > 1 and sys.argv[1] == "scan":
|
if len(sys.argv) > 1 and sys.argv[1] == "scan":
|
||||||
_run_scan(sys.argv[2:])
|
_run_scan(sys.argv[2:])
|
||||||
return
|
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).
|
# Default: start the web UI (parse_known_args tolerates stray flags).
|
||||||
args, _ = _web_parser().parse_known_args()
|
args, _ = _web_parser().parse_known_args()
|
||||||
_run_web(args.web_host, args.web_port)
|
_run_web(args.web_host, args.web_port)
|
||||||
|
|||||||
+145
@@ -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()
|
||||||
@@ -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())
|
||||||
@@ -322,6 +322,32 @@ def unblock_entry(conn: sqlite3.Connection, cidr: str, *,
|
|||||||
return removed
|
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], *,
|
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``.
|
||||||
@@ -465,6 +491,20 @@ def save_firewall_snapshot(conn: sqlite3.Connection, *,
|
|||||||
return snapshot_id
|
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
|
# WHOIS cache
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
+35
-5
@@ -42,18 +42,42 @@ 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], 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
|
Uses ``record_blocklist`` (add-only) so the ledger is the authoritative
|
||||||
analysis always sees the current policy."""
|
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:
|
try:
|
||||||
from ..db import get_connection
|
from ..db import get_connection
|
||||||
from ..db import repositories as repo
|
from ..db import repositories as repo
|
||||||
conn = get_connection()
|
conn = get_connection()
|
||||||
repo.sync_blocklist(conn, list(desired_v4) + list(desired_v6),
|
return repo.record_blocklist(
|
||||||
added_by=added_by)
|
conn, list(desired_v4) + list(desired_v6), added_by=added_by)
|
||||||
except Exception as exc: # never break the scan for a DB issue
|
except Exception as exc: # never break the scan for a DB issue
|
||||||
print(f"WARN: could not sync blocklist to db ({exc})")
|
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
|
@dataclass
|
||||||
@@ -256,6 +280,12 @@ def run_scan(params: ScanParams) -> ScanResult:
|
|||||||
f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
|
f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.")
|
||||||
|
|
||||||
# Timestamped change report — only when the router was reachable.
|
# Timestamped change report — only when the router was reachable.
|
||||||
|
# Source of truth for the cumulative set: the SQLite |ledger| first
|
||||||
|
# (B001), falling back to the git-history reconstruction only if the
|
||||||
|
# DB is unavailable.
|
||||||
|
complete_v4, complete_v6 = _db_complete_blocklist(
|
||||||
|
blocked_ipv4, blocked_ipv6)
|
||||||
|
if not complete_v4 and not complete_v6:
|
||||||
complete_v4, complete_v6 = cumulative_blocklist(
|
complete_v4, complete_v6 = cumulative_blocklist(
|
||||||
blocklist_v4_path, blocklist_v6_path,
|
blocklist_v4_path, blocklist_v6_path,
|
||||||
blocked_ipv4, blocked_ipv6)
|
blocked_ipv4, blocked_ipv6)
|
||||||
|
|||||||
@@ -23,6 +23,11 @@ endblock %} {% block content %}
|
|||||||
<div class="panel-head"><h2>Firewall rules</h2></div>
|
<div class="panel-head"><h2>Firewall rules</h2></div>
|
||||||
<div id="firewall-rules"></div>
|
<div id="firewall-rules"></div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<div class="panel">
|
||||||
|
<div class="panel-head"><h2>Snapshot history (SQLite)</h2></div>
|
||||||
|
<div id="fw-snapshots"></div>
|
||||||
|
</div>
|
||||||
{% endblock %} {% block scripts %}
|
{% endblock %} {% block scripts %}
|
||||||
<script>
|
<script>
|
||||||
async function loadState() {
|
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) {
|
function renderAddressLists(addr) {
|
||||||
const box = document.getElementById("address-lists");
|
const box = document.getElementById("address-lists");
|
||||||
box.innerHTML = "";
|
box.innerHTML = "";
|
||||||
@@ -172,5 +216,6 @@ endblock %} {% block content %}
|
|||||||
.getElementById("apply-btn")
|
.getElementById("apply-btn")
|
||||||
.addEventListener("click", () => sync(false));
|
.addEventListener("click", () => sync(false));
|
||||||
loadState();
|
loadState();
|
||||||
|
loadSnapshots();
|
||||||
</script>
|
</script>
|
||||||
{% endblock %}
|
{% endblock %}
|
||||||
@@ -265,6 +265,19 @@ def register_routes(app: Flask,
|
|||||||
"snapshot_dir": MIKROTIK_DATA_DIR,
|
"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")
|
@app.post("/api/firewall/sync")
|
||||||
def firewall_sync():
|
def firewall_sync():
|
||||||
return _sync_firewall(dry_run=True)
|
return _sync_firewall(dry_run=True)
|
||||||
|
|||||||
+54
-6
@@ -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]]:
|
def latest_scans(limit: int = 10) -> list[dict[str, Any]]:
|
||||||
"""Return the most recently started jobs, newest first.
|
"""Return the most recently started jobs, newest first.
|
||||||
|
|
||||||
Appends a file-derived "latest scan" entry when the newest report files on
|
Merges three sources, newest-first:
|
||||||
disk reflect a scan newer than any known job (e.g. after a restart or a
|
1. the in-memory job store (currently running / last-run jobs),
|
||||||
CLI run), so the dashboard always shows the latest completed scan.
|
2. the SQLite ``scans`` table (persisted by run_scan/analyzers — S001),
|
||||||
|
3. a file-derived "latest scan" fallback for pre-DB runs.
|
||||||
|
Any DB-backed scan not already represented in memory is appended so the
|
||||||
|
dashboard shows scans from CLI runs and across restarts.
|
||||||
"""
|
"""
|
||||||
load_persisted()
|
load_persisted()
|
||||||
with _jobs_lock:
|
with _jobs_lock:
|
||||||
jobs = sorted(_jobs.values(),
|
jobs = sorted(_jobs.values(),
|
||||||
key=lambda j: j.started or "", reverse=True)
|
key=lambda j: j.started or "", reverse=True)
|
||||||
entries = [j.to_dict() for j in jobs]
|
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()
|
file_entry = latest_file_scan()
|
||||||
if file_entry is not None:
|
if file_entry is not None:
|
||||||
newest = jobs[0] if jobs else None
|
newest = (entries or [None])[0]
|
||||||
newest_ts = _iso_key(
|
newest_key = (
|
||||||
(newest.finished or newest.started) if newest else None)
|
(newest.get("finished") or newest.get("started")) if newest else None)
|
||||||
|
newest_ts = _iso_key(newest_key)
|
||||||
# Only surface the file-derived entry when it is genuinely newer than
|
# Only surface the file-derived entry when it is genuinely newer than
|
||||||
# every known job (tolerance avoids duplicating the run that produced
|
# every known job (tolerance avoids duplicating the run that produced
|
||||||
# it, where collected_at lands a moment after the job finished).
|
# it, where collected_at lands a moment after the job finished).
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in new issue
Block a user