#!/usr/bin/env python3 """Native MCP diagnostic, public adaptation, 2026-09-12. Linux, Python 3.10+ standard library, and standalone Claude Code / OpenCode binaries. The predecessor ran once with Claude Code 2.1.270 (claude-opus-5) and once with OpenCode 1.18.30 (opencode/muse-spark-1.3-contributor-free). This portable copy was tested without models; other client versions and future model availability are unverified. Start with five local tests; no AI client is started by selftest: python3 -B native_probe.py selftest Prepare one NEW directory; both binaries must be installed and resolvable: python3 -B native_probe.py prepare /tmp/native-probe-NEW Optional: --claude /absolute/path/to/claude --opencode /absolute/path/to/opencode Only run invokes a model. Claude uses the existing native login; OpenCode uses a fresh home and the specified contributor-free provider/model. Each run can make several model requests, with up to eight turns/steps and a 240-second process timeout. It records an attempt before starting and never retries the process. Inspect a failed attempt instead of resending. python3 -B native_probe.py run /tmp/native-probe-NEW opus python3 -B native_probe.py verify /tmp/native-probe-NEW opus python3 -B native_probe.py run /tmp/native-probe-NEW muse python3 -B native_probe.py verify /tmp/native-probe-NEW muse Success means verification.json contains verified: true. The verifier checks the native input, client-reported model, matching tool calls, HTTP trace, output file, and process completion. Native evidence formats are version dependent; an unfamiliar or incomplete trace fails verification. The three MCP tools accept only brief.txt, one fixed loopback /status URL, and report.json (8192-byte limit). This restricts tool capabilities; it is not OS process isolation or a test of native Bash. Client-reported model identity is not provider-side attestation. Keep raw run folders local: native logs can contain machine paths or account metadata. Changes after the two native probes: PATH-based binary discovery, the current Python interpreter for child processes, embedded tests, this documentation, and a locally tested nonzero exit on process failure. """ import argparse import datetime import fcntl import hashlib import http.client import http.server import json import os from pathlib import Path import signal import shutil import tempfile import unittest import sqlite3 import stat import subprocess import sys import threading from urllib.parse import urlsplit MODELS = {"opus": "claude-opus-5", "muse": "opencode/muse-spark-1.3-contributor-free"} PROMPT = ('Technical diagnostic, one task. Use only the lab MCP tools.\n' 'Read brief.txt, call get_status, then use write_report to write a JSON object ' 'with exactly these keys: file_token, service_token, echoed_text. ' 'Copy the tokens from the two tool results. Set echoed_text to this exact text: ' 'Кавычки "и пробелы"; literal $HOME and `printf`.\n' 'Finish with DONE. Do not call any other tools.\n') BRIEF = 'Synthetic file. file_token: FILE-8f27c1\n' STATUS = {"service_token": "SERVICE-4e9bd2", "status": "ok"} EXPECTED = {"file_token": "FILE-8f27c1", "service_token": "SERVICE-4e9bd2", "echoed_text": 'Кавычки "и пробелы"; literal $HOME and `printf`.'} TOOLS = [ {"name": "read_file", "description": "Read the supplied brief.txt file.", "inputSchema": {"type": "object", "properties": {"name": {"type": "string", "enum": ["brief.txt"]}}, "required": ["name"], "additionalProperties": False}}, {"name": "get_status", "description": "GET /status from the fixed local synthetic HTTP service.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "write_report", "description": "Write content to the sole output file, report.json.", "inputSchema": {"type": "object", "properties": {"content": {"type": "string", "maxLength": 8192}}, "required": ["content"], "additionalProperties": False}}, ] def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def sha(data): return hashlib.sha256(data).hexdigest() def dump(path, value): Path(path).write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n") def append(path, value): with Path(path).open("a") as stream: stream.write(json.dumps(value, ensure_ascii=False) + "\n") def rows(path): return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()] def file_io(workspace, name, content=None): flags = os.O_RDONLY if content is None else os.O_WRONLY fd = os.open(workspace / name, flags | os.O_NOFOLLOW) try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: raise ValueError("Only regular files without links are allowed") if content is None: result = os.read(fd, 8193) if len(result) > 8192: raise ValueError("Input exceeds 8192 bytes") return result.decode("utf-8") data = content.encode("utf-8") if len(data) > 8192: raise ValueError("Output exceeds 8192 bytes") os.ftruncate(fd, 0) with os.fdopen(fd, "wb", closefd=False) as stream: stream.write(data) stream.flush() os.fsync(fd) return {"written": "report.json", "bytes": len(data), "sha256": sha(data)} finally: os.close(fd) def call_tool(workspace, service_url, name, arguments): if not isinstance(arguments, dict): raise ValueError("Arguments must be an object") if name == "read_file" and arguments == {"name": "brief.txt"}: return file_io(workspace, "brief.txt") if name == "write_report" and set(arguments) == {"content"} and isinstance(arguments["content"], str): return file_io(workspace, "report.json", arguments["content"]) if name == "get_status" and arguments == {}: endpoint = urlsplit(service_url) if (endpoint.scheme != "http" or endpoint.hostname != "127.0.0.1" or endpoint.username or endpoint.password or endpoint.path != "/status" or endpoint.query or endpoint.fragment or endpoint.port is None): raise ValueError("Only the fixed local /status endpoint is allowed") connection = http.client.HTTPConnection("127.0.0.1", endpoint.port, timeout=3) try: connection.request("GET", "/status") response = connection.getresponse() body = response.read(8193) if response.status != 200 or len(body) > 8192: raise ValueError("Unexpected service response; no redirect or retry") return json.loads(body) finally: connection.close() raise ValueError("Tool or arguments are outside this diagnostic's allowlist") def mcp(workspace, trace, service_url): for line in sys.stdin: request = json.loads(line) append(trace, {"at": now(), "direction": "request", "body": request}) if "id" not in request: continue method = request.get("method") response = {"jsonrpc": "2.0", "id": request["id"]} if method == "initialize": response["result"] = {"protocolVersion": request["params"]["protocolVersion"], "capabilities": {"tools": {}}, "serverInfo": {"name": "native-probe", "version": "1"}} elif method == "ping": response["result"] = {} elif method == "tools/list": response["result"] = {"tools": TOOLS} elif method == "tools/call": try: params = request["params"] result = call_tool(workspace, service_url, params["name"], params.get("arguments", {})) text = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False) response["result"] = {"content": [{"type": "text", "text": text}]} except (ValueError, OSError, http.client.HTTPException) as exc: response["result"] = {"isError": True, "content": [{"type": "text", "text": str(exc)}]} else: response["error"] = {"code": -32601, "message": "Method not found"} append(trace, {"at": now(), "direction": "response", "body": response}) print(json.dumps(response, ensure_ascii=False), flush=True) class SyntheticService: def __init__(self, trace): class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): code = 200 if self.path == "/status" else 404 body = json.dumps(STATUS if code == 200 else {"error": "not_found"}).encode() append(trace, {"at": now(), "method": "GET", "path": self.path, "status": code, "body": json.loads(body), "sha256": sha(body)}) self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *_): pass self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_port}/status" def __enter__(self): self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() return self.url def __exit__(self, *_): self.server.shutdown() self.server.server_close() self.thread.join() def prepare(root, claude, opencode): root.mkdir(parents=True, exist_ok=False) binaries = {} for model, requested in (("opus", claude), ("muse", opencode)): found = shutil.which(requested) if found is None: raise ValueError("Executable not found: " + requested) binaries[model] = str(Path(found).resolve(strict=True)) for model in MODELS: folder = root / model (folder / "workspace").mkdir(parents=True) (folder / "workspace/brief.txt").write_text(BRIEF) (folder / "workspace/report.json").touch() (folder / "client-home").mkdir(mode=0o700) (root / "prompt.txt").write_bytes(PROMPT.encode()) dump(root / "plan.json", {"created_at": now(), "kind": "technical diagnostic, not a research trial", "planned_native_runs": list(MODELS), "models": MODELS, "binaries": binaries, "binary_sha256": {key: sha(Path(value).read_bytes()) for key, value in binaries.items()}, "script_sha256": sha(Path(__file__).read_bytes()), "prompt_sha256": sha(PROMPT.encode()), "expected_report": EXPECTED, "automatic_runner_retries": 0, "timeout_seconds": 240}) return {"state": "prepared", "root": str(root), "model_runs": 0} def client(root, model, service_url): folder = root / model workspace = folder / "workspace" plan = json.loads((root / "plan.json").read_text()) binary = plan["binaries"][model] if sha(Path(binary).read_bytes()) != plan["binary_sha256"][model]: raise ValueError("Pinned client changed") script = str(Path(__file__).resolve()) mcp_args = [sys.executable, "-B", script, "mcp", str(workspace), "--trace", str(folder / "mcp.jsonl"), "--url", service_url] env = {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8", "SHELL": "/bin/sh"} if model == "opus": env.update({key: os.environ[key] for key in ("HOME", "USER", "LOGNAME") if key in os.environ}) env.update(CLAUDE_CODE_DISABLE_CLAUDE_MDS="1", CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC="1", CLAUDE_CODE_SKIP_PROMPT_HISTORY="1", ENABLE_TOOL_SEARCH="false", DISABLE_AUTOUPDATER="1") settings = {"fastMode": False, "autoMemoryEnabled": False, "disableAllHooks": True, "claudeMdExcludes": ["/**"], "permissions": {"allow": ["mcp__lab__*"]}} config = {"mcpServers": {"lab": {"type": "stdio", "command": mcp_args[0], "args": mcp_args[1:]}}} dump(folder / "settings.json", settings) dump(folder / "mcp-config.json", config) args = [binary, "--restricted", "--tools", "", "--settings", str(folder / "settings.json"), "--setting-sources", "", "--strict-mcp-config", "--mcp-config", str(folder / "mcp-config.json"), "--disable-slash-commands", "--no-chrome", "--permission-mode", "dontAsk", "--model", MODELS[model], "--effort", "medium", "--no-session-persistence", "--max-turns", "8", "--max-budget-usd", "2", "--prompt-suggestions", "false", "--input-format", "stream-json", "--replay-user-messages", "--output-format", "stream-json", "--verbose", "-p"] stdin = (json.dumps({"type": "user", "message": {"role": "user", "content": PROMPT}}, ensure_ascii=False) + "\n").encode() else: # This client needs no owner credentials. Its entire home is fresh. permission = {"*": "deny", "lab_*": "allow"} config = {"$schema": "https://opencode.ai/config.json", "model": MODELS[model], "small_model": MODELS[model], "enabled_providers": ["opencode"], "share": "disabled", "autoupdate": False, "provider": {"opencode": {"whitelist": [MODELS[model].split("/", 1)[1]]}}, "permission": permission, "agent": {"probe": {"mode": "primary", "steps": 8, "permission": permission}}, "mcp": {"lab": {"type": "local", "command": mcp_args, "enabled": True}}} dump(folder / "opencode.json", config) client_home = str(folder / "client-home") env.update(HOME=client_home, USER="probe", LOGNAME="probe", XDG_CONFIG_HOME=client_home + "/config", XDG_DATA_HOME=client_home + "/data", XDG_STATE_HOME=client_home + "/state", XDG_CACHE_HOME=client_home + "/cache", OPENCODE_CONFIG=str(folder / "opencode.json"), OPENCODE_CONFIG_CONTENT=json.dumps(config), OPENCODE_DISABLE_DEFAULT_PLUGINS="true", OPENCODE_DISABLE_AUTOUPDATE="true") args = [binary, "run", "--pure", "--agent", "probe", "--model", MODELS[model], "--title", "Native technical diagnostic", "--format", "json"] stdin = PROMPT.encode() # No argv message: this native CLI quotes argv entries containing spaces. return args, env, stdin def run(root, model): folder = root / model plan = json.loads((root / "plan.json").read_text()) if (sha(Path(__file__).read_bytes()) != plan["script_sha256"] or (root / "prompt.txt").read_bytes() != PROMPT.encode() or (folder / "workspace/brief.txt").read_text() != BRIEF or (folder / "workspace/report.json").read_bytes()): raise ValueError("Prepared inputs or runner changed") with (folder / "writer.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) if (folder / "attempt.json").exists(): raise ValueError("An attempt already exists; inspect it without retrying") with SyntheticService(folder / "http.jsonl") as url: args, env, stdin = client(root, model, url) dump(folder / "command.json", args) (folder / "stdin.bin").write_bytes(stdin) receipt = {"state": "attempt_recorded", "started_at": now(), "model": MODELS[model], "prompt_sha256": sha(PROMPT.encode()), "stdin_sha256": sha(stdin)} with (folder / "attempt.json").open("x") as stream: json.dump(receipt, stream) process = None try: with (folder / "events.jsonl").open("xb") as stdout, (folder / "stderr.txt").open("xb") as stderr: process = subprocess.Popen(args, env=env, cwd=folder / "workspace", stdin=subprocess.PIPE, stdout=stdout, stderr=stderr, start_new_session=True) try: process.communicate(stdin, timeout=plan["timeout_seconds"]) except subprocess.TimeoutExpired: receipt["timed_out"] = True os.killpg(process.pid, signal.SIGTERM) try: process.wait(timeout=5) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) process.wait() receipt["returncode"] = process.returncode receipt["state"] = "finished" if process.returncode == 0 and not receipt.get("timed_out") else "incomplete" except (Exception, KeyboardInterrupt) as exc: if process is not None and process.poll() is None: os.killpg(process.pid, signal.SIGTERM) try: process.wait(timeout=5) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) process.wait() receipt.update(state="not_started" if process is None else "incomplete", error=str(exc)) receipt["finished_at"] = now() dump(folder / "receipt.json", receipt) return receipt def text_content(content): if isinstance(content, str): return content return "".join(block.get("text", "") for block in content if block.get("type") == "text") def _verify(root, model): """Read native evidence, independently of the process exit status and model's claim.""" folder = root / model events = rows(folder / "events.jsonl") mcp_rows = rows(folder / "mcp.jsonl") checks = {} native_calls = [] if model == "opus": delivered = [text_content(e["message"].get("content", [])) for e in events if e.get("type") == "user"] delivered = [text for text in delivered if text] messages = [e["message"] for e in events if e.get("type") == "assistant"] models = sorted({message.get("model") for message in messages}) for message in messages: native_calls.extend({"id": b["id"], "name": b["name"], "arguments": b["input"]} for b in message.get("content", []) if b.get("type") == "tool_use") native_results = [b for e in events if e.get("type") == "user" for b in e["message"].get("content", []) if isinstance(b, dict) and b.get("type") == "tool_result"] checks["all_native_tool_results"] = (len(native_results) == len(native_calls) and {b["tool_use_id"] for b in native_results} == {c["id"] for c in native_calls} and not any(b.get("is_error") for b in native_results)) terminal = [e for e in events if e.get("type") == "result"] checks["terminal_success"] = (len(terminal) == 1 and events[-1] == terminal[0] and terminal[0].get("subtype") == "success" and not terminal[0].get("is_error")) init = [e for e in events if e.get("type") == "system" and e.get("subtype") == "init"] checks["only_scoped_tools_exposed"] = (len(init) == 1 and set(init[0].get("tools", [])) == {"mcp__lab__" + tool["name"] for tool in TOOLS}) session_id = terminal[0].get("session_id") if terminal else None final_text = terminal[0].get("result") if terminal else None usage = terminal[0].get("modelUsage") if terminal else None prefix = "mcp__lab__" else: sessions = {e["sessionID"] for e in events if "sessionID" in e} if len(sessions) != 1: raise ValueError("No unique native session ID") session_id = sessions.pop() db_path = folder / "client-home/data/opencode/opencode.db" with sqlite3.connect(db_path.resolve().as_uri() + "?mode=ro", uri=True) as db: messages = [{"id": r[0], **json.loads(r[1])} for r in db.execute( "SELECT id,data FROM message WHERE session_id=? ORDER BY time_created,id", (session_id,))] user_ids = {m["id"] for m in messages if m["role"] == "user"} delivered = [json.loads(r[1])["text"] for r in db.execute( "SELECT message_id,data FROM part WHERE session_id=? ORDER BY time_created,id", (session_id,)) if r[0] in user_ids and json.loads(r[1]).get("type") == "text"] assistants = [m for m in messages if m["role"] == "assistant"] models = sorted({m["providerID"] + "/" + m["modelID"] for m in assistants}) tool_parts = [e["part"] for e in events if e.get("type") == "tool_use"] native_calls = [{"id": p["callID"], "name": p["tool"], "arguments": p["state"]["input"]} for p in tool_parts] checks["all_native_tool_results"] = all(p["state"]["status"] == "completed" for p in tool_parts) checks["terminal_success"] = (events[-1].get("type") == "step_finish" and events[-1].get("part", {}).get("reason") == "stop" and not any(e.get("type") == "error" for e in events)) final_id = events[-1].get("part", {}).get("messageID") final_text = "".join(e["part"]["text"] for e in events if e.get("type") == "text" and e["part"].get("messageID") == final_id) # The native config readback is recorded separately before launch; the stream does not enumerate schemas. usage = [{k: m.get(k) for k in ("id", "tokens", "cost", "finish")} for m in assistants] prefix = "lab_" checks["native_model_identity"] = models == [MODELS[model]] checks["exact_native_prompt"] = delivered == [PROMPT] dump(folder / "native-input-receipt.json", {"session_id": session_id, "texts": delivered, "sha256": [sha(text.encode()) for text in delivered], "models": models, "source": "native user replay" if model == "opus" else "native session SQLite, read only"}) requests = [r["body"] for r in mcp_rows if r["direction"] == "request" and r["body"].get("method") == "tools/call"] replies = [r["body"] for r in mcp_rows if r["direction"] == "response"] replies_by_id = {r["id"]: r for r in replies} checks["complete_mcp_trace"] = (len(replies_by_id) == len(replies) and all( r["id"] in replies_by_id and "result" in replies_by_id[r["id"]] and not replies_by_id[r["id"]]["result"].get("isError") for r in requests)) checks["native_calls_match_mcp"] = sorted( json.dumps({"name": c["name"], "arguments": c["arguments"]}, sort_keys=True) for c in native_calls ) == sorted(json.dumps({"name": prefix + r["params"]["name"], "arguments": r["params"].get("arguments", {})}, sort_keys=True) for r in requests) checks["exact_three_operations"] = sorted(r["params"]["name"] for r in requests) == sorted(t["name"] for t in TOOLS) http = rows(folder / "http.jsonl") checks["one_successful_http_read"] = (len(http) == 1 and http[0]["method"] == "GET" and http[0]["path"] == "/status" and http[0]["status"] == 200 and http[0]["body"] == STATUS) checks["report_matches_both_sources"] = json.loads((folder / "workspace/report.json").read_text()) == EXPECTED checks["brief_unchanged"] = (folder / "workspace/brief.txt").read_bytes() == BRIEF.encode() checks["only_expected_workspace_files"] = sorted(p.name for p in (folder / "workspace").iterdir()) == ["brief.txt", "report.json"] checks["clean_process_exit"] = json.loads((folder / "receipt.json").read_text())["state"] == "finished" result = {"checked_at": now(), "verified": all(checks.values()), "checks": checks, "session_id": session_id, "native_assistant_records": len(messages) if model == "opus" else len(assistants), "native_tool_calls": native_calls, "final_text": final_text, "usage": usage, "scope": "MCP capabilities and native event records; not an OS sandbox or provider-side attestation"} dump(folder / "verification.json", result) return result def verify(root, model): try: return _verify(root, model) except (OSError, ValueError, KeyError, TypeError, IndexError, sqlite3.Error) as exc: result = {"checked_at": now(), "verified": False, "error": type(exc).__name__ + ": " + str(exc), "next_action": "Inspect saved evidence; do not repeat the native invocation"} dump(root / model / "verification.json", result) return result def main(): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="action", required=True) p = sub.add_parser("prepare", help="Create a fresh, frozen synthetic packet; no model calls") p.add_argument("root", type=Path) p.add_argument("--claude", default="claude") p.add_argument("--opencode", default="opencode") p = sub.add_parser("mcp", help="Serve the three scoped tools; does not invoke a model") p.add_argument("workspace", type=Path) p.add_argument("--trace", required=True, type=Path) p.add_argument("--url", required=True) p = sub.add_parser("run", help="Invoke exactly one native model session; requires prior authorization") p.add_argument("root", type=Path) p.add_argument("model", choices=MODELS) p = sub.add_parser("verify", help="Check saved prompt, native calls, HTTP, and output; no model calls") p.add_argument("root", type=Path) p.add_argument("model", choices=MODELS) sub.add_parser("selftest", help="Run five local tests; no model clients") args = parser.parse_args() if args.action == "selftest": suite = unittest.defaultTestLoader.loadTestsFromTestCase(ToolsTest) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 2 elif args.action == "mcp": mcp(args.workspace.resolve(), args.trace.resolve(), args.url) elif args.action == "prepare": print(json.dumps(prepare(args.root.resolve(), args.claude, args.opencode))) elif args.action == "verify": result = verify(args.root.resolve(), args.model) print(json.dumps(result, ensure_ascii=False)) return 0 if result["verified"] else 2 else: result = run(args.root.resolve(), args.model) print(json.dumps(result)) return 0 if result["state"] == "finished" else 2 SCRIPT = Path(__file__).resolve() probe = sys.modules[__name__] class ToolsTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="native-probe-test-") self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.work = self.root / "workspace" self.work.mkdir() (self.work / "brief.txt").write_text(probe.BRIEF) (self.work / "report.json").touch() self.canary = self.root / "private-canary.txt" self.canary.write_text("synthetic outside canary") def invoke(self, url, name, arguments): return probe.call_tool(self.work, url, name, arguments) def test_full_stdio_protocol_and_effects(self): with probe.SyntheticService(self.root / "http.jsonl") as url: requests = [ {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "no-model-test", "version": "1"}}}, {"jsonrpc": "2.0", "method": "notifications/initialized"}, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, *[{"jsonrpc": "2.0", "id": i, "method": "tools/call", "params": {"name": name, "arguments": args}} for i, (name, args) in enumerate([ ("read_file", {"name": "brief.txt"}), ("get_status", {}), ("write_report", {"content": json.dumps(probe.EXPECTED, ensure_ascii=False)}), ("read_file", {"name": "../private-canary.txt"}), ], 3)], ] run = subprocess.run([sys.executable, "-B", str(SCRIPT), "mcp", str(self.work), "--trace", str(self.root / "mcp.jsonl"), "--url", url], input="".join(json.dumps(x) + "\n" for x in requests), text=True, capture_output=True, timeout=10) self.assertEqual(run.returncode, 0, run.stderr) responses = [json.loads(line) for line in run.stdout.splitlines()] self.assertEqual([r["id"] for r in responses], list(range(1, 7))) self.assertEqual(responses[1]["result"]["tools"], probe.TOOLS) self.assertEqual(responses[2]["result"]["content"][0]["text"], probe.BRIEF) self.assertEqual(json.loads(responses[3]["result"]["content"][0]["text"]), probe.STATUS) self.assertTrue(responses[-1]["result"]["isError"]) self.assertEqual(json.loads((self.work / "report.json").read_text()), probe.EXPECTED) self.assertEqual(self.canary.read_text(), "synthetic outside canary") trace = probe.rows(self.root / "mcp.jsonl") self.assertEqual([row["body"] for row in trace if row["direction"] == "request"], requests) self.assertEqual([row["body"] for row in trace if row["direction"] == "response"], responses) http = probe.rows(self.root / "http.jsonl") self.assertEqual([(r["method"], r["path"], r["status"]) for r in http], [("GET", "/status", 200)]) def test_disallowed_arguments_cannot_read_write_or_send_http(self): with probe.SyntheticService(self.root / "http.jsonl") as url: cases = [("read_file", {"name": path}) for path in (str(self.canary), "../private-canary.txt", "./brief.txt", "report.json")] cases += [("write_report", {"content": "oops", "path": str(self.canary)}), ("write_report", {"content": 3}), ("write_report", {"content": "я" * 8192}), ("get_status", {"url": "https://example.invalid/"}), ("get_status", {"method": "POST"}), ("bash", {"command": "true"}), ("read_file", [])] for name, arguments in cases: with self.subTest(name=name, arguments=str(arguments)[:120]), self.assertRaises(ValueError): self.invoke(url, name, arguments) self.assertFalse((self.root / "http.jsonl").exists()) self.assertEqual((self.work / "report.json").read_bytes(), b"") self.assertEqual((self.work / "brief.txt").read_text(), probe.BRIEF) def test_links_cannot_redirect_allowed_files(self): for filename, tool, arguments in [("brief.txt", "read_file", {"name": "brief.txt"}), ("report.json", "write_report", {"content": "changed"})]: for kind in ("symlink", "hardlink"): with self.subTest(filename=filename, kind=kind): target = self.work / filename target.unlink() if kind == "symlink": target.symlink_to(self.canary) else: os.link(self.canary, target) with self.assertRaises((OSError, ValueError)): self.invoke("http://127.0.0.1:1/status", tool, arguments) self.assertEqual(self.canary.read_text(), "synthetic outside canary") target.unlink() target.touch() def test_service_configuration_rejects_redirectable_or_external_addresses(self): for url in ["https://example.invalid/status", "http://localhost:80/status", "http://127.0.0.1:80/other", "http://user@127.0.0.1:80/status", "http://127.0.0.1:80/status?target=other"]: with self.subTest(url=url), self.assertRaises(ValueError): self.invoke(url, "get_status", {}) def test_failed_process_is_nonzero_and_cannot_be_repeated(self): # /usr/bin/false is the process under test, never an AI client. root = self.root / "failed-run" prepare = subprocess.run([sys.executable, "-B", str(SCRIPT), "prepare", str(root), "--claude", "/usr/bin/false", "--opencode", "/usr/bin/false"], capture_output=True, text=True) self.assertEqual(prepare.returncode, 0, prepare.stderr) command = [sys.executable, "-B", str(SCRIPT), "run", str(root), "opus"] failed = subprocess.run(command, capture_output=True, text=True, timeout=10) self.assertEqual(failed.returncode, 2) self.assertEqual(json.loads(failed.stdout)["state"], "incomplete") attempt = (root / "opus/attempt.json").read_bytes() receipt = (root / "opus/receipt.json").read_bytes() again = subprocess.run(command, capture_output=True, text=True, timeout=10) self.assertNotEqual(again.returncode, 0) self.assertIn("An attempt already exists", again.stderr) self.assertEqual((root / "opus/attempt.json").read_bytes(), attempt) self.assertEqual((root / "opus/receipt.json").read_bytes(), receipt) if __name__ == "__main__": sys.exit(main())