test(mosaic): define adversarial Claude recovery gate contracts

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:*",

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