dbbe35b3fc
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>
125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
# 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()
|