feat(antirootkit): alerting + IOC matching (Task 8)

Add api/alerts.py: ioc_match (exact-IP v1 lookup), build_alert
(assembles exe/pid/score/reasons/dest/ioc from an ExecEvent), and
emit (fans an alert out to injected soc_post/mailer/mesh sinks,
each isolated in its own try/except so one failing sink never
blocks the others or raises out of emit).

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
CyberMind-FR 2026-07-28 08:54:10 +02:00
parent 6f3b8b1036
commit b5a44e6d49
2 changed files with 144 additions and 0 deletions

View File

@ -0,0 +1,52 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# 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

View File

@ -0,0 +1,92 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# 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