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.
This commit is contained in:
zwitschi committed 2026-10-24 19:48:20 +02:00
1 parent 0137264fe4
commit 6036fa8b1e
17 files changed
+1292 -5

No files matched your search

+6
View File
@@ -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.
+5
View File
@@ -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/
+9 -2
View File
@@ -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**
+62 -1
View File
@@ -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}")
+15
View File
@@ -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
+7
View File
@@ -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
+62
View File
@@ -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
+478
View File
@@ -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()))
+268
View File
@@ -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}")
+73 -2
View File
@@ -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,
+191
View File
@@ -0,0 +1,191 @@
{% extends "base.html" %} {% block title %}Log Analysis — Intrusion Detection{%
endblock %} {% block content %}
<h1>Log Analysis</h1>
<p class="muted">
Detailed analysis of persisted log data from the SQLite store. Select an IP
to see its full request history across all log sources.
</p>
<div class="panel">
<div class="panel-head">
<h2>Top IPs</h2>
<span class="muted" id="top-disclaimer"></span>
</div>
<table class="table">
<thead>
<tr>
<th>IP</th>
<th class="num">Requests</th>
<th class="num">Errors</th>
<th class="num">Malicious</th>
<th></th>
</tr>
</thead>
<tbody id="top-ips"></tbody>
</table>
</div>
<div class="panel">
<div class="panel-head">
<h2>Cross-source IPs</h2>
<span class="muted">Seen in more than one log source</span>
</div>
<table class="table">
<thead>
<tr>
<th>IP</th>
<th class="num">Sources</th>
<th>Source list</th>
<th class="num">Total</th>
</tr>
</thead>
<tbody id="cross-source"></tbody>
</table>
</div>
<div class="panel" id="ip-panel" hidden>
<div class="panel-head">
<h2 id="ip-title">Request history</h2>
<a class="btn ghost" href="#" id="ip-close">✕ Close</a>
</div>
<table class="table">
<thead>
<tr>
<th>Time</th>
<th>Source</th>
<th>Method</th>
<th>Path</th>
<th class="num">Status</th>
<th>UA</th>
</tr>
</thead>
<tbody id="ip-timeline"></tbody>
</table>
</div>
{% endblock %} {% block scripts %}
<script>
function showIp(ip) {
const panel = document.getElementById("ip-panel");
const body = document.getElementById("ip-timeline");
document.getElementById("ip-title").textContent = `Request history — ${ip}`;
panel.hidden = false;
body.innerHTML =
'<tr><td colspan="6" class="muted">Loading…</td></tr>';
api(`/api/analysis/ip/${encodeURIComponent(ip)}`)
.then((data) => {
body.innerHTML = "";
const tl = data.timeline || [];
if (!tl.length) {
body.innerHTML =
'<tr><td colspan="6" class="muted">No persisted requests for this IP.</td></tr>';
return;
}
for (const r of tl) {
const status = r.status ?? "—";
const row = el(
"tr",
el("td", {
class: "mono",
text: r.ts || "—",
}),
el("td", { text: r.source }),
el("td", { text: r.method || "—" }),
el("td", { class: "mono", text: r.path || "—" }),
el(
"td",
{
class: "num",
text: status,
style: status >= 400 ? "color:var(--danger)" : "",
},
),
el("td", { class: "muted", text: r.ua || "—" }),
);
if (r.is_error) row.classList.add("row-error");
body.append(row);
}
})
.catch((e) => toast(e.message, "error"));
panel.scrollIntoView({ behavior: "smooth", block: "start" });
}
async function loadTop() {
const body = document.getElementById("top-ips");
body.innerHTML = "";
try {
const data = await api("/api/analysis/top-ips?limit=50");
body.innerHTML = "";
const ips = data.top_ips || [];
document.getElementById("top-disclaimer").textContent =
`showing top ${ips.length} IPs`;
if (!ips.length) {
body.innerHTML =
'<tr><td colspan="5" class="muted">No persisted requests yet — run a scan to populate the SQLite store.</td></tr>';
return;
}
for (const r of ips) {
body.append(
el(
"tr",
el("td", { class: "mono", text: r.ip }),
el("td", { class: "num", text: r.total }),
el("td", { class: "num", text: r.errors }),
el("td", { class: "num", text: r.malicious_hits }),
el(
"td",
el(
"button",
{ class: "btn ghost", onclick: () => showIp(r.ip) },
"Detail",
),
),
),
);
}
} catch (e) {
body.innerHTML =
'<tr><td colspan="5" class="muted b">SQLite analysis unavailable.</td></tr>';
toast(e.message, "error");
}
}
async function loadCrossSource() {
const body = document.getElementById("cross-source");
body.innerHTML = "";
try {
const data = await api("/api/analysis/cross-source?limit=100");
const ips = data.ips || [];
if (!ips.length) {
body.innerHTML =
'<tr><td colspan="4" class="muted">No IPs seen across multiple sources yet.</td></tr>';
return;
}
for (const r of ips) {
body.append(
el(
"tr",
el("td", { class: "mono", text: r.ip }),
el("td", { class: "num", text: r.sources }),
el("td", { text: r.source_list }),
el("td", { class: "num", text: r.total }),
),
);
}
} catch (e) {
body.innerHTML =
'<tr><td colspan="4" class="muted b">SQLite analysis unavailable.</td></tr>';
toast(e.message, "error");
}
}
document
.getElementById("ip-close")
.addEventListener("click", (e) => {
e.preventDefault();
document.getElementById("ip-panel").hidden = true;
});
loadTop();
loadCrossSource();
</script>
{% endblock %}
+5
View File
@@ -40,6 +40,11 @@
class="{{ 'active' if request.path == '/graph' }}"
>Graph</a
>
<a
href="/analysis"
class="{{ 'active' if request.path == '/analysis' }}"
>Analysis</a
>
</nav>
</header>
<main class="container">{% block content %}{% endblock %}</main>
+7
View File
@@ -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) {
+5
View File
@@ -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")
+17
View File
@@ -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})")
+36
View File
@@ -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
+46
View File
@@ -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/<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 -------------------------------------------------------------