diff --git a/apps/api/src/application/identity/device-identity-service.test.ts b/apps/api/src/application/identity/device-identity-service.test.ts new file mode 100644 index 0000000..8805402 --- /dev/null +++ b/apps/api/src/application/identity/device-identity-service.test.ts @@ -0,0 +1,245 @@ +import Database from 'better-sqlite3'; +import { describe, expect, it } from 'vitest'; + +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { DeviceIdentityError, DeviceIdentityService } from './device-identity-service.js'; + +interface Fixture { + readonly db: Database.Database; + readonly service: DeviceIdentityService; + readonly advance: (minutes: number) => void; +} + +function fixture(): Fixture { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + let clock = new Date('2026-09-05T02:00:00.000Z'); + const service = new DeviceIdentityService({ db, now: () => clock }); + return { + db, + service, + advance: (minutes: number) => { + clock = new Date(clock.getTime() + minutes * 60_000); + }, + }; +} + +function manualFixture() { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + let clock = new Date('2026-09-05T02:00:00.000Z'); + const service = new DeviceIdentityService({ + db, + now: () => clock, + authorizationMode: () => 'manual', + }); + return { + db, + service, + advance: (minutes: number) => { + clock = new Date(clock.getTime() + minutes * 60_000); + }, + }; +} + +function register(db: Database.Database, id: string): void { + db.prepare( + `INSERT INTO instances + (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) + VALUES (?,?,?,'password',1,1,?,?)`, + ).run(id, id, `http://${id}.lan:8080`, '2026-09-05T02:00:00.000Z', '2026-09-05T02:00:00.000Z'); +} + +const device = (overrides: Record = {}) => ({ + imei: '860000000000001', + manufacturer: 'Quectel', + model: 'RM500Q', + revision: 'RM500QEAAAR13', + ...overrides, +}); + +describe('DeviceIdentityService', () => { + it('baselines the first report instead of calling it a conflict', () => { + const { db, service } = fixture(); + register(db, 'a'); + + const identity = service.observe('a', { ...device(), origin: 'http://a.lan:8080' }); + + expect(identity).toMatchObject({ + instanceId: 'a', + status: 'confirmed', + reasons: [], + imei: '860000000000001', + model: 'RM500Q', + origin: 'http://a.lan:8080', + }); + expect(service.isBlocked('a')).toBe(false); + }); + + it('keeps a device that reports no identity out of the guard', () => { + const { db, service } = fixture(); + register(db, 'a'); + + expect(service.observe('a', { agent: 'simadmin-agent 1.9.6' })).toBeUndefined(); + expect(service.observe('a', { imei: ' ' })).toBeUndefined(); + expect(service.list()).toEqual([]); + }); + + it('holds a first report until an operator confirms it in manual authorization mode', () => { + const { db, service } = manualFixture(); + register(db, 'a'); + + const identity = service.observe('a', device()); + + expect(identity).toMatchObject({ + instanceId: 'a', + status: 'pending', + reasons: [], + imei: '860000000000001', + }); + expect(service.isBlocked('a')).toBe(true); + expect(service.confirm('a')).toMatchObject({ status: 'confirmed', reasons: [] }); + expect(service.isBlocked('a')).toBe(false); + }); + + it('holds a node whose hardware no longer matches the record', () => { + const { db, service } = fixture(); + register(db, 'a'); + service.observe('a', device()); + + const swapped = service.observe('a', device({ model: 'RG500Q', imei: '860000000000009' })); + + expect(swapped?.status).toBe('pending'); + expect(swapped?.reasons).toEqual(['hardware_swapped']); + expect(swapped?.changes).toEqual([ + { field: 'imei', from: '860000000000001', to: '860000000000009' }, + { field: 'model', from: 'RM500Q', to: 'RG500Q' }, + ]); + expect(service.isBlocked('a')).toBe(true); + }); + + it('compares only the fields a partial report actually carries', () => { + const { db, service } = fixture(); + register(db, 'a'); + service.observe('a', device()); + + const partial = service.observe('a', { model: 'RM500Q' }); + + expect(partial?.status).toBe('confirmed'); + expect(partial?.reasons).toEqual([]); + // The dropped fields stay on the row: a silent read must not erase what we already knew. + expect(partial?.imei).toBe('860000000000001'); + }); + + it('asks both records when one identity is claimed twice', () => { + const { db, service } = fixture(); + register(db, 'a'); + register(db, 'b'); + service.observe('a', device()); + + const newcomer = service.observe('b', device()); + + expect(newcomer?.status).toBe('pending'); + expect(newcomer?.reasons).toEqual(['imei_claimed']); + expect(newcomer?.conflicts).toEqual([{ reason: 'imei_claimed', instanceIds: ['a'] }]); + expect(service.get('a')?.status).toBe('pending'); + expect(service.get('a')?.conflicts).toEqual([{ reason: 'imei_claimed', instanceIds: ['b'] }]); + expect(service.summarize()).toEqual({ tracked: 2, pending: ['a', 'b'] }); + }); + + it('accepts the twins that were on screen when the operator confirmed', () => { + const { db, service } = fixture(); + register(db, 'a'); + register(db, 'b'); + service.observe('a', device()); + service.observe('b', device()); + + expect(service.confirm('a')).toMatchObject({ status: 'confirmed', reasons: [] }); + service.observe('a', device()); + expect(service.get('a')?.status).toBe('confirmed'); + + // A third claimant is new information, so the guard comes back down. + register(db, 'c'); + service.observe('c', device()); + expect(service.get('a')?.status).toBe('pending'); + // 'b' was on screen when the operator confirmed; only the newcomer is still a question. + expect(service.get('a')?.conflicts).toEqual([{ reason: 'imei_claimed', instanceIds: ['c'] }]); + }); + + it('re-baselines swapped hardware on confirmation', () => { + const { db, service } = fixture(); + register(db, 'a'); + service.observe('a', device()); + service.observe('a', device({ model: 'RG500Q' })); + + const confirmed = service.confirm('a'); + + expect(confirmed).toMatchObject({ status: 'confirmed', reasons: [], changes: [] }); + expect(service.isBlocked('a')).toBe(false); + service.observe('a', device({ model: 'RG500Q' })); + expect(service.get('a')?.status).toBe('confirmed'); + service.observe('a', device({ model: 'EM120' })); + expect(service.get('a')?.reasons).toEqual(['hardware_swapped']); + }); + + it('sanitizes device-controlled evidence before it is stored', () => { + const { db, service } = fixture(); + register(db, 'a'); + + const identity = service.observe('a', { + imei: ' 860000000000001\u0000\u001b ', + model: 'x'.repeat(200), + revision: 42, + manufacturer: null, + }); + + expect(identity?.imei).toBe('860000000000001'); + expect(identity?.model).toHaveLength(64); + expect(identity?.revision).toBe('42'); + expect(identity?.manufacturer).toBe(''); + }); + + it('reports a missing identity rather than inventing one on confirm', () => { + const { db, service } = fixture(); + register(db, 'a'); + + expect(() => service.confirm('a')).toThrow(DeviceIdentityError); + try { + service.confirm('a'); + } catch (error) { + expect(error).toBeInstanceOf(DeviceIdentityError); + expect((error as DeviceIdentityError).code).toBe('NOT_FOUND'); + } + }); + + it('forgets the identity with the node it belonged to', () => { + const { db, service } = fixture(); + register(db, 'a'); + register(db, 'b'); + service.observe('a', device()); + service.observe('b', device()); + + db.prepare('DELETE FROM instances WHERE id=?').run('b'); + + // The row goes with the node, and the survivor is released the next time anything reads it. + expect(service.list().map((identity) => identity.instanceId)).toEqual(['a']); + expect(service.get('a')).toMatchObject({ status: 'pending' }); + expect(service.observe('a', device())).toMatchObject({ status: 'confirmed', reasons: [] }); + }); + + it('advances the observation time without touching the confirmation time', () => { + const { db, service, advance } = fixture(); + register(db, 'a'); + const first = service.observe('a', device()); + expect(first?.observedAt).toBe('2026-09-05T02:00:00.000Z'); + + advance(5); + const second = service.observe('a', device()); + + expect(second?.observedAt).toBe('2026-09-05T02:05:00.000Z'); + expect(second?.confirmedAt).toBe('2026-09-05T02:00:00.000Z'); + expect(second?.fingerprint).toBe(first?.fingerprint); + }); +}); diff --git a/apps/api/src/application/identity/device-identity-service.ts b/apps/api/src/application/identity/device-identity-service.ts new file mode 100644 index 0000000..b66e338 --- /dev/null +++ b/apps/api/src/application/identity/device-identity-service.ts @@ -0,0 +1,461 @@ +import { createHash } from 'node:crypto'; + +import type Database from 'better-sqlite3'; + +export type IdentityStatus = 'confirmed' | 'pending'; + +/** + * `hardware_swapped` is the record that used to answer at this address reporting different + * hardware now; `imei_claimed` is a second record answering with the same identity. Neither can + * be resolved from here, which is why both stop control traffic until an operator says which + * device is the real one. + */ +export type IdentityReason = 'hardware_swapped' | 'imei_claimed'; + +const IDENTITY_FIELDS = ['imei', 'manufacturer', 'model', 'revision'] as const; +type IdentityField = (typeof IDENTITY_FIELDS)[number]; +type IdentityValues = Partial>; + +const MAX_FIELD_LENGTH: Readonly> = { + imei: 32, + manufacturer: 64, + model: 64, + revision: 64, + agent: 64, + origin: 256, +}; + +const CONTROL_CHARACTERS = /[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/gu; + +export interface DeviceIdentityEvidence { + readonly imei?: unknown; + readonly manufacturer?: unknown; + readonly model?: unknown; + readonly revision?: unknown; + readonly agent?: unknown; + readonly origin?: unknown; +} + +export interface DeviceIdentityConflict { + readonly reason: IdentityReason; + readonly instanceIds: readonly string[]; +} + +export interface DeviceIdentityFieldChange { + readonly field: IdentityField; + readonly from: string; + readonly to: string; +} + +export interface DeviceIdentity { + readonly instanceId: string; + readonly status: IdentityStatus; + readonly reasons: readonly IdentityReason[]; + readonly conflicts: readonly DeviceIdentityConflict[]; + readonly imei: string; + readonly manufacturer: string; + readonly model: string; + readonly revision: string; + readonly agent: string; + readonly origin: string; + readonly fingerprint: string; + readonly confirmedFingerprint: string; + readonly changes: readonly DeviceIdentityFieldChange[]; + readonly observedAt: string; + readonly confirmedAt: string; +} + +export type DeviceIdentityErrorCode = 'NOT_FOUND' | 'VALIDATION_FAILED'; + +export class DeviceIdentityError extends Error { + constructor( + readonly code: DeviceIdentityErrorCode, + message: string, + ) { + super(message); + this.name = 'DeviceIdentityError'; + } +} + +export interface DeviceIdentityOptions { + readonly db: Database.Database; + readonly now?: () => Date; + /** Follows the connection setting: manual authorization makes a first report wait. */ + readonly authorizationMode?: () => 'auto' | 'manual'; +} + +interface IdentityRow { + instance_id: string; + status: string; + reasons: string; + imei: string; + manufacturer: string; + model: string; + revision: string; + agent: string; + origin: string; + fingerprint: string; + confirmed_fingerprint: string; + confirmed_values_json: string; + confirmed_peers_json: string; + detail_json: string; + observed_at: string; + confirmed_at: string; +} + +/** Device payloads are untrusted: only a trimmed, control-free, length-capped string is kept. */ +function text(value: unknown, limit: number): string { + if (typeof value !== 'string' && typeof value !== 'number') return ''; + const raw = typeof value === 'number' ? String(value) : value; + return raw.replace(CONTROL_CHARACTERS, '').trim().slice(0, limit); +} + +function identityValues(evidence: DeviceIdentityEvidence): IdentityValues { + const values: IdentityValues = {}; + for (const field of IDENTITY_FIELDS) { + const value = text(evidence[field], MAX_FIELD_LENGTH[field]); + if (value) values[field] = value; + } + return values; +} + +function digest(values: IdentityValues): string { + const canonical = IDENTITY_FIELDS.filter((field) => values[field] !== undefined) + .map((field) => `${field}=${values[field]}`) + .join('\n'); + return createHash('sha256').update(canonical, 'utf8').digest('hex').slice(0, 16); +} + +function stringArray(value: unknown): string[] { + return Array.isArray(value) + ? value.filter((item): item is string => typeof item === 'string') + : []; +} + +function parseValues(value: string): IdentityValues { + try { + const parsed: unknown = JSON.parse(value); + if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) return {}; + const record = parsed as Record; + const values: IdentityValues = {}; + for (const field of IDENTITY_FIELDS) { + const entry = text(record[field], MAX_FIELD_LENGTH[field]); + if (entry) values[field] = entry; + } + return values; + } catch { + return {}; + } +} + +/** The evidence the console shows: who else claims this identity and what changed. */ +interface IdentityDetail { + readonly peers: readonly string[]; + readonly changes: readonly DeviceIdentityFieldChange[]; +} + +function parseDetail(value: string): IdentityDetail { + let parsed: unknown = {}; + try { + parsed = JSON.parse(value); + } catch { + return { peers: [], changes: [] }; + } + if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) + return { peers: [], changes: [] }; + const record = parsed as Record; + const entries = Array.isArray(record.changes) ? record.changes : []; + const changes = entries.filter( + (entry): entry is DeviceIdentityFieldChange => + entry !== null && + typeof entry === 'object' && + !Array.isArray(entry) && + IDENTITY_FIELDS.includes((entry as Record).field as IdentityField) && + typeof (entry as Record).from === 'string' && + typeof (entry as Record).to === 'string', + ); + return { peers: stringArray(record.peers), changes }; +} + +function parseReasons(value: string): IdentityReason[] { + const allowed: IdentityReason[] = ['hardware_swapped', 'imei_claimed']; + try { + const parsed: unknown = JSON.parse(value); + if (!Array.isArray(parsed)) return []; + return parsed.filter((item): item is IdentityReason => + allowed.includes(item as IdentityReason), + ); + } catch { + return []; + } +} + +/** + * Tracks the hardware behind each registered node and holds the control plane still when that + * hardware stops matching the record. Observations come from device reads the console is making + * anyway, so a node only gains an identity row once something has actually looked at it. + */ +export class DeviceIdentityService { + readonly #db: Database.Database; + readonly #clock: () => Date; + readonly #authorizationMode: () => 'auto' | 'manual'; + readonly #select: Database.Statement; + readonly #selectPeers: Database.Statement; + readonly #insert: Database.Statement; + readonly #update: Database.Statement; + + constructor(options: DeviceIdentityOptions) { + this.#db = options.db; + this.#clock = options.now ?? (() => new Date()); + this.#authorizationMode = options.authorizationMode ?? (() => 'auto'); + this.#select = options.db.prepare('SELECT * FROM device_identities WHERE instance_id=?'); + this.#selectPeers = options.db.prepare( + "SELECT instance_id FROM device_identities WHERE imei=? AND instance_id<>? AND imei<>'' ORDER BY instance_id", + ); + this.#insert = options.db.prepare( + `INSERT INTO device_identities (instance_id,status,reasons,imei,manufacturer,model,revision, + agent,origin,fingerprint,confirmed_fingerprint,confirmed_values_json,confirmed_peers_json, + detail_json,observed_at,confirmed_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + ); + this.#update = options.db.prepare( + `UPDATE device_identities SET status=?,reasons=?,imei=?,manufacturer=?,model=?,revision=?, + agent=?,origin=?,fingerprint=?,confirmed_fingerprint=?,confirmed_values_json=?, + confirmed_peers_json=?,detail_json=?,observed_at=?,confirmed_at=?,updated_at=? + WHERE instance_id=?`, + ); + } + + /** + * Records what a device just said about itself. Returns undefined when the payload carried no + * identity at all, which keeps a device that does not report its IMEI out of the guard rather + * than inventing a fingerprint from nothing. + */ + observe(instanceId: string, evidence: DeviceIdentityEvidence): DeviceIdentity | undefined { + if (typeof instanceId !== 'string' || !instanceId) return undefined; + const values = identityValues(evidence); + const agent = text(evidence.agent, MAX_FIELD_LENGTH.agent); + const origin = text(evidence.origin, MAX_FIELD_LENGTH.origin); + if (Object.keys(values).length === 0) return undefined; + const now = this.#clock().toISOString(); + const fingerprint = digest(values); + const manualFirstReport = this.#authorizationMode() === 'manual'; + this.#db.transaction(() => { + const existing = this.#select.get(instanceId) as IdentityRow | undefined; + if (!existing) { + this.#insert.run( + instanceId, + manualFirstReport ? 'pending' : 'confirmed', + '[]', + values.imei ?? '', + values.manufacturer ?? '', + values.model ?? '', + values.revision ?? '', + agent, + origin, + fingerprint, + fingerprint, + JSON.stringify(values), + '[]', + '{"peers":[]}', + now, + now, + now, + now, + ); + // A first report is a baseline, but it can still walk into a twin that is already + // registered, so the new row goes through the same evaluation as every other read. + const inserted = this.#select.get(instanceId) as IdentityRow; + this.#write(inserted, values, agent, origin, fingerprint, now, now, manualFirstReport); + } else { + this.#write(existing, values, agent, origin, fingerprint, now, now); + } + this.#syncClaims(instanceId, values.imei ?? '', now); + })(); + return this.get(instanceId); + } + + list(): readonly DeviceIdentity[] { + const rows = this.#db + .prepare('SELECT * FROM device_identities ORDER BY status DESC, instance_id') + .all() as IdentityRow[]; + return rows.map((row) => this.#present(row)); + } + + get(instanceId: string): DeviceIdentity | undefined { + const row = this.#select.get(instanceId) as IdentityRow | undefined; + return row ? this.#present(row) : undefined; + } + + /** Control operations only; reading a device is always allowed, unbinding is the way out. */ + isBlocked(instanceId: string): boolean { + const row = this.#select.get(instanceId) as IdentityRow | undefined; + return row?.status === 'pending'; + } + + summarize(): { readonly tracked: number; readonly pending: readonly string[] } { + const rows = this.#db + .prepare('SELECT instance_id,status FROM device_identities ORDER BY instance_id') + .all() as Array<{ instance_id: string; status: string }>; + return { + tracked: rows.length, + pending: rows.filter((row) => row.status === 'pending').map((row) => row.instance_id), + }; + } + + /** + * The operator says "this is the device I mean". The current report becomes the baseline and + * the other claimants that were visible at that moment are accepted, so a known twin does not + * reopen the question on the next read while a third one still does. + */ + confirm(instanceId: string): DeviceIdentity { + const row = this.#select.get(instanceId) as IdentityRow | undefined; + if (!row) throw new DeviceIdentityError('NOT_FOUND', 'Device identity was not found'); + const now = this.#clock().toISOString(); + const values = identityValues({ + imei: row.imei, + manufacturer: row.manufacturer, + model: row.model, + revision: row.revision, + }); + const fingerprint = row.fingerprint || digest(values); + this.#db.transaction(() => { + this.#update.run( + 'confirmed', + '[]', + row.imei, + row.manufacturer, + row.model, + row.revision, + row.agent, + row.origin, + fingerprint, + fingerprint, + JSON.stringify(values), + JSON.stringify(this.#peers(instanceId, row.imei)), + '{"peers":[]}', + row.observed_at, + now, + now, + instanceId, + ); + this.#syncClaims(instanceId, row.imei, now); + })(); + const updated = this.#select.get(instanceId) as IdentityRow | undefined; + if (!updated) throw new DeviceIdentityError('NOT_FOUND', 'Device identity was not found'); + return this.#present(updated); + } + + #peers(instanceId: string, imei: string): string[] { + if (!imei) return []; + return (this.#selectPeers.all(imei, instanceId) as Array<{ instance_id: string }>).map( + (row) => row.instance_id, + ); + } + + /** Re-evaluated on every read, so a conflict that walks away stops holding the node. */ + #write( + existing: IdentityRow, + values: IdentityValues, + agent: string, + origin: string, + fingerprint: string, + observedAt: string, + updatedAt: string, + forcePending = false, + ): void { + const confirmed = parseValues(existing.confirmed_values_json); + const accepted = parseStringArray(existing.confirmed_peers_json); + const changes: DeviceIdentityFieldChange[] = []; + for (const field of IDENTITY_FIELDS) { + const before = confirmed[field]; + const after = values[field]; + if (before !== undefined && after !== undefined && before !== after) + changes.push({ field, from: before, to: after }); + } + const peers = this.#peers(existing.instance_id, values.imei ?? ''); + const unaccepted = peers.filter((peer) => !accepted.includes(peer)); + const reasons: IdentityReason[] = []; + if (changes.length > 0) reasons.push('hardware_swapped'); + if (unaccepted.length > 0) reasons.push('imei_claimed'); + this.#update.run( + reasons.length > 0 || forcePending ? 'pending' : 'confirmed', + JSON.stringify(reasons), + values.imei ?? existing.imei, + values.manufacturer ?? existing.manufacturer, + values.model ?? existing.model, + values.revision ?? existing.revision, + agent || existing.agent, + origin || existing.origin, + fingerprint, + existing.confirmed_fingerprint, + existing.confirmed_values_json, + existing.confirmed_peers_json, + JSON.stringify({ peers: unaccepted, changes }), + observedAt, + existing.confirmed_at, + updatedAt, + existing.instance_id, + ); + } + + /** + * A twin is told about itself: both records wait for the operator, not just the newcomer. Each + * peer is re-read from its own row, so a claim that walks away releases the peer it was holding + * and a peer that also drifted keeps its hardware_swapped reason. + */ + #syncClaims(instanceId: string, imei: string, now: string): void { + if (!imei) return; + for (const { instance_id: peerId } of this.#selectPeers.all(imei, instanceId) as Array<{ + instance_id: string; + }>) { + const peer = this.#select.get(peerId) as IdentityRow | undefined; + if (!peer) continue; + this.#write( + peer, + identityValues(peer), + peer.agent, + peer.origin, + peer.fingerprint || digest(identityValues(peer)), + peer.observed_at, + now, + ); + } + } + + #present(row: IdentityRow): DeviceIdentity { + const reasons = parseReasons(row.reasons); + const detail = parseDetail(row.detail_json); + const conflicts: DeviceIdentityConflict[] = []; + if (reasons.includes('hardware_swapped')) + conflicts.push({ reason: 'hardware_swapped', instanceIds: [] }); + if (reasons.includes('imei_claimed')) + conflicts.push({ reason: 'imei_claimed', instanceIds: detail.peers }); + return { + instanceId: row.instance_id, + status: row.status === 'pending' ? 'pending' : 'confirmed', + reasons, + conflicts, + imei: row.imei, + manufacturer: row.manufacturer, + model: row.model, + revision: row.revision, + agent: row.agent, + origin: row.origin, + fingerprint: row.fingerprint, + confirmedFingerprint: row.confirmed_fingerprint, + changes: detail.changes, + observedAt: row.observed_at, + confirmedAt: row.confirmed_at, + }; + } +} + +function parseStringArray(value: string): string[] { + try { + const parsed: unknown = JSON.parse(value); + return Array.isArray(parsed) ? stringArray(parsed) : []; + } catch { + return []; + } +} diff --git a/apps/api/src/application/instances/device-action-service.test.ts b/apps/api/src/application/instances/device-action-service.test.ts index 54cc4e9..0b8bcf1 100644 --- a/apps/api/src/application/instances/device-action-service.test.ts +++ b/apps/api/src/application/instances/device-action-service.test.ts @@ -18,6 +18,7 @@ function fixture( replies: readonly Reply[] = [{ status: 200, body: { status: 'success' } }], extra: { readonly ensureSession?: (id: string, origin: string, force?: boolean) => Promise; + readonly identities?: { isBlocked(instanceId: string): boolean }; } = {}, ) { const db = new Database(':memory:'); @@ -51,6 +52,7 @@ function fixture( }; }, ...(extra.ensureSession ? { ensureSession: extra.ensureSession } : {}), + ...(extra.identities ? { identities: extra.identities } : {}), now: () => new Date('2026-09-04T00:00:00.000Z'), id: () => 'audit-row-1', }); @@ -197,4 +199,23 @@ describe('DeviceActionService', () => { expect(JSON.stringify(summary)).not.toContain('office'); expect(summary.every((item) => item.redacted === true)).toBe(true); }); + + it('holds every control while the device identity is disputed', async () => { + const { service, calls } = fixture(undefined, { + identities: { isBlocked: () => true }, + }); + await expect( + service.execute('node-a', 'cell-lock.unlock-all', {}, context), + ).rejects.toMatchObject({ code: 'IDENTITY_UNCONFIRMED' }); + expect(calls).toHaveLength(0); + }); + + it('still lets an operator release the binding on a disputed device', async () => { + const { service, calls } = fixture(undefined, { + identities: { isBlocked: () => true }, + }); + const result = await service.execute('node-a', 'hub.unbind', {}, context); + expect(result.ok).toBe(true); + expect(calls).toHaveLength(1); + }); }); diff --git a/apps/api/src/application/instances/device-action-service.ts b/apps/api/src/application/instances/device-action-service.ts index 12d8fe2..2f20d84 100644 --- a/apps/api/src/application/instances/device-action-service.ts +++ b/apps/api/src/application/instances/device-action-service.ts @@ -20,7 +20,8 @@ export type DeviceActionErrorCode = | 'VALIDATION_FAILED' | 'SESSION_INVALID' | 'UPSTREAM_FAILED' - | 'NOT_DISPATCHED'; + | 'NOT_DISPATCHED' + | 'IDENTITY_UNCONFIRMED'; export class DeviceActionError extends Error { constructor( @@ -192,6 +193,12 @@ export interface DeviceActionServiceOptions { readonly ensureSession?: (instanceId: string, origin: string, force?: boolean) => Promise; readonly now?: () => Date; readonly id?: () => string; + /** + * The identity guard. A node whose reported hardware stopped matching its record, or whose + * identity is claimed by another record, is not safe to control: the command would land on a + * device the operator is not looking at. Unbinding stays available because it is the way out. + */ + readonly identities?: { isBlocked(instanceId: string): boolean }; } /** @@ -239,6 +246,8 @@ export class DeviceActionService { const session = this.options.sessions.sessionFor(instanceId); if (session && session.origin !== instance.origin) throw new DeviceActionError('SESSION_INVALID'); + if (action.id !== 'hub.unbind' && this.options.identities?.isBlocked(instanceId)) + throw new DeviceActionError('IDENTITY_UNCONFIRMED'); const values = this.#validate(action, params); const { path, query, body } = this.#assemble(action, values); diff --git a/apps/api/src/application/observability/metrics-service.test.ts b/apps/api/src/application/observability/metrics-service.test.ts new file mode 100644 index 0000000..9967b75 --- /dev/null +++ b/apps/api/src/application/observability/metrics-service.test.ts @@ -0,0 +1,138 @@ +import Database from 'better-sqlite3'; +import { afterEach, describe, expect, it } from 'vitest'; + +import { DeviceIdentityService } from '../identity/device-identity-service.js'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { MetricsService } from './metrics-service.js'; + +let db: Database.Database | undefined; + +const NOW = '2026-09-05T02:00:00.000Z'; + +function fixture() { + db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + const identities = new DeviceIdentityService({ db }); + const metrics = new MetricsService({ + db, + version: '1.9.6', + sources: { + connections: { + summarize: () => [ + { + instanceId: 'node-a', + total: 40, + success: 38, + failed: 2, + averageDurationMs: 12, + lastObservedAt: NOW, + lastErrorCode: null, + availabilityPercent: 95, + }, + ], + }, + identities, + }, + now: () => new Date(NOW), + }); + return { db, identities, metrics }; +} + +function seedNode(database: Database.Database, id: string, name: string, enabled = 1): void { + database + .prepare( + `INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) + VALUES (?,?,'http://' || ? || ':8080','password',?,1,?,?)`, + ) + .run(id, name, id, enabled, NOW, NOW); +} + +afterEach(() => { + db?.close(); + db = undefined; +}); + +describe('MetricsService', () => { + it('renders the fleet, queue and job gauges with every state present', () => { + const { db: database, metrics } = fixture(); + seedNode(database, 'node-a', '机房 A'); + seedNode(database, 'node-b', '机房 B', 0); + database + .prepare( + `INSERT INTO sms_outbox (id,instance_id,phone_number,content,status,available_at,created_at,updated_at) + VALUES ('s1','node-a','+8613800000000','hi','failed',?,?,?)`, + ) + .run(NOW, NOW, NOW); + database + .prepare( + `INSERT INTO notification_queue (id,event_type,status,payload_json,available_at,created_at,updated_at) + VALUES ('q1','sms','pending','{}',?,?,?)`, + ) + .run(NOW, NOW, NOW); + database + .prepare( + `INSERT INTO jobs (id,operation_id,risk_level,status,requested_by,request_id,parameters_digest,created_at,updated_at) + VALUES ('j1','node.reboot','R2','running','op','r1','digest',?,?)`, + ) + .run(NOW, NOW); + + const text = metrics.render(); + expect(text).toContain('multi_simadmin_nodes_total 2'); + expect(text).toContain('multi_simadmin_node_enabled{node="node-b"} 0'); + expect(text).toContain('multi_simadmin_node_info{node="node-a",name="机房 A"} 1'); + expect(text).toContain('multi_simadmin_node_probe_total{node="node-a",outcome="success"} 38'); + expect(text).toContain('multi_simadmin_node_availability_ratio{node="node-a"} 0.950'); + expect(text).toContain('multi_simadmin_sms_outbox{status="failed"} 1'); + expect(text).toContain('multi_simadmin_sms_outbox{status="queued"} 0'); + expect(text).toContain('multi_simadmin_notification_queue{status="pending"} 1'); + expect(text).toContain('multi_simadmin_jobs{status="running"} 1'); + expect(text).toContain('multi_simadmin_jobs{status="succeeded"} 0'); + expect(text).toContain('multi_simadmin_scheduled_tasks{status="armed"} 0'); + expect(text).toContain('multi_simadmin_build_info{version="1.9.6"} 1'); + // A node with no probe history is counted rather than silently missing. + expect(text).toContain('multi_simadmin_node_unprobed 1'); + }); + + it('reports the identity guard for each node it has seen', () => { + const { db: database, identities, metrics } = fixture(); + seedNode(database, 'node-a', 'A'); + seedNode(database, 'node-b', 'B'); + const report = { imei: '860000000000009', model: 'RM500Q' }; + identities.observe('node-a', report); + identities.observe('node-b', report); + + const text = metrics.render(); + expect(text).toContain('multi_simadmin_node_identity_pending{node="node-a"} 1'); + expect(text).toContain('multi_simadmin_node_identity_pending{node="node-b"} 1'); + expect(text).toContain('multi_simadmin_identity_pending_total 2'); + }); + + it('escapes label values so a node name cannot forge a series', () => { + const { db: database, metrics } = fixture(); + seedNode(database, 'node-a', 'evil" name\nwith break'); + + const text = metrics.render(); + expect(text).toContain('name="evil\\" name\\nwith break"'); + // The injected newline must not become its own sample line. + expect(text.split('\n').filter((line) => line.includes('with break'))).toHaveLength(1); + }); + + it('emits a valid exposition: help and type precede every series', () => { + const { db: database, metrics } = fixture(); + seedNode(database, 'node-a', 'A'); + const lines = metrics.render().trimEnd().split('\n'); + const names = new Set(); + for (const line of lines) { + if (line.startsWith('# ')) { + const [, kind, name] = line.split(' '); + if (kind === 'TYPE') expect(names.has(name ?? ''), name).toBe(false); + continue; + } + const name = line.slice(0, line.search(/[{ ]/u)); + expect(line, name).toMatch(/^[a-z_][a-z0-9_]*(\{[^{}]*\})? [-0-9.e+]+$/u); + names.add(name); + } + expect(names.size).toBeGreaterThan(10); + }); +}); diff --git a/apps/api/src/application/observability/metrics-service.ts b/apps/api/src/application/observability/metrics-service.ts new file mode 100644 index 0000000..24442bf --- /dev/null +++ b/apps/api/src/application/observability/metrics-service.ts @@ -0,0 +1,244 @@ +import type Database from 'better-sqlite3'; + +import type { ConnectionSummary } from '../system/connection-log-service.js'; +import type { DeviceIdentity } from '../identity/device-identity-service.js'; + +/** + * Prometheus exposition for the fused control plane. + * + * The scrape must never reach a device: every value here comes from rows the console already + * writes (instances, probes, queues, jobs, identity records). A monitoring system polling every + * 15 seconds should not be able to wake a modem. + */ + +const PREFIX = 'multi_simadmin'; + +export interface MetricsSource { + readonly connections: { summarize(): readonly ConnectionSummary[] }; + readonly identities: { list(): readonly DeviceIdentity[] }; +} + +export interface MetricsOptions { + readonly db: Database.Database; + readonly version: string; + readonly sources: MetricsSource; + readonly now?: () => Date; +} + +interface CountRow { + readonly status: string; + readonly count: number; +} + +/** Label values are operator- or device-controlled, so they are escaped the Prometheus way. */ +function label(value: string): string { + return value.replace(/\\/gu, '\\\\').replace(/"/gu, '\\"').replace(/\n/gu, '\\n'); +} + +const number = (value: unknown): number => { + const parsed = typeof value === 'number' ? value : Number(value); + return Number.isFinite(parsed) ? parsed : 0; +}; + +function sample(name: string, help: string, type: 'gauge' | 'counter', lines: readonly string[]) { + return [`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`, ...lines]; +} + +export class MetricsService { + readonly #db: Database.Database; + readonly #version: string; + readonly #sources: MetricsSource; + readonly #clock: () => Date; + + constructor(options: MetricsOptions) { + this.#db = options.db; + this.#version = options.version; + this.#sources = options.sources; + this.#clock = options.now ?? (() => new Date()); + } + + render(): string { + const nodes = this.#db + .prepare('SELECT id,name,enabled FROM instances ORDER BY name, id') + .all() as Array<{ + id: string; + name: string; + enabled: number; + }>; + const ids = new Set(nodes.map((node) => node.id)); + const lines: string[] = []; + + lines.push( + ...sample(`${PREFIX}_build_info`, 'Build identity of the control plane.', 'gauge', [ + `${PREFIX}_build_info{version="${label(this.#version)}"} 1`, + ]), + ...sample(`${PREFIX}_scrape_timestamp_seconds`, 'Wall-clock time of this scrape.', 'gauge', [ + `${PREFIX}_scrape_timestamp_seconds ${Math.floor(this.#clock().getTime() / 1000)}`, + ]), + ); + + const info = nodes.map( + (node) => `${PREFIX}_node_info{node="${label(node.id)}",name="${label(node.name)}"} 1`, + ); + lines.push( + ...sample(`${PREFIX}_node_info`, 'Registered nodes and their addresses.', 'gauge', info), + ...sample(`${PREFIX}_nodes_total`, 'Number of registered nodes.', 'gauge', [ + `${PREFIX}_nodes_total ${nodes.length}`, + ]), + ...sample( + `${PREFIX}_node_enabled`, + 'Whether the node is enabled (1) or paused (0).', + 'gauge', + nodes.map( + (node) => `${PREFIX}_node_enabled{node="${label(node.id)}"} ${node.enabled ? 1 : 0}`, + ), + ), + ); + + const probes = this.#sources.connections.summarize(); + lines.push( + ...sample( + `${PREFIX}_node_probe_total`, + 'Health probes recorded in the retained window, by outcome.', + 'counter', + probes.flatMap((probe) => [ + `${PREFIX}_node_probe_total{node="${label(probe.instanceId)}",outcome="success"} ${probe.success}`, + `${PREFIX}_node_probe_total{node="${label(probe.instanceId)}",outcome="failed"} ${probe.failed}`, + ]), + ), + ...sample( + `${PREFIX}_node_availability_ratio`, + 'Share of successful probes in the retained window.', + 'gauge', + probes.map( + (probe) => + `${PREFIX}_node_availability_ratio{node="${label(probe.instanceId)}"} ${(probe.availabilityPercent / 100).toFixed(3)}`, + ), + ), + ...sample( + `${PREFIX}_node_probe_duration_ms`, + 'Average probe latency in the retained window.', + 'gauge', + probes.map( + (probe) => + `${PREFIX}_node_probe_duration_ms{node="${label(probe.instanceId)}"} ${number(probe.averageDurationMs)}`, + ), + ), + ); + + const identities = this.#sources.identities.list(); + const pending = identities.filter((identity) => identity.status === 'pending'); + lines.push( + ...sample( + `${PREFIX}_node_identity_tracked`, + 'Whether the node has ever reported its hardware identity.', + 'gauge', + identities.map( + (identity) => `${PREFIX}_node_identity_tracked{node="${label(identity.instanceId)}"} 1`, + ), + ), + ...sample( + `${PREFIX}_node_identity_pending`, + '1 while the identity guard holds the node and control actions are refused.', + 'gauge', + identities.map( + (identity) => + `${PREFIX}_node_identity_pending{node="${label(identity.instanceId)}"} ${identity.status === 'pending' ? 1 : 0}`, + ), + ), + ...sample( + `${PREFIX}_identity_pending_total`, + 'Nodes waiting for an operator to confirm the device behind them.', + 'gauge', + [`${PREFIX}_identity_pending_total ${pending.length}`], + ), + ); + + lines.push( + ...this.#statusMetric( + `${PREFIX}_sms_outbox`, + 'Queued cross-node SMS sends, by state.', + 'SELECT status, COUNT(*) AS count FROM sms_outbox GROUP BY status', + ['queued', 'sending', 'sent', 'failed', 'cancelled'], + ), + ...this.#statusMetric( + `${PREFIX}_notification_queue`, + 'Central notification queue, by state.', + 'SELECT status, COUNT(*) AS count FROM notification_queue GROUP BY status', + ['pending', 'sending', 'succeeded', 'failed', 'cancelled'], + ), + ...this.#statusMetric( + `${PREFIX}_jobs`, + 'Fleet operations, by state.', + 'SELECT status, COUNT(*) AS count FROM jobs GROUP BY status', + [ + 'queued', + 'running', + 'cancelling', + 'succeeded', + 'partially-succeeded', + 'failed', + 'cancelled', + 'unknown-result', + ], + ), + ...this.#statusMetric( + `${PREFIX}_scheduled_tasks`, + 'Automation tasks, by state.', + `SELECT CASE WHEN enabled=1 THEN 'armed' ELSE 'paused' END AS status, + COUNT(*) AS count FROM scheduled_tasks GROUP BY enabled`, + ['armed', 'paused'], + ), + ); + + const groups = this.#count('SELECT COUNT(*) AS count FROM device_groups'); + const tags = this.#count('SELECT COUNT(*) AS count FROM tag_registry'); + const messages = this.#count('SELECT COUNT(*) AS count FROM sms_messages'); + const probed = new Set(probes.map((probe) => probe.instanceId)); + const unprobed = [...ids].filter((id) => !probed.has(id)).length; + lines.push( + ...sample(`${PREFIX}_device_groups`, 'Node groups.', 'gauge', [ + `${PREFIX}_device_groups ${groups}`, + ]), + ...sample(`${PREFIX}_device_tags`, 'Labels in the tag registry.', 'gauge', [ + `${PREFIX}_device_tags ${tags}`, + ]), + ...sample( + `${PREFIX}_sms_messages`, + 'SMS rows synchronised into the central message centre.', + 'gauge', + [`${PREFIX}_sms_messages ${messages}`], + ), + ...sample( + `${PREFIX}_node_unprobed`, + 'Registered nodes without a health probe in the retained window.', + 'gauge', + [`${PREFIX}_node_unprobed ${unprobed}`], + ), + ); + + return `${lines.join('\n')}\n`; + } + + #count(sql: string): number { + const row = this.#db.prepare(sql).get() as { count?: number } | undefined; + return number(row?.count ?? 0); + } + + /** Every state is emitted, including the empty ones, so a query never sees a series vanish. */ + #statusMetric( + name: string, + help: string, + sql: string, + states: readonly string[], + ): readonly string[] { + const rows = this.#db.prepare(sql).all() as readonly CountRow[]; + const counts = new Map(rows.map((row) => [String(row.status), number(row.count)])); + return sample( + name, + help, + 'gauge', + states.map((state) => `${name}{status="${label(state)}"} ${counts.get(state) ?? 0}`), + ); + } +} diff --git a/apps/api/src/infrastructure/database/database.test.ts b/apps/api/src/infrastructure/database/database.test.ts index 8f9c085..0b6c2a9 100644 --- a/apps/api/src/infrastructure/database/database.test.ts +++ b/apps/api/src/infrastructure/database/database.test.ts @@ -66,6 +66,7 @@ describe('database migrations', () => { 'console_auth_config', 'console_auth_sessions', 'device_groups', + 'device_identities', 'event_journal', 'instance_tags', 'instances', diff --git a/apps/api/src/infrastructure/database/migrations.ts b/apps/api/src/infrastructure/database/migrations.ts index b2ab1a5..5f27145 100644 --- a/apps/api/src/infrastructure/database/migrations.ts +++ b/apps/api/src/infrastructure/database/migrations.ts @@ -638,6 +638,38 @@ export const MIGRATIONS: readonly Migration[] = [ "ALTER TABLE notification_channels ADD COLUMN secret_fields TEXT NOT NULL DEFAULT ''", ], }, + { + id: 17, + name: 'device-identity-guard', + statements: [ + // The Hub refuses to control a device whose identity drifted: the same IMEI claimed by + // two records, or hardware swapped behind an address that is still registered. Both are + // recorded here and only an operator can clear them, because the machine cannot tell + // which of two claimants is the real device. + `CREATE TABLE device_identities ( + instance_id TEXT PRIMARY KEY REFERENCES instances(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'confirmed' CHECK (status IN ('confirmed','pending')), + reasons TEXT NOT NULL DEFAULT '[]', + imei TEXT NOT NULL DEFAULT '', + manufacturer TEXT NOT NULL DEFAULT '', + model TEXT NOT NULL DEFAULT '', + revision TEXT NOT NULL DEFAULT '', + agent TEXT NOT NULL DEFAULT '', + origin TEXT NOT NULL DEFAULT '', + fingerprint TEXT NOT NULL DEFAULT '', + confirmed_fingerprint TEXT NOT NULL DEFAULT '', + confirmed_values_json TEXT NOT NULL DEFAULT '{}', + confirmed_peers_json TEXT NOT NULL DEFAULT '[]', + detail_json TEXT NOT NULL DEFAULT '{}', + observed_at TEXT NOT NULL, + confirmed_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + 'CREATE INDEX idx_device_identities_status ON device_identities(status, instance_id)', + 'CREATE INDEX idx_device_identities_imei ON device_identities(imei, instance_id)', + ], + }, ]; const createMigrationsTable = `CREATE TABLE schema_migrations ( diff --git a/apps/api/src/interface/http/device-action-routes.ts b/apps/api/src/interface/http/device-action-routes.ts index a5ae5e4..e1a59be 100644 --- a/apps/api/src/interface/http/device-action-routes.ts +++ b/apps/api/src/interface/http/device-action-routes.ts @@ -7,6 +7,8 @@ import { export interface DeviceActionRoutesOptions { readonly actions: DeviceActionService; + /** The identity guard, so the console can grey controls out before the operator tries one. */ + readonly identities?: { isBlocked(instanceId: string): boolean }; } function problem( @@ -21,7 +23,7 @@ function problem( .type('application/problem+json') .send({ type: 'about:blank', - title: status === 404 ? 'Not Found' : 'Bad Request', + title: titleFor(status), status, code, detail, @@ -35,6 +37,8 @@ const statusFor = (code: DeviceActionError['code']): number => { return 404; case 'SESSION_INVALID': return 401; + case 'IDENTITY_UNCONFIRMED': + return 409; case 'UPSTREAM_FAILED': case 'NOT_DISPATCHED': return 502; @@ -43,6 +47,9 @@ const statusFor = (code: DeviceActionError['code']): number => { } }; +const titleFor = (status: number): string => + status === 404 ? 'Not Found' : status === 409 ? 'Conflict' : 'Bad Request'; + export function registerDeviceActionRoutes( app: FastifyInstance, options: DeviceActionRoutesOptions, @@ -51,7 +58,10 @@ export function registerDeviceActionRoutes( const instanceId = (request.params as { instanceId?: unknown }).instanceId; if (typeof instanceId !== 'string' || instanceId.length === 0) return problem(request, reply, 400, 'VALIDATION_FAILED', 'Instance id is required'); - return { actions: options.actions.list() }; + return { + actions: options.actions.list(), + identityBlocked: options.identities?.isBlocked(instanceId) === true, + }; }); app.post( @@ -98,7 +108,9 @@ export function registerDeviceActionRoutes( reply, statusFor(error.code), error.code, - 'The device action could not be completed.', + error.code === 'IDENTITY_UNCONFIRMED' + ? '设备身份待确认,请先核对 IMEI 与硬件信息后解除阻止。' + : 'The device action could not be completed.', ); throw error; } diff --git a/apps/api/src/interface/http/fleet-routes.ts b/apps/api/src/interface/http/fleet-routes.ts index d3077ca..00f0e8e 100644 --- a/apps/api/src/interface/http/fleet-routes.ts +++ b/apps/api/src/interface/http/fleet-routes.ts @@ -20,6 +20,11 @@ import { NotificationServiceError, type InstanceNotificationService, } from '../../application/notifications/instance-notification-service.js'; +import type { + DeviceIdentity, + DeviceIdentityService, + IdentityReason, +} from '../../application/identity/device-identity-service.js'; export interface FleetMessageDevice { readonly id: string; readonly name: string; @@ -68,6 +73,14 @@ export interface FleetRoutesOptions { readonly connections?: ConnectionProbe; /** Offline send queue; absent means sends fail fast instead of waiting for the device. */ readonly outbox?: SmsOutboxService; + /** Identity guard; absent means the fleet list has no opinion about hardware drift. */ + readonly identities?: DeviceIdentityService; +} + +export interface FleetIdentitySummary { + readonly tracked: boolean; + readonly status: 'confirmed' | 'pending'; + readonly reasons: readonly IdentityReason[]; } /** One queued send, with the node label the queue table itself does not store. */ @@ -184,6 +197,11 @@ function connectionSummary(state: ConnectionState | undefined): FleetConnectionS }; } +function identitySummary(identity: DeviceIdentity | undefined): FleetIdentitySummary { + if (!identity) return { tracked: false, status: 'confirmed', reasons: [] }; + return { tracked: true, status: identity.status, reasons: identity.reasons }; +} + function record(value: unknown): Record | undefined { if (!value || typeof value !== 'object' || Array.isArray(value)) return undefined; return value as Record; @@ -357,6 +375,7 @@ export function registerFleetRoutes(app: FastifyInstance, options: FleetRoutesOp ...instance, connection: connectionSummary(reachability?.get(instance.id)), resources: await options.resources.get(instance.id), + identity: identitySummary(options.identities?.get(instance.id)), })); return { items }; }); diff --git a/apps/api/src/interface/http/identity-routes.test.ts b/apps/api/src/interface/http/identity-routes.test.ts new file mode 100644 index 0000000..4a249bb --- /dev/null +++ b/apps/api/src/interface/http/identity-routes.test.ts @@ -0,0 +1,173 @@ +import Database from 'better-sqlite3'; +import Fastify from 'fastify'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { DeviceIdentityService } from '../../application/identity/device-identity-service.js'; +import type { InstanceModuleService } from '../../application/instances/instance-module-service.js'; +import type { InstanceService } from '../../application/instances/instance-service.js'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { registerIdentityRoutes } from './identity-routes.js'; + +let subject: { app: ReturnType; db: Database.Database } | undefined; + +const nodes: Record = { + 'node-a': '机房 A', + 'node-b': '机房 B', +}; + +function fixture(read?: (instanceId: string) => Promise) { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + // Identity rows hang off instances, so the fixture nodes have to exist for real. + const seed = db.prepare( + `INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) + VALUES (?,?,'http://' || ? || ':8080','password',1,1,'2026-09-05T02:00:00.000Z','2026-09-05T02:00:00.000Z')`, + ); + for (const [id, name] of Object.entries(nodes)) seed.run(id, name, id); + const identities = new DeviceIdentityService({ db }); + const app = Fastify(); + registerIdentityRoutes(app, { + identities, + instances: { + list: async () => ({ + items: Object.entries(nodes).map(([id, name]) => ({ id, name })), + }), + } as unknown as InstanceService, + ...(read + ? { + modules: { + read: async (instanceId: string) => { + await read(instanceId); + }, + } as unknown as InstanceModuleService, + } + : {}), + }); + subject = { app, db }; + return { app, identities, db }; +} + +afterEach(async () => { + await subject?.app.close(); + subject?.db.close(); + subject = undefined; +}); + +const report = { imei: '860000000000001', model: 'RM500Q', manufacturer: 'Quectel' }; + +describe('identity routes', () => { + it('serves the guard view with node labels for every claimant', async () => { + const { app, identities } = fixture(); + identities.observe('node-a', report); + identities.observe('node-b', report); + + const listed = await app.inject({ method: 'GET', url: '/api/v1/fleet/identities' }); + + expect(listed.statusCode).toBe(200); + const body = listed.json() as { + summary: { tracked: number; pending: number }; + items: Array>; + }; + expect(body.summary).toEqual({ tracked: 2, pending: 2 }); + expect(body.items).toHaveLength(2); + expect(body.items[0]).toMatchObject({ + instanceId: 'node-a', + instanceName: '机房 A', + status: 'pending', + reasons: ['imei_claimed'], + conflicts: [{ reason: 'imei_claimed', instanceIds: ['node-b'], instanceNames: ['机房 B'] }], + imei: '860000000000001', + }); + }); + + it('puts held nodes first no matter which one reported last', async () => { + const { app, identities } = fixture(); + // node-b registers first, so an insertion-order listing would lead with the wrong record. + identities.observe('node-b', report); + identities.observe('node-a', report); + await app.inject({ method: 'POST', url: '/api/v1/fleet/identities/node-b/confirm' }); + + const listed = await app.inject({ method: 'GET', url: '/api/v1/fleet/identities' }); + const body = listed.json() as { items: Array<{ instanceId: string; status: string }> }; + + expect(body.items.map((item) => [item.instanceId, item.status])).toEqual([ + ['node-a', 'pending'], + ['node-b', 'confirmed'], + ]); + }); + + it('clears the guard for the node the operator vouches for', async () => { + const { app, identities } = fixture(); + identities.observe('node-a', report); + identities.observe('node-b', report); + + const confirmed = await app.inject({ + method: 'POST', + url: '/api/v1/fleet/identities/node-a/confirm', + }); + + expect(confirmed.statusCode).toBe(200); + expect(confirmed.json()).toMatchObject({ + identity: { instanceId: 'node-a', status: 'confirmed', reasons: [] }, + }); + expect(identities.isBlocked('node-a')).toBe(false); + expect(identities.isBlocked('node-b')).toBe(true); + }); + + it('answers 404 when there is nothing to confirm', async () => { + const { app } = fixture(); + const response = await app.inject({ + method: 'POST', + url: '/api/v1/fleet/identities/ghost/confirm', + }); + expect(response.statusCode).toBe(404); + expect(response.json()).toMatchObject({ code: 'NOT_FOUND', status: 404 }); + }); + + it('refreshes by reading the device once', async () => { + const observed = vi.fn(); + const { app, identities } = fixture(async (instanceId) => { + observed(instanceId); + identities.observe(instanceId, report); + }); + const response = await app.inject({ + method: 'POST', + url: '/api/v1/fleet/identities/node-a/refresh', + }); + + expect(response.statusCode).toBe(200); + expect(observed).toHaveBeenCalledTimes(1); + expect(observed).toHaveBeenCalledWith('node-a'); + expect(response.json()).toMatchObject({ + observed: true, + identity: { instanceId: 'node-a', status: 'confirmed', model: 'RM500Q' }, + }); + }); + + it('says so when a device answers without any identity at all', async () => { + const { app } = fixture(async () => {}); + + const response = await app.inject({ + method: 'POST', + url: '/api/v1/fleet/identities/node-a/refresh', + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ observed: false }); + }); + + it('reports a device that could not be reached for a refresh', async () => { + const { app } = fixture(async () => { + throw new Error('ECONNREFUSED'); + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/v1/fleet/identities/node-a/refresh', + }); + + expect(response.statusCode).toBe(502); + expect(response.json()).toMatchObject({ code: 'UPSTREAM_FAILED' }); + }); +}); diff --git a/apps/api/src/interface/http/identity-routes.ts b/apps/api/src/interface/http/identity-routes.ts new file mode 100644 index 0000000..ead20d6 --- /dev/null +++ b/apps/api/src/interface/http/identity-routes.ts @@ -0,0 +1,175 @@ +import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; + +import { + DeviceIdentityError, + type DeviceIdentity, + type DeviceIdentityService, +} from '../../application/identity/device-identity-service.js'; +import type { InstanceService } from '../../application/instances/instance-service.js'; +import type { InstanceModuleService } from '../../application/instances/instance-module-service.js'; + +export interface IdentityRoutesOptions { + readonly identities: DeviceIdentityService; + readonly instances: InstanceService; + /** Optional so a control plane without device reads can still show what it already knows. */ + readonly modules?: InstanceModuleService; +} + +interface IdentityView { + readonly instanceId: string; + readonly instanceName: string; + readonly status: DeviceIdentity['status']; + readonly reasons: DeviceIdentity['reasons']; + readonly conflicts: ReadonlyArray<{ + readonly reason: string; + readonly instanceIds: readonly string[]; + readonly instanceNames: readonly string[]; + }>; + readonly imei: string; + readonly manufacturer: string; + readonly model: string; + readonly revision: string; + readonly agent: string; + readonly origin: string; + readonly fingerprint: string; + readonly confirmedFingerprint: string; + readonly changes: DeviceIdentity['changes']; + readonly observedAt: string; + readonly confirmedAt: string; +} + +const PAGE_SIZE = 100; +const MAX_PAGES = 100; + +/** + * Pending records lead, then the node label the operator actually reads. Instance ids are random + * uuids, so sorting by them alone would shuffle the panel between two opens. + */ +function compareViews(a: IdentityView, b: IdentityView): number { + if (a.status !== b.status) return a.status === 'pending' ? -1 : 1; + const byName = (a.instanceName || a.instanceId).localeCompare( + b.instanceName || b.instanceId, + 'zh-Hans-CN', + { numeric: true }, + ); + return byName !== 0 ? byName : a.instanceId.localeCompare(b.instanceId); +} + +async function namesById(instances: InstanceService): Promise> { + const names = new Map(); + for (let page = 1; page <= MAX_PAGES; page += 1) { + const current = await instances.list({ page, pageSize: PAGE_SIZE }); + for (const instance of current.items) names.set(instance.id, instance.name); + if (current.items.length < PAGE_SIZE) break; + } + return names; +} + +function view( + identity: DeviceIdentity, + names: ReadonlyMap, + fallback: string, +): IdentityView { + return { + instanceId: identity.instanceId, + instanceName: names.get(identity.instanceId) ?? fallback, + status: identity.status, + reasons: identity.reasons, + conflicts: identity.conflicts.map((conflict) => ({ + reason: conflict.reason, + instanceIds: conflict.instanceIds, + instanceNames: conflict.instanceIds.map((id) => names.get(id) ?? id), + })), + imei: identity.imei, + manufacturer: identity.manufacturer, + model: identity.model, + revision: identity.revision, + agent: identity.agent, + origin: identity.origin, + fingerprint: identity.fingerprint, + confirmedFingerprint: identity.confirmedFingerprint, + changes: identity.changes, + observedAt: identity.observedAt, + confirmedAt: identity.confirmedAt, + }; +} + +function problem( + request: FastifyRequest, + reply: FastifyReply, + status: number, + code: string, + detail: string, +): FastifyReply { + return reply + .code(status) + .type('application/problem+json') + .send({ + type: 'about:blank', + title: status === 404 ? 'Not Found' : 'Bad Request', + status, + code, + detail, + requestId: request.id, + }); +} + +function instanceIdOf(request: FastifyRequest): string { + const params = request.params as { instanceId?: unknown }; + return typeof params.instanceId === 'string' ? params.instanceId : ''; +} + +/** + * The identity guard's console surface: what each node claims to be, which claims disagree, and + * the one action that clears a disagreement once the operator has looked at the evidence. + */ +export function registerIdentityRoutes(app: FastifyInstance, options: IdentityRoutesOptions): void { + app.get('/api/v1/fleet/identities', async () => { + const names = await namesById(options.instances); + const items = options.identities + .list() + .map((identity) => view(identity, names, '')) + .sort(compareViews); + const summary = options.identities.summarize(); + return { + items, + summary: { tracked: summary.tracked, pending: summary.pending.length }, + }; + }); + + app.post('/api/v1/fleet/identities/:instanceId/confirm', async (request, reply) => { + const instanceId = instanceIdOf(request); + if (!instanceId) + return problem(request, reply, 400, 'VALIDATION_FAILED', 'Instance id is required'); + try { + const identity = options.identities.confirm(instanceId); + const names = await namesById(options.instances); + return { identity: view(identity, names, instanceId) }; + } catch (error) { + if (error instanceof DeviceIdentityError) + return problem(request, reply, 404, error.code, '该节点还没有可确认的身份记录。'); + throw error; + } + }); + + // A refresh is one overview read: the device answers with its own hardware report and the + // guard re-decides, which is also how a node first earns an identity row. + app.post('/api/v1/fleet/identities/:instanceId/refresh', async (request, reply) => { + const instanceId = instanceIdOf(request); + if (!instanceId) + return problem(request, reply, 400, 'VALIDATION_FAILED', 'Instance id is required'); + if (!options.modules) + return problem(request, reply, 400, 'VALIDATION_FAILED', 'Device reads are not available'); + try { + await options.modules.read(instanceId, 'overview'); + } catch { + return problem(request, reply, 502, 'UPSTREAM_FAILED', '设备未能返回身份信息,请稍后重试。'); + } + const identity = options.identities.get(instanceId); + const names = await namesById(options.instances); + return { + observed: identity !== undefined, + ...(identity ? { identity: view(identity, names, instanceId) } : {}), + }; + }); +} diff --git a/apps/api/src/interface/http/metrics-routes.test.ts b/apps/api/src/interface/http/metrics-routes.test.ts new file mode 100644 index 0000000..07d793a --- /dev/null +++ b/apps/api/src/interface/http/metrics-routes.test.ts @@ -0,0 +1,64 @@ +import Database from 'better-sqlite3'; +import Fastify from 'fastify'; +import { afterEach, describe, expect, it } from 'vitest'; + +import { ConsoleAuthService } from '../../application/auth/console-auth-service.js'; +import { DeviceIdentityService } from '../../application/identity/device-identity-service.js'; +import { MetricsService } from '../../application/observability/metrics-service.js'; +import { migrateDatabase } from '../../infrastructure/database/migrations.js'; +import { registerConsoleAuth } from './console-auth-routes.js'; +import { PROMETHEUS_CONTENT_TYPE, registerMetricsRoutes } from './metrics-routes.js'; + +let subject: { app: ReturnType; db: Database.Database } | undefined; + +function fixture() { + const db = new Database(':memory:'); + 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 ('node-a','机房 A','http://10.0.0.9:8080','password',1,1,'2026-09-05T02:00:00.000Z','2026-09-05T02:00:00.000Z')`, + ).run(); + const identities = new DeviceIdentityService({ db }); + const metrics = new MetricsService({ + db, + version: '0.1.0', + sources: { connections: { summarize: () => [] }, identities }, + }); + const app = Fastify(); + // The real guard, so the test proves the exemption rather than asserting a predicate. + registerConsoleAuth(app, new ConsoleAuthService({ db })); + registerMetricsRoutes(app, { metrics }); + subject = { app, db }; + return { app, db, identities }; +} + +afterEach(async () => { + await subject?.app.close(); + subject?.db.close(); + subject = undefined; +}); + +describe('metrics routes', () => { + it('serves the exposition in the format Prometheus expects', async () => { + const { app } = fixture(); + const response = await app.inject({ method: 'GET', url: '/api/v1/metrics' }); + + expect(response.statusCode).toBe(200); + expect(response.headers['content-type']).toContain(PROMETHEUS_CONTENT_TYPE); + expect(response.body).toContain('multi_simadmin_nodes_total 1'); + expect(response.body.endsWith('\n')).toBe(true); + }); + + it('stays open to a scraper while the console itself demands a session', async () => { + const { app, db } = fixture(); + const auth = new ConsoleAuthService({ db }); + await auth.updateSettings({ enabled: true, newPassword: 'passw0rd-1' }); + + const guarded = await app.inject({ method: 'GET', url: '/api/v1/instances' }); + expect(guarded.statusCode).toBe(401); + + const scraped = await app.inject({ method: 'GET', url: '/api/v1/metrics' }); + expect(scraped.statusCode).toBe(200); + }); +}); diff --git a/apps/api/src/interface/http/metrics-routes.ts b/apps/api/src/interface/http/metrics-routes.ts new file mode 100644 index 0000000..b36bf84 --- /dev/null +++ b/apps/api/src/interface/http/metrics-routes.ts @@ -0,0 +1,21 @@ +import type { FastifyInstance } from 'fastify'; + +import type { MetricsService } from '../../application/observability/metrics-service.js'; + +/** The content type Prometheus expects; anything else makes the scrape fail silently. */ +export const PROMETHEUS_CONTENT_TYPE = 'text/plain; version=0.0.4; charset=utf-8'; + +export interface MetricsRoutesOptions { + readonly metrics: MetricsService; +} + +/** + * Monitoring surface, modelled on the Hub's `/metrics`: no browser session, no cookies, just + * text. It is deliberately read-only and built from rows the console already writes, so a + * scraper polling every few seconds cannot wake a device. + */ +export function registerMetricsRoutes(app: FastifyInstance, options: MetricsRoutesOptions): void { + app.get('/api/v1/metrics', async (_request, reply) => { + return reply.type(PROMETHEUS_CONTENT_TYPE).send(options.metrics.render()); + }); +}