@ -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:<token> " | " apikey:<key> " ) , 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 )
