From 1cf0585111f9d7a7e02b901e85dd8f68951189ed Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:13:15 +0100 Subject: [PATCH 1/8] 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. --- apps/labeler/src/notification-endpoints.ts | 29 ++++++++++------ .../test/notification-endpoints.test.ts | 33 +++++++++++++++++++ 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/apps/labeler/src/notification-endpoints.ts b/apps/labeler/src/notification-endpoints.ts index b3abed96d7..4dd0f52acf 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; an RFC 8058 one-click unsubscribe POST + * carries `c` in the header URL's query instead, so the POST parser reads the + * body first and falls back to the query. * - 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; + // The capability travels in the header URL's query string. An RFC 8058 + // one-click POST (`List-Unsubscribe=One-Click` in the body) carries `c` ONLY + // there, so the query is the fallback when the body omits it; a browser form + // POST supplies `c` in the body, which wins. + const query = new URL(request.url).searchParams; + 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"); + const recipientHash = + typeof bodyHash === "string" && bodyHash.length > 0 ? bodyHash : (query.get("c") ?? ""); + if (action !== "confirm") return { recipientHash, token: "" }; + const bodyToken = form?.get("t"); + const token = + typeof bodyToken === "string" && bodyToken.length > 0 ? bodyToken : (query.get("t") ?? ""); + return { recipientHash, token }; } /** diff --git a/apps/labeler/test/notification-endpoints.test.ts b/apps/labeler/test/notification-endpoints.test.ts index 588cafd8ff..057c29116d 100644 --- a/apps/labeler/test/notification-endpoints.test.ts +++ b/apps/labeler/test/notification-endpoints.test.ts @@ -204,6 +204,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", () => { From 102e29d2c0d229af9ccf9b901b7c6fabb041c40b Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:15:29 +0100 Subject: [PATCH 2/8] 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. --- apps/labeler/src/notification-triggers.ts | 23 ++++++--- .../test/notification-triggers.test.ts | 48 +++++++++++++++++++ 2 files changed, 65 insertions(+), 6 deletions(-) diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index 6726f89789..228a44588f 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/apps/labeler/src/notification-triggers.ts @@ -577,17 +577,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 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("/"); if (did === undefined || did.length === 0) return null; - return { did, slug: rkey ?? "" }; + return { did, slug: packageSlugFromRkey(rkey ?? "") }; } if (uri.startsWith("did:")) { return { did: uri, slug: uri.split(":").at(-1) ?? uri }; @@ -595,6 +597,15 @@ export function contactTargetFromUri(uri: string): ContactTarget | null { return null; } +/** The parent package slug from a record rkey: the part before the `:` version + * delimiter for a release (`gallery:1.2.0` → `gallery`), or the rkey verbatim for + * a package (no delimiter) or any rkey whose delimiter is leading (no slug to + * take — left as-is so resolution degrades to the publisher tier). */ +function packageSlugFromRkey(rkey: string): string { + const delimiter = rkey.indexOf(":"); + return delimiter > 0 ? rkey.slice(0, delimiter) : rkey; +} + function logTrigger(source: NotificationSource, did: string, outcome: string): void { console.log("[notifications]", { action: "trigger", diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 265652f459..3c454647ce 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, @@ -45,6 +46,8 @@ interface AggregatorOpts { email?: string; verifications?: unknown[]; verifyStatus?: number; + /** A package `security[]` contact served by getPackage for a specific slug. */ + packageSecurity?: { slug: string; email: string }; } function aggregatorFor(opts: AggregatorOpts): AggregatorClient { @@ -55,6 +58,14 @@ 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, @@ -368,6 +379,43 @@ 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); + }); +}); + 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"; From 9049a11e3e83480526f42a7b0190f0cc0e2a87a5 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:21:31 +0100 Subject: [PATCH 3/8] 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. --- apps/labeler/src/notification-triggers.ts | 86 ++++++++++++--- apps/labeler/src/operational-events.ts | 3 +- .../test/notification-triggers.test.ts | 101 ++++++++++++++++++ 3 files changed, 176 insertions(+), 14 deletions(-) diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index 228a44588f..b6d6cb457a 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/apps/labeler/src/notification-triggers.ts @@ -39,8 +39,14 @@ import type { NotificationSender, NotificationSource, SendContext, + SendOutcome, } 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"; @@ -321,12 +327,57 @@ export async function notifyEmergencyTakedown( ): Promise { const target = contactTargetFromUri(input.uri); if (!target) return; - await runTrigger( + const result = await runTrigger( deps, { type: "operator", id: input.actionId }, 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, so raise a dedicated operator signal that manual outreach is needed. + if (!input.neg && result.outcome === "no_contact") { + await emitTakedownNoContactEvent(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 so the + * failed takedown notice surfaces to operators. Fire-and-forget: an insert + * failure is swallowed and logged, never propagated into the deferred tail. */ +async function emitTakedownNoContactEvent( + deps: NotifyDeps, + actionId: string, + uri: string, +): Promise { + const now = (deps.now ?? (() => new Date()))(); + try { + 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, + }), + buildOutboxInsert(deps.db, { eventId, channel: OPERATOR_ALERT_CHANNEL, now }), + ]); + } catch (error) { + console.error("[notifications] takedown no-contact event failed", { + actionId, + error: error instanceof Error ? error.message : String(error), + }); + } } /** Reconsideration outcome notice (console `reconsiderations/:id/resolve`). Fired @@ -362,12 +413,13 @@ export async function notifyProlongedError( // An unparseable URI is terminal, not transient (the URI never changes), so it // counts as processed — the cron marks it rather than re-attempting forever. if (!target) return true; - return runTrigger( + const result = await runTrigger( deps, { type: "issuance", id: assessment.id }, target, prolongedErrorNoticeContent(deps, { uri: assessment.uri, cid: assessment.cid }), ); + return result.terminal; } /** @@ -466,30 +518,38 @@ function operatorActionNeg(metadataJson: string): boolean { } } +/** The outcome of a trigger. `terminal` is false only when a TRANSIENT error was + * thrown before any row was claimed (the prolonged-error cron uses it to decide + * whether to stamp its fire-once mark). `outcome` is the `sendNotification` + * status when a send actually ran — undefined on a dedup hit or a thrown error — + * so a caller can react to a specific delivery result (e.g. `no_contact`). */ +interface TriggerResult { + terminal: boolean; + outcome?: SendOutcome["status"]; +} + /** * Dedup, resolve verification (fail-closed), send, swallow+log. Shared by every * trigger so the dedup and verified-skip policy is applied uniformly and a * notification failure can never escape into the label path. * - * Returns whether the trigger reached a TERMINAL outcome — a dedup hit or a - * normal `sendNotification` return (sent, confirmation-sent, undeliverable, or a - * claimed-then-failed row the sweep now owns) — versus a thrown TRANSIENT error - * (an aggregator read or a pre-claim D1 write that failed before any row was - * claimed). The prolonged-error cron uses this to decide whether to stamp its - * fire-once mark: a transient failure returns `false` so the next tick retries - * instead of being silently swallowed. Fire-and-forget callers ignore it. + * A TERMINAL result is a dedup hit or a normal `sendNotification` return (sent, + * confirmation-sent, undeliverable, or a claimed-then-failed row the sweep now + * owns); a non-terminal result is a thrown TRANSIENT error (an aggregator read or + * a pre-claim D1 write that failed before any row was claimed). Fire-and-forget + * callers ignore the result. */ async function runTrigger( deps: NotifyDeps, source: NotificationSource, target: ContactTarget, notice: NoticeContent, -): Promise { +): Promise { const now = deps.now ?? (() => new Date()); try { if (await sourceAlreadyProcessed(deps.db, source)) { logTrigger(source, target.did, "deduped"); - return true; + return { terminal: true }; } const verifiedPublisher = await isVerifiedPublisher(deps.aggregator, target.did, now()); const ctx: SendContext = { @@ -504,7 +564,7 @@ async function runTrigger( const request: NotificationRequest = { source, target, notice }; const outcome = await sendNotification(ctx, request); logTrigger(source, target.did, outcome.status); - return true; + return { terminal: true, outcome: outcome.status }; } catch (error) { console.error("[notifications] trigger failed", { sourceType: source.type, @@ -512,7 +572,7 @@ async function runTrigger( did: target.did, error: error instanceof Error ? error.message : String(error), }); - return false; + return { terminal: false }; } } diff --git a/apps/labeler/src/operational-events.ts b/apps/labeler/src/operational-events.ts index 557170cb76..ef24612e8b 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"; diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 3c454647ce..4f11199fe1 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -416,6 +416,107 @@ describe("package slug parse", () => { }); }); +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); + }); +}); + 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"; From cd4b3d9858b3bda3d694875639a793b2a74d9943 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:26:21 +0100 Subject: [PATCH 4/8] fix(labeler): gate double-opt-in bypass on trusted, current verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/labeler/src/notification-triggers.ts | 79 ++++++-- .../test/notification-triggers.test.ts | 187 +++++++++++++++--- 2 files changed, 232 insertions(+), 34 deletions(-) diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index b6d6cb457a..e73e71b33f 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, @@ -66,9 +71,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 @@ -88,6 +102,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, }; } @@ -551,7 +569,12 @@ async function runTrigger( logTrigger(source, target.did, "deduped"); return { terminal: 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, @@ -591,21 +614,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, @@ -615,6 +662,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 { diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 4f11199fe1..d351909564 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -46,6 +46,10 @@ 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 }; } @@ -69,7 +73,10 @@ function aggregatorFor(opts: AggregatorOpts): AggregatorClient { 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" }), { @@ -118,7 +125,11 @@ function throwingSender(): RecordingSender { }; } -function deps(aggregator: AggregatorClient, sender: RecordingSender): NotifyDeps { +function deps( + aggregator: AggregatorClient, + sender: RecordingSender, + trustedVerificationIssuers?: ReadonlySet, +): NotifyDeps { return { db: db(), aggregator, @@ -126,6 +137,7 @@ function deps(aggregator: AggregatorClient, sender: RecordingSender): NotifyDeps pepper: PEPPER, serviceUrl: SERVICE, reconsiderationUrl: RECON, + ...(trustedVerificationIssuers ? { trustedVerificationIssuers } : {}), }; } @@ -157,17 +169,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", + }, +]; +/** An in-force claim from an issuer NOT in the trust set (self-assertable). */ +const UNTRUSTED_CLAIM = [ + { + issuer: "did:plc:attacker", + handle: HANDLE, + displayName: DISPLAY_NAME, + createdAt: "2026-01-01T00:00:00.000Z", + }, ]; -const EXPIRED = [ +const EXPIRED_TRUSTED_CLAIM = [ { - issuer: "did:plc:issuer", - handle: "x.test", + 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 () => { @@ -292,35 +336,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("an EXPIRED claim falls back to double opt-in", async () => { + 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 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, ); @@ -333,13 +465,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); @@ -347,7 +482,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, ); From d1961b477afa9d27f13fe9bcdbaccb7cf88d0a29 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 15:03:21 +0100 Subject: [PATCH 5/8] 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. --- apps/labeler/src/notification-triggers.ts | 95 ++++++++++++------- .../test/notification-triggers.test.ts | 49 ++++++++++ 2 files changed, 110 insertions(+), 34 deletions(-) diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index e73e71b33f..cfef3c03b3 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/apps/labeler/src/notification-triggers.ts @@ -44,7 +44,6 @@ import type { NotificationSender, NotificationSource, SendContext, - SendOutcome, } from "./notification-send.js"; import { sendNotification } from "./notification-send.js"; import { @@ -345,34 +344,44 @@ export async function notifyEmergencyTakedown( ): Promise { const target = contactTargetFromUri(input.uri); if (!target) return; - const result = await runTrigger( + await runTrigger( deps, { type: "operator", id: input.actionId }, 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, so raise a dedicated operator signal that manual outreach is needed. - if (!input.neg && result.outcome === "no_contact") { - await emitTakedownNoContactEvent(deps, input.actionId, input.uri); - } + // 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 so the - * failed takedown notice surfaces to operators. Fire-and-forget: an insert - * failure is swallowed and logged, never propagated into the deferred tail. */ -async function emitTakedownNoContactEvent( +/** + * 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. Fire-and-forget — an + * error is swallowed and logged, never propagated into the deferred tail (a later + * replay retries anyway, since the event row is still absent). + */ +async function ensureTakedownNoContactAlert( deps: NotifyDeps, actionId: string, uri: string, ): Promise { - const now = (deps.now ?? (() => new Date()))(); 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, { @@ -391,13 +400,40 @@ async function emitTakedownNoContactEvent( buildOutboxInsert(deps.db, { eventId, channel: OPERATOR_ALERT_CHANNEL, now }), ]); } catch (error) { - console.error("[notifications] takedown no-contact event failed", { + 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 * only for `granted`/`denied`; the caller never invokes it for `withdrawn`. Source * is the resolve operator action, so a mutation replay dedups on it. */ @@ -431,13 +467,12 @@ export async function notifyProlongedError( // An unparseable URI is terminal, not transient (the URI never changes), so it // counts as processed — the cron marks it rather than re-attempting forever. if (!target) return true; - const result = await runTrigger( + return runTrigger( deps, { type: "issuance", id: assessment.id }, target, prolongedErrorNoticeContent(deps, { uri: assessment.uri, cid: assessment.cid }), ); - return result.terminal; } /** @@ -536,38 +571,30 @@ function operatorActionNeg(metadataJson: string): boolean { } } -/** The outcome of a trigger. `terminal` is false only when a TRANSIENT error was - * thrown before any row was claimed (the prolonged-error cron uses it to decide - * whether to stamp its fire-once mark). `outcome` is the `sendNotification` - * status when a send actually ran — undefined on a dedup hit or a thrown error — - * so a caller can react to a specific delivery result (e.g. `no_contact`). */ -interface TriggerResult { - terminal: boolean; - outcome?: SendOutcome["status"]; -} - /** * Dedup, resolve verification (fail-closed), send, swallow+log. Shared by every * trigger so the dedup and verified-skip policy is applied uniformly and a * notification failure can never escape into the label path. * - * A TERMINAL result is a dedup hit or a normal `sendNotification` return (sent, - * confirmation-sent, undeliverable, or a claimed-then-failed row the sweep now - * owns); a non-terminal result is a thrown TRANSIENT error (an aggregator read or - * a pre-claim D1 write that failed before any row was claimed). Fire-and-forget - * callers ignore the result. + * Returns whether the trigger reached a TERMINAL outcome — a dedup hit or a + * normal `sendNotification` return (sent, confirmation-sent, undeliverable, or a + * claimed-then-failed row the sweep now owns) — versus a thrown TRANSIENT error + * (an aggregator read or a pre-claim D1 write that failed before any row was + * claimed). The prolonged-error cron uses this to decide whether to stamp its + * fire-once mark: a transient failure returns `false` so the next tick retries + * instead of being silently swallowed. Fire-and-forget callers ignore it. */ async function runTrigger( deps: NotifyDeps, source: NotificationSource, target: ContactTarget, notice: NoticeContent, -): Promise { +): Promise { const now = deps.now ?? (() => new Date()); try { if (await sourceAlreadyProcessed(deps.db, source)) { logTrigger(source, target.did, "deduped"); - return { terminal: true }; + return true; } const verifiedPublisher = await isVerifiedPublisher( deps.aggregator, @@ -587,7 +614,7 @@ async function runTrigger( const request: NotificationRequest = { source, target, notice }; const outcome = await sendNotification(ctx, request); logTrigger(source, target.did, outcome.status); - return { terminal: true, outcome: outcome.status }; + return true; } catch (error) { console.error("[notifications] trigger failed", { sourceType: source.type, @@ -595,7 +622,7 @@ async function runTrigger( did: target.did, error: error instanceof Error ? error.message : String(error), }); - return { terminal: false }; + return false; } } diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index d351909564..1992d9e59b 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -112,6 +112,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: [], @@ -658,6 +679,34 @@ describe("emergency takedown with no resolvable contact", () => { expect(await takedownEvents(actionId)).toHaveLength(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", () => { From 8406a03141b6f442da34d42f582081e35fd6764d Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 16:26:06 +0100 Subject: [PATCH 6/8] 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. --- apps/labeler/src/notification-endpoints.ts | 26 +++++++++---------- .../test/notification-endpoints.test.ts | 18 +++++++++++++ 2 files changed, 31 insertions(+), 13 deletions(-) diff --git a/apps/labeler/src/notification-endpoints.ts b/apps/labeler/src/notification-endpoints.ts index 4dd0f52acf..198bb6efc5 100644 --- a/apps/labeler/src/notification-endpoints.ts +++ b/apps/labeler/src/notification-endpoints.ts @@ -15,9 +15,9 @@ * - 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 browser POST carries the - * token/hash in hidden form fields; an RFC 8058 one-click unsubscribe POST - * carries `c` in the header URL's query instead, so the POST parser reads the - * body first and falls back to the query. + * 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. @@ -116,11 +116,6 @@ async function readPostParams( request: Request, action: NotificationAction, ): Promise { - // The capability travels in the header URL's query string. An RFC 8058 - // one-click POST (`List-Unsubscribe=One-Click` in the body) carries `c` ONLY - // there, so the query is the fallback when the body omits it; a browser form - // POST supplies `c` in the body, which wins. - const query = new URL(request.url).searchParams; let form: FormData | null = null; try { form = await request.formData(); @@ -128,12 +123,17 @@ async function readPostParams( form = null; } const bodyHash = form?.get("c"); - const recipientHash = - typeof bodyHash === "string" && bodyHash.length > 0 ? bodyHash : (query.get("c") ?? ""); - if (action !== "confirm") return { recipientHash, token: "" }; + 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 = - typeof bodyToken === "string" && bodyToken.length > 0 ? bodyToken : (query.get("t") ?? ""); + const token = action === "confirm" && typeof bodyToken === "string" ? bodyToken : ""; return { recipientHash, token }; } diff --git a/apps/labeler/test/notification-endpoints.test.ts b/apps/labeler/test/notification-endpoints.test.ts index 057c29116d..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); From c96d88f972401843668f493efc44980152a1616d Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 16:26:22 +0100 Subject: [PATCH 7/8] 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. --- ..._operational_events_action_type_unique.sql | 13 +++ apps/labeler/src/notification-triggers.ts | 49 +++++++--- apps/labeler/src/operational-events.ts | 27 +++++- .../test/notification-triggers.test.ts | 89 +++++++++++++++++++ apps/labeler/test/operational-events.test.ts | 8 +- 5 files changed, 170 insertions(+), 16 deletions(-) create mode 100644 apps/labeler/migrations/0012_operational_events_action_type_unique.sql 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..e176ac7d65 --- /dev/null +++ b/apps/labeler/migrations/0012_operational_events_action_type_unique.sql @@ -0,0 +1,13 @@ +-- At-most-one operational event per (action_id, event_type). An action-scoped +-- event — the emergency alerts, pause/resume, dead-letter controls, +-- reconsideration open/resolve, and the deferred `takedown-no-contact` alert — +-- is emitted once per operator action; this unique index is the hard guarantee +-- behind the check-then-insert emitters, so a concurrent replay that passes the +-- existence check still converges to a single row (paired with ON CONFLICT DO +-- NOTHING on the idempotent insert). +-- +-- NULL `action_id` rows (e.g. `assessment-prolonged-error`, keyed off the +-- escalation row instead) are naturally exempt: SQLite treats each NULL as +-- distinct in a UNIQUE index, so it never constrains them. +CREATE UNIQUE INDEX idx_operational_events_action_type + ON operational_events(action_id, event_type); diff --git a/apps/labeler/src/notification-triggers.ts b/apps/labeler/src/notification-triggers.ts index cfef3c03b3..56263fc8cc 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/apps/labeler/src/notification-triggers.ts @@ -369,9 +369,12 @@ const OPERATOR_ALERT_CHANNEL = "deployment-alert"; * 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. Fire-and-forget — an - * error is swallowed and logged, never propagated into the deferred tail (a later - * replay retries anyway, since the event row is still absent). + * 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, @@ -396,8 +399,14 @@ async function ensureTakedownNoContactAlert( "Emergency takedown has no resolvable publisher contact; manual outreach required.", }, now, + idempotentOnActionType: true, + }), + buildOutboxInsert(deps.db, { + eventId, + channel: OPERATOR_ALERT_CHANNEL, + now, + gateOnEventPresent: true, }), - buildOutboxInsert(deps.db, { eventId, channel: OPERATOR_ALERT_CHANNEL, now }), ]); } catch (error) { console.error("[notifications] takedown no-contact alert failed", { @@ -719,8 +728,8 @@ 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 - * parent package slug: a package rkey IS the slug, and a release rkey is - * `slug:version`, so the slug is the part before the first `:`. That lets + * 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) @@ -729,9 +738,9 @@ function assessmentUrl(serviceUrl: string, uri: string, cid?: string): string { */ 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: packageSlugFromRkey(rkey ?? "") }; + return { did, slug: packageSlugFromRecord(collection, rkey ?? "") }; } if (uri.startsWith("did:")) { return { did: uri, slug: uri.split(":").at(-1) ?? uri }; @@ -739,13 +748,25 @@ export function contactTargetFromUri(uri: string): ContactTarget | null { return null; } -/** The parent package slug from a record rkey: the part before the `:` version - * delimiter for a release (`gallery:1.2.0` → `gallery`), or the rkey verbatim for - * a package (no delimiter) or any rkey whose delimiter is leading (no slug to - * take — left as-is so resolution degrades to the publisher tier). */ -function packageSlugFromRkey(rkey: string): string { +/** Canonical package-slug shape, mirroring the aggregator's release-rkey ingest + * validation (`records-consumer`'s PACKAGE_SLUG_RE). */ +const PACKAGE_SLUG_RE = /^[a-zA-Z][a-zA-Z0-9_-]*$/; + +/** + * 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 — 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 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(":"); - return delimiter > 0 ? rkey.slice(0, delimiter) : rkey; + if (delimiter <= 0 || delimiter === rkey.length - 1) return rkey; + const slug = rkey.slice(0, delimiter); + return PACKAGE_SLUG_RE.test(slug) ? slug : rkey; } function logTrigger(source: NotificationSource, did: string, outcome: string): void { diff --git a/apps/labeler/src/operational-events.ts b/apps/labeler/src/operational-events.ts index ef24612e8b..63633c8117 100644 --- a/apps/labeler/src/operational-events.ts +++ b/apps/labeler/src/operational-events.ts @@ -80,6 +80,14 @@ export interface OperationalEventInsert { * ticks are not serialized against each other). */ gateOnUnalertedEscalation?: { assessmentId: string }; + /** + * Appends `ON CONFLICT (action_id, event_type) DO NOTHING` to a plain insert so + * a concurrent replay converges to one row against the `(action_id, event_type)` + * unique index (migration 0012). Only valid 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. + */ + idempotentOnActionType?: boolean; } export interface OutboxInsert { @@ -90,6 +98,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.idempotentOnActionType} + * so an ON CONFLICT no-op event leaves no orphan outbox row in the same batch. */ + gateOnEventPresent?: boolean; } export interface StoredOperationalEvent { @@ -226,10 +238,13 @@ export function buildOperationalEventInsert( .bind(...values, input.gateOnUnalertedEscalation.assessmentId); } + const onConflict = input.idempotentOnActionType + ? ` ON CONFLICT (action_id, event_type) DO NOTHING` + : ""; return db .prepare( `INSERT INTO operational_events (${columns}) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)${onConflict}`, ) .bind(...values); } @@ -272,6 +287,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-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 1992d9e59b..219e7c51af 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -21,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; @@ -578,6 +583,45 @@ describe("package slug parse", () => { 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("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", () => { @@ -680,6 +724,51 @@ describe("emergency takedown with no resolvable contact", () => { 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, + idempotentOnActionType: 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"); diff --git a/apps/labeler/test/operational-events.test.ts b/apps/labeler/test/operational-events.test.ts index 560dc080a1..6edf9824c7 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); From 065fa7bc19db5bf7a9a1f33a21aeca3f178b9de0 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 17:54:35 +0100 Subject: [PATCH 8/8] 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. --- ..._operational_events_action_type_unique.sql | 25 ++++++------ apps/labeler/src/notification-triggers.ts | 28 ++++++++----- apps/labeler/src/operational-events.ts | 17 ++++---- .../test/notification-triggers.test.ts | 39 ++++++++++++++++++- apps/labeler/test/operational-events.test.ts | 31 +++++++++++++++ 5 files changed, 111 insertions(+), 29 deletions(-) diff --git a/apps/labeler/migrations/0012_operational_events_action_type_unique.sql b/apps/labeler/migrations/0012_operational_events_action_type_unique.sql index e176ac7d65..7abf31a41e 100644 --- a/apps/labeler/migrations/0012_operational_events_action_type_unique.sql +++ b/apps/labeler/migrations/0012_operational_events_action_type_unique.sql @@ -1,13 +1,14 @@ --- At-most-one operational event per (action_id, event_type). An action-scoped --- event — the emergency alerts, pause/resume, dead-letter controls, --- reconsideration open/resolve, and the deferred `takedown-no-contact` alert — --- is emitted once per operator action; this unique index is the hard guarantee --- behind the check-then-insert emitters, so a concurrent replay that passes the --- existence check still converges to a single row (paired with ON CONFLICT DO --- NOTHING on the idempotent insert). +-- 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). -- --- NULL `action_id` rows (e.g. `assessment-prolonged-error`, keyed off the --- escalation row instead) are naturally exempt: SQLite treats each NULL as --- distinct in a UNIQUE index, so it never constrains them. -CREATE UNIQUE INDEX idx_operational_events_action_type - ON operational_events(action_id, event_type); +-- 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-triggers.ts b/apps/labeler/src/notification-triggers.ts index 56263fc8cc..0978d08dc5 100644 --- a/apps/labeler/src/notification-triggers.ts +++ b/apps/labeler/src/notification-triggers.ts @@ -399,7 +399,7 @@ async function ensureTakedownNoContactAlert( "Emergency takedown has no resolvable publisher contact; manual outreach required.", }, now, - idempotentOnActionType: true, + idempotentTakedownNoContact: true, }), buildOutboxInsert(deps.db, { eventId, @@ -748,25 +748,35 @@ export function contactTargetFromUri(uri: string): ContactTarget | null { return null; } -/** Canonical package-slug shape, mirroring the aggregator's release-rkey ingest - * validation (`records-consumer`'s PACKAGE_SLUG_RE). */ +// 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 — 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 stays whole so it can - * never strip to a DIFFERENT package's slug — it misses at `getPackage` and - * resolution degrades to the publisher tier. + * 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); - return PACKAGE_SLUG_RE.test(slug) ? slug : rkey; + 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 { diff --git a/apps/labeler/src/operational-events.ts b/apps/labeler/src/operational-events.ts index 63633c8117..9aa36fad96 100644 --- a/apps/labeler/src/operational-events.ts +++ b/apps/labeler/src/operational-events.ts @@ -81,13 +81,14 @@ export interface OperationalEventInsert { */ gateOnUnalertedEscalation?: { assessmentId: string }; /** - * Appends `ON CONFLICT (action_id, event_type) DO NOTHING` to a plain insert so - * a concurrent replay converges to one row against the `(action_id, event_type)` - * unique index (migration 0012). Only valid with a non-null `actionId` and no + * 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. */ - idempotentOnActionType?: boolean; + idempotentTakedownNoContact?: boolean; } export interface OutboxInsert { @@ -99,7 +100,7 @@ export interface OutboxInsert { /** 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.idempotentOnActionType} + * 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; } @@ -238,8 +239,10 @@ export function buildOperationalEventInsert( .bind(...values, input.gateOnUnalertedEscalation.assessmentId); } - const onConflict = input.idempotentOnActionType - ? ` ON CONFLICT (action_id, event_type) DO NOTHING` + // 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( diff --git a/apps/labeler/test/notification-triggers.test.ts b/apps/labeler/test/notification-triggers.test.ts index 219e7c51af..230079599b 100644 --- a/apps/labeler/test/notification-triggers.test.ts +++ b/apps/labeler/test/notification-triggers.test.ts @@ -598,6 +598,43 @@ describe("package slug parse", () => { 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"; @@ -745,7 +782,7 @@ describe("emergency takedown with no resolvable contact", () => { labelValue: "!takedown", payload: { reason: "manual outreach required" }, now, - idempotentOnActionType: true, + idempotentTakedownNoContact: true, }), buildOutboxInsert(database, { eventId, diff --git a/apps/labeler/test/operational-events.test.ts b/apps/labeler/test/operational-events.test.ts index 6edf9824c7..b3fe18bb4a 100644 --- a/apps/labeler/test/operational-events.test.ts +++ b/apps/labeler/test/operational-events.test.ts @@ -128,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();