From 6036fa8b1eae4e7b54b2cd056e4eccc3d42a4a14 Mon Sep 17 00:00:00 2001 From: Zwitschi Date: Sat, 24 Oct 2026 19:48:20 +0200 Subject: [PATCH] feat: Implement SQLite schema and migrations for intrusion detection system - Added schema definition and migration logic in app/db/schema.py. - Created tables for log sources, requests, IP stats, scans, blocklist, whitelist, and firewall rules. - Introduced functions to apply migrations and ensure database integrity. feat: Enhance scan functionality with database integration - Added _sync_blocklist_to_db function to synchronize blocklist with SQLite. - Updated run_scan function to handle database connections and scan records. - Implemented error handling for database operations during scans. feat: Introduce log analysis page and API endpoints - Created analysis.html template for displaying log analysis results. - Added routes for fetching top IPs, IP request timelines, and cross-source IPs. - Integrated analysis data into the dashboard and status endpoints. fix: Ensure whitelists are synchronized with the database - Implemented sync_whitelists_to_db function to persist in-memory whitelists. - Updated constants.py to define the path for the SQLite database. refactor: Modularize analysis helpers for better maintainability - Created _analysis_helpers function to encapsulate SQLite query logic. - Updated routes.py to utilize new analysis helper functions for API responses. --- .env.example | 6 + .gitignore | 5 + README.md | 11 +- app/analyzer.py | 63 ++++- app/data/cloudflare/ips-v4.txt | 15 ++ app/data/cloudflare/ips-v6.txt | 7 + app/db/__init__.py | 62 +++++ app/db/repositories.py | 478 +++++++++++++++++++++++++++++++++ app/db/schema.py | 268 ++++++++++++++++++ app/scan/__init__.py | 75 +++++- app/templates/analysis.html | 191 +++++++++++++ app/templates/base.html | 5 + app/templates/dashboard.html | 7 + app/utils/constants.py | 5 + app/utils/whitelists.py | 17 ++ app/web/__init__.py | 36 +++ app/web/routes.py | 46 ++++ 17 files changed, 1292 insertions(+), 5 deletions(-) create mode 100644 app/data/cloudflare/ips-v4.txt create mode 100644 app/data/cloudflare/ips-v6.txt create mode 100644 app/db/__init__.py create mode 100644 app/db/repositories.py create mode 100644 app/db/schema.py create mode 100644 app/templates/analysis.html diff --git a/.env.example b/.env.example index 8e78739..b38d7be 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,12 @@ SUBNET_THRESHOLD=3 REPORT_RETENTION_DAYS=30 REPORT_RETENTION_MAX_FILES=10 +# --- SQLite store --------------------------------------------------------- + +# Local SQLite database canonical store (logs, scans, blocklist/whitelist and +# firewall history). Created automatically on first run. Default: data/ids.db +#IDS_DB_PATH=/absolute/path/to/ids.db + # --- Scan / flagging defaults ---------------------------------------------- # Rows in the top-IP report tables. diff --git a/.gitignore b/.gitignore index 7917295..1bddfc2 100644 --- a/.gitignore +++ b/.gitignore @@ -20,5 +20,10 @@ app/data/* !app/data/cloudflare/ !app/data/paloalto/ +# SQLite database (canonical store) + WAL/SHM sidecars +app/data/*.db +app/data/*.db-wal +app/data/*.db-shm + # generated reports app/reports/ diff --git a/README.md b/README.md index 68f6952..bfaee33 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,10 @@ intrusion-detection/ │ ├── scan/ scan pipeline │ │ ├── __init__.py consolidated scanner (run_scan, ScanParams, ScanResult) │ │ └── scan_state.py incremental log scan state (load/save) +│ ├── db/ SQLite store (see .github/instructions/DATABASE.md) +│ │ ├── __init__.py connection factory + migrations runner +│ │ ├── schema.py DDL for all tables (PRAGMA user_version migrations) +│ │ └── repositories.py queries: logs, scans, blocklist, whitelist, firewall │ ├── utils/ shared utilities │ │ ├── constants.py all config constants (paths, env vars, thresholds) │ │ ├── files.py read_lines, find_logs, write_csv/json/lines @@ -67,8 +71,9 @@ intrusion-detection/ │ ├── cloudflare/ │ │ ├── ips-v4.txt │ │ └── ips-v6.txt -│ └── paloalto/ -│ └── scanner-ips.txt +│ ├── paloalto/ +│ │ └── scanner-ips.txt +│ └── ids.db local SQLite store (log/scan/blocklist/whitelist history; auto-created) ├── reports/ generated reports (per-run + per-source) │ ├── copyparty/ │ ├── npm/ @@ -253,6 +258,8 @@ Pages: | `/blocklist` | view the consolidated blocklist, add/remove entries | | `/firewall` | view address lists + rules, preview/apply a sync | | `/whitelist` | view/manage provider whitelist CIDRs | +| `/graph` | visualize the 0.0.0.0/8 address space (blocked/whitelisted pixels) | +| `/analysis` | detailed persisted-log analysis (top IPs, drill-down, cross-source) | Scans run in a background thread (only one at a time) and report progress via the JSON API, so the UI never blocks. Firewall sync is **dry-run by default** diff --git a/app/analyzer.py b/app/analyzer.py index 0b6bb89..907db42 100644 --- a/app/analyzer.py +++ b/app/analyzer.py @@ -60,11 +60,66 @@ class Analyzer: self.flagged: dict[str, IPStats] = {} self.per_file: dict[str, int] = {} + # --- sqlite persistence ------------------------------------------------ + + def persist_to_db(self, scan_id: int | None = None) -> None: + """Ingest this analyzer's requests + per-IP stats into SQLite. + + Lazily imported to avoid cycles. When ``scan_id`` is None a standalone + scan record is created for this analyzer so individual + (non-consolidated) runs are still tracked. Never raises: a DB failure + logs a warning but does not break the scan. + """ + if not self.requests and not self.stats: + return + try: + from .db import get_connection + from .db import repositories as repo + from .utils.ip_utils import ip_version as _ip_ver + conn = get_connection() + if scan_id is None: + scan_id = repo.start_scan(conn, self.name, + params={"full": self.full}) + source = self.name + rows: list[dict] = [] + for r in self.requests: + rows.append({ + "source": source, "scan_id": scan_id, + "log_file": ((r.extra or {}).get("log_file") + or (self.files[0] if self.files else None)), + "ts": r.ts, "ip": r.ip, + "version": _ip_ver(r.ip) or 0, + "method": r.method, "path": r.path, + "status": r.status, + "upstream_status": r.upstream_status, + "host": r.host, "user": r.user, "ua": r.ua, + "referer": r.referer, "denied": r.denied, + "banned": r.banned, "is_error": r.is_error, + "extra": r.extra, + }) + repo.insert_requests(conn, rows) + repo.insert_ip_stats(conn, scan_id, + list(self.stats.values()), + set(self.block_candidates())) + repo.set_scan_source_counts( + conn, scan_id, source, len(self.requests), len(self.files)) + repo.finish_scan(conn, scan_id, status="done", + log_files=len(self.files), + request_count=len(self.requests)) + except Exception as exc: # never break the scan because of DB issues + if not self.quiet: + print(f"WARN: sqlite ingest skipped ({exc})") # --- hooks ------------------------------------------------------------- def parse_line(self, line: str) -> Request | None: raise NotImplementedError + def attr_source(self, req: Request, log_file: str) -> None: + """Record the originating log file onto a parsed request.""" + if req.extra is None: + req.extra = {} + req.extra["log_file"] = req.extra.get("log_file", log_file) + def flag(self, stats: dict[str, IPStats]) -> dict[str, IPStats]: raise NotImplementedError @@ -111,6 +166,7 @@ class Analyzer: for line in read_lines(full_path): req = self.parse_line(line) if req is not None: + self.attr_source(req, name) self.requests.append(req) count += 1 else: @@ -125,6 +181,7 @@ class Analyzer: for line, resume in read_lines_from(full_path, start): req = self.parse_line(line) if req is not None: + self.attr_source(req, name) self.requests.append(req) count += 1 last_resume = resume @@ -262,7 +319,7 @@ class Analyzer: # --- entrypoint -------------------------------------------------------- - def run(self) -> dict[str, Any]: + def run(self, db_scan_id: int | None = None) -> dict[str, Any]: validate_dir(self.log_path) ensure_dirs(self.report_dir, self.data_dir) refresh_whitelists() @@ -273,6 +330,10 @@ class Analyzer: if not self.quiet: print(self.text_report()) + # Ingest into the SQLite store (creates its own scan record when no + # consolidated ``db_scan_id`` is supplied). + self.persist_to_db(scan_id=db_scan_id) + self.write_data() report_path = self.write_report() print(f"data files: {self.data_dir}") diff --git a/app/data/cloudflare/ips-v4.txt b/app/data/cloudflare/ips-v4.txt new file mode 100644 index 0000000..626248f --- /dev/null +++ b/app/data/cloudflare/ips-v4.txt @@ -0,0 +1,15 @@ +173.245.48.0/20 +103.21.244.0/22 +103.22.200.0/22 +103.31.4.0/22 +141.101.64.0/18 +108.162.192.0/18 +190.93.240.0/20 +188.114.96.0/20 +197.234.240.0/22 +198.41.128.0/17 +162.158.0.0/15 +104.16.0.0/13 +104.24.0.0/14 +172.64.0.0/13 +131.0.72.0/22 \ No newline at end of file diff --git a/app/data/cloudflare/ips-v6.txt b/app/data/cloudflare/ips-v6.txt new file mode 100644 index 0000000..4584f6a --- /dev/null +++ b/app/data/cloudflare/ips-v6.txt @@ -0,0 +1,7 @@ +2400:cb00::/32 +2606:4700::/32 +2803:f800::/32 +2405:b500::/32 +2405:8100::/32 +2a06:98c0::/29 +2c0f:f248::/32 \ No newline at end of file diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..652904c --- /dev/null +++ b/app/db/__init__.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +"""Local SQLite database layer for the intrusion-detection system. + +Owns the shared connection factory (WAL mode, busy timeout, single mutable +connection reused across the background-scan thread and web readers) and +applies schema migrations on open. Query helpers live in :mod:`repositories`. +""" + +from __future__ import annotations + +import os +import sqlite3 +import threading + +from ..utils.constants import DATA_DB_PATH +from .schema import apply_migrations + +_connection: sqlite3.Connection | None = None +_lock = threading.RLock() + + +def connect(path: str | None = None) -> sqlite3.Connection: + """Open an initialized SQLite connection at ``path``. + + Enables WAL for concurrent readers during the single-writer background + scan and sets a busy timeout so short write contention does not raise. + Foreign keys are on and migrations are applied. + """ + db_path = path or DATA_DB_PATH + os.makedirs(os.path.dirname(os.path.abspath(db_path)) or ".", exist_ok=True) + conn = sqlite3.connect( + db_path, + check_same_thread=False, # background scan thread + web handlers + isolation_level=None, # autocommit; use explicit BEGIN for batches + ) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + conn.execute("PRAGMA synchronous=NORMAL") + conn.row_factory = sqlite3.Row + apply_migrations(conn) + return conn + + +def get_connection() -> sqlite3.Connection: + """Return the shared, lazily-created connection (thread-safe).""" + global _connection + with _lock: + if _connection is None: + _connection = connect() + return _connection + + +def close() -> None: + """Close the shared connection, if any.""" + global _connection + with _lock: + if _connection is not None: + try: + _connection.close() + except sqlite3.Error: + pass + _connection = None diff --git a/app/db/repositories.py b/app/db/repositories.py new file mode 100644 index 0000000..070fb1f --- /dev/null +++ b/app/db/repositories.py @@ -0,0 +1,478 @@ +#!/usr/bin/env python3 +"""Query helpers for the intrusion-detection SQLite store. + +Functions are grouped by domain and take an explicit ``sqlite3.Connection`` so +callers control transaction boundaries. Bulk inserts are batched with +``executemany`` inside a single transaction for speed. +""" + +from __future__ import annotations + +import json +import sqlite3 +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any, Iterator, Sequence + +from ..utils.models import IPStats + + +def utcnow() -> str: + """ISO-8601 timestamp with seconds precision (UTC).""" + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +@contextmanager +def transaction(conn: sqlite3.Connection) -> Iterator[sqlite3.Connection]: + """Context manager for an explicit transaction (commit or rollback).""" + conn.execute("BEGIN") + try: + yield conn + except Exception: + conn.execute("ROLLBACK") + raise + else: + conn.execute("COMMIT") + + +# --------------------------------------------------------------------------- +# Scans +# --------------------------------------------------------------------------- + +def start_scan(conn: sqlite3.Connection, kind: str, *, + params: dict[str, Any] | None = None, + full: bool = False, dry_run: bool = False) -> int: + """Record a scan start and return its id.""" + cur = conn.execute( + "INSERT INTO scans (kind, started_at, status, full, dry_run, params_json) " + "VALUES (?, ?, 'running', ?, ?, ?)", + (kind, utcnow(), int(full), int(dry_run), + json.dumps(params) if params is not None else None)) + return int(cur.lastrowid) + + +def finish_scan(conn: sqlite3.Connection, scan_id: int, *, + status: str, result: dict[str, Any] | None = None, + log_files: int = 0, request_count: int = 0, + error: str | None = None) -> None: + """Mark a scan complete with outcome counters.""" + conn.execute( + "UPDATE scans SET finished_at=?, status=?, result_json=?, " + "log_files=?, request_count=?, error=? WHERE id=?", + (utcnow(), status, + json.dumps(result) if result is not None else None, + log_files, request_count, error, scan_id)) + + +def set_scan_source_counts(conn: sqlite3.Connection, scan_id: int, + source: str, requests: int, files: int) -> None: + """Record per-source request/file counts for a scan.""" + conn.execute( + "INSERT INTO scans_sources (scan_id, source, requests, files) " + "VALUES (?, ?, ?, ?) " + "ON CONFLICT(scan_id, source) DO UPDATE SET " + "requests=excluded.requests, files=excluded.files", + (scan_id, source, requests, files)) + + +def latest_scans(conn: sqlite3.Connection, limit: int = 10) -> list[dict[str, Any]]: + """Most recent scan metadata rows.""" + rows = conn.execute( + "SELECT id, kind, started_at, finished_at, status, full, dry_run, " + "log_files, request_count, error FROM scans " + "ORDER BY id DESC LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + +def get_scan(conn: sqlite3.Connection, scan_id: int) -> dict[str, Any] | None: + row = conn.execute( + "SELECT id, kind, started_at, finished_at, status, full, dry_run, " + "params_json, result_json, log_files, request_count, error " + "FROM scans WHERE id=?", (scan_id,)).fetchone() + return dict(row) if row is not None else None + + +# --------------------------------------------------------------------------- +# Logs + raw requests +# --------------------------------------------------------------------------- + +def _parse_ts(ts: str) -> tuple[str, float | None]: + """Best-effort ISO timestamp extraction from a log line timestamp string.""" + if not ts: + return (ts, None) + epoch = None + cleaned = ts.strip() + # Accept a trailing timezone like +HH:MM or Z; fall back to naive parse. + try: + epoch = datetime.fromisoformat(cleaned.replace("Z", "+00:00")).timestamp() + except ValueError: + try: + epoch = datetime.fromisoformat(cleaned).timestamp() + except ValueError: + epoch = None + return (ts, epoch) + + +def insert_requests(conn: sqlite3.Connection, rows: Sequence[dict[str, Any]]) -> int: + """Bulk-insert ``requests`` rows. Returns inserted count. + + Each ``row`` maps keys: source, scan_id, log_file, ts, ip, version, + method, path, status, upstream_status, host, user, ua, referer, denied, + banned, is_error, malicious_hits, extra(dict|None) — matching + :mod:`app.utils.models.Request` plus provenance fields. + """ + if not rows: + return 0 + created = utcnow() + tuples: list[tuple] = [] + for r in rows: + ts, epoch = _parse_ts(r.get("ts") or "") + extra = r.get("extra") + tuples.append(( + r.get("source"), r.get("scan_id"), r.get("log_file"), ts, epoch, + r.get("ip"), r.get("version"), + r.get("method"), r.get("path"), + r.get("status"), r.get("upstream_status"), + r.get("host"), r.get("user"), r.get("ua"), r.get("referer"), + int(bool(r.get("denied"))), int(bool(r.get("banned"))), + int(bool(r.get("is_error"))), + int(r.get("malicious_hits") or 0), + json.dumps(extra) if extra is not None else None, + created, + )) + with transaction(conn): + conn.executemany( + "INSERT INTO requests (source, scan_id, log_file, ts, ts_epoch, ip, " + "version, method, path, status, upstream_status, host, user, ua, " + "referer, denied, banned, is_error, malicious_hits, extra, created_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + tuples) + return len(tuples) + + +def query_requests(conn: sqlite3.Connection, *, + ip: str | None = None, source: str | None = None, + since: str | None = None, until: str | None = None, + path: str | None = None, limit: int = 500) -> list[dict[str, Any]]: + """Return matching request rows (optionally filtered), newest last.""" + clauses: list[str] = [] + args: list[Any] = [] + if ip: + clauses.append("ip = ?") + args.append(ip) + if source: + clauses.append("source = ?") + args.append(source) + if since: + clauses.append("COALESCE(ts, '') >= ?") + args.append(since) + if until: + clauses.append("COALESCE(ts, '') <= ?") + args.append(until) + if path: + clauses.append("path LIKE ?") + args.append(path) + sql = "SELECT * FROM requests" + if clauses: + sql += " WHERE " + " AND ".join(clauses) + sql += " ORDER BY id LIMIT ?" + args.append(int(limit)) + return [dict(r) for r in conn.execute(sql, args).fetchall()] + + +# --------------------------------------------------------------------------- +# IP aggregates +# --------------------------------------------------------------------------- + +def insert_ip_stats(conn: sqlite3.Connection, scan_id: int, + stats: Sequence[IPStats], + blocked: set[str]) -> int: + """Persist per-IP aggregates for a scan. ``blocked`` are direct blocked IPs.""" + if not stats: + return 0 + tuples: list[tuple] = [] + for s in stats: + tuples.append(( + scan_id, s.ip, s.version, s.total, s.errors, s.c4xx, s.c5xx, + s.denied, s.banned, s.first, s.last, s.malicious_hits, + json.dumps(dict(s.paths)), json.dumps(dict(s.methods)), + json.dumps(dict(s.hosts)), json.dumps(dict(s.uas)), + json.dumps(sorted(s.users)), s.whitelist or None, + int(s.ip in blocked), + )) + with transaction(conn): + conn.executemany( + "INSERT INTO ip_stats (scan_id, ip, version, total, errors, c4xx, " + "c5xx, denied, banned, first_ts, last_ts, malicious_hits, paths_json, " + "methods_json, hosts_json, uas_json, users_json, whitelist, blocked) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + tuples) + return len(tuples) + + +def ip_timeline(conn: sqlite3.Connection, ip: str, + limit: int = 2000) -> list[dict[str, Any]]: + """Chronological request history for one IP across all sources.""" + rows = conn.execute( + "SELECT source, log_file, ts, ts_epoch, method, path, status, host, " + "ua, denied, banned, is_error, malicious_hits FROM requests " + "WHERE ip = ? ORDER BY COALESCE(ts_epoch, 0), id LIMIT ?", + (ip, limit)).fetchall() + return [dict(r) for r in rows] + + +def count_requests(conn: sqlite3.Connection) -> int: + """Total persisted request records.""" + return int(conn.execute("SELECT COUNT(*) FROM requests").fetchone()[0]) + + +def count_scans(conn: sqlite3.Connection) -> int: + """Total persisted scan records.""" + return int(conn.execute("SELECT COUNT(*) FROM scans").fetchone()[0]) + + +def correlate_across_sources(conn: sqlite3.Connection, + limit: int = 100) -> list[dict[str, Any]]: + """IPs seen in more than one log source, with per-source counts.""" + rows = conn.execute( + "SELECT ip, COUNT(DISTINCT source) AS sources, " + "GROUP_CONCAT(DISTINCT source) AS source_list, COUNT(*) AS total " + "FROM requests WHERE ip IS NOT NULL AND ip != '' " + "GROUP BY ip HAVING sources > 1 " + "ORDER BY total DESC LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + +def top_ips(conn: sqlite3.Connection, since: str | None = None, + limit: int = 50) -> list[dict[str, Any]]: + """Top IPs by request count within an optional time window.""" + rows = conn.execute( + "SELECT ip, version, COUNT(*) AS total, " + "SUM(is_error) AS errors, SUM(malicious_hits) AS malicious_hits, " + "MIN(COALESCE(ts_epoch, 0)) AS first_epoch, " + "MAX(COALESCE(ts_epoch, 0)) AS last_epoch " + "FROM requests WHERE ip IS NOT NULL AND ip != '' " + + ("AND COALESCE(ts, '') >= ? " if since else "") + + "GROUP BY ip ORDER BY total DESC LIMIT ?", + ((since,) if since else ()) + (limit,)).fetchall() + return [dict(r) for r in rows] + + +# --------------------------------------------------------------------------- +# Blocklist ledger +# --------------------------------------------------------------------------- + +def active_blocklist(conn: sqlite3.Connection) -> list[dict[str, Any]]: + rows = conn.execute( + "SELECT id, cidr, version, added_at, added_by, last_seen " + "FROM blocklist_entries WHERE status='active' ORDER BY cidr").fetchall() + return [dict(r) for r in rows] + + +def block_entry(conn: sqlite3.Connection, cidr: str, *, + added_by: str | None = None, last_seen: str | None = None, + reasons: list[dict[str, Any]] | None = None) -> None: + """Add a CIDR to the grow-only ledger (idempotent when already active). + + Manages its own transaction for standalone use.""" + with transaction(conn): + _block_entry_inner(conn, cidr, added_by=added_by, + last_seen=last_seen, reasons=reasons) + + +def _block_entry_inner(conn: sqlite3.Connection, cidr: str, *, + added_by: str | None = None, + last_seen: str | None = None, + reasons: list[dict[str, Any]] | None = None) -> None: + """Low-level block insert; caller owns the transaction.""" + now = utcnow() + version = _cidr_version(cidr) + existing = conn.execute( + "SELECT id FROM blocklist_entries WHERE cidr = ? AND status='active'", + (cidr,)).fetchone() + if existing is not None: + entry_id = int(existing["id"]) + conn.execute( + "UPDATE blocklist_entries SET last_seen = COALESCE(?, last_seen) " + "WHERE id = ?", (last_seen, entry_id)) + if reasons: + insert_reasons(conn, entry_id, last_seen or "", reasons) + return + cur = conn.execute( + "INSERT INTO blocklist_entries (cidr, version, status, added_at, " + "added_by, last_seen) VALUES (?, ?, 'active', ?, ?, ?)", + (cidr, version, now, added_by, last_seen)) + entry_id = int(cur.lastrowid) + if reasons: + insert_reasons(conn, entry_id, last_seen or "", reasons) + + +def unblock_entry(conn: sqlite3.Connection, cidr: str, *, + actor: str | None = None) -> bool: + """Remove a CIDR from active blocklist (idempotent). Returns True if removed.""" + now = utcnow() + cur = conn.execute( + "UPDATE blocklist_entries SET status='removed', removed_at=? " + "WHERE cidr=? AND status='active'", (now, cidr)) + removed = cur.rowcount > 0 + conn.execute( + "INSERT INTO blocklist_events (action, cidr, actor, fired_at, detail) " + "VALUES ('remove', ?, ?, ?, ?)", + (cidr, actor, now, None)) + return removed + + +def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *, + added_by: str | None = None) -> dict[str, Any]: + """Reconcile active blocklist to ``desired_cidrs``. + + Adds missing entries; marks active entries absent from ``desired_cidrs`` + as removed (only ones the system added) so the ledger stays grow-only but + reflects the current policy. Returns the add/remove delta. + """ + desired = set(desired_cidrs) + current = {r["cidr"] for r in active_blocklist(conn)} + added = sorted(desired - current) + removed = sorted(current - desired) + now = utcnow() + with transaction(conn): + for cidr in added: + _block_entry_inner(conn, cidr, added_by=added_by, last_seen=now) + conn.executemany( + "UPDATE blocklist_entries SET status='removed', removed_at=? " + "WHERE cidr=? AND status='active'", + [(now, cidr) for cidr in removed]) + return {"added": added, "removed": removed} + + +def insert_reasons(conn: sqlite3.Connection, entry_id: int, scan_session: str, + reasons: list[dict[str, Any]]) -> None: + tuples = [ + (entry_id, r.get("scan_id"), r.get("ip") or "", r.get("reason"), + r.get("pattern"), r.get("severity"), r.get("sample_path"), + int(r.get("hits") or 0)) + for r in reasons + ] + conn.executemany( + "INSERT INTO blocklist_reasons (entry_id, scan_id, ip, reason, pattern, " + "severity, sample_path, hits) VALUES (?,?,?,?,?,?,?,?)", + tuples) + + +def blocklist_history(conn: sqlite3.Connection, limit: int = 500) -> list[dict[str, Any]]: + rows = conn.execute( + "SELECT id, cidr, version, status, added_at, removed_at, added_by, " + "last_seen FROM blocklist_entries ORDER BY id DESC LIMIT ?", + (limit,)).fetchall() + return [dict(r) for r in rows] + + +def _cidr_version(cidr: str) -> int: + try: + import ipaddress + net = ipaddress.ip_network(cidr, strict=False) + return net.version + except ValueError: + return 4 if ":" not in str(cidr) else 6 + + +# --------------------------------------------------------------------------- +# Whitelists +# --------------------------------------------------------------------------- + +def set_whitelist(conn: sqlite3.Connection, provider: str, + cidrs: Sequence[str], source: str | None = None) -> None: + """Replace the CIDR set for a provider (mark old inactive, upsert new).""" + now = utcnow() + with transaction(conn): + conn.execute( + "UPDATE whitelist_entries SET active=0 WHERE provider=?", (provider,)) + for cidr in cidrs: + version = _cidr_version(cidr) + conn.execute( + "INSERT INTO whitelist_entries (provider, cidr, version, source, " + "active, updated_at) VALUES (?,?,?,?,1,?) " + "ON CONFLICT(provider, cidr) DO UPDATE SET active=1, " + "source=excluded.source, updated_at=excluded.updated_at", + (provider, cidr, version, source, now)) + conn.execute( + "INSERT INTO whitelist_sources (provider, source_type, last_refresh_at, " + "entry_count) VALUES (?, ?, ?, ?) ON CONFLICT(provider) DO UPDATE SET " + "source_type=excluded.source_type, last_refresh_at=excluded.last_refresh_at, " + "entry_count=excluded.entry_count", + (provider, source, now, len(cidrs))) + + +def is_whitelisted_db(conn: sqlite3.Connection, ip: str) -> bool: + """Check whether an IP falls in any active whitelist CIDR.""" + import ipaddress + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return False + rows = conn.execute( + "SELECT cidr FROM whitelist_entries WHERE active=1").fetchall() + for r in rows: + try: + net = ipaddress.ip_network(r["cidr"], strict=False) + except ValueError: + continue + if addr.version == net.version and addr in net: + return True + return False + + +# --------------------------------------------------------------------------- +# Firewall snapshots +# --------------------------------------------------------------------------- + +def save_firewall_snapshot(conn: sqlite3.Connection, *, + address_lists_v4: list[dict], + address_lists_v6: list[dict], + rules_v4: list[dict], rules_v6: list[dict]) -> int: + """Persist a full firewall snapshot; returns the snapshot id.""" + now = utcnow() + cur = conn.execute( + "INSERT INTO firewall_snapshots (captured_at, kind, trigger) " + "VALUES (?, 'snapshot', 'scan')", (now,)) + snapshot_id = int(cur.lastrowid) + with transaction(conn): + tuples = [ + (snapshot_id, "ipv4", e.get("list"), e.get("address"), + int(bool(e.get("dynamic"))), e.get("timeout")) + for e in address_lists_v4 + ] + [ + (snapshot_id, "ipv6", e.get("list"), e.get("address"), + int(bool(e.get("dynamic"))), e.get("timeout")) + for e in address_lists_v6 + ] + conn.executemany( + "INSERT INTO firewall_address_list_entries (snapshot_id, network, " + "list, address, dynamic, timeout) VALUES (?,?,?,?,?,?)", tuples) + rule_tuples = [ + (snapshot_id, "ipv4", idx, e.get("chain"), e.get("action"), + e.get("dst-list"), None, e.get("comment"), json.dumps(e)) + for idx, e in enumerate(rules_v4) + ] + [ + (snapshot_id, "ipv6", idx, e.get("chain"), e.get("action"), + e.get("dst-list"), None, e.get("comment"), json.dumps(e)) + for idx, e in enumerate(rules_v6) + ] + conn.executemany( + "INSERT INTO firewall_rule_entries (snapshot_id, network, rule_index, " + "chain, action, dst_list, proto_port, comment, raw_json) " + "VALUES (?,?,?,?,?,?,?,?,?)", rule_tuples) + return snapshot_id + + +# --------------------------------------------------------------------------- +# WHOIS cache +# --------------------------------------------------------------------------- + +def set_whois(conn: sqlite3.Connection, ip: str, name: str | None, + cidr: str | None) -> None: + conn.execute( + "INSERT INTO whois_info (ip, name, cidr, fetched_at) VALUES (?,?,?,?) " + "ON CONFLICT(ip) DO UPDATE SET name=excluded.name, cidr=excluded.cidr, " + "fetched_at=excluded.fetched_at", + (ip, name, cidr, utcnow())) diff --git a/app/db/schema.py b/app/db/schema.py new file mode 100644 index 0000000..e3a5f5f --- /dev/null +++ b/app/db/schema.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +"""SQLite schema and migrations for the intrusion-detection system. + +Definitions live here as the single source of truth and are applied through +``PRAGMA user_version`` migrations (see ``apply_migrations``). Each entry is an +ordered list of DDL statements; the version is bumped in lock-step so the +migrations are idempotent. +""" + +from __future__ import annotations + +import sqlite3 + +from typing import Iterable + +#: Latest schema version. Bump by one whenever a new migration is appended. +SCHEMA_VERSION = 1 + +#: 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; +#: append a new one instead. +MIGRATIONS: dict[int, list[str]] = { + 1: [ + # ------------------------------------------------------------------ + # log_sources — canonical provenance + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS log_sources ( + id INTEGER PRIMARY KEY, + name TEXT UNIQUE NOT NULL, + log_path TEXT, + enabled INTEGER NOT NULL DEFAULT 1 + ) + """, + """ + INSERT OR IGNORE INTO log_sources (name, log_path) + VALUES ('copyparty', NULL), ('npm', NULL) + """, + # ------------------------------------------------------------------ + # requests — one row per parsed request; insert-only + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS requests ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + scan_id INTEGER, + log_file TEXT, + ts TEXT, + ts_epoch REAL, + ip TEXT NOT NULL, + version INTEGER, + method TEXT, + path TEXT, + status INTEGER, + upstream_status INTEGER, + host TEXT, + user TEXT, + ua TEXT, + referer TEXT, + denied INTEGER NOT NULL DEFAULT 0, + banned INTEGER NOT NULL DEFAULT 0, + is_error INTEGER NOT NULL DEFAULT 0, + malicious_hits INTEGER NOT NULL DEFAULT 0, + extra TEXT, + created_at TEXT NOT NULL + ) + """, + "CREATE INDEX IF NOT EXISTS idx_requests_ip_ts ON requests (ip, ts)", + "CREATE INDEX IF NOT EXISTS idx_requests_source_ts ON requests (source, ts)", + "CREATE INDEX IF NOT EXISTS idx_requests_ts_epoch ON requests (ts_epoch)", + "CREATE INDEX IF NOT EXISTS idx_requests_path ON requests (path)", + "CREATE INDEX IF NOT EXISTS idx_requests_scan ON requests (scan_id)", + # ------------------------------------------------------------------ + # ip_stats — per-IP aggregate per scan + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS ip_stats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + scan_id INTEGER, + ip TEXT NOT NULL, + version INTEGER, + total INTEGER NOT NULL DEFAULT 0, + errors INTEGER NOT NULL DEFAULT 0, + c4xx INTEGER NOT NULL DEFAULT 0, + c5xx INTEGER NOT NULL DEFAULT 0, + denied INTEGER NOT NULL DEFAULT 0, + banned INTEGER NOT NULL DEFAULT 0, + first_ts TEXT, + last_ts TEXT, + malicious_hits INTEGER NOT NULL DEFAULT 0, + paths_json TEXT, + methods_json TEXT, + hosts_json TEXT, + uas_json TEXT, + users_json TEXT, + whitelist TEXT, + blocked INTEGER NOT NULL DEFAULT 0 + ) + """, + "CREATE UNIQUE INDEX IF NOT EXISTS idx_ip_stats_scan_ip ON ip_stats (scan_id, ip)", + # ------------------------------------------------------------------ + # scans + scans_sources — one row per analysis run + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS scans ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + status TEXT NOT NULL DEFAULT 'running', + full INTEGER NOT NULL DEFAULT 0, + dry_run INTEGER NOT NULL DEFAULT 0, + params_json TEXT, + result_json TEXT, + log_files INTEGER NOT NULL DEFAULT 0, + request_count INTEGER NOT NULL DEFAULT 0, + error TEXT + ) + """, + "CREATE INDEX IF NOT EXISTS idx_scans_started ON scans (started_at)", + """ + CREATE TABLE IF NOT EXISTS scans_sources ( + scan_id INTEGER NOT NULL, + source TEXT NOT NULL, + requests INTEGER NOT NULL DEFAULT 0, + files INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (scan_id, source) + ) + """, + # ------------------------------------------------------------------ + # blocklist — grow-only ledger + reasons + audit events + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS blocklist_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + cidr TEXT NOT NULL, + version INTEGER, + status TEXT NOT NULL DEFAULT 'active', + added_at TEXT NOT NULL, + removed_at TEXT, + added_by TEXT, + last_seen TEXT + ) + """, + """CREATE UNIQUE INDEX IF NOT EXISTS + idx_blocklist_active_cidr ON blocklist_entries (cidr) WHERE status='active'""", + """ + CREATE TABLE IF NOT EXISTS blocklist_reasons ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + entry_id INTEGER, + scan_id INTEGER, + ip TEXT NOT NULL, + reason TEXT, + pattern TEXT, + severity INTEGER, + sample_path TEXT, + hits INTEGER NOT NULL DEFAULT 0 + ) + """, + "CREATE INDEX IF NOT EXISTS idx_blereasons_ip ON blocklist_reasons (ip)", + """ + CREATE TABLE IF NOT EXISTS blocklist_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + action TEXT NOT NULL, + cidr TEXT NOT NULL, + actor TEXT, + fired_at TEXT NOT NULL, + detail TEXT + ) + """, + "CREATE INDEX IF NOT EXISTS idx_blevents_cidr ON blocklist_events (cidr)", + # ------------------------------------------------------------------ + # whitelist — current entries + source refresh tracking + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS whitelist_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + cidr TEXT NOT NULL, + version INTEGER, + source TEXT, + active INTEGER NOT NULL DEFAULT 1, + updated_at TEXT + ) + """, + "CREATE UNIQUE INDEX IF NOT EXISTS idx_wl_provider_cidr ON whitelist_entries (provider, cidr)", + """ + CREATE TABLE IF NOT EXISTS whitelist_sources ( + provider TEXT PRIMARY KEY, + source_type TEXT, + url TEXT, + file TEXT, + last_refresh_at TEXT, + entry_count INTEGER NOT NULL DEFAULT 0, + refresh_error TEXT + ) + """, + # ------------------------------------------------------------------ + # firewall — snapshots of address lists + rules + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS firewall_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + captured_at TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT 'snapshot', + trigger TEXT NOT NULL DEFAULT 'scan' + ) + """, + """ + CREATE TABLE IF NOT EXISTS firewall_address_list_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshot_id INTEGER, + network TEXT, + list TEXT, + address TEXT, + dynamic INTEGER NOT NULL DEFAULT 0, + timeout INTEGER + ) + """, + "CREATE INDEX IF NOT EXISTS idx_fwal_snapshot ON firewall_address_list_entries (snapshot_id)", + "CREATE INDEX IF NOT EXISTS idx_fwal_list_addr ON firewall_address_list_entries (list, address)", + """ + CREATE TABLE IF NOT EXISTS firewall_rule_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshot_id INTEGER, + network TEXT, + rule_index INTEGER, + chain TEXT, + action TEXT, + dst_list TEXT, + proto_port TEXT, + comment TEXT, + raw_json TEXT + ) + """, + "CREATE INDEX IF NOT EXISTS idx_fwr_snapshot ON firewall_rule_entries (snapshot_id)", + # ------------------------------------------------------------------ + # whois — structured cache + # ------------------------------------------------------------------ + """ + CREATE TABLE IF NOT EXISTS whois_info ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ip TEXT UNIQUE NOT NULL, + name TEXT, + cidr TEXT, + fetched_at TEXT NOT NULL + ) + """, + ], +} + + +def apply_migrations(conn: sqlite3.Connection) -> None: + """Apply all pending migrations, bumping ``user_version`` in lock-step. + + Safe to call on every process start: it only runs migrations whose version + is greater than the current ``PRAGMA user_version``. + """ + conn.execute("PRAGMA foreign_keys=ON") + current = conn.execute("PRAGMA user_version").fetchone()[0] + for version in range(current + 1, SCHEMA_VERSION + 1): + statements = MIGRATIONS.get(version) + if statements is None: + raise RuntimeError(f"missing migration for schema version {version}") + with conn: + for ddl in statements: + conn.execute(ddl) + conn.execute(f"PRAGMA user_version={version}") diff --git a/app/scan/__init__.py b/app/scan/__init__.py index 895cfd9..2a778ca 100644 --- a/app/scan/__init__.py +++ b/app/scan/__init__.py @@ -40,6 +40,22 @@ from ..mikrotik import split_cidr_versions from ..utils.whois import consolidate_by_subnet +def _sync_blocklist_to_db(connected: bool, desired_v4: list[str], + desired_v6: list[str], added_by: str | None = None) -> None: + """Best-effort sync of the consolidated blocklist targets into SQLite. + + Keep the DB ledger in sync regardless of whether a router sync happened so + analysis always sees the current policy.""" + 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) + except Exception as exc: # never break the scan for a DB issue + print(f"WARN: could not sync blocklist to db ({exc})") + + @dataclass class ScanParams: """Configuration for a consolidated scan. @@ -93,6 +109,24 @@ def run_scan(params: ScanParams) -> ScanResult: """ errors: list[str] = [] + # --- Step 0: open the SQLite store for this consolidated run ------------- + db_scan_id: int | None = None + try: + from ..db import get_connection + from ..db import repositories as repo + from ..utils.whitelists import sync_whitelists_to_db + conn = get_connection() + sync_whitelists_to_db() + params_map = { + "full": params.full, "dry_run": params.dry_run, + "update_firewall": params.update_firewall, + "quiet": params.quiet, "top": params.top, + } + db_scan_id = repo.start_scan(conn, "ids", params=params_map, + full=params.full, dry_run=params.dry_run) + except Exception as exc: + print(f"WARN: sqlite store unavailable ({exc})") + # --- Step 1-3: connect to MikroTik RouterOS and capture current state --- # If the router is unreachable we fall back to scan-only mode: still # generate reports and blocklists, but skip the firewall sync. @@ -134,7 +168,7 @@ def run_scan(params: ScanParams) -> ScanResult: print("=== Running copyparty scan ===") cp_exists = True try: - cp_analyzer.run() + cp_analyzer.run(db_scan_id=db_scan_id) except Exception as e: errors.append(f"copyparty scan failed: {e}") print(f"Error running copyparty scan: {e}") @@ -143,7 +177,7 @@ def run_scan(params: ScanParams) -> ScanResult: print("\n=== Running NPM scan ===") npm_exists = True try: - npm_analyzer.run() + npm_analyzer.run(db_scan_id=db_scan_id) except Exception as e: errors.append(f"NPM scan failed: {e}") print(f"Error running NPM scan: {e}") @@ -155,6 +189,14 @@ def run_scan(params: ScanParams) -> ScanResult: print(f"\n{message}") if firewall is not None: firewall.disconnect() + if db_scan_id is not None: + try: + from ..db import get_connection + from ..db import repositories as repo + repo.finish_scan(get_connection(), db_scan_id, status="error", + error="; ".join(errors)) + except Exception: + pass return ScanResult(ok=False, errors=errors) # Consolidate block candidates @@ -240,6 +282,35 @@ def run_scan(params: ScanParams) -> ScanResult: f"(keep {REPORT_RETENTION_DAYS}d / " f"max {REPORT_RETENTION_MAX_FILES} per dir)") + # --- Step 9: persist consolidated results to SQLite --------------------- + # Sync the blocklist ledger and close out the scan record regardless of + # whether the router sync was attempted. + _sync_blocklist_to_db( + firewall is not None, blocked_ipv4, blocked_ipv6, + added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}")) + if db_scan_id is not None: + try: + from ..db import get_connection + from ..db import repositories as repo + conn = get_connection() + result_payload = { + "ok": True, + "cp_candidates": cp_candidates, + "npm_candidates": npm_candidates, + "blocked_ips": len(final_blocked_ips), + "blocked_targets": len(final_blocked_targets), + "firewall_connected": firewall is not None, + } + repo.finish_scan( + conn, db_scan_id, status="done", + result=result_payload, log_files=0, + request_count=(cp_analyzer and len(cp_analyzer.requests) or 0) + + (npm_analyzer and len(npm_analyzer.requests) or 0), + error="; ".join(errors) or None, + ) + except Exception as exc: + print(f"WARN: could not finalize scan in db ({exc})") + return ScanResult( ok=True, cp_exists=cp_exists, diff --git a/app/templates/analysis.html b/app/templates/analysis.html new file mode 100644 index 0000000..41428f6 --- /dev/null +++ b/app/templates/analysis.html @@ -0,0 +1,191 @@ +{% extends "base.html" %} {% block title %}Log Analysis — Intrusion Detection{% +endblock %} {% block content %} +

Log Analysis

+

+ Detailed analysis of persisted log data from the SQLite store. Select an IP + to see its full request history across all log sources. +

+ +
+
+

Top IPs

+ +
+ + + + + + + + + + + +
IPRequestsErrorsMalicious
+
+ +
+
+

Cross-source IPs

+ Seen in more than one log source +
+ + + + + + + + + + +
IPSourcesSource listTotal
+
+ + +{% endblock %} {% block scripts %} + +{% endblock %} diff --git a/app/templates/base.html b/app/templates/base.html index d8fc8ea..fad8223 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -40,6 +40,11 @@ class="{{ 'active' if request.path == '/graph' }}" >Graph + Analysis
{% block content %}{% endblock %}
diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html index 2cb35ee..d2fb8a9 100644 --- a/app/templates/dashboard.html +++ b/app/templates/dashboard.html @@ -57,6 +57,13 @@ endblock %} {% block content %} s.firewall_snapshot && s.firewall_snapshot.exists ? "ok" : "muted", }, ]; + if (s.analysis && s.analysis.available) { + cards.push({ + label: "SQLite stores", + value: `${s.analysis.request_count ?? 0} req / ${s.analysis.scan_count ?? 0} scans`, + tone: "ok", + }); + } const box = document.getElementById("status-cards"); box.innerHTML = ""; for (const c of cards) { diff --git a/app/utils/constants.py b/app/utils/constants.py index 8c33eb5..edce68e 100644 --- a/app/utils/constants.py +++ b/app/utils/constants.py @@ -53,6 +53,11 @@ MIKROTIK_DATA_DIR = os.path.join(DATA_DIR, "mikrotik") WHOIS_CACHE_PATH = os.path.join(DATA_DIR, "whois_cache.json") +# Local SQLite database (see .github/instructions/DATABASE.md). The default +# lives under data/ and holds log records, scan metadata, blocklist/whitelist +# and firewall history. Override with IDS_DB_PATH (used by app/db). +DATA_DB_PATH = os.environ.get("IDS_DB_PATH", os.path.join(DATA_DIR, "ids.db")) + # Incremental-scan state: absolute log path -> last processed byte offset SCAN_STATE_PATH = os.path.join(DATA_DIR, "scan_state.json") diff --git a/app/utils/whitelists.py b/app/utils/whitelists.py index a55a15c..7ed212d 100644 --- a/app/utils/whitelists.py +++ b/app/utils/whitelists.py @@ -275,3 +275,20 @@ def refresh_whitelists(force: bool = False) -> bool: update_whitelists() _refreshed = True return True + + +def sync_whitelists_to_db() -> None: + """Persist the current in-memory ``WHITELISTS`` into the SQLite store. + + Best-effort (never raises): keeps the DB whitelist tables consistent with + the in-memory/provider state so lookups and analysis can rely on them. + """ + try: + from ..db import get_connection + from ..db import repositories as repo + conn = get_connection() + for provider, cidrs in WHITELISTS.items(): + repo.set_whitelist(conn, provider, cidrs, + source="builtin/refresh") + except Exception as exc: # pragma: no cover - defensive + print(f"WARN: could not sync whitelists to db ({exc})") diff --git a/app/web/__init__.py b/app/web/__init__.py index 64a4cfe..2eba8a9 100644 --- a/app/web/__init__.py +++ b/app/web/__init__.py @@ -131,6 +131,41 @@ def _sync_firewall(dry_run: bool): return _with_router(_sync) +def _analysis_helpers(): + """SQLite-backed log-analysis query helpers (None when DB unavailable). + + Functions take the shared DB connection each call so routes stay stateless + across the background scan thread. + """ + try: + from ..db import get_connection + from ..db import repositories as repo + except Exception: + return None + + def top_ips(since: str | None = None, limit: int = 50) -> list: + return repo.top_ips(get_connection(), since=since, limit=limit) + + def ip_timeline(ip: str) -> list: + return repo.ip_timeline(get_connection(), ip=ip) + + def cross_source(limit: int = 100) -> list: + return repo.correlate_across_sources(get_connection(), limit=limit) + + def conn(): + return get_connection() + + def request_count(conn) -> int: + return repo.count_requests(conn) + + def scan_count(conn) -> int: + return repo.count_scans(conn) + + return {"top_ips": top_ips, "ip_timeline": ip_timeline, + "cross_source": cross_source, "conn": conn, + "request_count": request_count, "scan_count": scan_count} + + # --------------------------------------------------------------------------- # app factory # --------------------------------------------------------------------------- @@ -155,6 +190,7 @@ def create_app() -> Flask: "_params_from_request": _params_from_request, "_with_router": _with_router, "_sync_firewall": _sync_firewall, + "_analysis": _analysis_helpers(), } register_routes(app, REPORT_SOURCES, SCAN_KINDS, web_scan, helpers) return app diff --git a/app/web/routes.py b/app/web/routes.py index aa12bbc..14f06b8 100644 --- a/app/web/routes.py +++ b/app/web/routes.py @@ -32,6 +32,40 @@ def register_routes(app: Flask, _with_router = helpers["_with_router"] _sync_firewall = helpers["_sync_firewall"] + # --- log analysis (SQLite-backed) -------------------------------------- + + _analysis = helpers.get("_analysis") + + @app.get("/analysis") + def analysis_page(): + return render_template("analysis.html") + + @app.get("/api/analysis/top-ips") + def analysis_top_ips(): + if _analysis is None: + return jsonify({"error": "analysis unavailable"}), 503 + since = (request.args.get("since") or "").strip() or None + limit = request.args.get("limit", 50, type=int) + return jsonify({"top_ips": _analysis["top_ips"](since=since, limit=limit)}) + + @app.get("/api/analysis/ip/") + def analysis_ip(ip): + if _analysis is None: + return jsonify({"error": "analysis unavailable"}), 503 + return jsonify({"ip": ip, + "timeline": _analysis["ip_timeline"](ip)}) + + @app.get("/api/analysis/cross-source") + def analysis_cross_source(): + if _analysis is None: + return jsonify({"error": "analysis unavailable"}), 503 + limit = request.args.get("limit", 100, type=int) + return jsonify({"ips": _analysis["cross_source"](limit=limit)}) + + @app.get("/api/analysis/manifest") + def analysis_manifest(): + return jsonify({"ok": _analysis is not None}) + # --- pages ------------------------------------------------------------- @app.get("/") @@ -68,6 +102,17 @@ def register_routes(app: Flask, def status(): v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt") v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt") + analysis = {} + if _analysis is not None: + try: + conn = _analysis["conn"]() + analysis = { + "available": True, + "request_count": _analysis["request_count"](conn), + "scan_count": _analysis["scan_count"](conn), + } + except Exception: + analysis = {"available": False} return jsonify({ "host": WEB_HOST, "port": WEB_PORT, @@ -79,6 +124,7 @@ def register_routes(app: Flask, "exists": os.path.isdir(MIKROTIK_DATA_DIR), "path": MIKROTIK_DATA_DIR, }, + "analysis": analysis, }) # --- scans -------------------------------------------------------------