diff --git a/apps/integration-service/migrations/0003_plugin_operator_context_replay_record.sql b/apps/integration-service/migrations/0003_plugin_operator_context_replay_record.sql new file mode 100644 index 00000000..b547c760 --- /dev/null +++ b/apps/integration-service/migrations/0003_plugin_operator_context_replay_record.sql @@ -0,0 +1,19 @@ +CREATE TABLE plugin_integration.plugin_operator_context_replay_record ( + evidence_id uuid PRIMARY KEY, + consumed_at timestamptz NOT NULL, + expires_at timestamptz NOT NULL, + CONSTRAINT plugin_operator_context_replay_lifetime_check + CHECK (expires_at >= consumed_at) +); + +CREATE INDEX plugin_operator_context_replay_expiry_index + ON plugin_integration.plugin_operator_context_replay_record (expires_at); + +COMMENT ON TABLE plugin_integration.plugin_operator_context_replay_record IS + 'Durable one-time plugin operator evidence consumed by the Integration service to prevent replay across service instances.'; +COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.evidence_id IS + 'Signed UUIDv4 evidence identity; primary-key uniqueness permits exactly one durable consumption.'; +COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.consumed_at IS + 'Instant when the winning Integration service instance consumed the signed evidence.'; +COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.expires_at IS + 'Retention deadline after which the evidence record is eligible for cleanup because its signature lifetime has closed.'; diff --git a/apps/integration-service/src/plugin-operator-application.test.ts b/apps/integration-service/src/plugin-operator-application.test.ts new file mode 100644 index 00000000..c97f183b --- /dev/null +++ b/apps/integration-service/src/plugin-operator-application.test.ts @@ -0,0 +1,427 @@ +import { createHmac, randomBytes, randomUUID } from 'node:crypto'; +import { describe, expect, it, vi } from 'vitest'; +import type { PluginManifest } from '@life-os/plugin-sdk'; +import { + PluginOperatorApplication, + PluginOperatorDependencyError, + type PluginCredentialOperatorPort, + type PluginInstallationOperatorPort, +} from './plugin-operator-application'; +import type { + PluginCredentialBindingView, + BindPluginCredentialInput, +} from './plugin-credential'; +import type { + InstallPluginInput, + PluginInstallationContext, + PluginInstallationRecord, +} from './plugin-installation'; +import type { + PluginOperatorReplayEvidence, + PluginOperatorReplayGuardPort, +} from './plugin-operator-replay'; + +const WORKSPACE_ID = '11111111-1111-4111-8111-111111111111'; +const USER_ID = '22222222-2222-4222-8222-222222222222'; +const INSTALLATION_ID = '33333333-3333-4333-8333-333333333333'; +const CREDENTIAL_BINDING_ID = '44444444-4444-4444-8444-444444444444'; +const SECRET = randomBytes(32).toString('base64url'); +const NOW_SECONDS = 1_786_334_400; +const EARLIER_ISSUED_AT_SECONDS = NOW_SECONDS - 30; +const NOW = new Date(NOW_SECONDS * 1_000).toISOString(); +const EXPIRES_AT = new Date((NOW_SECONDS + 60) * 1_000).toISOString(); +const EARLIER_EXPIRES_AT = new Date( + (EARLIER_ISSUED_AT_SECONDS + 60) * 1_000, +).toISOString(); + +const MANIFEST: PluginManifest = Object.freeze({ + pluginId: 'com.example.calendar', + displayName: 'Example Calendar', + contractVersion: '1.0', + subscriptions: Object.freeze(['lifeos.calendar.event.v1']), +}); + +const INSTALLATION_RECORD: PluginInstallationRecord = Object.freeze({ + installationId: INSTALLATION_ID, + workspaceId: WORKSPACE_ID, + installedByUserId: USER_ID, + pluginId: MANIFEST.pluginId, + pluginContractVersion: MANIFEST.contractVersion, + manifestSha256: 'a'.repeat(64), + grantedCapabilities: Object.freeze(['lifeos.calendar.event.v1']), + status: 'active', + installedAt: NOW, + revokedAt: null, +}); + +const CREDENTIAL_VIEW: PluginCredentialBindingView = Object.freeze({ + credentialBindingId: CREDENTIAL_BINDING_ID, + installationId: INSTALLATION_ID, + workspaceId: WORKSPACE_ID, + installedByUserId: USER_ID, + credentialName: 'oauth.access-token', + status: 'active', + boundAt: NOW, + revokedAt: null, +}); + +function signedHeaders( + method: 'GET' | 'POST', + path: string, + issuedAt = String(NOW_SECONDS), + evidenceId = randomUUID(), +): { + readonly workspaceId: string; + readonly userId: string; + readonly evidenceId: string; + readonly issuedAt: string; + readonly signature: string; +} { + return { + workspaceId: WORKSPACE_ID, + userId: USER_ID, + evidenceId, + issuedAt, + signature: createHmac('sha256', SECRET) + .update( + `life-os.integration-operator-context.v1\n${WORKSPACE_ID}\n${USER_ID}\n${evidenceId}\n${issuedAt}\n${method}\n${path}`, + 'utf8', + ) + .digest('base64url'), + }; +} + +function installationPort(): PluginInstallationOperatorPort & { + install: ReturnType; + getInstallation: ReturnType; + revoke: ReturnType; +} { + return { + install: vi.fn(async (_input: InstallPluginInput) => INSTALLATION_RECORD), + getInstallation: vi.fn( + async (_context: PluginInstallationContext, _installationId: string) => + INSTALLATION_RECORD, + ), + revoke: vi.fn( + async (_context: PluginInstallationContext, _installationId: string) => + ({ + ...INSTALLATION_RECORD, + status: 'revoked' as const, + revokedAt: NOW, + }), + ), + }; +} + +function credentialPort(): PluginCredentialOperatorPort & { + bind: ReturnType; + revoke: ReturnType; +} { + return { + bind: vi.fn(async (_input: BindPluginCredentialInput) => CREDENTIAL_VIEW), + revoke: vi.fn( + async (_context: PluginInstallationContext, _credentialBindingId: string) => + ({ ...CREDENTIAL_VIEW, status: 'revoked' as const, revokedAt: NOW }), + ), + }; +} + +function replayGuard(): PluginOperatorReplayGuardPort & { + consume: ReturnType; +} { + const consumed = new Set(); + return { + consume: vi.fn(async (evidence: PluginOperatorReplayEvidence) => { + if (consumed.has(evidence.evidenceId)) { + return false; + } + consumed.add(evidence.evidenceId); + return true; + }), + }; +} + +function application( + installations = installationPort(), + credentials: PluginCredentialOperatorPort | undefined = credentialPort(), + replay: PluginOperatorReplayGuardPort | undefined = replayGuard(), + nowSeconds = NOW_SECONDS, +): PluginOperatorApplication { + return new PluginOperatorApplication( + installations, + credentials, + SECRET, + replay, + () => nowSeconds, + ); +} + +function applicationWithoutCredentials( + installations = installationPort(), + replay: PluginOperatorReplayGuardPort | undefined = replayGuard(), +): PluginOperatorApplication { + return new PluginOperatorApplication( + installations, + undefined, + SECRET, + replay, + () => NOW_SECONDS, + ); +} + +describe('authenticated plugin operator composition', () => { + it('forwards installation input only after durable one-time evidence consumption', async () => { + const installations = installationPort(); + const replay = replayGuard(); + const app = application(installations, credentialPort(), replay); + const headers = signedHeaders('POST', '/v1/plugins/installations'); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + const result = await app.install(headers, input); + + expect(result).toEqual(INSTALLATION_RECORD); + expect(replay.consume).toHaveBeenCalledWith({ + evidenceId: headers.evidenceId, + consumedAt: NOW, + expiresAt: EXPIRES_AT, + }); + expect(installations.install).toHaveBeenCalledWith({ + ...input, + trustedContext: { + workspaceId: WORKSPACE_ID, + actorUserId: USER_ID, + }, + }); + }); + + it('expires replay evidence from signed issuance rather than delayed consumption', async () => { + const replay = replayGuard(); + const app = application(installationPort(), credentialPort(), replay); + const headers = signedHeaders( + 'POST', + '/v1/plugins/installations', + String(EARLIER_ISSUED_AT_SECONDS), + ); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + await expect(app.install(headers, input)).resolves.toEqual( + INSTALLATION_RECORD, + ); + expect(replay.consume).toHaveBeenCalledWith({ + evidenceId: headers.evidenceId, + consumedAt: NOW, + expiresAt: EARLIER_EXPIRES_AT, + }); + }); + + it('rejects reusing the same valid signed evidence across application instances', async () => { + const installations = installationPort(); + const replay = replayGuard(); + const first = application(installations, credentialPort(), replay); + const second = application(installations, credentialPort(), replay); + const headers = signedHeaders('POST', '/v1/plugins/installations'); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + await expect(first.install(headers, input)).resolves.toEqual( + INSTALLATION_RECORD, + ); + await expect(second.install(headers, input)).rejects.toMatchObject({ + name: 'IntegrationOperatorContextError', + kind: 'invalid', + }); + expect(installations.install).toHaveBeenCalledTimes(1); + }); + + it('allows distinct signed evidence for otherwise identical same-second requests', async () => { + const installations = installationPort(); + const replay = replayGuard(); + const app = application(installations, credentialPort(), replay); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + await expect( + app.install(signedHeaders('POST', '/v1/plugins/installations'), input), + ).resolves.toEqual(INSTALLATION_RECORD); + await expect( + app.install(signedHeaders('POST', '/v1/plugins/installations'), input), + ).resolves.toEqual(INSTALLATION_RECORD); + expect(installations.install).toHaveBeenCalledTimes(2); + }); + + it('fails closed before downstream authority when the replay store is unavailable', async () => { + const installations = installationPort(); + const app = new PluginOperatorApplication( + installations, + credentialPort(), + SECRET, + undefined, + () => NOW_SECONDS, + ); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + await expect( + app.install(signedHeaders('POST', '/v1/plugins/installations'), input), + ).rejects.toMatchObject({ + name: 'IntegrationOperatorContextError', + kind: 'unavailable', + }); + expect(installations.install).not.toHaveBeenCalled(); + }); + + it('classifies replay-store failures as verifier unavailability without granting authority', async () => { + const installations = installationPort(); + const replay: PluginOperatorReplayGuardPort = { + consume: vi.fn(async () => { + throw new Error('database unavailable'); + }), + }; + const app = application(installations, credentialPort(), replay); + const input = { + installationId: INSTALLATION_ID, + manifest: MANIFEST, + grantedCapabilities: ['lifeos.calendar.event.v1'], + } as const; + + await expect( + app.install(signedHeaders('POST', '/v1/plugins/installations'), input), + ).rejects.toMatchObject({ + name: 'IntegrationOperatorContextError', + kind: 'unavailable', + }); + expect(installations.install).not.toHaveBeenCalled(); + }); + + it('rejects replaying a read signature as revocation before application authority', async () => { + const installations = installationPort(); + const app = application(installations); + const readPath = `/v1/plugins/installations/${INSTALLATION_ID}`; + + await expect( + app.revokeInstallation( + signedHeaders('GET', readPath), + INSTALLATION_ID, + ), + ).rejects.toMatchObject({ + name: 'IntegrationOperatorContextError', + kind: 'invalid', + }); + expect(installations.revoke).not.toHaveBeenCalled(); + }); + + it('rejects ambiguous dynamic identifiers before installation lookup', async () => { + const installations = installationPort(); + const app = application(installations); + const ambiguousId = `${INSTALLATION_ID}?workspace=${WORKSPACE_ID}`; + + await expect( + app.getInstallation( + signedHeaders( + 'GET', + `/v1/plugins/installations/${INSTALLATION_ID}`, + ), + ambiguousId, + ), + ).rejects.toMatchObject({ + kind: 'invalid', + }); + expect(installations.getInstallation).not.toHaveBeenCalled(); + }); + + it('keeps missing secret-store composition explicitly unavailable after valid authentication', async () => { + const installations = installationPort(); + const app = applicationWithoutCredentials(installations); + const input = { + credentialBindingId: CREDENTIAL_BINDING_ID, + installationId: INSTALLATION_ID, + credentialName: 'oauth.access-token', + secretValue: 'provider-token-value', + } as const; + + await expect( + app.bindCredential( + signedHeaders('POST', '/v1/plugins/credential-bindings'), + input, + ), + ).rejects.toBeInstanceOf(PluginOperatorDependencyError); + }); + + it('does not reveal dependency availability to an invalid operator context', async () => { + const app = applicationWithoutCredentials(); + const input = { + credentialBindingId: CREDENTIAL_BINDING_ID, + installationId: INSTALLATION_ID, + credentialName: 'oauth.access-token', + secretValue: 'provider-token-value', + } as const; + const forged = { + ...signedHeaders('POST', '/v1/plugins/credential-bindings'), + signature: 'A'.repeat(43), + }; + + await expect(app.bindCredential(forged, input)).rejects.toMatchObject({ + name: 'IntegrationOperatorContextError', + kind: 'invalid', + }); + }); + + it('forwards credential material only after exact signed operator authority', async () => { + const credentials = credentialPort(); + const app = application(installationPort(), credentials); + const input = { + credentialBindingId: CREDENTIAL_BINDING_ID, + installationId: INSTALLATION_ID, + credentialName: 'oauth.access-token', + secretValue: 'provider-token-value', + } as const; + + const result = await app.bindCredential( + signedHeaders('POST', '/v1/plugins/credential-bindings'), + input, + ); + + expect(result).toEqual(CREDENTIAL_VIEW); + expect(credentials.bind).toHaveBeenCalledWith({ + ...input, + trustedContext: { + workspaceId: WORKSPACE_ID, + actorUserId: USER_ID, + }, + }); + }); + + it('binds credential revocation to its exact dynamic route', async () => { + const credentials = credentialPort(); + const app = application(installationPort(), credentials); + const path = `/v1/plugins/credential-bindings/${CREDENTIAL_BINDING_ID}/revoke`; + + const result = await app.revokeCredential( + signedHeaders('POST', path), + CREDENTIAL_BINDING_ID, + ); + + expect(result.status).toBe('revoked'); + expect(credentials.revoke).toHaveBeenCalledWith( + { workspaceId: WORKSPACE_ID, actorUserId: USER_ID }, + CREDENTIAL_BINDING_ID, + ); + }); +}); diff --git a/apps/integration-service/src/plugin-operator-application.ts b/apps/integration-service/src/plugin-operator-application.ts new file mode 100644 index 00000000..c1df6ddd --- /dev/null +++ b/apps/integration-service/src/plugin-operator-application.ts @@ -0,0 +1,204 @@ +import type { + BindPluginCredentialInput, + PluginCredentialBindingView, +} from './plugin-credential'; +import type { + InstallPluginInput, + PluginInstallationContext, + PluginInstallationRecord, +} from './plugin-installation'; +import { + IntegrationOperatorContextError, + PLUGIN_OPERATOR_CONTEXT_MAXIMUM_AGE_SECONDS, + requireVerifiedPluginOperatorContext, + type IntegrationOperatorContextHeaders, +} from './plugin-operator-context'; +import type { PluginOperatorReplayGuardPort } from './plugin-operator-replay'; + +/** Installation lifecycle authority consumed after signed operator verification. */ +export interface PluginInstallationOperatorPort { + /** Installs only with trusted tenant/user context supplied by this application boundary. */ + install(input: InstallPluginInput): Promise; + /** Reads one installer-owned installation inside trusted tenant/user authority. */ + getInstallation( + trustedContext: PluginInstallationContext, + installationId: string, + ): Promise; + /** Revokes one installer-owned installation inside trusted tenant/user authority. */ + revoke( + trustedContext: PluginInstallationContext, + installationId: string, + ): Promise; +} + +/** Credential lifecycle authority consumed after signed operator verification. */ +export interface PluginCredentialOperatorPort { + /** Binds secret material only after authenticated installation authority is derived. */ + bind(input: BindPluginCredentialInput): Promise; + /** Revokes one installer-owned credential binding inside trusted tenant/user authority. */ + revoke( + trustedContext: PluginInstallationContext, + credentialBindingId: string, + ): Promise; +} + +/** Fixed dependency failure that never discloses credential material or provider details. */ +export class PluginOperatorDependencyError extends Error { + /** Creates the bounded failure returned when credential composition is unavailable. */ + constructor() { + super('Plugin credential capability is unavailable'); + this.name = 'PluginOperatorDependencyError'; + } +} + +/** Operator-selected installation fields; authenticated authority is never accepted from the body. */ +export type PluginOperatorInstallInput = Omit; + +/** Operator-selected credential fields; authenticated authority is never accepted from the body. */ +export type PluginOperatorCredentialInput = Omit< + BindPluginCredentialInput, + 'trustedContext' +>; + +/** Converts a verified Unix second to one canonical persistence instant or fails closed. */ +function canonicalInstant(seconds: number): string { + const value = new Date(seconds * 1_000); + if (!Number.isFinite(value.getTime())) { + throw new IntegrationOperatorContextError('unavailable'); + } + return value.toISOString(); +} + +/** + * Composes cryptographically verified operator identity with host-owned plugin + * installation and credential applications. + * + * Every method constructs the exact server-owned method/path binding before any + * downstream authority is invoked. Tenant/user identifiers are derived only from + * the signed gateway context; request bodies and dynamic identifiers cannot widen + * that authority. Signed UUIDv4 evidence is atomically consumed through a + * service-owned replay guard before lifecycle authority is granted, so separate + * service instances cannot independently accept the same request evidence. + */ +export class PluginOperatorApplication { + /** Creates the operator boundary over host-owned lifecycle, replay, and verifier state. */ + constructor( + private readonly installations: PluginInstallationOperatorPort, + private readonly credentials: PluginCredentialOperatorPort | undefined, + private readonly contextSecret: unknown, + private readonly replayGuard: PluginOperatorReplayGuardPort | undefined, + private readonly nowSeconds: () => number = () => Math.floor(Date.now() / 1000), + ) {} + + /** Installs a plugin only under a signed POST collection authority. */ + async install( + headers: IntegrationOperatorContextHeaders, + input: PluginOperatorInstallInput, + ): Promise { + const trustedContext = await this.requireContext( + headers, + 'POST', + '/v1/plugins/installations', + ); + return this.installations.install({ ...input, trustedContext }); + } + + /** Reads one installation only under the exact signed dynamic GET authority. */ + async getInstallation( + headers: IntegrationOperatorContextHeaders, + installationId: string, + ): Promise { + const trustedContext = await this.requireContext( + headers, + 'GET', + `/v1/plugins/installations/${installationId}`, + ); + return this.installations.getInstallation(trustedContext, installationId); + } + + /** Revokes one installation only under the exact signed dynamic POST authority. */ + async revokeInstallation( + headers: IntegrationOperatorContextHeaders, + installationId: string, + ): Promise { + const trustedContext = await this.requireContext( + headers, + 'POST', + `/v1/plugins/installations/${installationId}/revoke`, + ); + return this.installations.revoke(trustedContext, installationId); + } + + /** Binds a credential only after exact signed authority and configured host secret storage. */ + async bindCredential( + headers: IntegrationOperatorContextHeaders, + input: PluginOperatorCredentialInput, + ): Promise { + const trustedContext = await this.requireContext( + headers, + 'POST', + '/v1/plugins/credential-bindings', + ); + const credentials = this.requireCredentials(); + return credentials.bind({ ...input, trustedContext }); + } + + /** Revokes a credential only under the exact signed dynamic POST authority. */ + async revokeCredential( + headers: IntegrationOperatorContextHeaders, + credentialBindingId: string, + ): Promise { + const trustedContext = await this.requireContext( + headers, + 'POST', + `/v1/plugins/credential-bindings/${credentialBindingId}/revoke`, + ); + return this.requireCredentials().revoke( + trustedContext, + credentialBindingId, + ); + } + + /** Verifies and atomically consumes one signed request identity before downstream authority. */ + private async requireContext( + headers: IntegrationOperatorContextHeaders, + method: 'GET' | 'POST', + path: string, + ): Promise { + const nowSeconds = this.nowSeconds(); + const verified = requireVerifiedPluginOperatorContext( + headers, + this.contextSecret, + { method, path }, + nowSeconds, + ); + if (!this.replayGuard) { + throw new IntegrationOperatorContextError('unavailable'); + } + const evidence = Object.freeze({ + evidenceId: verified.evidenceId, + consumedAt: canonicalInstant(nowSeconds), + expiresAt: canonicalInstant( + verified.issuedAtSeconds + PLUGIN_OPERATOR_CONTEXT_MAXIMUM_AGE_SECONDS, + ), + }); + let consumed: boolean; + try { + consumed = await this.replayGuard.consume(evidence); + } catch { + throw new IntegrationOperatorContextError('unavailable'); + } + if (!consumed) { + throw new IntegrationOperatorContextError('invalid'); + } + return verified.trustedContext; + } + + /** Returns configured credential authority only after caller authentication succeeds. */ + private requireCredentials(): PluginCredentialOperatorPort { + if (!this.credentials) { + throw new PluginOperatorDependencyError(); + } + return this.credentials; + } +} diff --git a/apps/integration-service/src/plugin-operator-context.test.ts b/apps/integration-service/src/plugin-operator-context.test.ts new file mode 100644 index 00000000..2d54eacd --- /dev/null +++ b/apps/integration-service/src/plugin-operator-context.test.ts @@ -0,0 +1,251 @@ +import { createHmac, randomBytes } from 'node:crypto'; +import { describe, expect, it } from 'vitest'; +import { + IntegrationOperatorContextError, + requireTrustedPluginOperatorContext, + requireVerifiedPluginOperatorContext, +} from './plugin-operator-context'; + +const WORKSPACE_ID = '11111111-1111-4111-8111-111111111111'; +const USER_ID = '22222222-2222-4222-8222-222222222222'; +const OTHER_USER_ID = '33333333-3333-4333-8333-333333333333'; +const EVIDENCE_ID = '77777777-7777-4777-8777-777777777777'; +const OTHER_EVIDENCE_ID = '88888888-8888-4888-8888-888888888888'; +const ISSUED_AT = '1786291200'; +const SECRET = randomBytes(32).toString('base64url'); +const INSTALL_PATH = '/v1/plugins/installations'; + +function signature( + method = 'POST', + path = INSTALL_PATH, + workspaceId = WORKSPACE_ID, + userId = USER_ID, + evidenceId = EVIDENCE_ID, + issuedAt = ISSUED_AT, +): string { + return createHmac('sha256', SECRET) + .update( + `life-os.integration-operator-context.v1\n${workspaceId}\n${userId}\n${evidenceId}\n${issuedAt}\n${method}\n${path}`, + 'utf8', + ) + .digest('base64url'); +} + +function context( + overrides: Partial< + Record< + 'workspaceId' | 'userId' | 'evidenceId' | 'issuedAt' | 'signature', + unknown + > + > = {}, +) { + return { + workspaceId: WORKSPACE_ID, + userId: USER_ID, + evidenceId: EVIDENCE_ID, + issuedAt: ISSUED_AT, + signature: signature(), + ...overrides, + }; +} + +function expectInvalid(operation: () => unknown): void { + let thrown: unknown; + try { + operation(); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(IntegrationOperatorContextError); + expect(thrown).toMatchObject({ kind: 'invalid' }); +} + +describe('trusted plugin operator context', () => { + it('returns normalized workspace-user authority only for the exact signed route', () => { + expect( + requireTrustedPluginOperatorContext( + context(), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + Number(ISSUED_AT), + ), + ).toEqual({ workspaceId: WORKSPACE_ID, actorUserId: USER_ID }); + }); + + it('returns the signed one-time evidence identity only after full verification', () => { + expect( + requireVerifiedPluginOperatorContext( + context(), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + Number(ISSUED_AT), + ), + ).toEqual({ + trustedContext: { workspaceId: WORKSPACE_ID, actorUserId: USER_ID }, + evidenceId: EVIDENCE_ID, + issuedAtSeconds: Number(ISSUED_AT), + }); + }); + + it('rejects cross-user, cross-workspace, evidence-id, method, and path replay as invalid evidence', () => { + const cases = [ + context({ userId: OTHER_USER_ID }), + context({ workspaceId: '44444444-4444-4444-8444-444444444444' }), + context({ evidenceId: OTHER_EVIDENCE_ID }), + context({ signature: signature('GET', INSTALL_PATH) }), + context({ + signature: signature('POST', '/v1/plugins/credential-bindings'), + }), + ]; + + for (const candidate of cases) { + expectInvalid(() => + requireTrustedPluginOperatorContext( + candidate, + SECRET, + { method: 'POST', path: INSTALL_PATH }, + Number(ISSUED_AT), + ), + ); + } + }); + + it('accepts only the bounded operator route surface', () => { + const installationId = '55555555-5555-4555-8555-555555555555'; + const bindingId = '66666666-6666-4666-8666-666666666666'; + const routes = [ + { method: 'POST', path: INSTALL_PATH }, + { method: 'GET', path: `/v1/plugins/installations/${installationId}` }, + { + method: 'POST', + path: `/v1/plugins/installations/${installationId}/revoke`, + }, + { method: 'POST', path: '/v1/plugins/credential-bindings' }, + { + method: 'POST', + path: `/v1/plugins/credential-bindings/${bindingId}/revoke`, + }, + ] as const; + + for (const route of routes) { + const signed = context({ + signature: signature(route.method, route.path), + }); + expect( + requireTrustedPluginOperatorContext( + signed, + SECRET, + route, + Number(ISSUED_AT), + ), + ).toEqual({ workspaceId: WORKSPACE_ID, actorUserId: USER_ID }); + } + + const invalidRoutes = [ + { method: 'DELETE', path: INSTALL_PATH }, + { method: 'POST', path: `${INSTALL_PATH}?workspace=other` }, + { method: 'POST', path: `${INSTALL_PATH}/not-a-uuid/revoke` }, + { method: 'POST', path: `${INSTALL_PATH}/../credential-bindings` }, + ]; + for (const route of invalidRoutes) { + expectInvalid(() => + requireTrustedPluginOperatorContext( + context(), + SECRET, + route, + Number(ISSUED_AT), + ), + ); + } + }); + + it('rejects a case-variant dynamic route signed for its lowercase alias', () => { + const canonicalId = '55555555-5555-4555-8555-55555555555a'; + const canonicalPath = `/v1/plugins/installations/${canonicalId}`; + const receivedPath = `/v1/plugins/installations/${canonicalId.toUpperCase()}`; + + expectInvalid(() => + requireTrustedPluginOperatorContext( + context({ signature: signature('GET', canonicalPath) }), + SECRET, + { method: 'GET', path: receivedPath }, + Number(ISSUED_AT), + ), + ); + }); + + it('classifies stale, future, malformed evidence identifiers, and non-canonical signatures as invalid', () => { + const now = Number(ISSUED_AT); + const staleIssuedAt = String(now - 61); + const futureIssuedAt = String(now + 6); + + expectInvalid(() => + requireTrustedPluginOperatorContext( + context({ + issuedAt: staleIssuedAt, + signature: signature( + 'POST', + INSTALL_PATH, + WORKSPACE_ID, + USER_ID, + EVIDENCE_ID, + staleIssuedAt, + ), + }), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + now, + ), + ); + expectInvalid(() => + requireTrustedPluginOperatorContext( + context({ + issuedAt: futureIssuedAt, + signature: signature( + 'POST', + INSTALL_PATH, + WORKSPACE_ID, + USER_ID, + EVIDENCE_ID, + futureIssuedAt, + ), + }), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + now, + ), + ); + expectInvalid(() => + requireTrustedPluginOperatorContext( + context({ evidenceId: 'not-a-uuid' }), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + now, + ), + ); + expectInvalid(() => + requireTrustedPluginOperatorContext( + context({ signature: `${signature()}=` }), + SECRET, + { method: 'POST', path: INSTALL_PATH }, + now, + ), + ); + }); + + it('classifies unavailable verifier state separately from invalid evidence', () => { + expect(() => + requireTrustedPluginOperatorContext( + context(), + 'short', + { method: 'POST', path: INSTALL_PATH }, + Number(ISSUED_AT), + ), + ).toThrowError( + expect.objectContaining({ + name: 'IntegrationOperatorContextError', + kind: 'unavailable', + }), + ); + }); +}); diff --git a/apps/integration-service/src/plugin-operator-context.ts b/apps/integration-service/src/plugin-operator-context.ts new file mode 100644 index 00000000..e0f49990 --- /dev/null +++ b/apps/integration-service/src/plugin-operator-context.ts @@ -0,0 +1,220 @@ +import { createHmac, timingSafeEqual } from 'node:crypto'; +import type { PluginInstallationContext } from './plugin-installation'; + +/** UUIDv4 grammar accepted for tenant, user, and one-time evidence identities; values normalize to lowercase. */ +const UUID_V4_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu; +/** Canonical unsigned decimal Unix-second grammar used by short-lived signed evidence. */ +const UNIX_SECONDS_PATTERN = /^(?:0|[1-9]\d{0,12})$/u; +/** Canonical unpadded base64url grammar for exactly one SHA-256 HMAC digest. */ +const BASE64URL_SHA256_PATTERN = /^[A-Za-z0-9_-]{43}$/u; +/** Exact lowercase installation item/revocation paths; case variants are never aliases. */ +const INSTALLATION_ROUTE_PATTERN = + /^\/v1\/plugins\/installations\/[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}(?:\/revoke)?$/u; +/** Exact lowercase credential-revocation path; case variants are never aliases. */ +const CREDENTIAL_ROUTE_PATTERN = + /^\/v1\/plugins\/credential-bindings\/[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\/revoke$/u; +/** Minimum UTF-8 verifier-key length required before any caller evidence is evaluated. */ +const MINIMUM_GATEWAY_SECRET_BYTES = 32; +/** Maximum age of otherwise valid operator evidence before it is classified invalid. */ +export const PLUGIN_OPERATOR_CONTEXT_MAXIMUM_AGE_SECONDS = 60; +/** Maximum tolerated positive clock skew before future evidence is classified invalid. */ +const MAXIMUM_FUTURE_SKEW_SECONDS = 5; +/** Exact installation collection path accepted only with POST. */ +const INSTALLATION_COLLECTION_PATH = '/v1/plugins/installations'; +/** Exact credential-binding collection path accepted only with POST. */ +const CREDENTIAL_COLLECTION_PATH = '/v1/plugins/credential-bindings'; + +/** Untrusted signed identity forwarded by the authenticated Integration host. */ +export interface IntegrationOperatorContextHeaders { + readonly workspaceId: unknown; + readonly userId: unknown; + readonly evidenceId: unknown; + readonly issuedAt: unknown; + readonly signature: unknown; +} + +/** Server-observed request identity included in operator-context verification. */ +export interface IntegrationOperatorRequestBinding { + readonly method: unknown; + readonly path: unknown; +} + +/** Verified authority plus the signed one-time evidence identity and issuance time. */ +export interface VerifiedPluginOperatorContext { + readonly trustedContext: PluginInstallationContext; + readonly evidenceId: string; + readonly issuedAtSeconds: number; +} + +/** Fixed, credential-free operator-context rejection safe for HTTP classification. */ +export class IntegrationOperatorContextError extends Error { + /** Creates an invalid-authority or verifier-unavailable failure. */ + constructor(readonly kind: 'invalid' | 'unavailable') { + super( + kind === 'invalid' + ? 'Plugin operator context is invalid' + : 'Plugin operator context is unavailable', + ); + this.name = 'IntegrationOperatorContextError'; + } +} + +/** Classifies malformed, forged, stale, replayed-route, or unsupported caller evidence as invalid. */ +function invalid(): never { + throw new IntegrationOperatorContextError('invalid'); +} + +/** Classifies verifier configuration or clock state that cannot authenticate callers as unavailable. */ +function unavailable(): never { + throw new IntegrationOperatorContextError('unavailable'); +} + +/** Requires UUIDv4 identity evidence and returns its canonical lowercase representation. */ +function requireUuidV4(value: unknown): string { + if (typeof value !== 'string' || !UUID_V4_PATTERN.test(value)) { + return invalid(); + } + return value.toLowerCase(); +} + +/** + * Accepts only the implemented plugin operator method/path surface. + * + * Collection paths are exact constants. Dynamic UUID paths must already use the + * canonical lowercase route grammar and are returned byte-for-byte so an HMAC for + * a normalized alias can never authenticate a different received route. + */ +function requireOperatorRoute( + binding: IntegrationOperatorRequestBinding, +): Readonly<{ method: 'GET' | 'POST'; path: string }> { + if ( + binding.method === 'POST' && + binding.path === INSTALLATION_COLLECTION_PATH + ) { + return Object.freeze({ + method: 'POST', + path: INSTALLATION_COLLECTION_PATH, + }); + } + if ( + binding.method === 'POST' && + binding.path === CREDENTIAL_COLLECTION_PATH + ) { + return Object.freeze({ + method: 'POST', + path: CREDENTIAL_COLLECTION_PATH, + }); + } + if ( + typeof binding.path === 'string' && + INSTALLATION_ROUTE_PATTERN.test(binding.path) + ) { + const isRevocation = binding.path.endsWith('/revoke'); + if ( + (isRevocation && binding.method === 'POST') || + (!isRevocation && binding.method === 'GET') + ) { + return Object.freeze({ + method: binding.method, + path: binding.path, + }); + } + } + if ( + binding.method === 'POST' && + typeof binding.path === 'string' && + CREDENTIAL_ROUTE_PATTERN.test(binding.path) + ) { + return Object.freeze({ + method: 'POST', + path: binding.path, + }); + } + return invalid(); +} + +/** + * Verifies tenant-and-user authority and one-time evidence bound to one exact request. + * + * The signed UUIDv4 evidence identifier makes otherwise identical same-second + * requests distinguishable without trusting caller-selected tenant or user data. + * This verifier remains stateless; its caller must atomically consume the returned + * evidence identifier before any downstream authority is invoked. + */ +export function requireVerifiedPluginOperatorContext( + headers: IntegrationOperatorContextHeaders, + secretValue: unknown, + requestBinding: IntegrationOperatorRequestBinding, + nowSeconds = Math.floor(Date.now() / 1000), +): VerifiedPluginOperatorContext { + if ( + typeof secretValue !== 'string' || + Buffer.byteLength(secretValue, 'utf8') < MINIMUM_GATEWAY_SECRET_BYTES || + !Number.isSafeInteger(nowSeconds) || + nowSeconds < 0 + ) { + return unavailable(); + } + + const binding = requireOperatorRoute(requestBinding); + if ( + typeof headers.issuedAt !== 'string' || + typeof headers.signature !== 'string' || + !UNIX_SECONDS_PATTERN.test(headers.issuedAt) || + !BASE64URL_SHA256_PATTERN.test(headers.signature) + ) { + return invalid(); + } + const workspaceId = requireUuidV4(headers.workspaceId); + const actorUserId = requireUuidV4(headers.userId); + const evidenceId = requireUuidV4(headers.evidenceId); + const issuedAtSeconds = Number(headers.issuedAt); + if ( + !Number.isSafeInteger(issuedAtSeconds) || + issuedAtSeconds > nowSeconds + MAXIMUM_FUTURE_SKEW_SECONDS || + issuedAtSeconds < nowSeconds - PLUGIN_OPERATOR_CONTEXT_MAXIMUM_AGE_SECONDS + ) { + return invalid(); + } + + const actual = Buffer.from(headers.signature, 'base64url'); + if (actual.toString('base64url') !== headers.signature) { + return invalid(); + } + const expected = createHmac('sha256', secretValue) + .update( + `life-os.integration-operator-context.v1\n${workspaceId}\n${actorUserId}\n${evidenceId}\n${headers.issuedAt}\n${binding.method}\n${binding.path}`, + 'utf8', + ) + .digest(); + if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) { + return invalid(); + } + return Object.freeze({ + trustedContext: Object.freeze({ workspaceId, actorUserId }), + evidenceId, + issuedAtSeconds, + }); +} + +/** + * Returns trusted tenant/user authority for callers that do not need replay metadata. + * + * This compatibility verifier does not consume evidence. Security-sensitive + * application boundaries must use `requireVerifiedPluginOperatorContext` and a + * service-owned replay guard before granting downstream authority. + */ +export function requireTrustedPluginOperatorContext( + headers: IntegrationOperatorContextHeaders, + secretValue: unknown, + requestBinding: IntegrationOperatorRequestBinding, + nowSeconds = Math.floor(Date.now() / 1000), +): PluginInstallationContext { + return requireVerifiedPluginOperatorContext( + headers, + secretValue, + requestBinding, + nowSeconds, + ).trustedContext; +} diff --git a/apps/integration-service/src/plugin-operator-replay-migration.test.ts b/apps/integration-service/src/plugin-operator-replay-migration.test.ts new file mode 100644 index 00000000..61d82812 --- /dev/null +++ b/apps/integration-service/src/plugin-operator-replay-migration.test.ts @@ -0,0 +1,201 @@ +import { spawnSync } from 'node:child_process'; +import { readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { beforeEach, describe, expect, it } from 'vitest'; + +const INSTALLATION_MIGRATION_PATH = join( + __dirname, + '..', + 'migrations', + '0001_plugin_installation_record.sql', +); +const CREDENTIAL_MIGRATION_PATH = join( + __dirname, + '..', + 'migrations', + '0002_plugin_credential_binding_record.sql', +); +const REPLAY_MIGRATION_PATH = join( + __dirname, + '..', + 'migrations', + '0003_plugin_operator_context_replay_record.sql', +); +const DATABASE_URL = process.env.INTEGRATION_DATABASE_URL; +const describeWithPostgres = DATABASE_URL ? describe : describe.skip; + +interface SqlExecution { + readonly status: number | null; + readonly stdout: string; + readonly stderr: string; +} + +function replayMigrationSql(): string { + return readFileSync(REPLAY_MIGRATION_PATH, 'utf8'); +} + +function executeSql(sql: string): SqlExecution { + if (!DATABASE_URL) { + throw new Error('An integration PostgreSQL URL is required'); + } + const target = new URL(DATABASE_URL); + const result = spawnSync( + 'psql', + [ + '-X', + '-v', + 'ON_ERROR_STOP=1', + '-h', + target.hostname, + '-p', + target.port || '5432', + '-U', + decodeURIComponent(target.username), + '-d', + decodeURIComponent(target.pathname.replace(/^\//u, '')), + '-Atq', + ], + { + input: sql, + encoding: 'utf8', + env: { + ...process.env, + PGPASSWORD: decodeURIComponent(target.password), + }, + }, + ); + if (result.error) { + throw result.error; + } + return { + status: result.status, + stdout: result.stdout, + stderr: result.stderr, + }; +} + +function requireSqlSuccess(sql: string): string { + const result = executeSql(sql); + if (result.status !== 0) { + throw new Error( + `Plugin operator replay PostgreSQL setup failed: ${result.stderr.slice(0, 500)}`, + ); + } + return result.stdout.trim(); +} + +function expectSqlFailure(sql: string): void { + expect(executeSql(sql).status).not.toBe(0); +} + +describe('plugin operator replay migration contract', () => { + it('owns only the bounded one-time evidence identity and lifetime in Integration persistence', () => { + const sql = replayMigrationSql(); + expect(sql).toContain( + 'CREATE TABLE plugin_integration.plugin_operator_context_replay_record', + ); + for (const column of [ + 'evidence_id uuid PRIMARY KEY', + 'consumed_at timestamptz NOT NULL', + 'expires_at timestamptz NOT NULL', + ]) { + expect(sql).toContain(column); + } + expect(sql).toContain('CHECK (expires_at >= consumed_at)'); + expect(sql).toContain( + 'ON plugin_integration.plugin_operator_context_replay_record (expires_at)', + ); + for (const documentedContract of [ + 'COMMENT ON TABLE plugin_integration.plugin_operator_context_replay_record', + 'COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.evidence_id', + 'COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.consumed_at', + 'COMMENT ON COLUMN plugin_integration.plugin_operator_context_replay_record.expires_at', + ]) { + expect(sql).toContain(documentedContract); + } + }); +}); + +describeWithPostgres('plugin operator replay PostgreSQL constraints', () => { + beforeEach(() => { + requireSqlSuccess('DROP SCHEMA IF EXISTS plugin_integration CASCADE;'); + requireSqlSuccess(readFileSync(INSTALLATION_MIGRATION_PATH, 'utf8')); + requireSqlSuccess(readFileSync(CREDENTIAL_MIGRATION_PATH, 'utf8')); + requireSqlSuccess(replayMigrationSql()); + }); + + it('persists only the allowlisted replay evidence columns with exact types and nullability', () => { + const columns = requireSqlSuccess(` + SELECT column_name || '|' || data_type || '|' || is_nullable + FROM information_schema.columns + WHERE table_schema = 'plugin_integration' + AND table_name = 'plugin_operator_context_replay_record' + ORDER BY ordinal_position; + `); + + expect(columns.split('\n')).toEqual([ + 'evidence_id|uuid|NO', + 'consumed_at|timestamp with time zone|NO', + 'expires_at|timestamp with time zone|NO', + ]); + }); + + it('permits exactly one durable winner for a UUIDv4 evidence identity', () => { + requireSqlSuccess(` + INSERT INTO plugin_integration.plugin_operator_context_replay_record ( + evidence_id, consumed_at, expires_at + ) VALUES ( + '77777777-7777-4777-8777-777777777777', + '2026-08-11T14:35:00.000Z', + '2026-08-11T14:36:00.000Z' + ); + `); + expectSqlFailure(` + INSERT INTO plugin_integration.plugin_operator_context_replay_record ( + evidence_id, consumed_at, expires_at + ) VALUES ( + '77777777-7777-4777-8777-777777777777', + '2026-08-11T14:35:01.000Z', + '2026-08-11T14:36:01.000Z' + ); + `); + expect( + requireSqlSuccess(` + SELECT count(*) + FROM plugin_integration.plugin_operator_context_replay_record + WHERE evidence_id = '77777777-7777-4777-8777-777777777777'::uuid; + `), + ).toBe('1'); + }); + + it('accepts a zero-length retention boundary when consumption and expiry are identical', () => { + requireSqlSuccess(` + INSERT INTO plugin_integration.plugin_operator_context_replay_record ( + evidence_id, consumed_at, expires_at + ) VALUES ( + '99999999-9999-4999-8999-999999999999', + '2026-08-11T14:35:00.000Z', + '2026-08-11T14:35:00.000Z' + ); + `); + expect( + requireSqlSuccess(` + SELECT count(*) + FROM plugin_integration.plugin_operator_context_replay_record + WHERE evidence_id = '99999999-9999-4999-8999-999999999999'::uuid; + `), + ).toBe('1'); + }); + + it('rejects replay lifetimes that expire before consumption', () => { + expectSqlFailure(` + INSERT INTO plugin_integration.plugin_operator_context_replay_record ( + evidence_id, consumed_at, expires_at + ) VALUES ( + '88888888-8888-4888-8888-888888888888', + '2026-08-11T14:35:00.000Z', + '2026-08-11T14:34:59.999Z' + ); + `); + }); +}); diff --git a/apps/integration-service/src/plugin-operator-replay.test.ts b/apps/integration-service/src/plugin-operator-replay.test.ts new file mode 100644 index 00000000..803433db --- /dev/null +++ b/apps/integration-service/src/plugin-operator-replay.test.ts @@ -0,0 +1,156 @@ +import { describe, expect, it } from 'vitest'; +import { + PluginOperatorReplayValidationError, + PostgresPluginOperatorReplayGuard, + type PluginOperatorReplaySqlClient, + type PluginOperatorReplaySqlResult, +} from './plugin-operator-replay'; + +const EVIDENCE_ID = '77777777-7777-4777-8777-777777777777'; +const LOWERCASE_EVIDENCE_ID = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee'; +const CONSUMED_AT = '2026-08-11T14:35:00.000Z'; +const EXPIRES_AT = '2026-08-11T14:36:00.000Z'; + +interface RecordedQuery { + readonly text: string; + readonly values: readonly unknown[] | undefined; +} + +class ScriptedSqlClient implements PluginOperatorReplaySqlClient { + readonly queries: RecordedQuery[] = []; + + constructor( + private readonly results: readonly PluginOperatorReplaySqlResult[], + ) {} + + async query( + text: string, + values?: readonly unknown[], + ): Promise> { + this.queries.push({ text, values }); + const result = this.results[this.queries.length - 1]; + if (!result) { + throw new Error('Unexpected replay-store SQL query'); + } + return result as PluginOperatorReplaySqlResult; + } +} + +function evidence( + overrides: Partial<{ + evidenceId: string; + consumedAt: string; + expiresAt: string; + }> = {}, +) { + return { + evidenceId: EVIDENCE_ID, + consumedAt: CONSUMED_AT, + expiresAt: EXPIRES_AT, + ...overrides, + }; +} + +describe('PostgresPluginOperatorReplayGuard', () => { + it('atomically consumes one evidence UUID after pruning only rows expired by the database clock', async () => { + const client = new ScriptedSqlClient([ + { rows: [], rowCount: 0 }, + { rows: [{ evidence_id: EVIDENCE_ID }], rowCount: 1 }, + ]); + const guard = new PostgresPluginOperatorReplayGuard(client); + + await expect(guard.consume(evidence())).resolves.toBe(true); + + expect(client.queries).toHaveLength(2); + expect(client.queries[0]?.text).toContain( + 'DELETE FROM plugin_integration.plugin_operator_context_replay_record', + ); + expect(client.queries[0]?.text).toContain('expires_at < now()'); + expect(client.queries[0]?.values).toBeUndefined(); + expect(client.queries[1]?.text).toContain( + 'INSERT INTO plugin_integration.plugin_operator_context_replay_record', + ); + expect(client.queries[1]?.text).toContain( + 'ON CONFLICT (evidence_id) DO NOTHING', + ); + expect(client.queries[1]?.values).toEqual([ + EVIDENCE_ID, + CONSUMED_AT, + EXPIRES_AT, + ]); + }); + + it('normalizes accepted UUID evidence to lowercase before persistence', async () => { + const client = new ScriptedSqlClient([ + { rows: [], rowCount: 0 }, + { rows: [{ evidence_id: LOWERCASE_EVIDENCE_ID }], rowCount: 1 }, + ]); + const guard = new PostgresPluginOperatorReplayGuard(client); + + await expect( + guard.consume( + evidence({ evidenceId: LOWERCASE_EVIDENCE_ID.toUpperCase() }), + ), + ).resolves.toBe(true); + + expect(client.queries[1]?.values).toEqual([ + LOWERCASE_EVIDENCE_ID, + CONSUMED_AT, + EXPIRES_AT, + ]); + }); + + it('returns false when another service instance already consumed the evidence UUID', async () => { + const client = new ScriptedSqlClient([ + { rows: [], rowCount: 0 }, + { rows: [], rowCount: 0 }, + ]); + const guard = new PostgresPluginOperatorReplayGuard(client); + + await expect(guard.consume(evidence())).resolves.toBe(false); + }); + + it('rejects malformed or contradictory evidence before issuing SQL', async () => { + for (const candidate of [ + evidence({ evidenceId: 'not-a-uuid' }), + evidence({ consumedAt: '2026-08-11 14:35:00Z' }), + evidence({ expiresAt: '2026-08-11T14:34:59.999Z' }), + ]) { + const client = new ScriptedSqlClient([]); + const guard = new PostgresPluginOperatorReplayGuard(client); + + await expect(guard.consume(candidate)).rejects.toBeInstanceOf( + PluginOperatorReplayValidationError, + ); + expect(client.queries).toHaveLength(0); + } + }); + + it('rejects ambiguous or corrupted INSERT evidence instead of granting authority', async () => { + for (const inserted of [ + { rows: [{ evidence_id: EVIDENCE_ID }], rowCount: null }, + { rows: [], rowCount: 1 }, + { + rows: [ + { evidence_id: EVIDENCE_ID }, + { evidence_id: EVIDENCE_ID }, + ], + rowCount: 2, + }, + { + rows: [{ evidence_id: '88888888-8888-4888-8888-888888888888' }], + rowCount: 1, + }, + ] satisfies readonly PluginOperatorReplaySqlResult[]) { + const client = new ScriptedSqlClient([ + { rows: [], rowCount: 0 }, + inserted, + ]); + const guard = new PostgresPluginOperatorReplayGuard(client); + + await expect(guard.consume(evidence())).rejects.toBeInstanceOf( + PluginOperatorReplayValidationError, + ); + } + }); +}); diff --git a/apps/integration-service/src/plugin-operator-replay.ts b/apps/integration-service/src/plugin-operator-replay.ts new file mode 100644 index 00000000..84904e2e --- /dev/null +++ b/apps/integration-service/src/plugin-operator-replay.ts @@ -0,0 +1,133 @@ +const UUID_V4_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu; +const ISO_INSTANT_PATTERN = + /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/u; + +/** One verified one-time operator evidence identity and its bounded lifetime. */ +export interface PluginOperatorReplayEvidence { + readonly evidenceId: string; + readonly consumedAt: string; + readonly expiresAt: string; +} + +/** Service-owned authority that atomically consumes signed operator evidence once. */ +export interface PluginOperatorReplayGuardPort { + /** Returns true only for the first durable consumption of this evidence identifier. */ + consume(evidence: PluginOperatorReplayEvidence): Promise; +} + +/** Result returned by the bounded replay-evidence SQL client. */ +export interface PluginOperatorReplaySqlResult { + readonly rows: readonly Row[]; + readonly rowCount: number | null; +} + +/** Minimal fixed-query SQL authority used by the PostgreSQL replay guard. */ +export interface PluginOperatorReplaySqlClient { + query( + text: string, + values?: readonly unknown[], + ): Promise>; +} + +/** Rejects malformed replay evidence before it can become persistence authority. */ +export class PluginOperatorReplayValidationError extends Error { + /** Creates a fixed failure without reflecting invalid caller evidence. */ + constructor() { + super('Plugin operator replay evidence is invalid'); + this.name = 'PluginOperatorReplayValidationError'; + } +} + +interface ReplayEvidenceRow { + evidence_id: unknown; +} + +/** Fails closed for malformed replay evidence without reflecting the value. */ +function invalid(): never { + throw new PluginOperatorReplayValidationError(); +} + +/** Requires a canonical UUIDv4 evidence identity. */ +function evidenceId(value: unknown): string { + if (typeof value !== 'string' || !UUID_V4_PATTERN.test(value)) { + return invalid(); + } + return value.toLowerCase(); +} + +/** Requires a canonical UTC millisecond instant. */ +function instant(value: unknown): string { + if (typeof value !== 'string' || !ISO_INSTANT_PATTERN.test(value)) { + return invalid(); + } + const parsed = new Date(value); + if (!Number.isFinite(parsed.getTime()) || parsed.toISOString() !== value) { + return invalid(); + } + return value; +} + +/** Validates one replay record and preserves its immutable time ordering. */ +function replayEvidence( + value: PluginOperatorReplayEvidence, +): PluginOperatorReplayEvidence { + const consumedAt = instant(value.consumedAt); + const expiresAt = instant(value.expiresAt); + if (new Date(expiresAt).getTime() < new Date(consumedAt).getTime()) { + return invalid(); + } + return Object.freeze({ + evidenceId: evidenceId(value.evidenceId), + consumedAt, + expiresAt, + }); +} + +/** + * PostgreSQL implementation of the one-time operator evidence guard. + * + * A primary-key insert is the distributed compare-and-set boundary: exactly one + * service instance can consume a signed evidence UUID. Expired rows are pruned + * against the database clock, while still-valid rows remain durable across + * processes so horizontal replicas cannot replay the same authority independently. + */ +export class PostgresPluginOperatorReplayGuard + implements PluginOperatorReplayGuardPort +{ + /** Creates the guard over a bounded parameterized SQL client. */ + constructor(private readonly client: PluginOperatorReplaySqlClient) {} + + /** Atomically consumes one evidence UUID and returns false for an existing winner. */ + async consume(evidence: PluginOperatorReplayEvidence): Promise { + const safe = replayEvidence(evidence); + await this.client.query( + `DELETE FROM plugin_integration.plugin_operator_context_replay_record + WHERE expires_at < now()`, + ); + const inserted = await this.client.query( + `INSERT INTO plugin_integration.plugin_operator_context_replay_record ( + evidence_id, consumed_at, expires_at + ) VALUES ($1::uuid, $2::timestamptz, $3::timestamptz) + ON CONFLICT (evidence_id) DO NOTHING + RETURNING evidence_id`, + [safe.evidenceId, safe.consumedAt, safe.expiresAt], + ); + if (inserted.rows.length > 1 || inserted.rowCount === null) { + return invalid(); + } + if (inserted.rows.length === 0) { + if (inserted.rowCount !== 0) { + return invalid(); + } + return false; + } + if ( + inserted.rowCount !== 1 || + evidenceId(inserted.rows[0]?.evidence_id) !== safe.evidenceId + ) { + return invalid(); + } + return true; + } +}