fix(labeler): route the PDS record fetch through an SSRF egress guard

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.

Author: ascorbic

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,13 +111,15 @@ export interface DiscoveryConsumerDeps {
 		cid: string;
 		didDocumentResolver: DidDocumentResolverLike;
 		fetch?: typeof fetch;
+		resolveHostname?: DnsResolver;
 	}) => Promise<VerifiedPdsRecord>;
 	/** Override for the delete-path absence check; defaults to
 	 * `confirmRecordAbsent`. Returns `true` when the record is verifiably gone. */
 	confirmDeleted?: (opts: {
 		uri: string;
 		didDocumentResolver: DidDocumentResolverLike;
 		fetch?: typeof fetch;
+		resolveHostname?: DnsResolver;
 	}) => Promise<boolean>;
 }
 
@@ -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<DiscoveryConsume
 		// bound wrapper rather than letting pds-verify.ts fall back to bare
 		// global `fetch`.
 		fetch: boundFetch,
+		// SSRF egress guard for the publisher-controlled PDS endpoint and any
+		// redirect it serves — the same DoH resolver artifact acquisition uses.
+		resolveHostname: cloudflareDohResolver,
 	};
 }
 

apps/labeler/src/pds-verify.ts

@@ -20,18 +20,28 @@
 import type { PublicKey } from "@atcute/crypto";
 import { type AtprotoDid, isDid } from "@atcute/lexicons/syntax";
 import { verifyRecord } from "@atcute/repo";
+import {
+	cloudflareDohResolver,
+	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. Each hop is independently
+ * re-validated against the SSRF egress rules before it is fetched. */
+const MAX_PDS_REDIRECTS = 3;
 
 export type VerificationFailureReason =
 	| "PDS_NETWORK_ERROR"
 	| "PDS_HTTP_ERROR"
 	| "RECORD_NOT_FOUND"
 	| "RESPONSE_TOO_LARGE"
-	| "INVALID_PROOF";
+	| "INVALID_PROOF"
+	| "PDS_HOST_BLOCKED";
 
 export class PdsVerificationError extends Error {
 	override readonly name = "PdsVerificationError";
@@ -57,6 +67,10 @@ export interface FetchAndVerifyOptions {
 	maxResponseBytes?: number;
 	/** Inject for tests; defaults to `globalThis.fetch`. */
 	fetch?: typeof fetch;
+	/** Resolves each hop's hostname so its addresses can be checked against the
+	 * private/reserved-IP blocklist before the request is made. Inject for
+	 * tests; defaults to the DoH resolver used by artifact acquisition. */
+	resolveHostname?: DnsResolver;
 }
 
 export interface VerifiedPdsRecord {
@@ -73,6 +87,7 @@ export async function fetchAndVerifyRecord(
 	const fetchImpl = opts.fetch ?? fetch;
 	const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS;
 	const maxResponseBytes = opts.maxResponseBytes ?? DEFAULT_MAX_RESPONSE_BYTES;
+	const resolveHostname = opts.resolveHostname ?? cloudflareDohResolver;
 
 	if (!isAtprotoDid(opts.did)) {
 		// Caller is expected to have validated this upstream (the resolver
@@ -84,7 +99,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,31 +132,126 @@ function buildGetRecordUrl(pds: string, did: string, collection: string, rkey: s
 	return url.toString();
 }
 
+/**
+ * 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.
+ */
+async function assertAllowedPdsUrl(url: string, resolveHostname: DnsResolver): Promise<void> {
+	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<Response> {
+	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<Uint8Array> {
+	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<Uint8Array>,
+	maxBytes: number,
+	deadline: number,
+	timeoutMs: number,
+): Promise<Uint8Array> {
 	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<Uint8Array>,
+	remainingMs: number,
+	timeoutMs: number,
+): Promise<ReadableStreamReadResult<Uint8Array>> {
+	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).