cultmesh-ts: gate unreadable writes, malformed refusals, payload tests

A frame that fails to decode or parse is judged as the write it looks like, so
an ungranted peer gets write_not_authorized instead of silence; a granted peer
gets onError and a malformed_document_put/delete refusal. Adds raw-frame tests
(int, 2**70, oversized claimed length, non-decoding, non-map document) and
the timing residual note.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultmesh-ts/CHANGELOG.md

@@ -10,7 +10,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
 - The package is renamed `cultmesh-ts` to `@gamecult/cultmesh-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultmesh-ts` and its subpaths; the API is unchanged. Depends on `@gamecult/cultnet-ts`.
 - `asSchemaAlias` is removed from `CultMeshDocumentHandle`, `CultMeshCollectionHandle` and their bound forms. Read the handle by the schema you mean.
 - `CultMeshDocumentCatalog` (`document`, `tryDocument`, `latest`, `replace`, `submitPrediction`, `authoritativeWriter`, `predictionWriter`, `watch`) and `CultMeshCollectionCatalog` (`collection`, `tryCollection`, `latest`, `watchChanges`) take `<TDocument = never, TSchema>` and yield `CultMeshCatalogValue<TDocument, TSchema>`: an explicit type argument wins, a CultCache document definition names its own value type, and a bare schema descriptor yields `unknown`. A value type is no longer inferred from a value or an expected result. The `parse` option is removed from these calls, and a lookup by type answers with that type's handle or none, not an alias of a compatible schema.
-- `createRudpDocumentServer` refuses every peer put and delete unless it is given a `writeAuthority`. A refused write is answered with the typed `write_not_authorized` naming the record key (the bytes pinned in `error-vectors.json`), and `onDocumentPutRaw` is never called; pass `writeAuthority: new CultMeshGrantAllWrites()` to keep the old accept-everything behaviour. The authority is asked before the put is sized, resolved or decoded, with the canonical schema id and the peer's address; one that throws, rejects or answers anything but `granted: true` refuses. A put carrying a wrong-typed `shardId` or `shardEpoch` is refused before the authority is asked. A peer `document_delete` now reaches the new `onDocumentDeleteRaw` once granted (without it, a text refusal) instead of the unsupported-request error.
+- `createRudpDocumentServer` refuses every peer put and delete unless it is given a `writeAuthority`. A refused write is answered with the typed `write_not_authorized` naming the record key (the bytes pinned in `error-vectors.json`), and `onDocumentPutRaw` is never called; pass `writeAuthority: new CultMeshGrantAllWrites()` to keep the old accept-everything behaviour. The authority is asked before the put is sized, resolved or decoded, with the canonical schema id and the peer's address; one that throws, rejects or answers anything but `granted: true` refuses. A frame the server cannot read is judged as the write it looks like (a put when it does not decode at all), so an ungranted peer gets `write_not_authorized` for it too; a granted peer gets `onError` and a `malformed_document_put` or `malformed_document_delete` refusal. A put carrying a wrong-typed `shardId` or `shardEpoch` is refused with `malformed_document_put` before the authority is asked. Residual: ungranted peers can still distinguish registered from unregistered schemas by timing. A peer `document_delete` now reaches the new `onDocumentDeleteRaw` once granted (without it, a text refusal) instead of the unsupported-request error.
 - Adding a document or collection to a catalog under a schema id or schema name another type already holds is refused with `SchemaConflictError`.
 
 ### Added

packages/cultmesh-ts/README.md

@@ -345,6 +345,8 @@ evidence for the requested provider identity.
 
 ## RUDP Helpers
 
+`createRudpDocumentServer` refuses every peer put and delete unless given a `writeAuthority` (`new CultMeshGrantAllWrites()` accepts all). The authority is asked before a write's payload is read or its schema resolved, and a refusal is the typed `write_not_authorized`. Residual: ungranted peers can still distinguish registered from unregistered schemas by timing.
+
 Node runtimes can build the shared CultNet reliable-UDP transport from the
 branded CultMesh entrypoint while the transport semantics stay owned by
 `@gamecult/cultnet-ts`:

packages/cultmesh-ts/src/index.ts

@@ -5641,7 +5641,21 @@ export class CultMesh {
       payload: Uint8Array,
     ): Promise<void> {
       if (record.closed) return;
-      const message = parseCultNetMessage(decodeWireValue(payload), wireContract);
+      let decoded: unknown;
+      let message: CultNetMessage;
+      try {
+        decoded = decodeWireValue(payload);
+        message = parseCultNetMessage(decoded, wireContract);
+      } catch (error) {
+        // A frame the server cannot read is judged as the write it looks like (a put when it does not even decode), so
+        // an ungranted peer hears the same refusal for it as for any other write, never silence or a parse error.
+        const write = readableWrite(decoded);
+        if (decoded !== undefined && !write) throw error;
+        if (await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) return;
+        reportError(error);
+        sendSchemaMessage(record, malformedWriteError(write?.operation ?? "put", "CultMesh RUDP document server could not read the write."));
+        return;
+      }
       switch (message.schemaVersion) {
         case "cultnet.snapshot_request.v0": {
           if (!options.getCache) {
@@ -5681,7 +5695,7 @@ export class CultMesh {
               (shardId !== undefined && typeof shardId !== "string") ||
               (shardEpoch !== undefined && !Number.isSafeInteger(shardEpoch))
             ) {
-              sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document put carries malformed shard fields."));
+              sendSchemaMessage(record, malformedWriteError("put", "Raw put messages must carry a string shardId and an integer shardEpoch."));
               return;
             }
             if (await writeRefused(record, "put", message.document.schemaId, message.document.recordKey, (shardId as string | undefined) ?? null, (shardEpoch as number | undefined) ?? null)) return;
@@ -5731,6 +5745,27 @@ export class CultMesh {
       }
     }
 
+    /** The put or delete a decoded frame claims to be, with the ids it carries when they are strings. */
+    function readableWrite(value: unknown): { operation: CultMeshRemoteWriteOperation; schemaId: string; recordKey: string } | undefined {
+      if (typeof value !== "object" || value === null) return undefined;
+      const frame = value as { schemaVersion?: unknown; schemaId?: unknown; recordKey?: unknown; document?: unknown };
+      const str = (v: unknown): string => (typeof v === "string" ? v : "");
+      if (frame.schemaVersion === "cultnet.document_delete.v0") {
+        return { operation: "delete", schemaId: str(frame.schemaId), recordKey: str(frame.recordKey) };
+      }
+      if (frame.schemaVersion === "cultnet.document_put_raw.v0") {
+        const document = (typeof frame.document === "object" && frame.document !== null ? frame.document : {}) as { schemaId?: unknown; recordKey?: unknown };
+        return { operation: "put", schemaId: str(document.schemaId), recordKey: str(document.recordKey) };
+      }
+      return undefined;
+    }
+
+    // Not a `code`: malformed_document_put/delete are outside the schema's closed code enum, which TS and Rust peers
+    // refuse to decode (docs/cultnet-error-contract-cut.md), so the name leads the text of a codeless error instead.
+    function malformedWriteError(operation: CultMeshRemoteWriteOperation, text: string): CultNetMessage {
+      return cultNetTextError(`malformed_document_${operation}: ${text}`);
+    }
+
     /**
      * Asks the write authority about a put or delete, before anything of the write is sized, resolved or decoded, and
      * answers a refusal with the typed `write_not_authorized` naming the record. True when the write is refused.

packages/cultmesh-ts/test/remote-write-authority.test.ts

@@ -9,10 +9,10 @@ import { documentPayloadFromBytes, defineDocumentType } from "@gamecult/cultcach
 import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
 import {
   CultNetDocumentRegistry,
+  CultNetPeer,
   cultNetWriteNotAuthorizedError,
   defineCultNetDocumentBinding,
   type CultNetMessage,
-  type CultNetPeer,
 } from "@gamecult/cultnet-ts";
 import {
   CultMesh,
@@ -99,7 +99,7 @@ async function rig(writeAuthority?: CultMeshRemoteWriteAuthority, extra: { maxPa
     ...extra,
   });
   await server.start();
-  const peer = await CultMesh.createRudpPeer("authority-peer", connectionId, `rudp://127.0.0.1:${server.bind.port}`, {
+  const peer: CultNetPeer = await CultMesh.createRudpPeer("authority-peer", connectionId, `rudp://127.0.0.1:${server.bind.port}`, {
     resendDelayMs: 25,
     resendPollMs: 5,
     connectTimeoutMs: 2_000,
@@ -273,7 +273,7 @@ test("wrong-typed shard fields on a put are refused before the authority and rea
     for (const fields of bad) r.peer.send(put(KEY_A, fields));
     await r.settle();
     assert.equal(r.replies.length, bad.length);
-    assert.ok(r.replies.every((reply) => reply.schemaVersion === "cultnet.error.v0"));
+    assert.ok(r.replies.every((reply) => reply.schemaVersion === "cultnet.error.v0" && reply.code === null && reply.error.startsWith("malformed_document_put:")));
     assert.deepEqual(authority.requests, [], "the authority was never asked");
     assert.equal(r.puts.length, 0);
   });
@@ -315,3 +315,87 @@ test("a granted delete with no delete handler is answered with a text refusal",
     assert.notEqual((replies[0] as { code?: string }).code, "write_not_authorized");
   } finally { peer.close(); server.close(); }
 });
+
+// Hostile frames, sent as raw bytes the typed peer would never build.
+const bin = (n: number, ...tail: number[]) => [0xc6, (n >>> 24) & 255, (n >>> 16) & 255, (n >>> 8) & 255, n & 255, ...tail]; // bin32 header
+const str = (t: string) => [0xa0 | t.length, ...Buffer.from(t)];
+function rawPut(payloadBytes: number[], key = KEY_A): Uint8Array {
+  // {schemaVersion, messageId, document: {schemaId, recordKey, storedAt, payloadEncoding, payload: <payloadBytes>}}
+  return Uint8Array.from([
+    0x83, ...str("schemaVersion"), ...str("cultnet.document_put_raw.v0"), ...str("messageId"), ...str("raw"),
+    ...str("document"), 0x85, ...str("schemaId"), ...str(SCHEMA_ID), ...str("recordKey"), ...str(key),
+    ...str("storedAt"), ...str("2026-11-07T00:00:00Z"), ...str("payloadEncoding"), ...str("messagepack"),
+    ...str("payload"), ...payloadBytes,
+  ]);
+}
+
+async function rawRig(writeAuthority?: CultMeshRemoteWriteAuthority) {
+  const connectionId = nextConnection++;
+  const puts: CultMeshRudpDocumentPut[] = [];
+  const errors: Error[] = [];
+  const server = CultMesh.createRudpDocumentServer("authority-raw", connectionId, {
+    documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+    writeAuthority, bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
+    onDocumentPutRaw: (d) => { puts.push(d); }, onError: (e) => errors.push(e),
+  });
+  await server.start();
+  const client = await CultMesh.createRudpClient("raw", connectionId, `rudp://127.0.0.1:${server.bind.port}`, { resendDelayMs: 25, resendPollMs: 5 });
+  client.connect();
+  const started = Date.now();
+  while (!client.connected) {
+    if (Date.now() - started > 2_000) throw new Error("not connected");
+    await sleep(5);
+  }
+  const replies: CultNetMessage[] = [];
+  new CultNetPeer(client, { wireContract: "cultnet.schema.v0" }).on("message", (m) => replies.push(m));
+  return { puts, errors, replies, send: (b: Uint8Array) => client.send("schema", b), close: () => { client.close(); server.close(); } };
+}
+
+const hostile: [string, Uint8Array][] = [
+  ["an int payload", rawPut([0x2a])],
+  ["a 2**70-sized integer payload", rawPut([0xcf, 255, 255, 255, 255, 255, 255, 255, 255])],
+  ["a bin whose claimed length exceeds the frame", rawPut(bin(1_000_000_000, 1, 2, 3))],
+  ["a frame that does not decode at all", Uint8Array.from([0xc1])],
+  ["a put whose document is not a map", Uint8Array.from([0x82, ...str("schemaVersion"), ...str("cultnet.document_put_raw.v0"), ...str("document"), 0x05])],
+];
+
+test("an ungranted peer's unreadable put is refused with write_not_authorized, never silence, and nothing is allocated for it", async () => {
+  for (const [name, frame] of hostile) {
+    const r = await rawRig(undefined);
+    try {
+      const before = process.memoryUsage().arrayBuffers;
+      r.send(frame);
+      await sleep(250);
+      assert.equal(r.replies.length, 1, name);
+      assert.equal((r.replies[0] as { code?: string }).code, "write_not_authorized", name);
+      assert.ok(process.memoryUsage().arrayBuffers - before < 50_000_000, `${name}: the claimed length was not allocated`);
+      assert.deepEqual([r.puts.length, r.errors.length], [0, 0], name);
+    } finally { r.close(); }
+  }
+});
+
+test("a granted peer's unreadable put reaches onError and gets a malformed_document_put refusal", async () => {
+  for (const [name, frame] of hostile) {
+    const r = await rawRig(new CultMeshGrantAllWrites());
+    try {
+      r.send(frame);
+      await sleep(250);
+      assert.equal(r.replies.length, 1, name);
+      assert.ok((r.replies[0] as { error: string }).error.startsWith("malformed_document_put:"), name);
+      assert.equal(r.errors.length, 1, name);
+      assert.equal(r.puts.length, 0, name);
+    } finally { r.close(); }
+  }
+});
+
+test("an ungranted peer's malformed delete is refused with write_not_authorized, a granted one gets malformed_document_delete", async () => {
+  const frame = Uint8Array.from([0x84, ...str("schemaVersion"), ...str("cultnet.document_delete.v0"), ...str("messageId"), ...str("d"), ...str("schemaId"), ...str(SCHEMA_ID), ...str("recordKey"), 0x07]);
+  for (const [authority, code] of [[undefined, "write_not_authorized"], [new CultMeshGrantAllWrites(), "malformed_document_delete:"]] as const) {
+    const r = await rawRig(authority);
+    try {
+      r.send(frame);
+      await sleep(250);
+      assert.ok((r.replies[0] as { error: string })?.error.startsWith(code), code);
+    } finally { r.close(); }
+  }
+});