Add HTTPS management console and response planning
This commit is contained in:
parent
a56d72edd6
commit
a7129e5666
16 changed files with 927 additions and 160 deletions
111
tests/test_respond.py
Normal file
111
tests/test_respond.py
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
# 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 incident, respond
|
||||
from enodia_sentinel.alert import Alert, Severity
|
||||
from enodia_sentinel.cli import main
|
||||
from enodia_sentinel.config import Config
|
||||
|
||||
|
||||
def _alert(sig, detail, pids=(), sid=1, sev=Severity.HIGH):
|
||||
return Alert(severity=sev, signature=sig, key=f"k:{sig}",
|
||||
detail=detail, pids=tuple(pids), sid=sid, classtype="test")
|
||||
|
||||
|
||||
class TestRespondPlan(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = tempfile.TemporaryDirectory()
|
||||
self.tmp = Path(self.dir.name)
|
||||
self.cfg = Config()
|
||||
self.cfg.log_dir = self.tmp
|
||||
self.iid = incident.record(self.cfg, "alert-1.log", [
|
||||
_alert("reverse_shell",
|
||||
"pid=4242 comm=bash stdio=net-socket peer=[8.8.8.8:4444]",
|
||||
pids=[4242], sid=100010, sev=Severity.CRITICAL),
|
||||
_alert("persistence",
|
||||
"persistence file modified: /etc/systemd/system/bad.service",
|
||||
sid=100015),
|
||||
_alert("new_suid",
|
||||
"SUID/SGID binary in writable dir: /tmp/suidsh",
|
||||
sid=100014, sev=Severity.CRITICAL),
|
||||
], lineage={4242, 100}, when=1000.0, host="h")
|
||||
(self.tmp / "alert-1.json").write_text(json.dumps({
|
||||
"time": "2026-06-10T10:00:00-07:00",
|
||||
"host": "h",
|
||||
"severity": "CRITICAL",
|
||||
"incident_id": self.iid,
|
||||
"alerts": [
|
||||
{"signature": "reverse_shell", "sid": 100010,
|
||||
"detail": "pid=4242 comm=bash stdio=net-socket peer=[8.8.8.8:4444]",
|
||||
"pids": [4242]},
|
||||
{"signature": "persistence", "sid": 100015,
|
||||
"detail": "persistence file modified: /etc/systemd/system/bad.service",
|
||||
"pids": []},
|
||||
{"signature": "new_suid", "sid": 100014,
|
||||
"detail": "SUID/SGID binary in writable dir: /tmp/suidsh",
|
||||
"pids": []},
|
||||
],
|
||||
"processes": [{"pid": 4242, "comm": "bash"}],
|
||||
}))
|
||||
|
||||
def tearDown(self):
|
||||
self.dir.cleanup()
|
||||
|
||||
def test_builds_dry_run_plan_from_incident(self):
|
||||
bundle = respond.load_bundle(self.cfg, self.iid)
|
||||
plan = respond.build_plan(bundle, self.cfg)
|
||||
self.assertEqual(plan["mode"], "dry-run")
|
||||
self.assertFalse(plan["apply_supported"])
|
||||
commands = [a["command"] for a in plan["actions"]]
|
||||
self.assertIn(["kill", "-STOP", "4242"], commands)
|
||||
self.assertIn(["kill", "-TERM", "4242"], commands)
|
||||
self.assertIn(["nft", "add", "rule", "inet", "filter", "output",
|
||||
"ip", "daddr", "8.8.8.8", "drop"], commands)
|
||||
self.assertIn(["systemctl", "disable", "--now", "bad.service"], commands)
|
||||
self.assertIn(["mv", "--", "/tmp/suidsh",
|
||||
"/tmp/suidsh.enodia-quarantine"], commands)
|
||||
self.assertNotIn(["mv", "--", "/etc/systemd/system/bad.service",
|
||||
"/etc/systemd/system/bad.service.enodia-quarantine"],
|
||||
commands)
|
||||
self.assertEqual(plan["actions"][0]["category"], "evidence")
|
||||
|
||||
def test_cli_json(self):
|
||||
os.environ["ENODIA_LOG_DIR"] = str(self.tmp)
|
||||
try:
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
code = main(["respond", "plan", self.iid, "--json"])
|
||||
self.assertEqual(code, 0)
|
||||
plan = json.loads(buf.getvalue())
|
||||
self.assertEqual(plan["incident_id"], self.iid)
|
||||
self.assertGreaterEqual(plan["summary"]["action_count"], 5)
|
||||
finally:
|
||||
os.environ.pop("ENODIA_LOG_DIR", None)
|
||||
|
||||
def test_cli_requires_incident_id(self):
|
||||
os.environ["ENODIA_LOG_DIR"] = str(self.tmp)
|
||||
try:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
code = main(["respond", "plan"])
|
||||
self.assertEqual(code, 2)
|
||||
finally:
|
||||
os.environ.pop("ENODIA_LOG_DIR", None)
|
||||
|
||||
def test_unknown_incident(self):
|
||||
os.environ["ENODIA_LOG_DIR"] = str(self.tmp)
|
||||
try:
|
||||
with redirect_stdout(io.StringIO()):
|
||||
code = main(["respond", "plan", "inc-nope"])
|
||||
self.assertEqual(code, 1)
|
||||
finally:
|
||||
os.environ.pop("ENODIA_LOG_DIR", None)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue