fix(labeler): Access custom-group roles + PDS-fetch SSRF guard (review round 1)

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,15 +134,26 @@ export async function verifyAccessRequest(
 	}
 
 	const principal = identity.kind === "service" ? identity.commonName : identity.email;
-	const principals = new Set<string>([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<string>([
+		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");
 
 	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);
 }

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,21 +111,25 @@ 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>;
 }
 
 /** 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<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,
 	};
 }