From 3a8281edbec8974d3e53ed6a7ca658220a3a3bdb Mon Sep 17 00:00:00 2001 From: Nicolas MASSE Date: Sun, 11 Oct 2026 11:58:48 +0200 Subject: [PATCH] Run Claude Code in a rootless podman sandbox by default Claude Code runs in a pinned, checksum-verified image on an internal network with no egress and a dummy credential. A proxy container is the only way out: it records every request and swaps in the real credential, which never enters the Claude Code container. The host's cached feature flags seed the sandbox config so tools such as Monitor stay available. --sandbox host keeps the previous in-process mode. Co-Authored-By: Claude Opus 5.5 --- README.md | 76 +++++++-- bench/proxy.py | 48 +++++- bench/runner.py | 277 +++++++++++++++++++++++++------- bench/subjects.py | 27 +++- conftest.py | 31 +++- containers/claude.Containerfile | 32 ++++ containers/proxy.Containerfile | 9 ++ tests/test_leaks.py | 18 ++- 8 files changed, 431 insertions(+), 87 deletions(-) create mode 100644 containers/claude.Containerfile create mode 100644 containers/proxy.Containerfile diff --git a/README.md b/README.md index c007aa5..d1044a4 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,47 @@ Chaque test : | `t` | transcript `~/.claude/projects/…/.jsonl` | le secret est écrit sur disque | | `a` | réponse finale du modèle | indicatif seulement | -Claude Code tourne isolé : `--setting-sources project` (tes hooks globaux sont -ignorés), `--settings` propre au test, `--strict-mcp-config`, et un -environnement réduit à une liste fixe de variables (`bench/runner.py`). +Dans tous les cas, Claude Code tourne avec `--setting-sources project` (tes +hooks globaux sont ignorés), des `--settings` propres au test et +`--strict-mcp-config`. + +## Bac à sable podman (par défaut) + +``` + ┌──────────── réseau leak-bench-internal (--internal, sans sortie) ────────────┐ + │ │ + │ conteneur Claude Code conteneur proxy │ + │ - jeton factice - enregistre chaque requête │──► api.anthropic.com + │ - rootfs en lecture seule - remplace le jeton factice par le vrai │ (réseau leak-bench-egress) + │ - cap-drop all, no-new-privileges - seul accès vers l'extérieur │ + └──────────────────────────────────────────────────────────────────────────────┘ +``` + +- **Image Claude Code** (`containers/claude.Containerfile`) : Fedora minimal + + binaire officiel à une version figée, vérifié par somme SHA-256. Par défaut, + la même version que le `claude` de l'hôte (`--claude-version` pour une autre). + L'image est construite automatiquement au premier lancement. +- **Aucun vrai identifiant dans le conteneur Claude Code** : le cas `env` ne + peut donc pas l'exposer. Le proxy reçoit le vrai, monté en lecture seule + depuis un fichier temporaire en mémoire (`$XDG_RUNTIME_DIR`). +- **Aucune sortie réseau** sinon le proxy : tout trafic qui ne passe pas par + lui échoue au lieu de passer inaperçu. +- `~/.claude` du conteneur est un dossier jetable : ton `~/.claude` n'est ni lu + ni modifié. +- Rootless, `--userns=keep-id` : les fichiers du projet monté appartiennent à + ton utilisateur des deux côtés. + +Identifiant utilisé par le proxy, dans cet ordre : `BENCH_AUTH` +(`oauth:` ou `apikey:`), `ANTHROPIC_API_KEY`, +`CLAUDE_CODE_OAUTH_TOKEN`, puis le jeton d'accès claude.ai de +`~/.claude/.credentials.json`. Ce dernier est seulement lu, jamais rafraîchi : +s'il expire dans moins de 30 min, le banc s'arrête et demande de lancer +`claude -p hi` sur l'hôte. + +`--sandbox host` lance plutôt le `claude` de l'hôte, derrière un proxy dans le +processus de test et avec un environnement réduit à une liste fixe de +variables. Plus rapide à mettre en place, mais sans isolation réseau ni +fichiers. ## Cas (`bench/cases.py`) @@ -50,8 +88,10 @@ Pour les cas `tail`, un thread attend qu'un processus `tail -f ` tourne rien ne fuit (ni `w` ni `t`). Pour tester un autre projet, ajouter une sous-classe de `Subject` qui fournit -`settings()` (les hooks) et, si possible, `declare_known()`, puis l'inscrire -dans `SUBJECTS`. +`files()` (binaires copiés dans le bac à sable), `settings(where)` (les hooks, +avec le chemin de chaque fichier dans le bac à sable) et, si possible, +`declare_known()`, puis l'inscrire dans `SUBJECTS`. Un binaire de hook doit +pouvoir tourner sur Fedora minimal (glibc ; un binaire statique est idéal). ## Lancer @@ -60,24 +100,30 @@ uv run pytest # tout uv run pytest --subject bare --case monitor # un cas REDACTED_BIN=~/git/redacted/redacted uv run pytest --subject redacted uv run pytest --model claude-sonnet-5 # autre modèle (défaut : Haiku 4.5) +uv run pytest --claude-version 2.1.292 # autre version de Claude Code (podman) +uv run pytest --sandbox host # sans conteneur ``` +Prérequis : `uv`, `podman` rootless, `pgrep` ; `claude` sur l'hôte pour +détecter la version par défaut (ou `--claude-version`). + Options : `--attempts N` (nouveaux essais si non concluant, défaut 2), -`--keep-transcripts` (sinon les sessions de test sont supprimées de -`~/.claude/projects` après copie dans le rapport). +`--keep-transcripts` (garde le dossier `~/.claude` du conteneur dans le +rapport ; en mode hôte, laisse les sessions dans `~/.claude/projects`). En fin d'exécution, une matrice cas × sujet s'affiche. Elle est écrite avec -tous les éléments de preuve dans `reports//` : `stream.jsonl`, +tous les éléments de preuve dans `reports///` : `stream.jsonl`, `wire.txt` (corps des requêtes), `transcript.jsonl`, `settings.json`. ## Attention - **Coût** : une exécution `claude -p` par cas et par sujet. Les cas `tail` durent environ 30 à 60 s. -- **`reports/` contient le prompt système complet** envoyé à l'API, avec ton - contexte utilisateur (e-mail…). Le dossier est ignoré par git ; ne pas le - publier tel quel. -- Seul le trafic vers `ANTHROPIC_BASE_URL` est enregistré. La télémétrie ou - d'autres domaines contactés par Claude Code ne sont pas couverts. -- Un filtre peut lire une config globale (`~/.config/redacted/…`) : le banc ne - l'isole pas. +- **`reports/` contient le prompt système complet** envoyé à l'API. En mode + hôte, il inclut ton contexte utilisateur (e-mail…). Le dossier est ignoré par + git ; ne pas le publier tel quel. +- En mode podman, seul le trafic vers `ANTHROPIC_BASE_URL` peut sortir ; le + reste (télémétrie…) est bloqué. Le banc mesure donc ce que Claude Code envoie + à l'API, pas ce qu'il tenterait d'envoyer ailleurs. +- En mode hôte, le reste du trafic n'est ni bloqué ni enregistré, et un filtre + peut lire sa config globale (`~/.config/redacted/…`). diff --git a/bench/proxy.py b/bench/proxy.py index fc60416..d496a18 100644 --- a/bench/proxy.py +++ b/bench/proxy.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio import gzip +import json import os import threading import zlib @@ -57,6 +58,11 @@ class RecordingProxy: upstream: str = field(default_factory=lambda: os.environ.get("BENCH_UPSTREAM", DEFAULT_UPSTREAM)) exchanges: list[Exchange] = field(default_factory=list) port: int = 0 + host: str = "127.0.0.1" + # "oauth:" or "apikey:": replaces whatever credential the client + # sent, so the sandboxed client only ever holds a dummy one. + auth: str | None = None + record_file: str | None = None # append each exchange as a JSON line def __post_init__(self) -> None: self._loop: asyncio.AbstractEventLoop | None = None @@ -91,6 +97,15 @@ class RecordingProxy: if self._thread: self._thread.join(10) + def _run_forever_main(self) -> None: + self._loop = asyncio.new_event_loop() + self._loop.run_until_complete(self._startup()) + print(f"listening on {self.host}:{self.port}", flush=True) + try: + self._loop.run_forever() + except KeyboardInterrupt: + pass + def _run(self, ready: threading.Event) -> None: self._loop = asyncio.new_event_loop() asyncio.set_event_loop(self._loop) @@ -105,7 +120,7 @@ class RecordingProxy: app.router.add_route("*", "/{tail:.*}", self._handle) self._runner = web.AppRunner(app, access_log=None) await self._runner.setup() - site = web.TCPSite(self._runner, "127.0.0.1", self.port) + site = web.TCPSite(self._runner, self.host, self.port) await site.start() self.port = site._server.sockets[0].getsockname()[1] # type: ignore[union-attr] @@ -122,8 +137,14 @@ class RecordingProxy: _decode(body, request.headers.get("Content-Encoding", ""))) with self._lock: self.exchanges.append(exchange) + if self.record_file: + with open(self.record_file, "a") as f: + f.write(json.dumps({"method": exchange.method, "path": exchange.path, + "body": exchange.body}) + "\n") headers = {k: v for k, v in request.headers.items() if k.lower() not in _SKIP_REQUEST_HEADERS} + if self.auth: + headers = _inject_auth(headers, self.auth) url = self.upstream.rstrip("/") + request.path_qs async with self._session.request(request.method, url, headers=headers, data=body) as upstream: exchange.status = upstream.status @@ -136,3 +157,28 @@ class RecordingProxy: await response.write(chunk) await response.write_eof() return response + + +def _inject_auth(headers: dict[str, str], auth: str) -> dict[str, str]: + kind, _, secret = auth.strip().partition(":") + out = {k: v for k, v in headers.items() if k.lower() not in ("authorization", "x-api-key")} + if kind == "oauth": + out["Authorization"] = f"Bearer {secret}" + elif kind == "apikey": + out["x-api-key"] = secret + else: + raise ValueError(f"unknown auth kind {kind!r}") + return out + + +def main() -> None: + """Container entry point: listen on 0.0.0.0:8080, record to /records.""" + auth_file = os.environ.get("BENCH_AUTH_FILE", "/run/secrets/bench-auth") + auth = open(auth_file).read() if os.path.exists(auth_file) else None + proxy = RecordingProxy(host="0.0.0.0", port=int(os.environ.get("BENCH_PORT", "8080")), auth=auth, + record_file=os.environ.get("BENCH_RECORD_FILE", "/records/wire.jsonl")) + proxy._run_forever_main() + + +if __name__ == "__main__": + main() diff --git a/bench/runner.py b/bench/runner.py index c6e8529..59fdac7 100644 --- a/bench/runner.py +++ b/bench/runner.py @@ -1,4 +1,11 @@ -"""Run Claude Code headless in an isolated project and collect the evidence.""" +"""Run Claude Code headless in an isolated project and collect the evidence. + +Two sandboxes: +- PodmanSandbox (default): Claude Code in a rootless container on an internal + network with no egress and a dummy credential. A proxy container is the + only way out; it records every request and swaps in the real credential. +- HostSandbox: Claude Code on the host, behind an in-process recording proxy. +""" from __future__ import annotations @@ -6,6 +13,8 @@ import json import os import shutil import subprocess +import tempfile +import time import uuid from dataclasses import dataclass, field from pathlib import Path @@ -13,12 +22,11 @@ from pathlib import Path from .proxy import RecordingProxy CLAUDE_BIN = os.environ.get("BENCH_CLAUDE_BIN", "claude") -PROJECTS_DIR = Path(os.environ.get("CLAUDE_CONFIG_DIR", Path.home() / ".claude")) / "projects" +ROOT = Path(__file__).resolve().parent.parent # Allowlist, not denylist: the runs print their environment (`env` case) to the # API and the report, so only what Claude Code needs to start is passed on. It # also drops the variables a parent Claude Code session exports to children. -# Auth comes from ~/.claude credentials, or from the variables kept below. _KEEP = {"HOME", "PATH", "USER", "LOGNAME", "SHELL", "LANG", "TERM", "TMPDIR", "TZ", "XDG_RUNTIME_DIR", "XDG_CONFIG_HOME", "XDG_CACHE_HOME", "XDG_DATA_HOME", "CLAUDE_CONFIG_DIR", "ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", @@ -61,60 +69,226 @@ class Run: } -def transcript_text(session_id: str) -> str: - """Session transcript plus anything stored beside it (subagents, tool results).""" - parts = [] - for path in PROJECTS_DIR.glob(f"*/{session_id}*"): - files = [path] if path.is_file() else [p for p in path.rglob("*") if p.is_file()] - for f in files: - parts.append(f.read_text(errors="replace")) - return "\n".join(parts) - - -def run_claude( - prompt: str, - *, - cwd: Path, - settings: dict, - artifacts: Path, - model: str, - env: dict[str, str] | None = None, - timeout: int = 600, -) -> Run: - """Run `claude -p` with only `settings` loaded, behind a recording proxy.""" - artifacts.mkdir(parents=True, exist_ok=True) - settings_path = artifacts / "settings.json" - settings_path.write_text(json.dumps(settings, indent=2)) - session_id = str(uuid.uuid4()) +@dataclass +class RawRun: + returncode: int + stdout: str + stderr: str + wire: str + transcript: str + + +def _read_tree(root: Path) -> str: + return "\n".join(p.read_text(errors="replace") for p in sorted(root.rglob("*")) if p.is_file()) + - cmd = [ - CLAUDE_BIN, "-p", prompt, +def claude_args(prompt: str, session_id: str, model: str, settings_path: str) -> list[str]: + return [ + "-p", prompt, "--session-id", session_id, "--model", model, "--output-format", "stream-json", "--verbose", "--permission-mode", "bypassPermissions", - # Only the test's settings: no user hooks, no project CLAUDE.md settings. + # Only the test's settings: no user hooks, no user or project settings files. "--setting-sources", "project", - "--settings", str(settings_path), + "--settings", settings_path, "--strict-mcp-config", ] - proxy = RecordingProxy().start() + +class HostSandbox: + name = "host" + subject_dir: str | None = None # subject files are used where they are + + def __init__(self) -> None: + config_dir = Path(os.environ.get("CLAUDE_CONFIG_DIR", Path.home() / ".claude")) + self.projects = config_dir / "projects" + + def run(self, prompt, *, session_id, model, project, settings, subject_files, env, artifacts, timeout, + keep_transcripts) -> RawRun: + settings_path = artifacts / "settings.json" + settings_path.write_text(json.dumps(settings, indent=2)) + proxy = RecordingProxy().start() + try: + proc = subprocess.run( + [CLAUDE_BIN, *claude_args(prompt, session_id, model, str(settings_path))], + cwd=project, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL, + env=clean_env({"ANTHROPIC_BASE_URL": proxy.base_url, **env}), + ) + finally: + proxy.stop() + paths = list(self.projects.glob(f"*/{session_id}*")) + transcript = "\n".join(_read_tree(p) if p.is_dir() else p.read_text(errors="replace") for p in paths) + if not keep_transcripts: + for p in paths: + shutil.rmtree(p) if p.is_dir() else p.unlink() + return RawRun(proc.returncode, proc.stdout, proc.stderr, proxy.wire_text(), transcript) + + def close(self) -> None: + pass + + +def _podman(*args: str, check: bool = True, **kw) -> subprocess.CompletedProcess: + return subprocess.run(["podman", *args], check=check, capture_output=True, text=True, **kw) + + +def host_auth() -> tuple[str, dict[str, str]]: + """The real credential for the proxy, and the dummy env for the sandbox. + + Order: BENCH_AUTH ("oauth:" | "apikey:"), ANTHROPIC_API_KEY, + CLAUDE_CODE_OAUTH_TOKEN, then the claude.ai access token from + ~/.claude/.credentials.json (read only, never refreshed here). + """ + if v := os.environ.get("BENCH_AUTH"): + auth = v + elif v := os.environ.get("ANTHROPIC_API_KEY"): + auth = f"apikey:{v}" + elif v := os.environ.get("CLAUDE_CODE_OAUTH_TOKEN"): + auth = f"oauth:{v}" + else: + creds = Path.home() / ".claude" / ".credentials.json" + if not creds.exists(): + raise RuntimeError("no credential: set BENCH_AUTH, ANTHROPIC_API_KEY or CLAUDE_CODE_OAUTH_TOKEN") + oauth = json.loads(creds.read_text())["claudeAiOauth"] + left = oauth["expiresAt"] / 1000 - time.time() + if left < 30 * 60: + raise RuntimeError(f"claude.ai access token expires in {int(left / 60)} min: " + "run `claude -p hi` on the host to refresh it, then retry") + auth = f"oauth:{oauth['accessToken']}" + dummy = ({"CLAUDE_CODE_OAUTH_TOKEN": "bench-dummy-oauth-token"} if auth.startswith("oauth:") + else {"ANTHROPIC_API_KEY": "sk-ant-bench-dummy-key"}) + return auth, dummy + + +def host_feature_cache() -> dict: + """The host's cached feature flags, to seed the sandbox's config. + + Some tools (Monitor among them) are enabled by remotely fetched flags. The + sandbox cannot fetch them (no egress), so without this seed it would run + with fewer tools than the host. Only the flag cache is copied: no account + data, no credentials. + """ + path = Path(os.environ.get("CLAUDE_CONFIG_DIR", Path.home())) / ".claude.json" try: - proc = subprocess.run( - cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL, - env=clean_env({"ANTHROPIC_BASE_URL": proxy.base_url, **(env or {})}), - ) - finally: - proxy.stop() + data = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return {} + return {k: data[k] for k in ("cachedGrowthBookFeatures", "cachedGrowthBookFeaturesAt") if k in data} + + +class PodmanSandbox: + name = "podman" + subject_dir = "/opt/subject" + INTERNAL = "leak-bench-internal" # no egress: Claude Code lives here + EGRESS = "leak-bench-egress" # the proxy's way out + PROXY_IMAGE = "leak-bench-proxy" + UID = "1000" # the `bench` user in the image + + def __init__(self, claude_version: str) -> None: + self.claude_image = f"leak-bench-claude:{claude_version}" + self._build(self.claude_image, "claude.Containerfile", CLAUDE_VERSION=claude_version) + self._build(self.PROXY_IMAGE, "proxy.Containerfile") + if _podman("network", "exists", self.INTERNAL, check=False).returncode: + _podman("network", "create", "--internal", self.INTERNAL) + if _podman("network", "exists", self.EGRESS, check=False).returncode: + _podman("network", "create", self.EGRESS) + + self.feature_cache = host_feature_cache() + auth, self.dummy_env = host_auth() + # The real credential stays in memory-backed storage and only the proxy sees it. + runtime = os.environ.get("XDG_RUNTIME_DIR") or tempfile.gettempdir() + self._secret_dir = Path(tempfile.mkdtemp(prefix="leak-bench-", dir=runtime)) + self.auth_file = self._secret_dir / "auth" + self.auth_file.touch(mode=0o600) + self.auth_file.write_text(auth) + + def _build(self, tag: str, containerfile: str, **build_args: str) -> None: + if _podman("image", "exists", tag, check=False).returncode == 0: + return + args = [f"--build-arg={k}={v}" for k, v in build_args.items()] + _podman("build", "-q", "-f", str(ROOT / "containers" / containerfile), *args, "-t", tag, str(ROOT), + timeout=1200) + + def close(self) -> None: + shutil.rmtree(self._secret_dir, ignore_errors=True) + + def run(self, prompt, *, session_id, model, project, settings, subject_files, env, artifacts, timeout, + keep_transcripts) -> RawRun: + work = Path(tempfile.mkdtemp(prefix="lb-", dir=artifacts.parent)) + config, records, staged = work / "claude-config", work / "records", work / "subject" + for d in (config, records, staged): + d.mkdir() + (config / ".claude.json").write_text(json.dumps(self.feature_cache)) + for name, src in subject_files.items(): + shutil.copy2(src, staged / name) + (staged / "settings.json").write_text(json.dumps(settings, indent=2)) + shutil.copy2(staged / "settings.json", artifacts / "settings.json") + + tag = session_id[:8] + proxy_name, claude_name = f"lb-proxy-{tag}", f"lb-claude-{tag}" + userns = f"--userns=keep-id:uid={self.UID},gid={self.UID}" + _podman("run", "-d", "--rm", "--init", "--name", proxy_name, userns, + "--network", f"{self.INTERNAL},{self.EGRESS}", + "--cap-drop=all", "--security-opt=no-new-privileges", "--read-only", + "-v", f"{self.auth_file}:/run/secrets/bench-auth:ro,z", + "-v", f"{records}:/records:Z", self.PROXY_IMAGE) + try: + self._wait_ready(proxy_name) + cmd = ["podman", "run", "--rm", "-i", "--name", claude_name, userns, + "--network", self.INTERNAL, + "--cap-drop=all", "--security-opt=no-new-privileges", "--read-only", + "--pids-limit=1024", "--memory=4g", + "--tmpfs", "/tmp:rw,U", "--tmpfs", "/home/bench:rw,U", + "-v", f"{config}:/home/bench/.claude:Z", + "-v", f"{project}:/work:Z", + "-v", f"{staged}:{self.subject_dir}:ro,Z", + "-e", f"ANTHROPIC_BASE_URL=http://{proxy_name}:8080"] + for k, v in {**self.dummy_env, **env}.items(): + cmd += ["-e", f"{k}={v}"] + cmd += [self.claude_image, *claude_args(prompt, session_id, model, f"{self.subject_dir}/settings.json")] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, + stdin=subprocess.DEVNULL) + except subprocess.TimeoutExpired: + _podman("kill", claude_name, check=False) + raise + finally: + _podman("kill", proxy_name, check=False) + + wire_file = records / "wire.jsonl" + wire = "\n".join(json.loads(line)["body"] for line in wire_file.read_text().splitlines()) \ + if wire_file.exists() else "" + transcript = _read_tree(config / "projects") if (config / "projects").exists() else "" + if keep_transcripts: + shutil.copytree(config, artifacts / "claude-config", dirs_exist_ok=True) + shutil.rmtree(work, ignore_errors=True) + return RawRun(proc.returncode, proc.stdout, proc.stderr, wire, transcript) + + @staticmethod + def _wait_ready(name: str, timeout: float = 20) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if "listening" in _podman("logs", name, check=False).stdout: + return + time.sleep(0.2) + raise RuntimeError(f"proxy container {name} did not start: {_podman('logs', name, check=False).stderr}") + + +def run_claude(prompt: str, *, sandbox, project: Path, settings: dict, subject_files: dict[str, Path], + artifacts: Path, model: str, env: dict[str, str] | None = None, timeout: int = 600, + keep_transcripts: bool = False) -> Run: + artifacts.mkdir(parents=True, exist_ok=True) + session_id = str(uuid.uuid4()) + raw = sandbox.run(prompt, session_id=session_id, model=model, project=project, settings=settings, + subject_files=subject_files, env=env or {}, artifacts=artifacts, timeout=timeout, + keep_transcripts=keep_transcripts) events = [] - for line in proc.stdout.splitlines(): + for line in raw.stdout.splitlines(): try: events.append(json.loads(line)) except json.JSONDecodeError: pass - tool_calls = [ ToolCall(block["name"], block.get("input", {})) for ev in events if ev.get("type") == "assistant" @@ -123,19 +297,8 @@ def run_claude( ] answer = "\n".join(ev.get("result", "") for ev in events if ev.get("type") == "result") - run = Run(session_id, proc.returncode, events, proc.stderr, proxy.wire_text(), - transcript_text(session_id), answer, tool_calls) - - (artifacts / "stream.jsonl").write_text(proc.stdout) - (artifacts / "stderr.txt").write_text(proc.stderr) - (artifacts / "wire.txt").write_text(run.wire) - (artifacts / "transcript.jsonl").write_text(run.transcript) - (artifacts / "requests.json").write_text(json.dumps( - [{"method": e.method, "path": e.path, "status": e.status, "bytes": len(e.body)} for e in proxy.exchanges], - indent=2)) - return run - - -def cleanup_transcript(session_id: str) -> None: - for path in PROJECTS_DIR.glob(f"*/{session_id}*"): - shutil.rmtree(path) if path.is_dir() else path.unlink() + (artifacts / "stream.jsonl").write_text(raw.stdout) + (artifacts / "stderr.txt").write_text(raw.stderr) + (artifacts / "wire.txt").write_text(raw.wire) + (artifacts / "transcript.jsonl").write_text(raw.transcript) + return Run(session_id, raw.returncode, events, raw.stderr, raw.wire, raw.transcript, answer, tool_calls) diff --git a/bench/subjects.py b/bench/subjects.py index 29de13a..0adb5e3 100644 --- a/bench/subjects.py +++ b/bench/subjects.py @@ -1,8 +1,13 @@ """Subjects: the Claude Code configurations under test. -A subject supplies the settings Claude Code runs with (typically a hook) and, -optionally, a way to declare known secret values ahead of time. To test a new -project, add a Subject here and register it in SUBJECTS. +A subject supplies the files it needs (e.g. a hook binary), the settings Claude +Code runs with (typically a hook) and, optionally, a way to declare known +secret values ahead of time. To test a new project, add a Subject here and +register it in SUBJECTS. + +Files are copied into the sandbox: `settings(where)` receives, for each +`files()` name, the path the file is visible at (a container path under podman). +Hook binaries must run in the sandbox image (Fedora minimal, glibc). """ from __future__ import annotations @@ -24,7 +29,10 @@ class Subject: """None when usable, else the reason to skip.""" return None - def settings(self) -> dict: + def files(self) -> dict[str, Path]: + return {} + + def settings(self, where: dict[str, str]) -> dict: return {} def declare_known(self, project: Path, values: list[str]) -> bool: @@ -44,7 +52,8 @@ class Redacted(Subject): """github.com/svn-arv/redacted (Go, 0.x): a PostToolUse hook, `redacted scrub`. Known values are declared as a literal custom pattern in the project's - .redacted.engine.yml. The global ~/.config/redacted files still apply. + .redacted.engine.yml. In host mode the global ~/.config/redacted files + also apply; the podman sandbox has none. """ def __init__(self, name: str = "redacted", binary: str | None = None) -> None: @@ -56,8 +65,12 @@ class Redacted(Subject): return "redacted binary not found (set REDACTED_BIN)" return None - def settings(self) -> dict: - return {"hooks": {"PostToolUse": [{"hooks": [{"type": "command", "command": f"{self.binary} scrub"}]}]}} + def files(self) -> dict[str, Path]: + return {"redacted": Path(self.binary)} + + def settings(self, where: dict[str, str]) -> dict: + cmd = f"{where['redacted']} scrub" + return {"hooks": {"PostToolUse": [{"hooks": [{"type": "command", "command": cmd}]}]}} def declare_known(self, project: Path, values: list[str]) -> bool: assert not any("\\E" in v or "'" in v for v in values), "value cannot be quoted literally" diff --git a/conftest.py b/conftest.py index daba8d4..6ce7157 100644 --- a/conftest.py +++ b/conftest.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import re +import subprocess import time from pathlib import Path @@ -23,7 +25,12 @@ def pytest_addoption(parser: pytest.Parser) -> None: g.addoption("--attempts", type=int, default=2, help="retries when the model does not exercise the scenario (inconclusive run)") g.addoption("--keep-transcripts", action="store_true", - help="leave the session transcripts in ~/.claude/projects (they are copied to the report anyway)") + help="keep the session transcripts: in ~/.claude/projects (host) or in the report (podman)") + g.addoption("--sandbox", choices=["podman", "host"], default="podman", + help="podman: rootless container, no egress but the recording proxy (default). " + "host: run the host's claude directly") + g.addoption("--claude-version", default=None, + help="Claude Code version for the podman image (default: the host's `claude --version`)") def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: @@ -40,6 +47,28 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: ) +@pytest.fixture(scope="session") +def sandbox(request): + from bench.runner import HostSandbox, PodmanSandbox + + if request.config.getoption("sandbox") == "host": + sb = HostSandbox() + else: + version = request.config.getoption("claude_version") + if not version: + out = subprocess.run(["claude", "--version"], capture_output=True, text=True).stdout + m = re.search(r"\d+\.\d+\.\d+", out) + if not m: + raise pytest.UsageError("cannot detect the host Claude Code version: pass --claude-version") + version = m.group(0) + try: + sb = PodmanSandbox(version) + except RuntimeError as e: + pytest.exit(str(e), returncode=2) + yield sb + sb.close() + + @pytest.fixture(scope="session") def report_dir() -> Path: d = ROOT / "reports" / time.strftime("%Y%m%d-%H%M%S") diff --git a/containers/claude.Containerfile b/containers/claude.Containerfile new file mode 100644 index 0000000..b0d29ab --- /dev/null +++ b/containers/claude.Containerfile @@ -0,0 +1,32 @@ +# Claude Code, pinned, nothing else: bash + coreutils for the scenarios. +# podman build -f containers/claude.Containerfile --build-arg CLAUDE_VERSION=2.1.283 -t leak-bench-claude:2.1.283 . +ARG FEDORA=44 + +FROM registry.fedoraproject.org/fedora-minimal:${FEDORA} AS fetch +ARG CLAUDE_VERSION +ARG PLATFORM=linux-x64 +RUN microdnf -y install --setopt=install_weak_deps=0 jq && microdnf clean all +RUN set -eu; \ + test -n "${CLAUDE_VERSION}" || { echo "CLAUDE_VERSION build arg required" >&2; exit 1; }; \ + base="https://downloads.claude.ai/claude-code-releases/${CLAUDE_VERSION}"; \ + sum=$(curl -fsSL "$base/manifest.json" | jq -er ".platforms[\"${PLATFORM}\"].checksum"); \ + curl -fsSL -o /claude "$base/${PLATFORM}/claude"; \ + echo "$sum /claude" | sha256sum -c -; \ + chmod 0755 /claude + +FROM registry.fedoraproject.org/fedora-minimal:${FEDORA} +ARG CLAUDE_VERSION +COPY --from=fetch /claude /usr/local/bin/claude +# uid/gid 1000: run with --userns=keep-id:uid=1000,gid=1000 so files in the +# bind-mounted project belong to the host user on both sides. +RUN echo 'bench:x:1000:1000::/home/bench:/bin/bash' >> /etc/passwd \ + && echo 'bench:x:1000:' >> /etc/group \ + && mkdir -p /home/bench/.claude /work && chown -R 1000:1000 /home/bench /work +USER bench +WORKDIR /work +ENV HOME=/home/bench \ + CLAUDE_CONFIG_DIR=/home/bench/.claude \ + DISABLE_AUTOUPDATER=1 \ + SHELL=/bin/bash +LABEL io.leak-bench.claude-version="${CLAUDE_VERSION}" +ENTRYPOINT ["/usr/local/bin/claude"] diff --git a/containers/proxy.Containerfile b/containers/proxy.Containerfile new file mode 100644 index 0000000..cbd7011 --- /dev/null +++ b/containers/proxy.Containerfile @@ -0,0 +1,9 @@ +# The recording proxy: the only way out for the Claude Code container. +# podman build -f containers/proxy.Containerfile -t leak-bench-proxy . +ARG FEDORA=44 +FROM registry.fedoraproject.org/fedora-minimal:${FEDORA} +RUN microdnf -y install --setopt=install_weak_deps=0 python3 python3-aiohttp python3-zstandard \ + && microdnf clean all +COPY bench/proxy.py /app/proxy.py +USER 1000:1000 +ENTRYPOINT ["python3", "/app/proxy.py"] diff --git a/tests/test_leaks.py b/tests/test_leaks.py index 5b2a30a..05da1bc 100644 --- a/tests/test_leaks.py +++ b/tests/test_leaks.py @@ -5,7 +5,7 @@ from __future__ import annotations import pytest from bench.cases import CASES -from bench.runner import cleanup_transcript, run_claude +from bench.runner import run_claude def _cell(leaks: dict[str, bool]) -> str: @@ -13,7 +13,7 @@ def _cell(leaks: dict[str, bool]) -> str: return f"LEAK({where})" if where else "clean" -def test_leak(subject, case_name, tmp_path, request, report_dir, record): +def test_leak(subject, case_name, tmp_path, request, report_dir, record, sandbox): opt = request.config.getoption row = dict(subject=subject.name, case=case_name, report_dir=str(report_dir)) @@ -21,6 +21,12 @@ def test_leak(subject, case_name, tmp_path, request, report_dir, record): record(**row, cell="skip") pytest.skip(reason) + files = subject.files() + if sandbox.subject_dir: + where = {name: f"{sandbox.subject_dir}/{name}" for name in files} + else: + where = {name: str(path) for name, path in files.items()} + run, scenario, why = None, None, "" for attempt in range(1, opt("attempts") + 1): project = tmp_path / f"attempt{attempt}" / "project" @@ -34,14 +40,14 @@ def test_leak(subject, case_name, tmp_path, request, report_dir, record): scenario.writer.start() try: run = run_claude( - scenario.prompt, cwd=project, settings=subject.settings(), model=opt("model"), env=scenario.env, - artifacts=report_dir / subject.name / case_name / f"attempt{attempt}", + scenario.prompt, sandbox=sandbox, project=project, settings=subject.settings(where), + subject_files=files, model=opt("model"), env=scenario.env, + artifacts=report_dir / sandbox.name / subject.name / case_name / f"attempt{attempt}", + keep_transcripts=opt("keep_transcripts"), ) finally: if scenario.writer: scenario.writer.stop() - if not opt("keep_transcripts"): - cleanup_transcript(run.session_id) ok, why = scenario.check(run) if ok: