packages/cultmesh-py/README.md

@@ -122,6 +122,33 @@ server = CultMesh.serve_node(
 )
 ```
 
+## Peer writes
+
+**Breaking (0.x):** a served node refuses every peer `document_put_raw` and `document_delete` unless a write
+authority grants it, mirroring C# `ICultNetRemoteWriteAuthority`. Before this, any connected peer could overwrite
+or delete the node's documents. The refusal is a typed `cultnet.error.v0` with code `write_not_authorized` and the
+record key in `details`; an authority that raises, or returns no decision, refuses too.
+
+Known residual: ungranted peers can still distinguish registered from unregistered schemas by timing (the node resolves the
+schema before the gate so the authority sees the canonical id); the refusal itself is identical.
+
+```python
+from cultmesh_py import CultMeshGrantAllWrites, CultMeshRemoteWriteDecision, CultMeshRemoteWriteRequest
+
+class OwnKeysOnly:
+    def decide(self, request: CultMeshRemoteWriteRequest) -> CultMeshRemoteWriteDecision:
+        if request.record_key.startswith("input:python:"):
+            return CultMeshRemoteWriteDecision.grant()
+        return CultMeshRemoteWriteDecision.refuse("not your key")
+
+server = CultMesh.serve_node(node, write_authority=OwnKeysOnly())
+```
+
+The request carries the peer (`endpoint`, `transport`), the operation, the canonical schema id the node
+resolved (never the alias the peer sent), the record key and the shard id and epoch as sent. `session.serve(...)`
+takes the same `write_authority`. A host that already authenticates its peers, or a test rig, opts in to
+`CultMeshGrantAllWrites()`; the daemon's equivalent is `--allow-peer-writes`. Without it the daemon is read-only to peers.
+
 The same local server can be launched as a package daemon when another runtime
 or operator process needs a CultNet endpoint without embedding Python glue:
 

packages/cultmesh-py/src/cultmesh_py/__init__.py

@@ -39,6 +39,14 @@ from .simulation import (
     CultMeshSimulationFactCommitter,
     simulation_fact_document,
 )
+from .write_authority import (
+    CultMeshGrantAllWrites,
+    CultMeshRemoteWriteAuthority,
+    CultMeshRemoteWriteDecision,
+    CultMeshRemoteWriteOperation,
+    CultMeshRemoteWriteRequest,
+    CultMeshServerPeer,
+)
 from cultnet_py.cultmesh_contracts import (
     PEER_EXCHANGE_REQUEST,
     PEER_EXCHANGE_RESPONSE,
@@ -89,6 +97,12 @@ __all__ = [
     "CultMeshSimulationFact",
     "CultMeshSimulationFactCommit",
     "CultMeshSimulationFactCommitter",
+    "CultMeshGrantAllWrites",
+    "CultMeshRemoteWriteAuthority",
+    "CultMeshRemoteWriteDecision",
+    "CultMeshRemoteWriteOperation",
+    "CultMeshRemoteWriteRequest",
+    "CultMeshServerPeer",
     "CultMeshAuthorityLease",
     "CultMeshAuthorityLeaseCatalog",
     "CultMeshHmacAuthorityLeaseVerifier",

packages/cultmesh-py/src/cultmesh_py/daemon.py

@@ -13,6 +13,7 @@ from cultnet_py import CultNetSimulationConsensusOptions, CultNetSimulationObser
 
 from .facade import CultMesh
 from .server import CultMeshLocalServer
+from .write_authority import CultMeshGrantAllWrites
 from cultnet_py.cultmesh_contracts import (
     CultMeshPeerCard,
     CultMeshPeerCatalog,
@@ -53,6 +54,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--simulation-quorum-ratio", type=float, default=0.5)
     parser.add_argument("--verse-id")
     parser.add_argument("--verse-display-name")
+    parser.add_argument("--allow-peer-writes", action="store_true", help="grant every connected peer put and delete; without it peer writes are refused")
     parser.add_argument("--role", action="append", dest="roles")
     args = parser.parse_args(argv)
 
@@ -117,6 +119,7 @@ def start_server(args: argparse.Namespace, parser: argparse.ArgumentParser) -> C
             display_name=args.display_name,
             max_snapshot_documents=args.max_snapshot_documents,
             max_snapshot_bytes=args.max_snapshot_bytes,
+            write_authority=CultMeshGrantAllWrites() if args.allow_peer_writes else None,
         )
     except ValueError as error:
         # The server owns its option rules; a limit it refuses is a usage error, not a crash.

packages/cultmesh-py/src/cultmesh_py/facade.py

@@ -35,6 +35,7 @@ from .node import (
 from .server import CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES, CultMeshLocalServer
 from .session import CultMeshGameSession, CultMeshGameSessionOptions
 from .simulation import CultMeshSimulationFactCommitter
+from .write_authority import CultMeshRemoteWriteAuthority
 from cultnet_py.cultmesh_contracts import (
     AuthorityLeaseVerifier,
     CultMeshAuthorityLeaseCatalog,
@@ -400,6 +401,7 @@ class CultMesh:
         verse_catalog: CultMeshVerseCatalog | None = None,
         peer_catalog: CultMeshPeerCatalog | None = None,
         observation_hub: CultNetSimulationObservationHub | None = None,
+        write_authority: CultMeshRemoteWriteAuthority | None = None,
         host: str = "127.0.0.1",
         port: int = 0,
         display_name: str | None = None,
@@ -416,6 +418,7 @@ class CultMesh:
             verse_catalog=verse_catalog or CultMeshVerseCatalog(),
             peer_catalog=peer_catalog or CultMeshPeerCatalog(),
             observation_hub=observation_hub,
+            write_authority=write_authority,
             host=host,
             port=port,
             display_name=display_name,

packages/cultmesh-py/src/cultmesh_py/node.py

@@ -650,6 +650,29 @@ class CultMeshDatabase:
         self._append_shard_log_delete(message)
         return message
 
+    def raw_put_target(self, message: dict[str, Any]) -> tuple[str, str] | None:
+        """The canonical schema id and record key a raw put would write, or None when it would not apply."""
+        document_record = message.get("document")
+        # Only a document of the wire shape is resolved, so an ungranted peer's payload is never
+        # converted or allocated here: a malformed one is the store's refusal, after the authority.
+        if (
+            not isinstance(document_record, dict)
+            or not isinstance(document_record.get("recordKey"), str)
+            or not isinstance(document_record.get("payload"), (bytes, bytearray, memoryview))
+        ):
+            return None
+        try:
+            prepared = self._raw_put_envelope(message)
+        except Exception:  # noqa: BLE001 - any other malformed shape
+            return None
+        return None if prepared is None else (prepared[0].catalog_entry().schema_id, prepared[1].key)
+
+    def raw_delete_target(self, message: dict[str, Any]) -> tuple[str, str] | None:
+        """The canonical schema id and record key a raw delete names, or None when its schema is not registered."""
+        document = self._document_for_schema(str(message.get("schemaId") or ""))
+        record_key = str(message.get("recordKey") or "")
+        return None if document is None or not record_key else (document.catalog_entry().schema_id, record_key)
+
     def apply_raw_put_message(self, message: dict[str, Any]) -> CultMeshDatabaseChange | None:
         prepared = self._raw_put_envelope(message)
         if prepared is None:

packages/cultmesh-py/src/cultmesh_py/server.py

@@ -10,6 +10,7 @@ from typing import Any
 import msgpack
 
 from cultnet_py import (
+    error_message,
     unsupported_schema_version_error,
     CultNetRawSnapshotResponse,
     CultNetSimulationObservationHub,
@@ -28,6 +29,12 @@ from cultnet_py import (
 )
 
 from .node import CultMeshNode
+from .write_authority import (
+    CultMeshRemoteWriteAuthority,
+    CultMeshRemoteWriteOperation,
+    CultMeshRemoteWriteRequest,
+    CultMeshServerPeer,
+)
 
 # The shortest message id this server answers a snapshot request under. It takes a
 # request's messageId as given, an empty one included, so no response it sends is smaller.
@@ -88,6 +95,9 @@ class CultMeshLocalServer:
     verse_catalog: CultMeshVerseCatalog = field(default_factory=CultMeshVerseCatalog)
     peer_catalog: CultMeshPeerCatalog = field(default_factory=CultMeshPeerCatalog)
     observation_hub: CultNetSimulationObservationHub | None = None
+    # Asked once per peer put and delete. None refuses every one of them (write_not_authorized); a node's own
+    # writes never come through here.
+    write_authority: CultMeshRemoteWriteAuthority | None = None
     host: str = "127.0.0.1"
     port: int = 0
     display_name: str | None = None
@@ -238,14 +248,15 @@ class CultMeshLocalServer:
             if server is None:
                 return
             try:
-                client, _ = server.accept()
+                client, address = server.accept()
             except TimeoutError:
                 continue
             except OSError:
                 return
-            threading.Thread(target=self._handle_connection, args=(client,), daemon=True).start()
+            peer = CultMeshServerPeer(f"{address[0]}:{address[1]}", "tcp")
+            threading.Thread(target=self._handle_connection, args=(client, peer), daemon=True).start()
 
-    def _handle_connection(self, client: socket.socket) -> None:
+    def _handle_connection(self, client: socket.socket, peer: CultMeshServerPeer) -> None:
         with client:
             stream = client.makefile("rwb")
             transport = TcpFramedTransportConnection(
@@ -267,7 +278,7 @@ class CultMeshLocalServer:
                 if not isinstance(message, dict):
                     continue
                 try:
-                    responses = self._handle_connection_message(message, subscriptions)
+                    responses = self._handle_connection_message(message, subscriptions, peer=peer)
                 except Exception as error:  # noqa: BLE001 - one message never ends the connection
                     # Whatever a peer's message makes the handling raise is that peer's failure:
                     # it is refused with fixed text and the connection serves on. The log names
@@ -363,7 +374,12 @@ class CultMeshLocalServer:
                         continue
                     if not isinstance(message, dict):
                         continue
-                    responses = self._handle_connection_message(message, peer.subscriptions, over_rudp=True)
+                    responses = self._handle_connection_message(
+                        message,
+                        peer.subscriptions,
+                        over_rudp=True,
+                        peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"),
+                    )
                     for response in responses:
                         if unsendable is None:
                             unsendable = self._send_rudp_schema_frame(
@@ -519,6 +535,7 @@ class CultMeshLocalServer:
         subscriptions: dict[str, _DatabaseSubscription],
         *,
         over_rudp: bool = False,
+        peer: CultMeshServerPeer | None = None,
     ) -> list[dict[str, Any]]:
         schema_version = message.get("schemaVersion")
         if schema_version == "cultnet.database_subscribe.v0":
@@ -542,9 +559,9 @@ class CultMeshLocalServer:
             subscriptions.pop(str(message.get("subscriptionId") or ""), None)
             return []
         if schema_version == "cultnet.document_put_raw.v0":
-            return self._handle_raw_put(message, subscriptions, over_rudp=over_rudp)
+            return self._handle_raw_put(message, subscriptions, over_rudp=over_rudp, peer=peer)
         if schema_version == "cultnet.document_delete.v0":
-            return self._handle_raw_delete(message, subscriptions)
+            return self._handle_raw_delete(message, subscriptions, peer=peer)
         if schema_version == "cultnet.simulation_observation.v0":
             return self._candidate_responses(message)
         response = self.handle_message(message)
@@ -556,6 +573,7 @@ class CultMeshLocalServer:
         subscriptions: dict[str, _DatabaseSubscription],
         *,
         over_rudp: bool = False,
+        peer: CultMeshServerPeer | None = None,
     ) -> list[dict[str, Any]]:
         document_record = message.get("document")
         if not isinstance(document_record, dict):
@@ -564,6 +582,19 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="malformed_document_put",
             )]
+        # Refusal precedes every other judgment of the put. The node resolves the schema first, because the
+        # authority must be asked about the canonical schema id (a payload-inferred alias included), as C#
+        # does; only a payload of the wire shape is resolved, so nothing of an ungranted peer's is
+        # allocated. A put the node cannot resolve is asked about under the ids as sent. Residual: the
+        # refusal is the same either way, but an ungranted peer can still tell registered from unregistered
+        # schemas by timing.
+        target = self.node.database.raw_put_target(message) or (
+            str(document_record.get("schemaId") or ""),
+            str(document_record.get("recordKey") or ""),
+        )
+        refusal = self._refuse_unauthorized_write(peer, CultMeshRemoteWriteOperation.PUT, target, message)
+        if refusal is not None:
+            return [refusal]
         unservable = self._unservable_put(message, over_rudp=over_rudp)
         if unservable is not None:
             return [unservable]
@@ -598,6 +629,8 @@ class CultMeshLocalServer:
         self,
         message: dict[str, Any],
         subscriptions: dict[str, _DatabaseSubscription],
+        *,
+        peer: CultMeshServerPeer | None = None,
     ) -> list[dict[str, Any]]:
         schema_id = str(message.get("schemaId") or "")
         record_key = str(message.get("recordKey") or "")
@@ -607,6 +640,10 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="malformed_document_delete",
             )]
+        target = self.node.database.raw_delete_target(message) or (schema_id, record_key)
+        refusal = self._refuse_unauthorized_write(peer, CultMeshRemoteWriteOperation.DELETE, target, message)
+        if refusal is not None:
+            return [refusal]
         change = self.node.database.apply_raw_delete_message(message)
         if change is None:
             return [self._error_response(
@@ -616,6 +653,50 @@ class CultMeshLocalServer:
             )]
         return self._database_delete_notifications(message, subscriptions, resolved_schema_id=change.schema_id)
 
+    def _refuse_unauthorized_write(
+        self,
+        peer: CultMeshServerPeer | None,
+        operation: CultMeshRemoteWriteOperation,
+        target: tuple[str, str],
+        message: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """None when the write may proceed, else the typed refusal for the sending peer.
+
+        No authority configured refuses; an authority that raises refuses too (fail closed) and is logged.
+        """
+        schema_id, record_key = target
+        reason: str | None
+        authority = self.write_authority
+        if authority is None:
+            reason = "no remote write authority is configured"
+        else:
+            request = CultMeshRemoteWriteRequest(
+                peer=peer or CultMeshServerPeer("", "unknown"),
+                operation=operation,
+                schema_id=schema_id,
+                record_key=record_key,
+                shard_id=message.get("shardId") if isinstance(message.get("shardId"), str) else None,
+                shard_epoch=message.get("shardEpoch")
+                if isinstance(message.get("shardEpoch"), int) and not isinstance(message.get("shardEpoch"), bool)
+                else None,
+            )
+            try:
+                decision = authority.decide(request)
+            except Exception as error:  # noqa: BLE001 - a failing authority refuses
+                _LOGGER.error("CultMesh remote write authority failed (%s); the write was refused.", type(error).__name__)
+                decision = None
+                reason = "the write authority failed"
+            else:
+                reason = getattr(decision, "reason", None)
+            if decision is not None and getattr(decision, "is_granted", False) is True:
+                return None
+        _LOGGER.warning("CultMesh %s of %s refused: write not authorized.", operation.value, schema_id)
+        return error_message(
+            f"write_not_authorized: {reason}" if reason and reason.strip() else "write_not_authorized",
+            code="write_not_authorized",
+            details={"field": "recordKey", "value": record_key},
+        ).to_wire()
+
     def _database_change_notifications(
         self,
         message: dict[str, Any],

packages/cultmesh-py/src/cultmesh_py/session.py

@@ -27,6 +27,7 @@ from cultnet_py.cultmesh_contracts import CultMeshAuthorityLeaseCatalog, CultMes
 
 if TYPE_CHECKING:
     from .server import CultMeshLocalServer
+    from .write_authority import CultMeshRemoteWriteAuthority
 
 
 @dataclass(frozen=True)
@@ -83,6 +84,7 @@ class CultMeshGameSession:
     def serve(
         self,
         *,
+        write_authority: CultMeshRemoteWriteAuthority | None = None,
         host: str = "127.0.0.1",
         port: int = 0,
         display_name: str | None = None,
@@ -96,6 +98,7 @@ class CultMeshGameSession:
             verse_catalog=self.verse_catalog if self._serve_verse_discovery else CultMeshVerseCatalog(),
             peer_catalog=self.peer_catalog if self._serve_peer_exchange else CultMeshPeerCatalog(),
             observation_hub=self.observation_hub if self._serve_simulation_observations else None,
+            write_authority=write_authority,
             host=host,
             port=port,
             display_name=display_name,

packages/cultmesh-py/src/cultmesh_py/write_authority.py

@@ -0,0 +1,66 @@
+"""The one question a CultMesh server asks before it applies a peer's write.
+
+Mirrors C# ``ICultNetRemoteWriteAuthority``: a server with no authority refuses every peer put and
+delete, so serving a node never lets a connected peer overwrite its documents by default.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Protocol
+
+
+class CultMeshRemoteWriteOperation(Enum):
+    PUT = "put"
+    DELETE = "delete"
+
+
+@dataclass(frozen=True)
+class CultMeshServerPeer:
+    """The connection a write arrived on: its endpoint and transport, nothing the peer claimed."""
+
+    endpoint: str
+    transport: str
+
+
+@dataclass(frozen=True)
+class CultMeshRemoteWriteRequest:
+    """One peer write as the authority sees it.
+
+    ``schema_id`` is the canonical id the node resolved for the write, never an alias the peer sent.
+    ``shard_id`` and ``shard_epoch`` are as the peer sent them.
+    """
+
+    peer: CultMeshServerPeer
+    operation: CultMeshRemoteWriteOperation
+    schema_id: str
+    record_key: str
+    shard_id: str | None = None
+    shard_epoch: int | None = None
+
+
+@dataclass(frozen=True)
+class CultMeshRemoteWriteDecision:
+    is_granted: bool
+    reason: str | None = None
+
+    @staticmethod
+    def grant() -> CultMeshRemoteWriteDecision:
+        return CultMeshRemoteWriteDecision(True)
+
+    @staticmethod
+    def refuse(reason: str | None = None) -> CultMeshRemoteWriteDecision:
+        return CultMeshRemoteWriteDecision(False, reason)
+
+
+class CultMeshRemoteWriteAuthority(Protocol):
+    def decide(self, request: CultMeshRemoteWriteRequest) -> CultMeshRemoteWriteDecision: ...
+
+
+class CultMeshGrantAllWrites:
+    """Grants every peer write. For a host that has already authenticated its peers (or a test rig); it is
+    an explicit choice because serving without an authority refuses writes."""
+
+    def decide(self, request: CultMeshRemoteWriteRequest) -> CultMeshRemoteWriteDecision:
+        return CultMeshRemoteWriteDecision.grant()

packages/cultmesh-py/tests/test_cultmesh.py

@@ -68,7 +68,7 @@ from cultnet_py import (
     snapshot_request,
 )
 from cultnet_py.cultmesh_contracts import VERSE_CATALOG_REQUEST
-from cultmesh_py import create_node
+from cultmesh_py import CultMeshGrantAllWrites, create_node
 from cultmesh_py import (
     CultMesh,
     CultMeshDatabase,
@@ -1631,7 +1631,7 @@ class CultMeshTests(unittest.TestCase):
         )
         primary = CultMesh.create_node(runtime_id="forward-primary")
         primary.database.register_document(document)
-        server = CultMesh.serve_node(primary)
+        server = CultMesh.serve_node(primary, write_authority=CultMeshGrantAllWrites())
         try:
             shard = CultNetShardDescriptor(
                 shard_id="forward",
@@ -3015,7 +3015,7 @@ class CultMeshTests(unittest.TestCase):
         source.database.put_raw_message(document, "note:live", CanonicalNote("initial"))
         target = CultMesh.create_node(runtime_id="subscription-alias-target")
         target.database.register_document(document)
-        server = CultMesh.serve_node(source)
+        server = CultMesh.serve_node(source, write_authority=CultMeshGrantAllWrites())
         try:
             raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
             with CultMesh.subscribe_document(target, raw_client, alias, "note:live") as subscription:
@@ -3533,6 +3533,7 @@ class CultMeshTests(unittest.TestCase):
                     sys.executable,
                     "-m",
                     "cultmesh_py.daemon",
+                    "--allow-peer-writes",
                     "--runtime-id",
                     "live-daemon-peer",
                     "--port",
@@ -3723,6 +3724,7 @@ class CultMeshTests(unittest.TestCase):
                     sys.executable,
                     "-m",
                     "cultmesh_py.daemon",
+                    "--allow-peer-writes",
                     "--runtime-id",
                     "forwarder-daemon-peer",
                     "--port",
@@ -3998,6 +4000,7 @@ class CultMeshTests(unittest.TestCase):
         server = CultMesh.serve_node(
             node,
             verse_catalog=verses,
+            write_authority=CultMeshGrantAllWrites(),
             peer_catalog=peers,
             observation_hub=observation_hub,
             display_name="Mesh Server",
@@ -4274,7 +4277,7 @@ class CultMeshTests(unittest.TestCase):
         node = CultMesh.create_node(runtime_id="mesh-alias-sub-server")
         node.database.register_document(document)
 
-        server = CultMesh.serve_node(node)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
         try:
             raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
             with raw_client.subscribe_database(
@@ -4323,7 +4326,7 @@ class CultMeshTests(unittest.TestCase):
             shard_epoch=1,
         )
 
-        server = CultMesh.serve_node(node)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
         try:
             raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
             with raw_client.subscribe_database(
@@ -4420,7 +4423,7 @@ class CultMeshTests(unittest.TestCase):
         def served_alone_bytes(key: str, body_length: int) -> int:
             node = CultMesh.create_node(runtime_id="mesh-served-bound-probe")
             node.database.register_document(document)
-            server = CultMesh.serve_node(node, enable_rudp=False)
+            server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites(), enable_rudp=False)
             try:
                 response, payload = exchange(server, put("probe", key, body_length), empty_id_snapshot)
             finally:
@@ -4434,7 +4437,7 @@ class CultMeshTests(unittest.TestCase):
 
         node = CultMesh.create_node(runtime_id="mesh-served-bound")
         node.database.register_document(document)
-        server = CultMesh.serve_node(node, max_snapshot_bytes=limit, enable_rudp=False)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites(), max_snapshot_bytes=limit, enable_rudp=False)
         try:
             refused, _ = exchange(server, put("put-over", "note:over", 101))
             stored_after_refusal = node.database.get(document, "note:over")
@@ -4498,7 +4501,7 @@ class CultMeshTests(unittest.TestCase):
         document = self._rudp_bound_note()
         node = CultMesh.create_node(runtime_id="mesh-rudp-large")
         node.database.register_document(document)
-        server = CultMesh.serve_node(node)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
         try:
             served, payload = self._rudp_exchange(
                 server,
@@ -4524,7 +4527,7 @@ class CultMeshTests(unittest.TestCase):
         def served_alone_bytes(body_length: int) -> int:
             node = CultMesh.create_node(runtime_id="mesh-rudp-bound-probe")
             node.database.register_document(document)
-            server = CultMesh.serve_node(node)
+            server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
             try:
                 _, payload = self._rudp_exchange(
                     server, self._rudp_note_put(document, "probe", key, body_length),
@@ -4554,7 +4557,7 @@ class CultMeshTests(unittest.TestCase):
 
         node = CultMesh.create_node(runtime_id="mesh-rudp-bound")
         node.database.register_document(document)
-        server = CultMesh.serve_node(node, rudp_max_fragment_bytes=100, rudp_max_pending_reliable_packets=8)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites(), rudp_max_fragment_bytes=100, rudp_max_pending_reliable_packets=8)
         try:
             refusals = [
                 self._rudp_exchange(server, self._rudp_note_put(document, f"put-{served}", "note:q", body_for(served)))[0]
@@ -4792,7 +4795,7 @@ class CultMeshTests(unittest.TestCase):
         body_for = self._rudp_body_for(document, "note:w")
         node = CultMesh.create_node(runtime_id="mesh-rudp-wide")
         node.database.register_document(document)
-        server = CultMesh.serve_node(node, rudp_max_fragment_bytes=1)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites(), rudp_max_fragment_bytes=1)
         try:
             transport = create_rudp_schema_transport(
                 host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id,
@@ -4857,7 +4860,7 @@ class CultMeshTests(unittest.TestCase):
         document = self._rudp_bound_note()
         node = CultMesh.create_node(runtime_id="mesh-rudp-malformed")
         node.database.register_document(document)
-        server = CultMesh.serve_node(node)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
         try:
             refusals = []
             for field_name, value in (
@@ -4917,7 +4920,7 @@ class CultMeshTests(unittest.TestCase):
         node = CultMesh.create_node(runtime_id="mesh-unstorable-put")
         node.database.register_document(document)
         node.database.register_document(global_document)
-        server = CultMesh.serve_node(node)
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
         try:
             rudp_refusals = [(put["messageId"], self._rudp_exchange(server, put)[0]) for put in cases()]
             rudp_after, _ = self._rudp_exchange(server, hello(runtime_id="rudp-after").to_wire())

packages/cultmesh-py/tests/test_remote_write_authority.py

@@ -0,0 +1,325 @@
+"""A peer's put or delete is applied only when the server's write authority grants it.
+
+Ports the behaviours of C# RemoteWriteAuthorityTests. The C# forwarding behaviour (a refused write is never
+forwarded) has no counterpart: this server does not forward non-primary writes. The handlers are reached with
+the peer the connection loop would pass, and one test drives the same path over a real TCP connection.
+"""
+
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+from typing import Any
+
+import msgpack
+from cultcache_py import define_database_entry_type
+from cultmesh_py import (
+    CultMesh,
+    CultMeshGrantAllWrites,
+    CultMeshRemoteWriteDecision,
+    CultMeshRemoteWriteOperation,
+    CultMeshRemoteWriteRequest,
+    CultMeshServerPeer,
+    create_node,
+)
+from cultmesh_py.server import CultMeshLocalServer
+from cultnet_py import CultNetPeerError, create_rudp_schema_transport, document_delete, document_put_raw, error_message
+
+SCHEMA_ID = "sha256:authority-note"
+ALIAS = "runtime.generated.authority-note.ui.42"
+SHARD_ID = "authority-shard"
+KEY_A = "tests:authority:a"
+KEY_B = "tests:authority:b"
+PEER = CultMeshServerPeer("127.0.0.1:50000", "tcp")
+
+_REPO = Path(__file__).resolve().parents[3]
+_VECTORS = json.loads((_REPO / "contracts/cultnet/interop/error-vectors.json").read_text(encoding="utf-8"))["vectors"]
+
+NOTE = define_database_entry_type(
+    "tests.authority_note",
+    [("schema_version", 0), ("text", 1)],
+    schema_id=SCHEMA_ID,
+    schema_name="tests.authority_note",
+    schema_version="tests.authority_note.v1",
+)
+
+
+class Recording:
+    def __init__(self, decide: Any) -> None:
+        self._decide = decide
+        self.requests: list[CultMeshRemoteWriteRequest] = []
+
+    def decide(self, request: CultMeshRemoteWriteRequest) -> Any:
+        self.requests.append(request)
+        return self._decide(request)
+
+
+def put_of(key: str, text: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]:
+    return document_put_raw(
+        message_id="wire",
+        key=key,
+        schema_id=schema_id,
+        stored_at="2026-11-07T00:00:00Z",
+        payload=NOTE.encode_payload({"schema_version": "tests.authority_note.v1", "text": text}),
+        shard_id=SHARD_ID,
+        shard_epoch=1,
+    ).to_wire()
+
+
+def delete_of(key: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]:
+    return document_delete(
+        message_id="wire-delete", schema_id=schema_id, record_key=key, shard_id=SHARD_ID, shard_epoch=1
+    ).to_wire()
+
+
+class RemoteWriteAuthorityTests(unittest.TestCase):
+    def build(self, authority: Any = None) -> CultMeshLocalServer:
+        node = create_node(runtime_id="authority-server")
+        node.database.register_document(NOTE)
+        return CultMeshLocalServer(node=node, port=0, write_authority=authority)
+
+    def text_of(self, server: CultMeshLocalServer, key: str) -> str | None:
+        value = server.node.cache.get(NOTE, key)
+        return None if value is None else value["text"]
+
+    def send(self, server: CultMeshLocalServer, message: dict[str, Any]) -> list[dict[str, Any]]:
+        subscriptions: dict[str, Any] = {}
+        return server._handle_connection_message(message, subscriptions, peer=PEER)
+
+    def assert_refused(self, replies: list[dict[str, Any]], key: str, *, error: str | None = None) -> None:
+        self.assertEqual(len(replies), 1)
+        reply = replies[0]
+        self.assertEqual(reply["schemaVersion"], "cultnet.error.v0")
+        self.assertEqual(reply["code"], "write_not_authorized")
+        self.assertTrue(reply["error"].startswith("write_not_authorized"))
+        self.assertEqual(reply["details"]["field"], "recordKey")
+        self.assertEqual(reply["details"]["value"], key)
+        if error is not None:
+            self.assertEqual(reply["error"], error)
+
+    def seed(self, server: CultMeshLocalServer, key: str, text: str) -> None:
+        # The host writes its own record through its own door, which asks no remote authority.
+        self.assertEqual(server.node.database.apply_raw_put_message(put_of(key, text)).change_kind, "added")
+
+    def test_with_no_authority_a_peers_put_and_delete_are_refused_and_change_nothing(self) -> None:
+        server = self.build(None)
+        self.seed(server, KEY_A, "server")
+
+        with self.assertLogs("cultmesh_py.server", level="WARNING") as logs:
+            self.assert_refused(self.send(server, put_of(KEY_A, "peer")), KEY_A)
+            self.assert_refused(self.send(server, delete_of(KEY_A)), KEY_A)
+            self.assert_refused(self.send(server, put_of(KEY_B, "peer")), KEY_B)
+
+        self.assertEqual(self.text_of(server, KEY_A), "server")
+        self.assertIsNone(self.text_of(server, KEY_B))
+        self.assertEqual(sum("write not authorized" in line for line in logs.output), 3)
+
+    def test_a_refusal_publishes_no_change_to_a_subscriber(self) -> None:
+        server = self.build(None)
+        server.start()
+        try:
+            client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
+            with client.subscribe_database(subscription_id="s", schema_ids=[SCHEMA_ID], include_snapshot=False) as sub:
+                with self.assertRaises(CultNetPeerError) as raised:
+                    sub.send(put_of(KEY_A, "peer"))
+                    sub.read_next()
+        finally:
+            server.stop()
+        self.assertEqual(raised.exception.response["code"], "write_not_authorized")
+        self.assertIsNone(self.text_of(server, KEY_A))
+
+    def test_a_granted_put_and_delete_commit_and_the_authority_saw_this_peer(self) -> None:
+        authority = Recording(lambda _: CultMeshRemoteWriteDecision.grant())
+        server = self.build(authority)
+
+        put_replies = self.send(server, put_of(KEY_A, "peer"))
+        self.assertEqual(self.text_of(server, KEY_A), "peer")
+        self.assertEqual(put_replies, [])
+        self.send(server, delete_of(KEY_A))
+        self.assertIsNone(self.text_of(server, KEY_A))
+
+        self.assertEqual(len(authority.requests), 2)
+        self.assertTrue(all(request.peer is PEER for request in authority.requests))
+
+    def test_the_request_carries_the_resolved_write_and_scoped_grants_refuse_their_neighbours(self) -> None:
+        authority = Recording(
+            lambda request: CultMeshRemoteWriteDecision.grant()
+            if request.operation is CultMeshRemoteWriteOperation.PUT and request.record_key == KEY_A
+            else CultMeshRemoteWriteDecision.refuse("scoped")
+        )
+        server = self.build(authority)
+        self.seed(server, KEY_A, "server")
+
+        self.send(server, put_of(KEY_A, "aliased", schema_id=ALIAS))
+
+        request = authority.requests[0]
+        self.assertEqual(request.schema_id, SCHEMA_ID, "the authority sees the canonical id, not the alias as sent")
+        self.assertIs(request.operation, CultMeshRemoteWriteOperation.PUT)
+        self.assertEqual(request.record_key, KEY_A)
+        self.assertEqual(request.shard_id, SHARD_ID)
+        self.assertEqual(request.shard_epoch, 1)
+        self.assertEqual(self.text_of(server, KEY_A), "aliased")
+
+        self.assert_refused(self.send(server, put_of(KEY_B, "other-key")), KEY_B)
+        self.assert_refused(self.send(server, delete_of(KEY_A)), KEY_A)
+        self.assertIsNone(self.text_of(server, KEY_B), "a grant scoped to key a refuses key b")
+        self.assertEqual(self.text_of(server, KEY_A), "aliased", "a grant scoped to put refuses delete")
+        self.assertIs(authority.requests[-1].operation, CultMeshRemoteWriteOperation.DELETE)
+        self.assertEqual(authority.requests[-1].schema_id, SCHEMA_ID)
+
+    def test_refusal_precedes_every_other_judgment_of_the_put(self) -> None:
+        refusing = self.build(None)
+        unowned = put_of("unrelated:authority", "x", schema_id="sha256:nobody-owns-this")
+        unowned["document"]["payload"] = msgpack.packb(["nobody.owns.this.v1", "x"], use_bin_type=True)
+        self.assert_refused(self.send(refusing, unowned), "unrelated:authority")
+        blank_key = put_of("", "x")
+        self.assert_refused(self.send(refusing, blank_key), "")
+
+        granting = self.build(Recording(lambda _: CultMeshRemoteWriteDecision.grant()))
+        reply = self.send(granting, unowned)
+        self.assertEqual(reply[0]["code"], "unregistered_document_put", "a granted put still meets the store's refusal")
+
+    def test_an_authority_that_raises_refuses_and_is_logged_as_an_error(self) -> None:
+        def explode(_: Any) -> Any:
+            raise RuntimeError("authority exploded")
+
+        server = self.build(Recording(explode))
+        self.seed(server, KEY_A, "server")
+
+        with self.assertLogs("cultmesh_py.server", level="ERROR") as logs:
+            self.assert_refused(self.send(server, put_of(KEY_A, "peer")), KEY_A)
+            self.assert_refused(self.send(server, delete_of(KEY_A)), KEY_A)
+
+        self.assertEqual(self.text_of(server, KEY_A), "server", "fail closed")
+        self.assertTrue(any("RuntimeError" in line for line in logs.output))
+
+    def test_a_refusal_with_no_reason_still_says_why(self) -> None:
+        server = self.build(Recording(lambda _: CultMeshRemoteWriteDecision.refuse()))
+        self.assert_refused(self.send(server, put_of(KEY_A, "peer")), KEY_A, error="write_not_authorized")
+
+    def test_the_refusal_names_the_key_and_the_code_in_the_pinned_bytes(self) -> None:
+        vector = next(v for v in _VECTORS if v["name"] == "write_not_authorized")
+        message = error_message(
+            "write_not_authorized: not granted",
+            code="write_not_authorized",
+            details={"field": "recordKey", "value": "some:key"},
+        )
+        self.assertEqual(message.to_bytes(), bytes.fromhex(vector["messagePackHex"]))
+
+        server = self.build(Recording(lambda _: CultMeshRemoteWriteDecision.refuse("not granted")))
+        reply = self.send(server, put_of("some:key", "peer"))[0]
+        self.assertEqual(msgpack.packb(reply, use_bin_type=True), bytes.fromhex(vector["messagePackHex"]))
+
+    def test_the_peer_receives_the_typed_refusal_over_tcp_and_a_grant_sends_nothing_back(self) -> None:
+        authority = Recording(
+            lambda request: CultMeshRemoteWriteDecision.grant()
+            if request.record_key == KEY_A
+            else CultMeshRemoteWriteDecision.refuse("only a")
+        )
+        server = self.build(authority)
+        server.start()
+        try:
+            client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
+            with client.subscribe_database(subscription_id="s", schema_ids=[SCHEMA_ID], include_snapshot=False) as sub:
+                sub.send(put_of(KEY_A, "peer"))
+                self.assertEqual(sub.read_next_change().record_key, KEY_A, "the granted write was answered by its change, not an error")
+                sub.send(put_of(KEY_B, "peer"))
+                with self.assertRaises(CultNetPeerError) as raised:
+                    sub.read_next()
+        finally:
+            server.stop()
+
+        reply = raised.exception.response
+        self.assertEqual(reply["error"], "write_not_authorized: only a")
+        self.assertEqual(reply["details"]["value"], KEY_B)
+        self.assertTrue(authority.requests[0].peer.endpoint.startswith("127.0.0.1:"))
+        self.assertEqual(authority.requests[0].peer.transport, "tcp")
+        self.assertEqual(self.text_of(server, KEY_A), "peer")
+        self.assertIsNone(self.text_of(server, KEY_B))
+
+    def test_a_delete_sent_under_a_schema_alias_asks_the_authority_with_the_canonical_id(self) -> None:
+        authority = Recording(lambda _: CultMeshRemoteWriteDecision.grant())
+        server = self.build(authority)
+        self.seed(server, KEY_A, "server")
+
+        self.send(server, delete_of(KEY_A, schema_id="tests.authority_note.v1"))
+
+        request = authority.requests[0]
+        self.assertIs(request.operation, CultMeshRemoteWriteOperation.DELETE)
+        self.assertEqual(request.schema_id, SCHEMA_ID)
+        self.assertIsNone(self.text_of(server, KEY_A), "the aliased delete was granted and applied")
+
+    def test_a_missing_decision_refuses_put_and_delete(self) -> None:
+        server = self.build(Recording(lambda _: None))
+        self.seed(server, KEY_A, "server")
+
+        self.assert_refused(self.send(server, put_of(KEY_A, "peer")), KEY_A)
+        self.assert_refused(self.send(server, delete_of(KEY_A)), KEY_A)
+        self.assert_refused(self.send(server, put_of(KEY_B, "peer")), KEY_B)
+
+        self.assertEqual(self.text_of(server, KEY_A), "server")
+        self.assertIsNone(self.text_of(server, KEY_B))
+
+    def test_a_peer_over_rudp_is_refused_with_its_rudp_endpoint_and_the_server_serves_on(self) -> None:
+        authority = Recording(lambda _: CultMeshRemoteWriteDecision.refuse("rudp no"))
+        server = self.build(authority)
+        server.start()
+        try:
+            transport = create_rudp_schema_transport(
+                host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id, timeout_seconds=4.0
+            )
+            try:
+                transport.send("schema", msgpack.packb(put_of(KEY_A, "peer"), use_bin_type=True))
+                reply = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+                transport.send("schema", msgpack.packb(delete_of(KEY_A), use_bin_type=True))
+                delete_reply = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            finally:
+                transport.close()
+        finally:
+            server.stop()
+        self.assertEqual(reply["error"], "write_not_authorized: rudp no")
+        self.assertEqual(delete_reply["code"], "write_not_authorized")
+        self.assertEqual([r.peer.transport for r in authority.requests], ["rudp", "rudp"])
+        self.assertTrue(all(r.peer.endpoint.startswith("127.0.0.1:") for r in authority.requests))
+        self.assertIsNone(self.text_of(server, KEY_A))
+
+    def test_a_put_with_an_absurd_payload_is_refused_by_the_gate_not_crashed_on(self) -> None:
+        server = self.build(None)
+        absurd = put_of(KEY_A, "x")
+        absurd["document"]["payload"] = 2**70
+        self.assert_refused(self.send(server, absurd), KEY_A)
+
+    def test_an_ungranted_peers_huge_payload_is_refused_without_allocating_it(self) -> None:
+        import tracemalloc
+
+        server = self.build(None)
+        huge = put_of(KEY_A, "x")
+        huge["document"]["payload"] = 10**9  # bytes(10**9) would allocate a gigabyte
+        tracemalloc.start()
+        try:
+            replies = self.send(server, huge)
+            _, peak = tracemalloc.get_traced_memory()
+        finally:
+            tracemalloc.stop()
+        self.assert_refused(replies, KEY_A)
+        self.assertLess(peak, 10_000_000)
+
+    def test_shard_fields_of_the_wrong_type_reach_the_authority_as_none(self) -> None:
+        authority = Recording(lambda _: CultMeshRemoteWriteDecision.grant())
+        server = self.build(authority)
+        message = put_of(KEY_A, "x")
+        message["shardId"] = {"not": "a string"}
+        message["shardEpoch"] = "7"
+        self.send(server, message)
+        self.assertIsNone(authority.requests[0].shard_id)
+        self.assertIsNone(authority.requests[0].shard_epoch)
+
+    def test_grant_all_is_an_explicit_choice_that_grants_every_write(self) -> None:
+        server = self.build(CultMeshGrantAllWrites())
+        self.send(server, put_of(KEY_A, "peer"))
+        self.assertEqual(self.text_of(server, KEY_A), "peer")
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultmesh-py/tests/test_rudp_server_generation.py

@@ -10,7 +10,7 @@ from dataclasses import dataclass
 
 import msgpack
 from cultcache_py import define_database_entry_type
-from cultmesh_py import create_node
+from cultmesh_py import CultMeshGrantAllWrites, create_node
 from cultmesh_py.server import CultMeshLocalServer
 from cultnet_py import (
     CultNetRudpSendOptions,
@@ -49,7 +49,7 @@ class RudpServerGenerationTests(unittest.TestCase):
         node = create_node(runtime_id="rudp-generation")
         node.database.register_document(NOTE)
         # Replies stay whole datagrams here, so each server packet the pump reads is one message.
-        self.server = CultMeshLocalServer(node=node, rudp_connection_id=CONNECTION_ID, rudp_max_fragment_bytes=60_000)
+        self.server = CultMeshLocalServer(node=node, rudp_connection_id=CONNECTION_ID, rudp_max_fragment_bytes=60_000, write_authority=CultMeshGrantAllWrites())
         self.server.start()
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.sock.bind(("127.0.0.1", 0))