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
+
+
+
+
+
+ | IP |
+ Requests |
+ Errors |
+ Malicious |
+ |
+
+
+
+
+
+
+
Cross-source IPs
+ Seen in more than one log source
+
+
+
+
+ | IP |
+ Sources |
+ Source list |
+ Total |
+
+
+
+
+
+
+
+
+
+ | Time |
+ Source |
+ Method |
+ Path |
+ Status |
+ UA |
+
+
+
+
+