diff --git a/packages/mosaic/framework/tools/lease-broker/lease_promote.py b/packages/mosaic/framework/tools/lease-broker/lease_promote.py new file mode 100644 index 00000000..2a416b33 --- /dev/null +++ b/packages/mosaic/framework/tools/lease-broker/lease_promote.py @@ -0,0 +1,269 @@ +#!/usr/bin/env python3 +"""Lease promotion client — the half the enforcement toolkit never shipped. + +The enforcement half (``daemon.py`` + ``mutator-gate.py``) ships and denies. The +promotion half has no production caller anywhere in the package: as of 0.0.48, +0.0.49 and 0.0.50-next.2207, ``begin_verification`` / ``observe_receipt`` / +``promote_lease`` are invoked only by ``broker-test-client.ts``, the acceptance +spec, unit tests, and two probes under ``docs/``. Consequence: **no lease on any +host can reach VERIFIED**, so every mutator is denied ``MUTATOR_UNVERIFIED`` by a +gate nothing can satisfy. + +THE PROTOCOL (``daemon.py:578-754``) +------------------------------------ + 1. ``begin_verification`` — broker revokes, mints a challenge, and returns the + exact ``receipt`` text the MODEL must emit + 2. *the model emits that text verbatim as its ENTIRE latest message* + 3. the runtime adapter ships that message to the daemon-owned observer socket + 4. ``observe_receipt`` -> ``PENDING_PROMOTION`` + 5. ``promote_lease`` -> ``VERIFIED`` + +THIS MODULE IMPLEMENTS 1, 4 AND 5 — NEVER 2 +------------------------------------------- +Step 2 is the security property, not a formality. ``is_verbatim_receipt`` uses +``hmac.compare_digest`` against the exact minted string — explicitly "not a +transcript substring" (``receipt_challenge.py``). Promotion therefore requires a +live model that received the challenge in its context and echoed it exactly. + +``receipt-observer-client.py`` will post ANY string as the latest assistant +message. A promotion client that posted its own receipt would satisfy the broker +while proving nothing — a gate-disabler indistinguishable from a working fix +unless someone looks for it. **This module never posts a receipt.** Emitting it +belongs to the runtime adapter, where a real model turn happens. + +The construction binds the exact normative source bytes, so a VERIFIED lease +means "this agent is running THIS law", not merely "this session id is known". +``h_source`` / ``h_payload`` are derived by the framework's own +``normative_fragments.build_payload`` rather than reimplemented: the broker +derives them the same way and any divergence yields ``PAYLOAD_BINDING_MISMATCH``. +There must be exactly one implementation. + +Usage +----- + lease_promote.py --begin # prints the receipt the MODEL must emit + lease_promote.py --complete # after the adapter observed it +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import socket +import sys +from pathlib import Path +from typing import Final + +# Isolated (`python -I`) adapter invocations must still import co-located +# framework modules; never depend on the caller's PYTHONPATH. +_MODULE_DIRECTORY = str(Path(__file__).resolve().parent) +if _MODULE_DIRECTORY not in sys.path: + sys.path.insert(0, _MODULE_DIRECTORY) + +from normative_fragments import NormativeFragment, build_payload # noqa: E402 + +MAX_FRAME: Final = 64 * 1024 +BROKER_TIMEOUT_SECONDS: Final = 3.0 +SCHEMA_VERSION: Final = 1 +MANIFEST_VERSION: Final = 1 +GENERATOR_VERSION: Final = "mosaic/lease_promote@1" +DEFAULT_TTL_SECONDS: Final = 300 + +# Normative sources whose exact bytes bind the lease. Sources absent on a given +# deployment are simply not part of the binding — never fabricated. +FRAGMENT_SOURCES: Final = ( + "CONSTITUTION.md", + "AGENTS.md", + "SOUL.md", + "USER.md", + "STANDARDS.md", + "TOOLS.md", +) + + +def mosaic_home() -> Path: + return Path(os.environ.get("MOSAIC_HOME") or Path.home() / ".config" / "mosaic") + + +def broker_socket() -> Path: + value = os.environ.get("MOSAIC_LEASE_BROKER_SOCKET") + if value: + return Path(value) + runtime_dir = os.environ.get("XDG_RUNTIME_DIR") + if runtime_dir: + return Path(runtime_dir) / "mosaic-lease" / "broker.sock" + return Path(f"/run/user/{os.getuid()}/mosaic-lease/broker.sock") + + +def session_identity() -> tuple[str, int, str]: + """Session id, CURRENT generation, runtime. + + The generation file wins over the env var, matching ``lease_generation.py``. + Sending a generation HIGHER than the broker's would revoke this session's own + authority (``daemon.py:342-344``), so this never guesses. + """ + session_id = os.environ["MOSAIC_LEASE_SESSION_ID"] + runtime = os.environ["MOSAIC_LEASE_RUNTIME"] + state_file = os.environ.get("MOSAIC_LEASE_GENERATION_FILE") + if state_file: + try: + return session_id, int(Path(state_file).read_text().strip()), runtime + except (OSError, ValueError): + pass + return session_id, int(os.environ["MOSAIC_RUNTIME_GENERATION"]), runtime + + +def build_construction(runtime: str) -> tuple[dict[str, object], object]: + """Assemble the wire construction and derive its hashes with the sole builder.""" + sources = list(FRAGMENT_SOURCES) + [f"runtime/{runtime}/RUNTIME.md"] + wire_fragments: list[dict[str, str]] = [] + objects: list[NormativeFragment] = [] + + for source_id in sources: + try: + content = (mosaic_home() / source_id).read_bytes() + except OSError: + continue + import hashlib + + digest = hashlib.sha256(content).hexdigest() + wire_fragments.append( + { + "source_id": source_id, + "content_base64": base64.b64encode(content).decode("ascii"), + "expected_sha256": digest, + } + ) + objects.append(NormativeFragment(source_id, content, digest)) + + if not wire_fragments: + raise RuntimeError("no normative sources found — refusing to build an empty binding") + + result = build_payload( + manifest_version=MANIFEST_VERSION, + generator_version=GENERATOR_VERSION, + fragments=objects, + ) + if result.injectionDecision != "ACCEPTED" or not result.promotion: + raise RuntimeError(f"construction refused locally: {result.source_reason}") + + return ( + { + "manifest_version": MANIFEST_VERSION, + "generator_version": GENERATOR_VERSION, + "fragments": wire_fragments, + }, + result, + ) + + +def broker_request(payload: dict[str, object]) -> dict[str, object]: + raw = (json.dumps(payload, separators=(",", ":")) + "\n").encode() + if len(raw) > MAX_FRAME: + raise ValueError( + f"request too large ({len(raw)} bytes); broker frame cap is {MAX_FRAME}" + ) + response = bytearray() + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: + connection.settimeout(BROKER_TIMEOUT_SECONDS) + connection.connect(str(broker_socket())) + connection.sendall(raw) + connection.shutdown(socket.SHUT_WR) + while len(response) <= MAX_FRAME: + chunk = connection.recv(4096) + 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 begin( + ttl_seconds: int = DEFAULT_TTL_SECONDS, + compaction_epoch: int = 0, + request_epoch: int = 0, +) -> dict[str, object]: + """Step 1. Returns the broker reply, including the exact ``receipt`` text.""" + session_id, generation, runtime = session_identity() + construction, derived = build_construction(runtime) + return broker_request( + { + "action": "begin_verification", + "session_id": session_id, + "runtime_generation": generation, + "runtime": runtime, + "ttl_seconds": ttl_seconds, + "binding": { + "compaction_epoch": compaction_epoch, + "request_epoch": request_epoch, + "h_source": derived.h_source, + "h_payload": derived.h_payload, + "schema_version": SCHEMA_VERSION, + }, + "construction": construction, + } + ) + + +def complete(challenge: str) -> dict[str, object]: + """Steps 4-5. Assumes the model already emitted the receipt and the adapter + shipped it to the observer socket.""" + session_id, generation, _ = session_identity() + observed = broker_request( + { + "action": "observe_receipt", + "session_id": session_id, + "runtime_generation": generation, + "receipt_challenge": challenge, + } + ) + if observed.get("ok") is not True or observed.get("state") != "PENDING_PROMOTION": + return {"stage": "observe_receipt", **observed} + promoted = broker_request( + { + "action": "promote_lease", + "session_id": session_id, + "runtime_generation": generation, + "receipt_challenge": challenge, + } + ) + return {"stage": "promote_lease", **promoted} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Mosaic lease promotion client.") + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument( + "--begin", + action="store_true", + help="mint a challenge; prints the receipt the MODEL must emit verbatim", + ) + group.add_argument( + "--complete", + metavar="CHALLENGE", + help="observe the emitted receipt and promote the lease", + ) + parser.add_argument("--ttl-seconds", type=int, default=DEFAULT_TTL_SECONDS) + arguments = parser.parse_args(argv) + + try: + if arguments.begin: + print(json.dumps(begin(ttl_seconds=arguments.ttl_seconds), indent=2)) + else: + print(json.dumps(complete(arguments.complete), indent=2)) + except KeyError as exc: + print(f"missing lease environment: {exc}; not a lease-gated session", file=sys.stderr) + return 2 + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + print(f"{type(exc).__name__}: {exc}", file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())