diff --git a/packages/secubox-antirootkit/api/alerts.py b/packages/secubox-antirootkit/api/alerts.py new file mode 100644 index 00000000..364d3568 --- /dev/null +++ b/packages/secubox-antirootkit/api/alerts.py @@ -0,0 +1,52 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: antirootkit alerting + IOC matching + +Pure/injectable building blocks: +- ioc_match: exact-IP IOC lookup (v1; domain/asn matching is out of scope) +- build_alert: assemble an alert dict from a suspicious ExecEvent +- emit: fan out an alert to SOC/mail/mesh sinks, resiliently + +Sinks are injected callables (soc_post, mailer, mesh) — this module never +touches the network or sends mail itself, and each sink call is isolated +so one failing sink never blocks the others. +""" + + +def ioc_match(dest_ip: str, ioc: dict) -> bool: + """Return True if dest_ip is in ioc.get("ips", []).""" + return dest_ip in (ioc or {}).get("ips", []) + + +def build_alert(ev, score: int, reasons: list, dest: str | None = None, ioc: dict = None) -> dict: + """Build an alert dict from a suspicious ExecEvent. + + ioc is the (optional) loaded IOC dict, e.g. antirootkit.toml [ioc]. + The "ioc" key is True only when a dest is given AND it matches ioc["ips"]. + """ + return { + "exe": ev.exe, + "pid": ev.pid, + "score": score, + "reasons": reasons, + "dest": dest, + "ioc": ioc_match(dest, ioc or {}) if dest else False, + } + + +def emit(alert: dict, soc_post, mailer, mesh) -> None: + """Fan an alert out to the SOC, mail and mesh sinks. + + Each sink is called in its own try/except: a sink raising is logged + away and never propagates, so it can never prevent the remaining + sinks from being called nor raise out of emit(). + """ + for sink in (soc_post, mailer, mesh): + try: + sink(alert) + except Exception: + pass diff --git a/packages/secubox-antirootkit/tests/test_alerts.py b/packages/secubox-antirootkit/tests/test_alerts.py new file mode 100644 index 00000000..062186fa --- /dev/null +++ b/packages/secubox-antirootkit/tests/test_alerts.py @@ -0,0 +1,92 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +from api import alerts +from api.execwatch import ExecEvent + + +def test_ioc_match(): + assert alerts.ioc_match("5.182.207.11", {"ips": ["5.182.207.11"]}) is True + assert alerts.ioc_match("1.1.1.1", {"ips": ["5.182.207.11"]}) is False + + +def test_ioc_match_empty_ioc(): + assert alerts.ioc_match("5.182.207.11", {}) is False + + +def test_emit_swallows_sink_error(): + a = {"x": 1} + + def boom(_): + raise RuntimeError("soc down") + + called = [] + alerts.emit( + a, + soc_post=boom, + mailer=lambda x: called.append("mail"), + mesh=lambda x: called.append("mesh"), + ) + assert "mail" in called and "mesh" in called + + +def test_emit_calls_all_sinks_on_success(): + a = {"x": 1} + called = [] + alerts.emit( + a, + soc_post=lambda x: called.append("soc"), + mailer=lambda x: called.append("mail"), + mesh=lambda x: called.append("mesh"), + ) + assert called == ["soc", "mail", "mesh"] + + +def test_emit_all_sinks_fail_never_raises(): + def boom(_): + raise RuntimeError("down") + + alerts.emit({"x": 1}, soc_post=boom, mailer=boom, mesh=boom) + + +def test_build_alert_shape_no_dest(): + e = ExecEvent(pid=7, ppid=1, uid=0, exe="/tmp/x", argv=[], success=True) + a = alerts.build_alert(e, score=4, reasons=["non-dpkg-exec-path"]) + assert a["exe"] == "/tmp/x" + assert a["pid"] == 7 + assert a["score"] == 4 + assert a["reasons"] == ["non-dpkg-exec-path"] + assert a["dest"] is None + assert a["ioc"] is False + + +def test_build_alert_ioc_true_when_dest_matches(): + e = ExecEvent(pid=7, ppid=1, uid=0, exe="/tmp/x", argv=[], success=True) + ioc = {"ips": ["5.182.207.11"]} + a = alerts.build_alert( + e, score=4, reasons=["non-dpkg-exec-path"], dest="5.182.207.11", ioc=ioc + ) + assert a["dest"] == "5.182.207.11" + assert a["ioc"] is True + assert a["ioc"] == alerts.ioc_match("5.182.207.11", ioc) + + +def test_build_alert_ioc_false_when_dest_not_in_ioc(): + e = ExecEvent(pid=7, ppid=1, uid=0, exe="/tmp/x", argv=[], success=True) + ioc = {"ips": ["5.182.207.11"]} + a = alerts.build_alert( + e, score=4, reasons=["non-dpkg-exec-path"], dest="1.1.1.1", ioc=ioc + ) + assert a["dest"] == "1.1.1.1" + assert a["ioc"] is False + + +def test_build_alert_ioc_false_when_no_ioc_given(): + e = ExecEvent(pid=7, ppid=1, uid=0, exe="/tmp/x", argv=[], success=True) + a = alerts.build_alert( + e, score=4, reasons=["non-dpkg-exec-path"], dest="5.182.207.11" + ) + assert a["dest"] == "5.182.207.11" + assert a["ioc"] is False