intrusion-detection/app/analyze.py

#!/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()