Add stable v1 schema IDs and docs

tests/test_schema_contracts.py

@@ -0,0 +1,222 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Compatibility tests for stable operator-facing JSON contracts."""
+import json
+import os
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from io import StringIO
+from pathlib import Path
+from unittest.mock import patch
+
+from enodia_sentinel import incident, respond, schemas, snapshot, web
+from enodia_sentinel.alert import Alert, Severity
+from enodia_sentinel.cli import main
+from enodia_sentinel.config import Config
+
+
+def _cfg(tmp: Path) -> Config:
+    c = Config()
+    c.log_dir = tmp
+    return c
+
+
+def _alert() -> Alert:
+    return Alert(
+        Severity.CRITICAL,
+        "reverse_shell",
+        "rsh:4242",
+        "pid=4242 peer=[8.8.8.8:4444]",
+        (4242,),
+        sid=100010,
+        classtype="command-and-control",
+    )
+
+
+def _assert_types(case: unittest.TestCase, obj: dict, expected: dict):
+    for key, typ in expected.items():
+        case.assertIn(key, obj)
+        case.assertIsInstance(obj[key], typ, key)
+
+
+class TestSchemaContracts(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.TemporaryDirectory()
+        self.tmp = Path(self.dir.name)
+        self.cfg = _cfg(self.tmp)
+
+    def tearDown(self):
+        self.dir.cleanup()
+
+    def _record_incident_with_snapshot(self) -> str:
+        alert = _alert()
+        iid = incident.record(
+            self.cfg, "alert-1.log", [alert], {4242, 100}, 1000.0, "host-a")
+        self.assertIsNotNone(iid)
+        (self.tmp / "alert-1.json").write_text(json.dumps({
+            "schema": schemas.ALERT_SNAPSHOT_V1,
+            "time": "2026-06-16T12:00:00-07:00",
+            "host": "host-a",
+            "severity": "CRITICAL",
+            "incident_id": iid,
+            "alerts": [alert.to_dict()],
+            "processes": [{"pid": 4242, "comm": "bash"}],
+        }))
+        return str(iid)
+
+    def test_alert_v1_contract(self):
+        alert = _alert().to_dict()
+        self.assertEqual(schemas.ALERT_V1, "enodia.alert.v1")
+        _assert_types(self, alert, {
+            "sid": int,
+            "severity": str,
+            "signature": str,
+            "classtype": str,
+            "key": str,
+            "detail": str,
+            "pids": list,
+        })
+
+    def test_alert_snapshot_v1_contract(self):
+        with patch("enodia_sentinel.snapshot._run", return_value=""):
+            path = snapshot.capture([_alert()], self._State(), self.cfg)
+        data = json.loads(path.with_suffix(".json").read_text())
+        self.assertEqual(data["schema"], schemas.ALERT_SNAPSHOT_V1)
+        _assert_types(self, data, {
+            "schema": str,
+            "time": str,
+            "host": str,
+            "severity": str,
+            "incident_id": str,
+            "alerts": list,
+            "processes": list,
+        })
+        self.assertEqual(data["alerts"][0]["signature"], "reverse_shell")
+
+    def test_incident_v1_contract_and_json_views(self):
+        iid = self._record_incident_with_snapshot()
+        inc = incident.load_index(self.cfg)[iid]
+        self.assertEqual(inc["schema"], schemas.INCIDENT_V1)
+        _assert_types(self, inc, {
+            "schema": str,
+            "id": str,
+            "host": str,
+            "first_ts": float,
+            "last_ts": float,
+            "first_seen": str,
+            "last_seen": str,
+            "severity": str,
+            "signatures": list,
+            "sids": list,
+            "pids": list,
+            "lineage": list,
+            "snapshots": list,
+            "alert_count": int,
+        })
+
+        view = web.get_incident(self.cfg, iid)
+        self.assertEqual(view["schema"], schemas.INCIDENT_VIEW_V1)
+        _assert_types(self, view, {
+            "schema": str,
+            "incident": dict,
+            "timeline": list,
+            "snapshots": list,
+        })
+
+        os.environ["ENODIA_LOG_DIR"] = str(self.tmp)
+        try:
+            buf = StringIO()
+            with redirect_stdout(buf):
+                code = main(["incident", "export", iid])
+            self.assertEqual(code, 0)
+            bundle = json.loads(buf.getvalue())
+        finally:
+            os.environ.pop("ENODIA_LOG_DIR", None)
+        self.assertEqual(bundle["schema"], schemas.INCIDENT_BUNDLE_V1)
+        _assert_types(self, bundle, {
+            "schema": str,
+            "incident": dict,
+            "snapshots": list,
+        })
+
+    def test_old_incident_records_get_schema_backfilled(self):
+        (self.tmp / "incidents.json").write_text(json.dumps({
+            "inc-old": {
+                "id": "inc-old",
+                "host": "host-a",
+                "first_ts": 1.0,
+                "last_ts": 1.0,
+                "first_seen": "2026-06-16T12:00:00-07:00",
+                "last_seen": "2026-06-16T12:00:00-07:00",
+                "severity": "HIGH",
+                "signatures": [],
+                "sids": [],
+                "pids": [],
+                "lineage": [],
+                "snapshots": [],
+                "alert_count": 0,
+            },
+        }))
+        inc = incident.load_index(self.cfg)["inc-old"]
+        self.assertEqual(inc["schema"], schemas.INCIDENT_V1)
+
+    def test_status_v1_contract(self):
+        status = web.daemon_status(self.cfg)
+        self.assertEqual(status["schema"], schemas.STATUS_V1)
+        _assert_types(self, status, {
+            "schema": str,
+            "version": str,
+            "running": bool,
+            "total_alerts": int,
+            "counts": dict,
+            "last_alert": (str, type(None)),
+            "ebpf": str,
+            "ebpf_exec": str,
+            "ebpf_syscall": str,
+            "host": str,
+            "heartbeat_age": (float, type(None)),
+            "heartbeat_stale": bool,
+        })
+
+    def test_response_plan_and_audit_contracts(self):
+        iid = self._record_incident_with_snapshot()
+        bundle = respond.load_bundle(self.cfg, iid)
+        plan = respond.build_plan(bundle, self.cfg)
+        self.assertEqual(plan["schema"], schemas.RESPONSE_PLAN_V1)
+        _assert_types(self, plan, {
+            "schema": str,
+            "plan_id": str,
+            "incident_id": str,
+            "created_at": str,
+            "mode": str,
+            "apply_supported": bool,
+            "summary": dict,
+            "actions": list,
+            "notes": list,
+        })
+        self.assertEqual(plan["actions"][0]["requires_review"], True)
+
+        artifacts = respond.persist_plan(self.cfg, plan, actor="test")
+        audit = Path(artifacts["audit_log"])
+        record = json.loads(audit.read_text().splitlines()[-1])
+        self.assertEqual(record["schema"], schemas.RESPONSE_AUDIT_V1)
+        _assert_types(self, record, {
+            "schema": str,
+            "time": str,
+            "event": str,
+            "actor": str,
+            "plan_id": str,
+            "incident_id": str,
+            "mode": str,
+            "apply_supported": bool,
+            "action_count": int,
+            "plan_path": str,
+        })
+
+    class _State:
+        def process(self, _pid):
+            return None
+
+
+if __name__ == "__main__":
+    unittest.main()