rein-aharness/tests/test_native_cli_boundary.py

"""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"
        )