fix(mosaic): harden literal Claude recovery gate mapping

This commit is contained in:
wjarvis mos-comms
2026-07-19 20:42:05 -05:00
parent 65e2bd71cf
commit b6deed04c7
5 changed files with 56 additions and 32 deletions

View File

@@ -8,7 +8,7 @@ import json
import os
import socket
import sys
import shlex
import re
from collections.abc import Callable, Mapping, Sequence
from pathlib import Path
from typing import BinaryIO, Final
@@ -23,6 +23,8 @@ from lease_generation import read_runtime_generation
MAX_FRAME: Final = 64 * 1024
BROKER_TIMEOUT_SECONDS: Final = 1.5
RECOVERY_TOOL: Final = "mosaic_context_recover"
_LITERAL_ABSOLUTE_PATH: Final = re.compile(r"/[A-Za-z0-9._/-]+\Z")
_SHELL_ACTIVE: Final = frozenset("$`~*?[]{}<>;|&" + '"' + "'" + "\\" + "\n\r\t")
def deny(code: str) -> int:
@@ -51,11 +53,13 @@ def read_tool_name(stream: BinaryIO | None = None) -> str:
def recovery_invocation_name(request: dict[str, object], recovery_command: Path | None) -> str:
"""Map only a direct Claude Bash invocation of the recovery executable.
"""Map only a byte-literal Claude recovery argv to ``RECOVERY_TOOL``.
The mapping is adapter-configured and remains broker-authenticated through
the ordinary authorization request. It never blesses Bash generally, shell
composition, a different executable, or unsupported recovery arguments.
Claude's Bash tool evaluates its raw command with a real shell. Therefore
the gate never attempts a second shell parser: any quote, expansion,
redirection, operator, glob, newline, or non-space whitespace is refused
before tokenizing. The remaining plain-space split is an exact argv proof,
not a best-effort interpretation of shell syntax.
"""
tool_name = request["tool_name"]
@@ -65,27 +69,33 @@ def recovery_invocation_name(request: dict[str, object], recovery_command: Path
if not isinstance(tool_input, dict) or set(tool_input) != {"command"}:
return str(tool_name)
command = tool_input.get("command")
if not isinstance(command, str) or not command:
if (
not isinstance(command, str)
or not command
or any(character in _SHELL_ACTIVE for character in command)
or command.startswith(" ")
or command.endswith(" ")
or " " in command
):
return str(tool_name)
try:
argv = shlex.split(command, posix=True)
except ValueError:
argv = command.split(" ")
if " ".join(argv) != command or len(argv) < 3 or argv[0] != "python3":
return str(tool_name)
if len(argv) < 3 or argv[0] != "python3":
return str(tool_name)
try:
if Path(argv[1]).resolve(strict=False) != recovery_command.resolve(strict=False):
return str(tool_name)
except OSError:
if argv[1] != str(recovery_command):
return str(tool_name)
phase = argv[2]
if phase == "complete" and len(argv) == 3:
return RECOVERY_TOOL
if phase != "begin" or len(argv) != 9:
return str(tool_name)
expected_flags = {"--construction", "--compaction-epoch", "--request-epoch"}
supplied_flags = set(argv[3::2])
if supplied_flags != expected_flags or any(not value for value in argv[4::2]):
if argv[3::2] != ["--construction", "--compaction-epoch", "--request-epoch"]:
return str(tool_name)
construction, compaction_epoch, request_epoch = argv[4::2]
if (
_LITERAL_ABSOLUTE_PATH.fullmatch(construction) is None
or not compaction_epoch.isdecimal()
or not request_epoch.isdecimal()
):
return str(tool_name)
return RECOVERY_TOOL