fix(#829): harden runtime launch guard against marker evasions

packages/mosaic/framework/tools/lease-broker/check-runtime-launches.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import json
 import re
+import shlex
 import sys
 from pathlib import Path
 from typing import Final, NamedTuple, Sequence
@@ -33,15 +34,22 @@ SKIPPED_DIRECTORIES: Final = {
     "dist",
     "node_modules",
 }
+DANGEROUS_PRIMITIVE: Final = "--dangerously-" + "skip-permissions"
+CHOKE_POINT_SUFFIX: Final = "framework/tools/lease-broker/launch-runtime.py"
+SHELL_SUFFIXES: Final = {".bash", ".sh", ".zsh"}
 
-# A launch line is gated only when it names the common wrapper, the TypeScript
-# adapter that invokes that wrapper, or the `mosaic` runtime command that reaches
-# the adapter. Merely mentioning a session ID or hook is not sufficient.
+# A launch line is gated only when it CALLS the common wrapper in command
+# position, invokes the TypeScript adapter, or constructs/runs a `mosaic`
+# runtime command. A marker in a comment, string argument, echo, or unrelated
+# variable can never satisfy these invocation-shaped patterns.
 GATED_PATTERNS: Final = (
-    re.compile(r"\bexecLeaseGatedRuntime\s*\("),
-    re.compile(r"\blaunch-runtime\.(?:sh|py)\b"),
+    re.compile(r"(?:^\s*|=>\s*)execLeaseGatedRuntime\s*\("),
     re.compile(
-        r"(?:\bexec\s+|\b(?:LAUNCH_COMMAND|launch_cmd)\s*=\s*\(?|\becho\s+[\"'])"
+        r"\bexecRuntime\s*\(\s*[\"']python3[\"']\s*,\s*"
+        r"\[\s*launcher\s*,.*[\"']--runtime[\"']\s*,\s*runtime\s*,\s*[\"']--[\"']"
+    ),
+    re.compile(
+        r"(?:^|[;&|]\s*|\bexec\s+|\b(?:LAUNCH_COMMAND|launch_cmd)\s*=\s*\(?|\becho\s+[\"'])"
         r"mosaic\s+(?:yolo\s+)?(?:claude|pi|claudex|[\"']?\$\{?runtime\}?[\"']?|[\"']?\$MOSAIC_AGENT_RUNTIME[\"']?)\b"
     ),
     re.compile(r"\[\s*[\"']mosaic[\"']\s*,\s*(?:[\"']yolo[\"']\s*,\s*)?(?:runtime|[\"'](?:claude|pi|claudex)[\"'])"),
@@ -67,6 +75,17 @@ DIRECT_PATTERNS: Final = (
     re.compile(r"\b(?:command|launchCommand|LAUNCH_COMMAND)\s*=\s*(?:\(|\[)\s*[\"']?(?:claude|pi)\b"),
     re.compile(r"\b(?:command|launchCommand|LAUNCH_COMMAND)\s*=\s*(?:\(|\[)\s*[\"']?\$(?:\{?runtime\}?|MOSAIC_AGENT_RUNTIME)\b"),
     re.compile(r"\breturn\s*\[\s*runtime\s*,\s*[\"'](?:-p|--dangerously)"),
+    re.compile(r"\$\(\s*(?:claude|pi)(?:\s|$)"),
+    re.compile(r"\beval\s+[\"'](?:claude|pi)(?:\s|[\"'])"),
+)
+RUNTIME_ASSIGNMENT: Final = re.compile(
+    r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=\s*[\"']?(?:claude|pi)(?:\s|[\"']|$)"
+)
+TYPESCRIPT_WRAPPER_INVOCATION: Final = re.compile(
+    r"(?m)^\s*execRuntime\s*\(\s*[\"']python3[\"']\s*,\s*"
+    r"\[\s*launcher\s*,[^\]]*[\"']--runtime[\"']\s*,\s*runtime\s*,\s*"
+    r"[\"']--[\"']\s*,\s*runtime(?:\s*,|\s*\])",
+    re.DOTALL,
 )
 
 
@@ -89,27 +108,215 @@ def is_test_path(path: Path) -> bool:
     )
 
 
-def is_gated_line(line: str) -> bool:
-    return any(pattern.search(line) for pattern in GATED_PATTERNS)
+def strip_comments(path: Path, line: str, in_block_comment: bool = False) -> tuple[str, bool]:
+    suffix = path.suffix.lower()
+    hash_comments = suffix in {".bash", ".py", ".sh", ".yaml", ".yml", ".zsh"}
+    slash_comments = suffix in {".cjs", ".js", ".mjs", ".ts", ".tsx"}
+    output: list[str] = []
+    quote: str | None = None
+    escaped = False
+    index = 0
+
+    while index < len(line):
+        if in_block_comment:
+            end = line.find("*/", index)
+            if end < 0:
+                return "".join(output), True
+            in_block_comment = False
+            index = end + 2
+            continue
+
+        character = line[index]
+        following = line[index + 1] if index + 1 < len(line) else ""
+        if quote is not None:
+            output.append(character)
+            if escaped:
+                escaped = False
+            elif character == "\\":
+                escaped = True
+            elif character == quote:
+                quote = None
+            index += 1
+            continue
+
+        if character in {"'", '"', "`"}:
+            quote = character
+            output.append(character)
+            index += 1
+            continue
+        if slash_comments and character == "/" and following == "/":
+            break
+        if slash_comments and character == "/" and following == "*":
+            in_block_comment = True
+            index += 2
+            continue
+        if hash_comments and character == "#":
+            if suffix == ".py" or index == 0 or line[index - 1].isspace():
+                break
+        output.append(character)
+        index += 1
+
+    return "".join(output), in_block_comment
 
 
-def is_direct_line(line: str) -> bool:
-    return any(pattern.search(line) for pattern in DIRECT_PATTERNS)
+def is_choke_point(path: Path) -> bool:
+    return path.as_posix().endswith(CHOKE_POINT_SUFFIX)
+
+
+def shell_commands(line: str) -> list[list[str]]:
+    candidate = line.rstrip().removesuffix("\\").rstrip()
+    try:
+        lexer = shlex.shlex(candidate, posix=True, punctuation_chars=";&|")
+        lexer.whitespace_split = True
+        lexer.commenters = ""
+        tokens = list(lexer)
+    except ValueError:
+        return []
+
+    commands: list[list[str]] = []
+    current: list[str] = []
+    for token in tokens:
+        if token and all(character in ";&|" for character in token):
+            if current:
+                commands.append(current)
+                current = []
+        else:
+            current.append(token)
+    if current:
+        commands.append(current)
+    return commands
+
+
+def is_wrapper_invocation(path: Path, line: str) -> bool:
+    if path.suffix.lower() not in SHELL_SUFFIXES:
+        return False
+    separator = re.search(r"\s--(?:\s|$)", line)
+    if separator is None:
+        return False
+    # Everything after the wrapper separator is opaque runtime argv and may
+    # contain an open quote continued on later physical lines. Parse only the
+    # complete command-position prefix through the separator.
+    wrapper_prefix = line[: separator.end()]
+    for command in shell_commands(wrapper_prefix):
+        if command and command[0] == "exec":
+            command = command[1:]
+        if not command:
+            continue
+        if command[0] == "python3":
+            if len(command) < 2 or not command[1].endswith("launch-runtime.py"):
+                continue
+            arguments = command[2:]
+        elif command[0].endswith(("launch-runtime.py", "launch-runtime.sh")):
+            arguments = command[1:]
+        else:
+            continue
+        try:
+            runtime_index = arguments.index("--runtime")
+            separator_index = arguments.index("--")
+        except ValueError:
+            continue
+        if runtime_index + 1 < len(arguments) and runtime_index < separator_index:
+            return True
+    return False
+
+
+def is_gated_line(path: Path, line: str) -> bool:
+    return is_wrapper_invocation(path, line) or any(
+        pattern.search(line) for pattern in GATED_PATTERNS
+    )
+
+
+def is_shell_direct_invocation(path: Path, line: str) -> bool:
+    if path.suffix.lower() not in SHELL_SUFFIXES:
+        return False
+    assignment = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
+    case_arm = re.match(r"^\s*[^;&()]+\)\s*", line)
+    command_source = line[case_arm.end() :] if case_arm is not None else line
+    for command in shell_commands(command_source):
+        index = 0
+        while index < len(command) and assignment.match(command[index]):
+            index += 1
+        while index < len(command) and command[index] in {"command", "exec", "nohup"}:
+            index += 1
+        if index < len(command) and command[index] == "env":
+            index += 1
+            while index < len(command) and (
+                command[index].startswith("-") or assignment.match(command[index])
+            ):
+                index += 1
+            while index < len(command) and command[index] in {"command", "exec", "nohup"}:
+                index += 1
+        if index < len(command) and Path(command[index]).name in {"claude", "pi"}:
+            return True
+    return False
+
+
+def is_direct_line(path: Path, line: str) -> bool:
+    return is_shell_direct_invocation(path, line) or any(
+        pattern.search(line) for pattern in DIRECT_PATTERNS
+    )
+
+
+def executes_runtime_variable(line: str, runtime_variables: set[str]) -> bool:
+    for variable in runtime_variables:
+        reference = rf"\$(?:{re.escape(variable)}|\{{{re.escape(variable)}\}})"
+        if re.search(rf"\beval\s+[\"']?{reference}", line):
+            return True
+        if re.match(rf"^\s*[\"']?{reference}[\"']?(?:\s|$)", line):
+            return True
+    return False
+
+
+def typescript_wrapper_lines(path: Path, source: str) -> set[int]:
+    if path.suffix.lower() not in {".js", ".mjs", ".ts", ".tsx"}:
+        return set()
+    return {
+        source.count("\n", 0, match.start()) + 1
+        for match in TYPESCRIPT_WRAPPER_INVOCATION.finditer(source)
+    }
+
+
+def classify_text(path: Path, source: str) -> list[LaunchSite]:
+    sites: list[LaunchSite] = []
+    validated_typescript_wrappers = typescript_wrapper_lines(path, source)
+    runtime_variables: set[str] = set()
+    gated_continuation = False
+    in_block_comment = False
+    for line_number, physical_line in enumerate(source.splitlines(), start=1):
+        line, in_block_comment = strip_comments(path, physical_line, in_block_comment)
+        stripped = line.strip()
+        if not stripped:
+            gated_continuation = False
+            continue
+
+        assignment = RUNTIME_ASSIGNMENT.match(line)
+        if assignment is not None:
+            runtime_variables.add(assignment.group(1))
+        primitive_violation = DANGEROUS_PRIMITIVE in line and not is_choke_point(path)
+        line_is_direct = is_direct_line(path, line) or executes_runtime_variable(
+            line, runtime_variables
+        )
+        line_is_gated = line_number in validated_typescript_wrappers or is_gated_line(path, line)
+
+        if primitive_violation:
+            sites.append(
+                LaunchSite(path, line_number, physical_line.rstrip(), "dangerous-primitive")
+            )
+        elif line_is_direct and not gated_continuation:
+            # Direct syntax always wins over a same-line marker. Only the command
+            # continuation of a previously validated wrapper may contain the raw
+            # runtime binary itself.
+            sites.append(LaunchSite(path, line_number, physical_line.rstrip(), "direct"))
+        elif line_is_gated:
+            sites.append(LaunchSite(path, line_number, physical_line.rstrip(), "gated"))
+
+        gated = line_is_gated or gated_continuation
+        gated_continuation = gated and line.rstrip().endswith("\\")
+    return sites
 
 
 def scan_text(path: Path, source: str) -> list[LaunchSite]:
-    violations: list[LaunchSite] = []
-    gated_continuation = False
-    for line_number, line in enumerate(source.splitlines(), start=1):
-        stripped = line.strip()
-        if not stripped or stripped.startswith(("#", "//", "*")):
-            gated_continuation = False
-            continue
-        gated = is_gated_line(line) or gated_continuation
-        if is_direct_line(line) and not gated:
-            violations.append(LaunchSite(path, line_number, line.rstrip(), "direct"))
-        gated_continuation = gated and line.rstrip().endswith("\\")
-    return violations
+    return [site for site in classify_text(path, source) if site.classification != "gated"]
 
 
 def source_files(root: Path):
@@ -143,27 +350,12 @@ def inventory_repository(root: Path) -> list[LaunchSite]:
     inventory: list[LaunchSite] = []
     for path in source_files(root):
         try:
-            lines = path.read_text(encoding="utf-8").splitlines()
+            source = path.read_text(encoding="utf-8")
         except UnicodeDecodeError:
             inventory.append(LaunchSite(path.relative_to(root), 0, "non-UTF-8 source", "unscannable"))
             continue
-        gated_continuation = False
-        for line_number, line in enumerate(lines, start=1):
-            stripped = line.strip()
-            if not stripped or stripped.startswith(("#", "//", "*")):
-                gated_continuation = False
-                continue
-            line_is_gated = is_gated_line(line)
-            gated = line_is_gated or gated_continuation
-            if line_is_gated and not stripped.startswith("function execLeaseGatedRuntime"):
-                inventory.append(
-                    LaunchSite(path.relative_to(root), line_number, line.rstrip(), "gated")
-                )
-            elif is_direct_line(line) and not gated:
-                inventory.append(
-                    LaunchSite(path.relative_to(root), line_number, line.rstrip(), "direct")
-                )
-            gated_continuation = gated and line.rstrip().endswith("\\")
+        relative_path = path.relative_to(root)
+        inventory.extend(classify_text(relative_path, source))
     return inventory
 
 

packages/mosaic/framework/tools/lease-broker/launch-runtime.py

@@ -14,6 +14,7 @@ from typing import Final
 
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5
+CLAUDE_DANGEROUS_FLAG: Final = "--dangerously-skip-permissions"
 
 
 def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
@@ -46,6 +47,7 @@ def main(
 ) -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--dangerous", action="store_true")
     parser.add_argument("command", nargs=argparse.REMAINDER)
     arguments = parser.parse_args(argv)
     command = arguments.command
@@ -54,6 +56,11 @@ def main(
     if not command:
         print("lease-gated runtime command is required", file=sys.stderr)
         return 64
+    if arguments.dangerous:
+        if arguments.runtime != "claude" or Path(command[0]).name != "claude":
+            print("dangerous mode is supported only for the Claude runtime", file=sys.stderr)
+            return 64
+        command = [command[0], CLAUDE_DANGEROUS_FLAG, *command[1:]]
 
     source_environment = os.environ if environ is None else environ
     try: