feat(#829): enforce whole mutator-class lease gate

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

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""Register a runtime parent with the lease broker, then exec without changing PID."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import sys
+from pathlib import Path
+from typing import Final
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+
+
+def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            if not chunk:
+                break
+            response.extend(chunk)
+    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+        raise ValueError("invalid broker reply")
+    value = json.loads(response)
+    if not isinstance(value, dict):
+        raise ValueError("invalid broker reply")
+    return value
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("command", nargs=argparse.REMAINDER)
+    arguments = parser.parse_args()
+    command = arguments.command
+    if command and command[0] == "--":
+        command = command[1:]
+    if not command:
+        print("lease-gated runtime command is required", file=sys.stderr)
+        return 64
+
+    try:
+        socket_path = Path(os.environ["MOSAIC_LEASE_BROKER_SOCKET"])
+        generation = int(os.environ.get("MOSAIC_RUNTIME_GENERATION", "1"))
+        if generation < 0:
+            raise ValueError("invalid generation")
+        reply = broker_request(
+            socket_path,
+            {
+                "action": "register_anchor",
+                "runtime_generation": generation,
+            },
+        )
+        session_id = reply.get("session_id")
+        if (
+            reply.get("ok") is not True
+            or not isinstance(session_id, str)
+            or len(session_id) != 64
+            or any(character not in "0123456789abcdef" for character in session_id)
+        ):
+            raise ValueError("registration refused")
+    except (KeyError, ValueError, OSError, json.JSONDecodeError):
+        print("Mosaic lease broker registration failed; runtime launch denied.", file=sys.stderr)
+        return 1
+
+    environment = dict(os.environ)
+    environment["MOSAIC_LEASE_SESSION_ID"] = session_id
+    environment["MOSAIC_RUNTIME_GENERATION"] = str(generation)
+    environment["MOSAIC_LEASE_RUNTIME"] = arguments.runtime
+    try:
+        os.execvpe(command[0], command, environment)
+    except OSError:
+        print("Mosaic lease-gated runtime exec failed.", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())