diff --git a/app.py b/app.py index f073bcb..35f6c6f 100644 --- a/app.py +++ b/app.py @@ -40,12 +40,22 @@ def _run_scan(argv: list[str]) -> None: scan_main(argv) +def _run_analyze(argv: list[str]) -> None: + from app.analyze import main as analyze_main + analyze_main(argv) + + def main() -> None: # `python app.py scan ...` runs the consolidated CLI scan. if len(sys.argv) > 1 and sys.argv[1] == "scan": _run_scan(sys.argv[2:]) return + # `python app.py analyze ...` prints a detailed report from the SQLite store. + if len(sys.argv) > 1 and sys.argv[1] == "analyze": + _run_analyze(sys.argv[2:]) + return + # Default: start the web UI (parse_known_args tolerates stray flags). args, _ = _web_parser().parse_known_args() _run_web(args.web_host, args.web_port) diff --git a/app/analyze.py b/app/analyze.py new file mode 100644 index 0000000..a083161 --- /dev/null +++ b/app/analyze.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""CLI detailed log-analysis report (L002). + +``python app.py analyze [--ip IP] [--since ISO] [--top N] [--json]`` reads the +persisted SQLite store (request history, per-IP aggregates, blocklist ledger) +and prints a detailed, interactive analysis. It complements the web UI's +``/analysis`` page for use on a headless server or in scripts. + +No scan is run here: this only queries what ``scan`` has already persisted. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime + + +def _fmt_ts(epoch: float | None, default: str = "-") -> str: + if not epoch: + return default + try: + return datetime.fromtimestamp(epoch).isoformat(timespec="seconds") + except (ValueError, OverflowError, OSError): + return default + + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="python app.py analyze", + description="Detailed log-analysis report from the SQLite store.") + p.add_argument("--ip", metavar="IP", default=None, + help="focus the report on a single IP (drill-down)") + p.add_argument("--since", metavar="ISO", default=None, + help="only analyse requests at/after this ISO timestamp") + p.add_argument("--top", type=int, default=20, + help="rows in the top-IP / threat tables (default: 20)") + p.add_argument("--json", action="store_true", + help="emit raw JSON instead of a human-readable report") + return p + + +def _report(conn, opts) -> dict: + from .db import repositories as repo + + data: dict = {} + data["scans"] = repo.latest_scans(conn, limit=10) + data["totals"] = { + "requests": repo.count_requests(conn), + "scans": repo.count_scans(conn), + } + + if opts.ip: + data["ip"] = { + "ip": opts.ip, + "timeline": repo.ip_timeline(conn, ip=opts.ip), + "active_blocklist": [e for e in repo.active_blocklist(conn) + if e["cidr"].startswith(opts.ip)], + } + else: + data["top_ips"] = repo.top_ips(conn, since=opts.since, limit=opts.top) + data["cross_source"] = repo.correlate_across_sources( + conn, limit=opts.top) + data["blocklist"] = repo.active_blocklist(conn) + data["recent_errors"] = repo.query_requests( + conn, since=opts.since, limit=opts.top) + return data + + +def _render_text(data: dict, opts) -> str: + lines = ["Detailed log analysis", "=" * 40, ""] + totals = data["totals"] + lines.append(f"SQLite store: {totals['requests']} requests across " + f"{totals['scans']} scan(s)") + if data["scans"]: + lines.append("Latest scans:") + for s in data["scans"]: + lines.append(f" #{s['id']:<4} {s['kind']:<8} {s['status']:<7} " + f"{s.get('started') or s.get('started_at') or '-'}") + lines.append("") + + if "ip" in data: + ip = data["ip"] + lines.append(f"== Request history for {ip['ip']} ==") + bl = ip["active_blocklist"] + if bl: + lines.append(f" (source: active blocklist entries " + f"{', '.join(e['cidr'] for e in bl)})") + lines.append(f"{'time':<24} {'src':<9} {'method':<8} {'status':<6} path") + for r in ip["timeline"][:opts.top]: + lines.append( + f"{r.get('ts') or '-':<24} {str(r.get('source')):<9} " + f"{str(r.get('method') or '-'):<8} " + f"{str(r.get('status') or '-'):<6} {r.get('path') or '-'}") + else: + lines.append("== Top IPs ==") + lines.append( + f"{'ip':<18} {'ver':<3} {'reqs':<6} {'err':<5} {'mal':<5} " + f"{'first':<22} last") + for r in data["top_ips"]: + lines.append( + f"{str(r['ip']):<18} {r.get('version') or '-':<3} " + f"{r['total']:<6} {r['errors']:<5} {r['malicious_hits']:<5} " + f"{_fmt_ts(r['first_epoch']):<22} {_fmt_ts(r['last_epoch'])}") + lines.append("") + + lines.append("== Cross-source IPs ==") + if data["cross_source"]: + for r in data["cross_source"]: + lines.append( + f" {r['ip']:<18} sources={r['sources']} " + f"({r['source_list']}) total={r['total']}") + else: + lines.append(" (none — no IP seen in more than one log source)") + + lines.append("") + lines.append("== Active blocklist (ledger) ==") + if data["blocklist"]: + for e in data["blocklist"]: + lines.append(f" {e['cidr']:<22} added={e['added_at']}") + else: + lines.append(" (empty)") + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> None: + opts = _build_parser().parse_args(argv) + try: + from .db import get_connection + conn = get_connection() + data = _report(conn, opts) + except Exception as exc: + print(f"error: could not read SQLite store ({exc}); " + f"run a scan first so data exists.", file=sys.stderr) + sys.exit(1) + + if opts.json: + print(json.dumps(data, indent=2, sort_keys=True, default=str)) + else: + print(_render_text(data, opts)) + + +if __name__ == "__main__": + main()