diff --git a/apps/labeler/migrations/0012_operational_events_action_type_unique.sql b/apps/labeler/migrations/0012_operational_events_action_type_unique.sql new file mode 100644 index 0000000000..7abf31a41e --- /dev/null +++ b/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'; diff --git a/apps/labeler/src/notification-endpoints.ts b/apps/labeler/src/notification-endpoints.ts index b3abed96d7..198bb6efc5 100644 --- a/apps/labeler/src/notification-endpoints.ts +++ b/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 { - 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 }; } /** diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index 6726f89789..0978d08dc5 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/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; now?: () => Date; } +/** No verification issuer is trusted — the conservative default for + * {@link NotifyDeps.trustedVerificationIssuers}. */ +const NO_TRUSTED_ISSUERS: ReadonlySet = 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 { 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 { + 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 { + 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 { + 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, + aggregator: Pick, + trustedIssuers: ReadonlySet, did: string, now: Date, ): Promise { + 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,17 +728,19 @@ 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 }; @@ -595,6 +748,37 @@ export function contactTargetFromUri(uri: string): ContactTarget | null { return null; } +// Canonical release-rkey shape, mirroring the aggregator's ingest validation +// (`records-consumer`'s `parseReleaseRkey`): `:` 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", diff --git a/apps/labeler/src/operational-events.ts b/apps/labeler/src/operational-events.ts index 557170cb76..9aa36fad96 100644 --- a/apps/labeler/src/operational-events.ts +++ b/apps/labeler/src/operational-events.ts @@ -14,7 +14,8 @@ export type OperationalEventType = | "dead-letter-quarantined" | "reconsideration-opened" | "reconsideration-resolved" - | "assessment-prolonged-error"; + | "assessment-prolonged-error" + | "takedown-no-contact"; export type OperationalEventSeverity = "critical" | "high" | "info"; @@ -79,6 +80,15 @@ export interface OperationalEventInsert { * ticks are not serialized against each other). */ gateOnUnalertedEscalation?: { assessmentId: string }; + /** + * Appends `ON CONFLICT ... DO NOTHING` targeting the partial unique index on + * `(action_id, event_type) WHERE event_type = 'takedown-no-contact'` (migration + * 0012), so a concurrent replay of the `takedown-no-contact` alert converges to + * one row. Only valid for that event type with a non-null `actionId` and no + * gate. Pair the outbox with {@link OutboxInsert.gateOnEventPresent} so a + * conflicted (no-op) event does not orphan an outbox row. + */ + idempotentTakedownNoContact?: boolean; } export interface OutboxInsert { @@ -89,6 +99,10 @@ export interface OutboxInsert { gateOnIssuedLabelActionId?: number; /** Same in-batch label gating as {@link OperationalEventInsert.gateOnIssuedLabelActionKey}. */ gateOnIssuedLabelActionKey?: string; + /** Insert only if the event row was actually written — `WHERE EXISTS (event + * with this id)`. Pairs with {@link OperationalEventInsert.idempotentTakedownNoContact} + * so an ON CONFLICT no-op event leaves no orphan outbox row in the same batch. */ + gateOnEventPresent?: boolean; } export interface StoredOperationalEvent { @@ -225,10 +239,15 @@ export function buildOperationalEventInsert( .bind(...values, input.gateOnUnalertedEscalation.assessmentId); } + // The conflict target repeats the partial index's predicate so SQLite resolves + // it to `idx_operational_events_takedown_no_contact` (migration 0012). + const onConflict = input.idempotentTakedownNoContact + ? ` ON CONFLICT (action_id, event_type) WHERE event_type = 'takedown-no-contact' DO NOTHING` + : ""; return db .prepare( `INSERT INTO operational_events (${columns}) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)${onConflict}`, ) .bind(...values); } @@ -271,6 +290,16 @@ export function buildOutboxInsert(db: D1Database, input: OutboxInsert): D1Prepar .bind(...values, input.gateOnIssuedLabelActionId); } + if (input.gateOnEventPresent) { + return db + .prepare( + `INSERT INTO notification_outbox (id, event_id, channel, created_at, created_at_epoch_ms) + SELECT ?, ?, ?, ?, ? + WHERE EXISTS (SELECT 1 FROM operational_events WHERE id = ?)`, + ) + .bind(...values, input.eventId); + } + return db .prepare( `INSERT INTO notification_outbox (id, event_id, channel, created_at, created_at_epoch_ms) diff --git a/apps/labeler/test/notification-endpoints.test.ts b/apps/labeler/test/notification-endpoints.test.ts index 588cafd8ff..f3b68f78de 100644 --- a/apps/labeler/test/notification-endpoints.test.ts +++ b/apps/labeler/test/notification-endpoints.test.ts @@ -145,6 +145,24 @@ describe("confirm", () => { expect((await getContactState(db(), hash))?.confirmState).toBe("confirmed"); }); + it("does NOT confirm when the credentials are only in the query (no form body)", async () => { + const token = "confirm-token-query-only"; + const hash = await seedPending(token); + + // A scanner POSTing the confirmation link URL with c/t in the query and no + // form body must not confirm — confirm has no query fallback. + const response = await SELF.fetch( + `https://labeler.test/notifications/confirm?c=${hash}&t=${encodeURIComponent(token)}`, + { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: "", + }, + ); + expect(response.status).toBe(200); + expect((await getContactState(db(), hash))?.confirmState).toBe("unconfirmed"); + }); + it("never confirms a suppressed contact even with a matching token", async () => { const token = "confirm-token-suppressed"; const hash = await seedPending(token); @@ -204,6 +222,39 @@ describe("unsubscribe", () => { .first<{ n: number }>(); expect(count?.n).toBe(0); }); + + it("suppresses on an RFC 8058 one-click POST carrying c in the query and the marker in the body", async () => { + const hash = await freshHash(); + await ensureContact(db(), hash, "2026-07-16T00:00:00.000Z"); + + const response = await SELF.fetch(`https://labeler.test/notifications/unsubscribe?c=${hash}`, { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: "List-Unsubscribe=One-Click", + }); + expect(response.status).toBe(200); + expect(await isSuppressed(db(), hash)).toBe(true); + expect(await reasonFor(hash)).toBe("unsubscribe"); + }); + + it("prefers the body hash over the query hash when both are present", async () => { + const bodyHash = await freshHash(); + const queryHash = await freshHash(); + await ensureContact(db(), bodyHash, "2026-07-16T00:00:00.000Z"); + await ensureContact(db(), queryHash, "2026-07-16T00:00:00.000Z"); + + const response = await SELF.fetch( + `https://labeler.test/notifications/unsubscribe?c=${queryHash}`, + { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ c: bodyHash }).toString(), + }, + ); + expect(response.status).toBe(200); + expect(await isSuppressed(db(), bodyHash)).toBe(true); + expect(await isSuppressed(db(), queryHash)).toBe(false); + }); }); describe("not-me", () => { diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 265652f459..230079599b 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -13,6 +13,7 @@ import { } from "../src/notification-contacts.js"; import type { ConfirmationPayload, NoticePayload, SendResult } from "../src/notification-send.js"; import { + contactTargetFromUri, notifyAssessmentOutcome, notifyEmergencyTakedown, notifyOperatorLabel, @@ -20,6 +21,11 @@ import { notifyOverrideRetract, type NotifyDeps, } from "../src/notification-triggers.js"; +import { + buildOperationalEventInsert, + buildOutboxInsert, + newOperationalEventId, +} from "../src/operational-events.js"; interface TestEnv { DB: D1Database; @@ -45,6 +51,12 @@ interface AggregatorOpts { email?: string; verifications?: unknown[]; verifyStatus?: number; + /** The publisher's current displayName, served by getPublisher — the live + * identity the verification binding is checked against. The publisher view + * carries no handle, so none is served. */ + displayName?: string; + /** A package `security[]` contact served by getPackage for a specific slug. */ + packageSecurity?: { slug: string; email: string }; } function aggregatorFor(opts: AggregatorOpts): AggregatorClient { @@ -55,10 +67,21 @@ function aggregatorFor(opts: AggregatorOpts): AggregatorClient { return new Response("err", { status: opts.verifyStatus }); return Response.json({ did: DID, verifications: opts.verifications ?? [], labels: [] }); } + if (url.includes("getPackage") && opts.packageSecurity !== undefined) { + const slug = new URL(url).searchParams.get("slug"); + if (slug === opts.packageSecurity.slug) { + return Response.json({ + profile: { security: [{ email: opts.packageSecurity.email }] }, + }); + } + } if (url.includes("getPublisher") && opts.email !== undefined) { return Response.json({ did: DID, - profile: { contact: [{ kind: "security", email: opts.email }] }, + profile: { + contact: [{ kind: "security", email: opts.email }], + ...(opts.displayName !== undefined ? { displayName: opts.displayName } : {}), + }, }); } return new Response(JSON.stringify({ error: "NotFound" }), { @@ -94,6 +117,27 @@ function recordingSender(result: SendResult = { ok: true, providerId: "p" }): Re }; } +/** Wraps a D1 database so its FIRST `batch()` throws (a transient failure), + * delegating every other call — and later batches — to the real db. */ +function dbThrowingFirstBatch(real: D1Database): D1Database { + let thrown = false; + return new Proxy(real, { + get(target, prop, receiver) { + if (prop === "batch") { + return async (statements: D1PreparedStatement[]) => { + if (!thrown) { + thrown = true; + throw new Error("transient batch failure"); + } + return target.batch(statements); + }; + } + const value = Reflect.get(target, prop, receiver); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + function throwingSender(): RecordingSender { return { confirmations: [], @@ -107,7 +151,11 @@ function throwingSender(): RecordingSender { }; } -function deps(aggregator: AggregatorClient, sender: RecordingSender): NotifyDeps { +function deps( + aggregator: AggregatorClient, + sender: RecordingSender, + trustedVerificationIssuers?: ReadonlySet, +): NotifyDeps { return { db: db(), aggregator, @@ -115,6 +163,7 @@ function deps(aggregator: AggregatorClient, sender: RecordingSender): NotifyDeps pepper: PEPPER, serviceUrl: SERVICE, reconsiderationUrl: RECON, + ...(trustedVerificationIssuers ? { trustedVerificationIssuers } : {}), }; } @@ -146,17 +195,49 @@ async function notificationRows(sourceId: string): Promise<{ kind: string; state return r.results ?? []; } -const IN_FORCE = [ - { issuer: "did:plc:issuer", handle: "x.test", createdAt: "2026-01-01T00:00:00.000Z" }, +const TRUSTED_ISSUER = "did:plc:trusted"; +const TRUSTED = new Set([TRUSTED_ISSUER]); +const HANDLE = "acme.test"; +const DISPLAY_NAME = "Acme"; + +/** A trusted, in-force claim whose bound displayName matches the publisher's + * current displayName. */ +const TRUSTED_CLAIM = [ + { + issuer: TRUSTED_ISSUER, + handle: HANDLE, + displayName: DISPLAY_NAME, + createdAt: "2026-01-01T00:00:00.000Z", + }, ]; -const EXPIRED = [ +/** An in-force claim from an issuer NOT in the trust set (self-assertable). */ +const UNTRUSTED_CLAIM = [ { - issuer: "did:plc:issuer", - handle: "x.test", + issuer: "did:plc:attacker", + handle: HANDLE, + displayName: DISPLAY_NAME, + createdAt: "2026-01-01T00:00:00.000Z", + }, +]; +const EXPIRED_TRUSTED_CLAIM = [ + { + issuer: TRUSTED_ISSUER, + handle: HANDLE, + displayName: DISPLAY_NAME, createdAt: "2020-01-01T00:00:00.000Z", expiresAt: "2021-01-01T00:00:00.000Z", }, ]; +/** Trusted + in-force but bound to a stale displayName (the publisher's + * displayName has drifted since verification). */ +const STALE_BINDING_CLAIM = [ + { + issuer: TRUSTED_ISSUER, + handle: HANDLE, + displayName: "Acme (old name)", + createdAt: "2026-01-01T00:00:00.000Z", + }, +]; describe("the five events each notify a confirmed publisher", () => { it("automated block → notice from source 'issuance'", async () => { @@ -281,35 +362,123 @@ describe("dedup", () => { }); }); -describe("verified-publisher skip", () => { - it("an in-force verification claim delivers the notice without a confirmation mail", async () => { +describe("verified-publisher skip (trusted issuer only)", () => { + async function confirmStateOf(email: string): Promise { + const hash = await recipientHash(PEPPER, email); + const contact = await db() + .prepare(`SELECT confirm_state FROM notification_contacts WHERE recipient_hash = ?`) + .bind(hash) + .first<{ confirm_state: string }>(); + return contact?.confirm_state; + } + + it("a trusted issuer claim with current bindings delivers the notice and upgrades the contact", async () => { const email = uniq("vok") + "@x.test"; const sender = recordingSender(); const a = assessmentRow({ state: "blocked" }); - // The contact is UNCONFIRMED; verification upgrades it in place. + // The contact is UNCONFIRMED; the trusted, current-binding claim upgrades it. await notifyAssessmentOutcome( - deps(aggregatorFor({ email, verifications: IN_FORCE }), sender), + deps( + aggregatorFor({ + email, + verifications: TRUSTED_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + TRUSTED, + ), a, ); expect(sender.notices).toHaveLength(1); expect(sender.confirmations).toHaveLength(0); - const hash = await recipientHash(PEPPER, email); - const contact = await db() - .prepare(`SELECT confirm_state FROM notification_contacts WHERE recipient_hash = ?`) - .bind(hash) - .first<{ confirm_state: string }>(); - expect(contact?.confirm_state).toBe("confirmed"); + expect(await confirmStateOf(email)).toBe("confirmed"); + }); + + it("a self-issued / untrusted-issuer claim does NOT bypass double opt-in", async () => { + const email = uniq("vself") + "@x.test"; + const sender = recordingSender(); + const a = assessmentRow({ state: "blocked" }); + + // A claim naming an arbitrary address, issued by an untrusted DID, even with + // a trust set configured for a DIFFERENT issuer. + await notifyAssessmentOutcome( + deps( + aggregatorFor({ + email, + verifications: UNTRUSTED_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + TRUSTED, + ), + a, + ); + + expect(sender.notices).toHaveLength(0); + expect(sender.confirmations).toHaveLength(1); + }); + + it("with no trusted issuer configured (the default), an in-force claim does NOT bypass", async () => { + const email = uniq("vnodef") + "@x.test"; + const sender = recordingSender(); + const a = assessmentRow({ state: "blocked" }); + + await notifyAssessmentOutcome( + deps( + aggregatorFor({ + email, + verifications: TRUSTED_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + ), + a, + ); + + expect(sender.notices).toHaveLength(0); + expect(sender.confirmations).toHaveLength(1); }); - it("an EXPIRED claim falls back to double opt-in", async () => { + it("an EXPIRED trusted claim falls back to double opt-in", async () => { const email = uniq("vexp") + "@x.test"; const sender = recordingSender(); const a = assessmentRow({ state: "blocked" }); await notifyAssessmentOutcome( - deps(aggregatorFor({ email, verifications: EXPIRED }), sender), + deps( + aggregatorFor({ + email, + verifications: EXPIRED_TRUSTED_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + TRUSTED, + ), + a, + ); + + expect(sender.notices).toHaveLength(0); + expect(sender.confirmations).toHaveLength(1); + }); + + it("a trusted claim with a STALE identity binding does NOT bypass", async () => { + const email = uniq("vstale") + "@x.test"; + const sender = recordingSender(); + const a = assessmentRow({ state: "blocked" }); + + // Claim's bound displayName has drifted from the publisher's current one. + await notifyAssessmentOutcome( + deps( + aggregatorFor({ + email, + verifications: STALE_BINDING_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + TRUSTED, + ), a, ); @@ -322,13 +491,16 @@ describe("verified-publisher skip", () => { const sender = recordingSender(); const a = assessmentRow({ state: "blocked" }); - await notifyAssessmentOutcome(deps(aggregatorFor({ email, verifyStatus: 500 }), sender), a); + await notifyAssessmentOutcome( + deps(aggregatorFor({ email, verifyStatus: 500 }), sender, TRUSTED), + a, + ); expect(sender.notices).toHaveLength(0); expect(sender.confirmations).toHaveLength(1); }); - it("a suppressed address gets NOTHING even when the publisher is verified", async () => { + it("a suppressed address gets NOTHING even when the publisher is trusted-verified", async () => { const email = uniq("vsupp") + "@x.test"; const hash = await recipientHash(PEPPER, email); await suppress(db(), hash, "not_me", "2026-07-16T00:00:00.000Z", 1_000); @@ -336,7 +508,15 @@ describe("verified-publisher skip", () => { const a = assessmentRow({ state: "blocked" }); await notifyAssessmentOutcome( - deps(aggregatorFor({ email, verifications: IN_FORCE }), sender), + deps( + aggregatorFor({ + email, + verifications: TRUSTED_CLAIM, + displayName: DISPLAY_NAME, + }), + sender, + TRUSTED, + ), a, ); @@ -368,6 +548,293 @@ describe("provider hard-bounce", () => { }); }); +describe("package slug parse", () => { + it("parses the parent package slug from a release rkey (slug:version)", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.release/gallery:1.2.0`, + ); + expect(target).toEqual({ did: DID, slug: "gallery" }); + }); + + it("keeps a package rkey (no version) as the slug", () => { + const target = contactTargetFromUri(`at://${DID}/com.emdashcms.experimental.package/gallery`); + expect(target).toEqual({ did: DID, slug: "gallery" }); + }); + + it("degrades safely on an rkey with no slug before the version delimiter", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.release/:1.2.0`, + ); + expect(target).toEqual({ did: DID, slug: ":1.2.0" }); + }); + + it("resolves the package's security contact for a release URI", async () => { + const email = uniq("pkgsec") + "@x.test"; + await seedConfirmed(email); + const sender = recordingSender(); + const releaseUri = `at://${DID}/com.emdashcms.experimental.package.release/gallery:1.2.0`; + const a = assessmentRow({ state: "blocked", uri: releaseUri }); + + await notifyAssessmentOutcome( + deps(aggregatorFor({ packageSecurity: { slug: "gallery", email } }), sender), + a, + ); + + expect(sender.notices).toHaveLength(1); + expect(sender.notices[0]?.to).toBe(email); + }); + + it("does not strip a colon-bearing rkey under a non-release collection", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.profile/gallery:evil`, + ); + expect(target).toEqual({ did: DID, slug: "gallery:evil" }); + }); + + it("does not strip a release rkey whose slug is malformed", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.release/1bad:2.0.0`, + ); + expect(target).toEqual({ did: DID, slug: "1bad:2.0.0" }); + }); + + it("does not strip a release rkey whose version is not valid semver", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.release/gallery:not-semver`, + ); + expect(target).toEqual({ did: DID, slug: "gallery:not-semver" }); + }); + + it("strips a release rkey with a valid prerelease semver version", () => { + const target = contactTargetFromUri( + `at://${DID}/com.emdashcms.experimental.package.release/gallery:1.2.0-beta.1`, + ); + expect(target).toEqual({ did: DID, slug: "gallery" }); + }); + + it("a release rkey with a non-semver version degrades to the publisher contact", async () => { + const publisherEmail = uniq("pubsem") + "@x.test"; + const galleryEmail = uniq("galsem") + "@x.test"; + await seedConfirmed(publisherEmail); + const sender = recordingSender(); + const uri = `at://${DID}/com.emdashcms.experimental.package.release/gallery:not-semver`; + const a = assessmentRow({ state: "blocked", uri }); + + await notifyAssessmentOutcome( + deps( + aggregatorFor({ + email: publisherEmail, + packageSecurity: { slug: "gallery", email: galleryEmail }, + }), + sender, + ), + a, + ); + + expect(sender.notices).toHaveLength(1); + expect(sender.notices[0]?.to).toBe(publisherEmail); + }); + + it("a malformed colon-bearing subject degrades to the publisher contact, not a wrong package", async () => { + const publisherEmail = uniq("pub") + "@x.test"; + const galleryEmail = uniq("gal") + "@x.test"; + await seedConfirmed(publisherEmail); + const sender = recordingSender(); + // Colon-bearing rkey under a non-release collection: must NOT resolve the + // gallery package's security contact by stripping to "gallery". + const uri = `at://${DID}/com.emdashcms.experimental.package.profile/gallery:evil`; + const a = assessmentRow({ state: "blocked", uri }); + + await notifyAssessmentOutcome( + deps( + aggregatorFor({ + email: publisherEmail, + packageSecurity: { slug: "gallery", email: galleryEmail }, + }), + sender, + ), + a, + ); + + expect(sender.notices).toHaveLength(1); + expect(sender.notices[0]?.to).toBe(publisherEmail); + }); +}); + +describe("emergency takedown with no resolvable contact", () => { + /** The committed operator action the takedown notify references — the + * operational_events.action_id FK requires it (satisfied in production because + * the emergency action commits before the deferred notify). */ + async function seedTakedownAction(actionId: string): Promise { + await db() + .prepare( + `INSERT INTO operator_actions + (id, actor_type, actor_id, role, action, subject_uri, label_value, reason, + idempotency_key, request_fingerprint, created_at, created_at_epoch_ms) + VALUES (?, 'human', 'op', 'admin', 'takedown', ?, '!takedown', 'r', ?, 'fp', ?, ?)`, + ) + .bind(actionId, RELEASE_URI, actionId, "2026-07-16T00:00:00.000Z", 1_000) + .run(); + } + + async function takedownEvents( + actionId: string, + ): Promise<{ eventType: string; severity: string; subjectUri: string | null }[]> { + const r = await db() + .prepare( + `SELECT event_type, severity, subject_uri FROM operational_events WHERE action_id = ?`, + ) + .bind(actionId) + .all<{ event_type: string; severity: string; subject_uri: string | null }>(); + return (r.results ?? []).map((row) => ({ + eventType: row.event_type, + severity: row.severity, + subjectUri: row.subject_uri, + })); + } + + it("emits a takedown-no-contact operational event and outbox row", async () => { + const sender = recordingSender(); + const actionId = uniq("oact"); + await seedTakedownAction(actionId); + + // aggregatorFor with no email/package resolves no contact. + await notifyEmergencyTakedown(deps(aggregatorFor({}), sender), { + actionId, + uri: RELEASE_URI, + neg: false, + }); + + expect(sender.notices).toHaveLength(0); + expect(await takedownEvents(actionId)).toEqual([ + { eventType: "takedown-no-contact", severity: "high", subjectUri: RELEASE_URI }, + ]); + const event = await db() + .prepare(`SELECT id FROM operational_events WHERE action_id = ?`) + .bind(actionId) + .first<{ id: string }>(); + const outbox = await db() + .prepare(`SELECT channel FROM notification_outbox WHERE event_id = ?`) + .bind(event?.id) + .first<{ channel: string }>(); + expect(outbox?.channel).toBe("deployment-alert"); + }); + + it("does not emit the event on a takedown RETRACT with no contact", async () => { + const sender = recordingSender(); + const actionId = uniq("oact"); + + await notifyEmergencyTakedown(deps(aggregatorFor({}), sender), { + actionId, + uri: RELEASE_URI, + neg: true, + }); + + expect(await takedownEvents(actionId)).toEqual([]); + }); + + it("does not emit the event when a contact resolves", async () => { + const email = uniq("tdok") + "@x.test"; + await seedConfirmed(email); + const sender = recordingSender(); + const actionId = uniq("oact"); + + await notifyEmergencyTakedown(deps(aggregatorFor({ email }), sender), { + actionId, + uri: RELEASE_URI, + neg: false, + }); + + expect(sender.notices).toHaveLength(1); + expect(await takedownEvents(actionId)).toEqual([]); + }); + + it("does not re-emit on a deduped replay", async () => { + const sender = recordingSender(); + const actionId = uniq("oact"); + await seedTakedownAction(actionId); + const d = deps(aggregatorFor({}), sender); + + await notifyEmergencyTakedown(d, { actionId, uri: RELEASE_URI, neg: false }); + await notifyEmergencyTakedown(d, { actionId, uri: RELEASE_URI, neg: false }); + + expect(await takedownEvents(actionId)).toHaveLength(1); + }); + + it("converges to one event when two replays both emit (unique index + ON CONFLICT)", async () => { + const actionId = uniq("oact"); + await seedTakedownAction(actionId); + const database = db(); + const now = new Date("2026-07-18T00:00:00.000Z"); + + // Two replays that both passed the existence check each build their own + // event+outbox batch for the same (action_id, event_type). The unique index + // + ON CONFLICT DO NOTHING must collapse them to a single event with no throw. + const build = () => { + const eventId = newOperationalEventId(); + return [ + buildOperationalEventInsert(database, { + id: eventId, + eventType: "takedown-no-contact", + severity: "high", + actionId, + subjectUri: RELEASE_URI, + labelValue: "!takedown", + payload: { reason: "manual outreach required" }, + now, + idempotentTakedownNoContact: true, + }), + buildOutboxInsert(database, { + eventId, + channel: "deployment-alert", + now, + gateOnEventPresent: true, + }), + ]; + }; + await database.batch(build()); + await database.batch(build()); + + expect(await takedownEvents(actionId)).toHaveLength(1); + const outbox = await database + .prepare( + `SELECT COUNT(*) n FROM notification_outbox nob + JOIN operational_events oe ON oe.id = nob.event_id WHERE oe.action_id = ?`, + ) + .bind(actionId) + .first<{ n: number }>(); + expect(outbox?.n).toBe(1); + }); + + it("recovers the alert on a replay after a transient first-pass emit failure", async () => { + const sender = recordingSender(); + const actionId = uniq("oact"); + await seedTakedownAction(actionId); + + // First pass: the undeliverable notifications row commits, but the alert + // batch fails transiently, so no event is recorded. + const firstPass: NotifyDeps = { + db: dbThrowingFirstBatch(db()), + aggregator: aggregatorFor({}), + sender, + pepper: PEPPER, + serviceUrl: SERVICE, + reconsiderationUrl: RECON, + }; + await notifyEmergencyTakedown(firstPass, { actionId, uri: RELEASE_URI, neg: false }); + expect(await takedownEvents(actionId)).toEqual([]); + + // Replay on a healthy db: the notifications row already exists (runTrigger + // dedups), but the alert is keyed on its own event row, so it recovers. + await notifyEmergencyTakedown(deps(aggregatorFor({}), sender), { + actionId, + uri: RELEASE_URI, + neg: false, + }); + expect(await takedownEvents(actionId)).toHaveLength(1); + }); +}); + describe("failure isolation", () => { it("a sender that THROWS never propagates out of the trigger (the label is safe)", async () => { const email = uniq("throw") + "@x.test"; diff --git a/apps/labeler/test/operational-events.test.ts b/apps/labeler/test/operational-events.test.ts index 560dc080a1..b3fe18bb4a 100644 --- a/apps/labeler/test/operational-events.test.ts +++ b/apps/labeler/test/operational-events.test.ts @@ -113,8 +113,14 @@ describe("operational_events store", () => { const otherId = `oact_other${counter}`; await insertOperatorAction(actionId); await insertOperatorAction(otherId); + // One action can raise events of DIFFERENT types (a takedown emits both + // `emergency-takedown` and the deferred `takedown-no-contact`); the + // (action_id, event_type) unique index forbids the SAME type twice. await buildOperationalEventInsert(testEnv.DB, eventInput({ actionId })).run(); - await buildOperationalEventInsert(testEnv.DB, eventInput({ actionId })).run(); + await buildOperationalEventInsert( + testEnv.DB, + eventInput({ actionId, eventType: "takedown-no-contact" }), + ).run(); await buildOperationalEventInsert(testEnv.DB, eventInput({ actionId: otherId })).run(); const rows = await getOperationalEventsByActionId(testEnv.DB, actionId); @@ -122,6 +128,37 @@ describe("operational_events store", () => { expect(rows.every((r) => r.actionId === actionId)).toBe(true); }); + it("allows duplicate (action_id, event_type) for a non-takedown event type", async () => { + // The 0012 unique index is PARTIAL to `takedown-no-contact`, so historical + // duplicates of other types are unconstrained — this is what keeps the + // migration safe on existing data. + const actionId = `oact_dup${counter}`; + await insertOperatorAction(actionId); + await buildOperationalEventInsert( + testEnv.DB, + eventInput({ actionId, eventType: "emergency-takedown" }), + ).run(); + await buildOperationalEventInsert( + testEnv.DB, + eventInput({ actionId, eventType: "emergency-takedown" }), + ).run(); + + const rows = await getOperationalEventsByActionId(testEnv.DB, actionId); + expect(rows).toHaveLength(2); + }); + + it("collapses a second takedown-no-contact for the same action to a no-op", async () => { + const actionId = `oact_tnc${counter}`; + await insertOperatorAction(actionId); + const input = () => + eventInput({ actionId, eventType: "takedown-no-contact", idempotentTakedownNoContact: true }); + await buildOperationalEventInsert(testEnv.DB, input()).run(); + await buildOperationalEventInsert(testEnv.DB, input()).run(); + + const rows = await getOperationalEventsByActionId(testEnv.DB, actionId); + expect(rows).toHaveLength(1); + }); + it("rejects UPDATE on a recorded event (immutable log)", async () => { const input = eventInput(); await buildOperationalEventInsert(testEnv.DB, input).run();