feat(go): port socket detectors

This commit is contained in:
Luna 2026-07-10 17:29:07 -07:00
parent fc9b5f0448
commit c6f7219de8
23 changed files with 693 additions and 28 deletions

View file

@ -1,6 +1,6 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Compare the first Go detector port with the Python reference oracle."""
"""Compare ported Go poll detectors with the Python reference oracle."""
from __future__ import annotations
import json
@ -19,15 +19,33 @@ from enodia_sentinel.config import Config # noqa: E402
from enodia_sentinel.detectors import ( # noqa: E402
credential_access,
deleted_exe,
egress,
input_snooper,
ld_preload,
reverse_shell,
stealth_network,
)
from enodia_sentinel.system import SystemState # noqa: E402
from enodia_sentinel.system import Socket, SystemState # noqa: E402
HOST = "parity-host"
TIMESTAMP = "2026-07-10T00:00:00-07:00"
class FixtureProcess(SimpleNamespace):
"""Duck-typed Python Process with Go-fixture stdio socket semantics."""
def stdio_socket_inode(self):
# Match Process.stdio_socket_inode: descriptors 0/1/2 only, in order.
for fd in ("0", "1", "2"):
target = self.fd_targets.get(fd, "")
if target.startswith("socket:[") and target.endswith("]"):
try:
return int(target[8:-1])
except ValueError:
continue
return None
def main() -> int:
fixture = ROOT / "tests/fixtures/go/process-detectors.json"
fixture_data = json.loads(fixture.read_text())
@ -55,17 +73,21 @@ def main() -> int:
process = dict(raw_process)
process.setdefault("environ", {})
process.setdefault("fd_targets", {})
processes.append(SimpleNamespace(**process))
processes.append(FixtureProcess(**process))
sockets = [Socket(**item) for item in fixture_data.get("sockets", [])]
cfg = Config()
state = SystemState(processes=processes, sockets=[])
state = SystemState(processes=processes, sockets=sockets)
alerts = list(reverse_shell.detect(state, cfg))
with patch.object(ld_preload, "Path") as path_class:
preload = path_class.return_value
preload.is_file.return_value = bool(fixture_data.get("ld_preload"))
preload.read_text.return_value = fixture_data.get("ld_preload", "")
alerts = list(ld_preload.detect(state, cfg))
alerts.extend(ld_preload.detect(state, cfg))
alerts.extend(deleted_exe.detect(state, cfg))
alerts.extend(input_snooper.detect(state, cfg))
alerts.extend(credential_access.detect(state, cfg))
alerts.extend(stealth_network.detect(state, cfg))
alerts.extend(egress.detect(state, cfg))
python_alerts = [
event.from_alert(alert, host=HOST, timestamp=TIMESTAMP)
for alert in alerts