From 1cc4a995eedee0a0414d2cf6f31fb9ff08480f78 Mon Sep 17 00:00:00 2001 From: chick Date: Fri, 17 Jul 2026 11:57:20 +0800 Subject: [PATCH] feat(api): advance capability and secure operations slices --- .../capability-observation-service.test.ts | 258 ++++++++++ .../capability-observation-service.ts | 216 ++++++++ .../delete-instance-operation.test.ts | 72 +++ .../operations/delete-instance-operation.ts | 68 ++- .../secure-operation-execution.test.ts | 215 ++++++++ .../operations/secure-operation-execution.ts | 466 ++++++++++++++++++ .../status/status-poll-scheduler.test.ts | 249 ++++++++++ .../status/status-poll-scheduler.ts | 220 +++++++++ apps/api/src/control-plane.operations.test.ts | 4 + apps/api/src/control-plane.test.ts | 1 + apps/api/src/control-plane.ts | 27 +- .../src/infrastructure/database/migrations.ts | 42 ++ .../safe-control-plane-upstream.test.ts | 9 +- .../transport/safe-control-plane-upstream.ts | 2 + .../transport/safe-upstream-gateway.test.ts | 30 +- .../transport/safe-upstream-gateway.ts | 32 ++ .../api/src/interface/http/instance-routes.ts | 11 +- .../src/interface/http/operation-routes.ts | 81 ++- .../http/secure-operation-routes.test.ts | 121 +++++ 19 files changed, 2102 insertions(+), 22 deletions(-) create mode 100644 apps/api/src/application/capabilities/capability-observation-service.test.ts create mode 100644 apps/api/src/application/capabilities/capability-observation-service.ts create mode 100644 apps/api/src/application/operations/secure-operation-execution.test.ts create mode 100644 apps/api/src/application/operations/secure-operation-execution.ts create mode 100644 apps/api/src/application/status/status-poll-scheduler.test.ts create mode 100644 apps/api/src/application/status/status-poll-scheduler.ts create mode 100644 apps/api/src/interface/http/secure-operation-routes.test.ts diff --git a/apps/api/src/application/capabilities/capability-observation-service.test.ts b/apps/api/src/application/capabilities/capability-observation-service.test.ts new file mode 100644 index 0000000..221ee0e --- /dev/null +++ b/apps/api/src/application/capabilities/capability-observation-service.test.ts @@ -0,0 +1,258 @@ +import Database from 'better-sqlite3'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { UpstreamError } from '../../infrastructure/transport/upstream-error.js'; +import { + CapabilityObservationError, + CapabilityObservationService, + parseHealthVersion, + type CapabilityOperationDescriptor, +} from './capability-observation-service.js'; + +const databases: Database.Database[] = []; +afterEach(() => { + for (const database of databases.splice(0)) database.close(); +}); + +const operation = ( + overrides: Partial = {}, +): CapabilityOperationDescriptor => ({ + operationId: 'getHealth', + method: 'GET', + pathTemplate: '/api/health', + riskLevel: 'R0', + ...overrides, +}); + +function fixture( + options: { + descriptor?: CapabilityOperationDescriptor; + response?: { + status: number; + headers: Readonly>; + body: string; + }; + failure?: unknown; + now?: string; + } = {}, +) { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + databases.push(db); + const created = '2026-07-17T10:00:00.000Z'; + db.prepare( + 'INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)', + ).run('instance-1', 'LAN', 'http://192.168.1.20:3000', 'none', 1, 1, created, created); + const events: string[] = []; + const instances = { + get: vi.fn(async (id: string) => { + events.push('instance'); + return id === 'instance-1' ? { id, origin: 'http://192.168.1.20:3000' } : undefined; + }), + }; + const registry = { + requireOperation: vi.fn((id: string) => { + events.push('operation'); + if (id !== 'getHealth') throw new Error('unknown operation'); + return options.descriptor ?? operation(); + }), + }; + const transport = { + get: vi.fn(async (url: string) => { + events.push('network'); + expect(url).toBe('http://192.168.1.20:3000/api/health'); + if (options.failure) throw options.failure; + return options.response ?? { status: 200, headers: {}, body: '{"version":"1.2.3"}' }; + }), + }; + const service = new CapabilityObservationService({ + db, + instances, + registry, + transport, + now: () => new Date(options.now ?? '2026-07-17T12:00:00.000Z'), + }); + return { db, events, instances, registry, transport, service }; +} + +const row = (db: Database.Database) => + db + .prepare( + 'SELECT instance_id,operation_id,state,upstream_version,detail_code,observed_at,created_at,updated_at FROM capabilities', + ) + .get(); + +describe('parseHealthVersion', () => { + it.each([ + ['{"version":"1.2.3"}', '1.2.3'], + ['{"data":{"version":"v2026.07.17-beta+5"}}', 'v2026.07.17-beta+5'], + ['{"version":" 2.0.0 "}', '2.0.0'], + ])('extracts a bounded version from supported health envelopes', (body, expected) => { + expect(parseHealthVersion(body)).toBe(expected); + }); + + it.each([ + ['not-json'], + ['null'], + ['{"version":7}'], + ['{"version":""}'], + [`{"version":"${'a'.repeat(129)}"}`], + ['{"version":"1.2.3\\nsecret"}'], + ['{"data":{"version":{"raw":"secret"}}}'], + ])('rejects malformed, unbounded, or unsafe values without leaking them', (body) => { + expect(parseHealthVersion(body)).toBeUndefined(); + }); +}); + +describe('CapabilityObservationService', () => { + it('resolves instance then operation before probing an R0 GET and persists one redacted supported row', async () => { + const { db, events, service } = fixture(); + await expect(service.observe('instance-1', 'getHealth')).resolves.toEqual({ + instanceId: 'instance-1', + operationId: 'getHealth', + state: 'supported', + upstreamVersion: '1.2.3', + detailCode: 'ACTIVE_PROBE_SUPPORTED', + observedAt: '2026-07-17T12:00:00.000Z', + }); + expect(events).toEqual(['instance', 'operation', 'network']); + expect(row(db)).toEqual({ + instance_id: 'instance-1', + operation_id: 'getHealth', + state: 'supported', + upstream_version: '1.2.3', + detail_code: 'ACTIVE_PROBE_SUPPORTED', + observed_at: '2026-07-17T12:00:00.000Z', + created_at: '2026-07-17T12:00:00.000Z', + updated_at: '2026-07-17T12:00:00.000Z', + }); + }); + + it.each([ + [401, 'auth-required', 'ACTIVE_PROBE_AUTH_REQUIRED'], + [403, 'auth-required', 'ACTIVE_PROBE_AUTH_REQUIRED'], + [404, 'unsupported', 'ACTIVE_PROBE_UNSUPPORTED'], + [500, 'degraded', 'ACTIVE_PROBE_UPSTREAM_DEGRADED'], + [503, 'degraded', 'ACTIVE_PROBE_UPSTREAM_DEGRADED'], + ] as const)( + 'classifies HTTP %i without persisting response content', + async (status, state, detailCode) => { + const body = '{"password":"do-not-store","version":"9.9.9"}'; + const { db, service } = fixture({ + response: { status, headers: { 'set-cookie': 'secret=1' }, body }, + }); + await expect(service.observe('instance-1', 'getHealth')).resolves.toMatchObject({ + state, + detailCode, + upstreamVersion: undefined, + }); + expect(JSON.stringify(row(db))).not.toContain('do-not-store'); + expect(JSON.stringify(row(db))).not.toContain('secret=1'); + }, + ); + + it.each([ + [new UpstreamError('UPSTREAM_TIMEOUT'), 'ACTIVE_PROBE_TIMEOUT'], + [new UpstreamError('UPSTREAM_UNAVAILABLE'), 'ACTIVE_PROBE_NETWORK_ERROR'], + [new Error('password=never-persist'), 'ACTIVE_PROBE_NETWORK_ERROR'], + ])('maps transport failures to unknown stable codes only', async (failure, detailCode) => { + const { db, service } = fixture({ failure }); + await expect(service.observe('instance-1', 'getHealth')).resolves.toMatchObject({ + state: 'unknown', + detailCode, + }); + expect(JSON.stringify(row(db))).not.toContain(failure.message); + }); + + it.each([ + [operation({ operationId: 'getHealth', method: 'POST', riskLevel: 'R1' })], + [operation({ operationId: 'getHealth', method: 'DELETE', riskLevel: 'R3' })], + [operation({ operationId: 'getHealth', method: 'GET', riskLevel: 'R1' })], + ])('never actively probes non-R0-GET operations and records unknown', async (descriptor) => { + const { db, service, transport } = fixture({ descriptor }); + await expect(service.observe('instance-1', 'getHealth')).resolves.toMatchObject({ + state: 'unknown', + detailCode: 'ACTIVE_PROBE_NOT_SAFE', + upstreamVersion: undefined, + }); + expect(transport.get).not.toHaveBeenCalled(); + expect(row(db)).toBeTruthy(); + }); + + it('does not let an older in-flight observation overwrite a newer completed observation', async () => { + const { db, service, transport } = fixture(); + let resolveOlder!: (response: { + status: number; + headers: Record; + body: string; + }) => void; + const olderResponse = new Promise<{ + status: number; + headers: Record; + body: string; + }>((resolve) => { + resolveOlder = resolve; + }); + transport.get + .mockImplementationOnce(async () => olderResponse) + .mockResolvedValueOnce({ status: 404, headers: {}, body: 'newer-sensitive-body' }); + + const older = service.observe('instance-1', 'getHealth'); + await vi.waitFor(() => expect(transport.get).toHaveBeenCalledTimes(1)); + const newer = service.observe('instance-1', 'getHealth'); + await expect(newer).resolves.toMatchObject({ + state: 'unsupported', + detailCode: 'ACTIVE_PROBE_UNSUPPORTED', + }); + + resolveOlder({ status: 200, headers: {}, body: '{"version":"older-secret-version"}' }); + await expect(older).resolves.toMatchObject({ state: 'supported' }); + expect(row(db)).toMatchObject({ + state: 'unsupported', + upstream_version: null, + detail_code: 'ACTIVE_PROBE_UNSUPPORTED', + }); + expect(JSON.stringify(row(db))).not.toContain('older-secret-version'); + expect(JSON.stringify(row(db))).not.toContain('newer-sensitive-body'); + }); + + it('upserts the single instance-operation row while preserving created_at', async () => { + const first = fixture(); + await first.service.observe('instance-1', 'getHealth'); + const second = new CapabilityObservationService({ + db: first.db, + instances: first.instances, + registry: first.registry, + transport: { get: async () => ({ status: 404, headers: {}, body: 'sensitive' }) }, + now: () => new Date('2026-07-17T13:00:00.000Z'), + }); + await second.observe('instance-1', 'getHealth'); + expect(first.db.prepare('SELECT COUNT(*) count FROM capabilities').get()).toEqual({ count: 1 }); + expect(row(first.db)).toMatchObject({ + state: 'unsupported', + upstream_version: null, + created_at: '2026-07-17T12:00:00.000Z', + updated_at: '2026-07-17T13:00:00.000Z', + }); + }); + + it('does no operation resolution, write, or network for an unknown instance', async () => { + const { db, service, registry, transport } = fixture(); + await expect(service.observe('missing', 'getHealth')).rejects.toEqual( + expect.objectContaining>({ code: 'INSTANCE_NOT_FOUND' }), + ); + expect(registry.requireOperation).not.toHaveBeenCalled(); + expect(transport.get).not.toHaveBeenCalled(); + expect(row(db)).toBeUndefined(); + }); + + it('does no write or network for an unknown operation', async () => { + const { db, service, transport } = fixture(); + await expect(service.observe('instance-1', 'missing')).rejects.toEqual( + expect.objectContaining>({ code: 'OPERATION_NOT_FOUND' }), + ); + expect(transport.get).not.toHaveBeenCalled(); + expect(row(db)).toBeUndefined(); + }); +}); diff --git a/apps/api/src/application/capabilities/capability-observation-service.ts b/apps/api/src/application/capabilities/capability-observation-service.ts new file mode 100644 index 0000000..2dd6738 --- /dev/null +++ b/apps/api/src/application/capabilities/capability-observation-service.ts @@ -0,0 +1,216 @@ +import type Database from 'better-sqlite3'; +import { UpstreamError } from '../../infrastructure/transport/upstream-error.js'; +import type { TransportResponse } from '../../infrastructure/transport/safe-instance-transport.js'; + +export type CapabilityState = + | 'supported' + | 'unsupported' + | 'auth-required' + | 'degraded' + | 'unknown'; + +export type CapabilityDetailCode = + | 'ACTIVE_PROBE_SUPPORTED' + | 'ACTIVE_PROBE_AUTH_REQUIRED' + | 'ACTIVE_PROBE_UNSUPPORTED' + | 'ACTIVE_PROBE_UPSTREAM_DEGRADED' + | 'ACTIVE_PROBE_TIMEOUT' + | 'ACTIVE_PROBE_NETWORK_ERROR' + | 'ACTIVE_PROBE_UNEXPECTED_STATUS' + | 'ACTIVE_PROBE_NOT_SAFE'; + +export interface CapabilityOperationDescriptor { + readonly operationId: string; + readonly method: string; + readonly pathTemplate: string; + readonly riskLevel: 'R0' | 'R1' | 'R2' | 'R3'; +} + +export interface CapabilityInstance { + readonly id: string; + readonly origin: string; +} + +export interface CapabilityObservation { + readonly instanceId: string; + readonly operationId: string; + readonly state: CapabilityState; + readonly upstreamVersion: string | undefined; + readonly detailCode: CapabilityDetailCode; + readonly observedAt: string; +} + +export interface CapabilityObservationOptions { + readonly db: Database.Database; + readonly instances: { get(instanceId: string): Promise }; + readonly registry: { requireOperation(operationId: string): CapabilityOperationDescriptor }; + readonly transport: { get(url: string): Promise }; + readonly now?: () => Date; +} + +export type CapabilityObservationErrorCode = 'INSTANCE_NOT_FOUND' | 'OPERATION_NOT_FOUND'; + +export class CapabilityObservationError extends Error { + constructor(readonly code: CapabilityObservationErrorCode) { + super(code); + this.name = 'CapabilityObservationError'; + } +} + +const MAX_VERSION_LENGTH = 128; +const VERSION_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._+-]*$/; + +/** Maps only documented health envelope locations into bounded, inert version metadata. */ +export function parseHealthVersion(body: string): string | undefined { + let value: unknown; + try { + value = JSON.parse(body); + } catch { + return undefined; + } + if (value === null || typeof value !== 'object' || Array.isArray(value)) return undefined; + const root = value as Record; + const data = root.data; + const candidate = + root.version ?? + (data !== null && typeof data === 'object' && !Array.isArray(data) + ? (data as Record).version + : undefined); + if (typeof candidate !== 'string') return undefined; + const normalized = candidate.trim(); + return normalized.length > 0 && + normalized.length <= MAX_VERSION_LENGTH && + VERSION_PATTERN.test(normalized) + ? normalized + : undefined; +} + +const joinUrl = (origin: string, path: string): string => + `${origin.replace(/\/$/, '')}/${path.replace(/^\//, '')}`; + +function classifyStatus(status: number): { + state: CapabilityState; + detailCode: CapabilityDetailCode; +} { + if (status >= 200 && status < 300) + return { state: 'supported', detailCode: 'ACTIVE_PROBE_SUPPORTED' }; + if (status === 401 || status === 403) + return { state: 'auth-required', detailCode: 'ACTIVE_PROBE_AUTH_REQUIRED' }; + if (status === 404) return { state: 'unsupported', detailCode: 'ACTIVE_PROBE_UNSUPPORTED' }; + if (status >= 500 && status < 600) + return { state: 'degraded', detailCode: 'ACTIVE_PROBE_UPSTREAM_DEGRADED' }; + return { state: 'unknown', detailCode: 'ACTIVE_PROBE_UNEXPECTED_STATUS' }; +} + +export class CapabilityObservationService { + readonly #db: Database.Database; + readonly #instances: CapabilityObservationOptions['instances']; + readonly #registry: CapabilityObservationOptions['registry']; + readonly #transport: CapabilityObservationOptions['transport']; + readonly #now: () => Date; + readonly #latestGeneration = new Map(); + #nextGeneration = 0; + + constructor(options: CapabilityObservationOptions) { + this.#db = options.db; + this.#instances = options.instances; + this.#registry = options.registry; + this.#transport = options.transport; + this.#now = options.now ?? (() => new Date()); + } + + async observe(instanceId: string, operationId: string): Promise { + // Resolution order is intentional: neither unknown identity crosses the network or persistence boundary. + const instance = await this.#instances.get(instanceId); + if (!instance) throw new CapabilityObservationError('INSTANCE_NOT_FOUND'); + + let operation: CapabilityOperationDescriptor; + try { + operation = this.#registry.requireOperation(operationId); + } catch { + throw new CapabilityObservationError('OPERATION_NOT_FOUND'); + } + + const observationKey = JSON.stringify([instanceId, operation.operationId]); + const generation = ++this.#nextGeneration; + this.#latestGeneration.set(observationKey, generation); + const observedAt = this.#now().toISOString(); + if (operation.method !== 'GET' || operation.riskLevel !== 'R0') { + return this.#persist( + { + instanceId, + operationId: operation.operationId, + state: 'unknown', + upstreamVersion: undefined, + detailCode: 'ACTIVE_PROBE_NOT_SAFE', + observedAt, + }, + observationKey, + generation, + ); + } + + let state: CapabilityState; + let detailCode: CapabilityDetailCode; + let upstreamVersion: string | undefined; + try { + const response = await this.#transport.get(joinUrl(instance.origin, operation.pathTemplate)); + ({ state, detailCode } = classifyStatus(response.status)); + if (state === 'supported' && operation.operationId === 'getHealth') { + upstreamVersion = parseHealthVersion(response.body); + } + } catch (error) { + state = 'unknown'; + detailCode = + error instanceof UpstreamError && error.code === 'UPSTREAM_TIMEOUT' + ? 'ACTIVE_PROBE_TIMEOUT' + : 'ACTIVE_PROBE_NETWORK_ERROR'; + } + return this.#persist( + { + instanceId, + operationId: operation.operationId, + state, + upstreamVersion, + detailCode, + observedAt, + }, + observationKey, + generation, + ); + } + + #persist( + observation: CapabilityObservation, + observationKey: string, + generation: number, + ): CapabilityObservation { + if (this.#latestGeneration.get(observationKey) !== generation) { + return Object.freeze({ ...observation }); + } + this.#db + .prepare( + `INSERT INTO capabilities + (instance_id,operation_id,state,upstream_version,detail_code,observed_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?) + ON CONFLICT(instance_id,operation_id) DO UPDATE SET + state=excluded.state, + upstream_version=excluded.upstream_version, + detail_code=excluded.detail_code, + observed_at=excluded.observed_at, + updated_at=excluded.updated_at`, + ) + .run( + observation.instanceId, + observation.operationId, + observation.state, + observation.upstreamVersion ?? null, + observation.detailCode, + observation.observedAt, + observation.observedAt, + observation.observedAt, + ); + this.#latestGeneration.delete(observationKey); + return Object.freeze({ ...observation }); + } +} diff --git a/apps/api/src/application/operations/delete-instance-operation.test.ts b/apps/api/src/application/operations/delete-instance-operation.test.ts index ee188b3..ef75c81 100644 --- a/apps/api/src/application/operations/delete-instance-operation.test.ts +++ b/apps/api/src/application/operations/delete-instance-operation.test.ts @@ -192,6 +192,78 @@ describe('DeleteInstanceOperation', () => { expect(db.prepare('SELECT COUNT(*) count FROM jobs').get()).toEqual({ count: 0 }); }); + it.each([ + ['risk_level', 'R2'], + ['target_origin', 'http://192.168.1.99'], + ['method', 'POST'], + ['path', '/api/v1/instances/someone-else'], + ['canonical_query', 'force=true'], + ['body_digest', 'a'.repeat(64)], + ['content_type', 'application/json'], + ['nonce', 'tampered-nonce'], + ])( + 'fails closed and invalidates a confirmation with tampered durable %s binding', + async (column, value) => { + const { db, instances, operation } = fixture(); + const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); + const prepared = await operation.prepare({ + operationId: 'deleteInstance', + targets: [{ instanceId: instance.id, revision: 1 }], + parameters: { parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] }, + }); + db.prepare(`UPDATE operation_preparations SET ${column}=? WHERE id=?`).run( + value, + prepared.id, + ); + + await expect( + operation.executeDelete({ + instanceId: instance.id, + revision: 1, + preparationId: prepared.id, + confirmationToken: prepared.confirmationToken, + actor: 'loopback-control-plane', + requestId: 'tampered-binding', + }), + ).rejects.toMatchObject({ code: 'CONFIRMATION_INVALID' }); + expect(await instances.get(instance.id)).toBeDefined(); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(prepared.id), + ).toEqual({ status: 'invalidated' }); + expect(db.prepare('SELECT COUNT(*) count FROM jobs').get()).toEqual({ count: 0 }); + }, + ); + + it('fails closed when the current target origin no longer matches the durable binding', async () => { + const { db, instances, operation } = fixture(); + const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); + const prepared = await operation.prepare({ + operationId: 'deleteInstance', + targets: [{ instanceId: instance.id, revision: 1 }], + parameters: { parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] }, + }); + db.prepare('UPDATE instances SET base_url=? WHERE id=?').run( + 'http://192.168.1.11', + instance.id, + ); + + await expect( + operation.executeDelete({ + instanceId: instance.id, + revision: 1, + preparationId: prepared.id, + confirmationToken: prepared.confirmationToken, + actor: 'loopback-control-plane', + requestId: 'changed-origin', + }), + ).rejects.toMatchObject({ code: 'CONFIRMATION_INVALID' }); + expect(await instances.get(instance.id)).toBeDefined(); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(prepared.id), + ).toEqual({ status: 'invalidated' }); + expect(db.prepare('SELECT COUNT(*) count FROM jobs').get()).toEqual({ count: 0 }); + }); + it('returns typed target errors after consuming a correctly bound confirmation', async () => { const { db, instances, operation } = fixture(); const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); diff --git a/apps/api/src/application/operations/delete-instance-operation.ts b/apps/api/src/application/operations/delete-instance-operation.ts index cca0580..ef30915 100644 --- a/apps/api/src/application/operations/delete-instance-operation.ts +++ b/apps/api/src/application/operations/delete-instance-operation.ts @@ -29,11 +29,19 @@ export class DeleteInstanceOperationError extends Error { interface PreparationRow { operation_id: string; + risk_level: string; status: string; target_instance_id: string; target_revision: number; + target_origin: string; + method: string; + path: string; + canonical_query: string; + body_digest: string; + content_type: string; parameter_schema_id: string; parameters_digest: string; + nonce: string; token_digest: string; expires_at: string; } @@ -128,8 +136,10 @@ export class DeleteInstanceOperation { this.db .prepare( `INSERT INTO operation_preparations - (id,operation_id,risk_level,status,target_instance_id,target_revision,parameter_schema_id,parameters_digest,token_digest,requested_by,request_id,expires_at,created_at,updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + (id,operation_id,risk_level,status,target_instance_id,target_revision,target_origin,method,path, + canonical_query,body_digest,content_type,parameter_schema_id,parameters_digest,nonce,token_digest, + requested_by,request_id,expires_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, ) .run( id, @@ -138,8 +148,15 @@ export class DeleteInstanceOperation { 'prepared', target.instanceId, target.revision, + current.origin, + 'DELETE', + `/api/v1/instances/${target.instanceId}`, + '', + digest(''), + '', DELETE_INSTANCE_PARAMETER_SCHEMA_ID, PARAMETERS_DIGEST, + `delete-${id}`, digest(token), ACTOR, requestId, @@ -169,8 +186,9 @@ export class DeleteInstanceOperation { this.db.transaction(() => { const row = this.db .prepare( - `SELECT operation_id,status,target_instance_id,target_revision,parameter_schema_id, - parameters_digest,token_digest,expires_at + `SELECT operation_id,risk_level,status,target_instance_id,target_revision,target_origin, + method,path,canonical_query,body_digest,content_type,parameter_schema_id, + parameters_digest,nonce,token_digest,expires_at FROM operation_preparations WHERE id=?`, ) .get(input.preparationId) as PreparationRow | undefined; @@ -180,15 +198,44 @@ export class DeleteInstanceOperation { 'Confirmation could not be accepted', ); - const secretValid = - row.expires_at > now && - equalDigest(row.token_digest, digest(input.confirmationToken)) && + const confirmationValid = + row.expires_at > now && equalDigest(row.token_digest, digest(input.confirmationToken)); + if (!confirmationValid) return; + + const expectedPath = `/api/v1/instances/${input.instanceId}`; + const bindingValid = row.operation_id === OPERATION_ID && + row.risk_level === 'R3' && row.target_instance_id === input.instanceId && row.target_revision === input.revision && + row.method === 'DELETE' && + row.path === expectedPath && + row.canonical_query === '' && + row.body_digest === digest('') && + row.content_type === '' && row.parameter_schema_id === DELETE_INSTANCE_PARAMETER_SCHEMA_ID && - row.parameters_digest === PARAMETERS_DIGEST; - if (!secretValid) return; + row.parameters_digest === PARAMETERS_DIGEST && + row.nonce === `delete-${input.preparationId}`; + if (!bindingValid) { + this.db + .prepare( + "UPDATE operation_preparations SET status='invalidated',consumed_at=?,updated_at=? WHERE id=? AND status='prepared'", + ) + .run(now, now, input.preparationId); + return; + } + + const current = this.db + .prepare('SELECT base_url,config_revision FROM instances WHERE id=?') + .get(input.instanceId) as { base_url: string; config_revision: number } | undefined; + if (current && current.base_url !== row.target_origin) { + this.db + .prepare( + "UPDATE operation_preparations SET status='invalidated',consumed_at=?,updated_at=? WHERE id=? AND status='prepared'", + ) + .run(now, now, input.preparationId); + return; + } // A correctly authenticated confirmation is one-shot, including stale/missing target results. this.db @@ -196,9 +243,6 @@ export class DeleteInstanceOperation { "UPDATE operation_preparations SET status='consumed',consumed_at=?,updated_at=? WHERE id=? AND status='prepared'", ) .run(now, now, input.preparationId); - const current = this.db - .prepare('SELECT config_revision FROM instances WHERE id=?') - .get(input.instanceId) as { config_revision: number } | undefined; if (!current) { targetError = 'NOT_FOUND'; return; diff --git a/apps/api/src/application/operations/secure-operation-execution.test.ts b/apps/api/src/application/operations/secure-operation-execution.test.ts new file mode 100644 index 0000000..cd5b8a5 --- /dev/null +++ b/apps/api/src/application/operations/secure-operation-execution.test.ts @@ -0,0 +1,215 @@ +import Database from 'better-sqlite3'; +import { createHash } from 'node:crypto'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { + OperationNotDispatchedError, + SecureOperationExecution, + secureOperationRegistry, + type SafeOperationTransport, +} from './secure-operation-execution.js'; + +const SCHEMA = 'simadmin.58e2204.postNetworkRegisterAuto.parameters.v1'; +const databases: Database.Database[] = []; +afterEach(() => databases.splice(0).forEach((db) => db.close())); + +function fixture(transport?: SafeOperationTransport) { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + databases.push(db); + db.prepare( + `INSERT INTO instances + (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) + VALUES ('i-1','one','http://192.168.1.10','none',1,3,'2026-07-17T00:00:00.000Z','2026-07-17T00:00:00.000Z')`, + ).run(); + let sequence = 0; + const request = vi.fn(async () => ({ status: 204 })); + const execution = new SecureOperationExecution({ + db, + registry: secureOperationRegistry, + transport: transport ?? { request }, + now: () => new Date('2026-07-17T12:00:00.000Z'), + idFactory: () => `id-${++sequence}`, + tokenFactory: () => Buffer.alloc(32, 9).toString('base64url'), + nonceFactory: () => `nonce-${++sequence}`, + }); + const input = { + operationId: 'postNetworkRegisterAuto', + targets: [{ instanceId: 'i-1', revision: 3 }], + parameters: { parameterSchemaId: SCHEMA, fields: [] }, + } as const; + return { db, execution, input, request }; +} + +const expectCode = async (promise: Promise, code: string) => + expect(promise).rejects.toMatchObject({ code }); + +describe('SecureOperationExecution generic R2 slice', () => { + it('prepares the sole audited zero-parameter operation without upstream or a job and persists full binding plus digest only', async () => { + const { db, execution, input, request } = fixture(); + const prepared = await execution.prepare(input, 'request-1'); + expect(prepared).toMatchObject({ + operationId: input.operationId, + risk: 'R2', + status: 'prepared', + targetCount: 1, + }); + expect(prepared.expiresAt).toBe('2026-07-17T12:05:00.000Z'); + expect(request).not.toHaveBeenCalled(); + expect(db.prepare('SELECT count(*) count FROM jobs').get()).toEqual({ count: 0 }); + const row = db + .prepare('SELECT * FROM operation_preparations WHERE id=?') + .get(prepared.id) as Record; + expect(row).toMatchObject({ + target_instance_id: 'i-1', + target_revision: 3, + target_origin: 'http://192.168.1.10', + operation_id: input.operationId, + risk_level: 'R2', + method: 'POST', + path: '/api/network/register-auto', + canonical_query: '', + body_digest: createHash('sha256').update('').digest('hex'), + content_type: '', + parameter_schema_id: SCHEMA, + status: 'prepared', + }); + expect(row.token_digest).toBe( + createHash('sha256').update(prepared.confirmationToken).digest('hex'), + ); + expect(JSON.stringify(row)).not.toContain(prepared.confirmationToken); + expect(typeof row.nonce).toBe('string'); + }); + + it('strictly rejects unknown shapes, operations, parameters, targets, and stale revisions', async () => { + const { execution, input } = fixture(); + await expectCode(execution.prepare({ ...input, extra: true } as never), 'VALIDATION_FAILED'); + await expectCode( + execution.prepare({ ...input, operationId: 'getDevice' }), + 'OPERATION_NOT_ALLOWED', + ); + await expectCode(execution.prepare({ ...input, targets: [] }), 'VALIDATION_FAILED'); + await expectCode( + execution.prepare({ ...input, targets: [{ instanceId: 'i-1', revision: 2 }] }), + 'REVISION_CONFLICT', + ); + await expectCode( + execution.prepare({ + ...input, + parameters: { ...input.parameters, fields: [{ fieldId: 'x', kind: 'string', value: 'x' }] }, + } as never), + 'VALIDATION_FAILED', + ); + }); + + it('does not consume on wrong token, expires atomically, and consumes exactly once after re-resolving bindings', async () => { + const { db, execution, input, request } = fixture(); + const prepared = await execution.prepare(input); + await expectCode( + execution.execute({ preparationId: prepared.id, confirmationToken: 'wrong' }, 'actor', 'r2'), + 'CONFIRMATION_INVALID', + ); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(prepared.id), + ).toEqual({ status: 'prepared' }); + const job = await execution.execute( + { preparationId: prepared.id, confirmationToken: prepared.confirmationToken }, + 'actor', + 'r3', + ); + expect(job).toMatchObject({ + operationId: input.operationId, + status: 'succeeded', + items: [{ targetId: 'i-1', state: 'succeeded' }], + }); + expect(request).toHaveBeenCalledWith({ + origin: 'http://192.168.1.10', + method: 'POST', + path: '/api/network/register-auto', + query: '', + contentType: '', + body: undefined, + }); + await expectCode( + execution.execute( + { preparationId: prepared.id, confirmationToken: prepared.confirmationToken }, + 'actor', + 'r4', + ), + 'CONFIRMATION_INVALID', + ); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(prepared.id), + ).toEqual({ status: 'consumed' }); + }); + + it('marks an expired preparation expired without transport', async () => { + const { db, execution, input, request } = fixture(); + const prepared = await execution.prepare(input); + db.prepare( + "UPDATE operation_preparations SET expires_at='2020-01-01T00:00:00.000Z' WHERE id=?", + ).run(prepared.id); + await expectCode( + execution.execute( + { preparationId: prepared.id, confirmationToken: prepared.confirmationToken }, + 'a', + 'r', + ), + 'CONFIRMATION_INVALID', + ); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(prepared.id), + ).toEqual({ status: 'expired' }); + expect(request).not.toHaveBeenCalled(); + }); + + it('durably creates running lineage before transport and maps ambiguity/interruption to unknown-result', async () => { + let observed = false; + const { db, execution, input } = fixture({ + request: async () => { + observed = + ( + db.prepare("SELECT count(*) count FROM jobs WHERE status='running'").get() as { + count: number; + } + ).count === 1; + throw new Error('socket reset'); + }, + }); + const prepared = await execution.prepare(input); + const job = await execution.execute( + { preparationId: prepared.id, confirmationToken: prepared.confirmationToken }, + 'a', + 'r', + ); + expect(observed).toBe(true); + expect(job.status).toBe('unknown-result'); + expect(job.attempts[0]?.state).toBe('unknown-result'); + expect(execution.reconcileInterruptedJobs()).toBe(0); + db.prepare("UPDATE jobs SET status='running',finished_at=NULL").run(); + db.prepare("UPDATE job_items SET status='running',finished_at=NULL").run(); + db.prepare("UPDATE job_attempts SET status='running',finished_at=NULL").run(); + expect(execution.reconcileInterruptedJobs()).toBe(1); + expect(db.prepare('SELECT status FROM jobs').get()).toEqual({ status: 'unknown-result' }); + }); + + it('maps a deterministic pre-dispatch rejection to failed rather than unknown-result', async () => { + const { execution, input, db } = fixture({ + request: async () => { + throw new OperationNotDispatchedError('UPSTREAM_REQUEST_INVALID'); + }, + }); + const prepared = await execution.prepare(input); + const job = await execution.execute( + { preparationId: prepared.id, confirmationToken: prepared.confirmationToken }, + 'a', + 'r', + ); + expect(job.status).toBe('failed'); + expect(job.items[0]?.state).toBe('failed'); + expect(db.prepare('SELECT result_code FROM job_items').get()).toEqual({ + result_code: 'UPSTREAM_NOT_DISPATCHED', + }); + }); +}); diff --git a/apps/api/src/application/operations/secure-operation-execution.ts b/apps/api/src/application/operations/secure-operation-execution.ts new file mode 100644 index 0000000..830b9ea --- /dev/null +++ b/apps/api/src/application/operations/secure-operation-execution.ts @@ -0,0 +1,466 @@ +import type Database from 'better-sqlite3'; +import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'; +import type { + ExecuteOperationRequest, + Job, + Preparation, + PrepareOperationRequest, +} from '@multi-simadmin/contracts'; +export interface SecureOperationDescriptor { + readonly operationId: string; + readonly title: string; + readonly riskLevel: string; + readonly method: string; + readonly pathTemplate: string; + readonly requestContentType: string; + readonly parameterSchemaId: string; +} +export interface SecureOperationRegistry { + requireExecutableOperation(operationId: string): SecureOperationDescriptor; +} +export const secureOperationRegistry: SecureOperationRegistry = { + requireExecutableOperation(operationId) { + if (operationId !== 'postNetworkRegisterAuto') throw new Error('UNKNOWN_OPERATION'); + return { + operationId: 'postNetworkRegisterAuto', + title: 'Register Network Automatically', + riskLevel: 'R2', + method: 'POST', + pathTemplate: '/api/network/register-auto', + requestContentType: 'none', + parameterSchemaId: 'simadmin.58e2204.postNetworkRegisterAuto.parameters.v1', + }; + }, +}; + +const ALLOWED_OPERATION = 'postNetworkRegisterAuto'; +const ACTOR = 'loopback-control-plane'; +const TTL_MS = 5 * 60 * 1000; +const EMPTY_DIGEST = createHash('sha256').update('').digest('hex'); + +export interface SafeOperationTransportRequest { + readonly origin: string; + readonly method: 'POST'; + readonly path: string; + readonly query: string; + readonly contentType: string; + readonly body: undefined; +} +export interface SafeOperationTransport { + request(request: SafeOperationTransportRequest): Promise<{ readonly status: number }>; +} +/** The request was rejected before any bytes could have reached the upstream. */ +export class OperationNotDispatchedError extends Error { + readonly dispatched = false; + constructor(readonly code: string) { + super(code); + this.name = 'OperationNotDispatchedError'; + } +} +export type SecureOperationExecutionErrorCode = + | 'VALIDATION_FAILED' + | 'OPERATION_NOT_ALLOWED' + | 'NOT_FOUND' + | 'REVISION_CONFLICT' + | 'CONFIRMATION_INVALID'; +export class SecureOperationExecutionError extends Error { + constructor( + readonly code: SecureOperationExecutionErrorCode, + message: string, + ) { + super(message); + this.name = 'SecureOperationExecutionError'; + } +} +interface Options { + readonly db: Database.Database; + readonly registry: SecureOperationRegistry; + readonly transport: SafeOperationTransport; + readonly now?: () => Date; + readonly idFactory?: () => string; + readonly tokenFactory?: () => string; + readonly nonceFactory?: () => string; +} +interface PreparationRow { + operation_id: string; + risk_level: string; + status: string; + target_instance_id: string; + target_revision: number; + target_origin: string; + method: string; + path: string; + canonical_query: string; + body_digest: string; + content_type: string; + parameter_schema_id: string; + parameters_digest: string; + nonce: string; + token_digest: string; + expires_at: string; +} +const digest = (value: string): string => createHash('sha256').update(value, 'utf8').digest('hex'); +const equalDigest = (left: string, right: string): boolean => { + const a = Buffer.from(left, 'hex'); + const b = Buffer.from(right, 'hex'); + return a.length === 32 && b.length === 32 && timingSafeEqual(a, b); +}; +const validRevision = (value: unknown): value is number => + typeof value === 'number' && Number.isSafeInteger(value) && value > 0; +const parameterDigest = (schema: string): string => + digest(JSON.stringify({ parameterSchemaId: schema, fields: [] })); + +export class SecureOperationExecution { + private readonly clock: () => Date; + private readonly id: () => string; + private readonly token: () => string; + private readonly nonce: () => string; + constructor(private readonly options: Options) { + this.clock = options.now ?? (() => new Date()); + this.id = options.idFactory ?? randomUUID; + this.token = options.tokenFactory ?? (() => randomBytes(32).toString('base64url')); + this.nonce = options.nonceFactory ?? randomUUID; + } + + async prepare(input: PrepareOperationRequest, requestId = this.id()): Promise { + if ( + !input || + typeof input !== 'object' || + Array.isArray(input) || + Object.keys(input).some((key) => !['operationId', 'targets', 'parameters'].includes(key)) || + typeof input.operationId !== 'string' + ) + this.validation(); + if (input.operationId !== ALLOWED_OPERATION) + throw new SecureOperationExecutionError( + 'OPERATION_NOT_ALLOWED', + 'Operation is not enabled for generic execution', + ); + let descriptor: SecureOperationDescriptor; + try { + descriptor = this.options.registry.requireExecutableOperation(input.operationId); + } catch { + throw new SecureOperationExecutionError( + 'OPERATION_NOT_ALLOWED', + 'Operation is not enabled for generic execution', + ); + } + if ( + descriptor.riskLevel !== 'R2' || + descriptor.method !== 'POST' || + descriptor.pathTemplate !== '/api/network/register-auto' || + descriptor.requestContentType !== 'none' || + !Array.isArray(input.targets) || + input.targets.length !== 1 + ) + this.validation(); + const target = input.targets[0]; + if ( + !target || + typeof target !== 'object' || + Array.isArray(target) || + Object.keys(target).some((key) => !['instanceId', 'revision'].includes(key)) || + typeof target.instanceId !== 'string' || + !target.instanceId || + !validRevision(target.revision) + ) + this.validation(); + const parameters = input.parameters; + if ( + !parameters || + typeof parameters !== 'object' || + Array.isArray(parameters) || + Object.keys(parameters).some((key) => !['parameterSchemaId', 'fields'].includes(key)) || + parameters.parameterSchemaId !== descriptor.parameterSchemaId || + !Array.isArray(parameters.fields) || + parameters.fields.length !== 0 + ) + this.validation(); + const instance = this.options.db + .prepare('SELECT base_url,config_revision FROM instances WHERE id=? AND enabled=1') + .get(target.instanceId) as { base_url: string; config_revision: number } | undefined; + if (!instance) throw new SecureOperationExecutionError('NOT_FOUND', 'Instance was not found'); + if (instance.config_revision !== target.revision) + throw new SecureOperationExecutionError( + 'REVISION_CONFLICT', + 'Instance revision does not match', + ); + const token = this.token(); + let tokenBytes: Buffer; + try { + tokenBytes = Buffer.from(token, 'base64url'); + } catch { + tokenBytes = Buffer.alloc(0); + } + if (tokenBytes.length < 32 || token.length < 20 || !/^[A-Za-z0-9_-]+$/.test(token)) + throw new Error('Confirmation token factory returned an unsafe token'); + const id = this.id(); + const created = this.clock(); + const now = created.toISOString(); + const expiresAt = new Date(created.getTime() + TTL_MS).toISOString(); + this.options.db + .prepare( + `INSERT INTO operation_preparations + (id,operation_id,risk_level,status,target_instance_id,target_revision,target_origin,method,path, + canonical_query,body_digest,content_type,parameter_schema_id,parameters_digest,nonce,token_digest, + requested_by,request_id,expires_at,created_at,updated_at) + VALUES (?,?,?,'prepared',?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + ) + .run( + id, + descriptor.operationId, + descriptor.riskLevel, + target.instanceId, + target.revision, + instance.base_url, + descriptor.method, + descriptor.pathTemplate, + '', + EMPTY_DIGEST, + '', + descriptor.parameterSchemaId, + parameterDigest(descriptor.parameterSchemaId), + this.nonce(), + digest(token), + ACTOR, + requestId, + expiresAt, + now, + now, + ); + return { + id, + status: 'prepared', + operationId: descriptor.operationId, + risk: 'R2', + expiresAt, + confirmationToken: token, + confirmationPrompt: `Execute ${descriptor.title} on ${target.instanceId}?`, + targetCount: 1, + }; + } + + async execute(input: ExecuteOperationRequest, actor: string, requestId: string): Promise { + if ( + !input || + typeof input !== 'object' || + Array.isArray(input) || + Object.keys(input).some((key) => !['preparationId', 'confirmationToken'].includes(key)) || + typeof input.preparationId !== 'string' || + !input.preparationId || + typeof input.confirmationToken !== 'string' || + !input.confirmationToken + ) + this.validation(); + const now = this.clock().toISOString(); + const ids = { job: this.id(), item: this.id(), attempt: this.id() }; + let bound: PreparationRow | undefined; + const outcome = this.options.db.transaction((): 'accepted' | 'invalid' => { + const row = this.options.db + .prepare('SELECT * FROM operation_preparations WHERE id=?') + .get(input.preparationId) as PreparationRow | undefined; + if (!row || row.status !== 'prepared') return 'invalid'; + if (row.expires_at <= now) { + this.options.db + .prepare( + "UPDATE operation_preparations SET status='expired',updated_at=? WHERE id=? AND status='prepared'", + ) + .run(now, input.preparationId); + return 'invalid'; + } + if (!equalDigest(row.token_digest, digest(input.confirmationToken))) return 'invalid'; + let descriptor: SecureOperationDescriptor; + try { + descriptor = this.options.registry.requireExecutableOperation(row.operation_id); + } catch { + return 'invalid'; + } + const instance = this.options.db + .prepare('SELECT base_url,config_revision FROM instances WHERE id=? AND enabled=1') + .get(row.target_instance_id) as { base_url: string; config_revision: number } | undefined; + const bindingValid = + descriptor.operationId === ALLOWED_OPERATION && + descriptor.riskLevel === 'R2' && + descriptor.method === row.method && + descriptor.pathTemplate === row.path && + descriptor.parameterSchemaId === row.parameter_schema_id && + row.canonical_query === '' && + row.content_type === '' && + row.body_digest === EMPTY_DIGEST && + row.parameters_digest === parameterDigest(row.parameter_schema_id) && + !!row.nonce && + !!instance && + instance.base_url === row.target_origin && + instance.config_revision === row.target_revision; + if (!bindingValid) { + this.options.db + .prepare( + "UPDATE operation_preparations SET status='invalidated',consumed_at=?,updated_at=? WHERE id=? AND status='prepared'", + ) + .run(now, now, input.preparationId); + return 'invalid'; + } + const consumed = this.options.db + .prepare( + "UPDATE operation_preparations SET status='consumed',consumed_at=?,updated_at=? WHERE id=? AND status='prepared'", + ) + .run(now, now, input.preparationId); + if (consumed.changes !== 1) return 'invalid'; + this.options.db + .prepare( + `INSERT INTO jobs + (id,parent_job_id,root_job_id,retry_of_job_id,operation_id,risk_level,status,requested_by,request_id,parameters_digest,created_at,started_at,updated_at) + VALUES (?,NULL,?,NULL,?,'R2','running',?,?,?,?,?,?)`, + ) + .run( + ids.job, + ids.job, + row.operation_id, + actor, + requestId, + row.parameters_digest, + now, + now, + now, + ); + this.options.db + .prepare( + "INSERT INTO job_attempts (id,job_id,status,started_at,created_at) VALUES (?,?,'running',?,?)", + ) + .run(ids.attempt, ids.job, now, now); + this.options.db + .prepare( + `INSERT INTO job_items (id,job_id,instance_id,attempt_number,status,created_at,started_at,updated_at) + VALUES (?,?,?,1,'running',?,?,?)`, + ) + .run(ids.item, ids.job, row.target_instance_id, now, now, now); + bound = row; + return 'accepted'; + })(); + if (outcome !== 'accepted' || !bound) + throw new SecureOperationExecutionError( + 'CONFIRMATION_INVALID', + 'Confirmation could not be accepted', + ); + let state: 'succeeded' | 'failed' | 'unknown-result'; + let code: string; + try { + const response = await this.options.transport.request({ + origin: bound.target_origin, + method: 'POST', + path: bound.path, + query: '', + contentType: '', + body: undefined, + }); + state = response.status >= 200 && response.status < 300 ? 'succeeded' : 'failed'; + code = state === 'succeeded' ? 'UPSTREAM_SUCCEEDED' : 'UPSTREAM_REJECTED'; + } catch (error) { + if (error instanceof OperationNotDispatchedError) { + state = 'failed'; + code = 'UPSTREAM_NOT_DISPATCHED'; + } else { + state = 'unknown-result'; + code = 'UPSTREAM_RESULT_UNKNOWN'; + } + } + this.finish(ids, state, code); + return this.job(ids.job); + } + + reconcileInterruptedJobs(): number { + const now = this.clock().toISOString(); + return this.options.db.transaction(() => { + const jobs = this.options.db + .prepare( + "SELECT id FROM jobs WHERE operation_id=? AND risk_level='R2' AND status='running'", + ) + .all(ALLOWED_OPERATION) as Array<{ id: string }>; + for (const row of jobs) this.finishByJob(row.id, now, 'unknown-result', 'INTERRUPTED'); + return jobs.length; + })(); + } + private finish( + ids: { job: string; item: string; attempt: string }, + state: string, + code: string, + ): void { + const now = this.clock().toISOString(); + this.options.db.transaction(() => { + this.options.db + .prepare( + "UPDATE job_items SET status=?,result_code=?,finished_at=?,updated_at=? WHERE id=? AND status='running'", + ) + .run(state, code, now, now, ids.item); + this.options.db + .prepare("UPDATE job_attempts SET status=?,finished_at=? WHERE id=? AND status='running'") + .run(state, now, ids.attempt); + this.options.db + .prepare( + "UPDATE jobs SET status=?,finished_at=?,updated_at=? WHERE id=? AND status='running'", + ) + .run(state, now, now, ids.job); + })(); + } + private finishByJob(jobId: string, now: string, state: string, code: string): void { + this.options.db + .prepare( + "UPDATE job_items SET status=?,result_code=?,finished_at=?,updated_at=? WHERE job_id=? AND status='running'", + ) + .run(state, code, now, now, jobId); + this.options.db + .prepare("UPDATE job_attempts SET status=?,finished_at=? WHERE job_id=? AND status='running'") + .run(state, now, jobId); + this.options.db + .prepare( + "UPDATE jobs SET status=?,finished_at=?,updated_at=? WHERE id=? AND status='running'", + ) + .run(state, now, now, jobId); + } + private job(id: string): Job { + const row = this.options.db + .prepare( + 'SELECT operation_id,status,root_job_id,retry_of_job_id,created_at FROM jobs WHERE id=?', + ) + .get(id) as { + operation_id: string; + status: Job['status']; + root_job_id: string; + retry_of_job_id: string | null; + created_at: string; + }; + const items = this.options.db + .prepare('SELECT id,instance_id,status FROM job_items WHERE job_id=?') + .all(id) as Array<{ + id: string; + instance_id: string; + status: 'succeeded' | 'failed' | 'unknown-result'; + }>; + const attempts = this.options.db + .prepare('SELECT id,status,started_at,finished_at FROM job_attempts WHERE job_id=?') + .all(id) as Array<{ + id: string; + status: 'succeeded' | 'failed' | 'unknown-result'; + started_at: string; + finished_at: string | null; + }>; + return { + id, + operationId: row.operation_id, + status: row.status, + rootJobId: row.root_job_id, + ...(row.retry_of_job_id ? { retryOfJobId: row.retry_of_job_id } : {}), + items: items.map((item) => ({ id: item.id, targetId: item.instance_id, state: item.status })), + attempts: attempts.map((attempt) => ({ + id: attempt.id, + state: attempt.status, + startedAt: attempt.started_at, + ...(attempt.finished_at ? { finishedAt: attempt.finished_at } : {}), + })), + createdAt: row.created_at, + }; + } + private validation(): never { + throw new SecureOperationExecutionError('VALIDATION_FAILED', 'Invalid operation request'); + } +} diff --git a/apps/api/src/application/status/status-poll-scheduler.test.ts b/apps/api/src/application/status/status-poll-scheduler.test.ts new file mode 100644 index 0000000..67e3ca9 --- /dev/null +++ b/apps/api/src/application/status/status-poll-scheduler.test.ts @@ -0,0 +1,249 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { StatusPollScheduler } from './status-poll-scheduler.js'; + +const deferred = () => { + let resolve!: () => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +}; + +const flush = async () => { + await Promise.resolve(); + await Promise.resolve(); +}; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('StatusPollScheduler', () => { + it('starts registered keys in FIFO order while respecting the global concurrency bound', async () => { + const runs: Array<{ id: string; done: ReturnType }> = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 1_000, + concurrency: 2, + task: (id) => { + const done = deferred(); + runs.push({ id, done }); + return done.promise; + }, + }); + scheduler.add('a'); + scheduler.add('b'); + scheduler.add('c'); + + scheduler.start(); + expect(runs.map(({ id }) => id)).toEqual(['a', 'b']); + runs[0]!.done.resolve(); + await flush(); + expect(runs.map(({ id }) => id)).toEqual(['a', 'b', 'c']); + + runs[1]!.done.resolve(); + runs[2]!.done.resolve(); + await scheduler.stop(); + }); + + it('maintains exactly one recurring timer chain when tasks resolve before the cadence', async () => { + vi.useFakeTimers(); + const task = vi.fn(async () => undefined); + const scheduler = new StatusPollScheduler({ intervalMs: 10, concurrency: 1, task }); + scheduler.add('a'); + scheduler.start(); + await flush(); + + expect(task).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(1); + + for (let expectedCalls = 2; expectedCalls <= 5; expectedCalls++) { + await vi.advanceTimersByTimeAsync(10); + expect(task).toHaveBeenCalledTimes(expectedCalls); + expect(vi.getTimerCount()).toBe(1); + } + + await scheduler.stop(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('is single-flight per key and coalesces arbitrarily many interval ticks into one rerun', async () => { + vi.useFakeTimers(); + const runs: ReturnType[] = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 10, + concurrency: 3, + task: async () => { + const done = deferred(); + runs.push(done); + await done.promise; + }, + }); + scheduler.add('a'); + scheduler.start(); + + await vi.advanceTimersByTimeAsync(100); + expect(runs).toHaveLength(1); + runs[0]!.resolve(); + await flush(); + expect(runs).toHaveLength(2); + + runs[1]!.resolve(); + await scheduler.stop(); + }); + + it('keeps equal-due recurring work FIFO instead of letting a busy key starve queued keys', async () => { + vi.useFakeTimers(); + const runs: Array<{ id: string; done: ReturnType }> = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 10, + concurrency: 1, + task: async (id) => { + const done = deferred(); + runs.push({ id, done }); + await done.promise; + }, + }); + scheduler.add('a'); + scheduler.add('b'); + scheduler.start(); + await vi.advanceTimersByTimeAsync(50); + runs[0]!.done.resolve(); + await flush(); + expect(runs.map(({ id }) => id)).toEqual(['a', 'b']); + + runs[1]!.done.resolve(); + await flush(); + expect(runs.map(({ id }) => id)).toEqual(['a', 'b', 'a']); + runs[2]!.done.resolve(); + await scheduler.stop(); + }); + + it('supports per-key interval updates and fences the obsolete generation', async () => { + vi.useFakeTimers(); + const calls: Array<{ signal: AbortSignal; done: ReturnType }> = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 100, + concurrency: 1, + task: async (_id, signal) => { + const done = deferred(); + calls.push({ signal, done }); + await done.promise; + }, + }); + scheduler.add('a'); + scheduler.start(); + scheduler.update('a', 10); + expect(calls[0]!.signal.aborted).toBe(true); + + await vi.advanceTimersByTimeAsync(50); + expect(calls).toHaveLength(1); + calls[0]!.done.resolve(); + await flush(); + expect(calls).toHaveLength(2); + + calls[1]!.done.resolve(); + await flush(); + await vi.advanceTimersByTimeAsync(9); + expect(calls).toHaveLength(2); + await vi.advanceTimersByTimeAsync(1); + expect(calls).toHaveLength(3); + calls[2]!.done.resolve(); + await scheduler.stop(); + }); + + it('remove cancels queued work and aborts active work; re-add is a fresh fenced generation', async () => { + const calls: Array<{ id: string; signal: AbortSignal; done: ReturnType }> = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 100, + concurrency: 1, + task: async (id, signal) => { + const done = deferred(); + calls.push({ id, signal, done }); + await done.promise; + }, + }); + scheduler.add('a'); + scheduler.add('b'); + scheduler.start(); + scheduler.remove('b'); + scheduler.remove('a'); + expect(calls[0]!.signal.aborted).toBe(true); + scheduler.add('a'); + + calls[0]!.done.resolve(); + await flush(); + expect(calls.map(({ id }) => id)).toEqual(['a', 'a']); + calls[1]!.done.resolve(); + await scheduler.stop(); + }); + + it('releases slots after rejection without an unhandled rejection', async () => { + const calls: string[] = []; + const unhandled = vi.fn(); + process.on('unhandledRejection', unhandled); + const scheduler = new StatusPollScheduler({ + intervalMs: 1_000, + concurrency: 1, + task: async (id) => { + calls.push(id); + if (id === 'a') throw new Error('probe failed'); + }, + }); + scheduler.add('a'); + scheduler.add('b'); + scheduler.start(); + await flush(); + expect(calls).toEqual(['a', 'b']); + expect(unhandled).not.toHaveBeenCalled(); + process.off('unhandledRejection', unhandled); + await scheduler.stop(); + }); + + it('start is idempotent and add/update reject invalid registration values', async () => { + const task = vi.fn(async () => undefined); + const scheduler = new StatusPollScheduler({ intervalMs: 20, concurrency: 1, task }); + expect(() => scheduler.add('')).toThrow(); + expect(() => scheduler.add('a', 0)).toThrow(); + expect(() => scheduler.update('missing', 10)).toThrow(); + scheduler.add('a'); + scheduler.start(); + scheduler.start(); + await flush(); + expect(task).toHaveBeenCalledTimes(1); + await scheduler.stop(); + }); + + it('stop is idempotent, clears timers, aborts active tasks, waits for settlement, and starts no queued work', async () => { + vi.useFakeTimers(); + const active = deferred(); + const calls: Array<{ id: string; signal: AbortSignal }> = []; + const scheduler = new StatusPollScheduler({ + intervalMs: 10, + concurrency: 1, + task: async (id, signal) => { + calls.push({ id, signal }); + await active.promise; + }, + }); + scheduler.add('a'); + scheduler.add('b'); + scheduler.start(); + const stopped = scheduler.stop(); + let settled = false; + void stopped.then(() => { + settled = true; + }); + await flush(); + expect(calls).toHaveLength(1); + expect(calls[0]!.signal.aborted).toBe(true); + expect(settled).toBe(false); + + active.resolve(); + await stopped; + await scheduler.stop(); + await vi.advanceTimersByTimeAsync(100); + expect(calls).toHaveLength(1); + }); +}); diff --git a/apps/api/src/application/status/status-poll-scheduler.ts b/apps/api/src/application/status/status-poll-scheduler.ts new file mode 100644 index 0000000..25f433f --- /dev/null +++ b/apps/api/src/application/status/status-poll-scheduler.ts @@ -0,0 +1,220 @@ +export type StatusPollTask = (instanceId: string, signal: AbortSignal) => Promise; + +export interface StatusPollSchedulerOptions { + task: StatusPollTask; + intervalMs: number; + concurrency: number; +} + +interface Registration { + generation: number; + intervalMs: number; + timer: ReturnType | undefined; +} + +interface QueueItem { + instanceId: string; + generation: number; +} + +interface ActiveRun { + generation: number; + controller: AbortController; + promise: Promise; + rerunGeneration?: number; +} + +/** + * A bounded, keyed polling scheduler. Registrations are the queue: repeated + * ticks only set one rerun bit, so backlog cannot grow with time. + */ +export class StatusPollScheduler { + readonly #task: StatusPollTask; + readonly #defaultIntervalMs: number; + readonly #concurrency: number; + readonly #registrations = new Map(); + readonly #queue: QueueItem[] = []; + readonly #queuedGeneration = new Map(); + readonly #active = new Map(); + #nextGeneration = 1; + #started = false; + #stopping = false; + #stopPromise?: Promise; + + constructor(options: StatusPollSchedulerOptions) { + this.#defaultIntervalMs = this.#validInterval(options.intervalMs); + if (!Number.isSafeInteger(options.concurrency) || options.concurrency < 1) { + throw new RangeError('concurrency must be a positive integer'); + } + this.#concurrency = options.concurrency; + this.#task = options.task; + } + + add(instanceId: string, intervalMs = this.#defaultIntervalMs): void { + this.#validId(instanceId); + const interval = this.#validInterval(intervalMs); + if (this.#registrations.has(instanceId)) { + throw new Error(`instance is already registered: ${instanceId}`); + } + if (this.#stopping) throw new Error('scheduler is stopping'); + + const registration: Registration = { + generation: this.#nextGeneration++, + intervalMs: interval, + timer: undefined, + }; + this.#registrations.set(instanceId, registration); + if (this.#started) this.#enqueue(instanceId, registration.generation); + } + + update(instanceId: string, intervalMs: number): void { + const interval = this.#validInterval(intervalMs); + const previous = this.#registrations.get(instanceId); + if (!previous) throw new Error(`instance is not registered: ${instanceId}`); + if (this.#stopping) throw new Error('scheduler is stopping'); + + if (previous.timer !== undefined) clearTimeout(previous.timer); + this.#cancelQueued(instanceId); + const active = this.#active.get(instanceId); + active?.controller.abort(); + + const registration: Registration = { + generation: this.#nextGeneration++, + intervalMs: interval, + timer: undefined, + }; + this.#registrations.set(instanceId, registration); + if (this.#started) this.#enqueue(instanceId, registration.generation); + } + + remove(instanceId: string): boolean { + const registration = this.#registrations.get(instanceId); + if (!registration) return false; + this.#registrations.delete(instanceId); + if (registration.timer !== undefined) clearTimeout(registration.timer); + this.#cancelQueued(instanceId); + this.#active.get(instanceId)?.controller.abort(); + return true; + } + + start(): void { + if (this.#started || this.#stopping) return; + this.#started = true; + for (const [instanceId, registration] of this.#registrations) { + this.#enqueue(instanceId, registration.generation); + } + } + + async stop(): Promise { + if (this.#stopPromise) return this.#stopPromise; + this.#stopping = true; + this.#started = false; + for (const registration of this.#registrations.values()) { + if (registration.timer !== undefined) clearTimeout(registration.timer); + registration.timer = undefined; + } + this.#queue.length = 0; + this.#queuedGeneration.clear(); + for (const run of this.#active.values()) run.controller.abort(); + + const settling = [...this.#active.values()].map(({ promise }) => promise); + this.#stopPromise = Promise.allSettled(settling).then(() => undefined); + return this.#stopPromise; + } + + #enqueue(instanceId: string, generation: number): void { + if (!this.#started || this.#stopping) return; + const registration = this.#registrations.get(instanceId); + if (!registration || registration.generation !== generation) return; + + const active = this.#active.get(instanceId); + if (active) { + active.rerunGeneration = generation; + return; + } + if (this.#queuedGeneration.get(instanceId) === generation) return; + + this.#cancelQueued(instanceId); + this.#queue.push({ instanceId, generation }); + this.#queuedGeneration.set(instanceId, generation); + this.#drain(); + } + + #drain(): void { + while (!this.#stopping && this.#active.size < this.#concurrency && this.#queue.length > 0) { + const item = this.#queue.shift()!; + if (this.#queuedGeneration.get(item.instanceId) === item.generation) { + this.#queuedGeneration.delete(item.instanceId); + } + const registration = this.#registrations.get(item.instanceId); + if (!registration || registration.generation !== item.generation) continue; + if (this.#active.has(item.instanceId)) { + this.#active.get(item.instanceId)!.rerunGeneration = item.generation; + continue; + } + this.#run(item, registration); + } + } + + #run(item: QueueItem, registration: Registration): void { + const controller = new AbortController(); + const active: ActiveRun = { + generation: item.generation, + controller, + promise: undefined as unknown as Promise, + }; + this.#active.set(item.instanceId, active); + this.#scheduleTick(item.instanceId, registration); + + let result: Promise; + try { + result = this.#task(item.instanceId, controller.signal); + } catch (error) { + result = Promise.reject(error); + } + const promise = Promise.resolve(result).then( + () => this.#finish(item.instanceId, active), + () => this.#finish(item.instanceId, active), + ); + active.promise = promise; + } + + #scheduleTick(instanceId: string, registration: Registration): void { + if (registration.timer !== undefined) return; + registration.timer = setTimeout(() => { + registration.timer = undefined; + if (this.#registrations.get(instanceId) !== registration || this.#stopping) return; + this.#scheduleTick(instanceId, registration); + this.#enqueue(instanceId, registration.generation); + }, registration.intervalMs); + } + + #finish(instanceId: string, completed: ActiveRun): void { + if (this.#active.get(instanceId) !== completed) return; + this.#active.delete(instanceId); + if (!this.#stopping) { + const current = this.#registrations.get(instanceId); + const rerun = completed.rerunGeneration; + if (current && rerun === current.generation) this.#enqueue(instanceId, rerun); + this.#drain(); + } + } + + #cancelQueued(instanceId: string): void { + this.#queuedGeneration.delete(instanceId); + for (let index = this.#queue.length - 1; index >= 0; index--) { + if (this.#queue[index]!.instanceId === instanceId) this.#queue.splice(index, 1); + } + } + + #validId(instanceId: string): void { + if (instanceId.length === 0) throw new TypeError('instanceId must not be empty'); + } + + #validInterval(intervalMs: number): number { + if (!Number.isFinite(intervalMs) || intervalMs <= 0) { + throw new RangeError('intervalMs must be greater than zero'); + } + return intervalMs; + } +} diff --git a/apps/api/src/control-plane.operations.test.ts b/apps/api/src/control-plane.operations.test.ts index 7316f3e..edfb1b3 100644 --- a/apps/api/src/control-plane.operations.test.ts +++ b/apps/api/src/control-plane.operations.test.ts @@ -19,6 +19,10 @@ it('registers the read-only operations catalog without touching upstream', async upstreamCalls += 1; throw new Error('unexpected'); }, + postNetworkRegisterAuto: async () => { + upstreamCalls += 1; + throw new Error('unexpected'); + }, }, }); const response = await app.inject('/api/v1/operations?pageSize=1'); diff --git a/apps/api/src/control-plane.test.ts b/apps/api/src/control-plane.test.ts index 6c85dba..31fb78c 100644 --- a/apps/api/src/control-plane.test.ts +++ b/apps/api/src/control-plane.test.ts @@ -86,6 +86,7 @@ describe('buildControlPlaneApp', () => { }); }, request: async () => ({ status: 200, headers: {}, body: '' }), + postNetworkRegisterAuto: async () => ({ status: 200 }), }, }); const created = await app.inject({ diff --git a/apps/api/src/control-plane.ts b/apps/api/src/control-plane.ts index bb49617..dadc983 100644 --- a/apps/api/src/control-plane.ts +++ b/apps/api/src/control-plane.ts @@ -1,6 +1,10 @@ import type Database from 'better-sqlite3'; import { registerOperationRoutes } from './interface/http/operation-routes.js'; import { operationCatalogRegistry } from './application/operations/operation-catalog-data.js'; +import { + SecureOperationExecution, + secureOperationRegistry, +} from './application/operations/secure-operation-execution.js'; import type { FastifyInstance } from 'fastify'; import { buildApp, type BuildAppOptions } from './app.js'; import { @@ -21,6 +25,7 @@ import { registerInstanceRoutes } from './interface/http/instance-routes.js'; export interface SafeControlPlaneUpstream extends ConnectionTransport { request: UpstreamSessionClientOptions['request']; + postNetworkRegisterAuto(origin: string): Promise<{ readonly status: number }>; } export interface ControlPlaneOptions { readonly db: Database.Database; @@ -54,11 +59,29 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane ? new DeleteInstanceOperation({ db: options.db, instances, now: options.now }) : new DeleteInstanceOperation({ db: options.db, instances }); deletion.reconcileInterruptedJobs(); + const secureExecution = new SecureOperationExecution({ + db: options.db, + registry: secureOperationRegistry, + transport: { + request: async ({ origin }) => { + const response = await options.upstream.postNetworkRegisterAuto(origin); + return { status: response.status }; + }, + }, + ...(options.now ? { now: options.now } : {}), + }); + secureExecution.reconcileInterruptedJobs(); const app = buildApp({ ...options.app, registerRoutes: (app) => { - registerInstanceRoutes(app, { instances, connections, login, deletion }); - registerOperationRoutes(app, operationCatalogRegistry); + registerInstanceRoutes(app, { + instances, + connections, + login, + deletion, + registerDeletionPreparationRoute: false, + }); + registerOperationRoutes(app, operationCatalogRegistry, secureExecution, deletion); }, }); Object.assign(app, { diff --git a/apps/api/src/infrastructure/database/migrations.ts b/apps/api/src/infrastructure/database/migrations.ts index fa47c9e..d4611a6 100644 --- a/apps/api/src/infrastructure/database/migrations.ts +++ b/apps/api/src/infrastructure/database/migrations.ts @@ -228,6 +228,48 @@ export const MIGRATIONS: readonly Migration[] = [ 'CREATE INDEX idx_operation_preparations_status_expires_at ON operation_preparations(status, expires_at)', ], }, + { + id: 5, + name: 'generic-secure-operation-preparation-binding', + statements: [ + `CREATE TABLE operation_preparations_v5 ( + id TEXT PRIMARY KEY, + operation_id TEXT NOT NULL, + risk_level TEXT NOT NULL CHECK (risk_level IN ('R2','R3')), + status TEXT NOT NULL CHECK (status IN ('prepared','consumed','expired','invalidated')), + target_instance_id TEXT NOT NULL, + target_revision INTEGER NOT NULL CHECK (target_revision > 0), + target_origin TEXT NOT NULL, + method TEXT NOT NULL, + path TEXT NOT NULL, + canonical_query TEXT NOT NULL, + body_digest TEXT NOT NULL CHECK (length(body_digest) = 64), + content_type TEXT NOT NULL, + parameter_schema_id TEXT NOT NULL, + parameters_digest TEXT NOT NULL CHECK (length(parameters_digest) = 64), + nonce TEXT NOT NULL UNIQUE, + token_digest TEXT NOT NULL CHECK (length(token_digest) = 64), + requested_by TEXT NOT NULL, + request_id TEXT NOT NULL, + expires_at TEXT NOT NULL, + consumed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `INSERT INTO operation_preparations_v5 + (id,operation_id,risk_level,status,target_instance_id,target_revision,target_origin,method,path, + canonical_query,body_digest,content_type,parameter_schema_id,parameters_digest,nonce,token_digest, + requested_by,request_id,expires_at,consumed_at,created_at,updated_at) + SELECT p.id,p.operation_id,p.risk_level,p.status,p.target_instance_id,p.target_revision, + COALESCE(i.base_url,''),'DELETE','/api/v1/instances/' || p.target_instance_id,'', + '${createHash('sha256').update('').digest('hex')}','',p.parameter_schema_id,p.parameters_digest, + 'legacy-' || p.id,p.token_digest,p.requested_by,p.request_id,p.expires_at,p.consumed_at,p.created_at,p.updated_at + FROM operation_preparations p LEFT JOIN instances i ON i.id=p.target_instance_id`, + 'DROP TABLE operation_preparations', + 'ALTER TABLE operation_preparations_v5 RENAME TO operation_preparations', + 'CREATE INDEX idx_operation_preparations_status_expires_at ON operation_preparations(status, expires_at)', + ], + }, ]; const createMigrationsTable = `CREATE TABLE schema_migrations ( diff --git a/apps/api/src/infrastructure/transport/safe-control-plane-upstream.test.ts b/apps/api/src/infrastructure/transport/safe-control-plane-upstream.test.ts index 377fa9e..3e347b6 100644 --- a/apps/api/src/infrastructure/transport/safe-control-plane-upstream.test.ts +++ b/apps/api/src/infrastructure/transport/safe-control-plane-upstream.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'; import { createSafeControlPlaneUpstream } from './safe-control-plane-upstream.js'; describe('createSafeControlPlaneUpstream', () => { - it('routes both health GET and login POST through the same safe transport', async () => { + it('routes health, login, and the audited R2 operation through the same safe transport', async () => { const calls: unknown[] = []; const upstream = createSafeControlPlaneUpstream({ get: async (url) => { @@ -22,6 +22,7 @@ describe('createSafeControlPlaneUpstream', () => { secret: '[REDACTED]', body: '[REDACTED]', }); + await upstream.postNetworkRegisterAuto('http://192.168.1.20'); expect(calls).toEqual([ { method: 'GET', url: 'http://192.168.1.20/api/health' }, { @@ -30,6 +31,12 @@ describe('createSafeControlPlaneUpstream', () => { headers: { 'content-type': 'application/json' }, body: '{"password":"[REDACTED]"}', }, + { + method: 'POST', + url: 'http://192.168.1.20/api/network/register-auto', + headers: {}, + body: '', + }, ]); }); }); diff --git a/apps/api/src/infrastructure/transport/safe-control-plane-upstream.ts b/apps/api/src/infrastructure/transport/safe-control-plane-upstream.ts index 77f1bd2..55184ac 100644 --- a/apps/api/src/infrastructure/transport/safe-control-plane-upstream.ts +++ b/apps/api/src/infrastructure/transport/safe-control-plane-upstream.ts @@ -4,6 +4,7 @@ import { SafeUpstreamGateway, type SafeUpstreamTransport } from './safe-upstream export interface SafeControlPlaneUpstream extends ConnectionTransport { request: UpstreamSessionClientOptions['request']; + postNetworkRegisterAuto(origin: string): Promise<{ readonly status: number }>; } export function createSafeControlPlaneUpstream( transport: SafeUpstreamTransport, @@ -12,5 +13,6 @@ export function createSafeControlPlaneUpstream( return { get: (url) => transport.get(url), request: (request) => gateway.request(request), + postNetworkRegisterAuto: (origin) => gateway.postNetworkRegisterAuto(origin), }; } diff --git a/apps/api/src/infrastructure/transport/safe-upstream-gateway.test.ts b/apps/api/src/infrastructure/transport/safe-upstream-gateway.test.ts index cf69ecb..cf1a83d 100644 --- a/apps/api/src/infrastructure/transport/safe-upstream-gateway.test.ts +++ b/apps/api/src/infrastructure/transport/safe-upstream-gateway.test.ts @@ -1,7 +1,35 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { SafeUpstreamGateway } from './safe-upstream-gateway.js'; describe('SafeUpstreamGateway', () => { + it('dispatches only the audited zero-body network registration operation through pinned POST', async () => { + const calls: unknown[] = []; + const gateway = new SafeUpstreamGateway({ + transport: { + get: async () => ({ status: 200, headers: {}, body: '' }), + post: async (url, headers, body) => { + calls.push({ url, headers, body }); + return { status: 204, headers: {}, body: '' }; + }, + }, + }); + const response = await gateway.postNetworkRegisterAuto('http://192.168.1.20:8080'); + expect(response.status).toBe(204); + expect(calls).toEqual([ + { url: 'http://192.168.1.20:8080/api/network/register-auto', headers: {}, body: '' }, + ]); + }); + + it('rejects malformed operation origins before calling transport', async () => { + const post = vi.fn(async () => ({ status: 204, headers: {}, body: '' })); + const gateway = new SafeUpstreamGateway({ + transport: { get: async () => ({ status: 200, headers: {}, body: '' }), post }, + }); + await expect( + gateway.postNetworkRegisterAuto('http://192.168.1.20/base?next=x'), + ).rejects.toMatchObject({ code: 'UPSTREAM_REQUEST_INVALID', dispatched: false }); + expect(post).not.toHaveBeenCalled(); + }); it('sends login password only as JSON through the pinned POST transport', async () => { const calls: unknown[] = []; const gateway = new SafeUpstreamGateway({ diff --git a/apps/api/src/infrastructure/transport/safe-upstream-gateway.ts b/apps/api/src/infrastructure/transport/safe-upstream-gateway.ts index 4ea95df..1c14bf8 100644 --- a/apps/api/src/infrastructure/transport/safe-upstream-gateway.ts +++ b/apps/api/src/infrastructure/transport/safe-upstream-gateway.ts @@ -4,6 +4,7 @@ import type { } from '../../application/connections/upstream-session-client.js'; import type { TransportResponse } from './safe-instance-transport.js'; import { UpstreamError } from './upstream-error.js'; +import { OperationNotDispatchedError } from '../../application/operations/secure-operation-execution.js'; export interface SafeUpstreamTransport { get(url: string): Promise; @@ -15,6 +16,34 @@ export interface SafeUpstreamTransport { } export class SafeUpstreamGateway { constructor(private readonly options: { readonly transport: SafeUpstreamTransport }) {} + async postNetworkRegisterAuto(origin: string): Promise { + let parsed: URL; + try { + parsed = new URL(origin); + } catch { + throw this.notDispatched(); + } + if ( + (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') || + parsed.username || + parsed.password || + parsed.pathname !== '/' || + parsed.search || + parsed.hash + ) + throw this.notDispatched(); + const url = `${parsed.origin}/api/network/register-auto`; + try { + return await this.options.transport.post(url, {}, ''); + } catch (error) { + if ( + error instanceof UpstreamError && + (error.code === 'UNSAFE_ORIGIN' || error.code === 'UNSAFE_RESOLUTION') + ) + throw new OperationNotDispatchedError(error.code); + throw error; + } + } async request(request: UpstreamRequest): Promise { const url = new URL(request.url); if (url.protocol !== 'https:') throw new UpstreamError('UPSTREAM_INSECURE_AUTH'); @@ -35,4 +64,7 @@ export class SafeUpstreamGateway { } throw new UpstreamError('UPSTREAM_REQUEST_INVALID'); } + private notDispatched(): OperationNotDispatchedError { + return new OperationNotDispatchedError('UPSTREAM_REQUEST_INVALID'); + } } diff --git a/apps/api/src/interface/http/instance-routes.ts b/apps/api/src/interface/http/instance-routes.ts index 9343003..ae0dfa0 100644 --- a/apps/api/src/interface/http/instance-routes.ts +++ b/apps/api/src/interface/http/instance-routes.ts @@ -25,6 +25,7 @@ export interface InstanceRoutesOptions { readonly connections?: ConnectionProbe; readonly login?: InstanceLoginService; readonly deletion?: DeleteInstanceOperation; + readonly registerDeletionPreparationRoute?: boolean; } const problem = ( request: FastifyRequest, @@ -379,10 +380,12 @@ export function registerInstanceRoutes(app: FastifyInstance, options: InstanceRo } if (options.deletion) { - app.post( - '/api/v1/operations/prepare', - wrap(async (request) => options.deletion!.prepare(request.body as never, request.id)), - ); + if (options.registerDeletionPreparationRoute !== false) { + app.post( + '/api/v1/operations/prepare', + wrap(async (request) => options.deletion!.prepare(request.body as never, request.id)), + ); + } app.delete('/api/v1/instances/:instanceId', async (request, reply) => { const handled = await wrap(async (wrappedRequest) => { const preparationId = wrappedRequest.headers['x-preparation-id']; diff --git a/apps/api/src/interface/http/operation-routes.ts b/apps/api/src/interface/http/operation-routes.ts index 8ba5505..9b985d3 100644 --- a/apps/api/src/interface/http/operation-routes.ts +++ b/apps/api/src/interface/http/operation-routes.ts @@ -1,7 +1,19 @@ -import type { OperationPageQuery } from '@multi-simadmin/contracts'; -import type { FastifyInstance, FastifyRequest } from 'fastify'; +import type { + ExecuteOperationRequest, + OperationPageQuery, + PrepareOperationRequest, +} from '@multi-simadmin/contracts'; +import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { OperationCatalogService } from '../../application/operations/operation-catalog-service.js'; import type { OperationCatalogRegistry } from '../../application/operations/operation-catalog-service.js'; +import { + DeleteInstanceOperation, + DeleteInstanceOperationError, +} from '../../application/operations/delete-instance-operation.js'; +import { + SecureOperationExecution, + SecureOperationExecutionError, +} from '../../application/operations/secure-operation-execution.js'; const capabilities = new Set(['query', 'command', 'job']); const keys = new Set([ @@ -89,6 +101,8 @@ function parseQuery(request: FastifyRequest): OperationPageQuery { export function registerOperationRoutes( app: FastifyInstance, registry: OperationCatalogRegistry, + execution?: SecureOperationExecution, + deletion?: DeleteInstanceOperation, ): void { const service = new OperationCatalogService(registry); app.get('/api/v1/operations', async (request, reply) => { @@ -106,4 +120,67 @@ export function registerOperationRoutes( }); } }); + if (!execution) return; + const problem = ( + request: FastifyRequest, + reply: FastifyReply, + error: SecureOperationExecutionError, + ) => { + const status = + error.code === 'NOT_FOUND' ? 404 : error.code === 'REVISION_CONFLICT' ? 409 : 400; + const title = status === 404 ? 'Not Found' : status === 409 ? 'Conflict' : 'Bad Request'; + return reply.code(status).type('application/problem+json').send({ + type: 'about:blank', + title, + status, + code: error.code, + detail: error.message, + requestId: request.id, + }); + }; + app.post('/api/v1/operations/prepare', async (request, reply) => { + try { + const body = request.body as { operationId?: unknown }; + if (body && typeof body === 'object' && body.operationId === 'deleteInstance' && deletion) + return reply.code(200).send(await deletion.prepare(request.body as never, request.id)); + return reply + .code(200) + .send(await execution.prepare(request.body as PrepareOperationRequest, request.id)); + } catch (error) { + if (error instanceof DeleteInstanceOperationError) { + const status = + error.code === 'NOT_FOUND' ? 404 : error.code === 'REVISION_CONFLICT' ? 412 : 400; + return reply + .code(status) + .type('application/problem+json') + .send({ + type: 'about:blank', + title: + status === 404 ? 'Not Found' : status === 412 ? 'Precondition Failed' : 'Bad Request', + status, + code: error.code, + detail: 'The requested operation could not be prepared.', + requestId: request.id, + }); + } + if (error instanceof SecureOperationExecutionError) return problem(request, reply, error); + throw error; + } + }); + app.post('/api/v1/operations/execute', async (request, reply) => { + try { + return reply + .code(202) + .send( + await execution.execute( + request.body as ExecuteOperationRequest, + 'loopback-control-plane', + request.id, + ), + ); + } catch (error) { + if (error instanceof SecureOperationExecutionError) return problem(request, reply, error); + throw error; + } + }); } diff --git a/apps/api/src/interface/http/secure-operation-routes.test.ts b/apps/api/src/interface/http/secure-operation-routes.test.ts new file mode 100644 index 0000000..9715196 --- /dev/null +++ b/apps/api/src/interface/http/secure-operation-routes.test.ts @@ -0,0 +1,121 @@ +import Database from 'better-sqlite3'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { buildApp } from '../../app.js'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { + SecureOperationExecution, + secureOperationRegistry, +} from '../../application/operations/secure-operation-execution.js'; +import { operationCatalogRegistry } from '../../application/operations/operation-catalog-data.js'; +import { registerOperationRoutes } from './operation-routes.js'; + +const resources: Array<{ close(): unknown }> = []; +afterEach(async () => { + for (const resource of resources.splice(0)) await resource.close(); +}); +function fixture() { + const db = new Database(':memory:'); + resources.push(db); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + db.prepare( + `INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) + VALUES ('i','one','http://192.168.1.10','none',1,1,'2026-07-17T00:00:00.000Z','2026-07-17T00:00:00.000Z')`, + ).run(); + const request = vi.fn(async () => ({ status: 200 })); + const execution = new SecureOperationExecution({ + db, + registry: secureOperationRegistry, + transport: { request }, + idFactory: (() => { + let n = 0; + return () => `id-${++n}`; + })(), + tokenFactory: () => Buffer.alloc(32, 4).toString('base64url'), + }); + const app = buildApp({ + registerRoutes: (server) => + registerOperationRoutes(server, operationCatalogRegistry, execution), + }); + resources.push(app); + return { app, db, request }; +} +const prepareBody = { + operationId: 'postNetworkRegisterAuto', + targets: [{ instanceId: 'i', revision: 1 }], + parameters: { + parameterSchemaId: 'simadmin.58e2204.postNetworkRegisterAuto.parameters.v1', + fields: [], + }, +}; + +describe('secure operation routes', () => { + it('returns 200 Preparation then 202 terminal Job with request IDs', async () => { + const { app, db, request } = fixture(); + const prepared = await app.inject({ + method: 'POST', + url: '/api/v1/operations/prepare', + payload: prepareBody, + }); + expect(prepared.statusCode).toBe(200); + expect(prepared.headers['x-request-id']).toBeTruthy(); + const preparation = prepared.json(); + expect(db.prepare('SELECT count(*) count FROM jobs').get()).toEqual({ count: 0 }); + const executed = await app.inject({ + method: 'POST', + url: '/api/v1/operations/execute', + payload: { + preparationId: preparation.id, + confirmationToken: preparation.confirmationToken, + }, + }); + expect(executed.statusCode).toBe(202); + expect(executed.headers['x-request-id']).toBeTruthy(); + expect(executed.json()).toMatchObject({ + operationId: 'postNetworkRegisterAuto', + status: 'succeeded', + }); + expect(request).toHaveBeenCalledOnce(); + }); + + it.each([ + ['/api/v1/operations/prepare', { ...prepareBody, extra: true }], + [ + '/api/v1/operations/prepare', + { ...prepareBody, targets: [{ instanceId: 'i', revision: 1, extra: true }] }, + ], + [ + '/api/v1/operations/prepare', + { ...prepareBody, parameters: { ...prepareBody.parameters, extra: true } }, + ], + ['/api/v1/operations/execute', { preparationId: 'x', confirmationToken: 'x', extra: true }], + ])('strictly rejects additional properties at %s', async (url, payload) => { + const { app } = fixture(); + const response = await app.inject({ method: 'POST', url, payload }); + expect(response.statusCode).toBe(400); + expect(response.headers['content-type']).toContain('application/problem+json'); + expect(response.headers['x-request-id']).toBeTruthy(); + expect(response.json()).toMatchObject({ + status: 400, + code: 'VALIDATION_FAILED', + requestId: response.headers['x-request-id'], + }); + }); + + it('returns Problem Details and does not consume a wrong confirmation', async () => { + const { app, db } = fixture(); + const preparation = ( + await app.inject({ method: 'POST', url: '/api/v1/operations/prepare', payload: prepareBody }) + ).json(); + const response = await app.inject({ + method: 'POST', + url: '/api/v1/operations/execute', + payload: { preparationId: preparation.id, confirmationToken: 'wrong' }, + }); + expect(response.statusCode).toBe(400); + expect(response.json()).toMatchObject({ status: 400, code: 'CONFIRMATION_INVALID' }); + expect( + db.prepare('SELECT status FROM operation_preparations WHERE id=?').get(preparation.id), + ).toEqual({ status: 'prepared' }); + }); +});