jason.woltje
abd2791f59
WI-2: mutator-class guard for the compaction directive-freshness mechanism — command-position parser (prefix x var-indirection unified) + primitive-anchored invariant backstop + all-tools-hook fail-close. Parser-complete on principle: realistic evasion matrix RED-regression-covered, residual exotic evasions proven backstop-caught (B-tests), lens-convergence reached. closes #829
212 lines
10 KiB
Python
212 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Contract tests for the permanent consequential-runtime launch guard."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import runpy
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
MOSAIC_ROOT = Path(__file__).parents[2]
|
|
REPO_ROOT = Path(__file__).parents[4]
|
|
GUARD_PATH = MOSAIC_ROOT / "framework/tools/lease-broker/check-runtime-launches.py"
|
|
SPEC = importlib.util.spec_from_file_location("runtime_launch_guard", GUARD_PATH)
|
|
if SPEC is None or SPEC.loader is None:
|
|
raise RuntimeError("unable to load runtime launch guard")
|
|
GUARD = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(GUARD)
|
|
|
|
|
|
class RuntimeLaunchGuardTest(unittest.TestCase):
|
|
def test_detects_direct_shell_and_process_api_launches(self) -> None:
|
|
cases = {
|
|
"shell-exec.sh": 'exec claude --dangerously-skip-permissions "prompt"\n',
|
|
"shell-print.sh": 'claude -p "prompt" | tee report.log\n',
|
|
"typescript.ts": "spawn('pi', ['--print', prompt]);\n",
|
|
"python.py": "subprocess.run(['claude', '-p', prompt])\n",
|
|
"dynamic.ts": "return [runtime, '-p', prompt];\n",
|
|
"plain-shell.sh": 'claude "$prompt"\n',
|
|
"node-exec.ts": 'exec("claude --print hello");\n',
|
|
"command-array.ts": "const launchCommand = ['pi', '--print', prompt];\n",
|
|
"python-system.py": 'os.system("claude -p prompt")\n',
|
|
"dynamic-shell.sh": 'exec "$runtime" "$prompt"\n',
|
|
"dynamic-spawn.ts": 'spawn(runtime, args);\n',
|
|
"dynamic-command.sh": 'LAUNCH_COMMAND=("$MOSAIC_AGENT_RUNTIME" --print)\n',
|
|
"absolute-shell.sh": 'exec /usr/local/bin/claude -p prompt\n',
|
|
"absolute-spawn.ts": "spawn('/opt/bin/pi', args);\n",
|
|
"terra-comment.sh": 'exec claude --dangerously-skip-permissions "terra-r3" # launch-runtime.py\n',
|
|
"python-comment.py": "subprocess.run(['claude', '-p', prompt]) # launch-runtime.py\n",
|
|
"typescript-comment.ts": "spawn('pi', args); // launch-runtime.py\n",
|
|
"marker-argument.sh": 'exec claude --dangerously-skip-permissions "launch-runtime.py"\n',
|
|
"marker-echo.sh": 'exec claude --dangerously-skip-permissions "prompt"; echo launch-runtime.py\n',
|
|
"marker-variable.sh": 'marker=launch-runtime.py; exec claude --dangerously-skip-permissions "prompt"\n',
|
|
"heredoc.sh": "cat <<'EOF'\nexec claude --dangerously-skip-permissions prompt # launch-runtime.py\nEOF\n",
|
|
"continued.sh": "exec \\\n claude --dangerously-skip-permissions prompt # launch-runtime.py\n",
|
|
"chain-semicolon.sh": "true; claude -p prompt\n",
|
|
"chain-and.sh": "true && claude -p prompt\n",
|
|
"chain-pipe.sh": "printf input | claude -p prompt\n",
|
|
"command-substitution.sh": "output=$(claude -p prompt)\n",
|
|
"eval.sh": "launcher='claude -p prompt'\neval \"$launcher\"\n",
|
|
"variable-exec.sh": "launcher=claude\n\"$launcher\" -p prompt\n",
|
|
"env-prefix.sh": "env SAFE=1 claude --help\n",
|
|
"command-prefix.sh": "command pi --help\n",
|
|
"nohup-prefix.sh": "nohup claude --help &\n",
|
|
}
|
|
for filename, source in cases.items():
|
|
with self.subTest(filename=filename):
|
|
violations = GUARD.scan_text(Path(filename), source)
|
|
self.assertNotEqual(violations, [], source)
|
|
|
|
def test_allows_only_explicit_gated_boundaries(self) -> None:
|
|
cases = {
|
|
"shell-helper.sh": 'exec "$GATED_RUNTIME" claude -- claude -p "prompt"\n',
|
|
"mosaic.sh": 'exec mosaic yolo "$runtime" "prompt"\n',
|
|
"launch.ts": "execLeaseGatedRuntime('claude', args);\n",
|
|
"coord.ts": "return ['mosaic', runtime, '-p', prompt];\n",
|
|
}
|
|
for filename, source in cases.items():
|
|
with self.subTest(filename=filename):
|
|
self.assertEqual(GUARD.scan_text(Path(filename), source), [])
|
|
|
|
def test_detects_prefixed_tracked_runtime_variable_execution(self) -> None:
|
|
multiline = {
|
|
"exec-quoted": 'exec "$v" -p x',
|
|
"exec-unquoted": "exec $v -p x",
|
|
"command": 'command "$v" -p x',
|
|
"nohup": 'nohup "$v" -p x',
|
|
"env": 'env A=1 "$v" -p x',
|
|
}
|
|
cases = {
|
|
**{f"multiline-{name}.sh": f"v=claude\n{command}\n" for name, command in multiline.items()},
|
|
**{f"same-line-{name}.sh": f"v=claude; {command}\n" for name, command in multiline.items()},
|
|
}
|
|
for filename, source in cases.items():
|
|
with self.subTest(filename=filename):
|
|
self.assertNotEqual(GUARD.scan_text(Path(filename), source), [], source)
|
|
|
|
def test_accepts_only_validated_multiline_typescript_wrapper_invocation(self) -> None:
|
|
source = """execRuntime(
|
|
'python3',
|
|
[launcher, ...dangerousArgs, '--runtime', runtime, '--', runtime, ...args],
|
|
environment,
|
|
);
|
|
"""
|
|
sites = GUARD.classify_text(Path("launch.ts"), source)
|
|
self.assertEqual(len(sites), 1)
|
|
self.assertEqual(sites[0].classification, "gated")
|
|
|
|
def test_marker_comments_strings_and_assignments_are_not_gated_sites(self) -> None:
|
|
harmless_sources = {
|
|
"comment.sh": "# launch-runtime.py --runtime claude --\n",
|
|
"echo.sh": "echo 'launch-runtime.py --runtime claude --'\n",
|
|
"assignment.sh": "marker='launch-runtime.py --runtime claude --'\n",
|
|
"argument.sh": "printf '%s' 'launch-runtime.py --runtime claude --'\n",
|
|
}
|
|
for filename, source in harmless_sources.items():
|
|
with self.subTest(filename=filename):
|
|
self.assertEqual(GUARD.classify_text(Path(filename), source), [])
|
|
|
|
def test_dangerous_primitive_backstops_parser_exotic_alias_indirection(self) -> None:
|
|
source = (
|
|
"alias hidden_runtime=claude\n"
|
|
"hidden_runtime --dangerously-skip-permissions -p x\n"
|
|
)
|
|
sites = GUARD.scan_text(Path("alias-launch.sh"), source)
|
|
self.assertEqual(len(sites), 1)
|
|
self.assertEqual(sites[0].classification, "dangerous-primitive")
|
|
|
|
def test_dangerous_primitive_is_owned_only_by_the_choke_point(self) -> None:
|
|
primitive = "--dangerously-skip-permissions"
|
|
self.assertNotEqual(GUARD.scan_text(Path("caller.ts"), f"args = ['{primitive}'];\n"), [])
|
|
self.assertEqual(
|
|
GUARD.scan_text(Path("framework/tools/lease-broker/launch-runtime.py"), f'FLAG = "{primitive}"\n'),
|
|
[],
|
|
)
|
|
|
|
def test_repository_has_no_ungated_consequential_runtime_launch(self) -> None:
|
|
violations = GUARD.scan_repository(REPO_ROOT)
|
|
self.assertEqual(
|
|
violations,
|
|
[],
|
|
"\n".join(GUARD.format_violation(violation) for violation in violations),
|
|
)
|
|
inventory = GUARD.inventory_repository(REPO_ROOT)
|
|
self.assertEqual(len(inventory), 14)
|
|
self.assertTrue(all(site.classification == "gated" for site in inventory))
|
|
|
|
def test_repository_walk_skips_tests_build_outputs_and_reports_unscannable_source(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
production = root / "packages/example/src/launch.sh"
|
|
production.parent.mkdir(parents=True)
|
|
production.write_text("exec claude -p prompt\n")
|
|
(production.parent / "launch.spec.ts").write_text("spawn('pi', [])\n")
|
|
dist = root / "packages/example/dist/launch.js"
|
|
dist.parent.mkdir(parents=True)
|
|
dist.write_text("exec('claude -p prompt')\n")
|
|
ignored_suffix = production.parent / "notes.txt"
|
|
ignored_suffix.write_text("claude -p prompt\n")
|
|
invalid = production.parent / "invalid.py"
|
|
invalid.write_bytes(b"\xff\xfe")
|
|
|
|
violations = GUARD.scan_repository(root)
|
|
formatted = [GUARD.format_violation(item) for item in violations]
|
|
self.assertEqual(len(violations), 2)
|
|
self.assertTrue(any("launch.sh:1: direct" in item for item in formatted))
|
|
self.assertTrue(any("invalid.py:0: unscannable" in item for item in formatted))
|
|
self.assertFalse(any("spec" in item or "dist" in item or "notes" in item for item in formatted))
|
|
|
|
inventory = GUARD.inventory_repository(root)
|
|
self.assertEqual({item.classification for item in inventory}, {"direct", "unscannable"})
|
|
|
|
def test_main_emits_machine_inventory_and_fails_on_a_direct_site(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "packages/example/launch.sh"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text(
|
|
'exec claude --dangerously-skip-permissions "terra-r3" # launch-runtime.py\n'
|
|
)
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
with redirect_stdout(stdout), redirect_stderr(stderr):
|
|
result = GUARD.main(["--root", str(root), "--json"])
|
|
payload = json.loads(stdout.getvalue())
|
|
self.assertEqual(result, 1)
|
|
self.assertEqual(payload["gated"], 0)
|
|
self.assertEqual(payload["total"], 1)
|
|
self.assertIn("ungated consequential runtime", stderr.getvalue())
|
|
|
|
def test_main_text_mode_reports_a_green_gated_inventory(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "packages/example/launch.sh"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("exec mosaic yolo claude prompt\n")
|
|
stdout = io.StringIO()
|
|
with redirect_stdout(stdout):
|
|
result = GUARD.main(["--root", str(root)])
|
|
self.assertEqual(result, 0)
|
|
self.assertIn("1 gated/1 total", stdout.getvalue())
|
|
|
|
def test_script_entrypoint_uses_current_directory_default(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "packages").mkdir()
|
|
with patch.object(sys, "argv", [str(GUARD_PATH)]), patch("pathlib.Path.cwd", return_value=root):
|
|
with redirect_stdout(io.StringIO()), self.assertRaises(SystemExit) as raised:
|
|
runpy.run_path(str(GUARD_PATH), run_name="__main__")
|
|
self.assertEqual(raised.exception.code, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|