test(mosaic): define adversarial Claude recovery gate contracts
This commit is contained in:
wjarvis mos-comms
@@ -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:*",
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user