R4 (e2e half): 3-hop nested-SSH circuit delivery + per-hop pcaps (gate item 1)

run_circuit_fixture stands up an isolated docker N-hop nested-SSH chain, pipes a
seed-deterministic self-generated payload through it, verifies end-to-end
delivery, captures + checksums a per-hop pcap, emits R3 events, and always tears
the circuit down. Containment is load-bearing: assert_isolated refuses
local/unknown engines up front, every hop is built through the ForwarderPlan
guard, only our own fixture bytes move between our own containers, and the runner
is wired for docker only (multipass e2e refused, not faked). Adds the sor-hop
fixture image (alpine + sshd + tcpdump, lab-relay only) and a docker-gated
acceptance test that skips where no daemon/image is present. Instrument-validation
gate item 1 GREEN: live 3-hop delivery verified, 3 distinct hops, pcaps checksum.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd_chat/sor/fixtures/hop.Dockerfile

@@ -0,0 +1,23 @@
+# R4 fixture — one isolated SOR relay hop.
+#
+# A minimal Alpine node running sshd, used ONLY as a lab relay for
+# self-generated fixture traffic inside the isolated engine (docker). It never
+# runs on the host (the forwarder guard refuses engine == local) and is torn
+# down after each run. openssh-client + tcpdump are present so the node can be a
+# nested-SSH jump host and so per-hop pcaps can be captured for the linkability
+# measurement (all traffic is SSH-encrypted self-traffic).
+FROM alpine:latest
+
+# Alpine's stock sshd_config ships `AllowTcpForwarding no` (and sshd honours the
+# FIRST occurrence of a keyword), so we strip any pre-set copies of the keywords
+# we care about before appending our nested-SSH relay policy — otherwise the
+# jump-host onward channel is refused ("stdio forwarding failed").
+RUN apk add --no-cache openssh openssh-client tcpdump \
+ && ssh-keygen -A \
+ && mkdir -p /root/.ssh && chmod 700 /root/.ssh \
+ && sed -i -E '/^[#[:space:]]*(AllowTcpForwarding|PermitRootLogin|PubkeyAuthentication|PasswordAuthentication|UseDNS)\b/d' /etc/ssh/sshd_config \
+ && printf '\n# --- SOR relay policy ---\nPermitRootLogin prohibit-password\nPubkeyAuthentication yes\nPasswordAuthentication no\nAllowTcpForwarding yes\nUseDNS no\n' \
+      >> /etc/ssh/sshd_config
+
+EXPOSE 22
+CMD ["/usr/sbin/sshd", "-D", "-e"]

cmd_chat/sor/forwarder.py

@@ -1,30 +1,43 @@
-"""R4 (partial) — containment guard for the SOR nested-SSH forwarder.
+"""R4 — SSH-tunnel data plane (nested-circuit forwarder), isolated-engine-only.
 
-This module lands **only instrument-validation gate item 5**: forwarders are
+Two parts, both gated by the same containment invariant (``assert engine !=
-*isolated-engine-only*. It is the guard that every forwarder MUST pass before it
+local`` or refuse):
-does anything — ``assert engine != local`` or refuse. It moves no traffic, opens
-no socket, spawns no engine, and builds no SSH chain.
 
-The traffic-moving half of R4 (gate item 1: a 3-hop self-traffic circuit across
+1. **The guard** (gate item 5): ``assert_isolated`` / ``isolation_prefix`` /
-the grid delivering a known payload e2e, with per-hop pcaps) is intentionally
+   ``ForwarderPlan`` — every forwarder must pass this before it does anything.
-**NOT** implemented here — it is HELD for a live isolated engine + grid (see
+   There is no code path here that returns an exec prefix for the host; ``local``
-OVERSEER-STATUS.md). Committing an unrunnable forwarder would violate the build
+   (the warned exception the *chat* sandbox allows, ``bridge.py:540``) is never
-discipline; this guard, by contrast, is fully verifiable offline: constructing a
+   valid for a SOR run. This half is fully verifiable offline.
-forwarder plan for a ``local`` (or unknown) engine refuses, for an isolated
-engine it yields the isolation exec prefix.
 
-Containment law (CLAUDE.md §Containment #1) is encoded structurally: there is no
+2. **The e2e circuit runner** (gate item 1): ``run_circuit_fixture`` stands up an
-code path in this module that returns an exec prefix for the host. ``local`` —
+   ``N``-hop nested-SSH chain of isolated **docker** containers, pipes a
-the explicit warned exception the *chat* sandbox allows (``bridge.py:540``) — is
+   seed-deterministic *self-generated* payload through it, captures a per-hop
-never valid for a SOR run. The allow-list is imported from ``provenance`` so the
+   pcap, verifies end-to-end delivery, checksums each pcap, emits R3 events, and
-forwarder and the run manifest can never disagree about what counts as isolated.
+   tears the circuit down. It moves only fixture traffic between our own
+   containers on our own engine — no external target, no real user data, no
+   forwarder on the host.
+
+Containment (CLAUDE.md §Containment) is load-bearing: the runner refuses any
+non-isolated engine up front, uses the docker control plane only to manage
+containers (the forwarder *processes* — ssh/tcpdump — run inside containers, not
+on the host), and always tears the circuit down. The isolated-engine allow-list
+is imported from ``provenance`` so the forwarder and the run manifest can never
+disagree about what counts as isolated.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+import hashlib
-from typing import List
+import json
+import shutil
+import subprocess
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional
 
+from cmd_chat.sor.config import Domain, SorRng
+from cmd_chat.sor.events import EventLog
 from cmd_chat.sor.provenance import ISOLATED_ENGINES
 
 # The host. Never a valid engine for a SOR forwarder — its whole purpose is to
@@ -98,3 +111,231 @@ class ForwarderPlan:
     def exec_prefix(self) -> List[str]:
         """The isolated exec argv prefix for this plan (validated at build time)."""
         return isolation_prefix(self.engine, self.container)
+
+
+# --------------------------------------------------------------------------- #
+# The e2e circuit runner (gate item 1).
+#
+# Stands up an N-hop nested-SSH chain of isolated docker containers, pipes a
+# seed-deterministic self-generated payload through it, captures a per-hop pcap,
+# verifies end-to-end delivery, checksums each pcap, emits R3 events, and tears
+# the circuit down. Every container it creates is a ForwarderPlan-gated hop; the
+# host (`local`) is refused up front. Traffic is our own fixture bytes moving
+# between our own containers on our own engine — no external target, no real
+# user data, no forwarder process on the host.
+# --------------------------------------------------------------------------- #
+
+# The fixture relay image (built from fixtures/hop.Dockerfile): alpine + sshd +
+# openssh-client + tcpdump. Only ever used as an isolated lab relay for
+# self-traffic; it never runs on the host.
+HOP_IMAGE = "sor-hop:latest"
+
+# ssh hardening applied inside the *client* container only (never the host): the
+# lab hops use throwaway host keys, so the client accepts-on-first-use and keeps
+# no known_hosts. This is a containment-internal convenience, not a security
+# posture we ship to anyone.
+_CLIENT_SSH_CONFIG = (
+    "Host *\n"
+    "    StrictHostKeyChecking accept-new\n"
+    "    UserKnownHostsFile /dev/null\n"
+    "    LogLevel ERROR\n"
+)
+
+
+class CircuitError(RuntimeError):
+    """The e2e circuit could not be stood up, delivered, or verified. Raised
+    (never silently swallowed) so a failed run is loud; teardown still runs."""
+
+
+def _docker(*args: str, check: bool = True, capture: bool = True,
+            input_bytes: Optional[bytes] = None, timeout: int = 120) -> subprocess.CompletedProcess:
+    """Run a single `docker ...` control-plane command. Args are passed as
+    separate argv items (never a shell string) so container/circuit names can't
+    inject shell metacharacters."""
+    return subprocess.run(
+        ["docker", *args],
+        check=check,
+        capture_output=capture,
+        input=input_bytes,
+        timeout=timeout,
+    )
+
+
+@dataclass
+class CircuitResult:
+    """The verifiable outcome of one e2e fixture run (all offline-checkable)."""
+
+    run_id: str
+    seed: int
+    engine: str
+    delivered: bool
+    payload_sha256: str
+    received_sha256: str
+    hop_containers: List[str]
+    pcaps: Dict[int, Path] = field(default_factory=dict)
+    pcap_sha256: Dict[int, str] = field(default_factory=dict)
+    events_sha256: Optional[str] = None
+
+
+def _seed_payload(seed: int, size: int = 4096) -> bytes:
+    """A self-generated, seed-deterministic payload (no real data). Drawn from
+    the R1 PADDING stream so the same seed yields the same bytes on every run."""
+    stream = SorRng(seed).stream(Domain.PADDING)
+    return bytes(stream.next_below(256) for _ in range(size))
+
+
+def run_circuit_fixture(
+    seed: int,
+    *,
+    engine: str = "docker",
+    hops: int = 3,
+    out_root: Optional[Path] = None,
+    payload_size: int = 4096,
+    keep: bool = False,
+) -> CircuitResult:
+    """Drive instrument-validation gate item 1: an ``hops``-hop nested-SSH circuit
+    of isolated containers delivers a seed-deterministic payload end-to-end, with
+    a per-hop pcap captured + checksummed and R3 events emitted.
+
+    Containment (load-bearing):
+      * ``assert_isolated(engine)`` up front — ``local`` and unknown engines are
+        refused before any container is created; only ``docker`` is wired for e2e.
+      * every hop is built through :class:`ForwarderPlan`, so each exec target is
+        re-validated as isolated.
+      * only self-generated fixture bytes move, between our own containers, and
+        the circuit is always torn down (``finally``) unless ``keep=True``.
+
+    Returns a :class:`CircuitResult`; raises :class:`CircuitError` on any failure
+    (teardown still runs). Requires a live docker daemon and the ``sor-hop`` image
+    — callers/tests that lack them should skip."""
+    assert_isolated(engine)
+    if engine != "docker":
+        # multipass e2e is not built; refuse rather than pretend.
+        raise CircuitError(
+            f"containment: e2e circuit runner only wired for docker, not {engine!r}"
+        )
+    if shutil.which("docker") is None:
+        raise CircuitError("docker control plane not found on PATH")
+    if hops < 3:
+        raise CircuitError(f"gate item 1 requires >=3 hops, got {hops}")
+
+    run_id = f"sorfix-{seed:016x}-{int(time.time())}"
+    out_root = Path(out_root) if out_root else Path("output/sor-runs")
+    run_dir = out_root / run_id
+    pcap_dir = run_dir / "pcap"
+    pcap_dir.mkdir(parents=True, exist_ok=True)
+
+    net = f"sorfix-net-{run_id}"
+    client = f"sorfix-client-{run_id}"
+    hop_names = [f"sorfix-hop{i}-{run_id}" for i in range(hops)]
+    hop_alias = [f"hop{i}" for i in range(hops)]
+
+    payload = _seed_payload(seed, payload_size)
+    payload_sha = hashlib.sha256(payload).hexdigest()
+
+    log = EventLog(run_dir, run_id, seed)
+    result = CircuitResult(
+        run_id=run_id,
+        seed=seed,
+        engine=engine,
+        delivered=False,
+        payload_sha256=payload_sha,
+        received_sha256="",
+        hop_containers=list(hop_names),
+    )
+
+    created: List[str] = []
+    net_created = False
+    try:
+        # 1) Isolated user-defined network so hops resolve each other by alias.
+        _docker("network", "create", net)
+        net_created = True
+
+        # 2) Bring up client + hop containers (each hop is a ForwarderPlan-gated
+        #    isolated target; building the plan re-asserts containment).
+        for name, alias in [(client, "client"), *zip(hop_names, hop_alias)]:
+            plan = ForwarderPlan(engine=engine, container=name)  # re-validates isolation
+            _docker(
+                "run", "-d", "--rm",
+                "--name", name,
+                "--hostname", alias,
+                "--network", net,
+                "--network-alias", alias,
+                "--cap-add", "NET_RAW",  # tcpdump; present by default but explicit
+                HOP_IMAGE,
+            )
+            created.append(name)
+            # Sanity: the plan's exec prefix targets this isolated container.
+            assert plan.exec_prefix()[:3] == ["docker", "exec", "-i"]
+
+        # 3) Client keypair + accept-on-first-use ssh config (client only).
+        _docker("exec", client, "sh", "-c",
+                "ssh-keygen -t ed25519 -N '' -f /root/.ssh/id_ed25519 -q")
+        _docker("exec", "-i", client, "sh", "-c",
+                "cat > /root/.ssh/config && chmod 600 /root/.ssh/config",
+                input_bytes=_CLIENT_SSH_CONFIG.encode())
+        pub = _docker("exec", client, "cat", "/root/.ssh/id_ed25519.pub").stdout
+
+        # 4) Authorize the client key on every hop.
+        for name in hop_names:
+            _docker("exec", "-i", name, "sh", "-c",
+                    "cat >> /root/.ssh/authorized_keys && chmod 600 /root/.ssh/authorized_keys",
+                    input_bytes=pub)
+
+        log.emit("circuit_build", circuit_id=run_id, hop_index=hops)
+
+        # 5) Start a per-hop pcap (adjacent-hop SSH ciphertext only — the nested
+        #    onion property). Detached so capture spans the transfer.
+        for i, name in enumerate(hop_names):
+            _docker("exec", "-d", name, "sh", "-c",
+                    "tcpdump -i eth0 -w /cap.pcap 'tcp port 22' >/dev/null 2>&1")
+            log.emit("hop_add", node_fp=hop_alias[i].encode().hex()[:8],
+                     circuit_id=run_id, hop_index=i, bytes_=payload_size)
+        time.sleep(1.0)  # let tcpdump bind before traffic
+
+        # 6) Stage payload in the client, push it through the nested-SSH chain to
+        #    the final hop, read it back. ProxyJump nests hop0->hop1->...->hopN-1.
+        _docker("exec", "-i", client, "sh", "-c",
+                "cat > /tmp/payload", input_bytes=payload)
+        jumps = ",".join(f"root@{a}" for a in hop_alias[:-1])
+        final = f"root@{hop_alias[-1]}"
+        t0 = time.time()
+        _docker("exec", client, "sh", "-c",
+                f"ssh -J {jumps} {final} 'cat > /tmp/recv' < /tmp/payload")
+        recv_sha = _docker("exec", hop_names[-1], "sha256sum", "/tmp/recv").stdout
+        latency_ms = (time.time() - t0) * 1000.0
+        received_sha = recv_sha.decode().split()[0]
+        result.received_sha256 = received_sha
+        result.delivered = received_sha == payload_sha
+        log.emit("bridge_forward", circuit_id=run_id, hop_index=hops - 1,
+                 bytes_=payload_size, latency_ms=latency_ms,
+                 decision="delivered" if result.delivered else "corrupt")
+
+        if not result.delivered:
+            raise CircuitError(
+                f"e2e delivery mismatch: sent {payload_sha[:12]} got {received_sha[:12]}"
+            )
+
+        # 7) Stop capture, copy each pcap out, checksum it (write-once artifact).
+        for i, name in enumerate(hop_names):
+            _docker("exec", name, "sh", "-c", "pkill tcpdump || true", check=False)
+        time.sleep(0.5)
+        for i, name in enumerate(hop_names):
+            dst = pcap_dir / f"hop{i}.pcap"
+            _docker("cp", f"{name}:/cap.pcap", str(dst))
+            result.pcaps[i] = dst
+            result.pcap_sha256[i] = hashlib.sha256(dst.read_bytes()).hexdigest()
+
+        result.events_sha256 = log.close()
+        return result
+    except subprocess.CalledProcessError as exc:  # noqa: PERF203
+        stderr = exc.stderr.decode(errors="replace") if exc.stderr else ""
+        raise CircuitError(f"docker control-plane step failed: {exc} {stderr}") from exc
+    finally:
+        if not log._closed:
+            log.close()
+        if not keep:
+            for name in created:
+                _docker("rm", "-f", name, check=False, timeout=60)
+            if net_created:
+                _docker("network", "rm", net, check=False, timeout=60)

tests/test_sor_forwarder_e2e.py

@@ -0,0 +1,110 @@
+"""R4 (e2e half) — 3-hop nested-SSH circuit delivery acceptance check.
+
+Acceptance predicate (instrument-validation gate item 1, the traffic-moving half
+of roadmap R4): a >=3-hop self-traffic circuit of *isolated* containers delivers
+a known, seed-deterministic payload end-to-end, and a per-hop pcap exists and
+checksums. This suite drives the real ``run_circuit_fixture`` against a live
+docker engine and asserts exactly that conjunct.
+
+Containment is still law here: the runner refuses ``local``/unknown engines up
+front (re-checked offline below), only self-generated fixture bytes move, they
+move only between our own containers on our own engine, and the circuit is always
+torn down. The e2e cell needs a live docker daemon + the ``sor-hop`` image; it is
+skipped (not failed) where either is absent, so the offline suite stays green on
+any host.
+"""
+
+import shutil
+import subprocess
+
+import pytest
+
+from cmd_chat.sor.config import Domain, SorRng
+from cmd_chat.sor.forwarder import (
+    HOP_IMAGE,
+    CircuitError,
+    _seed_payload,
+    run_circuit_fixture,
+)
+
+
+def _docker_ready() -> bool:
+    if shutil.which("docker") is None:
+        return False
+    try:
+        subprocess.run(["docker", "info"], capture_output=True, timeout=15, check=True)
+    except Exception:
+        return False
+    return True
+
+
+def _image_present() -> bool:
+    try:
+        out = subprocess.run(
+            ["docker", "image", "inspect", HOP_IMAGE],
+            capture_output=True, timeout=15, check=False,
+        )
+    except Exception:
+        return False
+    return out.returncode == 0
+
+
+live = pytest.mark.skipif(
+    not (_docker_ready() and _image_present()),
+    reason="requires a live docker daemon and the sor-hop:latest image",
+)
+
+
+# --------------------------------------------------------------------------- #
+# Offline: containment refusals hold even for the e2e entry point.
+# --------------------------------------------------------------------------- #
+def test_e2e_refuses_local_engine():
+    with pytest.raises(Exception):
+        run_circuit_fixture(1, engine="local")
+
+
+def test_e2e_refuses_non_docker_isolated_engine():
+    # multipass is isolated but the e2e runner is only wired for docker.
+    with pytest.raises(CircuitError):
+        run_circuit_fixture(1, engine="multipass")
+
+
+def test_e2e_refuses_fewer_than_three_hops():
+    with pytest.raises(CircuitError):
+        run_circuit_fixture(1, engine="docker", hops=2)
+
+
+def test_seed_payload_is_deterministic_self_traffic():
+    # Same seed -> identical fixture bytes (no real data); different seed differs.
+    assert _seed_payload(42) == _seed_payload(42)
+    assert _seed_payload(42) != _seed_payload(43)
+    # It is exactly the R1 PADDING stream (self-generated, reproducible).
+    stream = SorRng(42).stream(Domain.PADDING)
+    assert _seed_payload(42, 16) == bytes(stream.next_below(256) for _ in range(16))
+
+
+# --------------------------------------------------------------------------- #
+# Gate item 1 — the live conjunct.
+# --------------------------------------------------------------------------- #
+@live
+def test_three_hop_circuit_delivers_and_checksums(tmp_path):
+    seed = 0xC0FFEE
+    result = run_circuit_fixture(seed, engine="docker", hops=3, out_root=tmp_path)
+
+    # (a) end-to-end delivery of the known payload.
+    assert result.delivered is True
+    assert result.received_sha256 == result.payload_sha256
+
+    # (b) a per-hop pcap exists and checksums (3 distinct hops).
+    assert set(result.pcaps.keys()) == {0, 1, 2}
+    for i in (0, 1, 2):
+        pcap = result.pcaps[i]
+        assert pcap.exists() and pcap.stat().st_size > 0
+        assert len(result.pcap_sha256[i]) == 64
+
+    # (c) provenance: the R3 event log was sealed.
+    assert result.events_sha256 and len(result.events_sha256) == 64
+
+    # (d) containment: three distinct isolated containers, engine never local.
+    assert result.engine == "docker"
+    assert len(set(result.hop_containers)) == 3