691 lines
27 KiB
Python
691 lines
27 KiB
Python
#!/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
|
|
import base64
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
import ipaddress
|
|
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])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Report artifacts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def insert_report_artifact(conn: sqlite3.Connection, *, source: str,
|
|
report_type: str, generated_at: str,
|
|
payload: Any, original_path: str,
|
|
content_sha256: str, scan_id: int | None = None,
|
|
schema_version: int = 1) -> int:
|
|
"""Insert a structured report artifact, idempotent by path and hash."""
|
|
with transaction(conn):
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO report_artifacts "
|
|
"(source, scan_id, report_type, generated_at, schema_version, "
|
|
"payload_json, original_path, content_sha256, created_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(source, scan_id, report_type, generated_at, schema_version,
|
|
json.dumps(payload, ensure_ascii=False), original_path,
|
|
content_sha256, utcnow()),
|
|
)
|
|
row = conn.execute(
|
|
"SELECT id FROM report_artifacts "
|
|
"WHERE original_path=? AND content_sha256=?",
|
|
(original_path, content_sha256),
|
|
).fetchone()
|
|
return int(row["id"])
|
|
|
|
|
|
def list_report_artifacts(conn: sqlite3.Connection, *,
|
|
source: str | None = None,
|
|
report_type: str | None = None,
|
|
since: str | None = None,
|
|
until: str | None = None,
|
|
limit: int = 50,
|
|
cursor: str | None = None) -> dict[str, Any]:
|
|
"""List report metadata using bounded keyset pagination."""
|
|
limit = min(max(int(limit), 1), 200)
|
|
clauses: list[str] = []
|
|
args: list[Any] = []
|
|
if source:
|
|
clauses.append("source=?")
|
|
args.append(source)
|
|
if report_type:
|
|
clauses.append("report_type=?")
|
|
args.append(report_type)
|
|
if since:
|
|
clauses.append("generated_at>=?")
|
|
args.append(since)
|
|
if until:
|
|
clauses.append("generated_at<=?")
|
|
args.append(until)
|
|
if cursor:
|
|
generated_at, artifact_id = _decode_report_cursor(cursor)
|
|
clauses.append("(generated_at<? OR (generated_at=? AND id<?))")
|
|
args.extend((generated_at, generated_at, artifact_id))
|
|
|
|
sql = (
|
|
"SELECT id, source, scan_id, report_type, generated_at, "
|
|
"schema_version, original_path, content_sha256 "
|
|
"FROM report_artifacts"
|
|
)
|
|
if clauses:
|
|
sql += " WHERE " + " AND ".join(clauses)
|
|
sql += " ORDER BY generated_at DESC, id DESC LIMIT ?"
|
|
rows = [dict(row) for row in conn.execute(sql, (*args, limit + 1))]
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
next_cursor = (
|
|
_encode_report_cursor(rows[-1]["generated_at"], rows[-1]["id"])
|
|
if has_more and rows else None
|
|
)
|
|
return {"items": rows, "next_cursor": next_cursor}
|
|
|
|
|
|
def get_report_artifact(conn: sqlite3.Connection,
|
|
artifact_id: int) -> dict[str, Any] | None:
|
|
"""Fetch one artifact and decode its structured payload."""
|
|
row = conn.execute(
|
|
"SELECT id, source, scan_id, report_type, generated_at, "
|
|
"schema_version, payload_json, original_path, content_sha256, created_at "
|
|
"FROM report_artifacts WHERE id=?", (artifact_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
artifact = dict(row)
|
|
artifact["payload"] = json.loads(artifact.pop("payload_json"))
|
|
return artifact
|
|
|
|
|
|
def _encode_report_cursor(generated_at: str, artifact_id: int) -> str:
|
|
raw = f"{generated_at}\n{artifact_id}".encode("utf-8")
|
|
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _decode_report_cursor(cursor: str) -> tuple[str, int]:
|
|
try:
|
|
padded = cursor + "=" * (-len(cursor) % 4)
|
|
generated_at, artifact_id = base64.urlsafe_b64decode(
|
|
padded.encode("ascii")).decode("utf-8").split("\n", 1)
|
|
return generated_at, int(artifact_id)
|
|
except (ValueError, UnicodeError) as exc:
|
|
raise ValueError("invalid report cursor") from exc
|
|
|
|
|
|
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 record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
|
|
added_by: str | None = None, reasons=None) -> dict[str, Any]:
|
|
"""Grow-only record: add any ``cidrs`` not already active to the ledger.
|
|
|
|
Unlike :func:`sync_blocklist`, this **never removes** entries. It mirrors
|
|
the router's cumulative model: once an IP/subnet has ever been blocked it
|
|
stays active even after it leaves the scan window (e.g. NPM ~30-day log
|
|
retention), exactly like the old git-history backfill. Returns the added
|
|
delta and the resulting active count.
|
|
"""
|
|
cidr_set = set(cidrs)
|
|
before = {r["cidr"] for r in active_blocklist(conn)}
|
|
to_add = sorted(cidr_set - before)
|
|
now = utcnow()
|
|
with transaction(conn):
|
|
for cidr in to_add:
|
|
_block_entry_inner(conn, cidr, added_by=added_by, last_seen=now,
|
|
reasons=reasons)
|
|
# Refresh last_seen on entries that are still desired so the ledger
|
|
# reflects recency for existing rows too.
|
|
conn.executemany(
|
|
"UPDATE blocklist_entries SET last_seen=? WHERE cidr=? AND status='active'",
|
|
[(now, cidr) for cidr in cidr_set.intersection(before)])
|
|
return {"added": to_add, "active": len(before) + len(to_add)}
|
|
|
|
|
|
def prune_old_single_ip_blocks(conn: sqlite3.Connection, *,
|
|
older_than_days: int = 30,
|
|
actor: str | None = "retention") -> list[str]:
|
|
"""Remove active single-host entries older than ``older_than_days``.
|
|
|
|
Keeps CIDR ranges/subnets intact. Host entries are detected via
|
|
``ipaddress.ip_network(..., strict=False)`` and only removed when the
|
|
network is exactly one address (/32 for IPv4, /128 for IPv6).
|
|
"""
|
|
cutoff = datetime.now(timezone.utc) - \
|
|
timedelta(days=max(0, older_than_days))
|
|
rows = conn.execute(
|
|
"SELECT cidr, added_at, last_seen FROM blocklist_entries "
|
|
"WHERE status='active'").fetchall()
|
|
|
|
to_remove: list[str] = []
|
|
for row in rows:
|
|
cidr = str(row["cidr"])
|
|
if not _is_single_host_cidr(cidr):
|
|
continue
|
|
seen = _parse_iso_ts(
|
|
row["last_seen"]) or _parse_iso_ts(row["added_at"])
|
|
if seen is None:
|
|
continue
|
|
if seen < cutoff:
|
|
to_remove.append(cidr)
|
|
|
|
if not to_remove:
|
|
return []
|
|
|
|
now = utcnow()
|
|
with transaction(conn):
|
|
conn.executemany(
|
|
"UPDATE blocklist_entries SET status='removed', removed_at=? "
|
|
"WHERE cidr=? AND status='active'",
|
|
[(now, cidr) for cidr in to_remove],
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO blocklist_events (action, cidr, actor, fired_at, detail) "
|
|
"VALUES ('remove', ?, ?, ?, ?)",
|
|
[(cidr, actor, now, f"single-ip-retention>{older_than_days}d")
|
|
for cidr in to_remove],
|
|
)
|
|
return sorted(to_remove)
|
|
|
|
|
|
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
|
|
|
|
|
|
def _is_single_host_cidr(cidr: str) -> bool:
|
|
try:
|
|
net = ipaddress.ip_network(cidr, strict=False)
|
|
return net.num_addresses == 1
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _parse_iso_ts(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
def recent_firewall_snapshots(conn: sqlite3.Connection,
|
|
limit: int = 10) -> list[dict[str, Any]]:
|
|
"""Most recent firewall snapshots with entry counts (F001)."""
|
|
rows = conn.execute(
|
|
"SELECT fs.id, fs.captured_at, fs.kind, fs.trigger, "
|
|
"(SELECT COUNT(*) FROM firewall_address_list_entries e "
|
|
" WHERE e.snapshot_id = fs.id) AS addr_entries, "
|
|
"(SELECT COUNT(*) FROM firewall_rule_entries r "
|
|
" WHERE r.snapshot_id = fs.id) AS rule_entries "
|
|
"FROM firewall_snapshots fs ORDER BY fs.id DESC LIMIT ?",
|
|
(limit,)).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHOIS cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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()))
|