280 lines
9 KiB
Python
280 lines
9 KiB
Python
# 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 assurance, incident, reconcile, 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_reconcile_v1_contract(self):
|
|
self.assertEqual(schemas.RECONCILE_V1, "enodia.reconcile.v1")
|
|
store = reconcile.ReconcileStore.load(self.cfg)
|
|
store.accept("fim", "/etc/hosts",
|
|
{"sha256": "a", "mode": 1, "uid": 0, "gid": 0},
|
|
"operator edit", "luna")
|
|
reconcile.ReconcileStore.invalidate_cache()
|
|
data = json.loads(
|
|
reconcile.ReconcileStore.store_path(self.cfg).read_text())
|
|
self.assertEqual(data["schema"], schemas.RECONCILE_V1)
|
|
_assert_types(self, data, {"schema": str, "records": list})
|
|
_assert_types(self, data["records"][0], {
|
|
"kind": str,
|
|
"target": str,
|
|
"fingerprint": dict,
|
|
"reason": str,
|
|
"actor": str,
|
|
"accepted_at": str,
|
|
"status": str,
|
|
})
|
|
|
|
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,
|
|
"enrichment": dict,
|
|
})
|
|
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,
|
|
"correlations": list,
|
|
})
|
|
|
|
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_integrity_v1_contract(self):
|
|
report = web.integrity_report(self.cfg, status={
|
|
"running": True,
|
|
"heartbeat_age": 1.0,
|
|
"heartbeat_stale": False,
|
|
})
|
|
self.assertEqual(report["schema"], schemas.INTEGRITY_V1)
|
|
_assert_types(self, report, {
|
|
"schema": str,
|
|
"generated_at": float,
|
|
"status": str,
|
|
"checks": dict,
|
|
"watchdog": dict,
|
|
"anchors": dict,
|
|
"sentinel_footprint": dict,
|
|
"hash_chain": dict,
|
|
"read_only": bool,
|
|
})
|
|
|
|
def test_hash_chain_v1_contract(self):
|
|
artifact = self.tmp / "artifact.json"
|
|
artifact.write_text("{}")
|
|
record = assurance.append_artifact(self.cfg, "test-artifact", artifact)
|
|
self.assertEqual(record["schema"], schemas.HASH_CHAIN_V1)
|
|
_assert_types(self, record, {
|
|
"schema": str,
|
|
"time": float,
|
|
"kind": str,
|
|
"path": str,
|
|
"size": int,
|
|
"sha256": str,
|
|
"prev_hash": str,
|
|
"chain_hash": str,
|
|
})
|
|
|
|
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()
|