enodia-sentinal/tests/test_reconcile_integration.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""Reconciliation wired into the daemon chokepoint, fim-check, and the web API."""
import contextlib
import io
import os
import tempfile
import time
import unittest
from pathlib import Path

from enodia_sentinel import fim, reconcile, web
from enodia_sentinel.alert import Alert, Severity
from enodia_sentinel.cli import _cmd_fim_check
from enodia_sentinel.config import Config
from enodia_sentinel.daemon import Sentinel


def _cfg(tmp: Path) -> Config:
    cfg = Config()
    cfg.log_dir = tmp
    return cfg


class DaemonChokepointTest(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        self.cfg = _cfg(Path(self.d.name))
        reconcile.ReconcileStore.invalidate_cache()

    def tearDown(self):
        self.d.cleanup()
        reconcile.ReconcileStore.invalidate_cache()

    def _accept(self, kind, target, fp, **kw):
        store = reconcile.ReconcileStore.load(self.cfg)
        store.accept(kind, target, fp, "reason", "luna", **kw)
        reconcile.ReconcileStore.invalidate_cache()

    def test_unacked_alert_passes_through(self):
        s = Sentinel(self.cfg)
        a = Alert(Severity.HIGH, "new_listener", "lis:8080/nginx", "d")
        self.assertEqual(len(s.fresh_alerts([a], time.time())), 1)

    def test_acked_listener_is_dropped(self):
        self._accept("listener", "8080/nginx", {"key": "8080/nginx"})
        s = Sentinel(self.cfg)
        a = Alert(Severity.HIGH, "new_listener", "lis:8080/nginx", "d")
        self.assertEqual(s.fresh_alerts([a], time.time()), [])

    def test_acked_fim_dropped_only_while_fingerprint_matches(self):
        live = {"sha256": "a", "mode": 33188, "uid": 0, "gid": 0, "size": 4}
        fp = reconcile.build_fingerprint("fim", "/etc/hosts", live)
        self._accept("fim", "/etc/hosts", fp)
        s = Sentinel(self.cfg)
        a = Alert(Severity.HIGH, "fim_modified", "fim:mod:/etc/hosts", "d")
        # matching live state -> suppressed
        self.assertEqual(
            s.fresh_alerts([a], time.time(), {"fim": {"/etc/hosts": live}}), [])
        # changed again -> re-alerts
        s2 = Sentinel(self.cfg)
        changed = dict(live, sha256="b")
        self.assertEqual(
            len(s2.fresh_alerts([a], time.time(), {"fim": {"/etc/hosts": changed}})), 1)


class FimCheckSuppressionTest(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        self.tmp = Path(self.d.name)
        self.watch = self.tmp / "watched"
        self.watch.mkdir()
        self.cfg = _cfg(self.tmp)
        self.cfg.fim_paths = (str(self.watch),)
        self.cfg.fim_pkg_verify = False
        reconcile.ReconcileStore.invalidate_cache()

    def tearDown(self):
        self.d.cleanup()
        reconcile.ReconcileStore.invalidate_cache()

    def _baseline(self):
        # Seed a baseline directly so the command doesn't shell out to pacman.
        base = fim.scan_paths(self.cfg.fim_path_list())
        self.cfg.fim_baseline.write_text(__import__("json").dumps(base))

    def _fim_check(self):
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            code = _cmd_fim_check(self.cfg, packages=False)
        return code, out.getvalue()

    def test_modified_path_reported_then_suppressed_after_accept(self):
        f = self.watch / "app.conf"
        f.write_text("v1")
        self._baseline()
        f.write_text("v2-tampered")
        code, out = self._fim_check()
        self.assertEqual(code, 1)
        self.assertIn(str(f), out)
        # accept the drift -> the path drops out of fim-check
        live = fim.scan_paths([str(f)]).get(str(f))
        fp = reconcile.build_fingerprint("fim", str(f), live)
        store = reconcile.ReconcileStore.load(self.cfg)
        store.accept("fim", str(f), fp, "rolled out v2", "luna")
        reconcile.ReconcileStore.invalidate_cache()
        code2, out2 = self._fim_check()
        self.assertEqual(code2, 0)
        self.assertNotIn(str(f), out2)


class WebReconciliationBlockTest(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        self.cfg = _cfg(Path(self.d.name))
        reconcile.ReconcileStore.invalidate_cache()

    def tearDown(self):
        self.d.cleanup()
        reconcile.ReconcileStore.invalidate_cache()

    def test_integrity_report_has_reconciliation_summary(self):
        store = reconcile.ReconcileStore.load(self.cfg)
        past = reconcile._utcnow() - __import__("datetime").timedelta(days=1)
        store.accept("listener", "1/a", {"key": "1/a"}, "ok now", "luna",
                     expires_at=reconcile._utcnow() + __import__("datetime").timedelta(days=1))
        store.accept("listener", "2/b", {"key": "2/b"}, "expired", "luna",
                     expires_at=past)
        reconcile.ReconcileStore.invalidate_cache()
        report = web.integrity_report(self.cfg)
        recon = report["reconciliation"]
        self.assertEqual(recon["total"], 2)
        self.assertEqual(recon["stale"], 1)
        self.assertEqual(report["checks"]["reconciliation"], "review")

    def test_integrity_report_never_writes_store(self):
        # The dashboard is read-only: rendering the integrity view must not
        # persist a TTL-lapsed ack, even though it now reports it as stale.
        past = reconcile._utcnow() - __import__("datetime").timedelta(days=1)
        store = reconcile.ReconcileStore.load(self.cfg)
        store.accept("listener", "2/b", {"key": "2/b"}, "expired", "luna",
                     expires_at=past)  # status persisted as "ok"
        reconcile.ReconcileStore.invalidate_cache()
        path = reconcile.ReconcileStore.store_path(self.cfg)
        before = path.read_bytes()
        web.integrity_report(self.cfg)
        self.assertEqual(path.read_bytes(), before,
                         "integrity_report wrote to the reconciliation store")


if __name__ == "__main__":
    unittest.main()