diff --git a/beehive/src/direct-membership.ts b/beehive/src/direct-membership.ts index a622a86a0a..31a455d1d4 100644 --- a/beehive/src/direct-membership.ts +++ b/beehive/src/direct-membership.ts @@ -114,10 +114,32 @@ export function fetchRelayTransport(fetcher: typeof globalThis.fetch = globalThi return async request => { const response = await fetcher(request.url, { method: request.method, headers: request.headers, body: request.body, redirect: 'error', signal: request.signal }); const declared = Number(response.headers.get('content-length')); - if (declared > MAX_BODY_BYTES) throw Error('Relay response exceeds bounded size'); - const bodyText = await response.text(); - if (bodyText.length > MAX_BODY_BYTES) throw Error('Relay response exceeds bounded size'); - return { status: response.status, bodyText }; + if (declared > MAX_BODY_BYTES) { + try { await response.body?.cancel(); } catch { /* release is best-effort; the refusal stands */ } + throw Error('Relay response exceeds bounded size'); + } + if (response.body === null) return { status: response.status, bodyText: '' }; + // A chunked body carries no usable Content-Length, so bytes are acquired + // incrementally and the reader is cancelled the moment the cap is exceeded + // or the stream fails: an oversized body is rejected while it is still + // arriving — never buffered whole first — and the connection is released + // instead of being left to drain an already-refused body. + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let received = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done || value === undefined) break; + received += value.byteLength; + if (received > MAX_BODY_BYTES) throw Error('Relay response exceeds bounded size'); + chunks.push(value); + } + } catch (error) { + try { await reader.cancel(); } catch { /* an already failed stream needs no release */ } + throw error; + } + return { status: response.status, bodyText: new TextDecoder().decode(Buffer.concat(chunks, received)) }; }; } @@ -143,11 +165,13 @@ function hexPublicKey(value: unknown, label: string): string { return value; } -function requestScope(timeoutMs: number, external?: AbortSignal) { +/** One bounded request's timeout/cancel scope, shared by the transport race + * and the signer await; `messages` names the boundary that aborted. */ +function requestScope(timeoutMs: number, external?: AbortSignal, messages: { readonly timeout: string; readonly cancelled: string } = { timeout: 'Relay HTTP request timeout', cancelled: 'Relay HTTP request cancelled' }) { const controller = new AbortController(); const abort = (message: string) => { if (!controller.signal.aborted) controller.abort(Error(message)); }; - const timer = setTimeout(() => abort('Relay HTTP request timeout'), timeoutMs); - const cancel = () => abort('Relay HTTP request cancelled'); + const timer = setTimeout(() => abort(messages.timeout), timeoutMs); + const cancel = () => abort(messages.cancelled); if (external?.aborted) cancel(); else external?.addEventListener('abort', cancel, { once: true }); return { signal: controller.signal, done: () => { clearTimeout(timer); external?.removeEventListener('abort', cancel); } }; @@ -215,12 +239,32 @@ function tagValues(event: Event, name: string): string[] { return event.tags.filter(tag => tag[0] === name && tag.length >= 2).map(tag => tag[1]); } +/** The injected signer is an async boundary exactly like the transport: its + * await is raced against the operation's own timeout and caller + * cancellation, so a signer that never settles cannot hold a claim or an + * evidence run open. Signer misuse still surfaces directly once the signer + * settles; only the settling itself is bounded here. */ +async function boundedSigner(signer: HostAuthorizationSigner, request: { url: string; method: 'POST'; payloadSha256Hex: string }, timeoutMs: number, signal?: AbortSignal): Promise { + const scope = requestScope(timeoutMs, signal, { timeout: 'Host authorization signer timed out', cancelled: 'Host authorization signer cancelled' }); + const attempt = signer.signHttpAuthentication(request); + void attempt.catch(() => {}); // an abandoned attempt must not reject unhandled later + try { + return await Promise.race([attempt, new Promise((_, reject) => { + // Same pre-aborted guard as the transport race: an abort whose event + // has already fired must reject immediately, not wait for a listener. + if (scope.signal.aborted) reject(scope.signal.reason); + else scope.signal.addEventListener('abort', () => reject(scope.signal.reason), { once: true }); + })]); + } finally { scope.done(); } +} + /** Signer output is fully re-verified before any request leaves the module, - * binding the signature to the exact host identity, URL, method and payload. */ -async function nip98Authorization(signer: HostAuthorizationSigner, url: string, body: string, now: number): Promise { + * binding the signature to the exact host identity, URL, method and payload; + * the signer's own await stays inside the caller's timeout/cancellation. */ +async function nip98Authorization(signer: HostAuthorizationSigner, url: string, body: string, now: number, timeoutMs: number, signal?: AbortSignal): Promise { const hostPublicKey = hexPublicKey(signer.hostPublicKey, 'host public key'); const payloadSha256Hex = digest(body).toString('hex'); - const serialized = await signer.signHttpAuthentication({ url, method: 'POST', payloadSha256Hex }); + const serialized = await boundedSigner(signer, { url, method: 'POST', payloadSha256Hex }, timeoutMs, signal); let event: Event; try { event = parseEvent(JSON.parse(serialized)); } catch { throw Error('Signer returned a malformed NIP-98 event'); } if (event.kind !== NIP98_KIND) throw Error('Signer returned a non-NIP-98 event kind'); @@ -289,12 +333,13 @@ export async function claimRelayMembershipInvite(input: { const policyReceipt = input.policyReceipt === undefined ? undefined : text(input.policyReceipt, 2048); const now = input.now ?? Math.floor(Date.now() / 1000); const transport = input.transport ?? fetchRelayTransport(); + const timeoutMs = boundedTimeout(input.timeoutMs); const url = `${origin}api/invites/claim`; const body = JSON.stringify(policyReceipt === undefined ? { code: inviteCode } : { code: inviteCode, policy_receipt: policyReceipt }); - const authorization = await nip98Authorization(input.signer, url, body, now); + const authorization = await nip98Authorization(input.signer, url, body, now, timeoutMs, input.signal); let response: { status: number; bodyText: string }; try { - response = await relayFetch({ transport, url, method: 'POST', headers: { accept: 'application/json', 'content-type': 'application/json', authorization }, body, timeoutMs: boundedTimeout(input.timeoutMs), signal: input.signal }); + response = await relayFetch({ transport, url, method: 'POST', headers: { accept: 'application/json', 'content-type': 'application/json', authorization }, body, timeoutMs, signal: input.signal }); } catch (error) { throw Error(`Invite claim request failed before a relay outcome; commit status unknown (${safeReason(error)})`); } @@ -370,7 +415,7 @@ export async function verifyDirectMembershipEvidence(input: { if (relaySelf !== null) { const url = `${origin}query`; const body = JSON.stringify([{ kinds: [ROSTER_KIND], authors: [relaySelf] }]); - const authorization = await nip98Authorization(input.signer, url, body, now); + const authorization = await nip98Authorization(input.signer, url, body, now, timeoutMs, input.signal); try { const response = await relayFetch({ transport, url, method: 'POST', headers: { accept: 'application/json', 'content-type': 'application/json', authorization }, body, timeoutMs, signal: input.signal }); if (response.status === 200) { diff --git a/beehive/test/direct-membership.test.ts b/beehive/test/direct-membership.test.ts index 53bd13635c..b1d1c15e0d 100644 --- a/beehive/test/direct-membership.test.ts +++ b/beehive/test/direct-membership.test.ts @@ -1,7 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; -import { createServer } from 'node:http'; +import { createServer, type ServerResponse } from 'node:http'; import { once } from 'node:events'; import { generateSecretKey, getPublicKey, finalizeEvent, verifyEvent, type Event } from 'nostr-tools/pure'; import { @@ -363,3 +363,73 @@ test('real loopback HTTP journey through fetchRelayTransport with server-side NI await once(server, 'close'); } }); + +test('chunked body over the acquisition cap is rejected mid-stream and the reader is cancelled', async () => { + // No Content-Length on this route, so the only enforceable bound is during + // acquisition, and the server deliberately never finishes the body: the + // transport must reject while bytes are still arriving (never wait for a + // completion that does not happen) and must cancel the reader, which the + // server observes as a connection teardown. A regression to read-then-check + // would hang on the unfinished body until the abort backstop rejects. + let serverResponse: ServerResponse | undefined; + const server = createServer((request, response) => { + assert.equal(request.url, '/unbounded'); + serverResponse = response; + response.writeHead(200, { 'content-type': 'application/json' }); // no content-length → chunked transfer + response.write(Buffer.alloc(300_000, 0x78)); // over the 256 KB cap; response.end() is never called + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + assert.ok(address && typeof address === 'object'); + try { + const backstop = new AbortController(); + setTimeout(() => backstop.abort(), 4000).unref?.(); // a regression must fail the suite, not hang it + const start = Date.now(); + await assert.rejects( + fetchRelayTransport()({ url: `http://127.0.0.1:${address.port}/unbounded`, method: 'GET', headers: {}, body: undefined, timeoutMs: 2000, signal: backstop.signal }), + /bounded size/, + ); + assert.ok(Date.now() - start < 2000, 'rejected while the body was still arriving, not after completion'); + // The cancelled reader released the connection server-side and the + // oversized body never completed. + assert.ok(serverResponse, 'server response observed'); + const tornDown = serverResponse.closed || await Promise.race([ + once(serverResponse, 'close').then(() => true), + new Promise(resolve => { setTimeout(() => resolve(false), 2000).unref?.(); }), + ]); + assert.ok(tornDown, 'the server observed the cancelled reader as a connection close'); + assert.equal(serverResponse.writableFinished, false, 'the oversized body never completed'); + } finally { + server.closeAllConnections(); + server.close(); + await once(server, 'close'); + } +}); + +test('an unsettled host signer is bounded by the operation timeout and cancellation like the transport', async () => { + const relayKey = generateSecretKey(), hostKey = generateSecretKey(); + const neverSettles: HostAuthorizationSigner = { hostPublicKey: getPublicKey(hostKey), signHttpAuthentication: () => new Promise(() => {}) }; + const fixture = scriptedRelay({ relayKey, enforced: true }); + const input = { relay: 'wss://relay.example.com', inviteCode: 'v2.fixture-code', transport: fixture.transport, now: NOW }; + const settlesWithin = (promise: Promise): Promise => Promise.race([promise, new Promise((_, reject) => { setTimeout(() => reject(Error('test bound exceeded')), 500).unref?.(); })]); + + // The injected signer is an async boundary: a key access that never settles + // cannot hold the claim open past the operation's own timeout, and no + // request leaves the module while it hangs. + await assert.rejects(settlesWithin(claimRelayMembershipInvite({ ...input, signer: neverSettles, timeoutMs: 40 })), /Host authorization signer timed out/); + assert.equal(fixture.requests.length, 0); + + // Pre-aborted and mid-flight caller cancellation both reach the signer await. + const preAborted = new AbortController(); + preAborted.abort(); + await assert.rejects(settlesWithin(claimRelayMembershipInvite({ ...input, signer: neverSettles, signal: preAborted.signal, timeoutMs: 5000 })), /Host authorization signer cancelled/); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 10).unref?.(); + await assert.rejects(settlesWithin(claimRelayMembershipInvite({ ...input, signer: neverSettles, signal: controller.signal, timeoutMs: 5000 })), /Host authorization signer cancelled/); + assert.equal(fixture.requests.length, 0); + + // The gated query's signer await is equally bounded: discovery runs, then + // the unsettled signer is abandoned before any POST /query leaves. + await assert.rejects(settlesWithin(verifyDirectMembershipEvidence({ relay: 'wss://relay.example.com', signer: neverSettles, transport: fixture.transport, timeoutMs: 40, now: NOW })), /Host authorization signer timed out/); + assert.deepEqual(fixture.requests.map(request => request.method), ['GET']); +});