packages/cultnet-py/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased (breaking; version 0.x, so the next minor)
 
+- Decodes the write receipt frames of `contracts/cultnet/interop/write-receipt.md`: `parse_message` validates `cultnet.write_admitted.v0` and `cultnet.error.v1` against their closed schemas (unknown keys, empty or non-string ids, an unknown operation or code are refused; `error.v1` requires `messageId`, string or null; `limit_reached` is `error.v1` only). Added `CULTNET_ERROR_V1_CODES`, `CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS`, `write_content_hash` and `hello_advertises_write_receipts`; `hello` takes `supported_message_versions`. `tests/test_write_receipt.py` replays the vectors' `contentHash`, `wire` and `schemaInvalid` sections. Not implemented here: the client write tracker (`scenarios`) and the server rules (`dedupe`, `blankFields`); this package has no receipt server and no write client.
 - Added `message_invalid` to `CULTNET_ERROR_CODES` and `message_invalid_error(field, reason)`, byte-identical to the C# `ForMessageInvalid`. `cultmesh-py` refuses a malformed put or delete with it instead of the out-of-schema `malformed_document_put` and `malformed_document_delete`, which are removed.
 - Added `CultNetRudpSocketTransportServer` (`cultnet_py.rudp_server`): one UDP socket serving many RUDP peers, one session per
   remote endpoint, ported from the C# `CultNetRudpSocketTransportServer` and held to its bytes by

packages/cultnet-py/src/cultnet_py/__init__.py

@@ -39,6 +39,8 @@ from .cultmesh_contracts import (
 from .framing import decode_frame, encode_frame, read_frame, write_frame
 from .messages import (
     CULTNET_ERROR_CODES,
+    CULTNET_ERROR_V1_CODES,
+    CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
     CultNetMessage,
     CultNetWitnessArtifactBundle,
     compute_simulation_claim_hash,
@@ -51,7 +53,9 @@ from .messages import (
     error_message,
     message_invalid_error,
     unsupported_schema_version_error,
+    write_content_hash,
     hello,
+    hello_advertises_write_receipts,
     login,
     login_success,
     parse_message,
@@ -176,6 +180,8 @@ __all__ = [
     "verse_catalog_request",
     "verse_from_wire",
     "CULTNET_ERROR_CODES",
+    "CULTNET_ERROR_V1_CODES",
+    "CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS",
     "CultNetMessage",
     "CultNetWitnessArtifactBundle",
     "CultNetDatabaseSubscription",
@@ -266,6 +272,8 @@ __all__ = [
     "message_invalid_error",
     "unsupported_schema_version_error",
     "hello",
+    "hello_advertises_write_receipts",
+    "write_content_hash",
     "INTEROP_WIRE_CONTRACT",
     "login",
     "login_success",

packages/cultnet-py/src/cultnet_py/messages.py

@@ -29,7 +29,11 @@ def parse_message(payload: bytes) -> CultNetMessage:
     body = dict(decoded)
     del body["schemaVersion"]
     if schema_version == "cultnet.error.v0":
-        _validate_error_body(body)
+        _validate_error_body(body, "cultnet.error.v0", CULTNET_ERROR_CODES)
+    elif schema_version == "cultnet.error.v1":
+        _validate_error_body(body, "cultnet.error.v1", CULTNET_ERROR_V1_CODES)
+    elif schema_version == "cultnet.write_admitted.v0":
+        _validate_write_admitted_body(body)
     return CultNetMessage(schema_version=schema_version, body=body)
 
 
@@ -47,6 +51,13 @@ CULTNET_ERROR_CODES: tuple[str, ...] = (
     "message_invalid",
 )
 
+# cultnet.error.v1 (contracts/cultnet/interop/cultnet.error.v1.schema.json): v0's codes plus limit_reached.
+CULTNET_ERROR_V1_CODES: tuple[str, ...] = (*CULTNET_ERROR_CODES, "limit_reached")
+
+# The two frames a receipt-capable hello lists in supportedMessageVersions; a peer lists both or neither
+# (contracts/cultnet/interop/write-receipt.md, "Capability: opting in").
+CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS: tuple[str, ...] = ("cultnet.write_admitted.v0", "cultnet.error.v1")
+
 _ERROR_DETAILS_KEYS = ("field", "value", "asOf", "current")
 
 
@@ -54,16 +65,48 @@ def _is_uint(value: Any) -> bool:
     return isinstance(value, int) and not isinstance(value, bool) and value >= 0
 
 
-def _validate_error_body(body: dict[str, Any]) -> None:
-    """Refuse a cultnet.error.v0 body that is not the full-key shape the schema requires."""
-    expected = ("error", "routingHint", "code", "details")
+def _is_name(value: Any) -> bool:
+    return isinstance(value, str) and value != ""
+
+
+def _validate_write_admitted_body(body: dict[str, Any]) -> None:
+    """Refuse a cultnet.write_admitted.v0 body that is not exactly the five keys of its closed schema."""
+    expected = ("messageId", "operation", "schemaId", "recordKey")
     if set(body) != set(expected):
-        raise ValueError(f"cultnet.error.v0 must carry exactly {list(expected)}, got {sorted(body)}")
+        raise ValueError(f"cultnet.write_admitted.v0 must carry exactly {list(expected)}, got {sorted(body)}")
+    for key in ("messageId", "schemaId", "recordKey"):
+        if not _is_name(body[key]):
+            raise ValueError(f"cultnet.write_admitted.v0 {key} must be a non-empty string")
+    if body["operation"] not in ("put", "delete"):
+        raise ValueError("cultnet.write_admitted.v0 operation must be put or delete")
+
+
+def write_content_hash(operation: str, schema_id: str, record_key: str, payload: bytes | None) -> bytes:
+    """SHA-256 of the MessagePack array [operation, schemaId, recordKey, payload] (write-receipt.md, rule 5).
+
+    payload is the raw document payload, packed as a MessagePack bin for a put and nil for a delete.
+    """
+    return hashlib.sha256(msgpack.packb([operation, schema_id, record_key, payload], use_bin_type=True)).digest()
+
+
+def hello_advertises_write_receipts(hello_message: CultNetMessage) -> bool:
+    """True when a hello lists both receipt frames: the only thing that makes a connection receipt-capable."""
+    listed = hello_message.body.get("supportedMessageVersions")
+    return isinstance(listed, list) and all(version in listed for version in CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS)
+
+
+def _validate_error_body(body: dict[str, Any], version: str, codes: tuple[str, ...]) -> None:
+    """Refuse a cultnet.error body that is not the full-key shape its closed schema requires."""
+    expected = ("error", "routingHint", "code", "details") + (("messageId",) if version == "cultnet.error.v1" else ())
+    if set(body) != set(expected):
+        raise ValueError(f"{version} must carry exactly {list(expected)}, got {sorted(body)}")
     if not isinstance(body["error"], str) or not body["error"]:
-        raise ValueError("cultnet.error.v0 error must be a non-empty string")
+        raise ValueError(f"{version} error must be a non-empty string")
     code = body["code"]
-    if code is not None and (not isinstance(code, str) or code not in CULTNET_ERROR_CODES):
-        raise ValueError(f"cultnet.error.v0 code {code!r} is not a recognized refusal code")
+    if code is not None and (not isinstance(code, str) or code not in codes):
+        raise ValueError(f"{version} code {code!r} is not a recognized refusal code")
+    if "messageId" in body and body["messageId"] is not None and not _is_name(body["messageId"]):
+        raise ValueError(f"{version} messageId must be a non-empty string or null")
     hint = body["routingHint"]
     if hint is not None:
         if (
@@ -72,17 +115,17 @@ def _validate_error_body(body: dict[str, Any]) -> None:
             or not isinstance(hint["reason"], str)
             or not (hint["shard"] is None or isinstance(hint["shard"], dict))
         ):
-            raise ValueError("cultnet.error.v0 routingHint must be null or {shard, reason}")
+            raise ValueError(f"{version} routingHint must be null or {{shard, reason}}")
     details = body["details"]
     if details is not None:
         if not isinstance(details, dict) or set(details) != set(_ERROR_DETAILS_KEYS):
-            raise ValueError(f"cultnet.error.v0 details must be null or carry exactly {list(_ERROR_DETAILS_KEYS)}")
+            raise ValueError(f"{version} details must be null or carry exactly {list(_ERROR_DETAILS_KEYS)}")
         for key in ("field", "value"):
             if details[key] is not None and not isinstance(details[key], str):
-                raise ValueError(f"cultnet.error.v0 details.{key} must be a string or null")
+                raise ValueError(f"{version} details.{key} must be a string or null")
         for key in ("asOf", "current"):
             if details[key] is not None and not _is_uint(details[key]):
-                raise ValueError(f"cultnet.error.v0 details.{key} must be a non-negative integer or null")
+                raise ValueError(f"{version} details.{key} must be a non-negative integer or null")
 
 
 def hello(
@@ -92,17 +135,18 @@ def hello(
     supported_schema_versions: list[str] | None = None,
     supported_mutation_contracts: list[dict[str, Any]] | None = None,
     transport_profiles: list[dict[str, Any]] | None = None,
+    supported_message_versions: list[str] | None = None,
 ) -> CultNetMessage:
-    return CultNetMessage(
-        "cultnet.hello.v0",
-        {
-            "runtimeId": runtime_id,
-            "supportedWireContracts": supported_wire_contracts or ["cultnet.schema.v0"],
-            "supportedSchemaVersions": supported_schema_versions or [],
-            "supportedMutationContracts": supported_mutation_contracts or [],
-            "transportProfiles": transport_profiles or [],
-        },
-    )
+    body: dict[str, Any] = {
+        "runtimeId": runtime_id,
+        "supportedWireContracts": supported_wire_contracts or ["cultnet.schema.v0"],
+        "supportedSchemaVersions": supported_schema_versions or [],
+        "supportedMutationContracts": supported_mutation_contracts or [],
+        "transportProfiles": transport_profiles or [],
+    }
+    if supported_message_versions is not None:
+        body["supportedMessageVersions"] = supported_message_versions
+    return CultNetMessage("cultnet.hello.v0", body)
 
 
 def login(
@@ -172,7 +216,7 @@ def error_message(
     }
     if details is not None and set(details) - set(_ERROR_DETAILS_KEYS):
         raise ValueError(f"cultnet.error.v0 details has unknown keys {sorted(set(details) - set(_ERROR_DETAILS_KEYS))}")
-    _validate_error_body(body)
+    _validate_error_body(body, "cultnet.error.v0", CULTNET_ERROR_CODES)
     return CultNetMessage("cultnet.error.v0", body)
 
 

packages/cultnet-py/tests/test_write_receipt.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+from typing import Any
+
+import msgpack
+
+from cultnet_py import (
+    CULTNET_ERROR_CODES,
+    CULTNET_ERROR_V1_CODES,
+    CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
+    error_message,
+    hello,
+    hello_advertises_write_receipts,
+    parse_message,
+    write_content_hash,
+)
+
+# Decoder-side conformance with contracts/cultnet/interop/write-receipt-vectors.json: `contentHash`, `wire` and
+# `schemaInvalid`. `dedupe` and `blankFields` are server obligations and `scenarios` is the client write tracker; `counts` is the file's own tally;
+# this runtime has neither a receipt server nor a write client, so it replays none of them.
+_INTEROP = Path(__file__).resolve().parents[3] / "contracts/cultnet/interop"
+_VECTORS = json.loads((_INTEROP / "write-receipt-vectors.json").read_text(encoding="utf-8"))
+_ERROR_V1_SCHEMA = json.loads((_INTEROP / "cultnet.error.v1.schema.json").read_text(encoding="utf-8"))
+
+
+def _pack(wire: dict[str, Any]) -> bytes:
+    return msgpack.packb(wire, use_bin_type=True)
+
+
+def _admitted(**override: Any) -> dict[str, Any]:
+    return {
+        "schemaVersion": "cultnet.write_admitted.v0",
+        "messageId": "put-1",
+        "operation": "put",
+        "schemaId": "note.v0",
+        "recordKey": "note:1",
+        **override,
+    }
+
+
+def _refused(**override: Any) -> dict[str, Any]:
+    return {
+        "schemaVersion": "cultnet.error.v1",
+        "error": "write_not_authorized: no",
+        "routingHint": None,
+        "code": "write_not_authorized",
+        "details": None,
+        "messageId": "put-1",
+        **override,
+    }
+
+
+class WriteReceiptVectorTests(unittest.TestCase):
+    def test_vector_sections_are_the_ones_this_file_accounts_for(self) -> None:
+        self.assertEqual(
+            set(_VECTORS) - {"provenance", "specification"},
+            {"constants", "contentHash", "dedupe", "wire", "blankFields", "schemaInvalid", "scenarios", "counts"},
+        )
+        self.assertEqual(len(_VECTORS["contentHash"]["cases"]), 4)
+        self.assertEqual(len(_VECTORS["wire"]), 7)
+        self.assertEqual(len(_VECTORS["schemaInvalid"]), 7)
+        # The file declares how many of each it holds; a loader that drops entries disagrees with it.
+        counts = _VECTORS["counts"]
+        self.assertEqual(counts["wire"], len(_VECTORS["wire"]))
+        self.assertEqual(counts["schemaInvalid"], len(_VECTORS["schemaInvalid"]))
+        self.assertEqual(counts["contentHashCases"], len(_VECTORS["contentHash"]["cases"]))
+        self.assertEqual(counts["scenarioCases"], len(_VECTORS["scenarios"]["cases"]))
+
+    def test_content_hash_matches_every_vector(self) -> None:
+        for case in _VECTORS["contentHash"]["cases"]:
+            payload = None if case["payloadHex"] is None else bytes.fromhex(case["payloadHex"])
+            digest = write_content_hash(case["operation"], case["schemaId"], case["recordKey"], payload)
+            self.assertEqual(digest.hex(), case["sha256"], case["name"])
+
+    def test_wire_vectors_decode_to_their_fields_in_order_and_reencode_byte_identical(self) -> None:
+        for vector in _VECTORS["wire"]:
+            raw = bytes.fromhex(vector["messagePackHex"])
+            message = parse_message(raw)
+            self.assertEqual(message.to_wire(), vector["message"], vector["name"])
+            self.assertEqual(list(message.to_wire()), list(vector["message"]), vector["name"])
+            self.assertEqual(message.to_bytes(), raw, vector["name"])
+
+    def test_schema_invalid_vectors_are_refused(self) -> None:
+        for vector in _VECTORS["schemaInvalid"]:
+            with self.assertRaises(ValueError, msg=vector["name"]):
+                parse_message(bytes.fromhex(vector["messagePackHex"]))
+
+    def test_a_payload_that_is_not_bytes_changes_the_hash(self) -> None:
+        # The payload is a bin, not an str and not a re-parsed document.
+        as_bin = write_content_hash("put", "note.v0", "note:1", b"\x81\xa2id\xa11")
+        self.assertNotEqual(as_bin, write_content_hash("put", "note.v0", "note:1", None))
+        self.assertEqual(len(as_bin), 32)
+
+
+class ErrorV1Tests(unittest.TestCase):
+    def test_code_set_equals_the_schema_enum_and_v0_is_unchanged(self) -> None:
+        schema_codes = [c for c in _ERROR_V1_SCHEMA["properties"]["code"]["enum"] if c is not None]
+        self.assertEqual(sorted(CULTNET_ERROR_V1_CODES), sorted(schema_codes))
+        self.assertNotIn("limit_reached", CULTNET_ERROR_CODES)
+        self.assertEqual(CULTNET_ERROR_V1_CODES[-1], "limit_reached")
+
+    def test_every_v1_code_decodes_and_an_unknown_one_does_not(self) -> None:
+        for code in CULTNET_ERROR_V1_CODES:
+            self.assertEqual(parse_message(_pack(_refused(code=code))).body["code"], code)
+        parse_message(_pack(_refused(code=None)))
+        with self.assertRaises(ValueError):
+            parse_message(_pack(_refused(code="invented")))
+        with self.assertRaises(ValueError):
+            parse_message(_pack(_refused(code=7)))
+
+    def test_limit_reached_is_v1_only(self) -> None:
+        v0 = {k: v for k, v in _refused(code="limit_reached", schemaVersion="cultnet.error.v0").items() if k != "messageId"}
+        with self.assertRaises(ValueError):
+            parse_message(_pack(v0))
+        with self.assertRaises(ValueError):
+            error_message("x", code="limit_reached")
+
+    def test_message_id_is_required_and_is_a_nonempty_string_or_null(self) -> None:
+        self.assertEqual(parse_message(_pack(_refused(messageId="w-é 1"))).body["messageId"], "w-é 1")
+        self.assertIsNone(parse_message(_pack(_refused(messageId=None))).body["messageId"])
+        omitted = {k: v for k, v in _refused().items() if k != "messageId"}
+        with self.assertRaises(ValueError):
+            parse_message(_pack(omitted))
+        for bad in ("", 5, b"put-1", ["put-1"], True):
+            with self.assertRaises(ValueError, msg=repr(bad)):
+                parse_message(_pack(_refused(messageId=bad)))
+
+    def test_v0_stays_closed_to_message_id(self) -> None:
+        v0 = _refused(schemaVersion="cultnet.error.v0", code="message_invalid")
+        with self.assertRaises(ValueError):
+            parse_message(_pack(v0))
+        del v0["messageId"]
+        self.assertEqual(parse_message(_pack(v0)).schema_version, "cultnet.error.v0")
+
+    def test_v1_is_closed_to_unknown_keys_and_keeps_the_shape_rules_of_v0(self) -> None:
+        for bad in (
+            _refused(extra=1),
+            _refused(error=""),
+            _refused(error=3),
+            _refused(routingHint={"shard": None}),
+            _refused(routingHint={"shard": 1, "reason": "r"}),
+            _refused(routingHint={"shard": None, "reason": 1}),
+            _refused(details={"field": None, "value": None, "asOf": None}),
+            _refused(details={"field": 1, "value": None, "asOf": None, "current": None}),
+            _refused(details={"field": None, "value": None, "asOf": -1, "current": None}),
+            _refused(details={"field": None, "value": None, "asOf": None, "current": True}),
+        ):
+            with self.assertRaises(ValueError, msg=str(bad)):
+                parse_message(_pack(bad))
+        for good in (
+            _refused(routingHint={"shard": None, "reason": "r"}),
+            _refused(details={"field": "messageId", "value": "x", "asOf": 1, "current": 2}),
+        ):
+            parse_message(_pack(good))
+
+    def test_the_refusal_reason_names_its_version(self) -> None:
+        with self.assertRaisesRegex(ValueError, "cultnet.error.v1 code"):
+            parse_message(_pack(_refused(code="invented")))
+        with self.assertRaisesRegex(ValueError, r"routingHint must be null or \{shard, reason\}"):
+            parse_message(_pack(_refused(routingHint={"shard": None})))
+
+
+class RefusalTextTests(unittest.TestCase):
+    """A refusal names its frame version and the rule it broke; a caller reads this text in a log."""
+
+    def assertRefused(self, wire: dict[str, Any], pattern: str) -> None:
+        with self.assertRaisesRegex(ValueError, pattern):
+            parse_message(_pack(wire))
+
+    def test_error_frames_name_their_own_version(self) -> None:
+        v0 = {k: v for k, v in _refused(schemaVersion="cultnet.error.v0").items() if k != "messageId"}
+        self.assertRefused({**v0, "code": "nope"}, r"^cultnet\.error\.v0 code")
+        self.assertRefused({**v0, "extra": 1}, r"^cultnet\.error\.v0 must carry exactly")
+        self.assertRefused(_refused(extra=1), r"^cultnet\.error\.v1 must carry exactly")
+        self.assertRefused(_refused(error=""), r"^cultnet\.error\.v1 error must be")
+        self.assertRefused(_refused(messageId=""), r"^cultnet\.error\.v1 messageId must be")
+        self.assertRefused(_refused(details={"field": None}), r"^cultnet\.error\.v1 details must be null or carry exactly")
+        self.assertRefused(
+            _refused(details={"field": 1, "value": None, "asOf": None, "current": None}), r"^cultnet\.error\.v1 details\.field must be"
+        )
+        self.assertRefused(
+            _refused(details={"field": None, "value": None, "asOf": -1, "current": None}), r"^cultnet\.error\.v1 details\.asOf must be"
+        )
+        with self.assertRaisesRegex(ValueError, r"^cultnet\.error\.v0 code"):
+            error_message("x", code="bogus")
+
+    def test_write_admitted_names_the_broken_rule(self) -> None:
+        self.assertRefused(_admitted(extra=1), r"^cultnet\.write_admitted\.v0 must carry exactly")
+        self.assertRefused(_admitted(recordKey=""), r"^cultnet\.write_admitted\.v0 recordKey must be a non-empty string")
+        self.assertRefused(_admitted(operation="upsert"), r"^cultnet\.write_admitted\.v0 operation must be put or delete$")
+
+
+class WriteAdmittedTests(unittest.TestCase):
+    def test_both_operations_decode(self) -> None:
+        for operation in ("put", "delete"):
+            self.assertEqual(parse_message(_pack(_admitted(operation=operation))).body["operation"], operation)
+
+    def test_every_key_is_required_and_no_other_is_allowed(self) -> None:
+        for key in ("messageId", "operation", "schemaId", "recordKey"):
+            with self.assertRaises(ValueError, msg=key):
+                parse_message(_pack({k: v for k, v in _admitted().items() if k != key}))
+        for extra in ("error", "code", "details"):
+            with self.assertRaises(ValueError, msg=extra):
+                parse_message(_pack(_admitted(**{extra: "x"})))
+
+    def test_each_string_is_nonempty_and_a_string(self) -> None:
+        for key in ("messageId", "schemaId", "recordKey"):
+            for bad in ("", None, 3, b"x"):
+                with self.assertRaises(ValueError, msg=f"{key}={bad!r}"):
+                    parse_message(_pack(_admitted(**{key: bad})))
+
+    def test_operation_is_put_or_delete(self) -> None:
+        for bad in ("upsert", "PUT", "", None, 1):
+            with self.assertRaises(ValueError, msg=repr(bad)):
+                parse_message(_pack(_admitted(operation=bad)))
+
+    def test_message_id_is_echoed_byte_for_byte(self) -> None:
+        self.assertEqual(parse_message(_pack(_admitted(messageId="  w é 1 "))).body["messageId"], "  w é 1 ")
+
+
+class HelloCapabilityTests(unittest.TestCase):
+    def test_capability_needs_both_frames(self) -> None:
+        both = list(CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS)
+        self.assertEqual(both, ["cultnet.write_admitted.v0", "cultnet.error.v1"])
+        self.assertTrue(hello_advertises_write_receipts(hello(runtime_id="r", supported_message_versions=both)))
+        self.assertTrue(hello_advertises_write_receipts(hello(runtime_id="r", supported_message_versions=[*both, "x"])))
+        self.assertFalse(hello_advertises_write_receipts(hello(runtime_id="r", supported_message_versions=both[:1])))
+        self.assertFalse(hello_advertises_write_receipts(hello(runtime_id="r", supported_message_versions=both[1:])))
+        self.assertFalse(hello_advertises_write_receipts(hello(runtime_id="r", supported_message_versions=[])))
+        self.assertFalse(hello_advertises_write_receipts(hello(runtime_id="r")))
+
+    def test_a_non_list_listing_is_not_a_capability(self) -> None:
+        wire = hello(runtime_id="r").to_wire()
+        wire["supportedMessageVersions"] = "cultnet.write_admitted.v0 cultnet.error.v1"
+        self.assertFalse(hello_advertises_write_receipts(parse_message(_pack(wire))))
+
+    def test_hello_writes_the_listing_only_when_given_and_round_trips(self) -> None:
+        self.assertNotIn("supportedMessageVersions", hello(runtime_id="r").body)
+        listed = hello(runtime_id="r", supported_message_versions=list(CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS))
+        self.assertEqual(parse_message(listed.to_bytes()), listed)
+
+
+if __name__ == "__main__":
+    unittest.main()