218 lines
7.7 KiB
Python
218 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail CI when production code launches Claude/Pi outside the lease gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Final, NamedTuple, Sequence
|
|
|
|
SCANNED_ROOTS: Final = ("packages", "apps", "plugins", "tools")
|
|
SCANNED_SUFFIXES: Final = {
|
|
".bash",
|
|
".cjs",
|
|
".js",
|
|
".json",
|
|
".mjs",
|
|
".py",
|
|
".sh",
|
|
".ts",
|
|
".tsx",
|
|
".yaml",
|
|
".yml",
|
|
".zsh",
|
|
}
|
|
SKIPPED_DIRECTORIES: Final = {
|
|
".git",
|
|
".next",
|
|
".turbo",
|
|
"coverage",
|
|
"dist",
|
|
"node_modules",
|
|
}
|
|
|
|
# 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.
|
|
GATED_PATTERNS: Final = (
|
|
re.compile(r"\bexecLeaseGatedRuntime\s*\("),
|
|
re.compile(r"\blaunch-runtime\.(?:sh|py)\b"),
|
|
re.compile(
|
|
r"(?:\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)[\"'])"),
|
|
)
|
|
|
|
DIRECT_PATTERNS: Final = (
|
|
# Shell/process command forms, including here-doc command examples that an
|
|
# operator could execute verbatim.
|
|
re.compile(r"^\s*(?:claude|pi)(?:\s|$)"),
|
|
re.compile(r"(?:^|[;&|]\s*|\bexec\s+|\bcommand\s+)(?:claude|pi)\s+(?:-p\b|--dangerously\b|--print\b)"),
|
|
re.compile(r"\bexec\s+(?:claude|pi)(?:\s|$)"),
|
|
re.compile(r"\bexec\s+(?:/[^\s/]+)+/(?:claude|pi)(?:\s|$)"),
|
|
re.compile(r"\bexec\s+[\"']?\$(?:\{?runtime\}?|MOSAIC_AGENT_RUNTIME)\b"),
|
|
# JS/TS and Python process APIs with a literal runtime binary.
|
|
re.compile(
|
|
r"\b(?:spawn|spawnSync|exec|execSync|execFile|execFileSync|execv|execvp|execvpe|Popen|run|call|system|check_call|check_output)\s*\(\s*(?:\[\s*)?[\"'](?:claude|pi)(?:[\"']|\s)"
|
|
),
|
|
re.compile(
|
|
r"\b(?:spawn|spawnSync|exec|execSync|execFile|execFileSync|execv|execvp|execvpe|Popen|run|call|system|check_call|check_output)\s*\(\s*(?:\[\s*)?[\"'](?:/[^\"'/]+)+/(?:claude|pi)[\"']"
|
|
),
|
|
# Launch-command arrays and the prior @mosaicstack/coord dynamic default.
|
|
re.compile(r"\b(?:spawn|spawnSync|exec|execSync|execFile|execFileSync)\s*\(\s*runtime\b"),
|
|
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)"),
|
|
)
|
|
|
|
|
|
class LaunchSite(NamedTuple):
|
|
path: Path
|
|
line_number: int
|
|
line: str
|
|
classification: str
|
|
|
|
|
|
def is_test_path(path: Path) -> bool:
|
|
name = path.name.lower()
|
|
return (
|
|
"__tests__" in path.parts
|
|
or ".spec." in name
|
|
or ".test." in name
|
|
or name.endswith("_unittest.py")
|
|
or name.startswith("test-")
|
|
or name.startswith("test_")
|
|
)
|
|
|
|
|
|
def is_gated_line(line: str) -> bool:
|
|
return any(pattern.search(line) for pattern in GATED_PATTERNS)
|
|
|
|
|
|
def is_direct_line(line: str) -> bool:
|
|
return any(pattern.search(line) for pattern in DIRECT_PATTERNS)
|
|
|
|
|
|
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
|
|
|
|
|
|
def source_files(root: Path):
|
|
for relative_root in SCANNED_ROOTS:
|
|
search_root = root / relative_root
|
|
if not search_root.is_dir():
|
|
continue
|
|
for path in search_root.rglob("*"):
|
|
if not path.is_file() or path.suffix.lower() not in SCANNED_SUFFIXES:
|
|
continue
|
|
if any(part in SKIPPED_DIRECTORIES for part in path.parts):
|
|
continue
|
|
if is_test_path(path):
|
|
continue
|
|
yield path
|
|
|
|
|
|
def scan_repository(root: Path) -> list[LaunchSite]:
|
|
violations: list[LaunchSite] = []
|
|
for path in source_files(root):
|
|
try:
|
|
source = path.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
violations.append(LaunchSite(path, 0, "non-UTF-8 source", "unscannable"))
|
|
continue
|
|
violations.extend(scan_text(path.relative_to(root), source))
|
|
return violations
|
|
|
|
|
|
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()
|
|
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("\\")
|
|
return inventory
|
|
|
|
|
|
def format_violation(violation: LaunchSite) -> str:
|
|
return f"{violation.path}:{violation.line_number}: {violation.classification}: {violation.line.strip()}"
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", type=Path, default=Path.cwd())
|
|
parser.add_argument("--json", action="store_true")
|
|
arguments = parser.parse_args(argv)
|
|
root = arguments.root.resolve()
|
|
inventory = inventory_repository(root)
|
|
violations = [site for site in inventory if site.classification != "gated"]
|
|
|
|
if arguments.json:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"gated": sum(site.classification == "gated" for site in inventory),
|
|
"total": len(inventory),
|
|
"sites": [
|
|
{
|
|
"path": str(site.path),
|
|
"line": site.line_number,
|
|
"classification": site.classification,
|
|
"source": site.line.strip(),
|
|
}
|
|
for site in inventory
|
|
],
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
else:
|
|
for site in inventory:
|
|
print(format_violation(site))
|
|
print(
|
|
f"runtime launch inventory: "
|
|
f"{len(inventory) - len(violations)} gated/{len(inventory)} total"
|
|
)
|
|
|
|
if violations:
|
|
print("ungated consequential runtime launch detected", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|