From 4379a364ce18d0626d70e23302fedfd6d90485a7 Mon Sep 17 00:00:00 2001 From: James Kurz Date: Sun, 11 Oct 2026 18:36:23 -0400 Subject: [PATCH] fix(auth): version staff provisioning audits by membership --- .../staff-provisioning.integration.test.ts | 214 ++++++++++++++++++ scripts/provision-staff.ts | 77 +++++-- 2 files changed, 268 insertions(+), 23 deletions(-) create mode 100644 packages/db/src/staff-provisioning.integration.test.ts diff --git a/packages/db/src/staff-provisioning.integration.test.ts b/packages/db/src/staff-provisioning.integration.test.ts new file mode 100644 index 00000000..09a3367f --- /dev/null +++ b/packages/db/src/staff-provisioning.integration.test.ts @@ -0,0 +1,214 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, expect, it, vi } from "vitest"; +import { provisionStaff } from "../../../scripts/provision-staff"; +import { createRuntimeDatabase } from "./client"; +import { accounts, organizations } from "./schema"; + +const url = + process.env.DIRECT_DATABASE_URL ?? + "postgresql://postgres:postgres@127.0.0.1:54322/postgres?sslmode=disable"; +const { db, client } = createRuntimeDatabase({ + url, + role: "clockwork_service", + ssl: false, +}); +const accountId = randomUUID(); +const organizationId = randomUUID(); +const workosOrganizationId = `org_${organizationId.replaceAll("-", "")}`; +const operatorId = randomUUID(); +const historicalOrganizationEventId = randomUUID(); +const identities: string[] = [operatorId]; +const consoleLog = vi.spyOn(console, "log").mockImplementation(() => undefined); +// Even an accidentally unmocked provider must never leave the local test. +const fetchSpy = vi + .spyOn(globalThis, "fetch") + .mockRejectedValue(new Error("Unexpected provider call")); + +async function audit( + id: string, + aggregateType: string, + aggregateId: string, + version: number, +) { + await client`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,after) + values(${id},${aggregateType},${aggregateId},${version},'test.staff.history',1,${JSON.stringify({ kind: "user", id: operatorId })}::jsonb,now(),${randomUUID()},'{}'::jsonb)`; +} + +beforeAll(async () => { + await db.insert(accounts).values({ + id: accountId, + legalName: `Provisioning ${accountId}`, + relationshipRoles: ["direct_client"], + registeredAddress: { + line1: "1 Main St", + city: "Wilmington", + region: "DE", + postalCode: "19801", + country: "US", + }, + billingContact: { name: "Billing", email: "billing@fil.one" }, + apContact: { name: "AP", email: "ap@fil.one" }, + invoiceDeliveryEmail: "billing@fil.one", + domain: `provisioning-${accountId}.test`, + country: "US", + currency: "USD", + } as typeof accounts.$inferInsert); + await db.insert(organizations).values({ + id: organizationId, + accountId, + name: "Provisioning audit", + side: "fil_one", + workosOrganizationId, + } as typeof organizations.$inferInsert); + await client`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled) + values(${operatorId},${`user_${operatorId}`},${`operator-${operatorId}@fil.one`},'Provisioning operator',true,true)`; + await client`insert into memberships(id,organization_id,user_id,role) + values(${randomUUID()},${organizationId},${operatorId},'commerce_admin')`; + await audit(historicalOrganizationEventId, "organization", organizationId, 1); +}); + +afterAll(async () => { + consoleLog.mockRestore(); + fetchSpy.mockRestore(); + await client`delete from memberships where organization_id=${organizationId}`; + // Retain identities referenced by immutable audit/notices, but remove staff eligibility. + await client`update commerce_users set is_internal_staff=false where id=any(${identities})`; + await client.end(); +}); + +async function fixture(kind: "create" | "restore" | "update") { + const id = randomUUID(); + const email = `provision-${id}@fil.one`; + const workosUserId = `user_${id.replaceAll("-", "")}`; + identities.push(id); + if (kind !== "create") { + await client`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled) + values(${id},${workosUserId},${email},'Existing staff',true,true)`; + } + const priorMembershipId = randomUUID(); + if (kind !== "create") { + await client`insert into memberships(id,organization_id,user_id,role) + values(${priorMembershipId},${organizationId},${id},'internal_operator')`; + await audit(randomUUID(), "membership", priorMembershipId, 1); + if (kind === "restore") + await client`delete from memberships where id=${priorMembershipId}`; + } + const environment = { + STAFF_PROVISIONING: JSON.stringify({ + email, + name: "Approved staff", + title: "Staff", + role: "commerce_admin", + updateRole: true, + }), + BOOTSTRAP_MANIFEST: JSON.stringify({ + environment: "staging", + targetDatabaseHost: new URL(url).hostname, + operatorUserId: operatorId, + organization: { id: organizationId, workosOrganizationId }, + }), + DIRECT_DATABASE_URL: url, + WORKOS_API_KEY: "test-provider-key", + DEPLOY_STAGE: "staging", + INTERNAL_EMAIL_DOMAINS: "fil.one", + }; + const provider = vi.fn(() => + Promise.resolve({ + workosUserId, + workosMembershipId: `om_${id.replaceAll("-", "")}`, + outcome: "existing" as const, + }), + ); + return { id, email, priorMembershipId, environment, provider }; +} + +async function membership(email: string) { + const [row] = await client< + { id: string; user_id: string; role: string; row_version: number }[] + >` + select m.id,m.user_id,m.role,m.row_version from memberships m join commerce_users u on u.id=m.user_id where u.email=${email}`; + if (!row) throw new Error("Expected provisioned membership"); + return row; +} + +it("updates the membership audit stream past existing organization and membership history", async () => { + const input = await fixture("update"); + await provisionStaff(input.environment, input.provider); + expect(await membership(input.email)).toMatchObject({ + id: input.priorMembershipId, + role: "commerce_admin", + row_version: 2, + }); + input.environment.STAFF_PROVISIONING = JSON.stringify({ + email: input.email, + name: "Approved staff", + title: "Staff", + role: "revenue", + updateRole: true, + }); + await provisionStaff(input.environment, input.provider); + expect(await membership(input.email)).toMatchObject({ + id: input.priorMembershipId, + role: "revenue", + row_version: 3, + }); + // Reapplying the same approved request verifies without another mutation/audit. + await provisionStaff(input.environment, input.provider); + const events = await client< + { + aggregate_version: number; + event_type: string; + after: Record; + outbox_count: number; + }[] + >` + select e.aggregate_version,e.event_type,e.after,(select count(*)::int from outbox_messages o where o.event_id=e.id) as outbox_count + from audit_events e where e.aggregate_type='membership' and e.aggregate_id=${input.priorMembershipId} order by e.aggregate_version`; + expect(events.map((event) => event.aggregate_version)).toEqual([1, 2, 3]); + expect(events.slice(1)).toMatchObject( + [2, 3].map((version) => ({ + aggregate_version: version, + event_type: "staff.role_changed", + after: { organizationId }, + outbox_count: 1, + })), + ); + expect(fetchSpy).not.toHaveBeenCalled(); + expect(input.provider).toHaveBeenCalledTimes(3); +}); + +it.each(["create", "restore"] as const)( + "%s uses a fresh membership stream despite existing organization history", + async (kind) => { + const input = await fixture(kind); + await provisionStaff(input.environment, input.provider); + const row = await membership(input.email); + identities.push(row.user_id); + expect(row).toMatchObject({ role: "commerce_admin", row_version: 1 }); + expect(row.id).not.toBe(input.priorMembershipId); + const events = await client< + { + aggregate_version: number; + event_type: string; + after: Record; + outbox_count: number; + }[] + >` + select e.aggregate_version,e.event_type,e.after,(select count(*)::int from outbox_messages o where o.event_id=e.id) as outbox_count + from audit_events e where e.aggregate_type='membership' and e.aggregate_id=${row.id}`; + expect(events).toMatchObject([ + { + aggregate_version: 1, + event_type: + kind === "create" ? "staff.provisioned" : "staff.reactivated", + after: { organizationId }, + outbox_count: 1, + }, + ]); + const prior = await client< + { id: string }[] + >`select id from audit_events where aggregate_type='organization' and aggregate_id=${organizationId} and aggregate_version=1`; + expect(prior.map(({ id }) => id)).toEqual([historicalOrganizationEventId]); + expect(fetchSpy).not.toHaveBeenCalled(); + }, +); diff --git a/scripts/provision-staff.ts b/scripts/provision-staff.ts index 787576fb..255f6329 100644 --- a/scripts/provision-staff.ts +++ b/scripts/provision-staff.ts @@ -1,4 +1,6 @@ import { randomUUID } from "node:crypto"; +import { realpathSync } from "node:fs"; +import { pathToFileURL } from "node:url"; import { assertStaffAdministratorRemains, assertStaffProvisioningTarget, @@ -13,24 +15,27 @@ import postgres from "postgres"; import { provisionWorkosStaff } from "../packages/integrations/src/workos/provision-staff"; let phase = "configuration"; -async function main() { - const raw = process.env.STAFF_PROVISIONING; +export async function provisionStaff( + environment: NodeJS.ProcessEnv = process.env, + provisionIdentity: typeof provisionWorkosStaff = provisionWorkosStaff, +) { + const raw = environment.STAFF_PROVISIONING; if (!raw) return; const person = StaffProvisioningSchema.parse(JSON.parse(raw)); const manifest = JSON.parse( - process.env.BOOTSTRAP_MANIFEST ?? "null", + environment.BOOTSTRAP_MANIFEST ?? "null", ) as StaffProvisioningManifest | null; - const url = process.env.DIRECT_DATABASE_URL, - key = process.env.WORKOS_API_KEY; + const url = environment.DIRECT_DATABASE_URL, + key = environment.WORKOS_API_KEY; if (!manifest || !url || !key) throw new Error("STAFF_PROVISIONING_CONFIGURATION_REQUIRED"); assertStaffProvisioningTarget({ manifest, databaseHost: new URL(url).hostname, - deployStage: process.env.DEPLOY_STAGE, + deployStage: environment.DEPLOY_STAGE, }); if ( - !staffEmailDomainAllowed(person.email, process.env.INTERNAL_EMAIL_DOMAINS) + !staffEmailDomainAllowed(person.email, environment.INTERNAL_EMAIL_DOMAINS) ) throw new Error("STAFF_PROVISIONING_DOMAIN_DENIED"); const client = postgres(url, { @@ -59,7 +64,7 @@ async function main() { }); console.log("staff provisioning: identity plan", { plan: plan.kind }); phase = "workos"; - const binding = await provisionWorkosStaff( + const binding = await provisionIdentity( key, manifest.organization.workosOrganizationId, person, @@ -74,13 +79,20 @@ async function main() { const actor = { kind: "user", id: manifest.operatorUserId }; const appendEvent = async ( tx: postgres.TransactionSql, + membership: { id: string; row_version: number }, eventType: string, after: Record, before?: Record, ) => { const eventId = randomUUID(); - await tx`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,before,after) values(${eventId},'organization',${manifest.organization.id},1,${eventType},1,${tx.json(actor)},now(),${randomUUID()},${before ? tx.json(before) : null},${tx.json(after)})`; - await tx`insert into outbox_messages(id,event_id,topic,payload) values(${randomUUID()},${eventId},${eventType},${tx.json(after)})`; + const associatedAfter = { + ...after, + organizationId: manifest.organization.id, + }; + // Staff changes share Team's membership stream and its persisted version. + // Organization version 1 may already belong to an unrelated earlier event. + await tx`insert into audit_events(id,aggregate_type,aggregate_id,aggregate_version,event_type,event_version,actor,occurred_at,request_id,before,after) values(${eventId},'membership',${membership.id},${membership.row_version},${eventType},1,${tx.json(actor)},now(),${randomUUID()},${before ? tx.json(before) : null},${tx.json(associatedAfter)})`; + await tx`insert into outbox_messages(id,event_id,topic,payload) values(${randomUUID()},${eventId},${eventType},${tx.json(associatedAfter)})`; }; if (plan.kind === "update_role") { phase = "update_role"; @@ -99,13 +111,16 @@ async function main() { otherAdministrators: administrators?.total ?? 0, }); phase = "role_update"; - const updated = - await tx`update memberships set role=${person.role} where user_id=${plan.userId} and organization_id=${manifest.organization.id} and role=${plan.from} returning id`; - if (updated.length !== 1) + const updated = await tx< + { id: string; row_version: number }[] + >`update memberships set role=${person.role} where user_id=${plan.userId} and organization_id=${manifest.organization.id} and role=${plan.from} returning id,row_version`; + const membership = updated[0]; + if (updated.length !== 1 || !membership) throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED"); phase = "role_audit"; await appendEvent( tx, + membership, "staff.role_changed", { email: person.email, role: person.role }, { email: person.email, role: plan.from }, @@ -132,8 +147,14 @@ async function main() { await tx`select id from memberships where user_id=${plan.userId}`; if (current.length) throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED"); - await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${randomUUID()},${manifest.organization.id},${plan.userId},${binding.workosMembershipId},${person.role})`; - await appendEvent(tx, "staff.reactivated", after); + const membershipId = randomUUID(); + await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${membershipId},${manifest.organization.id},${plan.userId},${binding.workosMembershipId},${person.role})`; + await appendEvent( + tx, + { id: membershipId, row_version: 1 }, + "staff.reactivated", + after, + ); return; } const duplicate = @@ -141,9 +162,15 @@ async function main() { if (duplicate.length) throw new Error("STAFF_PROVISIONING_RETRY_REQUIRED"); const id = randomUUID(); + const membershipId = randomUUID(); await tx`insert into commerce_users(id,workos_user_id,email,name,is_internal_staff,mfa_enrolled) values(${id},${binding.workosUserId},${person.email},${person.name},true,false)`; - await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${randomUUID()},${manifest.organization.id},${id},${binding.workosMembershipId},${person.role})`; - await appendEvent(tx, "staff.provisioned", after); + await tx`insert into memberships(id,organization_id,user_id,workos_membership_id,role) values(${membershipId},${manifest.organization.id},${id},${binding.workosMembershipId},${person.role})`; + await appendEvent( + tx, + { id: membershipId, row_version: 1 }, + "staff.provisioned", + after, + ); }); console.log( `staff provisioning: identity and ${person.role} membership ${plan.kind === "restore" ? "restored" : "created"}; email verification and MFA remain required at login`, @@ -152,10 +179,14 @@ async function main() { await client.end(); } } -main().catch((error: unknown) => { - console.error("staff provisioning failed", { - phase, - code: staffProvisioningFailureCode(error), +if ( + process.argv[1] && + import.meta.url === pathToFileURL(realpathSync(process.argv[1])).href +) + provisionStaff().catch((error: unknown) => { + console.error("staff provisioning failed", { + phase, + code: staffProvisioningFailureCode(error), + }); + process.exitCode = 1; }); - process.exitCode = 1; -});