import type Database from 'better-sqlite3'; import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'; import type { EventEnvelope } from '../events/event-journal.js'; 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; } const EXECUTABLE_OPERATIONS: Readonly> = Object.freeze({ postNetworkRegisterAuto: Object.freeze({ operationId: 'postNetworkRegisterAuto', title: 'Register Network Automatically', riskLevel: 'R2', method: 'POST', pathTemplate: '/api/network/register-auto', requestContentType: 'none', parameterSchemaId: 'simadmin.58e2204.postNetworkRegisterAuto.parameters.v1', }), postServiceRestart: Object.freeze({ operationId: 'postServiceRestart', title: 'Restart Service', riskLevel: 'R3', method: 'POST', pathTemplate: '/api/service/restart', requestContentType: 'none', parameterSchemaId: 'simadmin.58e2204.postServiceRestart.parameters.v1', }), postBasebandRestart: Object.freeze({ operationId: 'postBasebandRestart', title: 'Restart Baseband', riskLevel: 'R3', method: 'POST', pathTemplate: '/api/baseband/restart', requestContentType: 'none', parameterSchemaId: 'simadmin.58e2204.postBasebandRestart.parameters.v1', }), postSystemReboot: Object.freeze({ operationId: 'postSystemReboot', title: 'Reboot System', riskLevel: 'R3', method: 'POST', pathTemplate: '/api/system/reboot', requestContentType: 'application/json', parameterSchemaId: 'simadmin.58e2204.postSystemReboot.parameters.v1', }), }); export const secureOperationRegistry: SecureOperationRegistry = { requireExecutableOperation(operationId) { const descriptor = EXECUTABLE_OPERATIONS[operationId]; if (!descriptor) throw new Error('UNKNOWN_OPERATION'); return descriptor; }, }; const ZERO_BODY_OPERATIONS = new Set([ 'postNetworkRegisterAuto', 'postServiceRestart', 'postBasebandRestart', ]); const SYSTEM_REBOOT_OPERATION = 'postSystemReboot'; const SYSTEM_REBOOT_DELAY_SECONDS = 3; 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; /** Serialized JSON body for audited JSON operations; undefined for zero-body POSTs. */ readonly body: string | undefined; /** Bound target used only for origin-scoped session attachment; never logged. */ readonly instanceId: string; } 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; /** Receives job terminal transitions so the SSE journal has a live producer. */ readonly emit?: (envelope: EventEnvelope) => void; } 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, fields: readonly unknown[] = []): 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(); 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.operationId !== input.operationId || (descriptor.riskLevel !== 'R2' && descriptor.riskLevel !== 'R3') || descriptor.method !== 'POST' || !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) ) this.validation(); let bodyDigest = EMPTY_DIGEST; let contentType = ''; let serializedBody: string | undefined; if (ZERO_BODY_OPERATIONS.has(descriptor.operationId)) { if (descriptor.requestContentType !== 'none' || parameters.fields.length !== 0) this.validation(); } else if (descriptor.operationId === SYSTEM_REBOOT_OPERATION) { if (descriptor.requestContentType !== 'application/json' || parameters.fields.length !== 1) this.validation(); const field = parameters.fields[0] as { fieldId?: unknown; kind?: unknown; value?: unknown }; if ( !field || typeof field !== 'object' || Array.isArray(field) || Object.keys(field).some((key) => !['fieldId', 'kind', 'value'].includes(key)) || field.fieldId !== 'delay_seconds' || field.kind !== 'number' || field.value !== SYSTEM_REBOOT_DELAY_SECONDS ) this.validation(); serializedBody = JSON.stringify({ delay_seconds: SYSTEM_REBOOT_DELAY_SECONDS }); bodyDigest = digest(serializedBody); contentType = 'application/json'; } else { 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, '', bodyDigest, contentType, descriptor.parameterSchemaId, parameterDigest(descriptor.parameterSchemaId, parameters.fields), this.nonce(), digest(token), ACTOR, requestId, expiresAt, now, now, ); return { id, status: 'prepared', operationId: descriptor.operationId, risk: descriptor.riskLevel as 'R2' | 'R3', 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 expectedBody = descriptor.operationId === SYSTEM_REBOOT_OPERATION ? JSON.stringify({ delay_seconds: SYSTEM_REBOOT_DELAY_SECONDS }) : undefined; const expectedBodyDigest = expectedBody ? digest(expectedBody) : EMPTY_DIGEST; const expectedContentType = descriptor.operationId === SYSTEM_REBOOT_OPERATION ? 'application/json' : ''; const expectedFields = descriptor.operationId === SYSTEM_REBOOT_OPERATION ? [{ fieldId: 'delay_seconds', kind: 'number', value: SYSTEM_REBOOT_DELAY_SECONDS }] : []; const bindingValid = !!EXECUTABLE_OPERATIONS[descriptor.operationId] && (descriptor.riskLevel === 'R2' || descriptor.riskLevel === 'R3') && descriptor.riskLevel === row.risk_level && descriptor.method === row.method && descriptor.pathTemplate === row.path && descriptor.parameterSchemaId === row.parameter_schema_id && row.canonical_query === '' && row.content_type === expectedContentType && row.body_digest === expectedBodyDigest && row.parameters_digest === parameterDigest(row.parameter_schema_id, expectedFields) && !!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,?,?,'running',?,?,?,?,?,?)`, ) .run( ids.job, ids.job, row.operation_id, row.risk_level, 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: bound.content_type, body: bound.operation_id === SYSTEM_REBOOT_OPERATION ? JSON.stringify({ delay_seconds: SYSTEM_REBOOT_DELAY_SECONDS }) : undefined, instanceId: bound.target_instance_id, }); 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); this.#emitTerminal(ids.job, state, requestId); 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 IN ('postNetworkRegisterAuto','postServiceRestart','postBasebandRestart','postSystemReboot') AND risk_level IN ('R2','R3') AND status='running'", ) .all() as Array<{ id: string }>; for (const row of jobs) { this.finishByJob(row.id, now, 'unknown-result', 'INTERRUPTED'); this.#emitTerminal(row.id, 'unknown-result', now); } 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); })(); } #emitTerminal(jobId: string, state: string, requestId: string): void { if (!this.options.emit) return; try { this.options.emit({ kind: 'job', id: this.id(), occurredAt: this.clock().toISOString(), requestId, jobId, }); } catch { // A journal failure must never fail the operation itself. } } 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'); } }