fix(labeler): notification contact-resolution, unsubscribe, and verification-trust fixes (review round 1) (#2116)

* fix(labeler): honor RFC 8058 one-click unsubscribe query param

A one-click unsubscribe POST carries the recipient hash (c) in the
List-Unsubscribe header URL's query string and only List-Unsubscribe=One-Click
in the body. The POST parser read c from the form body only, so a one-click
request returned success without suppressing the recipient. Read c (and the
confirm token) from the body first, falling back to the query.

* fix(labeler): resolve package contacts for release-subject notices

A release subject's rkey is slug:version, but contactTargetFromUri passed the
whole rkey as the package slug, so getPackage missed the package and its
security[]/authors[] contacts were skipped in favor of the publisher-profile
fallback. Parse the parent package slug (before the version delimiter) so a
release notice reaches the package's contacts; a package rkey and a malformed
rkey both degrade safely.

* fix(labeler): alert operators when a takedown notice has no contact

An emergency takedown issuance that resolves no publisher contact recorded only
a generic undeliverable notification row, so the signal that manual outreach is
needed was easy to miss. Emit a dedicated takedown-no-contact operational event
(severity high) plus its operator-alert outbox row on that path. A takedown
retract and a takedown that does resolve a contact emit nothing, and a deduped
replay does not re-emit.

* fix(labeler): gate double-opt-in bypass on trusted, current verification

A publisher's verification claim upgraded a contact past double opt-in on the
strength of any in-force claim. Verification claims are self-assertable — the
aggregator indexes any issuer, including self-issued ones — so a publisher could
name a victim address in its profile, self-issue a claim, and have an
unconfirmed notice sent there. Upgrade only on a claim whose issuer is in the
configured trust set, that is unexpired, and whose bound displayName still
matches the publisher's current identity. No issuer is trusted by default, so
every address falls back to double opt-in.

The publisher view carries a current displayName but no handle (the aggregator's
identity-event ingestion is unbuilt), so the binding checks displayName only;
handle-binding is deferred until the view carries a handle.

* fix(labeler): make takedown no-contact alert recoverable

The takedown-no-contact operator alert was emitted only on the first pass's
send outcome, after the undeliverable notifications row had committed. A
transient failure of that fire-and-forget event write lost the alert
permanently: a deferred replay dedups on the existing notifications row and
never reached the emit again. Key the alert on its own operational_events row
instead, and run the check on every takedown issuance (including the dedup
replay path), so a replay after a failed emit recovers the signal while a normal
replay does not duplicate it.

* fix(labeler): scope unsubscribe query fallback, keep confirm body-only

The one-click query fallback was applied to the confirm and not-me POST parsers
too, so a scanner POSTing the original confirmation link URL (with c and t in
the query) could confirm an address without the user submitting the form.
Restrict the query fallback to the RFC 8058 one-click unsubscribe path; confirm
and not-me require their capability from the POST body.

* fix(labeler): validate release rkey shape; make no-contact alert dedup atomic

Contact resolution stripped a subject rkey at the first colon without checking
the collection or the canonical release rkey shape, so a malformed colon-bearing
subject could resolve a DIFFERENT package's contacts. Only strip a canonical
release record (release collection + well-formed slug:version); anything else
keeps its rkey and degrades to the publisher tier.

The takedown-no-contact alert used an existence-check-then-insert with no
uniqueness guard, so two concurrent replays could double-emit. Add a
UNIQUE(action_id, event_type) index (migration 0012) and make the insert
idempotent (ON CONFLICT DO NOTHING), with the outbox gated on the event being
written; concurrency and replay now converge to exactly one event.

* fix(labeler): scope dedup index to takedown-no-contact; require semver

Migration 0012 created a GLOBAL unique index on (action_id, event_type), which
would fail at migration time on any existing DB holding historical duplicate
rows of other event types. Make it PARTIAL (WHERE event_type =
'takedown-no-contact') and point the insert's ON CONFLICT target at that
predicate, so it stays safe on historical data while still deduping the alert.

The release-rkey parse validated only the slug and a non-empty suffix, so
'gallery:not-semver' still resolved the gallery package. Validate the complete
canonical shape (slug + percent-decoded semver version, mirroring the
aggregator's parseReleaseRkey) before stripping; anything else degrades to the
publisher tier.

Author: ascorbic

apps/labeler/migrations/0012_operational_events_action_type_unique.sql

@@ -0,0 +1,14 @@
+-- At-most-one `takedown-no-contact` operational event per action. That deferred
+-- alert is raised from a fire-and-forget tail with a check-then-insert, so a
+-- concurrent replay could double-emit; this unique index is the hard guarantee
+-- that collapses them to one row (paired with ON CONFLICT DO NOTHING on the
+-- insert).
+--
+-- PARTIAL, scoped to `takedown-no-contact` only: the prior schema allowed
+-- duplicate (action_id, event_type) rows and historical data may hold them for
+-- OTHER event types, which a global unique index would reject at migration time.
+-- Restricting the predicate to the one event type that needs the guarantee keeps
+-- the migration safe on existing data while still enforcing the dedup we require.
+CREATE UNIQUE INDEX idx_operational_events_takedown_no_contact
+	ON operational_events(action_id, event_type)
+	WHERE event_type = 'takedown-no-contact';

apps/labeler/src/notification-endpoints.ts

@@ -14,8 +14,10 @@
  * Safety properties:
  *   - GET renders a confirmation page with a POST form; only POST mutates. An
  *     email scanner's or link-prefetcher's automated GET therefore never
- *     confirms, unsubscribes, or suppresses. The token/hash live in hidden form
- *     fields, so the mutating POST carries them in the body, not the URL.
+ *     confirms, unsubscribes, or suppresses. The browser POST carries the
+ *     token/hash in hidden form fields; only the RFC 8058 one-click UNSUBSCRIBE
+ *     POST carries `c` in the header URL's query, so the query fallback is scoped
+ *     to that path — confirm and not-me require their capability from the body.
  *   - CSRF: a cross-site POST cannot supply a valid recipient hash (or confirm
  *     token) for a victim, so possession of the capability is the CSRF defense;
  *     a custom header (unsendable from a plain email-client form) is not used.
@@ -114,18 +116,25 @@ async function readPostParams(
 	request: Request,
 	action: NotificationAction,
 ): Promise<RequestParams> {
-	let form: FormData;
+	let form: FormData | null = null;
 	try {
 		form = await request.formData();
 	} catch {
-		return { recipientHash: "", token: "" };
+		form = null;
 	}
-	const recipientHash = form.get("c");
-	const token = form.get("t");
-	return {
-		recipientHash: typeof recipientHash === "string" ? recipientHash : "",
-		token: action === "confirm" && typeof token === "string" ? token : "",
-	};
+	const bodyHash = form?.get("c");
+	let recipientHash = typeof bodyHash === "string" ? bodyHash : "";
+	// RFC 8058 one-click unsubscribe POSTs to the List-Unsubscribe header URL with
+	// `c` only in its query string and `List-Unsubscribe=One-Click` in the body.
+	// That query fallback is scoped to unsubscribe ONLY: confirm and not-me require
+	// their capability from the POST body, so a scanner cannot confirm (or suppress
+	// via not-me) an address by POSTing a link URL without the rendered form.
+	if (action === "unsubscribe" && recipientHash.length === 0) {
+		recipientHash = new URL(request.url).searchParams.get("c") ?? "";
+	}
+	const bodyToken = form?.get("t");
+	const token = action === "confirm" && typeof bodyToken === "string" ? bodyToken : "";
+	return { recipientHash, token };
 }
 
 /**

apps/labeler/src/notification-triggers.ts

@@ -16,10 +16,15 @@
  *     `notifications` row is skipped, so a Workflow-step retry or a mutation
  *     replay does not re-notify. `sendNotification`'s notice claim closes the
  *     residual concurrent race atomically.
- *   - Verified-publisher skip: a publisher with an in-force verification claim
- *     bypasses double opt-in (the notice goes out directly). The verification read
- *     fails CLOSED — any error leaves the publisher on the normal confirmation
- *     path, never looser.
+ *   - Verified-publisher skip: a publisher whose CURRENT identity is vouched for
+ *     by an in-force verification claim from a TRUSTED issuer bypasses double
+ *     opt-in (the notice goes out directly). A self-issued or otherwise untrusted
+ *     claim carries no authority — verification claims are self-assertable and the
+ *     aggregator indexes any issuer — so only a claim whose issuer is in the
+ *     configured trust set AND whose bound displayName still matches the
+ *     publisher's current identity upgrades a contact. The read fails CLOSED, and
+ *     with no trusted issuer configured (the default) NOTHING upgrades: every
+ *     address stays on the normal confirmation path.
  *
  * All notice copy is public-safe: subject line, label effect, the assessment's
  * public summary, and the public assessment + reconsideration URLs. No findings,
@@ -41,6 +46,11 @@ import type {
 	SendContext,
 } from "./notification-send.js";
 import { sendNotification } from "./notification-send.js";
+import {
+	buildOperationalEventInsert,
+	buildOutboxInsert,
+	newOperationalEventId,
+} from "./operational-events.js";
 import { getOperatorActionById } from "./operator-actions.js";
 import { getLabelDefinition } from "./policy.js";
 import type { ContactTarget } from "./publisher-contact.js";
@@ -60,9 +70,18 @@ export interface NotifyDeps {
 	serviceUrl: string;
 	/** The monitored reconsideration URL from the moderation policy. */
 	reconsiderationUrl: string;
+	/** DIDs whose verification claims may upgrade a contact past double opt-in.
+	 * Verification claims are self-assertable (the aggregator indexes any issuer),
+	 * so a claim upgrades only when its issuer is in this set. Empty (the default)
+	 * trusts no issuer, so every address stays on the confirmation path. */
+	trustedVerificationIssuers?: ReadonlySet<string>;
 	now?: () => Date;
 }
 
+/** No verification issuer is trusted — the conservative default for
+ * {@link NotifyDeps.trustedVerificationIssuers}. */
+const NO_TRUSTED_ISSUERS: ReadonlySet<string> = new Set();
+
 /**
  * Build the production {@link NotifyDeps} from the Worker env: the real Cloudflare
  * Email Sending adapter over the `EMAIL` binding, the aggregator client over the
@@ -82,6 +101,10 @@ export async function createNotifyDeps(env: Env): Promise<NotifyDeps> {
 		pepper,
 		serviceUrl: env.LABELER_SERVICE_URL,
 		reconsiderationUrl: moderationPolicy.contact.reconsiderationUrl,
+		// No verification issuer is trusted to bypass double opt-in: the labeler
+		// issues no first-party verification and the aggregator indexes any
+		// self-asserted claim, so every address goes through confirmation.
+		trustedVerificationIssuers: NO_TRUSTED_ISSUERS,
 	};
 }
 
@@ -327,6 +350,97 @@ export async function notifyEmergencyTakedown(
 		target,
 		emergencyNoticeContent(deps, input),
 	);
+	// A takedown ISSUANCE (not a retract) that resolves no contact is the most
+	// consequential delivery failure — the undeliverable audit row alone is easy to
+	// miss. Raise a dedicated operator alert, keyed on ITS OWN operational_events
+	// row rather than the notifications-row dedup: this runs on every issuance,
+	// including a replay that `runTrigger` short-circuited on the existing
+	// notifications row, so a first pass whose event write failed transiently
+	// recovers on the next replay instead of losing the signal forever.
+	if (!input.neg) await ensureTakedownNoContactAlert(deps, input.actionId, input.uri);
+}
+
+/** The operator-alert channel for the emergency no-contact signal — mirrors the
+ * emergency-action alert channel in `console-mutation-api`. */
+const OPERATOR_ALERT_CHANNEL = "deployment-alert";
+
+/**
+ * Raise the `takedown-no-contact` operational event + its outbox row when a
+ * takedown issuance resolved no contact and the alert is not already recorded.
+ * Idempotent and recoverable: the operational_events row (not the notifications
+ * row) is the dedup key, so a first pass whose event batch failed re-emits on a
+ * later replay, while a normal replay does not duplicate. The insert is guarded
+ * by the `(action_id, event_type)` unique index + ON CONFLICT DO NOTHING, so two
+ * concurrent replays that both pass the existence check still converge to one
+ * event (the outbox is gated on the event being written, so a conflict leaves no
+ * orphan). Fire-and-forget — an error is swallowed and logged, never propagated
+ * into the deferred tail (a later replay retries anyway, the event still absent).
+ */
+async function ensureTakedownNoContactAlert(
+	deps: NotifyDeps,
+	actionId: string,
+	uri: string,
+): Promise<void> {
+	try {
+		if (!(await sourceResolvedNoContact(deps.db, actionId))) return;
+		if (await takedownNoContactAlertExists(deps.db, actionId)) return;
+		const now = (deps.now ?? (() => new Date()))();
+		const eventId = newOperationalEventId();
+		await deps.db.batch([
+			buildOperationalEventInsert(deps.db, {
+				id: eventId,
+				eventType: "takedown-no-contact",
+				severity: "high",
+				actionId,
+				subjectUri: uri,
+				labelValue: "!takedown",
+				payload: {
+					reason:
+						"Emergency takedown has no resolvable publisher contact; manual outreach required.",
+				},
+				now,
+				idempotentTakedownNoContact: true,
+			}),
+			buildOutboxInsert(deps.db, {
+				eventId,
+				channel: OPERATOR_ALERT_CHANNEL,
+				now,
+				gateOnEventPresent: true,
+			}),
+		]);
+	} catch (error) {
+		console.error("[notifications] takedown no-contact alert failed", {
+			actionId,
+			error: error instanceof Error ? error.message : String(error),
+		});
+	}
+}
+
+/** Whether this operator action's notification resolved to NO contact: the send
+ * path records that (and only that) undeliverable case with a NULL
+ * `recipient_hash` (suppressed/declined carry a hash). */
+async function sourceResolvedNoContact(db: D1Database, actionId: string): Promise<boolean> {
+	const row = await db
+		.prepare(
+			`SELECT 1 FROM notifications
+			 WHERE source_type = 'operator' AND source_id = ? AND recipient_hash IS NULL LIMIT 1`,
+		)
+		.bind(actionId)
+		.first<{ 1: number }>();
+	return row !== null;
+}
+
+/** Whether a `takedown-no-contact` alert already exists for this action — the
+ * event's dedup key, decoupled from the notifications-row dedup. */
+async function takedownNoContactAlertExists(db: D1Database, actionId: string): Promise<boolean> {
+	const row = await db
+		.prepare(
+			`SELECT 1 FROM operational_events
+			 WHERE action_id = ? AND event_type = 'takedown-no-contact' LIMIT 1`,
+		)
+		.bind(actionId)
+		.first<{ 1: number }>();
+	return row !== null;
 }
 
 /** Reconsideration outcome notice (console `reconsiderations/:id/resolve`). Fired
@@ -491,7 +605,12 @@ async function runTrigger(
 			logTrigger(source, target.did, "deduped");
 			return true;
 		}
-		const verifiedPublisher = await isVerifiedPublisher(deps.aggregator, target.did, now());
+		const verifiedPublisher = await isVerifiedPublisher(
+			deps.aggregator,
+			deps.trustedVerificationIssuers ?? NO_TRUSTED_ISSUERS,
+			target.did,
+			now(),
+		);
 		const ctx: SendContext = {
 			db: deps.db,
 			aggregator: deps.aggregator,
@@ -531,21 +650,45 @@ async function sourceAlreadyProcessed(
 }
 
 /**
- * Whether the publisher holds an IN-FORCE verification claim — reuses the
- * history-context notion of "in force" (a claim whose `expiresAt` is absent or in
- * the future). FAILS CLOSED: a `null` view (redacted / unverified), an empty
- * claim set, or ANY read error returns `false`, so the address stays on the
- * stricter double-opt-in path.
+ * Whether the publisher may bypass double opt-in on the strength of a TRUSTED
+ * verification claim. A claim upgrades a contact only when it is
+ *   (a) issued by a DID in `trustedIssuers` — verification claims are
+ *       self-assertable and the aggregator indexes any issuer, so an untrusted or
+ *       self-issued claim carries no authority;
+ *   (b) in force — no expiry, or an expiry still in the future; and
+ *   (c) bound to the publisher's CURRENT identity — its `displayName` still
+ *       matches `getPublisher`'s live value, so a displayName drift since the
+ *       verification does not carry trust. The claim also binds a `handle`, but
+ *       the publisher view carries no current handle to compare it against (the
+ *       aggregator's identity-event ingestion is unbuilt), so handle-binding is
+ *       deferred until it does.
+ * Note this vouches for the publisher's identity, not that they own the contact
+ * address — the trust set is the operator's explicit decision to accept that.
+ *
+ * FAILS CLOSED: an empty trust set (the default), a `null` verification/publisher
+ * view, an empty claim set, unknown current handle/displayName, or ANY read error
+ * returns `false`, so the address stays on the stricter double-opt-in path.
  */
 export async function isVerifiedPublisher(
-	aggregator: Pick<AggregatorClient, "getPublisherVerification">,
+	aggregator: Pick<AggregatorClient, "getPublisherVerification" | "getPublisher">,
+	trustedIssuers: ReadonlySet<string>,
 	did: string,
 	now: Date,
 ): Promise<boolean> {
+	if (trustedIssuers.size === 0) return false;
 	try {
 		const state = await aggregator.getPublisherVerification(did);
 		if (!state || state.verifications.length === 0) return false;
-		return state.verifications.some((claim) => isInForce(claim.expiresAt, now));
+		const trustedInForce = state.verifications.filter(
+			(claim) => trustedIssuers.has(claim.issuer) && isInForce(claim.expiresAt, now),
+		);
+		if (trustedInForce.length === 0) return false;
+
+		const publisher = await aggregator.getPublisher(did);
+		if (!publisher) return false;
+		const displayName = readDisplayName(publisher.profile);
+		if (displayName === undefined) return false;
+		return trustedInForce.some((claim) => claim.displayName === displayName);
 	} catch (error) {
 		console.error("[notifications] verification read failed, using double opt-in", {
 			did,
@@ -555,6 +698,14 @@ export async function isVerifiedPublisher(
 	}
 }
 
+/** The `displayName` a publisher-profile record carries, or undefined when the
+ * profile is not a `{ displayName: string }`-shaped object. */
+function readDisplayName(profile: unknown): string | undefined {
+	if (typeof profile !== "object" || profile === null) return undefined;
+	const value = (profile as { displayName?: unknown }).displayName;
+	return typeof value === "string" ? value : undefined;
+}
+
 /** A claim is in force when it has no expiry or its expiry is still in the future
  * (mirrors `history-context`'s `isExpired`). */
 function isInForce(expiresAt: string | undefined, now: Date): boolean {
@@ -577,24 +728,57 @@ function assessmentUrl(serviceUrl: string, uri: string, cid?: string): string {
 /**
  * The `{ did, slug }` a notice's contact resolution needs, parsed from the
  * subject URI. A record URI (`at://did/collection/rkey`) yields the DID and the
- * rkey as the slug — for a package record the rkey IS the package slug (tier-1/2
- * resolve); for a release the rkey misses the package tiers and resolution falls
- * through to the DID-keyed publisher-profile contact (tier 3), which is the
- * reliable channel. A bare DID subject (publisher-level action) resolves by DID
- * alone. Anything unparseable returns null and the notice is skipped.
+ * parent package slug: a package rkey IS the slug, and a canonical release rkey
+ * is `slug:version`, so the slug is the part before the first `:`. That lets
+ * `getPackage` reach the package's `security[]`/`authors[]` contacts (tier-1/2)
+ * for a release subject instead of falling straight through to the DID-keyed
+ * publisher profile (tier 3). A bare DID subject (publisher-level action)
+ * resolves by DID alone. Anything unparseable returns null and the notice is
+ * skipped.
  */
 export function contactTargetFromUri(uri: string): ContactTarget | null {
 	if (uri.startsWith("at://")) {
-		const [did, , rkey] = uri.slice("at://".length).split("/");
+		const [did, collection, rkey] = uri.slice("at://".length).split("/");
 		if (did === undefined || did.length === 0) return null;
-		return { did, slug: rkey ?? "" };
+		return { did, slug: packageSlugFromRecord(collection, rkey ?? "") };
 	}
 	if (uri.startsWith("did:")) {
 		return { did: uri, slug: uri.split(":").at(-1) ?? uri };
 	}
 	return null;
 }
 
+// Canonical release-rkey shape, mirroring the aggregator's ingest validation
+// (`records-consumer`'s `parseReleaseRkey`): `<slug>:<semver>` with the version
+// percent-decoded before the semver check. Kept in sync by the pinned tests.
+const PACKAGE_SLUG_RE = /^[a-zA-Z][a-zA-Z0-9_-]*$/;
+const SEMVER_RE = /^(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?$/;
+
+/**
+ * The parent package slug a subject resolves against. Only a CANONICAL release
+ * record — the release collection AND a `slug:version` rkey whose slug is
+ * well-formed AND whose version is valid semver — is stripped to its package slug
+ * (`gallery:1.2.0` → `gallery`). Every other subject keeps its rkey verbatim: a
+ * package rkey IS the slug, and a non-release collection or a malformed
+ * colon-bearing rkey (`gallery:not-semver`) stays whole so it can never strip to
+ * a DIFFERENT package's slug — it misses at `getPackage` and resolution degrades
+ * to the publisher tier.
+ */
+function packageSlugFromRecord(collection: string | undefined, rkey: string): string {
+	if (collection !== NSID.packageRelease) return rkey;
+	const delimiter = rkey.indexOf(":");
+	if (delimiter <= 0 || delimiter === rkey.length - 1) return rkey;
+	const slug = rkey.slice(0, delimiter);
+	if (!PACKAGE_SLUG_RE.test(slug)) return rkey;
+	let version: string;
+	try {
+		version = decodeURIComponent(rkey.slice(delimiter + 1));
+	} catch {
+		return rkey;
+	}
+	return SEMVER_RE.test(version) ? slug : rkey;
+}
+
 function logTrigger(source: NotificationSource, did: string, outcome: string): void {
 	console.log("[notifications]", {
 		action: "trigger",