#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Compare ported Go poll and exec-rule output with the Python oracle.""" from __future__ import annotations import json import os import subprocess import sys from pathlib import Path from types import SimpleNamespace from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from enodia_sentinel import __version__, correlation, event, ruleops # noqa: E402 from enodia_sentinel.config import Config # noqa: E402 from enodia_sentinel.detectors import ( # noqa: E402 credential_access, deleted_exe, egress, first_seen, input_snooper, ld_preload, memory_obfuscation, new_listener, new_suid, persistence, reverse_shell, stealth_network, ) from enodia_sentinel.events.exec_event import ExecEvent # noqa: E402 from enodia_sentinel.events.rules import ExecRuleEngine # noqa: E402 from enodia_sentinel.events.syscall_event import SyscallEvent # noqa: E402 from enodia_sentinel.events.syscall_rules import SyscallRuleEngine # noqa: E402 from enodia_sentinel.system import MemoryMap, 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()) env = os.environ.copy() env.pop("ENODIA_CONFIG", None) env["GOCACHE"] = "/tmp/enodia-go-cache" command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(ROOT / "tests/fixtures/go/no-such-config.toml"), "--once", "--fixture", str(fixture), "--host", HOST, "--timestamp", TIMESTAMP, ] result = subprocess.run( command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if result.returncode: print(result.stderr, file=sys.stderr, end="") return result.returncode go_events = [json.loads(line) for line in result.stdout.splitlines() if line] go_alerts = [record for record in go_events if record["event_type"] == "alert"] processes = [] for raw_process in fixture_data["processes"]: process = dict(raw_process) process.setdefault("environ", {}) process.setdefault("fd_targets", {}) process["memory_maps"] = [MemoryMap(**mapping) for mapping in process.get("memory_maps", [])] processes.append(FixtureProcess(**process)) sockets = [Socket(**item) for item in fixture_data.get("sockets", [])] cfg = Config() state = SystemState( processes=processes, sockets=sockets, listener_baseline=set(fixture_data["listener_baseline"]) if "listener_baseline" in fixture_data else None, suid_baseline=set(fixture_data["suid_baseline"]) if "suid_baseline" in fixture_data else None, suid_binaries=fixture_data.get("suid_binaries"), persist_since=fixture_data.get("persist_since"), ) 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.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(memory_obfuscation.detect(state, cfg)) alerts.extend(egress.detect(state, cfg)) first_seen_data = { "schema": "enodia.first_seen.v1", "initialized": fixture_data.get("first_seen_initialized", False), "public_destinations": fixture_data.get("first_seen_public_destinations", {}), "listener_ports": fixture_data.get("first_seen_listener_ports", {}), } with patch.object(first_seen, "_load", return_value=first_seen_data), \ patch.object(first_seen, "_save"): alerts.extend(first_seen.detect(state, cfg)) alerts.extend(new_listener.detect(state, cfg)) with patch.object(persistence, "_iter_files", return_value=list(fixture_data.get("persistence_files", {}))): with patch.object( persistence.os, "lstat", side_effect=lambda path: SimpleNamespace( st_mtime=fixture_data["persistence_files"][path]), ): alerts.extend(persistence.detect(state, cfg)) alerts.extend(new_suid.detect(state, cfg)) python_alerts = [ event.from_alert(alert, host=HOST, timestamp=TIMESTAMP) for alert in alerts ] if go_alerts != python_alerts: print("Go/Python alert parity mismatch", file=sys.stderr) print(json.dumps({"go": go_alerts, "python": python_alerts}, indent=2), file=sys.stderr) return 1 status_events = [record for record in go_events if record["event_type"] == "status"] if len(status_events) != 1 or status_events[0].get("schema") != "enodia.event.v1": print("Go sidecar did not emit one enodia.event.v1 status record", file=sys.stderr) return 1 status = status_events[0].get("status", {}) if status.get("schema") != "enodia.status.v1" or status.get("version") != __version__: print("Go status schema/version drifted from Python", file=sys.stderr) return 1 # Exec rules use a separate replay mode so the event model and rule engine # stay independently testable before the Go side has a kernel event source. exec_fixture = ROOT / "tests/fixtures/go/exec-events.json" exec_rules = ROOT / "tests/fixtures/go/exec-rules.toml" exec_config = ROOT / "tests/fixtures/go/exec-rules-config.toml" exec_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(exec_config), "--exec-events", str(exec_fixture), "--host", HOST, "--timestamp", TIMESTAMP, ] exec_result = subprocess.run( exec_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if exec_result.returncode: print(exec_result.stderr, file=sys.stderr, end="") return exec_result.returncode go_exec_alerts = [ json.loads(line) for line in exec_result.stdout.splitlines() if line ] exec_cfg = Config() exec_cfg.exec_rules_file = str(exec_rules) exec_engine = ExecRuleEngine.load(exec_cfg.exec_rules_file) exec_alerts = [] for raw in json.loads(exec_fixture.read_text()): raw["argv"] = tuple(raw.get("argv", ())) exec_alerts.extend(exec_engine.match(ExecEvent(**raw))) python_exec_alerts = [ event.from_alert(alert, host=HOST, timestamp=TIMESTAMP) for alert in exec_alerts ] if go_exec_alerts != python_exec_alerts: print("Go/Python exec-rule parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_exec_alerts, "python": python_exec_alerts}, indent=2, ), file=sys.stderr) return 1 syscall_fixture = ROOT / "tests/fixtures/go/syscall-events.json" syscall_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(ROOT / "tests/fixtures/go/no-such-config.toml"), "--syscall-events", str(syscall_fixture), "--host", HOST, "--timestamp", TIMESTAMP, ] syscall_result = subprocess.run( syscall_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if syscall_result.returncode: print(syscall_result.stderr, file=sys.stderr, end="") return syscall_result.returncode go_syscall_alerts = [ json.loads(line) for line in syscall_result.stdout.splitlines() if line ] syscall_engine = SyscallRuleEngine() syscall_alerts = [] for raw in json.loads(syscall_fixture.read_text()): args = list(raw.pop("args", ())) args.extend([0] * (6 - len(args))) syscall_alerts.extend(syscall_engine.match(SyscallEvent( **raw, arg0=args[0], arg1=args[1], arg2=args[2], arg3=args[3], arg4=args[4], arg5=args[5], ))) python_syscall_alerts = [ event.from_alert(alert, host=HOST, timestamp=TIMESTAMP) for alert in syscall_alerts ] if go_syscall_alerts != python_syscall_alerts: print("Go/Python syscall-rule parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_syscall_alerts, "python": python_syscall_alerts}, indent=2, ), file=sys.stderr) return 1 host_fixture = ROOT / "tests/fixtures/go/host-events.json" host_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(ROOT / "tests/fixtures/go/no-such-config.toml"), "--host-events", str(host_fixture), "--host", HOST, "--timestamp", TIMESTAMP, ] host_result = subprocess.run( host_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if host_result.returncode: print(host_result.stderr, file=sys.stderr, end="") return host_result.returncode go_host_alerts = [ json.loads(line) for line in host_result.stdout.splitlines() if line ] host_alerts = [] for raw in json.loads(host_fixture.read_text()): host_alerts.extend(ruleops.test_event(Config(), raw)) python_host_alerts = [ event.from_alert(alert, host=HOST, timestamp=TIMESTAMP) for alert in host_alerts ] if go_host_alerts != python_host_alerts: print("Go/Python host-rule parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_host_alerts, "python": python_host_alerts}, indent=2, ), file=sys.stderr) return 1 catalog_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(exec_config), "--rules-list", ] catalog_result = subprocess.run( catalog_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if catalog_result.returncode: print(catalog_result.stderr, file=sys.stderr, end="") return catalog_result.returncode go_catalog = json.loads(catalog_result.stdout) python_catalog = ruleops.list_rules(exec_cfg) if go_catalog != python_catalog: print("Go/Python rule-catalog parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_catalog, "python": python_catalog}, indent=2, ), file=sys.stderr) return 1 mixed_fixture = ROOT / "tests/fixtures/go/mixed-events.jsonl" mixed_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(exec_config), "--event-stream", str(mixed_fixture), "--host", HOST, "--timestamp", TIMESTAMP, ] mixed_result = subprocess.run( mixed_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if mixed_result.returncode: print(mixed_result.stderr, file=sys.stderr, end="") return mixed_result.returncode go_mixed_alerts = [ json.loads(line) for line in mixed_result.stdout.splitlines() if line ] mixed_alerts = [] for line in mixed_fixture.read_text().splitlines(): if line.strip(): mixed_alerts.extend(ruleops.test_event(exec_cfg, json.loads(line))) python_mixed_alerts = [ event.from_alert(alert, host=HOST, timestamp=TIMESTAMP) for alert in mixed_alerts ] if go_mixed_alerts != python_mixed_alerts: print("Go/Python mixed-event pipeline parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_mixed_alerts, "python": python_mixed_alerts}, indent=2, ), file=sys.stderr) return 1 correlation_fixture = ROOT / "tests/fixtures/go/correlation-incidents.json" correlation_command = [ "go", "run", "./cmd/enodia-sentinel-go", "--config", str(exec_config), "--correlate", str(correlation_fixture), ] correlation_result = subprocess.run( correlation_command, cwd=ROOT / "go-agent", env=env, capture_output=True, text=True, timeout=60, ) if correlation_result.returncode: print(correlation_result.stderr, file=sys.stderr, end="") return correlation_result.returncode go_correlations = json.loads(correlation_result.stdout) python_correlations = [ correlation.correlate(incident) for incident in json.loads(correlation_fixture.read_text()) ] if go_correlations != python_correlations: print("Go/Python correlation parity mismatch", file=sys.stderr) print(json.dumps( {"go": go_correlations, "python": python_correlations}, indent=2, ), file=sys.stderr) return 1 signatures = sorted({record["alert"]["signature"] for record in go_alerts}) exec_sids = sorted(record["alert"]["sid"] for record in go_exec_alerts) syscall_sids = sorted(record["alert"]["sid"] for record in go_syscall_alerts) host_sids = sorted(record["alert"]["sid"] for record in go_host_alerts) print( f"parity ok: {len(go_alerts)} poll alerts across {', '.join(signatures)}; " f"{len(go_exec_alerts)} exec alerts across SIDs {exec_sids}; " f"{len(go_syscall_alerts)} syscall alerts across SIDs {syscall_sids}; " f"{len(go_host_alerts)} host alerts across SIDs {host_sids}; " f"{len(go_catalog)} catalog records; {len(go_mixed_alerts)} mixed-stream alerts" f"; {sum(len(item) for item in go_correlations)} correlations" ) return 0 if __name__ == "__main__": raise SystemExit(main())