diff --git a/apps/labeler/src/access-auth.ts b/apps/labeler/src/access-auth.ts index c9f2aa22f2..4db9b83b92 100644 --- a/apps/labeler/src/access-auth.ts +++ b/apps/labeler/src/access-auth.ts @@ -110,7 +110,10 @@ export async function verifyAccessRequest( } const sub = payload.sub; - if (typeof sub !== "string" || sub.length === 0) + // Cloudflare Access sets sub: "" for non-identity (service-token) JWTs, so + // only reject a non-string sub here; the human path below additionally + // requires it to be non-empty. + if (typeof sub !== "string") throw new AccessAuthError("invalid-token", "Access assertion is missing sub"); const commonName = payload.common_name; @@ -120,6 +123,8 @@ export async function verifyAccessRequest( if (typeof commonName === "string" && commonName.length > 0) { identity = { kind: "service", commonName, sub, roles: [] }; } else if (typeof email === "string" && email.length > 0) { + if (sub.length === 0) + throw new AccessAuthError("invalid-token", "Access assertion is missing sub"); identity = { kind: "human", email, sub, roles: [] }; } else { throw new AccessAuthError( @@ -129,7 +134,13 @@ export async function verifyAccessRequest( } const principal = identity.kind === "service" ? identity.commonName : identity.email; - const principals = new Set([principal, ...groupPrincipals(payload.groups)]); + // Group membership authorizes humans only. A service token carries no IdP + // identity — it is authorized solely by common_name — so a `custom.groups` + // on a service token must never confer a role. + const principals = new Set([ + principal, + ...(identity.kind === "human" ? groupPrincipals(payload.custom) : []), + ]); const roles: OperatorRole[] = []; if (config.admins.some((admin) => principals.has(admin))) roles.push("admin"); if (config.reviewers.some((reviewer) => principals.has(reviewer))) roles.push("reviewer"); @@ -137,7 +148,12 @@ export async function verifyAccessRequest( return { ...identity, roles }; } -function groupPrincipals(groups: unknown): readonly string[] { +// Cloudflare Access surfaces IdP group membership inside the verified token's +// `custom` object (as a `groups` claim configured on the SAML/OIDC integration), +// never as a top-level claim. Read the group principals from there. +function groupPrincipals(custom: unknown): readonly string[] { + if (!isRecord(custom)) return []; + const groups = custom.groups; if (!Array.isArray(groups)) return []; return groups.filter((entry): entry is string => typeof entry === "string" && entry.length > 0); } diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index ee42c02e59..18395e7b7b 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -32,6 +32,7 @@ import { PlcDidDocumentResolver, } from "@atcute/identity-resolver"; import type { LabelSigner } from "@emdash-cms/registry-moderation"; +import { cloudflareDohResolver, type DnsResolver } from "emdash/security/ssrf"; import { AssessmentDispatchError, @@ -94,6 +95,9 @@ export interface DiscoveryConsumerDeps { * (assessment-dispatch.ts). */ assessmentWorkflow: AssessmentWorkflowBinding; fetch?: typeof fetch; + /** Resolves each PDS hop's hostname for the SSRF egress guard; defaults to + * the DoH resolver used by artifact acquisition. */ + resolveHostname?: DnsResolver; now?: () => Date; /** * Optional override for the record-verification step. Used by tests to @@ -107,6 +111,7 @@ export interface DiscoveryConsumerDeps { cid: string; didDocumentResolver: DidDocumentResolverLike; fetch?: typeof fetch; + resolveHostname?: DnsResolver; }) => Promise; /** Override for the delete-path absence check; defaults to * `confirmRecordAbsent`. Returns `true` when the record is verifiably gone. */ @@ -114,14 +119,17 @@ export interface DiscoveryConsumerDeps { uri: string; didDocumentResolver: DidDocumentResolverLike; fetch?: typeof fetch; + resolveHostname?: DnsResolver; }) => Promise; } /** Subset of `cloudflare:workers` `Message` we use; defining inline so tests * don't need to import workerd types. */ export interface MessageController { + /** 1 on first delivery, incremented on each redelivery. */ + readonly attempts: number; ack(): void; - retry(): void; + retry(options?: { delaySeconds?: number }): void; } /** Subset of a `MessageBatch`. Workers' real batch object satisfies this. */ @@ -137,6 +145,7 @@ export type DiscoveryDeadLetterReason = | "RESPONSE_TOO_LARGE" | "INVALID_PROOF" | "PDS_HTTP_ERROR" + | "PDS_HOST_BLOCKED" | "DELETE_RECORD_PRESENT" | RecordVerificationFailureReason | "UNEXPECTED_ERROR"; @@ -215,6 +224,7 @@ export async function processDiscoveryMessage( uri, didDocumentResolver: deps.didDocumentResolver, ...(deps.fetch ? { fetch: deps.fetch } : {}), + ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); if (!absent) { await writeDeadLetter( @@ -293,7 +303,10 @@ async function classifyDiscoveryError( } if (err instanceof PdsVerificationError) { if (isTransient(err.reason, err.status)) { - controller.retry(); + // Back off so a DNS-propagation or PDS blip has time to clear before + // max_retries is exhausted; an immediate re-delivery would burn every + // attempt in a couple of batches. + controller.retry({ delaySeconds: transientRetryDelaySeconds(controller.attempts) }); return; } let mapped: DiscoveryDeadLetterReason; @@ -349,6 +362,7 @@ async function verifyAndCreateRun( cid: job.cid, didDocumentResolver: deps.didDocumentResolver, ...(deps.fetch ? { fetch: deps.fetch } : {}), + ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); await createSubject(deps.db, { @@ -508,6 +522,20 @@ function jobUri(job: DiscoveryJob): string { return `at://${job.did}/${job.collection}/${job.rkey}`; } +/** Capped exponential backoff for a transient PDS failure. The queue has no + * `retry_delay`, so a bare `retry()` re-delivers in the next batch — with only + * `max_retries` attempts, an immediate loop can exhaust every attempt before + * ordinary DNS propagation completes (the empty-answer/resolver-failure path + * routes propagation lag here). Delaying each retry gives the host time to + * resolve without an unbounded backlog. */ +const RETRY_BASE_DELAY_SECONDS = 15; +const RETRY_MAX_DELAY_SECONDS = 300; + +function transientRetryDelaySeconds(attempts: number): number { + const exponent = Math.max(0, attempts - 1); + return Math.min(RETRY_BASE_DELAY_SECONDS * 2 ** exponent, RETRY_MAX_DELAY_SECONDS); +} + /** * Translate a permanent `PdsVerificationError.reason` to its * `DiscoveryDeadLetterReason` counterpart. Transient reasons @@ -521,6 +549,7 @@ function mapPdsReason(reason: VerificationFailureReason): DiscoveryDeadLetterRea case "RESPONSE_TOO_LARGE": case "INVALID_PROOF": case "PDS_HTTP_ERROR": + case "PDS_HOST_BLOCKED": return reason; case "PDS_NETWORK_ERROR": throw new Error( @@ -554,6 +583,9 @@ async function createProductionDiscoveryDeps(env: Env): Promise { + let parsed: URL; + try { + parsed = new URL(url); + } catch { + throw new PdsVerificationError("PDS_HOST_BLOCKED", `PDS URL is not a valid URL: ${url}`); + } + if (parsed.protocol !== "https:") { + throw new PdsVerificationError( + "PDS_HOST_BLOCKED", + `PDS URL must use https, got ${parsed.protocol}`, + ); + } + try { + await resolveAndValidateExternalUrl(url, { resolver: resolveHostname }); + } catch (err) { + if (err instanceof SsrfError) { + // Two transient resolver outcomes: the resolver itself failed (DoH + // network error, SERVFAIL, timeout), or it returned an empty answer + // (NXDOMAIN / NODATA, which a host mid-DNS-propagation produces). + // Both retry rather than permanently dead-lettering a legitimate + // record. A true private/reserved-address or scheme block is permanent. + if ( + err.message.startsWith(SSRF_RESOLVER_FAILURE_PREFIX) || + err.message === SSRF_EMPTY_ANSWER_MESSAGE + ) { + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + `PDS host resolution failed: ${err.message}`, + undefined, + err, + ); + } + throw new PdsVerificationError( + "PDS_HOST_BLOCKED", + `PDS host rejected: ${err.message}`, + undefined, + err, + ); + } + throw err; + } +} + +/** + * Fetch `initialUrl`, following redirects manually so every hop — the + * publisher-controlled PDS endpoint and any `Location` it names — is + * re-validated against the SSRF egress rules before the request is made. A + * hop pointing at a forbidden scheme or address rejects with + * `PDS_HOST_BLOCKED`. + */ +async function fetchWithRedirectGuard( + fetchImpl: typeof fetch, + initialUrl: string, + signal: AbortSignal, + timeoutMs: number, + resolveHostname: DnsResolver, +): Promise { + let currentUrl = initialUrl; + for (let hop = 0; ; hop++) { + await assertAllowedPdsUrl(currentUrl, resolveHostname); + + let response: Response; + try { + response = await fetchImpl(currentUrl, { + signal, + 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 (response.status < 300 || response.status >= 400) return response; + + if (hop >= MAX_PDS_REDIRECTS) { + throw new PdsVerificationError( + "PDS_HTTP_ERROR", + `PDS exceeded ${MAX_PDS_REDIRECTS} redirects`, + response.status, + ); + } + const location = response.headers.get("location"); + if (location === null) { + throw new PdsVerificationError( + "PDS_HTTP_ERROR", + `PDS redirect ${response.status} without a location header`, + response.status, + ); + } + try { + currentUrl = new URL(location, currentUrl).toString(); + } catch { + throw new PdsVerificationError( + "PDS_HOST_BLOCKED", + `PDS redirect location is not a valid URL: ${location}`, + response.status, + ); + } + } +} + async function fetchCar( fetchImpl: typeof fetch, url: string, timeoutMs: number, maxBytes: number, + resolveHostname: DnsResolver, ): Promise { + const deadline = Date.now() + timeoutMs; 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, + response = await fetchWithRedirectGuard( + fetchImpl, + url, + controller.signal, + timeoutMs, + resolveHostname, ); } finally { clearTimeout(timer); @@ -172,12 +311,34 @@ async function fetchCar( if (!reader) { throw new PdsVerificationError("INVALID_PROOF", "PDS response body is null"); } + return readCarBody(reader, maxBytes, deadline, timeoutMs); +} + +/** + * Buffer the CAR body, bounding both its size (`maxBytes`) and its total wall + * time (`deadline`) — the header-phase abort timer is already cleared by the + * time we stream, so without this a slow-drip PDS could hold the read open + * indefinitely. Each read is raced against the remaining budget; exhausting it + * rejects as a transient `PDS_NETWORK_ERROR` so the consumer retries. + */ +async function readCarBody( + reader: ReadableStreamDefaultReader, + maxBytes: number, + deadline: number, + timeoutMs: number, +): Promise { const chunks: Uint8Array[] = []; let total = 0; try { - // biome-ignore lint/correctness/noConstantCondition: drains the stream - while (true) { - const { done, value } = await reader.read(); + for (;;) { + const remaining = deadline - Date.now(); + if (remaining <= 0) { + throw new PdsVerificationError( + "PDS_NETWORK_ERROR", + `PDS body read exceeded ${timeoutMs}ms`, + ); + } + const { done, value } = await readWithDeadline(reader, remaining, timeoutMs); if (done) break; total += value.byteLength; if (total > maxBytes) { @@ -193,10 +354,11 @@ async function fetchCar( await reader.cancel().catch(() => { /* swallow — we already have a primary error to surface */ }); - // A read error after headers (socket drop, stream abort) is transient — - // re-wrap it so the consumer retries rather than dead-lettering it as an - // unexpected failure. Our own PdsVerificationErrors (e.g. too-large) - // carry their own classification and pass through. + // A read error after headers (socket drop, stream abort, stall) is + // transient — re-wrap it so the consumer retries rather than + // dead-lettering it as an unexpected failure. Our own + // PdsVerificationErrors (too-large, budget exceeded) carry their own + // classification and pass through. if (err instanceof PdsVerificationError) throw err; throw new PdsVerificationError( "PDS_NETWORK_ERROR", @@ -217,6 +379,44 @@ async function fetchCar( return out; } +/** + * Race a single `reader.read()` against a per-read timeout. Handlers are + * attached to the read promise so a read that loses the race cannot later + * surface as an unhandled rejection. + */ +function readWithDeadline( + reader: ReadableStreamDefaultReader, + remainingMs: number, + timeoutMs: number, +): Promise> { + return new Promise((resolve, reject) => { + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + reject( + new PdsVerificationError("PDS_NETWORK_ERROR", `PDS body read exceeded ${timeoutMs}ms`), + ); + }, remainingMs); + void reader.read().then( + (result) => { + if (settled) return undefined; + settled = true; + clearTimeout(timer); + resolve(result); + return undefined; + }, + (err: unknown) => { + if (settled) return undefined; + settled = true; + clearTimeout(timer); + reject(err); + return undefined; + }, + ); + }); +} + /** * Map a `PdsVerificationError.reason` to "should the consumer retry?". Network * blips and 5xx are transient; everything else is permanent (forensics + ack). diff --git a/apps/labeler/src/record-verification.ts b/apps/labeler/src/record-verification.ts index cd71a5e470..3a103863c4 100644 --- a/apps/labeler/src/record-verification.ts +++ b/apps/labeler/src/record-verification.ts @@ -22,6 +22,7 @@ import { } from "@atcute/crypto"; import { type DidDocument, getAtprotoVerificationMaterial, getPdsEndpoint } from "@atcute/identity"; import { type Did, isDid } from "@atcute/lexicons/syntax"; +import type { DnsResolver } from "emdash/security/ssrf"; import { fetchAndVerifyRecord, @@ -77,6 +78,9 @@ export interface FetchAndVerifyExactRecordOptions { didDocumentResolver: DidDocumentResolverLike; /** Inject for tests; defaults to `globalThis.fetch`. */ fetch?: typeof fetch; + /** Inject for tests; defaults to the DoH resolver used by artifact + * acquisition. Threaded into the SSRF egress guard around the PDS fetch. */ + resolveHostname?: DnsResolver; timeoutMs?: number; maxResponseBytes?: number; } @@ -178,6 +182,7 @@ async function fetchAndVerifyLatestRecord( rkey, publicKey, ...(opts.fetch ? { fetch: opts.fetch } : {}), + ...(opts.resolveHostname ? { resolveHostname: opts.resolveHostname } : {}), ...(opts.timeoutMs !== undefined ? { timeoutMs: opts.timeoutMs } : {}), ...(opts.maxResponseBytes !== undefined ? { maxResponseBytes: opts.maxResponseBytes } : {}), }); diff --git a/apps/labeler/test/access-auth.test.ts b/apps/labeler/test/access-auth.test.ts index 8618d3ae31..bb2e86a90e 100644 --- a/apps/labeler/test/access-auth.test.ts +++ b/apps/labeler/test/access-auth.test.ts @@ -28,6 +28,7 @@ interface MintOverrides { email?: string; common_name?: string; groups?: unknown; + custom?: unknown; issuer?: string; audience?: string; expiresInSeconds?: number; @@ -87,8 +88,10 @@ describe("verifyAccessRequest", () => { }); }); - it("accepts a valid service token identified by common_name", async () => { - const token = await mintToken({ common_name: "ci-automation" }, signKey); + it("accepts a valid service token identified by common_name (empty sub)", async () => { + // Cloudflare Access sets sub: "" for non-identity (service-token) JWTs and + // identifies the token via common_name (the CF-Access-Client-Id). + const token = await mintToken({ common_name: "ci-automation", sub: "" }, signKey); const identity = await verifyAccessRequest( requestWith({ "Cf-Access-Jwt-Assertion": token }), baseConfig({ admins: ["ci-automation"] }), @@ -97,14 +100,27 @@ describe("verifyAccessRequest", () => { expect(identity).toEqual({ kind: "service", commonName: "ci-automation", - sub: "user-sub-1", + sub: "", roles: ["admin"], }); }); - it("maps roles from a groups claim", async () => { + it("rejects a human token with an empty sub", async () => { + // An empty sub is only legitimate for a service token (common_name path); + // a human/email identity must carry a non-empty subject. + const token = await mintToken({ email: "admin@example.com", sub: "" }, signKey); + await expect( + verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": token }), + baseConfig(), + resolver, + ), + ).rejects.toMatchObject({ reason: "invalid-token" }); + }); + + it("maps roles from a groups claim nested under the verified custom object", async () => { const token = await mintToken( - { email: "someone@example.com", groups: ["emdash-labeler-reviewers"] }, + { email: "someone@example.com", custom: { groups: ["emdash-labeler-reviewers"] } }, signKey, ); const identity = await verifyAccessRequest( @@ -115,6 +131,35 @@ describe("verifyAccessRequest", () => { expect(identity.roles).toEqual(["reviewer"]); }); + it("ignores a top-level groups claim (Access places IdP groups under custom)", async () => { + const token = await mintToken( + { email: "someone@example.com", groups: ["emdash-labeler-admins"] }, + signKey, + ); + const identity = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": token }), + baseConfig(), + resolver, + ); + expect(identity.roles).toEqual([]); + }); + + it("maps both admin and reviewer when custom.groups lists both", async () => { + const token = await mintToken( + { + email: "someone@example.com", + custom: { groups: ["emdash-labeler-admins", "emdash-labeler-reviewers"] }, + }, + signKey, + ); + const identity = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": token }), + baseConfig(), + resolver, + ); + expect(identity.roles).toEqual(["admin", "reviewer"]); + }); + it("grants only reviewer role for a reviewer-only principal", async () => { const token = await mintToken({ email: "reviewer@example.com" }, signKey); const identity = await verifyAccessRequest( @@ -330,33 +375,39 @@ describe("verifyAccessRequest", () => { ).rejects.toMatchObject({ reason: "invalid-token" }); }); - it("ignores a groups claim of the wrong shape without crashing", async () => { - const stringShape = await mintToken( - { email: "reviewer@example.com", groups: "emdash-labeler-reviewers" }, - signKey, - ); - const numberArrayShape = await mintToken( - { email: "reviewer@example.com", groups: [1, 2, 3] }, - signKey, - ); + it("ignores malformed custom / groups shapes without crashing", async () => { + // custom itself is not an object; custom.groups is a string not an array; + // custom.groups is a non-string array; custom is absent. None should throw + // and none should leak a group principal — the reviewer role here comes + // only from the email allowlist. + const shapes: unknown[] = [ + "custom-is-a-string", + { groups: "emdash-labeler-reviewers" }, + { groups: [1, 2, 3] }, + { groups: { nested: true } }, + ]; + for (const custom of shapes) { + const token = await mintToken({ email: "reviewer@example.com", custom }, signKey); + const identity = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": token }), + baseConfig(), + resolver, + ); + expect(identity.roles).toEqual(["reviewer"]); + } - const identityA = await verifyAccessRequest( - requestWith({ "Cf-Access-Jwt-Assertion": stringShape }), - baseConfig(), - resolver, - ); - const identityB = await verifyAccessRequest( - requestWith({ "Cf-Access-Jwt-Assertion": numberArrayShape }), + const noCustom = await mintToken({ email: "reviewer@example.com" }, signKey); + const identity = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": noCustom }), baseConfig(), resolver, ); - expect(identityA.roles).toEqual(["reviewer"]); - expect(identityB.roles).toEqual(["reviewer"]); + expect(identity.roles).toEqual(["reviewer"]); }); - it("does not treat an empty-string group entry as a principal", async () => { + it("does not treat an empty-string custom group entry as a principal", async () => { const token = await mintToken( - { email: "nobody@example.com", groups: ["", "not-a-configured-group"] }, + { email: "nobody@example.com", custom: { groups: ["", "not-a-configured-group"] } }, signKey, ); // An empty-string allowlist entry can only match if an empty group leaks in as a principal. @@ -368,6 +419,40 @@ describe("verifyAccessRequest", () => { expect(identity.roles).toEqual([]); }); + it("never lets a service token gain a role from custom.groups (service = common_name only)", async () => { + // A service token carries no IdP identity, so a `custom.groups` that + // happens to match an allowlist entry must NOT confer a role — otherwise + // moving the group claim under `custom` opens a fail-open path. + const unauthorized = await mintToken( + { common_name: "unknown-service", sub: "", custom: { groups: ["emdash-labeler-admins"] } }, + signKey, + ); + const identityA = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": unauthorized }), + baseConfig(), + resolver, + ); + expect(identityA.roles).toEqual([]); + + // A service token authorized by common_name keeps exactly that role; a + // reviewer group in its custom object adds nothing. + const authorized = await mintToken( + { common_name: "ci-automation", sub: "", custom: { groups: ["emdash-labeler-reviewers"] } }, + signKey, + ); + const identityB = await verifyAccessRequest( + requestWith({ "Cf-Access-Jwt-Assertion": authorized }), + baseConfig({ admins: ["ci-automation"] }), + resolver, + ); + expect(identityB).toEqual({ + kind: "service", + commonName: "ci-automation", + sub: "", + roles: ["admin"], + }); + }); + it("never includes the raw token in a thrown error message", async () => { const token = await mintToken({ email: "admin@example.com" }, otherKey); try { diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index d194a55d49..e98706172d 100644 --- a/apps/labeler/test/console-mutation-api.test.ts +++ b/apps/labeler/test/console-mutation-api.test.ts @@ -1743,6 +1743,7 @@ describe("console mutation: automation replay and idempotency", () => { class KillSwitchMessage implements MessageController { acked = 0; retried = 0; + readonly attempts = 1; ack() { this.acked += 1; } diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index e76529aa8f..5325c4d221 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -115,11 +115,14 @@ class StubResolver implements DidDocumentResolverLike { class FakeMessage implements MessageController { acked = 0; retried = 0; + retryDelaySeconds: number | undefined; + constructor(readonly attempts = 1) {} ack() { this.acked += 1; } - retry() { + retry(options?: { delaySeconds?: number }) { this.retried += 1; + this.retryDelaySeconds = options?.delaySeconds; } } @@ -394,6 +397,8 @@ describe("processDiscoveryMessage: verification failures", () => { expect(msg.acked).toBe(1); expect(msg.retried).toBe(0); + // A permanent failure dead-letters immediately — no retry, so no delay. + expect(msg.retryDelaySeconds).toBeUndefined(); const dl = await testEnv.DB.prepare(`SELECT reason FROM dead_letters WHERE rkey = ?`) .bind(job.rkey) @@ -477,12 +482,40 @@ describe("processDiscoveryMessage: verification failures", () => { expect(msg.retried).toBe(1); expect(msg.acked).toBe(0); + // The transient retry must carry a backoff delay so DNS propagation / PDS + // blips have time to clear before max_retries is exhausted. + expect(msg.retryDelaySeconds).toBeGreaterThan(0); const dl = await testEnv.DB.prepare(`SELECT COUNT(*) AS n FROM dead_letters WHERE rkey = ?`) .bind(job.rkey) .first<{ n: number }>(); expect(dl?.n).toBe(0); }); + it("scales the transient retry backoff with delivery attempts", async () => { + const job = await jobFor({ rkey: rkey() }); + const deps = await buildDeps(); + const transient = { + ...deps, + verify: () => + Promise.reject( + new PdsVerificationError( + "PDS_NETWORK_ERROR", + "PDS host resolution failed: Hostname resolved to no addresses", + ), + ), + }; + + const first = new FakeMessage(1); + const later = new FakeMessage(4); + await processDiscoveryMessage(job, first, transient); + await processDiscoveryMessage(job, later, transient); + + expect(first.retryDelaySeconds).toBeGreaterThan(0); + // A later delivery attempt backs off longer (capped), giving slow + // propagation more time before the DLQ. + expect(later.retryDelaySeconds).toBeGreaterThan(first.retryDelaySeconds!); + }); + it("transient PDS 5xx retries, no dead letter", async () => { const job = await jobFor({ rkey: rkey() }); const deps = await buildDeps(); diff --git a/apps/labeler/test/pds-verify.test.ts b/apps/labeler/test/pds-verify.test.ts new file mode 100644 index 0000000000..bf6919e3a9 --- /dev/null +++ b/apps/labeler/test/pds-verify.test.ts @@ -0,0 +1,238 @@ +/** + * pds-verify SSRF-egress + HTTP-shaping unit tests. + * + * The PDS endpoint (and any redirect it serves) is publisher-controlled, so the + * CAR fetch is routed through the same SSRF egress guard artifact acquisition + * uses: HTTPS-only, DoH resolution with private/reserved-IP rejection, and + * per-hop redirect re-resolution. These tests inject a fake `fetch` and a fake + * `resolveHostname` so no real network or DoH round-trip happens. + * + * The crypto handoff to `@atcute/repo`'s `verifyRecord` is not exercised + * end-to-end (a valid signed CAR would re-implement what the library already + * tests, and its fixture can't load inside `@cloudflare/vitest-pool-workers`); + * reaching `verifyRecord` at all is proof the request passed the egress guard. + */ + +import { P256PrivateKeyExportable, P256PublicKey } 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 PUBLIC_IP = "93.184.216.34"; + +let publicKey: P256PublicKey; + +beforeAll(async () => { + const kp = await P256PrivateKeyExportable.createKeypair(); + publicKey = await P256PublicKey.importRaw(await kp.exportPublicKey("raw")); +}); + +const resolvePublic: DnsResolver = () => Promise.resolve([PUBLIC_IP]); + +async function captureRejection(promise: Promise): Promise { + try { + await promise; + } catch (err) { + if (err instanceof PdsVerificationError) return err; + throw err; + } + throw new Error("expected promise to reject with PdsVerificationError"); +} + +function buildOpts(overrides: { + fetch: typeof fetch; + pds?: string; + resolveHostname?: DnsResolver; + timeoutMs?: number; + maxResponseBytes?: number; +}) { + return { + pds: overrides.pds ?? TEST_PDS, + did: TEST_DID, + collection: "com.emdashcms.experimental.package.profile", + rkey: "demo", + publicKey, + resolveHostname: overrides.resolveHostname ?? resolvePublic, + fetch: overrides.fetch, + ...(overrides.timeoutMs !== undefined ? { timeoutMs: overrides.timeoutMs } : {}), + ...(overrides.maxResponseBytes !== undefined + ? { maxResponseBytes: overrides.maxResponseBytes } + : {}), + }; +} + +function redirectTo(location: string, status = 302): Response { + return new Response(null, { status, headers: { location } }); +} + +describe("fetchAndVerifyRecord — SSRF egress guard", () => { + it("rejects a non-HTTPS PDS with PDS_HOST_BLOCKED before fetching", async () => { + let called = false; + const fetchImpl: typeof fetch = () => { + called = true; + return Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + }; + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, pds: "http://pds.test.example" })), + ); + expect(err.reason).toBe("PDS_HOST_BLOCKED"); + expect(called).toBe(false); + }); + + it("rejects a PDS whose host resolves to a private address", async () => { + let called = false; + const fetchImpl: typeof fetch = () => { + called = true; + return Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + }; + const err = await captureRejection( + fetchAndVerifyRecord( + buildOpts({ fetch: fetchImpl, resolveHostname: () => Promise.resolve(["10.0.0.1"]) }), + ), + ); + expect(err.reason).toBe("PDS_HOST_BLOCKED"); + // A true address block is permanent — the consumer dead-letters it. + expect(isTransient(err.reason, err.status)).toBe(false); + expect(called).toBe(false); + }); + + it("classifies a resolver failure (DoH blip) as transient PDS_NETWORK_ERROR, not a permanent block", async () => { + let called = false; + const fetchImpl: typeof fetch = () => { + called = true; + return Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + }; + // A throwing resolver stands in for a DoH network error / SERVFAIL / + // timeout. `resolveAndValidateExternalUrl` wraps it in an SsrfError whose + // message prefix we key on; if core changes that wording this test fails + // (the mapping would silently fall through to a permanent block instead). + const err = await captureRejection( + fetchAndVerifyRecord( + buildOpts({ + fetch: fetchImpl, + resolveHostname: () => Promise.reject(new Error("DoH request timed out")), + }), + ), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + // The consumer retries transient reasons rather than dead-lettering them. + expect(isTransient(err.reason, err.status)).toBe(true); + expect(called).toBe(false); + }); + + it("classifies an empty DNS answer (NXDOMAIN/NODATA) as transient, not a permanent block", async () => { + let called = false; + const fetchImpl: typeof fetch = () => { + called = true; + return Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + }; + // An empty resolver answer is what a host mid-DNS-propagation yields. + // `resolveAndValidateExternalUrl` raises the pinned "Hostname resolved to + // no addresses" SsrfError; it must retry (so the record survives a + // temporary negative answer) rather than dead-letter. A core wording + // change would fail this test instead of silently mis-classifying. + const err = await captureRejection( + fetchAndVerifyRecord( + buildOpts({ fetch: fetchImpl, resolveHostname: () => Promise.resolve([]) }), + ), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(isTransient(err.reason, err.status)).toBe(true); + expect(called).toBe(false); + }); + + it("rejects a redirect that points at a private address (per-hop re-resolution)", async () => { + let calls = 0; + const fetchImpl: typeof fetch = () => { + calls += 1; + return Promise.resolve(redirectTo("https://internal.example/xrpc")); + }; + const resolveByHost: DnsResolver = (hostname) => + Promise.resolve(hostname === "internal.example" ? ["10.0.0.1"] : [PUBLIC_IP]); + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, resolveHostname: resolveByHost })), + ); + expect(err.reason).toBe("PDS_HOST_BLOCKED"); + // Only the first hop was fetched; the private redirect target is blocked + // before its request is made. + expect(calls).toBe(1); + }); + + it("rejects a redirect that downgrades to a non-HTTPS scheme", async () => { + const fetchImpl: typeof fetch = () => + Promise.resolve(redirectTo("http://pds.test.example/xrpc")); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_HOST_BLOCKED"); + }); + + it("stops after the redirect limit with PDS_HTTP_ERROR", async () => { + const fetchImpl: typeof fetch = () => + Promise.resolve(redirectTo("https://loop.test.example/next")); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_HTTP_ERROR"); + }); +}); + +describe("fetchAndVerifyRecord — the guard permits public hosts", () => { + it("follows an allowed HTTPS redirect and re-resolves each hop", async () => { + let calls = 0; + const fetchImpl: typeof fetch = () => { + calls += 1; + if (calls === 1) return Promise.resolve(redirectTo("https://mirror.test.example/xrpc")); + return Promise.resolve(new Response("", { status: 404 })); + }; + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + // Reaching the 404 proves the redirect was followed and both hops passed + // the egress guard. + expect(err.reason).toBe("RECORD_NOT_FOUND"); + expect(err.status).toBe(404); + expect(calls).toBe(2); + }); + + it("maps a 404 to RECORD_NOT_FOUND with status", async () => { + const fetchImpl: typeof fetch = () => Promise.resolve(new Response("", { status: 404 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("RECORD_NOT_FOUND"); + expect(err.status).toBe(404); + }); + + it("maps a 5xx to PDS_HTTP_ERROR with status", async () => { + const fetchImpl: typeof fetch = () => Promise.resolve(new Response("", { status: 503 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_HTTP_ERROR"); + expect(err.status).toBe(503); + }); + + it("surfaces a network error from an allowed host as PDS_NETWORK_ERROR", async () => { + const fetchImpl: typeof fetch = () => Promise.reject(new TypeError("connection refused")); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + }); + + it("bounds a slow-drip body read by the timeout budget", async () => { + // The stream yields one chunk then stalls forever: the next read never + // resolves, so only the per-read time budget can end it. + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + }); + const fetchImpl: typeof fetch = () => Promise.resolve(new Response(stream, { status: 200 })); + const err = await captureRejection( + fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl, timeoutMs: 40 })), + ); + expect(err.reason).toBe("PDS_NETWORK_ERROR"); + expect(err.message).toMatch(/exceeded 40ms/); + }); + + it("hands successful body bytes to verifyRecord (INVALID_PROOF on garbage)", async () => { + const garbage = new Uint8Array([1, 2, 3, 4, 5]); + const fetchImpl: typeof fetch = () => Promise.resolve(new Response(garbage, { status: 200 })); + const err = await captureRejection(fetchAndVerifyRecord(buildOpts({ fetch: fetchImpl }))); + expect(err.reason).toBe("INVALID_PROOF"); + expect(err.cause).toBeDefined(); + }); +}); diff --git a/apps/labeler/test/record-verification.test.ts b/apps/labeler/test/record-verification.test.ts index aee123cd1b..d67725b6a6 100644 --- a/apps/labeler/test/record-verification.test.ts +++ b/apps/labeler/test/record-verification.test.ts @@ -160,6 +160,7 @@ describe("fetchAndVerifyExactRecord: PDS fetch propagation", () => { cid: "bafkreiplaceholder00000000000000000000000000000000000000000", didDocumentResolver: resolver, fetch: () => Promise.resolve(new Response("", { status: 404 })), + resolveHostname: () => Promise.resolve(["93.184.216.34"]), }), ).rejects.toBeInstanceOf(PdsVerificationError); try { @@ -168,6 +169,7 @@ describe("fetchAndVerifyExactRecord: PDS fetch propagation", () => { cid: "bafkreiplaceholder00000000000000000000000000000000000000000", didDocumentResolver: resolver, fetch: () => Promise.resolve(new Response("", { status: 404 })), + resolveHostname: () => Promise.resolve(["93.184.216.34"]), }); throw new Error("expected rejection"); } catch (err) {