SmartestHome/digest-engine/ingest/opnsense_ids.py

"""OPNsense IDS ingestion — a Suricata alert summary for the preceding digest window.

READ-ONLY INVARIANT: this module touches exactly two OPNsense API endpoints —
`GET /api/ids/service/status` and `POST /api/ids/service/query_alerts` — and no
others. `query_alerts` is a POST because OPNsense routes all filtered queries
that way, not because it changes anything: it runs
`/usr/local/opnsense/scripts/suricata/queryAlertLog.py`, which opens
`/var/log/suricata/eve.json` and reads it backwards. Nothing here starts, stops,
reconfigures, reloads or clears anything on the firewall, and nothing here may
ever be changed to, per docs/project-plan.md Phase 12 step 8. The endpoints that
*would* mutate state (`/ids/service/{start,stop,restart,reconfigure,reload_rules,
update_rules,drop_alert_log}`) are named here only so it is obvious they are
deliberately not used.

Sourcing, all verified against OPNsense source 2026-07-28:

  Suricata is **core**, not a plugin. The premise that this needs `os-suricata`
  installed is wrong: the IDS module ships in opnsense/core
  (src/opnsense/mvc/app/controllers/OPNsense/IDS/) and the GUI lives at
  Services -> Intrusion Detection on a stock install. The only IDS-related
  plugins in opnsense/plugins are ruleset *content* packages
  (os-intrusion-detection-content-et-pro and friends); ET Open and the abuse.ch
  lists need no plugin at all.

  Alerts are queryable over the API. `POST /api/ids/service/query_alerts` takes
  `rowCount`, `current` (1-based page), `searchPhrase` and `fileid`, and returns
  `{"rows": [...], "total": n, "rowCount": n, "current": n}`. SSH or file access
  to /var/log/suricata/ is NOT required, which is why this is a pull over HTTPS
  like every other ingestion module here.

  There is **no server-side time filter**. `searchPhrase` is a substring match
  against the signature, action, source IP and destination IP only. Rows come
  back newest-first (the backend uses a reverse log reader), so the digest window
  is applied client-side by paging until a row falls out of it. That is also why
  `max_alerts_scanned` exists: a noisy WAN interface can produce more alerts in
  six hours than is sane to page through or to put in an LLM context.

  **Severity is not available.** queryAlertLog.py flattens each eve.json record
  down to `alert` (the signature text), `alert_sid` and `alert_action` before it
  returns, discarding `alert.severity` and `alert.category`. `get_alert_info`
  goes through the same flattening, so it does not help. Recovering severity
  would mean reading eve.json directly over SSH, which is a bigger access grant
  than this is worth. Alerts are therefore ranked by count, and the prompt is
  told it cannot see severity.

  Packet capture is deliberately not triggered from here. OPNsense does expose
  Interfaces: Diagnostics: Packet Capture over the API
  (`/api/diagnostics/packet_capture/...`), but `set`, `start`, `stop` and
  `remove` are all POSTs that write a job file to /tmp/captures and spawn
  tcpdump. Starting a capture is a write action on the firewall and is barred by
  the invariant above, quite apart from it being the wrong design — see the
  intrusion-detection section of digest-engine/README.md for how to keep a
  rotating local capture on OPNsense itself and merely *point* at it from the
  digest via `packet_capture_reference`.

Configuration is a JSON file, not env, because the user asked for `IDSconf.json`
specifically. Path comes from OPNSENSE_IDS_CONF_PATH (default /data/IDSconf.json)
and the file is git-ignored; IDSconf.json.example is the committed template.
"""

import ipaddress
import json
import logging
import math
import os
from collections import Counter
from datetime import datetime, timedelta, timezone

import requests

LOG = logging.getLogger(__name__)

DEFAULT_CONF_PATH = "/data/IDSconf.json"

STATUS_PATH = "/api/ids/service/status"
QUERY_ALERTS_PATH = "/api/ids/service/query_alerts"

HTTP_TIMEOUT = 45
ROWS_PER_PAGE = 500

DEFAULT_MAX_ALERTS_SCANNED = 5000
DEFAULT_TOP_SIGNATURES = 8
DEFAULT_TOP_HOSTS = 5


def _load_config(path):
    try:
        with open(path, encoding="utf-8") as handle:
            config = json.load(handle)
    except FileNotFoundError:
        LOG.warning("opnsense_ids: %s not found, skipping (copy IDSconf.json.example there)", path)
        return None
    except Exception:
        LOG.warning("opnsense_ids: %s could not be parsed, skipping", path, exc_info=True)
        return None

    missing = [key for key in ("base_url", "api_key", "api_secret") if not str(config.get(key, "")).strip()]
    if missing:
        LOG.warning("opnsense_ids: %s is missing %s, skipping", path, ", ".join(missing))
        return None

    return config


def _parse_timestamp(raw):
    if not raw:
        return None
    try:
        parsed = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
    except ValueError:
        return None
    # Suricata always writes an offset, but a naive value would raise on the
    # comparison against the cutoff and lose the whole run over one bad line.
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)


def _service_status(session, base_url, verify):
    response = session.get(base_url + STATUS_PATH, timeout=HTTP_TIMEOUT, verify=verify)
    response.raise_for_status()
    return response.json().get("status", "unknown")


def _collect_alerts(session, base_url, verify, cutoff, max_pages):
    rows = []

    for page in range(1, max_pages + 1):
        response = session.post(
            base_url + QUERY_ALERTS_PATH,
            data={"rowCount": ROWS_PER_PAGE, "current": page},
            timeout=HTTP_TIMEOUT,
            verify=verify,
        )
        response.raise_for_status()
        page_rows = response.json().get("rows") or []

        if not page_rows:
            # An empty first page means the log holds no alerts at all. An empty
            # later page means eve.json ran out before the window did, i.e. it
            # rotated mid-window and older alerts are in a file this does not read.
            return rows, bool(rows)

        for row in page_rows:
            timestamp = _parse_timestamp(row.get("timestamp"))
            if timestamp is None:
                continue
            if timestamp < cutoff:
                return rows, False
            rows.append(row)

        if len(page_rows) < ROWS_PER_PAGE:
            return rows, True

    return rows, True


def _host_role(ip):
    # is_global rather than "not is_private" so CGNAT and the reserved ranges land
    # on the household side of the split rather than being reported as the internet.
    try:
        return "remote" if ipaddress.ip_address(ip).is_global else "local"
    except ValueError:
        return None


def _summarise(rows, truncated, status, config, window_start, window_end, lookback_hours):
    actions = Counter()
    interfaces = Counter()
    local_hosts = Counter()
    remote_hosts = Counter()
    signatures = {}

    for row in rows:
        action = row.get("alert_action") or "unknown"
        actions[action] += 1

        interface = row.get("in_iface")
        if interface:
            interfaces[interface] += 1

        for key in ("src_ip", "dest_ip"):
            ip = row.get(key)
            if not ip:
                continue
            role = _host_role(ip)
            if role == "local":
                local_hosts[ip] += 1
            elif role == "remote":
                remote_hosts[ip] += 1

        sid = row.get("alert_sid")
        entry = signatures.setdefault(
            sid,
            {
                "sid": sid,
                "signature": row.get("alert"),
                "count": 0,
                "actions": Counter(),
                "sources": Counter(),
                "destinations": Counter(),
                "last_seen": row.get("timestamp"),
            },
        )
        entry["count"] += 1
        entry["actions"][action] += 1
        if row.get("src_ip"):
            entry["sources"][row["src_ip"]] += 1
        if row.get("dest_ip"):
            entry["destinations"][row["dest_ip"]] += 1

    top_signatures = sorted(signatures.values(), key=lambda item: item["count"], reverse=True)
    top_signatures = top_signatures[: int(config.get("top_signatures", DEFAULT_TOP_SIGNATURES))]
    top_hosts = int(config.get("top_hosts", DEFAULT_TOP_HOSTS))

    summary = {
        "source": "opnsense_ids",
        "category": "network_security",
        "observed_at": window_end.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "window_start": window_start.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "window_hours": lookback_hours,
        "ids_status": status,
        "alert_count": len(rows),
        "alerts_by_action": dict(actions),
        "interfaces": dict(interfaces),
        "top_signatures": [
            {
                "sid": entry["sid"],
                "signature": entry["signature"],
                "count": entry["count"],
                "actions": dict(entry["actions"]),
                "top_sources": [ip for ip, _ in entry["sources"].most_common(3)],
                "top_destinations": [ip for ip, _ in entry["destinations"].most_common(3)],
                "last_seen": entry["last_seen"],
            }
            for entry in top_signatures
        ],
        "top_local_hosts": [{"ip": ip, "alerts": count} for ip, count in local_hosts.most_common(top_hosts)],
        "top_remote_hosts": [{"ip": ip, "alerts": count} for ip, count in remote_hosts.most_common(top_hosts)],
        "window_truncated": truncated,
        "caveat": (
            "Suricata alerts are signature matches on traffic crossing the monitored "
            "interfaces, not confirmed compromise — false positives are normal and a "
            "single alert is not an incident. Severity is not exposed by the OPNsense "
            "API, so these are ranked by frequency only. Encrypted traffic is largely "
            "opaque to signature matching, and a quiet window is not evidence that "
            "nothing happened."
        ),
    }

    if truncated:
        summary["truncation_note"] = (
            "More alerts existed in this window than were read (max_alerts_scanned "
            "reached, or the Suricata log rotated mid-window). Counts are a lower bound."
        )

    reference = str(config.get("packet_capture_reference", "")).strip()
    if reference:
        summary["packet_capture_reference"] = reference

    return summary


def fetch(lookback_hours):
    config = _load_config(os.environ.get("OPNSENSE_IDS_CONF_PATH", DEFAULT_CONF_PATH))
    if config is None:
        return []

    base_url = str(config["base_url"]).strip().rstrip("/")
    verify = config.get("verify_tls", True)
    if verify is False:
        LOG.warning("opnsense_ids: verify_tls is false, the API session is unauthenticated against MITM")

    window_end = datetime.now(timezone.utc)
    window_start = window_end - timedelta(hours=float(lookback_hours))
    max_pages = max(1, math.ceil(int(config.get("max_alerts_scanned", DEFAULT_MAX_ALERTS_SCANNED)) / ROWS_PER_PAGE))

    session = requests.Session()
    session.auth = (str(config["api_key"]).strip(), str(config["api_secret"]).strip())

    try:
        status = _service_status(session, base_url, verify)
    except Exception:
        # Without the status call a run cannot tell "no alerts" from "Suricata was
        # off the whole window", which is the difference between reassuring and
        # meaningless. Carry on so the alert query still gets its chance.
        LOG.warning("opnsense_ids: service status query failed", exc_info=True)
        status = "unknown"

    try:
        rows, truncated = _collect_alerts(session, base_url, verify, window_start, max_pages)
    except Exception:
        LOG.warning("opnsense_ids: alert query failed, skipping", exc_info=True)
        return []

    interface_filter = [str(item).strip() for item in config.get("interfaces", []) if str(item).strip()]
    if interface_filter:
        rows = [row for row in rows if row.get("in_iface") in interface_filter]

    summary = _summarise(rows, truncated, status, config, window_start, window_end, lookback_hours)
    LOG.info("opnsense_ids: %d alert(s) in the last %sh, IDS %s", summary["alert_count"], lookback_hours, status)
    return [summary]