From ba2716c11397b30e43f4ed9573b96acc24674bad Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Sun, 19 Jul 2026 16:30:23 -0500 Subject: [PATCH] feat(#832): enforce broker-minted receipt challenges --- .../832-receipt-challenge-protocol.md | 2 +- .../framework/tools/lease-broker/daemon.py | 110 ++++++++++++++---- .../tools/lease-broker/receipt_challenge.py | 43 +++++++ 3 files changed, 132 insertions(+), 23 deletions(-) create mode 100644 packages/mosaic/framework/tools/lease-broker/receipt_challenge.py diff --git a/docs/scratchpads/832-receipt-challenge-protocol.md b/docs/scratchpads/832-receipt-challenge-protocol.md index 1cbe9a61..8de808a9 100644 --- a/docs/scratchpads/832-receipt-challenge-protocol.md +++ b/docs/scratchpads/832-receipt-challenge-protocol.md @@ -10,4 +10,4 @@ 3. Implement the broker-minted receipt challenge and exact receipt observation/consume/promote path. 4. Run unit, framework-shell, compile, lint, and type checks; push after the required queue guard; report to `mosaic-100`. - **Risks:** The standalone harness must drive the real daemon without a divergent fixture. If that is impossible, stop and flag Mos. -- **Evidence:** Pending red run, green checks, and push. +- **Evidence:** RED recorded in `/home/hermes/agent-work/reviews/832-wi5-red-receipt-challenge.log`: receipt module absent before implementation. GREEN: receipt T26/T29 unittest (2 tests), normative-fragments unittest (5 tests), state-store regression (10 tests), `py_compile`, Mosaic package lint, and Mosaic package typecheck. Coverage tooling is unavailable (`python3 -m coverage`: module not installed); no probe was fired. Push pending. diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py index 1ac8e22f..596b7559 100644 --- a/packages/mosaic/framework/tools/lease-broker/daemon.py +++ b/packages/mosaic/framework/tools/lease-broker/daemon.py @@ -20,6 +20,13 @@ import time from pathlib import Path from typing import Final +# This script is loaded both as an executable and through isolated stdlib tests. +# Keep its co-located receipt implementation importable in both modes. +_MODULE_DIRECTORY = str(Path(__file__).resolve().parent) +if _MODULE_DIRECTORY not in sys.path: + sys.path.insert(0, _MODULE_DIRECTORY) +from receipt_challenge import is_verbatim_receipt, latest_assistant_digest, receipt_for + MAX_FRAME: Final = 64 * 1024 MAX_STATE: Final = 4 * 1024 * 1024 MAX_PENDING_TOKENS: Final = 256 @@ -90,6 +97,14 @@ def valid_binding(binding: object) -> bool: ) +def valid_receipt_evidence(evidence: object) -> bool: + return ( + isinstance(evidence, dict) + and set(evidence) == {"h_latest_assistant"} + and is_hex_256(evidence["h_latest_assistant"]) + ) + + def validate_state(value: object) -> dict[str, object]: if not isinstance(value, dict) or set(value) != {"version", "sessions", "tokens"}: raise BrokerFailure("STATE_INTEGRITY") @@ -124,7 +139,11 @@ def validate_state(value: object) -> dict[str, object]: for token_value, token in tokens.items(): if not is_hex_256(token_value) or not isinstance(token, dict): raise BrokerFailure("STATE_INTEGRITY") - if set(token) != {"session_id", "runtime_generation", "binding", "consumed"}: + token_fields = set(token) + if token_fields not in ( + {"session_id", "runtime_generation", "binding", "consumed"}, + {"session_id", "runtime_generation", "binding", "consumed", "evidence"}, + ): raise BrokerFailure("STATE_INTEGRITY") session_id = token["session_id"] generation = token["runtime_generation"] @@ -137,6 +156,9 @@ def validate_state(value: object) -> dict[str, object]: raise BrokerFailure("STATE_INTEGRITY") if not valid_binding(token["binding"]) or token["consumed"] is not False: raise BrokerFailure("STATE_INTEGRITY") + evidence = token.get("evidence") + if evidence is not None and not valid_receipt_evidence(evidence): + raise BrokerFailure("STATE_INTEGRITY") return value @@ -348,6 +370,9 @@ class Broker: "runtime_generation": generation, "binding": copy.deepcopy(binding), "consumed": False, + # Receipt evidence is durably committed before this challenge may + # be consumed and promotion made externally visible. + "evidence": None, } return token @@ -443,47 +468,88 @@ class Broker: raise BrokerFailure("INVALID_LEASE_TTL") # Revoke-first is a broker operation, not advisory adapter order. self.revoke_session_authority(session_id) - token = self.mint_token(session_id, request["runtime_generation"], binding) + cycle_binding = copy.deepcopy(binding) + cycle_binding["runtime_generation"] = request["runtime_generation"] + challenge = self.mint_token(session_id, request["runtime_generation"], binding) self.leases[session_id] = { "state": LEASE_PENDING, "runtime": runtime, "runtime_generation": request["runtime_generation"], "binding": copy.deepcopy(binding), - "promotion_token": token, + "receipt_challenge": challenge, "ttl_seconds": ttl_seconds, } + # The broker constructs both the bound challenge and the delivery + # text. The model's role is an exact copy, never hashing its output. return { "ok": True, "state": LEASE_PENDING, - "promotion_token": token, + "receipt_challenge": challenge, + "receipt": receipt_for(challenge, cycle_binding), + "binding": cycle_binding, } - if action == "promote_lease": + if action == "observe_receipt": session_id, _ = self.authenticate(peer_pid, request) - promotion_token = request.get("promotion_token") - lease = self.leases.get(session_id) - if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING: - raise BrokerFailure("INVALID_LEASE_TRANSITION") - expected_token = lease.get("promotion_token") - if ( - not isinstance(promotion_token, str) - or not isinstance(expected_token, str) - or not secrets.compare_digest(promotion_token, expected_token) - ): - raise BrokerFailure("PROMOTION_TOKEN_MISMATCH") - token = self.store.tokens().get(promotion_token) + challenge = request.get("receipt_challenge") + message = request.get("latest_assistant_message") + if not isinstance(challenge, str) or not isinstance(message, str): + raise BrokerFailure("INVALID_RECEIPT") + token = self.store.tokens().get(challenge) if ( not isinstance(token, dict) or token.get("session_id") != session_id or token.get("runtime_generation") != request.get("runtime_generation") - or token.get("binding") != lease.get("binding") or token.get("consumed") is not False ): - raise BrokerFailure("PROMOTION_TOKEN_INVALID") - del self.store.tokens()[promotion_token] + raise BrokerFailure("RECEIPT_REPLAY") + lease = self.leases.get(session_id) + if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING: + raise BrokerFailure("RECEIPT_REPLAY") + expected_challenge = lease.get("receipt_challenge") + binding = lease.get("binding") + if ( + not isinstance(expected_challenge, str) + or not secrets.compare_digest(challenge, expected_challenge) + or not isinstance(binding, dict) + or token.get("binding") != binding + ): + raise BrokerFailure("RECEIPT_REPLAY") + receipt_binding = copy.deepcopy(binding) + receipt_binding["runtime_generation"] = request["runtime_generation"] + # This accepts one exact current-cycle assistant entry only. It is + # deliberately not a transcript search and rejects quoted/extra text. + if not is_verbatim_receipt(message, challenge, receipt_binding): + raise BrokerFailure("RECEIPT_MISMATCH") + evidence = {"h_latest_assistant": latest_assistant_digest(message)} + token["evidence"] = evidence lease["state"] = LEASE_PENDING_PROMOTION + lease["evidence"] = copy.deepcopy(evidence) + return {"ok": True, "state": LEASE_PENDING_PROMOTION} + if action == "promote_lease": + session_id, _ = self.authenticate(peer_pid, request) + challenge = request.get("receipt_challenge") + if not isinstance(challenge, str): + raise BrokerFailure("INVALID_RECEIPT") + token = self.store.tokens().get(challenge) + if ( + not isinstance(token, dict) + or token.get("session_id") != session_id + or token.get("runtime_generation") != request.get("runtime_generation") + or token.get("consumed") is not False + ): + raise BrokerFailure("RECEIPT_REPLAY") + lease = self.leases.get(session_id) + if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING_PROMOTION: + raise BrokerFailure("INVALID_LEASE_TRANSITION") + expected_challenge = lease.get("receipt_challenge") + if not isinstance(expected_challenge, str) or not secrets.compare_digest(challenge, expected_challenge): + raise BrokerFailure("RECEIPT_REPLAY") + if token.get("binding") != lease.get("binding") or not valid_receipt_evidence(token.get("evidence")): + raise BrokerFailure("PROMOTION_TOKEN_INVALID") + del self.store.tokens()[challenge] lease["expires_at"] = time.monotonic() + int(lease["ttl_seconds"]) - # handle() commits token consumption before finish_promotion() makes - # VERIFIED externally visible: promote-last by construction. + # handle() commits the evidence-backed consumption before + # finish_promotion() makes VERIFIED externally visible: promote-last. return {"ok": True, "state": LEASE_PENDING_PROMOTION} if action == "revoke_lease": session_id, _ = self.authenticate(peer_pid, request) diff --git a/packages/mosaic/framework/tools/lease-broker/receipt_challenge.py b/packages/mosaic/framework/tools/lease-broker/receipt_challenge.py new file mode 100644 index 00000000..4c2f5c3d --- /dev/null +++ b/packages/mosaic/framework/tools/lease-broker/receipt_challenge.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +"""Broker-side construction and verification for one-time receipt challenges.""" + +from __future__ import annotations + +import hashlib +import hmac +import struct +from typing import Final + + +LATEST_ASSISTANT_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_LATEST_ASSISTANT/v1\x00" + + +def _length_frame(value: bytes) -> bytes: + return struct.pack(">Q", len(value)) + value + + +def receipt_for(challenge: str, binding: dict[str, object]) -> str: + """Return the sole receipt text a model may copy for this broker cycle.""" + + return ( + "MOSAIC-RECEIPT{" + f"challenge={challenge}; " + f"H_payload={binding['h_payload']}; " + f"gen={binding['runtime_generation']}; " + f"cep={binding['compaction_epoch']}" + "}" + ) + + +def is_verbatim_receipt(message: str, challenge: str, binding: dict[str, object]) -> bool: + """Require the exact one current-cycle receipt, not a transcript substring.""" + + expected = receipt_for(challenge, binding) + return hmac.compare_digest(message, expected) + + +def latest_assistant_digest(message: str) -> str: + """Record the broker-computed digest of the exact observed assistant entry.""" + + encoded = message.encode("utf-8") + return hashlib.sha256(LATEST_ASSISTANT_DOMAIN_SEPARATOR + _length_frame(encoded)).hexdigest()