feat(api): advance capability and secure operations slices
This commit is contained in:
chick
@@ -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> = {},
|
||||
): CapabilityOperationDescriptor => ({
|
||||
operationId: 'getHealth',
|
||||
method: 'GET',
|
||||
pathTemplate: '/api/health',
|
||||
riskLevel: 'R0',
|
||||
...overrides,
|
||||
});
|
||||
|
||||
function fixture(
|
||||
options: {
|
||||
descriptor?: CapabilityOperationDescriptor;
|
||||
response?: {
|
||||
status: number;
|
||||
headers: Readonly<Record<string, string | undefined>>;
|
||||
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<string, string>;
|
||||
body: string;
|
||||
}) => void;
|
||||
const olderResponse = new Promise<{
|
||||
status: number;
|
||||
headers: Record<string, string>;
|
||||
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<Partial<CapabilityObservationError>>({ 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<Partial<CapabilityObservationError>>({ code: 'OPERATION_NOT_FOUND' }),
|
||||
);
|
||||
expect(transport.get).not.toHaveBeenCalled();
|
||||
expect(row(db)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -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<CapabilityInstance | undefined> };
|
||||
readonly registry: { requireOperation(operationId: string): CapabilityOperationDescriptor };
|
||||
readonly transport: { get(url: string): Promise<TransportResponse> };
|
||||
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<string, unknown>;
|
||||
const data = root.data;
|
||||
const candidate =
|
||||
root.version ??
|
||||
(data !== null && typeof data === 'object' && !Array.isArray(data)
|
||||
? (data as Record<string, unknown>).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<string, number>();
|
||||
#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<CapabilityObservation> {
|
||||
// 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 });
|
||||
}
|
||||
}
|
||||
@@ -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' });
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<unknown>, 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<string, unknown>;
|
||||
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',
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<Preparation> {
|
||||
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<Job> {
|
||||
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');
|
||||
}
|
||||
}
|
||||
@@ -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<void>((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<typeof deferred> }> = [];
|
||||
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<typeof deferred>[] = [];
|
||||
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<typeof deferred> }> = [];
|
||||
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<typeof deferred> }> = [];
|
||||
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<typeof deferred> }> = [];
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,220 @@
|
||||
export type StatusPollTask = (instanceId: string, signal: AbortSignal) => Promise<void>;
|
||||
|
||||
export interface StatusPollSchedulerOptions {
|
||||
task: StatusPollTask;
|
||||
intervalMs: number;
|
||||
concurrency: number;
|
||||
}
|
||||
|
||||
interface Registration {
|
||||
generation: number;
|
||||
intervalMs: number;
|
||||
timer: ReturnType<typeof setTimeout> | undefined;
|
||||
}
|
||||
|
||||
interface QueueItem {
|
||||
instanceId: string;
|
||||
generation: number;
|
||||
}
|
||||
|
||||
interface ActiveRun {
|
||||
generation: number;
|
||||
controller: AbortController;
|
||||
promise: Promise<void>;
|
||||
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<string, Registration>();
|
||||
readonly #queue: QueueItem[] = [];
|
||||
readonly #queuedGeneration = new Map<string, number>();
|
||||
readonly #active = new Map<string, ActiveRun>();
|
||||
#nextGeneration = 1;
|
||||
#started = false;
|
||||
#stopping = false;
|
||||
#stopPromise?: Promise<void>;
|
||||
|
||||
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<void> {
|
||||
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<void>,
|
||||
};
|
||||
this.#active.set(item.instanceId, active);
|
||||
this.#scheduleTick(item.instanceId, registration);
|
||||
|
||||
let result: Promise<void>;
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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');
|
||||
|
||||
@@ -86,6 +86,7 @@ describe('buildControlPlaneApp', () => {
|
||||
});
|
||||
},
|
||||
request: async () => ({ status: 200, headers: {}, body: '' }),
|
||||
postNetworkRegisterAuto: async () => ({ status: 200 }),
|
||||
},
|
||||
});
|
||||
const created = await app.inject({
|
||||
|
||||
@@ -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, {
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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: '',
|
||||
},
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<TransportResponse>;
|
||||
@@ -15,6 +16,34 @@ export interface SafeUpstreamTransport {
|
||||
}
|
||||
export class SafeUpstreamGateway {
|
||||
constructor(private readonly options: { readonly transport: SafeUpstreamTransport }) {}
|
||||
async postNetworkRegisterAuto(origin: string): Promise<UpstreamResponse> {
|
||||
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<UpstreamResponse> {
|
||||
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');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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'];
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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' });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user