From 3a8600e7702f11a09a05a06cc5b0f814654083df Mon Sep 17 00:00:00 2001 From: branover Date: Fri, 17 Jul 2026 12:41:11 -0400 Subject: [PATCH 1/2] feat: audit the real connection outcome of egress probes (net_list_egress) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit net_list_egress previously showed only the POLICY decision — allowed=True the moment a destination passed the loopback/private allowlist — so a probe that then failed at the socket layer (e.g. "No route to host" to a lab-private subnet) left no trace: the audit log misleadingly read "allowed: true" for a connection that never happened. This records a follow-up EgressEvent with the REAL outcome, tool suffixed :connected / :connect_failed, distinct from the pre-flight policy-allow event. The outcome is derived by INSPECTING the probe result, not by catching an exception: the sandbox probes report a socket failure by RETURNING {"ok": false, "error": ...} (http_probe/tcp_probe catch and serialize it), so an exception-only check would miss exactly the failures this audit exists to record (it would log :connected for a failed connection). _probe_outcome reads the probe's own success signal across all three result shapes (http single {response}, web_poc {steps}, tcp/udp flat {ok}): "connected" if any attempt produced a real transport response (an HTTP status incl. 4xx/5xx, a TCP banner, an ok UDP send), else "connect_failed". Both events are durable=True so the audit survives even a caller that rolls back on the ensuing error. The try/except remains a backstop for a HARD failure (the probe/container dying before it emits JSON → run_channel_probe raises). Applied to run_http_request, run_web_poc, and the tcp/udp socket probe. Tests use the REAL probe result shapes (returned error dicts, not a raising fake) + a hard-failure durability case; a neuter check confirms they fail if the classifier regresses to always-connected. Full offline suite green (1717 passed). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/hexgraph/engine/targets/surfaces.py | 96 ++++++++++++-- tests/test_egress.py | 168 +++++++++++++++++++++++- 2 files changed, 255 insertions(+), 9 deletions(-) diff --git a/src/hexgraph/engine/targets/surfaces.py b/src/hexgraph/engine/targets/surfaces.py index fffc4efc..cf780172 100644 --- a/src/hexgraph/engine/targets/surfaces.py +++ b/src/hexgraph/engine/targets/surfaces.py @@ -233,6 +233,78 @@ def _egress_gate(session, project: Project, target: Target, *, tool: str, task_i return base_url, scope, dest +def _probe_outcome(result) -> tuple[bool, str | None]: + """Classify a channel-probe RESULT as connected vs connection-failed by INSPECTING it. + + Crucial: the sandbox probes report a socket-level failure (connection refused / timeout / + 'No route to host') by RETURNING `{"ok": false, "error": ...}` — they do NOT raise — so an + exception-only check would miss exactly the failures this audit exists to record. We read the + probe's own success signal across its result shapes: + - http single request -> `{"response": {...}}` (http_probe) + - http multi-step web_poc -> `{"steps": [ {...}, ... ]}` + - tcp / udp -> the flat `{"ok": bool, "error"?: ...}` dict itself + Rule: we REACHED the service if any attempt produced a real transport response — an HTTP + `status` (even 4xx/5xx is a genuine reply), or `ok: true` (a TCP banner, or a fire-and-forget + UDP send that the probe deems ok even with no datagram back). We FAILED to connect only when + every attempt reported a socket-level error and none produced a response. Returns + `(connected, error_detail)`.""" + if not isinstance(result, dict): + return True, None # unrecognized shape — never fabricate a failure + steps = result.get("steps") + if isinstance(steps, list): + responses = [r for r in steps if isinstance(r, dict)] + elif isinstance(result.get("response"), dict): + responses = [result["response"]] + else: + responses = [result] + if any(("status" in r) or (r.get("ok") is True) for r in responses): + return True, None + err = next((r.get("error") for r in responses if r.get("error")), None) + return False, err + + +def _probe_with_outcome_audit(session, project: Project, target: Target, *, task_id, tool: str, + dest: str, run): + """Run `run()` (the actual sandboxed network probe) and audit its REAL outcome as a + follow-up EgressEvent, distinct from `_egress_gate`'s pre-flight policy-allow event. + + Without this, `net_list_egress` only ever showed the POLICY decision (allowed=True the + moment the destination passed the allowlist check) — if the probe itself then failed at + the socket layer (e.g. the sandbox container had no route to a lab-private subnet), that + failure surfaced to the caller but left NO trace in the audit log, so `net_list_egress` + misleadingly read "allowed: true" for a request that never actually connected. Recording the + real result here (tool suffixed `:connected`/`:connect_failed`) makes that gap diagnosable + directly from the audit log instead of needing to reproduce the failure by hand. + + The outcome is derived by INSPECTING the probe result (`_probe_outcome`), because the probes + report a connection failure by RETURNING `{"ok": false, ...}`, not by raising; the try/except + is only the backstop for a HARD failure — the container/probe dying before it can emit JSON, in + which case `run_channel_probe` raises. Both events are `durable=True` so the audit survives even + a caller that rolls back on the ensuing error (audit-of-action, the rationale the deny path + already commits under).""" + from hexgraph.engine.audit import record_egress + + try: + result = run() + except Exception as exc: + record_egress(session, project_id=project.id, target_id=target.id, task_id=task_id, + dest=dest, allowed=True, tool=f"{tool}:connect_failed", + detail=f"policy allowed egress, but the probe failed to run: {exc}", + durable=True) + raise + connected, err = _probe_outcome(result) + if connected: + record_egress(session, project_id=project.id, target_id=target.id, task_id=task_id, + dest=dest, allowed=True, tool=f"{tool}:connected", + detail="connection succeeded", durable=True) + else: + record_egress(session, project_id=project.id, target_id=target.id, task_id=task_id, + dest=dest, allowed=True, tool=f"{tool}:connect_failed", + detail=f"policy allowed egress, but the connection attempt failed: " + f"{err or 'unknown error'}", durable=True) + return result + + # Per-session cookie jars for free-form authed exploration via http_request. A fresh # sandbox container runs each request (no state between calls), so the host keeps the jar # and re-injects it. Keyed by (target_id, session_id); in-process (single-user local tool), @@ -272,7 +344,8 @@ def run_http_request(session: Session, project: Project, target: Target, *, requ from hexgraph import settings from hexgraph.sandbox.executor import get_executor - base_url, scope, dest = _egress_gate(session, project, target, tool="http_request", task_id=task_id) + tool = "http_request" + base_url, scope, dest = _egress_gate(session, project, target, tool=tool, task_id=task_id) runner = runner or get_executor() timeout = int(settings.get("features.network.timeout", 30) or 30) @@ -293,8 +366,10 @@ def run_http_request(session: Session, project: Project, target: Target, *, requ channel = {"base_url": base_url, "allow": sorted(scope.allow), "timeout": timeout, "request": req} - result = runner.run_channel_probe("http_probe.py", channel=channel, - net_container=_rehost_container(target)) + result = _probe_with_outcome_audit( + session, project, target, task_id=task_id, tool=tool, dest=dest, + run=lambda: runner.run_channel_probe("http_probe.py", channel=channel, + net_container=_rehost_container(target))) resp = result.get("response") or result if jar is not None and isinstance(resp, dict): jar.update(_parse_set_cookie(resp.get("set_cookie") or [])) @@ -312,13 +387,16 @@ def run_web_poc(session: Session, project: Project, target: Target, *, steps: li from hexgraph import settings from hexgraph.sandbox.executor import get_executor - base_url, scope, dest = _egress_gate(session, project, target, tool="web_poc", task_id=task_id) + tool = "web_poc" + base_url, scope, dest = _egress_gate(session, project, target, tool=tool, task_id=task_id) runner = runner or get_executor() timeout = int(settings.get("features.network.timeout", 30) or 30) channel = {"base_url": base_url, "allow": sorted(scope.allow), "timeout": timeout, "steps": steps, "oracle": oracle} - return runner.run_channel_probe("http_probe.py", channel=channel, - net_container=_rehost_container(target)) + return _probe_with_outcome_audit( + session, project, target, task_id=task_id, tool=tool, dest=dest, + run=lambda: runner.run_channel_probe("http_probe.py", channel=channel, + net_container=_rehost_container(target))) def _device_host(target: Target) -> str | None: @@ -385,8 +463,10 @@ def _run_socket_probe(session: Session, project: Project, target: Target, *, tra channel["oracle"] = oracle if read_bytes is not None: channel["read_bytes"] = int(read_bytes) - return runner.run_channel_probe("tcp_probe.py", channel=channel, - net_container=net_container) + return _probe_with_outcome_audit( + session, project, target, task_id=task_id, tool=tool, dest=dest, + run=lambda: runner.run_channel_probe("tcp_probe.py", channel=channel, + net_container=net_container)) def run_tcp_probe(session: Session, project: Project, target: Target, *, port: int, diff --git a/tests/test_egress.py b/tests/test_egress.py index 764f19c3..38e3d9d2 100644 --- a/tests/test_egress.py +++ b/tests/test_egress.py @@ -42,7 +42,8 @@ def _load_egress(): from hexgraph.agent import mcp_tools as M from hexgraph.engine.audit import list_egress, record_egress from hexgraph.engine.targets.ingest import create_project -from hexgraph.engine.targets.surfaces import register_web_surface +from hexgraph.engine.targets.surfaces import (register_web_surface, run_http_request, + run_tcp_probe, run_web_poc) from hexgraph.engine.tasks import create_task from hexgraph.engine.worker import run_task_sync @@ -128,6 +129,171 @@ def test_web_recon_denied_by_default_and_audited(hg_home): assert events[0].allowed is False and events[0].dest == "127.0.0.1:8080" +class _ChannelRunner: + """Stand-in sandbox executor returning a CANNED run_channel_probe result — the REAL contract: + a socket-level failure comes back as a JSON dict ({"ok": false, "error": ...}), NOT as a raised + exception (the sandbox probes catch and serialize it). `result` is that dict.""" + + def __init__(self, result): + self.result = result + + def run_channel_probe(self, *a, **kw): + return self.result + + +class _HardFailRunner: + """A HARD failure — the container/probe dies before emitting valid JSON, so run_channel_probe + itself RAISES (SandboxError). The audit's try/except backstop must still record connect_failed + (durably) and re-raise.""" + + def run_channel_probe(self, *a, **kw): + from hexgraph.sandbox.runner import SandboxError + + raise SandboxError("probe http_probe did not emit valid JSON") + + +# The real http_probe shapes: a socket failure is a nested {"response": {"ok": false, "error": ...}} +# (NOT a raise); a real reply — including an HTTP error status — carries a `status`. +_HTTP_CONNFAIL = {"tool": "http_probe", "base_url": "http://127.0.0.1:8080", + "response": {"ok": False, "error": "URLError", "dest": "127.0.0.1:8080"}} +_HTTP_OK = {"tool": "http_probe", "base_url": "http://127.0.0.1:8080", + "response": {"status": 200, "body": "ok"}} +_HTTP_500 = {"tool": "http_probe", "base_url": "http://127.0.0.1:8080", + "response": {"status": 500, "body": "boom"}} +# tcp/udp results are FLAT (`ok` at top level, no nested "response"). +_TCP_CONNFAIL = {"tool": "tcp_probe", "host": "127.0.0.1", "port": 8080, "ok": False, + "error": "OSError: [Errno 113] No route to host"} +_TCP_OK = {"tool": "tcp_probe", "host": "127.0.0.1", "port": 8080, "ok": True, "raw": "banner"} + + +def _web_surface(s, name): + p = create_project(s, name=name) + surface = register_web_surface(s, p, "http://127.0.0.1:8080", + endpoints=[{"method": "GET", "path": "/"}]) + return p, surface + + +def _egress_tools(pid): + with session_scope() as s: + return [e.tool for e in (s.query(EgressEvent) + .filter(EgressEvent.project_id == pid) + .order_by(EgressEvent.created_at).all())] + + +def test_http_request_audits_connect_failed_from_returned_error(hg_home): + """The REAL 'No route to host despite allowed:true' case: the probe RETURNS + {"ok": false, "error": ...} (it does NOT raise), so run_http_request returns normally — and + the outcome audit must STILL record :connect_failed. This is exactly what the old + exception-only check missed (it would have logged :connected).""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "connfail") + pid = p.id + resp = run_http_request(s, p, surface, request={"method": "GET", "path": "/"}, + runner=_ChannelRunner(_HTTP_CONNFAIL)) + assert resp.get("ok") is False # returned, not raised — as in production + + tools = _egress_tools(pid) + assert "http_request" in tools # pre-flight policy-allow event + assert "http_request:connect_failed" in tools # the real-outcome event + assert "http_request:connected" not in tools # must NOT mislabel a failure as connected + with session_scope() as s: + failed = next(e for e in s.query(EgressEvent) + .filter(EgressEvent.tool == "http_request:connect_failed").all()) + assert failed.allowed is True # policy DID allow it + assert "URLError" in (failed.detail or "") # the real socket error is captured + + +def test_http_request_audits_connected_on_real_response(hg_home): + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "connok") + pid = p.id + resp = run_http_request(s, p, surface, request={"method": "GET", "path": "/"}, + runner=_ChannelRunner(_HTTP_OK)) + assert resp["status"] == 200 + tools = _egress_tools(pid) + assert "http_request:connected" in tools + assert "http_request:connect_failed" not in tools + + +def test_http_error_status_counts_as_connected(hg_home): + """An HTTP 4xx/5xx is a REAL reply — the connection succeeded, the server just answered with an + error. It must audit :connected, never :connect_failed (guards the 'reached the service' rule).""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "http500") + pid = p.id + run_http_request(s, p, surface, request={"method": "GET", "path": "/"}, + runner=_ChannelRunner(_HTTP_500)) + tools = _egress_tools(pid) + assert "http_request:connected" in tools + assert "http_request:connect_failed" not in tools + + +def test_tcp_probe_audits_connect_failed_from_flat_result(hg_home): + """tcp/udp results are a FLAT {"ok": false, ...} (no nested 'response') — the classifier must + read that shape too, else a network-fuzz/service connect failure would mislabel as connected.""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "tcpfail") + pid = p.id + run_tcp_probe(s, p, surface, port=8080, runner=_ChannelRunner(_TCP_CONNFAIL)) + tools = _egress_tools(pid) + assert "tcp_probe:connect_failed" in tools + assert "tcp_probe:connected" not in tools + + +def test_tcp_probe_audits_connected_on_flat_ok(hg_home): + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "tcpok") + pid = p.id + run_tcp_probe(s, p, surface, port=8080, runner=_ChannelRunner(_TCP_OK)) + tools = _egress_tools(pid) + assert "tcp_probe:connected" in tools + + +def test_web_poc_multistep_connect_failed_when_no_step_reaches(hg_home): + """A multi-step web_poc result is `{"steps": [...]}`; if EVERY step is a socket-level failure + the outcome is :connect_failed. (If any step got a real response, the connection was made — see + the connected variant of this rule in _probe_outcome.)""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + steps_result = {"tool": "http_probe", "base_url": "http://127.0.0.1:8080", + "steps": [{"ok": False, "error": "URLError"}, {"ok": False, "error": "URLError"}]} + with session_scope() as s: + p, surface = _web_surface(s, "pocfail") + pid = p.id + run_web_poc(s, p, surface, steps=[{"method": "GET", "path": "/"}], oracle={}, + runner=_ChannelRunner(steps_result)) + tools = _egress_tools(pid) + assert "web_poc:connect_failed" in tools + assert "web_poc:connected" not in tools + + +def test_outcome_audit_is_durable_across_a_hard_failure_rollback(hg_home): + """A HARD failure (probe/container dies before emitting JSON) makes run_channel_probe RAISE. The + connect_failed audit is durable=True, so it must SURVIVE even though the caller's session_scope + ROLLS BACK on the propagating error — the whole point is a durable trace. Without durable=True + the rollback would discard it.""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + from hexgraph.db.models import Project, Target + from hexgraph.sandbox.runner import SandboxError + + with session_scope() as s: + p, surface = _web_surface(s, "hardfail") + s.commit() # commit stable ids that outlive the rollback below + pid, sid = p.id, surface.id + + with pytest.raises(SandboxError): + with session_scope() as s: # THIS scope rolls back on the raised error + run_http_request(s, s.get(Project, pid), s.get(Target, sid), + request={"method": "GET", "path": "/"}, runner=_HardFailRunner()) + + tools = _egress_tools(pid) + assert "http_request:connect_failed" in tools # survived the rollback (durable=True) + + def test_web_recon_refuses_public_target(hg_home): # even with network enabled, a public target is refused by the scope guard settings.update_settings({"features": {"network": {"enabled": True}}}) From e5a031973c9deebc199e86ac59d57c8280fcc3ae Mon Sep 17 00:00:00 2001 From: branover Date: Fri, 17 Jul 2026 12:53:29 -0400 Subject: [PATCH 2/2] fix: address review nits on the egress outcome audit (N1-N3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - N2 (edge bug): _probe_outcome now returns connected for an EMPTY response set (e.g. an empty steps list) instead of fabricating a :connect_failed — any([]) is False. Regression test added. - N3 (test fidelity): _TCP_OK uses the real tcp-success shape (ok:True + _decode's response/encoding), not a "raw" key the probe never emits. - N1 (completeness): document why run_web_recon is intentionally NOT wrapped in the outcome audit — surface_probe returns a multi-endpoint liveness sweep, not a single connect outcome, so _probe_outcome doesn't model it; wrapping it is a follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/hexgraph/engine/targets/surfaces.py | 7 +++++++ tests/test_egress.py | 18 +++++++++++++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/hexgraph/engine/targets/surfaces.py b/src/hexgraph/engine/targets/surfaces.py index cf780172..89e1133f 100644 --- a/src/hexgraph/engine/targets/surfaces.py +++ b/src/hexgraph/engine/targets/surfaces.py @@ -257,6 +257,8 @@ def _probe_outcome(result) -> tuple[bool, str | None]: responses = [result["response"]] else: responses = [result] + if not responses: + return True, None # nothing to classify (e.g. an empty steps list) — never fabricate a failure if any(("status" in r) or (r.get("ok") is True) for r in responses): return True, None err = next((r.get("error") for r in responses if r.get("error")), None) @@ -558,6 +560,11 @@ def run_web_recon(session: Session, project: Project, target: Target, task=None, "endpoints": [{"method": e.get("method", "GET"), "path": e.get("path", "/")} for e in endpoints], "timeout": timeout} + # NOT wrapped in _probe_with_outcome_audit (unlike http_request/web_poc/tcp): surface_probe + # returns a MULTI-endpoint liveness sweep ({"probes": [{"alive": bool, ...}, ...]}), not a + # single connect outcome, so _probe_outcome doesn't model this shape — a naive wrap would + # mislabel a fully-alive recon as :connect_failed. Modelling the sweep (connected = any probe + # alive) and wrapping it is a follow-up; the pre-flight policy-allow event above still audits. result = runner.run_channel_probe("surface_probe.py", channel=channel, net_container=_rehost_container(target)) diff --git a/tests/test_egress.py b/tests/test_egress.py index 38e3d9d2..dc6bb56b 100644 --- a/tests/test_egress.py +++ b/tests/test_egress.py @@ -163,7 +163,9 @@ def run_channel_probe(self, *a, **kw): # tcp/udp results are FLAT (`ok` at top level, no nested "response"). _TCP_CONNFAIL = {"tool": "tcp_probe", "host": "127.0.0.1", "port": 8080, "ok": False, "error": "OSError: [Errno 113] No route to host"} -_TCP_OK = {"tool": "tcp_probe", "host": "127.0.0.1", "port": 8080, "ok": True, "raw": "banner"} +# tcp SUCCESS: the real probe emits ok:True + the _decode fields (response/encoding), not "raw". +_TCP_OK = {"tool": "tcp_probe", "host": "127.0.0.1", "port": 8080, "ok": True, + "response": "banner", "encoding": "utf-8"} def _web_surface(s, name): @@ -271,6 +273,20 @@ def test_web_poc_multistep_connect_failed_when_no_step_reaches(hg_home): assert "web_poc:connected" not in tools +def test_web_poc_empty_steps_is_not_a_connection_failure(hg_home): + """An empty `steps` result must NOT be labelled a connection failure — there was nothing to + classify, and `any([])` is False, so the classifier must default to connected (regression guard).""" + settings.update_settings({"features": {"network": {"enabled": True}}}) + with session_scope() as s: + p, surface = _web_surface(s, "pocempty") + pid = p.id + run_web_poc(s, p, surface, steps=[{"method": "GET", "path": "/"}], oracle={}, + runner=_ChannelRunner({"tool": "http_probe", "steps": []})) + tools = _egress_tools(pid) + assert "web_poc:connect_failed" not in tools + assert "web_poc:connected" in tools + + def test_outcome_audit_is_durable_across_a_hard_failure_rollback(hg_home): """A HARD failure (probe/container dies before emitting JSON) makes run_channel_probe RAISE. The connect_failed audit is durable=True, so it must SURVIVE even though the caller's session_scope