diff --git a/cmd_chat/sor/fixtures/hop.Dockerfile b/cmd_chat/sor/fixtures/hop.Dockerfile new file mode 100644 index 0000000..a9546f7 --- /dev/null +++ b/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"] diff --git a/cmd_chat/sor/forwarder.py b/cmd_chat/sor/forwarder.py index d3dac6b..a0c4c95 100644 --- a/cmd_chat/sor/forwarder.py +++ b/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 -*isolated-engine-only*. It is the guard that every forwarder MUST pass before it -does anything — ``assert engine != local`` or refuse. It moves no traffic, opens -no socket, spawns no engine, and builds no SSH chain. +Two parts, both gated by the same containment invariant (``assert engine != +local`` or refuse): -The traffic-moving half of R4 (gate item 1: a 3-hop self-traffic circuit across -the grid delivering a known payload e2e, with per-hop pcaps) is intentionally -**NOT** implemented here — it is HELD for a live isolated engine + grid (see -OVERSEER-STATUS.md). Committing an unrunnable forwarder would violate the build -discipline; this guard, by contrast, is fully verifiable offline: constructing a -forwarder plan for a ``local`` (or unknown) engine refuses, for an isolated -engine it yields the isolation exec prefix. +1. **The guard** (gate item 5): ``assert_isolated`` / ``isolation_prefix`` / + ``ForwarderPlan`` — every forwarder must pass this before it does anything. + There is no code path here that returns an exec prefix for the host; ``local`` + (the warned exception the *chat* sandbox allows, ``bridge.py:540``) is never + valid for a SOR run. This half is fully verifiable offline. -Containment law (CLAUDE.md §Containment #1) is encoded structurally: there is no -code path in this module that returns an exec prefix for the host. ``local`` — -the explicit warned exception the *chat* sandbox allows (``bridge.py:540``) — is -never valid for a SOR run. The allow-list is imported from ``provenance`` so the -forwarder and the run manifest can never disagree about what counts as isolated. +2. **The e2e circuit runner** (gate item 1): ``run_circuit_fixture`` 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. 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 -from typing import List +import hashlib +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) diff --git a/tests/test_sor_forwarder_e2e.py b/tests/test_sor_forwarder_e2e.py new file mode 100644 index 0000000..d634c86 --- /dev/null +++ b/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