feat: audit the real connection outcome of egress probes (net_list_egress)

src/hexgraph/engine/targets/surfaces.py

@@ -233,6 +233,80 @@ 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 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)
+    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 +346,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 +368,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 +389,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 +465,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,
@@ -478,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))
 

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,187 @@ 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 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):
+    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_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
+    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}}})