From 4820ac74c9b4e8f8ff336170e27c23cd2a8d1b30 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 19 Oct 2026 16:09:15 +0200 Subject: [PATCH] cultmesh-ts/content: serve a private copy of the hashed bytes, open the store non-blocking, test the input-shaped mutants The handler copies the store's bytes before hashing and sends that copy, so a reused store buffer cannot change the answer. The directory store opens with O_NONBLOCK so a FIFO named by a hash is refused by fstat instead of holding a libuv thread. New tests: exact record-key compare and blank keys, empty-hash prefix lookalikes, single store read, buffer reuse, FIFO (POSIX, uid-independent). Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-ts/README.md | 5 +- packages/cultmesh-ts/src/content.ts | 9 ++- packages/cultmesh-ts/test/content.test.ts | 77 +++++++++++++++++++++++ 3 files changed, 86 insertions(+), 5 deletions(-) diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 63858126..de7b2794 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -528,8 +528,9 @@ process, and chunks are never served by path. The directory store reads a flat `/` layout. The publisher writes it from `packContent`'s chunks (an artifact of one chunk is the file named by its content hash). The store never writes, lists or caches. TLS and concurrency -bounds belong to the host: memory is at most `maxChunkBytes` per in-flight -request. `onServerFault` reports every 500 (`hash-mismatch`, `too-large`, +bounds belong to the host: memory is at most two buffers of `maxChunkBytes` per +in-flight request (the store's, and the private copy that is hashed and sent, so a +host store may reuse its buffer). `onServerFault` reports every 500 (`hash-mismatch`, `too-large`, `store-error`) for health. ```ts diff --git a/packages/cultmesh-ts/src/content.ts b/packages/cultmesh-ts/src/content.ts index c130cbb8..3191a285 100644 --- a/packages/cultmesh-ts/src/content.ts +++ b/packages/cultmesh-ts/src/content.ts @@ -1,3 +1,4 @@ +import { constants } from "node:fs"; import { open } from "node:fs/promises"; import type { IncomingMessage, ServerResponse } from "node:http"; import { join } from "node:path"; @@ -70,7 +71,9 @@ export function createContentChunkHandler(store: ContentChunkStore, options: Con catch (error) { return fault(hex, error instanceof ContentChunkTooLargeError ? "too-large" : "store-error", error); } if (!bytes) return refusal(404); if (bytes.length > maxChunkBytes) return fault(hex, "too-large"); - return await sha256Hex(bytes) === hex ? chunkAnswer(bytes) : fault(hex, "hash-mismatch"); + // The copy is the provider's own: the bytes hashed are exactly the bytes sent, whatever the store does with its buffer afterwards. + const owned = new Uint8Array(bytes); + return await sha256Hex(owned) === hex ? chunkAnswer(owned) : fault(hex, "hash-mismatch"); }; return async ({ method, url }) => { @@ -80,12 +83,12 @@ export function createContentChunkHandler(store: ContentChunkStore, options: Con }; } -/** A read-only flat directory: the chunk whose hash is `` lives in `/`. It writes, lists and caches nothing. */ +/** A read-only flat directory (opened non-blocking, so a FIFO named by a hash is refused by fstat instead of holding a thread): the chunk whose hash is `` lives in `/`. It writes, lists and caches nothing. */ export function createDirectoryChunkStore(root: string): ContentChunkStore { return { async readChunk(chunkHash, maxBytes) { if (parseSha256Hex(chunkHash) !== chunkHash) throw new Error("chunkHash must be canonical lowercase hex."); - const file = await open(join(root, chunkHash), "r").catch(error => { if (error?.code === "ENOENT") return undefined; throw error; }); + const file = await open(join(root, chunkHash), constants.O_RDONLY | constants.O_NONBLOCK).catch(error => { if (error?.code === "ENOENT") return undefined; throw error; }); if (!file) return undefined; try { const stat = await file.stat(); diff --git a/packages/cultmesh-ts/test/content.test.ts b/packages/cultmesh-ts/test/content.test.ts index c7f98ac5..16720955 100644 --- a/packages/cultmesh-ts/test/content.test.ts +++ b/packages/cultmesh-ts/test/content.test.ts @@ -6,6 +6,7 @@ import type { AddressInfo } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test, { mock } from "node:test"; +import { execFileSync } from "node:child_process"; import { EMPTY_SHA256, chunkRecordKey, packContent, sha256Hex } from "cultnet-ts/content"; @@ -260,3 +261,79 @@ test("directory store closes every handle it opens, whether it returns bytes, fi assert.deepEqual(opened.map(handle => handle.fd), opened.map(() => -1)); } finally { spy.mock.restore(); await rm(root, { recursive: true, force: true }); } }); + +test("a recordKey is compared exactly: padded and re-cased forms of the right key answer 400, tab and newline keys are blank and answer 200", async () => { + const chunk = await packed("hello chunk"); + const store = spyStore({ [chunk.chunkHash]: chunk.payload }); + const key = chunkRecordKey(chunk.chunkHash); + const q = (value: string) => `/c?chunkHash=${chunk.chunkHash}&recordKey=${encodeURIComponent(value)}`; + for (const forged of [` ${key}`, `${key} `, `\t${key}`, key.toUpperCase(), key.replace("mesh", "Mesh")]) { + assert.equal((await ask(store, q(forged))).status, 400, JSON.stringify(forged)); + } + assert.deepEqual(store.calls, []); + for (const blank of ["\t", "\n", "\r\n", " \t "]) assert.equal((await ask(store, q(blank))).status, 200, JSON.stringify(blank)); + assert.equal((await ask(store, q(key))).status, 200); +}); + +test("only the empty chunk's own hash is answered without the store: hashes sharing its prefix or differing in the last digit reach the store", async () => { + const flipped = EMPTY_SHA256.slice(0, -1) + (EMPTY_SHA256.endsWith("5") ? "6" : "5"); + const prefixed = EMPTY_SHA256.slice(0, 8) + "0".repeat(56); + for (const hash of [flipped, prefixed]) { + assert.equal(hash.length, 64); + const store = spyStore({}); + const answer = await ask(store, `/c?chunkHash=${hash}`); + assert.equal(answer.status, 404, hash); + assert.deepEqual(store.calls, [hash]); + } +}); + +test("a GET reads the store once and serves the bytes it hashed, not a later read", async () => { + const first = await packed("first read"); + const second = await packed("second read"); + const reads: Uint8Array[] = [first.payload, second.payload]; + const store: ContentChunkStore = { readChunk: async () => reads.shift() }; + const answer = await ask(store, `/c?chunkHash=${first.chunkHash}`); + assert.equal(answer.status, 200); + assert.deepEqual(answer.body, first.payload); + assert.equal(reads.length, 1, "the store was read exactly once"); +}); + +test("a store that reuses its buffer cannot change the bytes after the hash: the answer is the hashed content, from a private copy", async () => { + const chunk = await packed("hello chunk"); + const shared = Buffer.from(chunk.payload); + // The overwrite lands in a microtask chain that runs after the handler resumes and has started hashing, before the digest resolves. + const store: ContentChunkStore = { + readChunk: async () => { + let hops = 0; + const hop = (): void => { if (++hops < 4) queueMicrotask(hop); else shared.fill(0x58); }; + queueMicrotask(hop); + return shared; + }, + }; + const answer = await ask(store, `/c?chunkHash=${chunk.chunkHash}`); + assert.equal(shared[0], 0x58, "the store's buffer was overwritten while the request was in flight"); + assert.equal(answer.status, 200); + assert.deepEqual(answer.body, chunk.payload); + assert.notEqual(answer.body?.buffer, shared.buffer); +}); + +test("a FIFO named by a hash is refused as not a regular file at once and holds no thread: other chunks are still served", { skip: process.platform === "win32" }, async () => { + const root = await mkdtemp(join(tmpdir(), "cm-content-")); + try { + const chunk = await packed("hello chunk"); + await writeFile(join(root, chunk.chunkHash), chunk.payload); + const fifoHash = await sha256Hex(new Uint8Array([9])); + execFileSync("mkfifo", [join(root, fifoHash)]); + const store = createDirectoryChunkStore(root); + const within = async (work: Promise): Promise => { + let timer: NodeJS.Timeout | undefined; + try { return await Promise.race([work, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("blocked")), 3000); })]); } + finally { clearTimeout(timer); } + }; + // More than the libuv pool (4) so a blocking open would starve the pool. + const refusals = Array.from({ length: 8 }, () => within(store.readChunk(fifoHash, 1 << 20)).then(() => "read", (error: Error) => error.message)); + assert.deepEqual(await Promise.all(refusals), Array(8).fill("Chunk is not a regular file.")); + assert.deepEqual(await within(store.readChunk(chunk.chunkHash, 1 << 20)), chunk.payload); + assert.equal((await within(createContentChunkHandler(store)({ method: "GET", url: `/?chunkHash=${fifoHash}` }))).status, 500); + } finally { await rm(root, { recursive: true, force: true }); } +});