#!/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 = 2 #: 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 ) """ ], 2: [ """ CREATE TABLE IF NOT EXISTS report_artifacts ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, scan_id INTEGER, report_type TEXT NOT NULL, generated_at TEXT NOT NULL, schema_version INTEGER NOT NULL DEFAULT 1, payload_json TEXT NOT NULL, original_path TEXT NOT NULL, content_sha256 TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE (original_path, content_sha256) ) """, "CREATE INDEX IF NOT EXISTS idx_report_artifacts_source_time " "ON report_artifacts (source, generated_at DESC, id DESC)", "CREATE INDEX IF NOT EXISTS idx_report_artifacts_type_time " "ON report_artifacts (report_type, generated_at DESC, id DESC)", """ CREATE TABLE IF NOT EXISTS requests ( 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}")