diff --git a/apps/aggregator/package.json b/apps/aggregator/package.json index 5e7f8ac93f..7137b0efbf 100644 --- a/apps/aggregator/package.json +++ b/apps/aggregator/package.json @@ -32,7 +32,8 @@ "@atcute/xrpc-server": "catalog:", "@atcute/xrpc-server-cloudflare": "catalog:", "@emdash-cms/registry-lexicons": "workspace:*", - "@emdash-cms/registry-moderation": "workspace:*" + "@emdash-cms/registry-moderation": "workspace:*", + "emdash": "workspace:*" }, "devDependencies": { "@cloudflare/vite-plugin": "catalog:", diff --git a/apps/aggregator/src/pds-verify.ts b/apps/aggregator/src/pds-verify.ts index 4b31bb0b0a..1c8965bd19 100644 --- a/apps/aggregator/src/pds-verify.ts +++ b/apps/aggregator/src/pds-verify.ts @@ -20,18 +20,40 @@ import type { PublicKey } from "@atcute/crypto"; import { type AtprotoDid, isDid } from "@atcute/lexicons/syntax"; import { verifyRecord } from "@atcute/repo"; +import { type DnsResolver, resolveAndValidateExternalUrl, SsrfError } from "emdash/security/ssrf"; const DEFAULT_TIMEOUT_MS = 15_000; /** 5 MB ceiling. Records and their proofs are tiny (sub-KB typical); this is * a defence against a hostile or broken PDS streaming an unbounded body. */ const DEFAULT_MAX_RESPONSE_BYTES = 5 * 1024 * 1024; +/** Redirect hops we'll follow before giving up. The PDS endpoint is + * publisher-controlled, so redirects are an SSRF pivot: every hop is + * re-validated, and this caps the chain. */ +const MAX_PDS_REDIRECTS = 3; +/** Message prefix `resolveAndValidateExternalUrl` uses when the injected DNS + * resolver itself throws (DoH network error / SERVFAIL / timeout) — a transient + * infrastructure failure, distinct from a disallowed address. Keyed on here to + * classify those as retryable; see `assertFetchableUrl`. */ +const RESOLVER_FAILURE_PREFIX = "Could not resolve hostname:"; +/** Message `resolveAndValidateExternalUrl` uses for an empty DNS answer + * (NXDOMAIN / NOERROR-NODATA / CNAME-only). Transient: a host mid-propagation + * would otherwise be permanently dead-lettered; a genuinely-gone host instead + * dead-letters via retry exhaustion. Not a disallowed address. */ +const EMPTY_ANSWER_MESSAGE = "Hostname resolved to no addresses"; +/** Delay, in seconds, before the consumer re-delivers a message that failed on a + * transient DNS resolution (empty answer / resolver outage). The records queue + * re-delivers immediately by default, so without this a host mid-propagation + * burns all `max_retries` in seconds and lands in the DLQ before DNS settles. + * A fixed pause spreads the retries across a propagation window. */ +const RESOLUTION_RETRY_DELAY_SECONDS = 60; export type VerificationFailureReason = | "PDS_NETWORK_ERROR" | "PDS_HTTP_ERROR" | "RECORD_NOT_FOUND" | "RESPONSE_TOO_LARGE" - | "INVALID_PROOF"; + | "INVALID_PROOF" + | "PDS_ADDRESS_BLOCKED"; export class PdsVerificationError extends Error { override readonly name = "PdsVerificationError"; @@ -40,6 +62,10 @@ export class PdsVerificationError extends Error { message: string, readonly status?: number, override readonly cause?: unknown, + /** When set, the consumer re-delivers the message after this many seconds + * instead of immediately. Set only for transient DNS-resolution failures + * that need time to propagate; other transient retries stay immediate. */ + readonly retryAfterSeconds?: number, ) { super(message); } @@ -51,12 +77,20 @@ export interface FetchAndVerifyOptions { collection: string; rkey: string; publicKey: PublicKey; - /** Default 15s. Aborts the fetch if the PDS is slow. */ + /** Default 15s wall-clock budget covering every redirect hop AND the body + * read — a slow-drip body is bounded by this, not just the initial fetch. */ timeoutMs?: number; /** Default 5 MB. Rejects with `RESPONSE_TOO_LARGE` if exceeded. */ maxResponseBytes?: number; /** Inject for tests; defaults to `globalThis.fetch`. */ fetch?: typeof fetch; + /** + * SSRF-safe hostname resolver applied to the PDS endpoint and every redirect + * target before we connect. Production injects `cloudflareDohResolver`; tests + * stub it. Absent means fail closed — the publisher-controlled fetch is + * refused rather than issued unprotected. + */ + resolveHostname?: DnsResolver; } export interface VerifiedPdsRecord { @@ -74,6 +108,16 @@ export async function fetchAndVerifyRecord( const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; const maxResponseBytes = opts.maxResponseBytes ?? DEFAULT_MAX_RESPONSE_BYTES; + // Fail closed: without a resolver we can't validate the publisher-controlled + // endpoint against the SSRF blocklist, so we refuse rather than fetch blind. + const resolveHostname = opts.resolveHostname; + if (!resolveHostname) { + throw new PdsVerificationError( + "PDS_ADDRESS_BLOCKED", + "refusing PDS fetch: no SSRF-safe hostname resolver was provided", + ); + } + if (!isAtprotoDid(opts.did)) { // Caller is expected to have validated this upstream (the resolver // rejects non-DID strings before reaching here), but the verifier's @@ -84,7 +128,7 @@ export async function fetchAndVerifyRecord( ); } const url = buildGetRecordUrl(opts.pds, opts.did, opts.collection, opts.rkey); - const carBytes = await fetchCar(fetchImpl, url, timeoutMs, maxResponseBytes); + const carBytes = await fetchCar(fetchImpl, url, timeoutMs, maxResponseBytes, resolveHostname); try { const result = await verifyRecord({ @@ -117,54 +161,179 @@ function buildGetRecordUrl(pds: string, did: string, collection: string, rkey: s return url.toString(); } +function isRedirectStatus(status: number): boolean { + return status === 301 || status === 302 || status === 303 || status === 307 || status === 308; +} + +/** + * True when an `SsrfError` message denotes a transient resolution failure (the + * resolver threw, or the DNS answer was empty) rather than a disallowed address. + * Keyed on the shared helper's messages; guarded by tests so a wording change + * there fails loudly instead of silently re-classifying. + */ +function isTransientResolutionFailure(message: string): boolean { + return message.startsWith(RESOLVER_FAILURE_PREFIX) || message === EMPTY_ANSWER_MESSAGE; +} + +/** + * Reject the URL unless it's HTTPS and resolves to a public address. The PDS + * endpoint (and every redirect target) is publisher-controlled, so this is the + * SSRF boundary: HTTP is refused (no plaintext downgrade to an internal + * listener the resolver can't see), and `resolveAndValidateExternalUrl` resolves + * the hostname and rejects any private/reserved address. IP/address logic lives + * in the shared helper — never duplicated here. + */ +async function assertFetchableUrl(url: string, resolver: DnsResolver): Promise { + let parsed: URL; + try { + parsed = new URL(url); + } catch { + throw new PdsVerificationError("PDS_ADDRESS_BLOCKED", `invalid PDS URL: ${url}`); + } + if (parsed.protocol !== "https:") { + throw new PdsVerificationError("PDS_ADDRESS_BLOCKED", `PDS URL must be https: ${url}`); + } + try { + await resolveAndValidateExternalUrl(url, { resolver }); + } catch (err) { + if (err instanceof SsrfError) { + // `resolveAndValidateExternalUrl` throws `SsrfError` for two very + // different classes of situation: a disallowed address (permanent — + // dead-letter) and a resolution failure (transient — retry). The + // transient class is a DoH network error / SERVFAIL / timeout (the + // resolver threw, prefixed `RESOLVER_FAILURE_PREFIX`) or an empty DNS + // answer (`EMPTY_ANSWER_MESSAGE` — a host mid-propagation). `SsrfError.code` + // is the same constant for all, so the only discriminator is the message. + // Tests assert both transient mappings so a future wording change in the + // shared helper breaks loudly here rather than silently dead-lettering + // legitimate records during a DoH blip or DNS propagation. + if (isTransientResolutionFailure(err.message)) { + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + `PDS host resolution failed: ${err.message}`, + undefined, + err, + RESOLUTION_RETRY_DELAY_SECONDS, + ); + } + throw new PdsVerificationError( + "PDS_ADDRESS_BLOCKED", + `PDS address rejected: ${err.message}`, + undefined, + err, + ); + } + throw err; + } +} + async function fetchCar( fetchImpl: typeof fetch, - url: string, + initialUrl: string, timeoutMs: number, maxBytes: number, + resolver: DnsResolver, ): Promise { + // One wall-clock budget spanning every redirect hop and the body read, so a + // PDS that dribbles headers-then-body can't hold the request open past it. const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); - let response: Response; try { - response = await fetchImpl(url, { - signal: controller.signal, - headers: { accept: "application/vnd.ipld.car" }, - }); - } catch (err) { - // Whether the fetch threw because we aborted (timeout) or because the - // network failed at the OS layer, the right caller behaviour is the - // same: retry. Lump them under PDS_NETWORK_ERROR. - throw new PdsVerificationError( - "PDS_NETWORK_ERROR", - err instanceof Error && err.name === "AbortError" - ? `PDS fetch aborted after ${timeoutMs}ms` - : `PDS fetch failed: ${err instanceof Error ? err.message : String(err)}`, - undefined, - err, - ); + let currentUrl = initialUrl; + for (let hop = 0; ; hop++) { + // Re-resolve and re-validate every hop: an allowed public host can + // 30x to a private one, so validating only the initial URL is a hole. + await assertFetchableUrl(currentUrl, resolver); + + let response: Response; + try { + response = await fetchImpl(currentUrl, { + signal: controller.signal, + // Intercept redirects so each target is re-validated before we + // follow it, rather than letting fetch chase them unchecked. + redirect: "manual", + headers: { accept: "application/vnd.ipld.car" }, + }); + } catch (err) { + // Whether the fetch threw because we aborted (timeout) or because + // the network failed at the OS layer, the right caller behaviour + // is the same: retry. Lump them under PDS_NETWORK_ERROR. + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + err instanceof Error && err.name === "AbortError" + ? `PDS fetch aborted after ${timeoutMs}ms` + : `PDS fetch failed: ${err instanceof Error ? err.message : String(err)}`, + undefined, + err, + ); + } + + if (isRedirectStatus(response.status)) { + // Free the redirect response's socket before following. + await response.body?.cancel().catch(() => {}); + if (hop >= MAX_PDS_REDIRECTS) { + throw new PdsVerificationError( + "PDS_ADDRESS_BLOCKED", + `PDS exceeded ${MAX_PDS_REDIRECTS} redirects`, + ); + } + const location = response.headers.get("location"); + if (!location) { + throw new PdsVerificationError( + "PDS_HTTP_ERROR", + `PDS redirect without a Location header`, + response.status, + ); + } + // The publisher fully controls this header; a malformed value + // (e.g. `http://` with an empty authority) makes `new URL` throw a + // raw TypeError. Classify it as a blocked redirect rather than + // letting it escape as an UNEXPECTED_ERROR the publisher can force. + try { + currentUrl = new URL(location, currentUrl).toString(); + } catch { + throw new PdsVerificationError( + "PDS_ADDRESS_BLOCKED", + `PDS redirect Location is not a valid URL: ${location}`, + ); + } + continue; + } + + if (response.status === 404) { + // Distinct from a generic 4xx. The publisher may have deleted the + // record between Jetstream emitting and us fetching, which is the + // common cause; other 4xx (auth, bad request) suggest programming + // errors. Both end up dead-lettered by the consumer (the audit + // trail is useful even for legitimate races so operators can spot + // systematic Jetstream-vs-PDS skew); the distinct reason code keeps + // them queryable separately. + throw new PdsVerificationError( + "RECORD_NOT_FOUND", + `PDS returned 404 for ${currentUrl}`, + 404, + ); + } + if (!response.ok) { + throw new PdsVerificationError( + "PDS_HTTP_ERROR", + `PDS returned ${response.status} for ${currentUrl}`, + response.status, + ); + } + + return await readCarBody(response, maxBytes, timeoutMs); + } } finally { clearTimeout(timer); } +} - if (response.status === 404) { - // Distinct from a generic 4xx. The publisher may have deleted the - // record between Jetstream emitting and us fetching, which is the - // common cause; other 4xx (auth, bad request) suggest programming - // errors. Both end up dead-lettered by the consumer (the audit trail - // is useful even for legitimate races so operators can spot - // systematic Jetstream-vs-PDS skew); the distinct reason code keeps - // them queryable separately. - throw new PdsVerificationError("RECORD_NOT_FOUND", `PDS returned 404 for ${url}`, 404); - } - if (!response.ok) { - throw new PdsVerificationError( - "PDS_HTTP_ERROR", - `PDS returned ${response.status} for ${url}`, - response.status, - ); - } - +async function readCarBody( + response: Response, + maxBytes: number, + timeoutMs: number, +): Promise { // Buffer the body up to the size limit. Don't trust Content-Length; a // hostile or buggy PDS could under-report and stream more bytes than // advertised. @@ -175,8 +344,7 @@ async function fetchCar( const chunks: Uint8Array[] = []; let total = 0; try { - // biome-ignore lint/correctness/noConstantCondition: drains the stream - while (true) { + for (;;) { const { done, value } = await reader.read(); if (done) break; total += value.byteLength; @@ -193,7 +361,18 @@ async function fetchCar( await reader.cancel().catch(() => { /* swallow — we already have a primary error to surface */ }); - throw err; + // Our own errors (too-large) carry their classification; pass them + // through. Anything else — a dropped socket, or the wall-clock deadline + // aborting a slow-drip body mid-download — is transient: retry. + if (err instanceof PdsVerificationError) throw err; + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + err instanceof Error && err.name === "AbortError" + ? `PDS body read aborted after ${timeoutMs}ms` + : `PDS stream failed mid-download: ${err instanceof Error ? err.message : String(err)}`, + undefined, + err, + ); } finally { reader.releaseLock(); } diff --git a/apps/aggregator/src/records-consumer.ts b/apps/aggregator/src/records-consumer.ts index eb2c0af03e..e44fe5e23d 100644 --- a/apps/aggregator/src/records-consumer.ts +++ b/apps/aggregator/src/records-consumer.ts @@ -43,6 +43,7 @@ import { PublisherProfile, PublisherVerification, } from "@emdash-cms/registry-lexicons"; +import { cloudflareDohResolver, type DnsResolver } from "emdash/security/ssrf"; import { createD1DidDocCache, DidResolver } from "./did-resolver.js"; import type { RecordsJob } from "./env.js"; @@ -63,6 +64,12 @@ export interface ConsumerDeps { db: D1Database; resolver: DidResolver; fetch?: typeof fetch; + /** + * SSRF-safe hostname resolver handed to `fetchAndVerifyRecord` so it can + * validate the publisher-controlled PDS endpoint (and any redirect) before + * connecting. Production wires `cloudflareDohResolver`. + */ + resolveHostname?: DnsResolver; now?: () => Date; /** * Optional override for the PDS-verification step. Used by tests to inject @@ -78,6 +85,7 @@ export interface ConsumerDeps { rkey: string; publicKey: import("@atcute/crypto").PublicKey; fetch?: typeof fetch; + resolveHostname?: DnsResolver; }) => Promise; } @@ -85,7 +93,7 @@ export interface ConsumerDeps { * don't need to import workerd types. */ export interface MessageController { ack(): void; - retry(): void; + retry(options?: { delaySeconds?: number }): void; } /** Subset of a `MessageBatch`. Workers' real batch object satisfies this. */ @@ -102,6 +110,7 @@ export type DeadLetterReason = | "RESPONSE_TOO_LARGE" | "INVALID_PROOF" | "PDS_HTTP_ERROR" + | "PDS_ADDRESS_BLOCKED" // structural checks (consumer-enforced) | "LEXICON_VALIDATION_FAILED" | "RKEY_MISMATCH" @@ -241,7 +250,12 @@ export async function processMessage( } catch (err) { if (err instanceof PdsVerificationError) { if (isTransient(err.reason, err.status)) { - controller.retry(); + // A transient DNS-resolution failure carries a re-delivery delay so + // retries span the DNS-propagation window instead of firing back to + // back; other transient failures (network, 5xx, timeout) retry now. + controller.retry( + err.retryAfterSeconds !== undefined ? { delaySeconds: err.retryAfterSeconds } : undefined, + ); return; } // Compute the mapped reason in its own try/catch — `mapPdsReason` @@ -311,6 +325,7 @@ async function verifyAndIngest(job: RecordsJob, deps: ConsumerDeps): Promise>; try { policy = await resolveRequestLabelerPolicy(env, request); } catch (err) { - if (!(err instanceof XRPCError)) throw err; - const errorResponse = err.toResponse(); + // Both branches take the same wrapper as a normal response so the + // CORS + `no-store` cache invariant holds on the error path too — an + // unwrapped throw would escape to workerd's bare 500, dropping both + // and leaving a takedown-relevant response cacheable. + const errorResponse = err instanceof XRPCError ? err.toResponse() : internalErrorResponse(err); const headers = new Headers(errorResponse.headers); headers.set("cache-control", NO_STORE); applyCorsHeaders(headers); diff --git a/apps/aggregator/test/pds-verify.test.ts b/apps/aggregator/test/pds-verify.test.ts index 68c2055d10..f3d62b2330 100644 --- a/apps/aggregator/test/pds-verify.test.ts +++ b/apps/aggregator/test/pds-verify.test.ts @@ -15,12 +15,21 @@ */ import { P256PublicKey, P256PrivateKeyExportable } from "@atcute/crypto"; +import type { DnsResolver } from "emdash/security/ssrf"; import { beforeAll, describe, expect, it } from "vitest"; import { fetchAndVerifyRecord, isTransient, PdsVerificationError } from "../src/pds-verify.js"; const TEST_DID = "did:plc:test00000000000000000000"; const TEST_PDS = "https://pds.test.example"; +const TEST_PDS_HOST = "pds.test.example"; +/** A public, non-reserved address so the default resolver stub passes the + * SSRF blocklist. Real production uses `cloudflareDohResolver`. */ +const PUBLIC_IP = "93.184.216.34"; + +/** Default resolver stub: every hostname maps to one public address. Tests that + * exercise the SSRF boundary pass their own. */ +const publicResolver: DnsResolver = async () => [PUBLIC_IP]; let publicKey: P256PublicKey; @@ -44,6 +53,8 @@ function buildOpts(overrides: { fetch: typeof fetch; timeoutMs?: number; maxResponseBytes?: number; + pds?: string; + resolveHostname?: DnsResolver; }) { return { pds: TEST_PDS, @@ -51,6 +62,7 @@ function buildOpts(overrides: { collection: "com.emdashcms.experimental.package.profile", rkey: "demo", publicKey, + resolveHostname: publicResolver, ...overrides, }; } @@ -149,6 +161,181 @@ describe("fetchAndVerifyRecord — HTTP path", () => { }); }); +describe("fetchAndVerifyRecord — SSRF egress hardening", () => { + const rejectFetch: typeof fetch = () => { + throw new Error("fetch must not be reached when the URL is blocked"); + }; + + it("fails closed with PDS_ADDRESS_BLOCKED when no resolver is provided", async () => { + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: rejectFetch, resolveHostname: undefined })), + ); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + }); + + it("rejects a non-HTTPS PDS endpoint with PDS_ADDRESS_BLOCKED", async () => { + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: rejectFetch, pds: "http://pds.test.example" })), + ); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + }); + + it("maps an empty DNS answer to the transient PDS_NETWORK_ERROR", async () => { + // An empty resolver answer (NXDOMAIN / NOERROR-NODATA / CNAME-only) is a + // host mid-propagation, not a disallowed address — it must retry, not + // dead-letter. A genuinely-gone host dead-letters via retry exhaustion. + // Keyed on the shared helper's `Hostname resolved to no addresses` wording; + // this breaks if that wording changes rather than silently mis-classifying. + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: rejectFetch, resolveHostname: async () => [] })), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(isTransient(err.reason, err.status)).toBe(true); + // Carries a re-delivery delay so retries span the DNS-propagation window. + expect(err.retryAfterSeconds).toBeGreaterThan(0); + }); + + it("maps a resolver infrastructure failure to the transient PDS_NETWORK_ERROR", async () => { + // A throwing resolver models a DoH network error / SERVFAIL / timeout — + // transient infrastructure, not a disallowed address. It must NOT dead-letter. + // If the shared helper's `Could not resolve hostname:` wording ever changes, + // this assertion breaks instead of silently mis-classifying as permanent. + const throwingResolver: DnsResolver = () => Promise.reject(new Error("DoH 503")); + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: rejectFetch, resolveHostname: throwingResolver })), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(isTransient(err.reason, err.status)).toBe(true); + expect(err.retryAfterSeconds).toBeGreaterThan(0); + }); + + it("rejects when the endpoint resolves to a private address", async () => { + const err = await captureRejection( + fetchAndVerifyRecord( + buildOpts({ fetch: rejectFetch, resolveHostname: async () => ["10.0.0.5"] }), + ), + ); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + }); + + it("re-validates a redirect target and blocks one resolving to a reserved address", async () => { + // Initial host resolves public; the redirect target resolves to the + // cloud-metadata address. The private target must never be fetched. + const resolver: DnsResolver = async (host) => + host === TEST_PDS_HOST ? [PUBLIC_IP] : ["169.254.169.254"]; + let evilFetched = false; + const redirectModes: RequestInit["redirect"][] = []; + const fetchImpl: typeof fetch = (input, init) => { + redirectModes.push(init?.redirect); + const href = + typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (href.includes(TEST_PDS_HOST)) { + return Promise.resolve( + new Response(null, { + status: 302, + headers: { location: "https://metadata.evil.example/x" }, + }), + ); + } + evilFetched = true; + return Promise.resolve(new Response(new Uint8Array([0]), { status: 200 })); + }; + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, resolveHostname: resolver })), + ); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + expect(evilFetched).toBe(false); + // `redirect: "manual"` is load-bearing: it's what forces our own per-hop + // re-validation instead of fetch auto-following unchecked. + expect(redirectModes).toEqual(["manual"]); + }); + + it("follows a redirect to a public target under redirect:manual on every hop", async () => { + // Both hosts resolve public; the redirect is followed to the second host, + // which returns garbage bytes (verifyRecord then rejects). The point is + // that each hop was issued with `redirect: "manual"`. + const redirectModes: RequestInit["redirect"][] = []; + const fetchImpl: typeof fetch = (input, init) => { + redirectModes.push(init?.redirect); + const href = + typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (href.includes(TEST_PDS_HOST)) { + return Promise.resolve( + new Response(null, { + status: 302, + headers: { location: "https://mirror.test.example/x" }, + }), + ); + } + return Promise.resolve(new Response(new Uint8Array([1, 2, 3]), { status: 200 })); + }; + await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, resolveHostname: publicResolver })), + ); + expect(redirectModes).toEqual(["manual", "manual"]); + }); + + it("rejects a malformed redirect Location with PDS_ADDRESS_BLOCKED", async () => { + // `http://` has an empty authority — `new URL` throws. A hostile publisher + // controls this header, so the parse failure must be a blocked redirect, + // not an escaping UNEXPECTED_ERROR, and the next hop must not be fetched. + let hops = 0; + const fetchImpl: typeof fetch = () => { + hops += 1; + return Promise.resolve(new Response(null, { status: 302, headers: { location: "http://" } })); + }; + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + expect(hops).toBe(1); + }); + + it("rejects with PDS_ADDRESS_BLOCKED when redirects exceed the hop limit", async () => { + // Every hop redirects to a fresh public host. After MAX_PDS_REDIRECTS + // followed hops the next redirect is refused (permanent — dead-letter). + let hops = 0; + const fetchImpl: typeof fetch = () => { + hops += 1; + return Promise.resolve( + new Response(null, { + status: 302, + headers: { location: `https://hop-${hops}.test.example/x` }, + }), + ); + }; + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, resolveHostname: publicResolver })), + ); + expect(err.reason).toBe("PDS_ADDRESS_BLOCKED"); + expect(err.message).toMatch(/exceeded 3 redirects/); + // Initial hop + 3 followed redirects are fetched; the 4th is refused + // before a fetch is issued. + expect(hops).toBe(4); + }); + + it("bounds a slow-drip body by the wall-clock deadline", async () => { + // One byte then silence; the read only unblocks when the deadline aborts + // the fetch signal, which errors the stream. + const fetchImpl: typeof fetch = (_input, init) => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + init?.signal?.addEventListener("abort", () => { + controller.error(new DOMException("aborted", "AbortError")); + }); + }, + }); + return Promise.resolve(new Response(stream, { status: 200 })); + }; + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, timeoutMs: 30 })), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(err.message).toMatch(/aborted after 30ms/); + // A non-resolution transient retries immediately — no propagation delay. + expect(err.retryAfterSeconds).toBeUndefined(); + }); +}); + describe("isTransient policy", () => { it("network errors retry", () => { expect(isTransient("PDS_NETWORK_ERROR", undefined)).toBe(true); diff --git a/apps/aggregator/test/read-api.test.ts b/apps/aggregator/test/read-api.test.ts index 22f8dccc6c..c76a9a6bf3 100644 --- a/apps/aggregator/test/read-api.test.ts +++ b/apps/aggregator/test/read-api.test.ts @@ -14,7 +14,7 @@ import { NSID } from "@emdash-cms/registry-lexicons"; import { applyD1Migrations, env, SELF } from "cloudflare:test"; -import { beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; interface TestEnv { DB: D1Database; @@ -1146,6 +1146,44 @@ describe("XRPC dispatcher", () => { }); }); +describe("XRPC dispatcher — unexpected policy-resolution failure", () => { + // A non-XRPC throw before dispatch (here: a D1 error because `labelers` + // is gone) must still take the CORS + `no-store` wrapper, not escape to + // workerd's bare 500. Capture the table's schema so the shared beforeEach + // (`DELETE FROM labelers`) keeps working after a test drops it. + let labelersSchema: string; + beforeAll(async () => { + const row = await testEnv.DB.prepare( + "SELECT sql FROM sqlite_master WHERE type='table' AND name='labelers'", + ).first<{ sql: string }>(); + labelersSchema = row!.sql; + }); + afterEach(async () => { + const exists = await testEnv.DB.prepare( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='labelers'", + ).first(); + if (!exists) await testEnv.DB.prepare(labelersSchema).run(); + }); + + it("wraps a non-XRPC failure in a 500 carrying CORS + no-store and no leaked internals", async () => { + await seedPackage({ slug: "demo" }); + // Resolving the default policy runs `SELECT did FROM labelers`; dropping + // the table makes that throw a non-XRPC D1 error before dispatch. + await testEnv.DB.prepare("DROP TABLE labelers").run(); + + const res = await SELF.fetch( + `https://test/xrpc/${NSID.aggregatorGetPackage}?did=${DID_A}&slug=demo`, + ); + expect(res.status).toBe(500); + expect(res.headers.get("cache-control")).toBe("private, no-store"); + expect(res.headers.get("access-control-allow-origin")).toBe("*"); + const body = (await res.json()) as { error: string; message?: string }; + expect(body.error).toBe("InternalServerError"); + // No internal detail (SQL text, table name, stack) leaks to the client. + expect(JSON.stringify(body)).not.toMatch(/labelers|no such table|SQL|SELECT/i); + }); +}); + describe("getPublisherVerification", () => { it("returns the verification claims naming a DID as subject, newest first", async () => { await seedVerification({ diff --git a/apps/aggregator/test/records-consumer.test.ts b/apps/aggregator/test/records-consumer.test.ts index e5609426c4..f5980d3b31 100644 --- a/apps/aggregator/test/records-consumer.test.ts +++ b/apps/aggregator/test/records-consumer.test.ts @@ -543,15 +543,20 @@ class MapDidDocCache implements DidDocCache { class FakeMessage implements MessageController { acked = 0; retried = 0; + retryDelaySeconds: number | undefined = undefined; ack() { this.acked += 1; } - retry() { + retry(options?: { delaySeconds?: number }) { this.retried += 1; + this.retryDelaySeconds = options?.delaySeconds; } } -function buildDeps(opts: { fetch: typeof fetch }): { +function buildDeps(opts: { + fetch: typeof fetch; + resolveHostname?: ConsumerDeps["resolveHostname"]; +}): { deps: ConsumerDeps; cache: MapDidDocCache; } { @@ -564,7 +569,15 @@ function buildDeps(opts: { fetch: typeof fetch }): { now: () => NOW, }); return { - deps: { db: testEnv.DB, resolver, fetch: opts.fetch, now: () => NOW }, + deps: { + db: testEnv.DB, + resolver, + fetch: opts.fetch, + // pds.test.example resolves public so the SSRF egress guard in + // `fetchAndVerifyRecord` passes and these tests reach the fetch stub. + resolveHostname: opts.resolveHostname ?? (async () => ["93.184.216.34"]), + now: () => NOW, + }, cache, }; } @@ -608,6 +621,8 @@ describe("processMessage dispatcher", () => { expect(msg.retried).toBe(1); expect(msg.acked).toBe(0); expect(await deadLetterCount()).toBe(0); + // A 5xx retries immediately — the propagation delay is DNS-only. + expect(msg.retryDelaySeconds).toBeUndefined(); }); it("retries on a network error", async () => { @@ -621,6 +636,45 @@ describe("processMessage dispatcher", () => { expect(msg.retried).toBe(1); expect(await deadLetterCount()).toBe(0); + expect(msg.retryDelaySeconds).toBeUndefined(); + }); + + it("retries when the SSRF host resolver fails (transient DoH outage, not dead-lettered)", async () => { + const { deps, cache } = buildDeps({ + // Reached only if the resolver somehow passed — it must not. + fetch: () => Promise.reject(new Error("fetch must not run when resolution fails")), + resolveHostname: () => Promise.reject(new Error("DoH 503")), + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + + await processMessage(jobFor(DID_A, NSID.packageProfile, "demo"), msg, deps); + + expect(msg.retried).toBe(1); + expect(msg.acked).toBe(0); + expect(await deadLetterCount()).toBe(0); + // Re-delivery is delayed so retries span the DNS-propagation window + // instead of burning max_retries before the record's host resolves. + expect(msg.retryDelaySeconds).toBeGreaterThan(0); + }); + + it("acks and dead-letters PDS_ADDRESS_BLOCKED when the endpoint resolves to a private address", async () => { + const { deps, cache } = buildDeps({ + fetch: () => Promise.reject(new Error("fetch must not run for a blocked address")), + resolveHostname: async () => ["10.0.0.5"], + }); + cache.seed(DID_A); + const msg = new FakeMessage(); + + await processMessage(jobFor(DID_A, NSID.packageProfile, "demo"), msg, deps); + + expect(msg.acked).toBe(1); + expect(msg.retried).toBe(0); + expect(await deadLetterCount()).toBe(1); + const row = await testEnv.DB.prepare(`SELECT reason FROM dead_letters`).first<{ + reason: string; + }>(); + expect(row?.reason).toBe("PDS_ADDRESS_BLOCKED"); }); it("forensics + acks on garbage CAR bytes (verifyRecord rejects → INVALID_PROOF)", async () => { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 47bd5d6a4c..5a1c40082c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -392,6 +392,9 @@ importers: '@emdash-cms/registry-moderation': specifier: workspace:* version: link:../../packages/registry-moderation + emdash: + specifier: workspace:* + version: link:../../packages/core devDependencies: '@cloudflare/vite-plugin': specifier: 'catalog:'