"""Opt-in installed Claude binary against a namespace-local synthetic API.""" from __future__ import annotations import hashlib import json import os import shutil import subprocess from pathlib import Path import pytest from llm_connect.models import RunConfig from rein_aharness.adapter import AgenticClaudeCodeAdapter from rein_aharness.native_limits import NativeLimitError, terminal_accounting pytestmark = pytest.mark.skipif( os.environ.get("REIN_REAL_CLAUDE") != "1", reason="opt-in installed CLI boundary proof", ) @pytest.mark.parametrize( "case", ["overrun", "tools", "guarded-overrun", "guarded-tools"] ) def test_installed_cli_boundary(tmp_path, case): binary = Path( os.environ.get("REIN_CLAUDE_PROOF_BINARY") or shutil.which("claude") or "" ).resolve(strict=True) version = subprocess.check_output([str(binary), "--version"], text=True).strip() assert version.startswith("2.1.266 "), ( "revalidate the fixture for this binary version before claiming the pinned proof" ) adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, model="claude-sonnet-4-6") cap = 0.01 if case in ("overrun", "guarded-overrun") else 1 argv = adapter._build_command( RunConfig(model_params={"max_budget_usd": cap, "max_turns": 4}) ) argv[0] = "/opt/claude" (tmp_path / "case.json").write_text(json.dumps({"case": case, "argv": argv})) (tmp_path / "passwd").write_text( f"fixture:x:{os.getuid()}:{os.getgid()}:Fixture:/work/home:/bin/sh\n" ) (tmp_path / "group").write_text(f"fixture:x:{os.getgid()}:\n") (tmp_path / "home").mkdir() shutil.copyfile( Path(__file__).with_name("native_cli_fixture.py"), tmp_path / "fixture.py" ) if case.startswith("guarded-"): import llm_connect.messages_gate import rein_aharness.request_admission import rein_aharness.spend_admission for package, modules in { "llm_connect": [llm_connect.messages_gate], "rein_aharness": [ rein_aharness.request_admission, rein_aharness.spend_admission, ], }.items(): target = tmp_path / package target.mkdir() (target / "__init__.py").write_text("") for module in modules: shutil.copyfile(module.__file__, target / Path(module.__file__).name) command = [ "bwrap", "--unshare-all", "--die-with-parent", "--new-session", "--ro-bind", "/usr", "/usr", "--ro-bind", "/lib", "/lib", "--ro-bind", "/lib64", "/lib64", "--symlink", "usr/bin", "/bin", "--proc", "/proc", "--dev", "/dev", "--tmpfs", "/tmp", "--dir", "/etc", "--ro-bind", str(tmp_path / "passwd"), "/etc/passwd", "--ro-bind", str(tmp_path / "group"), "/etc/group", "--dir", "/opt", "--ro-bind", str(binary), "/opt/claude", "--bind", str(tmp_path), "/work", "--chdir", "/work", "--clearenv", "--setenv", "PATH", "/usr/bin:/bin", "/usr/bin/python3", "/work/fixture.py", ] proc = subprocess.run( command, capture_output=True, text=True, timeout=55, check=False ) assert proc.returncode == 0, proc.stderr[-1000:] result = json.loads(proc.stdout) assert result["network_namespace"] != os.readlink("/proc/self/ns/net") assert result["markers"] == { "hook-ran": False, "mcp-ran": False, "denied-ran": False, } requests = [row for row in result["requests"] if "count_tokens" not in row["path"]] expected_tools = {"Read", "Write", "Edit", "Glob", "Grep", "Bash"} for request in requests: # The CLI may add its terminal EndConversation tool; it cannot add work tools. assert set(request["tools"]) - {"EndConversation"} <= expected_tools assert {"Bash", "Read", "Edit"} <= set(request["tools"]) assert request["model"] == "claude-sonnet-4-6" assert not request["ambient_context_loaded"] terminal = result["terminal"] if case.startswith("guarded-"): assert all("refusal" not in row for row in result["guard_attempts"]), ( json.dumps(result["guard_attempts"]) ) if case == "guarded-overrun": assert result["guard_attempts"], result assert all( row.get("liability_microusd") == 1_080_000 for row in result["guard_attempts"] ), result assert not requests and not result["request_reservations"] assert result["returncode"] != 0 and terminal["is_error"] assert terminal["total_cost_usd"] == 0 elif case == "overrun": assert len(requests) == 1 assert proc.returncode == 0 and result["returncode"] == 1 assert terminal["subtype"] == "error_max_budget_usd" assert terminal["total_cost_usd"] == pytest.approx(0.18015) assert terminal["total_cost_usd"] > cap with pytest.raises(NativeLimitError) as caught: terminal_accounting(terminal, max_budget_usd=cap, max_turns=4) assert caught.value.cost_usd == terminal["total_cost_usd"] else: assert result["returncode"] == 0 and terminal["subtype"] == "success" assert len(requests) == 2 assert {"id": "tool_allowed", "is_error": False} in requests[1]["tool_results"] assert {"id": "tool_denied", "is_error": True} in requests[1]["tool_results"] assert {"id": "tool_create", "is_error": False} in requests[1]["tool_results"] assert result["created_file"] == "fixture-created\n" usage, cost = terminal_accounting(terminal, max_budget_usd=cap, max_turns=4) assert usage["total_tokens"] > 0 and cost > 0 if case == "guarded-tools": assert len(result["request_reservations"]) == 2 assert all( row["state"] == "charged" for row in result["request_reservations"] ) assert ( sum(row["liability_microusd"] for row in result["request_reservations"]) == 2_160_000 ) if output_dir := os.environ.get("REIN_CLAUDE_PROOF_OUTPUT"): target = Path(output_dir) target.mkdir(parents=True, exist_ok=True) (target / f"{case}.json").write_text( json.dumps( { "case": case, "binary_version": version, "binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(), "argv": argv, "scope": "synthetic API in isolated network namespace; no real credential or provider request", **result, }, indent=2, ) + "\n" )