Add baseline reconciliation: accept audited drift with a reason

Implements the v0.9 roadmap item from the approved design spec. Operators
accept a specific FIM/package/listener/SUID drift item with a mandatory
reason; the ack suppresses that one alert only while the live state still
matches the recorded fingerprint. Content kinds (fim/pkgfile) re-alert on
further change; identity kinds (listener/suid) retire on TTL or revoke.

- reconcile.py: ReconcileStore (mtime-cached, fails closed on missing/corrupt
  store), fingerprint builders, and the filter_alerts chokepoint.
- CLI: baseline accept/revoke/list with --reason/--expires/--force/--stale/--json.
- Wired at the daemon sweep + eBPF chokepoint, fim-check, and /api/integrity.
- RECONCILE_V1 schema, config knob, COMMAND_REFERENCE/SCHEMAS/OPERATIONS/ROADMAP
  docs, and reconcile unit/CLI/integration tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-06-27 10:33:55 -07:00
parent 3d2047fde2
commit dbbe35b3fc
18 changed files with 1155 additions and 28 deletions

125
tests/test_reconcile_cli.py Normal file
View file

@ -0,0 +1,125 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tests for the `baseline accept/revoke/list` CLI surface."""
import contextlib
import io
import json
import os
import tempfile
import unittest
from pathlib import Path
from enodia_sentinel import reconcile
from enodia_sentinel.cli import main
def _run(*argv) -> tuple[int, str, str]:
out, err = io.StringIO(), io.StringIO()
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
code = main(list(argv))
return code, out.getvalue(), err.getvalue()
class BaselineCLITest(unittest.TestCase):
def setUp(self):
self.d = tempfile.TemporaryDirectory()
self._old = os.environ.get("ENODIA_LOG_DIR")
os.environ["ENODIA_LOG_DIR"] = self.d.name
reconcile.ReconcileStore.invalidate_cache()
def tearDown(self):
if self._old is None:
os.environ.pop("ENODIA_LOG_DIR", None)
else:
os.environ["ENODIA_LOG_DIR"] = self._old
self.d.cleanup()
reconcile.ReconcileStore.invalidate_cache()
def _store(self):
from enodia_sentinel.config import Config
reconcile.ReconcileStore.invalidate_cache()
return reconcile.ReconcileStore.load(Config.load())
def test_accept_fim_path_round_trips(self):
f = Path(self.d.name) / "watched.conf"
f.write_text("hello")
code, _out, _err = _run("baseline", "accept", "fim", str(f),
"--reason", "operator edit")
self.assertEqual(code, 0)
recs = self._store().list()
self.assertEqual(len(recs), 1)
self.assertEqual(recs[0].kind, "fim")
self.assertEqual(recs[0].reason, "operator edit")
self.assertIn("sha256", recs[0].fingerprint)
def test_accept_requires_reason(self):
f = Path(self.d.name) / "x"
f.write_text("y")
code, _out, err = _run("baseline", "accept", "fim", str(f))
self.assertEqual(code, 2)
self.assertIn("reason", err.lower())
def test_accept_missing_live_data_requires_force(self):
code, _out, err = _run("baseline", "accept", "listener", "65000/ghost",
"--reason", "temp")
self.assertEqual(code, 1)
self.assertIn("force", err.lower())
self.assertEqual(self._store().list(), [])
def test_accept_force_overrides_missing_live_data(self):
code, _out, _err = _run("baseline", "accept", "listener",
"65000/ghost", "--reason", "temp", "--force")
self.assertEqual(code, 0)
self.assertEqual(len(self._store().list()), 1)
def test_accept_invalid_expires_errors(self):
code, _out, err = _run("baseline", "accept", "listener", "1/a",
"--reason", "r", "--force", "--expires", "soon")
self.assertEqual(code, 2)
self.assertIn("duration", err.lower())
def test_revoke_removes_then_reports_missing(self):
_run("baseline", "accept", "listener", "1/a", "--reason", "r", "--force")
code, _o, _e = _run("baseline", "revoke", "listener", "1/a")
self.assertEqual(code, 0)
code2, _o2, err2 = _run("baseline", "revoke", "listener", "1/a")
self.assertEqual(code2, 1)
self.assertIn("not found", err2.lower())
def test_list_json_emits_records(self):
f = Path(self.d.name) / "c.conf"
f.write_text("z")
_run("baseline", "accept", "fim", str(f), "--reason", "edit")
code, out, _err = _run("baseline", "list", "--json")
self.assertEqual(code, 0)
data = json.loads(out)
self.assertEqual(len(data["records"]), 1)
self.assertEqual(data["records"][0]["kind"], "fim")
def test_list_stale_exit_code(self):
# no acks -> exit 0
code, _o, _e = _run("baseline", "list", "--stale")
self.assertEqual(code, 0)
# an expired ack -> exit 1
_run("baseline", "accept", "listener", "1/a", "--reason", "r",
"--force", "--expires", "0m")
code2, _o2, _e2 = _run("baseline", "list", "--stale")
self.assertEqual(code2, 1)
def test_bare_baseline_still_builds(self):
# The legacy `baseline` (no action) must keep routing to the rebuild
# path. Stub the expensive filesystem scan; we only assert the dispatch.
from enodia_sentinel import cli
called = []
orig = cli.Sentinel.build_baselines
cli.Sentinel.build_baselines = lambda self: called.append(True)
try:
code, out, _err = _run("baseline")
finally:
cli.Sentinel.build_baselines = orig
self.assertEqual(code, 0)
self.assertTrue(called)
self.assertIn("Baselines written", out)
if __name__ == "__main__":
unittest.main()