test(#832): define observer and payload admission regressions

packages/mosaic/src/lease-broker/receipt_challenge_unittest.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
-"""RED-first T26/T29 contract tests for the shipped receipt-challenge path."""
+"""RED-first contracts for the shipped receipt challenge and observer seam."""
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import importlib.util
 import os
 import tempfile
@@ -13,7 +15,7 @@ from pathlib import Path
 TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
 DAEMON_PATH = TOOLS / "daemon.py"
 FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
-RECEIPT_PATH = TOOLS / "receipt_challenge.py"
+OBSERVER_PATH = TOOLS / "receipt_observer.py"
 
 
 def load_module(name: str, path: Path):
@@ -28,16 +30,15 @@ def load_module(name: str, path: Path):
 
 DAEMON = load_module("lease_broker_receipt_daemon", DAEMON_PATH)
 FRAGMENTS = load_module("lease_broker_normative_fragments", FRAGMENTS_PATH)
-RECEIPTS = load_module("lease_broker_receipt_challenge", RECEIPT_PATH)
 
 
-class ReceiptChallengeTest(unittest.TestCase):
+class BrokerFixture(unittest.TestCase):
     def setUp(self) -> None:
         self.temporary = tempfile.TemporaryDirectory()
         root = Path(self.temporary.name)
         os.chmod(root, 0o700)
-        self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
         self.peer = (os.getpid(), os.getuid(), os.getgid())
+        self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
         registered = self.broker.handle(self.peer, {
             "action": "register_anchor",
             "runtime_generation": 7,
@@ -48,83 +49,137 @@ class ReceiptChallengeTest(unittest.TestCase):
     def tearDown(self) -> None:
         self.temporary.cleanup()
 
-    def binding(self, compaction_epoch: int, request_epoch: int) -> dict[str, object]:
+    def construction(self) -> tuple[dict[str, object], dict[str, object]]:
-        construction = FRAGMENTS.build_payload(
+        content = b"Constitution\n"
-            manifest_version=1,
+        expected_sha256 = hashlib.sha256(content).hexdigest()
-            generator_version="wi5-receipt-test",
+        construction = {
-            fragments=[
+            "manifest_version": 1,
-                FRAGMENTS.NormativeFragment(
+            "generator_version": "wi5-receipt-test",
-                    "authority/constitution",
+            "fragments": [{
-                    b"Constitution\n",
+                "source_id": "authority/constitution",
-                    "b1f7989bf4a3feee87aaf841d3c94385337f12cdbf356e0db33f68edfe535f16",
+                "content_base64": base64.b64encode(content).decode("ascii"),
-                ),
+                "expected_sha256": expected_sha256,
-            ],
+            }],
+        }
+        result = FRAGMENTS.build_payload(
+            manifest_version=construction["manifest_version"],
+            generator_version=construction["generator_version"],
+            fragments=[FRAGMENTS.NormativeFragment("authority/constitution", content, expected_sha256)],
         )
-        self.assertEqual(construction.injectionDecision, "ACCEPTED")
+        self.assertEqual(result.injectionDecision, "ACCEPTED")
-        self.assertTrue(construction.promotion)
+        self.assertTrue(result.promotion)
-        return {
+        return construction, {
-            "compaction_epoch": compaction_epoch,
+            "compaction_epoch": 3,
-            "request_epoch": request_epoch,
+            "request_epoch": 8,
-            "h_source": construction.h_source,
+            "h_source": result.h_source,
-            "h_payload": construction.h_payload,
+            "h_payload": result.h_payload,
             "schema_version": 1,
         }
 
-    def begin(self, binding: dict[str, object]) -> dict[str, object]:
+    def begin(self, binding: dict[str, object], construction: dict[str, object]) -> dict[str, object]:
         response = self.broker.handle(self.peer, {
             "action": "begin_verification",
             "session_id": self.session_id,
             "runtime_generation": 7,
             "runtime": "pi",
             "binding": binding,
+            "construction": construction,
         })
         self.assertEqual(response["state"], DAEMON.LEASE_PENDING)
         self.assertIsInstance(response.get("receipt_challenge"), str)
         self.assertIsInstance(response.get("receipt"), str)
         return response
 
-    def observe(self, challenge: str, message: str) -> dict[str, object]:
+
+class BuildPayloadAdmissionTest(BrokerFixture):
+    def test_b3_forged_h_source_or_h_payload_is_refused_against_shipped_build_payload(self) -> None:
+        construction, trusted = self.construction()
+        for field in ("h_source", "h_payload"):
+            with self.subTest(field=field):
+                forged = dict(trusted)
+                forged[field] = "f" * 64
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, "PAYLOAD_BINDING_MISMATCH"):
+                    self.begin(forged, construction)
+
+
+class ReceiptObserverTest(BrokerFixture):
+    def setUp(self) -> None:
+        super().setUp()
+        observers = load_module("lease_broker_test_observer", OBSERVER_PATH)
+        self.observer = observers.TestReceiptObserver()
+        self.broker = DAEMON.Broker(self.broker.store, observer=self.observer)
+
+    def record(self, message: str) -> None:
+        self.observer.record_latest_assistant_message(self.session_id, 7, message)
+
+    def observe(self, challenge: str, **untrusted: object) -> dict[str, object]:
         return self.broker.handle(self.peer, {
             "action": "observe_receipt",
             "session_id": self.session_id,
             "runtime_generation": 7,
             "receipt_challenge": challenge,
-            "latest_assistant_message": message,
+            **untrusted,
         })
 
-    def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
+    def promote(self, challenge: str) -> dict[str, object]:
-        stale = self.begin(self.binding(compaction_epoch=3, request_epoch=8))
+        return self.broker.handle(self.peer, {
-        current = self.begin(self.binding(compaction_epoch=4, request_epoch=9))
-        self.assertNotEqual(stale["receipt_challenge"], current["receipt_challenge"])
-
-        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
-            self.observe(current["receipt_challenge"], stale["receipt"])
-        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
-            self.broker.handle(self.peer, {
             "action": "promote_lease",
             "session_id": self.session_id,
             "runtime_generation": 7,
-                "receipt_challenge": current["receipt_challenge"],
+            "receipt_challenge": challenge,
         })
 
+    def test_b2_echoed_request_observation_is_refused_but_observer_source_promotes(self) -> None:
+        construction, binding = self.construction()
+        cycle = self.begin(binding, construction)
+        challenge = cycle["receipt_challenge"]
+        receipt = cycle["receipt"]
+
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_RECEIPT"):
+            self.observe(challenge, latest_assistant_message=receipt)
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_OBSERVATION_UNAVAILABLE"):
+            self.observe(challenge)
+
+        self.record(receipt)
+        self.assertEqual(self.observe(challenge)["state"], DAEMON.LEASE_PENDING_PROMOTION)
+        self.assertEqual(self.promote(challenge)["state"], DAEMON.LEASE_VERIFIED)
+        self.assertEqual(self.broker.handle(self.peer, {
+            "action": "authorize_tool",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "runtime": "pi",
+            "tool_name": "bash",
+        })["decision"], "allow")
+
+    def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
+        construction, stale_binding = self.construction()
+        stale = self.begin(stale_binding, construction)
+        current_binding = dict(stale_binding)
+        current_binding["compaction_epoch"] = 4
+        current_binding["request_epoch"] = 9
+        current = self.begin(current_binding, construction)
+        self.assertNotEqual(stale["receipt_challenge"], current["receipt_challenge"])
+
+        self.record(stale["receipt"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
+            self.observe(current["receipt_challenge"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
+            self.promote(current["receipt_challenge"])
+
     def test_t29_altered_model_hash_cannot_promote_against_shipped_binding(self) -> None:
-        cycle = self.begin(self.binding(compaction_epoch=5, request_epoch=10))
+        construction, binding = self.construction()
+        cycle = self.begin(binding, construction)
         expected = cycle["receipt"]
-        self.assertIsInstance(expected, str)
         altered_hash = "f" * 64
         self.assertNotEqual(altered_hash, cycle["binding"]["h_payload"])
         altered = expected.replace(cycle["binding"]["h_payload"], altered_hash, 1)
         self.assertNotEqual(altered, expected)
 
+        self.record(altered)
         with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
-            self.observe(cycle["receipt_challenge"], altered)
+            self.observe(cycle["receipt_challenge"])
         with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
-            self.broker.handle(self.peer, {
+            self.promote(cycle["receipt_challenge"])
-                "action": "promote_lease",
-                "session_id": self.session_id,
-                "runtime_generation": 7,
-                "receipt_challenge": cycle["receipt_challenge"],
-            })
 
 
 if __name__ == "__main__":