From 65e2bd71cf360b6f45c86eec0b06ae24494832d8 Mon Sep 17 00:00:00 2001 From: wjarvis mos-comms Date: Sun, 19 Jul 2026 20:35:41 -0500 Subject: [PATCH] test(mosaic): define adversarial Claude recovery gate contracts --- packages/mosaic/package.json | 2 +- .../recovery_b1_adversarial_unittest.py | 205 ++++++++++++++++++ 2 files changed, 206 insertions(+), 1 deletion(-) create mode 100644 packages/mosaic/src/lease-broker/recovery_b1_adversarial_unittest.py diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index 9805efb2..12a6b879 100644 --- a/packages/mosaic/package.json +++ b/packages/mosaic/package.json @@ -25,7 +25,7 @@ "lint": "eslint src", "typecheck": "tsc --noEmit", "test": "vitest run --passWithNoTests && pnpm run test:framework-shell", - "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" + "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" }, "dependencies": { "@mosaicstack/brain": "workspace:*", diff --git a/packages/mosaic/src/lease-broker/recovery_b1_adversarial_unittest.py b/packages/mosaic/src/lease-broker/recovery_b1_adversarial_unittest.py new file mode 100644 index 00000000..4577a136 --- /dev/null +++ b/packages/mosaic/src/lease-broker/recovery_b1_adversarial_unittest.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +"""RED-first adversarial Claude B1 gate contracts. + +This private harness drives the shipped gate and daemon out of process. It +never contacts a live broker/runtime and executes a shell payload only after a +regression has already (incorrectly) received the recovery exemption. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + + +TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker" +FRAMEWORK = Path(__file__).parents[2] / "framework" +DAEMON = TOOLS / "daemon.py" +GATE = TOOLS / "mutator-gate.py" +RECOVERY = TOOLS / "recover-context.py" +SKILL = FRAMEWORK / "skills/mosaic-context-refresh/SKILL.md" + + +def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: + connection.settimeout(3.0) + connection.connect(str(socket_path)) + connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode()) + connection.shutdown(socket.SHUT_WR) + response = bytearray() + while True: + chunk = connection.recv(4096) + if not chunk: + break + response.extend(chunk) + if not response.endswith(b"\n") or response.count(b"\n") != 1: + raise AssertionError(f"unframed broker response: {bytes(response)!r}") + reply = json.loads(response[:-1]) + if not isinstance(reply, dict): + raise AssertionError("broker response is not an object") + return reply + + +def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None: + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + if socket_path.exists(): + return + if process.poll() is not None: + output = process.stdout.read() if process.stdout is not None else "" + raise RuntimeError(f"daemon exited before READY: {output}") + time.sleep(0.02) + raise TimeoutError("daemon did not create private socket") + + +class ClaudeRecoveryGateAdversarialTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + os.chmod(self.root, 0o700) + self.socket = self.root / "broker.sock" + self.daemon = subprocess.Popen( + [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.root / "state.json")], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + wait_ready(self.daemon, self.socket) + registered = request(self.socket, {"action": "register_anchor", "runtime_generation": 1}) + self.session_id = registered["session_id"] + self.environment = { + **os.environ, + "MOSAIC_LEASE_BROKER_SOCKET": str(self.socket), + "MOSAIC_LEASE_SESSION_ID": self.session_id, + "MOSAIC_RUNTIME_GENERATION": "1", + "MOSAIC_LEASE_RUNTIME": "claude", + } + # This private path is only a gate-classifier identity; if a regression + # blesses a payload, bash invokes no installed/live recovery command. + self.recovery_path = self.root / "recover-context.py" + self.canonical = ( + f"python3 {self.recovery_path} begin " + f"--construction {self.root / 'construction.json'} --compaction-epoch 0 --request-epoch 0" + ) + + def tearDown(self) -> None: + if self.daemon.poll() is None: + self.daemon.terminate() + try: + self.daemon.wait(timeout=3.0) + except subprocess.TimeoutExpired: + self.daemon.kill() + self.daemon.wait() + if self.daemon.stdout is not None: + self.daemon.stdout.close() + self.temporary.cleanup() + + def gate(self, command: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [ + sys.executable, + "-I", + "-S", + "-B", + str(GATE), + "--runtime", + "claude", + "--recovery-command", + str(self.recovery_path), + ], + input=json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}), + text=True, + capture_output=True, + env=self.environment, + check=False, + ) + + def command_for(self, position: str, value: str) -> str: + argv = [ + "python3", + str(self.recovery_path), + "begin", + "--construction", + str(self.root / "construction.json"), + "--compaction-epoch", + "0", + "--request-epoch", + "0", + ] + positions = { + "executable": 0, + "path": 1, + "phase": 2, + "construction_flag": 3, + "construction": 4, + "compaction_epoch_flag": 5, + "compaction_epoch": 6, + "request_epoch_flag": 7, + "request_epoch": 8, + } + try: + argv[positions[position]] = value + except KeyError as exc: + raise AssertionError(f"unknown argv position {position}") from exc + return " ".join(argv) + + def test_canonical_literal_and_shipped_skill_invocation_are_ungated(self) -> None: + self.assertEqual(self.gate(self.canonical).returncode, 0) + self.assertIn( + "python3 /home/hermes/.config/mosaic/tools/lease-broker/recover-context.py begin", + SKILL.read_text(encoding="utf-8"), + ) + + def test_every_shell_active_vector_in_every_argv_position_falls_through_to_bash_deny(self) -> None: + positions = ( + "executable", + "path", + "phase", + "construction_flag", + "construction", + "compaction_epoch_flag", + "compaction_epoch", + "request_epoch_flag", + "request_epoch", + ) + marker = self.root / "PWNED" + marker_vector = f"$(touch${{IFS}}{marker})" + vectors = { + "command-substitution": marker_vector, + "backtick": f"`touch${{IFS}}{marker}`", + "process-substitution": f"<(touch${{IFS}}{marker})", + "parameter-expansion": "${IFS}", + "home-expansion": "${HOME}", + "arithmetic-expansion": "$((1+1))", + "brace-expansion": "{a,b}", + "tilde-expansion": "~", + "glob": "*", + "redirection": ">", + "semicolon": f";touch${{IFS}}{marker}", + "and": f"&&touch${{IFS}}{marker}", + "pipe": f"|touch${{IFS}}{marker}", + "embedded-newline": "literal\nnext", + "quoting-trick": "'literal'", + } + for position in positions: + for kind, vector in vectors.items(): + with self.subTest(position=position, kind=kind): + marker.unlink(missing_ok=True) + command = self.command_for(position, vector) + gated = self.gate(command) + if gated.returncode == 0 and kind in {"command-substitution", "backtick", "process-substitution", "semicolon", "and", "pipe"}: + subprocess.run(["bash", "-c", command], cwd=self.root, env=self.environment, check=False) + self.assertEqual(gated.returncode, 2, f"unexpected recovery exemption: {command!r}") + self.assertFalse(marker.exists(), f"shell payload executed: {command!r}") + + +if __name__ == "__main__": + unittest.main()