From ed43eca0996a9ed1ba9977e2b0b95af0e779935e Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:12:30 +0100 Subject: [PATCH 1/6] fix(labeler): read Access group claims from the verified custom object Cloudflare Access places IdP group membership inside the verified application token's `custom` object (a `groups` claim configured on the SAML/OIDC integration), never as a top-level `groups` claim. The role mapper read `payload.groups`, which Access never populates, so the group-only reviewer/admin allowlist granted no roles to human operators whose access is conferred by group. Read the `groups` array from the verified `payload.custom` object, guarding the `custom` record and the array shape strictly. Restrict group extraction to human identities: a service token is authorized solely by common_name and carries no IdP identity, so a `custom.groups` on a service token must never confer a role (moving the group claim under `custom` would otherwise open a fail-open path). Email/common_name resolution is otherwise unchanged. --- apps/labeler/src/access-auth.ts | 15 +++- apps/labeler/test/access-auth.test.ts | 114 +++++++++++++++++++++----- 2 files changed, 105 insertions(+), 24 deletions(-) diff --git a/apps/labeler/src/access-auth.ts b/apps/labeler/src/access-auth.ts index c9f2aa22f2..a044fcba32 100644 --- a/apps/labeler/src/access-auth.ts +++ b/apps/labeler/src/access-auth.ts @@ -129,7 +129,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 +143,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/test/access-auth.test.ts b/apps/labeler/test/access-auth.test.ts index 8618d3ae31..24f20f49ae 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; @@ -102,9 +103,9 @@ describe("verifyAccessRequest", () => { }); }); - it("maps roles from a groups claim", async () => { + 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 +116,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 +360,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 +404,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", 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", 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: "user-sub-1", + roles: ["admin"], + }); + }); + it("never includes the raw token in a thrown error message", async () => { const token = await mintToken({ email: "admin@example.com" }, otherKey); try { From 5913738d68db9c7639d0c4e7fab871aab3f1b7e7 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:12:30 +0100 Subject: [PATCH 2/6] fix(labeler): route the PDS record fetch through an SSRF egress guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The PDS endpoint is resolved from a publisher-controlled DID document, and the CAR fetch followed redirects with the default fetch policy and no scheme, DNS, or reserved-address checks. A hostile DID/PDS document could point the endpoint (or a redirect it serves) at a private/reserved address or a non-HTTPS host, turning the verifier into a blind SSRF probe. Validate every hop before fetching, sharing the DoH resolver (`cloudflareDohResolver`) that artifact acquisition uses and routing each hop through core's `resolveAndValidateExternalUrl`: HTTPS-only, DoH resolution, and private/reserved-IP rejection. Redirects are now followed manually (`redirect: "manual"`) so each `Location` is re-validated per-hop. A blocked hop rejects with the new permanent `PDS_HOST_BLOCKED` reason (dead-lettered, never retried). This shares the resolver but NOT artifact acquisition's address predicate: `fetchVerifiedResource`'s stricter `isForbiddenAddress` is not exported from @emdash-cms/registry-verification, so this path uses core's `isPrivateIp`, which blocks loopback, RFC1918, link-local/metadata, and IPv4-mapped/NAT64 forms but not CGNAT (100.64.0.0/10), benchmarking (198.18.0.0/15), or multicast/reserved (224.0.0.0/4, ff00::/8) — an accepted residual for PDS egress; closing it would mean exporting a published-package predicate. Also bound the body read by a wall-clock budget: the header-phase abort timer is cleared once headers arrive, so each streamed chunk is now raced against the remaining fetch timeout, preventing a slow-drip PDS from holding the read open indefinitely. The resolver is threaded through the discovery consumer's closure deps; the existing 404/5xx/size status classification the delete and retry paths depend on is preserved. --- apps/labeler/src/discovery-consumer.ts | 13 ++ apps/labeler/src/pds-verify.ts | 219 ++++++++++++++++-- apps/labeler/src/record-verification.ts | 5 + apps/labeler/test/pds-verify.test.ts | 202 ++++++++++++++++ apps/labeler/test/record-verification.test.ts | 2 + 5 files changed, 417 insertions(+), 24 deletions(-) create mode 100644 apps/labeler/test/pds-verify.test.ts diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index ee42c02e59..3d0f7961e7 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,6 +119,7 @@ export interface DiscoveryConsumerDeps { uri: string; didDocumentResolver: DidDocumentResolverLike; fetch?: typeof fetch; + resolveHostname?: DnsResolver; }) => Promise; } @@ -137,6 +143,7 @@ export type DiscoveryDeadLetterReason = | "RESPONSE_TOO_LARGE" | "INVALID_PROOF" | "PDS_HTTP_ERROR" + | "PDS_HOST_BLOCKED" | "DELETE_RECORD_PRESENT" | RecordVerificationFailureReason | "UNEXPECTED_ERROR"; @@ -215,6 +222,7 @@ export async function processDiscoveryMessage( uri, didDocumentResolver: deps.didDocumentResolver, ...(deps.fetch ? { fetch: deps.fetch } : {}), + ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); if (!absent) { await writeDeadLetter( @@ -349,6 +357,7 @@ async function verifyAndCreateRun( cid: job.cid, didDocumentResolver: deps.didDocumentResolver, ...(deps.fetch ? { fetch: deps.fetch } : {}), + ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); await createSubject(deps.db, { @@ -521,6 +530,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 +564,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) { + 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 +282,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 +325,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 +350,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/pds-verify.test.ts b/apps/labeler/test/pds-verify.test.ts new file mode 100644 index 0000000000..1d8554fb9d --- /dev/null +++ b/apps/labeler/test/pds-verify.test.ts @@ -0,0 +1,202 @@ +/** + * 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, 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"); + expect(called).toBe(false); + }); + + it("rejects when the resolver returns no addresses (fails closed)", async () => { + const fetchImpl: typeof fetch = () => + Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + const err = await captureRejection( + fetchAndVerifyRecord( + buildOpts({ fetch: fetchImpl, resolveHostname: () => Promise.resolve([]) }), + ), + ); + expect(err.reason).toBe("PDS_HOST_BLOCKED"); + }); + + 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) { From 791fbea06797ffcaf2d23e25b2cc6f906f20fd03 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 15:08:27 +0100 Subject: [PATCH 3/6] fix(labeler): retry transient PDS host-resolution failures The PDS SSRF egress guard mapped every `SsrfError` from `resolveAndValidateExternalUrl` to the permanent `PDS_HOST_BLOCKED` reason. But that helper raises `SsrfError` for two distinct cases: a permanent private/reserved-address or scheme block, and a transient resolver failure (DoH network error, SERVFAIL, timeout). A transient DoH blip while resolving a publisher's PDS host therefore permanently dead-lettered a legitimate record instead of retrying it. Core exposes no structured discriminator (`SsrfError` carries a single `code`), so key on the resolver-failure message prefix ("Could not resolve hostname:") to map that case to the transient `PDS_NETWORK_ERROR` reason, which the consumer retries via `isTransient`. Genuine address/scheme blocks stay `PDS_HOST_BLOCKED` (permanent). A test pins the prefix so a core wording change fails loudly rather than silently mis-classifying. --- apps/labeler/src/pds-verify.ts | 24 ++++++++++++++++++++++-- apps/labeler/test/pds-verify.test.ts | 28 +++++++++++++++++++++++++++- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/apps/labeler/src/pds-verify.ts b/apps/labeler/src/pds-verify.ts index 692731f188..0092be28a7 100644 --- a/apps/labeler/src/pds-verify.ts +++ b/apps/labeler/src/pds-verify.ts @@ -34,6 +34,13 @@ const DEFAULT_MAX_RESPONSE_BYTES = 5 * 1024 * 1024; /** Redirect hops we'll follow before giving up. Each hop is independently * re-validated against the SSRF egress rules before it is fetched. */ const MAX_PDS_REDIRECTS = 3; +/** `resolveAndValidateExternalUrl` raises `SsrfError` both for a permanent + * address/scheme block and for a transient resolver failure (DoH network error, + * SERVFAIL, timeout). Core exposes no structured discriminator, so this prefix + * — the exact wording it wraps a thrown resolver in — separates the retryable + * case from the permanent one. A test pins it so a core wording change fails + * loudly instead of silently mis-classifying. */ +const SSRF_RESOLVER_FAILURE_PREFIX = "Could not resolve hostname:"; export type VerificationFailureReason = | "PDS_NETWORK_ERROR" @@ -135,8 +142,9 @@ function buildGetRecordUrl(pds: string, did: string, collection: string, rkey: s /** * Reject a PDS URL that is not HTTPS or whose host resolves to a * private/reserved address. The publisher controls the PDS endpoint (and any - * redirect it serves), so this reuses the same DNS-aware SSRF validator that - * guards artifact acquisition rather than duplicating any address logic. + * redirect it serves), so this validates every hop through core's DNS-aware + * `resolveAndValidateExternalUrl` (private/reserved-address rejection) rather + * than duplicating any address logic. */ async function assertAllowedPdsUrl(url: string, resolveHostname: DnsResolver): Promise { let parsed: URL; @@ -155,6 +163,18 @@ async function assertAllowedPdsUrl(url: string, resolveHostname: DnsResolver): P await resolveAndValidateExternalUrl(url, { resolver: resolveHostname }); } catch (err) { if (err instanceof SsrfError) { + // A resolver failure (DoH network error, SERVFAIL, timeout) is + // transient infrastructure — retry it rather than permanently + // dead-lettering a legitimate record. A true address/scheme block is + // permanent. + if (err.message.startsWith(SSRF_RESOLVER_FAILURE_PREFIX)) { + 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}`, diff --git a/apps/labeler/test/pds-verify.test.ts b/apps/labeler/test/pds-verify.test.ts index 1d8554fb9d..9f959e34ce 100644 --- a/apps/labeler/test/pds-verify.test.ts +++ b/apps/labeler/test/pds-verify.test.ts @@ -17,7 +17,7 @@ import { P256PrivateKeyExportable, P256PublicKey } from "@atcute/crypto"; import type { DnsResolver } from "emdash/security/ssrf"; import { beforeAll, describe, expect, it } from "vitest"; -import { fetchAndVerifyRecord, PdsVerificationError } from "../src/pds-verify.js"; +import { fetchAndVerifyRecord, isTransient, PdsVerificationError } from "../src/pds-verify.js"; const TEST_DID = "did:plc:test00000000000000000000"; const TEST_PDS = "https://pds.test.example"; @@ -94,6 +94,32 @@ describe("fetchAndVerifyRecord — SSRF egress guard", () => { ), ); 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); }); From b0ca6296b45af6b5c14f593e1c120652e86c8669 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 16:08:50 +0100 Subject: [PATCH 4/6] fix(labeler): accept the empty sub Cloudflare Access sets for service tokens MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cloudflare Access issues non-identity (service-token) application JWTs with `sub: ""` and identifies the token via `common_name` (the CF-Access-Client-Id). The verifier rejected any empty `sub` before it reached the common_name branch, so every real service-token request failed `invalid-token` and the service-identity path was unreachable in production — the prior service-token tests passed only because they minted a synthetic non-empty subject. Require `sub` to be a string, but only reject an empty `sub` on the human/email path; the service path (common_name present) now proceeds with the empty subject Access actually sends. Tests use the realistic service-token shape (empty sub + common_name), and a human token with an empty sub is still rejected. --- apps/labeler/src/access-auth.ts | 7 ++++++- apps/labeler/test/access-auth.test.ts | 27 +++++++++++++++++++++------ 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/apps/labeler/src/access-auth.ts b/apps/labeler/src/access-auth.ts index a044fcba32..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( diff --git a/apps/labeler/test/access-auth.test.ts b/apps/labeler/test/access-auth.test.ts index 24f20f49ae..bb2e86a90e 100644 --- a/apps/labeler/test/access-auth.test.ts +++ b/apps/labeler/test/access-auth.test.ts @@ -88,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"] }), @@ -98,11 +100,24 @@ describe("verifyAccessRequest", () => { expect(identity).toEqual({ kind: "service", commonName: "ci-automation", - sub: "user-sub-1", + sub: "", roles: ["admin"], }); }); + 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", custom: { groups: ["emdash-labeler-reviewers"] } }, @@ -409,7 +424,7 @@ describe("verifyAccessRequest", () => { // 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", custom: { groups: ["emdash-labeler-admins"] } }, + { common_name: "unknown-service", sub: "", custom: { groups: ["emdash-labeler-admins"] } }, signKey, ); const identityA = await verifyAccessRequest( @@ -422,7 +437,7 @@ describe("verifyAccessRequest", () => { // 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", custom: { groups: ["emdash-labeler-reviewers"] } }, + { common_name: "ci-automation", sub: "", custom: { groups: ["emdash-labeler-reviewers"] } }, signKey, ); const identityB = await verifyAccessRequest( @@ -433,7 +448,7 @@ describe("verifyAccessRequest", () => { expect(identityB).toEqual({ kind: "service", commonName: "ci-automation", - sub: "user-sub-1", + sub: "", roles: ["admin"], }); }); From 748822a89965435b9879ca4496c8cef84e54683e Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 16:08:51 +0100 Subject: [PATCH 5/6] fix(labeler): retry an empty DNS answer for a PDS host MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The PDS SSRF guard mapped every non-resolver-failure `SsrfError` to the permanent `PDS_HOST_BLOCKED` reason. But `resolveAndValidateExternalUrl` also raises `SsrfError` ("Hostname resolved to no addresses") when the resolver returns an empty answer (NXDOMAIN / NODATA), which a host mid-DNS-propagation produces transiently — so a legitimate record was permanently dead-lettered on a temporary negative answer. Classify the empty-answer message as the transient `PDS_NETWORK_ERROR` too (the consumer retries it via `isTransient`; a genuinely absent host dead-letters through retry exhaustion). Keyed on a pinned message constant like the resolver-failure prefix, so a core wording change fails a test. Genuine private/reserved-address and scheme blocks stay permanent. --- apps/labeler/src/pds-verify.ts | 19 ++++++++++++++----- apps/labeler/test/pds-verify.test.ts | 18 ++++++++++++++---- 2 files changed, 28 insertions(+), 9 deletions(-) diff --git a/apps/labeler/src/pds-verify.ts b/apps/labeler/src/pds-verify.ts index 0092be28a7..44c987db81 100644 --- a/apps/labeler/src/pds-verify.ts +++ b/apps/labeler/src/pds-verify.ts @@ -41,6 +41,11 @@ const MAX_PDS_REDIRECTS = 3; * case from the permanent one. A test pins it so a core wording change fails * loudly instead of silently mis-classifying. */ const SSRF_RESOLVER_FAILURE_PREFIX = "Could not resolve hostname:"; +/** `resolveAndValidateExternalUrl` raises this exact `SsrfError` when the + * resolver returns no addresses (NXDOMAIN / NODATA). A host mid-DNS-propagation + * produces a temporary negative answer, so this is retryable — not a permanent + * block. Pinned by a test so a core wording change fails loudly. */ +const SSRF_EMPTY_ANSWER_MESSAGE = "Hostname resolved to no addresses"; export type VerificationFailureReason = | "PDS_NETWORK_ERROR" @@ -163,11 +168,15 @@ async function assertAllowedPdsUrl(url: string, resolveHostname: DnsResolver): P await resolveAndValidateExternalUrl(url, { resolver: resolveHostname }); } catch (err) { if (err instanceof SsrfError) { - // A resolver failure (DoH network error, SERVFAIL, timeout) is - // transient infrastructure — retry it rather than permanently - // dead-lettering a legitimate record. A true address/scheme block is - // permanent. - if (err.message.startsWith(SSRF_RESOLVER_FAILURE_PREFIX)) { + // 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}`, diff --git a/apps/labeler/test/pds-verify.test.ts b/apps/labeler/test/pds-verify.test.ts index 9f959e34ce..bf6919e3a9 100644 --- a/apps/labeler/test/pds-verify.test.ts +++ b/apps/labeler/test/pds-verify.test.ts @@ -123,15 +123,25 @@ describe("fetchAndVerifyRecord — SSRF egress guard", () => { expect(called).toBe(false); }); - it("rejects when the resolver returns no addresses (fails closed)", async () => { - const fetchImpl: typeof fetch = () => - Promise.resolve(new Response(new Uint8Array([1]), { status: 200 })); + 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_HOST_BLOCKED"); + 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 () => { From ca40128b06f620deffc65024dfdfe3e323145ca0 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 17:56:31 +0100 Subject: [PATCH 6/6] fix(labeler): back off transient PDS retries so DNS propagation can complete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The discovery consumer retried transient PDS failures with a bare `retry()`, and the queue consumer has no `retry_delay`, so Cloudflare Queues re-delivered immediately in the next batch. With only max_retries attempts, ordinary DNS propagation (now routed here via the empty-answer / resolver-failure transient classification) could exhaust every attempt before the host resolved — after which the record dead-lettered as UNEXPECTED_ERROR. Pass a capped exponential `delaySeconds` (15s, doubling per delivery attempt, capped at 300s) when retrying a transient PDS failure, giving propagation time without an unbounded backlog. The delay applies only to the transient PDS retry path; permanent failures still dead-letter immediately with no retry, and the other retry paths (label issuance, workflow dispatch, DID resolution) are unchanged. --- apps/labeler/src/discovery-consumer.ts | 23 ++++++++++-- .../labeler/test/console-mutation-api.test.ts | 1 + apps/labeler/test/discovery-consumer.test.ts | 35 ++++++++++++++++++- 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 3d0f7961e7..18395e7b7b 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -126,8 +126,10 @@ export interface DiscoveryConsumerDeps { /** 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. */ @@ -301,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; @@ -517,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 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();