44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
#!/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()
|