From 083dde24a1ece24c3eb63fb273e431c2f03b6420 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Sun, 19 Jul 2026 15:05:43 -0500 Subject: [PATCH 1/2] test(#831): add failing tests for verbatim-hashed fragments (B_payload/H_payload, no-self-ref, source-invalidation) --- packages/mosaic/package.json | 2 +- .../normative_fragments_unittest.py | 159 ++++++++++++++++++ 2 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 packages/mosaic/src/lease-broker/normative_fragments_unittest.py diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index 3ffda774..20bb7384 100644 --- a/packages/mosaic/package.json +++ b/packages/mosaic/package.json @@ -25,7 +25,7 @@ "lint": "eslint src", "typecheck": "tsc --noEmit", "test": "vitest run --passWithNoTests && pnpm run test:framework-shell", - "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" + "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh" }, "dependencies": { "@mosaicstack/brain": "workspace:*", diff --git a/packages/mosaic/src/lease-broker/normative_fragments_unittest.py b/packages/mosaic/src/lease-broker/normative_fragments_unittest.py new file mode 100644 index 00000000..0b910a33 --- /dev/null +++ b/packages/mosaic/src/lease-broker/normative_fragments_unittest.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""RED-first contract tests for verbatim-hashed normative fragments.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import tempfile +import unittest +from pathlib import Path + + +MODULE_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/normative_fragments.py" + + +def shipped_module(): + # Each test reaches the shipped implementation; no test doubles or local + # reimplementation of construction are allowed on this admission surface. + assert MODULE_PATH.is_file(), f"shipped construction module is missing: {MODULE_PATH}" + spec = importlib.util.spec_from_file_location("normative_fragments", MODULE_PATH) + if spec is None or spec.loader is None: + raise RuntimeError("unable to load normative fragment construction") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def fragment(module, source_id: str, content: bytes): + return module.NormativeFragment( + source_id=source_id, + content=content, + expected_sha256=hashlib.sha256(content).hexdigest(), + ) + + +def valid_fragments(module): + return [ + fragment(module, "authority/constitution", b"Constitution\n"), + fragment(module, "authority/runtime", b"Runtime\n"), + ] + + +class NormativeFragmentsTest(unittest.TestCase): + def test_t24_claude_and_pi_builders_are_byte_identical_and_one_way(self) -> None: + module = shipped_module() + fragments = valid_fragments(module) + + claude = module.build_for_claude( + manifest_version=1, + generator_version="wi4-test", + fragments=fragments, + ) + pi = module.build_for_pi( + manifest_version=1, + generator_version="wi4-test", + fragments=fragments, + ) + + self.assertEqual(claude.injectionDecision, "ACCEPTED") + self.assertTrue(claude.promotion) + self.assertEqual(claude.b_payload, pi.b_payload) + self.assertEqual(claude.h_payload, pi.h_payload) + self.assertEqual( + claude.h_payload, + hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + module.length_frame([claude.b_payload])).hexdigest(), + ) + self.assertNotIn(b"h_payload", claude.b_payload) + + mutated_fragment = module.build_for_claude( + manifest_version=1, + generator_version="wi4-test", + fragments=[ + fragment(module, "authority/constitution", b"Constitution changed\n"), + fragment(module, "authority/runtime", b"Runtime\n"), + ], + ) + mutated_metadata = module.build_for_claude( + manifest_version=2, + generator_version="wi4-test", + fragments=fragments, + ) + self.assertNotEqual(claude.h_payload, mutated_fragment.h_payload) + self.assertNotEqual(claude.h_payload, mutated_metadata.h_payload) + + def test_length_framing_and_domain_separation_prevent_ambiguous_construction(self) -> None: + module = shipped_module() + left = module.length_frame([b"ab", b"c"]) + right = module.length_frame([b"a", b"bc"]) + + self.assertNotEqual(left, right) + self.assertNotEqual( + hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(), + hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + right).hexdigest(), + ) + self.assertNotEqual( + hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(), + hashlib.sha256(b"other-context\x00" + left).hexdigest(), + ) + + def test_source_invalidation_missing_refuses_real_construction_and_promotion(self) -> None: + module = shipped_module() + missing = module.NormativeFragment( + source_id="authority/missing", + content=None, + expected_sha256=hashlib.sha256(b"missing").hexdigest(), + ) + + result = module.build_for_claude( + manifest_version=1, + generator_version="wi4-test", + fragments=[missing], + ) + + self.assertEqual(result.injectionDecision, "REFUSED") + self.assertFalse(result.promotion) + self.assertEqual(result.source_reason, "missing") + self.assertIsNone(result.b_payload) + self.assertIsNone(result.h_payload) + + def test_source_invalidation_oversize_refuses_real_construction_and_promotion(self) -> None: + module = shipped_module() + content = b"x" * (module.MAX_FRAGMENT_BYTES + 1) + oversize = fragment(module, "authority/oversize", content) + + result = module.build_for_pi( + manifest_version=1, + generator_version="wi4-test", + fragments=[oversize], + ) + + self.assertEqual(result.injectionDecision, "REFUSED") + self.assertFalse(result.promotion) + self.assertEqual(result.source_reason, "oversize") + self.assertIsNone(result.b_payload) + self.assertIsNone(result.h_payload) + + def test_source_invalidation_hash_mismatch_refuses_real_construction_and_promotion(self) -> None: + module = shipped_module() + mismatch = module.NormativeFragment( + source_id="authority/hash-mismatch", + content=b"trusted bytes", + expected_sha256=hashlib.sha256(b"different bytes").hexdigest(), + ) + + result = module.build_for_claude( + manifest_version=1, + generator_version="wi4-test", + fragments=[mismatch], + ) + + self.assertEqual(result.injectionDecision, "REFUSED") + self.assertFalse(result.promotion) + self.assertEqual(result.source_reason, "hash-mismatch") + self.assertIsNone(result.b_payload) + self.assertIsNone(result.h_payload) + + +if __name__ == "__main__": + unittest.main() -- 2.49.1 From c63baec14aceb64a117c3e31bcac46e626197cf9 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Sun, 19 Jul 2026 15:12:47 -0500 Subject: [PATCH 2/2] feat(#831): implement verbatim-hashed normative fragments construction --- .../tools/lease-broker/normative_fragments.py | 176 ++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 packages/mosaic/framework/tools/lease-broker/normative_fragments.py diff --git a/packages/mosaic/framework/tools/lease-broker/normative_fragments.py b/packages/mosaic/framework/tools/lease-broker/normative_fragments.py new file mode 100644 index 00000000..0d1c118e --- /dev/null +++ b/packages/mosaic/framework/tools/lease-broker/normative_fragments.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""Fail-closed construction of verbatim-hashed normative fragments. + +This is the single construction path for the Claude and Pi adapters. The +payload contains only versioned source metadata and exact validated source +bytes. ``h_payload`` is derived afterwards and is deliberately not representable +as a payload input, preventing cryptographic self-reference. +""" + +from __future__ import annotations + +import hashlib +import hmac +import struct +from collections.abc import Sequence +from typing import Final + +MAX_FRAGMENT_BYTES: Final = 64 * 1024 +HASH_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_PAYLOAD/v1\x00" +SOURCE_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_SOURCE/v1\x00" +_PAYLOAD_VERSION_LABEL: Final = b"MOSAIC/B_PAYLOAD/v1" + + +class NormativeFragment: + """A source identity, its expected digest, and its exact resolved bytes.""" + + def __init__(self, source_id: str, content: bytes | None, expected_sha256: str) -> None: + self.source_id = source_id + self.content = content + self.expected_sha256 = expected_sha256 + + +class ConstructionResult: + """A source-admission decision and, only when admitted, derived payload values. + + ``promotion`` means the construction has produced the only values a later + receipt protocol may use to attempt promotion. It never performs broker + promotion itself. A REFUSED result has no payload/hash values, so it cannot + advance to that later protocol. + """ + + def __init__( + self, + injection_decision: str, + promotion: bool, + source_reason: str | None, + b_payload: bytes | None, + h_payload: str | None, + h_source: str | None, + ) -> None: + self.injectionDecision = injection_decision + self.promotion = promotion + self.source_reason = source_reason + self.b_payload = b_payload + self.h_payload = h_payload + self.h_source = h_source + + +def length_frame(parts: Sequence[bytes]) -> bytes: + """Encode a finite ordered byte sequence with unambiguous 64-bit framing.""" + + framed = bytearray(struct.pack(">Q", len(parts))) + for part in parts: + if not isinstance(part, bytes): + raise TypeError("length framing requires bytes") + framed.extend(struct.pack(">Q", len(part))) + framed.extend(part) + return bytes(framed) + + +def _sha256_hex(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _valid_digest(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +def _source_reason(fragment: NormativeFragment) -> str | None: + if not isinstance(fragment.source_id, str) or not fragment.source_id: + return "missing" + if fragment.content is None: + return "missing" + if not isinstance(fragment.content, bytes): + return "missing" + if len(fragment.content) > MAX_FRAGMENT_BYTES: + return "oversize" + if not _valid_digest(fragment.expected_sha256): + return "hash-mismatch" + actual = _sha256_hex(fragment.content) + if not hmac.compare_digest(actual, fragment.expected_sha256): + return "hash-mismatch" + return None + + +def _refused(reason: str) -> ConstructionResult: + return ConstructionResult( + injection_decision="REFUSED", + promotion=False, + source_reason=reason, + b_payload=None, + h_payload=None, + h_source=None, + ) + + +def build_payload( + *, + manifest_version: int, + generator_version: str, + fragments: Sequence[NormativeFragment], +) -> ConstructionResult: + """Construct B_payload then H_payload after fail-closed source validation. + + The sequence order is caller-supplied resolved-source order and is encoded + directly. Changing source order, source identity, metadata, or any exact + fragment byte therefore changes the framed B_payload and its derived hash. + """ + + if type(manifest_version) is not int or manifest_version < 0: + return _refused("missing") + if not isinstance(generator_version, str) or not generator_version: + return _refused("missing") + + validated = list(fragments) + if not validated: + return _refused("missing") + for fragment in validated: + if not isinstance(fragment, NormativeFragment): + return _refused("missing") + reason = _source_reason(fragment) + if reason is not None: + return _refused(reason) + + source_identities = [ + length_frame([ + fragment.source_id.encode("utf-8"), + fragment.expected_sha256.encode("ascii"), + ]) + for fragment in validated + ] + h_source = _sha256_hex(SOURCE_DOMAIN_SEPARATOR + length_frame(source_identities)) + payload_parts = [ + _PAYLOAD_VERSION_LABEL, + str(manifest_version).encode("ascii"), + generator_version.encode("utf-8"), + *source_identities, + *(fragment.content for fragment in validated), + ] + # h_payload is intentionally not an argument or field in payload_parts. + b_payload = length_frame(payload_parts) + h_payload = _sha256_hex(HASH_DOMAIN_SEPARATOR + length_frame([b_payload])) + return ConstructionResult( + injection_decision="ACCEPTED", + promotion=True, + source_reason=None, + b_payload=b_payload, + h_payload=h_payload, + h_source=h_source, + ) + + +def build_for_claude(**kwargs: object) -> ConstructionResult: + """Claude adapter entrypoint; delegates to the sole shared constructor.""" + + return build_payload(**kwargs) # type: ignore[arg-type] + + +def build_for_pi(**kwargs: object) -> ConstructionResult: + """Pi adapter entrypoint; delegates to the sole shared constructor.""" + + return build_payload(**kwargs) # type: ignore[arg-type] -- 2.49.1