diff --git a/.changeset/delivery-update-op-tenant-classification.md b/.changeset/delivery-update-op-tenant-classification.md new file mode 100644 index 0000000000..2b3304a106 --- /dev/null +++ b/.changeset/delivery-update-op-tenant-classification.md @@ -0,0 +1,66 @@ +--- +"@objectstack/service-messaging": minor +"@objectstack/plugin-webhooks": minor +--- + +fix(service-messaging,plugin-webhooks): the `update`-op tenant-audit surface on the delivery outboxes is classified — `ack` is a dispatcher sweep, `redeliver` threads the caller's tenant (#10740) + +**BREAKING** signature change on `IHttpOutbox.redeliver` and +`MessagingService.redeliverHttp`, shipped as `minor` under the repo's +launch-window convention for breaking changes. + +`sys_http_delivery` and `sys_notification_delivery` carry three single-record +(`multi: false`) writes that the SQL driver audits under the **`update`** op — +a different op, and a different throttle key, from the `updateMany` half +classified previously. Their correct classifications are **opposite**, and +treating them as one sweep is the dangerous reading: + +| site | reachable from | classification | +| --- | --- | --- | +| `SqlNotificationOutbox.ack` | dispatcher tick only | global sweep | +| `SqlHttpOutbox.ack` | dispatcher tick only | global sweep | +| `SqlHttpOutbox.redeliver` | `POST /api/v1/webhooks/redeliver` | request-contextual | + +**The two `ack` sites** are declared global sweeps through a new +`dispatcherAckOptions()` helper, sibling to `dispatcherSweepOptions()` and +deliberately not the same function — that one returns `& { multi: true }`, so a +`multi: false` site cannot borrow it by accident. The warrant was re-derived +against the current tree rather than inherited: `ack` has exactly two callers, +both inside `runPartition()` on a `setInterval` tick holding a per-partition +cluster lock, so no request context exists to thread; and the row being acked +was claimed by a sweep that crosses organizations by construction +(`hash(refId | notificationId | digestKey) mod N` is a load-spreading key, and +one outbox per environment drains the whole queue). Passing the claimed row's +own `organization_id` is documented at the helper as the tempting wrong answer: +a predicate read off the row you are about to write matches exactly that row, +adds no isolation, and silences the audit anyway — the appearance of scoping +without the substance. + +**`redeliver` is not that**, and it is the reason this shipped separately. The +route in front of it is served to any authenticated user, so on a walled +deployment (`OS_TENANCY_POSTURE=isolated|group`) an unscoped replay is an +authenticated user writing another organization's delivery row — the case the +tenant audit exists to catch. It now carries the caller's tenant, applied to +the rows it reads as well as the row it writes, and it must never be given +`bypassTenantAudit`: a scoped write and a bypassed write produce the same +silence in the log, so the flag would convert a detectable hole into an +undetectable one. The webhook route resolves the session's +`activeOrganizationId` and threads it. + +Behaviour change at the endpoint: a delivery row outside the caller's +organization is now **not found** (`RESOURCE_NOT_FOUND`, HTTP 404) rather than +replayed. It is deliberately invisible rather than forbidden, so the endpoint +is not an existence oracle for other tenants' delivery ids. An in-tenant +redelivery is unchanged. + +Migrating a caller: `redeliver(id, guard?)` becomes +`redeliver(id, { tenantId, guard? })`, and `redeliverHttp(id)` becomes +`redeliverHttp(id, { tenantId })`. `tenantId` is a **required** property typed +`string | undefined`, so omitting it does not compile — a caller with no tenant +has to write `tenantId: undefined` and mean it. That is the point of the shape: +an optional property would let the dangerous case, a request path that simply +forgot, type-check in silence. Passing `undefined` leaves the write unscoped +and the audit line still fires, which is the intended reporting behaviour on a +deployment that cannot resolve an organization for the caller. + + diff --git a/packages/plugins/plugin-webhooks/src/webhook-drop-durable-record.test.ts b/packages/plugins/plugin-webhooks/src/webhook-drop-durable-record.test.ts index b4d78c77fd..68034daf16 100644 --- a/packages/plugins/plugin-webhooks/src/webhook-drop-durable-record.test.ts +++ b/packages/plugins/plugin-webhooks/src/webhook-drop-durable-record.test.ts @@ -200,7 +200,11 @@ describe('dropped webhook subscription leaves a durable, unsendable record (#806 await new HttpDispatcher({ nodeId: 'n1', outbox, fetchImpl: impl, partitionCount: 1 }).tick(); expect(calls).toHaveLength(0); - await expect(messaging.redeliverHttp(row.id)).rejects.toMatchObject({ + // [#10740] `redeliverHttp` now requires the requesting caller's tenant. + // This fixture has no organization and no tenancy posture, so + // `undefined` is the honest value — required rather than optional + // precisely so that answer is written down instead of defaulted into. + await expect(messaging.redeliverHttp(row.id, { tenantId: undefined })).rejects.toMatchObject({ code: 'DELIVERY_NEVER_SENT', }); // The refusal did not mutate the row on its way out. diff --git a/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts b/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts index c52589c393..93076eda31 100644 --- a/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts +++ b/packages/plugins/plugin-webhooks/src/webhook-outbox-plugin.ts @@ -28,7 +28,17 @@ import { interface MessagingHttpSurface { isHttpDeliveryReady(): boolean; enqueueHttp(input: EnqueueHttpInput): Promise; - redeliverHttp(id: string): Promise<{ id: string; status: string }>; + /** + * [#10740] Takes the REQUESTING caller's organization. Declared with the + * required-but-nullable `tenantId` the service declares, so this plugin + * cannot call the endpoint's backing method without deciding what tenant + * the request carries — the omission this structural view would otherwise + * type-check happily. + */ + redeliverHttp( + id: string, + options: { tenantId: string | undefined }, + ): Promise<{ id: string; status: string }>; /** * [#8069] Where this plugin's veto over redelivering `source: 'webhook'` * rows is installed. Declared REQUIRED on this structural view even though @@ -347,8 +357,21 @@ export class WebhookOutboxPlugin implements Plugin { /** * Mount POST /api/v1/webhooks/redeliver on the host Hono app, if one is - * available. Delegates to `messaging.redeliverHttp(deliveryId)`. Auth is the - * better-auth session cookie — every authenticated user counts. + * available. Delegates to `messaging.redeliverHttp(deliveryId, …)`. Auth is + * the better-auth session cookie — every authenticated user counts. + * + * [#10740] Which is precisely why the caller's ACTIVE ORGANIZATION is + * resolved here and threaded into the call. `sys_http_delivery` is + * tenant-scoped, and this is the one door on it a request can reach: an + * unscoped replay from here is an authenticated user reaching another + * organization's delivery row on a walled deployment. With the tenant + * threaded, a row outside the caller's organization is simply not found. + * + * ⚠️ A session with no active organization threads `undefined`, and the + * driver's tenant-audit line then fires for that write. That is deliberate: + * the deployment could not tell us who is asking, and reporting the gap is + * the correct outcome. ⛔ It is never repaired with `bypassTenantAudit`, + * which would silence the report without closing anything. */ private registerAdminRoutes(ctx: PluginContext): void { const http = this.tryGetService(ctx, ['http-server']); @@ -361,8 +384,9 @@ export class WebhookOutboxPlugin implements Plugin { if (!rawApp || !messaging) return; rawApp.post('/api/v1/webhooks/redeliver', async (c: any) => { - const userId = await this.resolveSessionUserId(ctx, c); - if (!userId) { + const session = await this.resolveSession(ctx, c); + const userId = session?.user?.id; + if (typeof userId !== 'string' || userId.length === 0) { return c.json( { success: false, error: { code: 'UNAUTHENTICATED', message: 'Sign in to redeliver webhook deliveries.' } }, 401, @@ -382,8 +406,20 @@ export class WebhookOutboxPlugin implements Plugin { ); } try { - const row = await messaging.redeliverHttp(deliveryId); - ctx.logger.info?.('[webhook-outbox] redelivered', { deliveryId, requestedBy: userId }); + // [#10740] `session.session.activeOrganizationId` is the + // canonical spelling of the caller's active organization + // (better-auth's organization plugin; see + // `plugin-auth/auth-schema-config.ts`). Read from the one + // place it lives — a `??` chain over alternative spellings + // would make a MISSING organization indistinguishable from a + // differently-shaped one, and the missing case is the one that + // must stay visible. + const activeOrg = session?.session?.activeOrganizationId; + const tenantId = typeof activeOrg === 'string' && activeOrg.length > 0 + ? activeOrg + : undefined; + const row = await messaging.redeliverHttp(deliveryId, { tenantId }); + ctx.logger.info?.('[webhook-outbox] redelivered', { deliveryId, requestedBy: userId, tenantId }); return c.json({ success: true, data: { id: row.id, status: row.status } }); } catch (err: any) { const code = err?.code; @@ -412,7 +448,18 @@ export class WebhookOutboxPlugin implements Plugin { ctx.logger.info?.('[webhook-outbox] redeliver endpoint mounted at POST /api/v1/webhooks/redeliver'); } - private async resolveSessionUserId(ctx: PluginContext, c: any): Promise { + /** + * [#10740] The better-auth session envelope (`{ user, session }`) for this + * request, or `undefined`. + * + * Widened from the previous `resolveSessionUserId` because the route now + * needs two facts from ONE lookup: who is asking (`user.id`, the + * authentication gate) and which organization they are asking as + * (`session.activeOrganizationId`, the tenant threaded into the write). + * Resolving them separately would mean two `getSession` calls that can + * disagree. + */ + private async resolveSession(ctx: PluginContext, c: any): Promise { try { const authService: any = this.tryGetService(ctx, ['auth']); if (!authService) return undefined; @@ -421,9 +468,7 @@ export class WebhookOutboxPlugin implements Plugin { api = await authService.getApi(); } if (!api?.getSession) return undefined; - const session = await api.getSession({ headers: c.req.raw.headers }); - const uid = session?.user?.id; - return typeof uid === 'string' && uid.length > 0 ? uid : undefined; + return await api.getSession({ headers: c.req.raw.headers }); } catch { return undefined; } diff --git a/packages/plugins/plugin-webhooks/src/webhook-redeliver-tenant-scope.test.ts b/packages/plugins/plugin-webhooks/src/webhook-redeliver-tenant-scope.test.ts new file mode 100644 index 0000000000..34cf523427 --- /dev/null +++ b/packages/plugins/plugin-webhooks/src/webhook-redeliver-tenant-scope.test.ts @@ -0,0 +1,167 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #10740 — `POST /api/v1/webhooks/redeliver` carries the CALLER'S tenant. + * + * This route is the reason `SqlHttpOutbox.redeliver` is classified + * request-contextual rather than as a dispatcher sweep: its auth gate is "any + * authenticated user", and `sys_http_delivery` is a tenant-scoped object. On a + * walled deployment (`OS_TENANCY_POSTURE=isolated`) an unscoped replay from + * here is an authenticated user writing another organization's delivery row — + * exactly what the driver's tenant audit exists to catch, and the one site + * where silencing that audit would convert a detectable hole into an + * undetectable one. + * + * ## What this file pins that the service-level test cannot + * The tenant has to come from the REQUEST. `service-messaging`'s + * `delivery-update-tenant-audit.integration.test.ts` proves the outbox applies + * whatever tenant it is handed, right down to the options that reach the + * driver; nothing there can prove the route hands it the right one, or hands + * it anything at all. Here the session is the only source of the value, so a + * route that dropped it would go red. + * + * It also pins the HTTP half of the ADR-0112 envelope: the service layer + * carries the `code`, and the `status` exists only at this boundary. Both are + * asserted for the cross-tenant refusal — a `code` assertion alone would not + * notice the refusal surfacing as a 500. + */ + +import { describe, it, expect } from 'vitest'; +import { WebhookOutboxPlugin } from './webhook-outbox-plugin.js'; + +/** Captures the handler `registerAdminRoutes` mounts, and lets us call it. */ +function mountRoute(opts: { + session: any; + redeliverHttp: (id: string, options: { tenantId: string | undefined }) => Promise; +}): { post: (body: any) => Promise<{ status: number; json: any }> } { + let handler: ((c: any) => Promise) | undefined; + const rawApp = { + post(path: string, h: (c: any) => Promise) { + if (path === '/api/v1/webhooks/redeliver') handler = h; + }, + }; + const services: Record = { + 'http-server': { getRawApp: () => rawApp }, + messaging: { + // `getMessaging` gates on this being a function. + enqueueHttp: async () => 'unused', + isHttpDeliveryReady: () => true, + registerRedeliverGuard: () => {}, + redeliverHttp: opts.redeliverHttp, + }, + auth: { api: { getSession: async () => opts.session } }, + }; + const ctx: any = { + getService: (n: string) => services[n], + logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, + }; + (new WebhookOutboxPlugin() as any).registerAdminRoutes(ctx); + if (!handler) throw new Error('route was not mounted'); + + return { + async post(body: any) { + let status = 200; + let json: any; + const c = { + req: { raw: { headers: new Headers() }, json: async () => body }, + json(payload: any, s?: number) { + json = payload; + if (s !== undefined) status = s; + return { status, json }; + }, + }; + await handler!(c); + return { status, json }; + }, + }; +} + +/** A better-auth session envelope: `{ user, session }`. */ +const sessionFor = (userId: string, activeOrganizationId?: string) => ({ + user: { id: userId }, + session: { userId, ...(activeOrganizationId ? { activeOrganizationId } : {}) }, +}); + +describe('POST /api/v1/webhooks/redeliver — the caller\'s tenant reaches the outbox (#10740)', () => { + it('threads the session\'s active organization into redeliverHttp', async () => { + const seen: Array<{ id: string; tenantId: string | undefined }> = []; + const route = mountRoute({ + session: sessionFor('user_1', 'org_a'), + async redeliverHttp(id, options) { + seen.push({ id, tenantId: options.tenantId }); + return { id, status: 'pending' }; + }, + }); + + const res = await route.post({ deliveryId: 'del_1' }); + + expect(res.status).toBe(200); + expect(res.json).toEqual({ success: true, data: { id: 'del_1', status: 'pending' } }); + // The whole point: not `undefined`, and not some other org. + expect(seen).toEqual([{ id: 'del_1', tenantId: 'org_a' }]); + }); + + it('refuses a cross-tenant delivery id with RESOURCE_NOT_FOUND and 404', async () => { + // The outbox scopes its reads by the tenant it is handed, so a row in + // another organization is INVISIBLE rather than forbidden — which is + // also what stops this endpoint being an existence oracle for other + // tenants' delivery ids. + const route = mountRoute({ + session: sessionFor('user_1', 'org_a'), + async redeliverHttp() { + const err: any = new Error("Delivery row 'del_other' not found"); + err.name = 'HttpRedeliverError'; + err.code = 'RESOURCE_NOT_FOUND'; + throw err; + }, + }); + + const res = await route.post({ deliveryId: 'del_other' }); + + // ADR-0112: `code` AND `status`. Either alone passes on a refusal that + // surfaced as a 500, or on a 404 carrying the wrong code. + expect(res.json?.error?.code).toBe('RESOURCE_NOT_FOUND'); + expect(res.status).toBe(404); + }); + + it('threads `undefined` for a session with no active organization — reported, never silenced', async () => { + // The honest half. The deployment could not tell the route which + // organization is asking, so the write goes out unscoped and the + // driver's tenant-audit line fires for it. ⛔ The repair for that is + // never `bypassTenantAudit`, which would hide the report and close + // nothing — so what this pins is that the route invents no tenant. + const seen: Array = []; + const route = mountRoute({ + session: sessionFor('user_1'), + async redeliverHttp(id, options) { + seen.push(options.tenantId); + return { id, status: 'pending' }; + }, + }); + + const res = await route.post({ deliveryId: 'del_1' }); + + expect(res.status).toBe(200); + expect(seen).toEqual([undefined]); + }); + + it('never reaches the outbox at all for an unauthenticated caller', async () => { + // The pre-existing gate, re-pinned because the session lookup was + // widened from "user id" to the whole envelope: a widening that lost + // the auth check would be invisible to every assertion above. + let called = 0; + const route = mountRoute({ + session: null, + async redeliverHttp(id) { + called += 1; + return { id, status: 'pending' }; + }, + }); + + const res = await route.post({ deliveryId: 'del_1' }); + + expect(res.status).toBe(401); + expect(res.json?.error?.code).toBe('UNAUTHENTICATED'); + expect(called).toBe(0); + }); +}); diff --git a/packages/services/service-messaging/src/delivery-headers-at-rest.integration.test.ts b/packages/services/service-messaging/src/delivery-headers-at-rest.integration.test.ts index df90fbbd50..feffb3e028 100644 --- a/packages/services/service-messaging/src/delivery-headers-at-rest.integration.test.ts +++ b/packages/services/service-messaging/src/delivery-headers-at-rest.integration.test.ts @@ -55,6 +55,15 @@ import { HttpDispatcher } from './http-dispatcher.js'; import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; import type { FetchImpl } from './http-sender.js'; +/** + * [#10740] `IHttpOutbox.redeliver` now requires its caller to state the + * requesting tenant. These fixtures boot an engine with no tenancy posture and + * no organization, so `undefined` is the honest answer — the property is + * required, not optional, precisely so that answer has to be written down. + */ +const NO_TENANT = { tenantId: undefined } as const; + + /** A credential a real deployment would put in `headers`. Distinctive on purpose. */ const BEARER = 'Bearer prod_tok_8118_do_not_serve'; /** The flow half's credential — "interpolated per run", so a run-scoped value. */ @@ -222,7 +231,7 @@ describe('sys_http_delivery — authored headers vs the data API (#8118)', () => // `redeliver()` itself returns the REDACTED view (it is an admin verb, // not a dispatch path)… - const redelivered = await outbox.redeliver(id); + const redelivered = await outbox.redeliver(id, NO_TENANT); expect(redelivered.status).toBe('pending'); expect(redelivered.headers).toBeUndefined(); diff --git a/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts b/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts new file mode 100644 index 0000000000..14d1ed1050 --- /dev/null +++ b/packages/services/service-messaging/src/delivery-update-tenant-audit.integration.test.ts @@ -0,0 +1,332 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #10740 — the `update`-op half of the delivery tenant-audit surface, where + * three single-record writes on two objects carry TWO OPPOSITE + * classifications. + * + * | site | reachable from | classification | + * | ------------------------------- | ------------------ | ------------------- | + * | `SqlNotificationOutbox.ack` | dispatcher tick | global sweep | + * | `SqlHttpOutbox.ack` | dispatcher tick | global sweep | + * | `SqlHttpOutbox.redeliver` | POST /api/v1/… | request-contextual | + * + * The `ack` pair is declared global (`dispatcherAckOptions`, warrant in + * `outbox-dispatcher-scope.ts`). `redeliver` is NOT: it is served to any + * authenticated user, so it threads the caller's tenant instead. ⛔ A + * `bypassTenantAudit` on that third site would convert a detectable hole into + * an undetectable one, which is why the assertions below are not written + * against the audit line alone. + * + * ## Why the audit line is not a sufficient assertion here + * A SCOPED write and a BYPASSED write produce the SAME silence in the log. + * An assertion that only checked "no `[tenant-audit]` line for `redeliver`" + * would therefore pass on the one implementation this card forbids. So the + * `redeliver` tests assert **the options that actually reached the driver** — + * `tenantId` present, `bypassTenantAudit` absent — through a spy installed on + * `SqlDriver.update` itself, the method that both applies the tenant scope and + * decides the audit. + * + * ## Why a real delivery is driven through + * `ack` runs only once a delivery has actually been processed, unlike the + * claim path's reap `UPDATE` which runs on every tick. A boot log on an empty + * queue shows the two `updateMany` lines and says NOTHING about this op — an + * audit line that is absent because nothing ran is NOT MEASURED, not a pass. + * Every `ack` test below therefore runs a real dispatcher tick and then pins + * that the attempt was recorded (`status`, `attempts`), so a silent run cannot + * be mistaken for a clean one. + * + * ## The vacuity traps closed explicitly + * 1. **"the audit was never armed."** Every silence assertion is followed by + * a positive control on the SAME object through the SAME driver: an + * unscoped by-id `update` that MUST produce the line. The gate throttles + * one warning per `${object}:${op}`, so the control runs last and only + * fires if the production path consumed no `update` warning of its own. + * 2. **"a fix that touches nothing."** Row state is pinned after every write. + * 3. **"a refusal that refuses everything."** The cross-tenant refusal is + * paired with a still-works leg: an in-tenant redeliver still succeeds. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; +import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; +import { HttpDispatcher } from './http-dispatcher.js'; +import { NotificationDispatcher } from './dispatcher.js'; +import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; +import type { FetchImpl } from './http-sender.js'; +import type { MessagingChannel } from './channel.js'; + +const OLD_POSTURE = process.env.OS_TENANCY_POSTURE; +const OLD_AUDIT = process.env.OS_TENANT_AUDIT; + +let engine: ObjectQL; +let driver: SqlDriver; +let warns: Array<{ msg: string; meta: any }>; +/** Every `options` bag that reached `SqlDriver.update` — the `update` op only. */ +let driverUpdates: Array<{ object: string; id: unknown; options: any }>; + +/** The audit line for the SINGLE-RECORD op, matched on object + op. */ +const auditedUpdate = (object: string): boolean => + warns.some((w) => w.msg.includes(`[tenant-audit] update on tenant-scoped object "${object}"`)); + +beforeEach(async () => { + // Read LIVE by `isMultiTenantMode()` (#5262), so this really arms the gate. + process.env.OS_TENANCY_POSTURE = 'isolated'; + delete process.env.OS_TENANT_AUDIT; + + driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + warns = []; + driverUpdates = []; + (driver as any).logger = { warn: (msg: string, meta: any) => warns.push({ msg, meta }) }; + + // Spy on the driver's by-id UPDATE — the method that calls + // `auditMissingTenant(object, 'update', options)` AND `applyTenantScope`. + // Recording here rather than at the engine is deliberate: the engine may + // add or withhold `tenantId` (`buildDriverOptions`), so the driver's + // argument is the only reading of what the write was actually scoped by. + const realUpdate = (driver as any).update.bind(driver); + (driver as any).update = async (object: string, id: unknown, data: any, options: any) => { + driverUpdates.push({ object, id, options }); + return realUpdate(object, id, data, options); + }; + + engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(HttpDelivery as any, '@objectstack/service-messaging'); + engine.registry.registerObject(NotificationDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); +}); + +afterEach(async () => { + try { await engine?.destroy(); } catch { /* noop */ } + if (OLD_POSTURE === undefined) delete process.env.OS_TENANCY_POSTURE; + else process.env.OS_TENANCY_POSTURE = OLD_POSTURE; + if (OLD_AUDIT === undefined) delete process.env.OS_TENANT_AUDIT; + else process.env.OS_TENANT_AUDIT = OLD_AUDIT; +}); + +/** + * The positive control for the `update` op. A by-id write with no `tenantId` + * and no bypass MUST produce the audit line on `object`, or this file cannot + * tell "classified" from "the gate was never armed". Run AFTER the assertion + * it guards — the gate throttles one warning per `${object}:${op}`. + */ +async function controlUnscopedUpdate(object: string, existingId: string): Promise { + // `where: { id }` with a scalar id routes through `driver.update` + // (`resolveEngineUpdateDispatch` → `by-id`), exactly as the production + // paths under test do. + // + // ⚠️ It must name a row that EXISTS. The engine's by-id branch raises + // `Record not found` before it ever reaches the driver, so a control + // pointed at a missing id never arms the gate it is meant to prove is + // armed — it fails as an error rather than reporting a vacuous suite, + // which is the only reason that mistake was visible here. + await engine.update(object, { attempts: 99 }, { where: { id: existingId } } as any); + expect( + auditedUpdate(object), + `positive control failed: an unscoped by-id update on ${object} produced no [tenant-audit] ` + + 'line, so every "no finding" assertion in this file is vacuous', + ).toBe(true); +} + +function okFetch(): { impl: FetchImpl; calls: string[] } { + const calls: string[] = []; + const impl: FetchImpl = async (url) => { + calls.push(url); + return { ok: true, status: 204, async text() { return ''; } }; + }; + return { impl, calls }; +} + +async function seedHttpRow(id: string, org: string, over: Record = {}): Promise { + const now = new Date(); + await engine.insert(SYS_HTTP_DELIVERY, { + id, + source: 'webhook', + ref_id: id, + dedup_key: id, + url: 'https://receiver.example/hook', + method: 'POST', + payload_json: '{}', + partition_key: 0, + status: 'pending', + attempts: 0, + organization_id: org, + created_at: now, + updated_at: now, + ...over, + } as any); +} + +/** A terminal, genuinely-attempted row — the only kind `redeliver` accepts. */ +async function seedDeadRow(id: string, org: string): Promise { + await seedHttpRow(id, org, { status: 'dead', attempts: 3, error: 'receiver down' }); +} + +// ─────────────────────────────────────────────────────────────────────────── +describe('ack — the two dispatcher sites are a classified global sweep (update op)', () => { + it('SqlHttpOutbox.ack records a REAL delivery in every organization, without a finding', async () => { + // The gate's own precondition: this object really is tenant-scoped. + expect((driver as any).resolveTenantField(SYS_HTTP_DELIVERY)).toBe('organization_id'); + + await seedHttpRow('h_a', 'org_a'); + await seedHttpRow('h_b', 'org_b'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + const { impl, calls } = okFetch(); + // A real tick: claim → POST → ack. Nothing is hand-called. + await new HttpDispatcher({ + nodeId: 'n1', outbox, fetchImpl: impl, partitionCount: 1, intervalMs: 10_000, + }).tick(); + + // ① The delivery ACTUALLY ran — otherwise the silence below is + // NOT MEASURED rather than clean. + expect(calls).toHaveLength(2); + const rows = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(rows.map((r) => `${r.id}:${r.organization_id}:${r.status}:${r.attempts}`).sort()).toEqual([ + 'h_a:org_a:success:1', + 'h_b:org_b:success:1', + ]); + // ② Both organizations' rows were acked by one dispatcher — the + // cross-organization reach is the operation's semantics. + const ackWrites = driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY); + expect(ackWrites.map((u) => u.id).sort()).toEqual(['h_a', 'h_b']); + // ③ …under the DECLARED classification, not an accidental silence. + expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true); + expect(ackWrites.every((u) => u.options?.tenantId === undefined)).toBe(true); + expect(auditedUpdate(SYS_HTTP_DELIVERY)).toBe(false); + + await controlUnscopedUpdate(SYS_HTTP_DELIVERY, 'h_a'); + }); + + it('SqlNotificationOutbox.ack records a REAL delivery in every organization, without a finding', async () => { + expect((driver as any).resolveTenantField(DELIVERY_OBJECT)).toBe('organization_id'); + + const outbox = new SqlNotificationOutbox(engine as any, { partitionCount: 1 }); + const idA = await outbox.enqueue({ + notificationId: 'n_a', recipientId: 'u_a', channel: 'inbox', organizationId: 'org_a', payload: {}, + } as any); + await outbox.enqueue({ + notificationId: 'n_b', recipientId: 'u_b', channel: 'inbox', organizationId: 'org_b', payload: {}, + } as any); + + let sent = 0; + const channel: MessagingChannel = { + id: 'inbox', + async send() { sent += 1; return { ok: true }; }, + }; + await new NotificationDispatcher({ + nodeId: 'n1', + outbox, + channels: { getChannel: (id: string) => (id === 'inbox' ? channel : undefined) } as any, + channelContext: { logger: { info: () => {}, warn: () => {}, error: () => {} } }, + partitionCount: 1, + intervalMs: 10_000, + }).tick(); + + // ① Both deliveries really were sent and really were acked. + expect(sent).toBe(2); + const rows = (await engine.find(DELIVERY_OBJECT, { where: {} })) as any[]; + expect(rows.map((r) => `${r.organization_id}:${r.status}:${r.attempts}`).sort()).toEqual([ + 'org_a:success:1', + 'org_b:success:1', + ]); + // ② Declared global, for both organizations' rows. + const ackWrites = driverUpdates.filter((u) => u.object === DELIVERY_OBJECT); + expect(ackWrites).toHaveLength(2); + expect(ackWrites.every((u) => u.options?.bypassTenantAudit === true)).toBe(true); + expect(auditedUpdate(DELIVERY_OBJECT)).toBe(false); + + await controlUnscopedUpdate(DELIVERY_OBJECT, idA); + }); +}); + +// ─────────────────────────────────────────────────────────────────────────── +describe('redeliver — the request-reachable site is SCOPED, never bypassed', () => { + /** + * The assertion this card exists for. A bypassed `redeliver` and a scoped + * `redeliver` are indistinguishable from the log, so the distinction is + * drawn where it is real: in the options the driver received. + */ + it('threads the caller\'s tenant to the driver and carries NO bypass', async () => { + expect((driver as any).resolveTenantField(SYS_HTTP_DELIVERY)).toBe('organization_id'); + await seedDeadRow('h_a', 'org_a'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + const replayed = await outbox.redeliver('h_a', { tenantId: 'org_a' }); + expect(replayed.status).toBe('pending'); + + const writes = driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY); + expect(writes).toHaveLength(1); + // ⛔ The forbidden implementation, named: a bypass here would silence + // the audit for an authenticated user's unscoped write. + expect(writes[0].options?.bypassTenantAudit).toBeUndefined(); + // …and the remedy that replaces it, present. + expect(writes[0].options?.tenantId).toBe('org_a'); + // The line is absent BECAUSE the write is scoped — the two assertions + // above are what make this one mean something. + expect(auditedUpdate(SYS_HTTP_DELIVERY)).toBe(false); + + await controlUnscopedUpdate(SYS_HTTP_DELIVERY, 'h_a'); + }); + + it('refuses a cross-tenant redeliver — the row is not found, not merely forbidden', async () => { + await seedDeadRow('h_b', 'org_b'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + // ADR-0112: assert the CODE, not merely that something threw. The HTTP + // status this code maps to (404) is pinned at the route, in + // plugin-webhooks' `webhook-redeliver-tenant-scope.test.ts`. + await expect(outbox.redeliver('h_b', { tenantId: 'org_a' })).rejects.toMatchObject({ + name: 'HttpRedeliverError', + code: 'RESOURCE_NOT_FOUND', + }); + + // The refusal ran before any write — the row is untouched, not reset. + const [row] = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(`${row.status}:${row.attempts}`).toBe('dead:3'); + expect(driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY)).toHaveLength(0); + }); + + it('still works: an in-tenant redeliver succeeds while a foreign one does not', async () => { + // The still-works leg. An implementation that refused EVERY redeliver + // would score green on the refusal test alone. + await seedDeadRow('h_a', 'org_a'); + await seedDeadRow('h_b', 'org_b'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + const replayed = await outbox.redeliver('h_a', { tenantId: 'org_a' }); + expect(`${replayed.id}:${replayed.status}:${replayed.attempts}`).toBe('h_a:pending:0'); + await expect(outbox.redeliver('h_b', { tenantId: 'org_a' })).rejects.toMatchObject({ + code: 'RESOURCE_NOT_FOUND', + }); + + // One row moved, the other did not — the predicate discriminates. + const rows = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(rows.map((r) => `${r.id}:${r.status}`).sort()).toEqual(['h_a:pending', 'h_b:dead']); + }); + + it('a tenant-less caller is NOT silenced — the audit still reports the gap', async () => { + // The honest half of `tenantId: string | undefined`. Passing + // `undefined` leaves the write unscoped, and the finding is REPORTED + // rather than suppressed. If this ever goes green-by-silence, someone + // has reached for `bypassTenantAudit` on this path. + await seedDeadRow('h_a', 'org_a'); + + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + await outbox.redeliver('h_a', { tenantId: undefined }); + + const writes = driverUpdates.filter((u) => u.object === SYS_HTTP_DELIVERY); + expect(writes).toHaveLength(1); + expect(writes[0].options?.bypassTenantAudit).toBeUndefined(); + expect(auditedUpdate(SYS_HTTP_DELIVERY)).toBe(true); + }); +}); diff --git a/packages/services/service-messaging/src/http-outbox-parked-record.integration.test.ts b/packages/services/service-messaging/src/http-outbox-parked-record.integration.test.ts index 6a09f9e17a..d76fabaf05 100644 --- a/packages/services/service-messaging/src/http-outbox-parked-record.integration.test.ts +++ b/packages/services/service-messaging/src/http-outbox-parked-record.integration.test.ts @@ -55,6 +55,15 @@ import { HttpRedeliverError, type IHttpOutbox } from './http-outbox.js'; import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; import type { FetchImpl } from './http-sender.js'; +/** + * [#10740] `IHttpOutbox.redeliver` now requires its caller to state the + * requesting tenant. These fixtures boot an engine with no tenancy posture and + * no organization, so `undefined` is the honest answer — the property is + * required, not optional, precisely so that answer has to be written down. + */ +const NO_TENANT = { tenantId: undefined } as const; + + const SECRET = 'whsec_8069_signing_key'; const DROP_REASON = @@ -131,7 +140,7 @@ describe('sys_http_delivery — parked drop records are not redeliverable (#8069 updated_at: now, }); - await expect(outbox.redeliver('parked_1')).rejects.toMatchObject({ + await expect(outbox.redeliver('parked_1', NO_TENANT)).rejects.toMatchObject({ name: 'HttpRedeliverError', code: 'DELIVERY_NEVER_SENT', }); @@ -178,7 +187,7 @@ describe('sys_http_delivery — parked drop records are not redeliverable (#8069 await new HttpDispatcher({ nodeId: 'n1', outbox, fetchImpl: impl, partitionCount: 1 }).tick(); expect(calls).toHaveLength(0); // …and an operator cannot conjure a first delivery out of it. - await expect(outbox.redeliver(id)).rejects.toMatchObject({ code: 'DELIVERY_NEVER_SENT' }); + await expect(outbox.redeliver(id, NO_TENANT)).rejects.toMatchObject({ code: 'DELIVERY_NEVER_SENT' }); }); it('converges duplicates like enqueue does — one discarded event, one record', async () => { @@ -221,9 +230,12 @@ describe('sys_http_delivery — parked drop records are not redeliverable (#8069 const seen: string[] = []; await expect( - outbox.redeliver(id, (row) => { - seen.push(row.refId); - return 'the sys_webhook subscription no longer exists'; + outbox.redeliver(id, { + ...NO_TENANT, + guard: (row) => { + seen.push(row.refId); + return 'the sys_webhook subscription no longer exists'; + }, }), ).rejects.toMatchObject({ code: 'DELIVERY_NOT_ELIGIBLE' }); @@ -244,7 +256,7 @@ describe('sys_http_delivery — parked drop records are not redeliverable (#8069 await outbox.ack(id, { success: false, dead: true, error: 'boom', durationMs: 1 }); await expect( - outbox.redeliver(id, () => { throw new Error('sys_webhook read failed'); }), + outbox.redeliver(id, { ...NO_TENANT, guard: () => { throw new Error('sys_webhook read failed'); } }), ).rejects.toMatchObject({ code: 'DELIVERY_NOT_ELIGIBLE' }); const [row] = await outbox.list(); expect(row.status).toBe('dead'); @@ -271,7 +283,7 @@ describe('sys_http_delivery — parked drop records are not redeliverable (#8069 }); await outbox.ack(id, { success: false, dead: true, error: 'receiver down', durationMs: 1 }); - const replayed = await outbox.redeliver(id); + const replayed = await outbox.redeliver(id, NO_TENANT); expect(replayed.status).toBe('pending'); const { impl, calls } = makeFetch(); @@ -307,8 +319,8 @@ describe.each<[string, () => IHttpOutbox]>([ // Never claimed… expect(await outbox.claim({ nodeId: 'n1', limit: 10, claimTtlMs: 1000 })).toHaveLength(0); // …and never redeliverable. - await expect(outbox.redeliver(id)).rejects.toBeInstanceOf(HttpRedeliverError); - await expect(outbox.redeliver(id)).rejects.toMatchObject({ code: 'DELIVERY_NEVER_SENT' }); + await expect(outbox.redeliver(id, NO_TENANT)).rejects.toBeInstanceOf(HttpRedeliverError); + await expect(outbox.redeliver(id, NO_TENANT)).rejects.toMatchObject({ code: 'DELIVERY_NEVER_SENT' }); }); it('refuses the parked discriminator at the delivery door', async () => { diff --git a/packages/services/service-messaging/src/http-outbox.ts b/packages/services/service-messaging/src/http-outbox.ts index 7e3c5d8df4..da55ea0add 100644 --- a/packages/services/service-messaging/src/http-outbox.ts +++ b/packages/services/service-messaging/src/http-outbox.ts @@ -216,6 +216,50 @@ export type RedeliverGuard = ( row: HttpDelivery, ) => Promise | string | undefined; +/** + * [#10740] Everything {@link IHttpOutbox.redeliver} needs from its CALLER — + * the requesting tenant, and the producer's veto. + * + * ## Why the tenant is a REQUIRED property typed `string | undefined` + * `redeliver` is the one door on `sys_http_delivery` that a request can reach: + * `POST /api/v1/webhooks/redeliver` is served to any authenticated user. The + * object is tenant-scoped (the kernel provisions `organization_id`), so on a + * walled deployment (`OS_TENANCY_POSTURE=isolated|group`) a `redeliver` that + * does not carry its caller's tenant is an unscoped write reachable by a user + * — precisely the finding the driver's `auditMissingTenant` gate exists to + * raise. + * + * ⛔ The remedy for that finding is this field, and it is NEVER + * `bypassTenantAudit`. Silencing the line would leave the hole and remove the + * only thing that reports it: a scoped write and a bypassed write are + * indistinguishable from the log, so the flag converts a detectable defect + * into an undetectable one. The dispatcher-side classification + * (`outbox-dispatcher-scope.ts`) does not extend here and must not be + * borrowed — that warrant rests on there being no request context at all. + * + * The property is REQUIRED so that omitting it cannot compile, and typed + * `string | undefined` so a genuinely tenant-less caller (an unwalled + * single-tenant deployment, an internal tool, a test) has to write + * `tenantId: undefined` and mean it. An optional property would let the + * dangerous case — a request path that simply forgot — pass type-checking + * silently, which is the shape this contract change exists to remove. + * + * ⚠️ Passing `undefined` is honest, not a bypass: the write stays unscoped and + * the audit line still fires. That is the intended behaviour on a deployment + * that cannot resolve an organization for the caller — the gap is reported + * rather than hidden. + */ +export interface RedeliverOptions { + /** + * Organization id of the caller requesting the replay, threaded to the + * driver as `DriverOptions.tenantId`. `undefined` only when the caller + * genuinely has no tenant (see the interface docs). + */ + tenantId: string | undefined; + /** Optional producer verdict; see {@link RedeliverGuard}. */ + guard?: RedeliverGuard; +} + export interface HttpClaimOptions { /** Identifier of the node doing the claim (for `claimedBy`). */ nodeId: string; @@ -441,12 +485,21 @@ export interface IHttpOutbox { * signature). Throws {@link HttpRedeliverError}. * * [#8069] Implementations MUST call {@link assertHttpRedeliverable} before - * writing anything, and MUST consult `guard` — the producer's verdict on - * whether the configuration this row depends on is still available — with - * the same "refuse before you write" ordering. A refused redelivery leaves - * the row exactly as it was. + * writing anything, and MUST consult `options.guard` — the producer's + * verdict on whether the configuration this row depends on is still + * available — with the same "refuse before you write" ordering. A refused + * redelivery leaves the row exactly as it was. + * + * [#10740] Implementations backed by a tenant-scoped store MUST apply + * `options.tenantId` to the rows they read and to the row they write, so a + * caller can only replay a delivery inside its own organization. A row in + * another tenant is INVISIBLE, not forbidden — the refusal surfaces as + * `RESOURCE_NOT_FOUND`, which is also what keeps this endpoint from being + * an existence oracle for other tenants' delivery ids. ⛔ An implementation + * must never reach for `bypassTenantAudit` here; see + * {@link RedeliverOptions}. * - * @param guard Optional producer verdict; see {@link RedeliverGuard}. + * @param options The caller's tenant and the producer's veto. */ - redeliver(id: string, guard?: RedeliverGuard): Promise; + redeliver(id: string, options: RedeliverOptions): Promise; } diff --git a/packages/services/service-messaging/src/http-signature-at-rest.integration.test.ts b/packages/services/service-messaging/src/http-signature-at-rest.integration.test.ts index 6b70f10266..e2eeeb57c7 100644 --- a/packages/services/service-messaging/src/http-signature-at-rest.integration.test.ts +++ b/packages/services/service-messaging/src/http-signature-at-rest.integration.test.ts @@ -45,6 +45,15 @@ import { HttpDispatcher } from './http-dispatcher.js'; import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; import type { FetchImpl } from './http-sender.js'; +/** + * [#10740] `IHttpOutbox.redeliver` now requires its caller to state the + * requesting tenant. These fixtures boot an engine with no tenancy posture and + * no organization, so `undefined` is the honest answer — the property is + * required, not optional, precisely so that answer has to be written down. + */ +const NO_TENANT = { tenantId: undefined } as const; + + /** The secret under test — distinctive enough that a byte-scan cannot miss it. */ const SECRET = 'whsec_7722_do_not_persist_me'; @@ -197,7 +206,7 @@ describe('sys_http_delivery — signing secret at rest (#7722)', () => { // Redeliver: a terminal row reset to pending and sent again, which is // where a resolve-at-send-time design would need the secret back. - await outbox.redeliver(id); + await outbox.redeliver(id, NO_TENANT); const second = makeFetch(); await new HttpDispatcher({ nodeId: 'n2', outbox, fetchImpl: second.impl, partitionCount: 1 }).tick(); diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 1a49daf1ac..c036cbbe57 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -121,6 +121,7 @@ export type { HttpAckFailure, UndeliverableHttpInput, RedeliverGuard, + RedeliverOptions, } from './http-outbox.js'; export { HttpRedeliverError, diff --git a/packages/services/service-messaging/src/memory-http-outbox.ts b/packages/services/service-messaging/src/memory-http-outbox.ts index 68a2ba904e..1b3a0d2106 100644 --- a/packages/services/service-messaging/src/memory-http-outbox.ts +++ b/packages/services/service-messaging/src/memory-http-outbox.ts @@ -13,7 +13,7 @@ import { type HttpDelivery, type HttpDeliveryStatus, type IHttpOutbox, - type RedeliverGuard, + type RedeliverOptions, type UndeliverableHttpInput, } from './http-outbox.js'; @@ -156,14 +156,26 @@ export class MemoryHttpOutbox implements IHttpOutbox { return all; } - async redeliver(id: string, guard?: RedeliverGuard): Promise { + /** + * [#10740] `options.tenantId` is accepted and deliberately not applied: + * this outbox is a `Map` of in-process rows with no tenant column and no + * driver behind it, so there is nothing to scope and nothing that could + * emit a tenant-audit finding. It is a test/dev double, never the + * request-reachable production store — `SqlHttpOutbox.redeliver` is the + * site that owes the isolation, and it applies the field. + * + * ⚠️ Stated rather than left implicit, because "the parameter is ignored" + * and "the isolation is missing" look identical from a call site. A future + * memory implementation that DOES store a tenant owes the predicate here. + */ + async redeliver(id: string, options: RedeliverOptions): Promise { const row = this.rows.get(id); if (!row) { throw new HttpRedeliverError(`Delivery row '${id}' not found`, 'RESOURCE_NOT_FOUND'); } // [#8069] Refuse BEFORE any mutation — a refused redelivery must leave // the row byte-identical, including its `dead` status and its reason. - await assertRedeliverAllowed({ ...row }, guard); + await assertRedeliverAllowed({ ...row }, options.guard); const now = Date.now(); row.status = 'pending'; row.attempts = 0; diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index a4d76ff31b..f542a73b8c 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -16,6 +16,7 @@ import type { HttpDeliveryStatus, IHttpOutbox, RedeliverGuard, + RedeliverOptions, } from './http-outbox.js'; import { INBOX_OBJECT, RECEIPT_OBJECT } from './inbox-channel.js'; @@ -280,12 +281,24 @@ export class MessagingService { * (a parked drop record — sending it would be a FIRST delivery, unsigned), * and for a row whose producer's {@link registerRedeliverGuard} verdict * refuses. Both refusals happen before anything is written. + * + * [#10740] `options.tenantId` is the REQUESTING caller's organization, and + * it is the whole reason this method takes a second argument: the door in + * front of it (`POST /api/v1/webhooks/redeliver`) is open to any + * authenticated user, and `sys_http_delivery` is tenant-scoped. The + * outbox applies it to the rows it reads and the row it writes, so a + * caller can only replay deliveries in its own organization; a row + * elsewhere is `RESOURCE_NOT_FOUND`. ⛔ There is no `bypassTenantAudit` + * anywhere on this path and there must not be — see `RedeliverOptions`. */ - async redeliverHttp(id: string): Promise { + async redeliverHttp(id: string, options: RedeliverOptions): Promise { if (!this.httpOutbox) { throw new Error('messaging: HTTP delivery outbox not configured'); } - return this.httpOutbox.redeliver(id, (row) => this.redeliverGuards.get(row.source)?.(row)); + return this.httpOutbox.redeliver(id, { + tenantId: options.tenantId, + guard: (row) => this.redeliverGuards.get(row.source)?.(row), + }); } /** List HTTP delivery rows (admin/tests). Empty when no outbox is wired. */ diff --git a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts index e8e59054e8..c788d780af 100644 --- a/packages/services/service-messaging/src/outbox-dispatcher-scope.ts +++ b/packages/services/service-messaging/src/outbox-dispatcher-scope.ts @@ -44,9 +44,15 @@ import type { EngineUpdateOptions } from '@objectstack/spec/data'; * touches". It silences a warning about a write that was already, and * correctly, environment-wide. It must never be reached for to quiet a write * that a request context could have scoped — that is the failure mode the - * audit exists to prevent, and the row-level writes on these same objects - * (`ack`, `redeliver`) are single-record `multi: false` writes that do **not** - * use this helper. + * audit exists to prevent. + * + * ⛔ **`multi: false` sites do not use this helper** — its return type says + * `multi: true` so they cannot, deliberately. The single-record writes on + * these same objects are audited under a DIFFERENT op (`update`, not + * `updateMany`) and they do **not** share one classification: + * {@link dispatcherAckOptions} carries the sweep warrant to the two `ack` + * sites, and `SqlHttpOutbox.redeliver` — request-reachable — carries a + * threaded tenant and no bypass at all. * * @param where Predicate identifying the rows this sweep claims or reaps. */ @@ -55,3 +61,57 @@ export function dispatcherSweepOptions( ): EngineUpdateOptions & { multi: true; bypassTenantAudit: true } { return { where, multi: true, bypassTenantAudit: true }; } + + +/** + * The write options for a dispatcher **`ack`** — the single-record + * (`multi: false`) write that records one delivery attempt's outcome, on + * `SqlNotificationOutbox.ack` and `SqlHttpOutbox.ack`. + * + * ## Why a second helper instead of {@link dispatcherSweepOptions} + * These are audited under the driver's **`update`** op, not `updateMany`, and + * the two ops are separate keys in `auditMissingTenant`'s throttle — so a + * classification made for the sweeps says nothing about these. The sweep + * helper's return type is `& { multi: true }`, which makes the confusion a + * compile error rather than a judgement call. + * + * ## The warrant, re-derived rather than inherited + * It is the same warrant as the sweeps', and every limb was re-checked + * against this tree: + * + * 1. **No request context exists to thread.** `ack` has exactly two callers, + * `NotificationDispatcher.dispatchOne` / `HttpDispatcher.dispatchOne`, + * both inside `runPartition()` — a `setInterval` tick holding the + * `notify.dispatcher.partition.` / `http.dispatcher.partition.` + * cluster lock. There is no HTTP request, no session and no active + * organization anywhere on that path. + * 2. **The row being acked was claimed by a deliberately global sweep.** + * `claim()` crosses organizations by construction (partitioning is + * `hash(refId | notificationId | digestKey) mod N`, a load-spreading key, + * never an org key), and one outbox per ENVIRONMENT drains the whole + * queue. An `ack` that could not write the row its own tick just claimed + * would leave that row `in_flight` until the visibility timeout, forever, + * for every organization but one. + * + * ## ⛔ Why not "just pass the claimed row's own organization_id" + * It is available on the notification claim result, so it is the tempting + * answer, and it is the WRONG one — worse than this bypass, not better. A + * predicate derived from the row you are about to write is tautological: it + * matches exactly the row it was read from and can never exclude anything, so + * it adds no isolation whatsoever. What it does add is the appearance of + * isolation — it silences the audit line, and the next reader finds a + * tenant-scoped write instead of a declared global one. The audit's question + * is "did the CALLER's tenant reach this write?", and on a dispatcher tick + * the honest answer is "there is no caller tenant", which is what this flag + * states. + * + * ⚠️ Diagnostics only, exactly as above: it never changes what the write + * touches. `ack` already targets one row by primary key. + * + * @param id Primary key of the delivery row this attempt outcome belongs to. + */ +export function dispatcherAckOptions( + id: string, +): EngineUpdateOptions & { multi: false; bypassTenantAudit: true } { + return { where: { id }, multi: false, bypassTenantAudit: true }; +} diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index 73068d9507..41c5a0b96e 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -4,7 +4,7 @@ import { randomUUID } from 'node:crypto'; import type { IDataEngine } from '@objectstack/spec/contracts'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; -import { dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; +import { dispatcherAckOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; import { deliveryBody, signBody } from './http-sender.js'; import { HttpRedeliverError, @@ -16,7 +16,7 @@ import { type HttpDelivery, type HttpDeliveryStatus, type IHttpOutbox, - type RedeliverGuard, + type RedeliverOptions, type UndeliverableHttpInput, } from './http-outbox.js'; import { SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; @@ -329,7 +329,10 @@ export class SqlHttpOutbox implements IHttpOutbox { next_retry_at: nextRetryAt, error, }, - { where: { id }, multi: false }, + // Single-record dispatcher write, audited under the `update` op. + // Declared a global-sweep site — no request context exists on the + // tick that reaches here. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherAckOptions(id), ); } @@ -341,14 +344,39 @@ export class SqlHttpOutbox implements IHttpOutbox { return rows.map((r) => this.toDelivery(r)); } - async redeliver(id: string, guard?: RedeliverGuard): Promise { - const current = (await this.engine.findOne(this.objectName, { where: { id } })) as DeliveryRow | null; + /** + * [#10740] The one request-reachable write on `sys_http_delivery`, and the + * only site in this class that is tenant-SCOPED rather than declared + * global. + * + * `POST /api/v1/webhooks/redeliver` serves any authenticated user, so on a + * walled deployment the caller's organization is exactly the predicate + * this write is missing — and `options.tenantId` supplies it, to the reads + * as well as the write. ⛔ `bypassTenantAudit` is NOT an option here: the + * two produce the same silence in the log, and the audit line is the only + * thing that would ever report this hole. See {@link RedeliverOptions}. + * + * Scoping the READS is what makes the refusal fail-closed and quiet: a row + * belonging to another organization is not found (`RESOURCE_NOT_FOUND`), + * so the endpoint neither replays it nor confirms it exists. It is also + * what keeps the guard honest — the producer veto never sees a row the + * caller may not read. + */ + async redeliver(id: string, options: RedeliverOptions): Promise { + // One options bag for every engine call below, so the read that decides + // the refusal and the write that acts on it can never disagree about + // which tenant is asking. + const scope = { tenantId: options.tenantId }; + const current = (await this.engine.findOne(this.objectName, { + where: { id }, + ...scope, + })) as DeliveryRow | null; if (!current) { throw new HttpRedeliverError(`Delivery row '${id}' not found`, 'RESOURCE_NOT_FOUND'); } // [#8069] Every refusal runs BEFORE the reset UPDATE — a refused // redelivery leaves the row exactly as it was, `dead` reason included. - await assertRedeliverAllowed(this.toDelivery(current), guard); + await assertRedeliverAllowed(this.toDelivery(current), options.guard); await this.engine.update( this.objectName, { @@ -362,9 +390,12 @@ export class SqlHttpOutbox implements IHttpOutbox { response_body: null, error: null, }, - { where: { id, status: { $in: ['success', 'failed', 'dead'] } }, multi: false }, + { where: { id, status: { $in: ['success', 'failed', 'dead'] } }, multi: false, ...scope }, ); - const after = (await this.engine.findOne(this.objectName, { where: { id } })) as DeliveryRow | null; + const after = (await this.engine.findOne(this.objectName, { + where: { id }, + ...scope, + })) as DeliveryRow | null; if (!after || after.status !== 'pending') { throw new HttpRedeliverError(`Delivery row '${id}' state changed during redeliver`, 'DELIVERY_NOT_ELIGIBLE'); } diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index 5ff09c982f..9e9a5790ad 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -12,7 +12,7 @@ import type { } from './outbox.js'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; -import { dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; +import { dispatcherAckOptions, dispatcherSweepOptions } from './outbox-dispatcher-scope.js'; export const DELIVERY_OBJECT = 'sys_notification_delivery'; @@ -246,7 +246,10 @@ export class SqlNotificationOutbox implements INotificationOutbox { next_attempt_at: nextAttemptAt, error, }, - { where: { id }, multi: false } as any, + // Single-record dispatcher write, audited under the `update` op. + // Declared a global-sweep site — no request context exists on the + // tick that reaches here. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherAckOptions(id) as any, ); }