jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

beehive: bound chunked body acquisition and signer settling

Browse Source 
fetchRelayTransport now acquires response bytes through the stream
reader and cancels it the moment the 256 KB cap is exceeded or the
stream fails, so a chunked body with no usable Content-Length is
rejected while it is still arriving instead of after response.text()
has already buffered it all. The injected host signer is an async
boundary like the transport: its await is raced against the same
per-request timeout and caller cancellation, so a signer that never
settles cannot hold a claim or evidence run open. Redirect refusal,
error redaction and abort semantics are unchanged.

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-10 14:38:18 -04:00
parent ac781fd953
commit 7176028b7c
2 changed files with 129 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
beehive/src/direct-membership.ts
View File

@@ -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<string> {
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<never>((_, 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<string> {
* 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<string> {
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) {

72
beehive/test/direct-membership.test.ts
View File

@@ -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<void>(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<boolean>(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<string>(() => {}) };
const fixture = scriptedRelay({ relayKey, enforced: true });
const input = { relay: 'wss://relay.example.com', inviteCode: 'v2.fixture-code', transport: fixture.transport, now: NOW };
const settlesWithin = <T>(promise: Promise<T>): Promise<T> => Promise.race([promise, new Promise<never>((_, 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']);
});