From be964eccb7124307f75979c3eb25ae7a33d1e1d9 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 11 Aug 2026 20:14:55 +0300 Subject: [PATCH] feat(device-plane): supervise pinned edge channels in core --- device-plane/docker-compose.device-plane.yml | 1 + .../services/device-control-core/Dockerfile | 2 + .../migrations/013_device_edge_channels.sql | 54 +++ .../services/device-control-core/src/app.mjs | 77 ++-- .../src/edge-channel-supervisor.mjs | 351 ++++++++++++++++++ .../src/gateway-ingest.mjs | 78 ++++ .../src/infrastructure-management.mjs | 110 +++++- .../src/infrastructure-repository.mjs | 91 ++++- .../src/postgres-repository.mjs | 31 ++ .../device-control-core/src/server.mjs | 132 ++++++- .../device-control-core/test/app.test.mjs | 6 + .../device-edge-channel-migration.test.mjs | 22 ++ .../test/edge-channel-supervisor.test.mjs | 115 ++++++ .../test/gateway-ingest.test.mjs | 87 +++++ .../test/infrastructure-management.test.mjs | 56 +++ .../test/infrastructure-repository.test.mjs | 42 +++ .../test/channel-integration.test.mjs | 21 +- .../device-gateway-core/src/runtime.mjs | 11 +- 18 files changed, 1217 insertions(+), 70 deletions(-) create mode 100644 device-plane/services/device-control-core/migrations/013_device_edge_channels.sql create mode 100644 device-plane/services/device-control-core/src/edge-channel-supervisor.mjs create mode 100644 device-plane/services/device-control-core/src/gateway-ingest.mjs create mode 100644 device-plane/services/device-control-core/test/device-edge-channel-migration.test.mjs create mode 100644 device-plane/services/device-control-core/test/edge-channel-supervisor.test.mjs create mode 100644 device-plane/services/device-control-core/test/gateway-ingest.test.mjs diff --git a/device-plane/docker-compose.device-plane.yml b/device-plane/docker-compose.device-plane.yml index 3aae261..b8b859c 100644 --- a/device-plane/docker-compose.device-plane.yml +++ b/device-plane/docker-compose.device-plane.yml @@ -44,6 +44,7 @@ services: DEVICE_DATABASE_PASSWORD_FILE: /run/nodedc-secrets/postgres-password DEVICE_DATABASE_POOL_SIZE: "10" DEVICE_DISCOVERY_INGEST_ENABLED: "true" + DEVICE_EDGE_CHANNEL_ENABLED: "false" DEVICE_GATEWAY_CORE_TOKEN_FILE: /run/nodedc-secrets/gateway-core-token DEVICE_IDENTIFIER_PEPPER_FILE: /run/nodedc-secrets/identifier-pepper volumes: diff --git a/device-plane/services/device-control-core/Dockerfile b/device-plane/services/device-control-core/Dockerfile index c0d85c3..67ea7aa 100644 --- a/device-plane/services/device-control-core/Dockerfile +++ b/device-plane/services/device-control-core/Dockerfile @@ -4,12 +4,14 @@ WORKDIR /app COPY package.json package-lock.json ./ COPY packages/device-protocol-contract ./packages/device-protocol-contract +COPY packages/device-edge-channel-contract ./packages/device-edge-channel-contract COPY packages/device-adapter-runtime/package.json ./packages/device-adapter-runtime/package.json COPY packages/device-adapter-catalog/package.json ./packages/device-adapter-catalog/package.json COPY packages/arusnavi-b2-adapter/package.json ./packages/arusnavi-b2-adapter/package.json COPY services/device-control-core ./services/device-control-core COPY services/device-gateway/package.json ./services/device-gateway/package.json COPY services/device-edge-relay/package.json ./services/device-edge-relay/package.json +COPY services/device-gateway-core ./services/device-gateway-core RUN npm ci --omit=dev --ignore-scripts diff --git a/device-plane/services/device-control-core/migrations/013_device_edge_channels.sql b/device-plane/services/device-control-core/migrations/013_device_edge_channels.sql new file mode 100644 index 0000000..1ec41d4 --- /dev/null +++ b/device-plane/services/device-control-core/migrations/013_device_edge_channels.sql @@ -0,0 +1,54 @@ +begin; + +alter table device_edges + add column if not exists channel_endpoint text, + add column if not exists channel_servername text, + add column if not exists channel_generation_ref text, + add column if not exists channel_trust_bundle_ref text, + add column if not exists channel_certificate_identities jsonb not null + default '[]'::jsonb, + add column if not exists channel_lifecycle_state text not null + default 'disabled'; + +do $$ +begin + if not exists ( + select 1 from pg_constraint + where conname = 'device_edges_channel_lifecycle_state_check' + ) then + alter table device_edges add constraint device_edges_channel_lifecycle_state_check + check (channel_lifecycle_state in ('disabled', 'active', 'revoked')); + end if; + if not exists ( + select 1 from pg_constraint + where conname = 'device_edges_channel_configuration_check' + ) then + alter table device_edges add constraint device_edges_channel_configuration_check + check ( + ( + channel_lifecycle_state = 'disabled' + and channel_endpoint is null + and channel_servername is null + and channel_generation_ref is null + and channel_trust_bundle_ref is null + and channel_certificate_identities = '[]'::jsonb + ) + or + ( + channel_lifecycle_state in ('active', 'revoked') + and length(btrim(channel_endpoint)) between 12 and 256 + and channel_servername ~ '^[A-Za-z0-9.-]{1,253}$' + and channel_generation_ref ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$' + and channel_trust_bundle_ref ~ '^edge-trust:[a-z][a-z0-9-]{1,62}$' + and jsonb_typeof(channel_certificate_identities) = 'array' + and jsonb_array_length(channel_certificate_identities) between 1 and 2 + ) + ); + end if; +end $$; + +create index if not exists device_edges_active_channel_idx + on device_edges (channel_lifecycle_state, updated_at desc) + where channel_lifecycle_state = 'active'; + +commit; diff --git a/device-plane/services/device-control-core/src/app.mjs b/device-plane/services/device-control-core/src/app.mjs index 29632f1..6a493b7 100644 --- a/device-plane/services/device-control-core/src/app.mjs +++ b/device-plane/services/device-control-core/src/app.mjs @@ -2,16 +2,11 @@ import { createHash, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; import { - assertSafeProjection, hashRestrictedIdentifier, maskRestrictedIdentifier, - normalizeAdapterAcceptance, - normalizeAdapterMessage, - normalizeDiscoverySignal, normalizeRestrictedIdentifier, - toSafeDiscoveryView, - toSafeAdapterMessageView, } from "../../../packages/device-protocol-contract/src/index.mjs"; +import { createDeviceGatewayIngest } from "./gateway-ingest.mjs"; import { normalizeManagementActor, } from "./project-management.mjs"; @@ -59,6 +54,8 @@ export function createControlCoreApp({ discoveryIngestEnabled = false, managementApiEnabled = false, managementToken = "", + gatewayIngest = null, + edgeChannelStatusProvider = null, } = {}) { if (!repository || typeof repository.health !== "function") { throw new TypeError("device_repository_required"); @@ -88,6 +85,24 @@ export function createControlCoreApp({ throw new TypeError("device_identifier_pepper_invalid"); } } + const ingest = discoveryIngestEnabled + ? gatewayIngest ?? createDeviceGatewayIngest({ repository, identifierPepper }) + : gatewayIngest; + if ( + ingest + && ( + typeof ingest.observeDiscovery !== "function" + || typeof ingest.acceptMessage !== "function" + ) + ) { + throw new TypeError("device_gateway_ingest_invalid"); + } + if ( + edgeChannelStatusProvider != null + && typeof edgeChannelStatusProvider !== "function" + ) { + throw new TypeError("device_edge_channel_status_provider_invalid"); + } const server = createServer(async (request, response) => { response.setHeader("Content-Type", "application/json; charset=utf-8"); @@ -108,6 +123,9 @@ export function createControlCoreApp({ database, discoveryIngest: discoveryIngestEnabled ? "enabled" : "disabled", managementApi: managementApiEnabled ? "enabled" : "disabled", + edgeChannels: edgeChannelStatusProvider + ? edgeChannelStatusProvider() + : { enabled: false, configured: 0, accepted: 0, degraded: 0 }, commandTransport: "disabled", }); } @@ -236,22 +254,11 @@ export function createControlCoreApp({ } const input = await readJsonBody(request, 32 * 1024); - const signal = normalizeDiscoverySignal(input); - const identifierDigest = hashRestrictedIdentifier( - signal.identifier, - identifierPepper, - ); - const safeView = assertSafeProjection(toSafeDiscoveryView(signal)); - const discovery = await repository.upsertQuarantineDiscovery({ - identifierDigest, - safeView, - sessionRef: signal.sessionRef, - routeRef: signal.routeRef ?? null, - }); + const discovery = await ingest.observeDiscovery(input); return writeJson(response, discovery.created ? 201 : 200, { ok: true, created: discovery.created, - discovery: assertSafeProjection(discovery.value), + discovery: discovery.value, }); } @@ -273,31 +280,7 @@ export function createControlCoreApp({ } const input = await readJsonBody(request, 1024 * 1024); - const message = normalizeAdapterMessage(input); - const identifierDigest = hashRestrictedIdentifier( - message.identifier, - identifierPepper, - ); - const safeView = assertSafeProjection(toSafeAdapterMessageView(message)); - const requestDigest = gatewayMessageRequestDigest({ - edgeRef: safeView.edgeRef, - adapterRef: safeView.adapterRef, - protocolProfileRef: safeView.protocolProfileRef, - protocol: safeView.protocol, - routeRef: safeView.routeRef ?? null, - idempotencyKey: safeView.idempotencyKey, - identifierKind: safeView.identifier.kind, - identifierDigest, - payloadSchemaRef: safeView.payloadSchemaRef, - payload: safeView.payload, - }); - const acceptance = normalizeAdapterAcceptance( - await repository.acceptAdapterMessage({ - identifierDigest, - requestDigest, - safeView, - }), - ); + const acceptance = await ingest.acceptMessage(input); return writeJson(response, acceptance.replayed ? 200 : 201, { ok: true, acceptance, @@ -421,12 +404,6 @@ function managementRequestDigest(value) { .digest("hex")}`; } -function gatewayMessageRequestDigest(value) { - return `sha256:${createHash("sha256") - .update(JSON.stringify(value), "utf8") - .digest("hex")}`; -} - function matchesBearer(header, expected) { if (typeof header !== "string" || !header.startsWith("Bearer ")) return false; const actual = Buffer.from(header.slice("Bearer ".length), "utf8"); diff --git a/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs b/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs new file mode 100644 index 0000000..a9ac683 --- /dev/null +++ b/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs @@ -0,0 +1,351 @@ +import { createHash, X509Certificate } from "node:crypto"; +import { lstat, readFile } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; + +import { + normalizeCertificateFingerprint, + normalizeCertificateIdentities, +} from "../../../packages/device-edge-channel-contract/src/index.mjs"; +import { + createDeviceGatewayCoreChannelClient, +} from "../../device-gateway-core/src/runtime.mjs"; + +const DEFAULT_TRUST_ROOT = "/run/nodedc-secrets/device-edge-channel/peers"; + +export function createDeviceEdgeChannelSupervisor(options = {}) { + const config = normalizeConfiguration(options); + const clients = new Map(); + const failures = new Map(); + let running = false; + let timer = null; + let reconcilePromise = null; + let requestedCount = 0; + let reconciliationFailures = 0; + let lastErrorCode = null; + + return Object.freeze({ + async start() { + if (running) return; + running = true; + await reconcile(); + schedule(); + }, + async stop() { + running = false; + clearTimeout(timer); + timer = null; + if (reconcilePromise) await reconcilePromise.catch(() => undefined); + const stopping = [...clients.values()].map(({ client }) => client.stop()); + clients.clear(); + failures.clear(); + await Promise.allSettled(stopping); + }, + reconcile, + status, + }); + + async function reconcile() { + if (!running) return status(); + if (reconcilePromise) return reconcilePromise; + reconcilePromise = performReconcile().finally(() => { + reconcilePromise = null; + }); + return reconcilePromise; + } + + async function performReconcile() { + let registrations; + try { + registrations = await config.repository + .listActiveEdgeChannelRegistrations(config.maxEdges); + if (!Array.isArray(registrations) || registrations.length > config.maxEdges) { + throw new TypeError("device_edge_channel_registration_set_invalid"); + } + registrations = registrations.map(normalizeRegistration); + if (new Set(registrations.map((item) => item.edgeRegistrationId)).size + !== registrations.length) { + throw new TypeError("device_edge_channel_registration_set_invalid"); + } + requestedCount = registrations.length; + } catch (error) { + reconciliationFailures += 1; + lastErrorCode = safeErrorCode(error); + return status(); + } + + const desiredIds = new Set(registrations.map((item) => item.edgeRegistrationId)); + for (const [edgeRegistrationId, active] of clients) { + if (!desiredIds.has(edgeRegistrationId)) { + clients.delete(edgeRegistrationId); + await active.client.stop().catch(() => undefined); + } + } + for (const edgeRegistrationId of failures.keys()) { + if (!desiredIds.has(edgeRegistrationId)) failures.delete(edgeRegistrationId); + } + + for (const registration of registrations) { + const digest = registrationDigest(registration); + const current = clients.get(registration.edgeRegistrationId); + if (current?.digest === digest) { + failures.delete(registration.edgeRegistrationId); + continue; + } + if (current) { + clients.delete(registration.edgeRegistrationId); + await current.client.stop().catch(() => undefined); + } + try { + const ca = await config.readPeerTrust({ + registration, + trustRoot: config.trustRoot, + }); + const client = config.clientFactory({ + registration, + tls: { + key: config.coreIdentity.key, + cert: config.coreIdentity.cert, + ca, + }, + coreIdentity: config.coreIdentity.identityRef, + observeDiscovery: config.gatewayIngest.observeDiscovery, + acceptMessage: config.gatewayIngest.acceptMessage, + }); + assertClient(client); + clients.set(registration.edgeRegistrationId, { client, digest }); + failures.delete(registration.edgeRegistrationId); + await client.start(); + } catch (error) { + const code = safeErrorCode(error); + failures.set(registration.edgeRegistrationId, code); + lastErrorCode = code; + } + } + return status(); + } + + function schedule() { + if (!running) return; + timer = setTimeout(async () => { + timer = null; + await reconcile().catch(() => undefined); + schedule(); + }, config.reconcileIntervalMs); + timer.unref?.(); + } + + function status() { + let accepted = 0; + let connecting = 0; + let degraded = failures.size; + const edges = []; + for (const [edgeRegistrationId, { client }] of clients) { + const clientStatus = client.status(); + if (clientStatus.channel === "accepted") accepted += 1; + else connecting += 1; + if (clientStatus.lastErrorCode) degraded += 1; + edges.push(Object.freeze({ + edgeRegistrationId, + channel: clientStatus.channel, + lastErrorCode: clientStatus.lastErrorCode ?? null, + })); + } + for (const [edgeRegistrationId, code] of failures) { + edges.push(Object.freeze({ + edgeRegistrationId, + channel: "absent", + lastErrorCode: code, + })); + } + edges.sort((left, right) => + left.edgeRegistrationId.localeCompare(right.edgeRegistrationId) + ); + return Object.freeze({ + enabled: true, + running, + configured: requestedCount, + accepted, + connecting, + degraded, + reconciliationFailures, + lastErrorCode, + commandTransport: "disabled", + edges: Object.freeze(edges), + }); + } +} + +export async function readPinnedEdgeTrust({ registration, trustRoot }) { + const match = registration.trustBundleRef.match( + /^edge-trust:([a-z][a-z0-9-]{1,62})$/, + ); + if (!match) throw new TypeError("device_edge_channel_trust_bundle_ref_invalid"); + const root = resolve(trustRoot); + const path = resolve(root, `${match[1]}.pem`); + if (dirname(path) !== root) { + throw new TypeError("device_edge_channel_trust_bundle_path_invalid"); + } + const state = await lstat(path); + if (state.isSymbolicLink() || !state.isFile() || state.size < 1 || state.size > 64 * 1024) { + throw new Error("device_edge_channel_trust_bundle_file_invalid"); + } + const pem = await readFile(path); + const blocks = pem.toString("utf8").match( + /-----BEGIN CERTIFICATE-----[\s\S]+?-----END CERTIFICATE-----/g, + ); + if (!blocks || blocks.length < 1 || blocks.length > 2) { + throw new Error("device_edge_channel_trust_bundle_invalid"); + } + const expected = new Set( + registration.certificateIdentities.map((item) => item.fingerprint), + ); + const observed = new Set(blocks.map((block) => normalizeCertificateFingerprint( + new X509Certificate(block).fingerprint256, + ))); + if ( + observed.size !== expected.size + || [...observed].some((fingerprint) => !expected.has(fingerprint)) + ) { + throw new Error("device_edge_channel_trust_bundle_identity_mismatch"); + } + return pem; +} + +function normalizeConfiguration(options) { + if ( + !options.repository + || typeof options.repository.listActiveEdgeChannelRegistrations !== "function" + ) { + throw new TypeError("device_edge_channel_repository_required"); + } + if ( + !options.gatewayIngest + || typeof options.gatewayIngest.observeDiscovery !== "function" + || typeof options.gatewayIngest.acceptMessage !== "function" + ) { + throw new TypeError("device_edge_channel_gateway_ingest_required"); + } + const coreIdentity = options.coreIdentity; + if ( + !coreIdentity + || !(typeof coreIdentity.key === "string" || Buffer.isBuffer(coreIdentity.key)) + || !(typeof coreIdentity.cert === "string" || Buffer.isBuffer(coreIdentity.cert)) + || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(coreIdentity.identityRef) + ) { + throw new TypeError("device_edge_channel_core_identity_invalid"); + } + const maxEdges = normalizeInteger(options.maxEdges, 1, 64, 32); + const reconcileIntervalMs = normalizeInteger( + options.reconcileIntervalMs, + 1_000, + 300_000, + 15_000, + ); + const trustRoot = resolve(options.trustRoot ?? DEFAULT_TRUST_ROOT); + return Object.freeze({ + repository: options.repository, + gatewayIngest: options.gatewayIngest, + coreIdentity: Object.freeze({ ...coreIdentity }), + maxEdges, + reconcileIntervalMs, + trustRoot, + readPeerTrust: options.readPeerTrust ?? readPinnedEdgeTrust, + clientFactory: options.clientFactory ?? createDeviceGatewayCoreChannelClient, + }); +} + +function normalizeRegistration(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError("device_edge_channel_registration_invalid"); + } + const endpoint = new URL(String(value.endpoint || "")); + if ( + endpoint.protocol !== "https:" + || endpoint.username + || endpoint.password + || endpoint.pathname !== "/" + || endpoint.search + || endpoint.hash + || endpoint.port !== "8443" + || endpoint.hostname !== String(value.servername || "").toLowerCase() + || !isPublicIpv4(endpoint.hostname) + ) { + throw new TypeError("device_edge_channel_registration_endpoint_invalid"); + } + if (value.lifecycleState !== "active") { + throw new TypeError("device_edge_channel_registration_inactive"); + } + return Object.freeze({ + edgeRegistrationId: normalizeRef(value.edgeRegistrationId), + endpoint: endpoint.toString(), + servername: endpoint.hostname, + channelGeneration: normalizeRef(value.channelGeneration), + trustBundleRef: normalizeTrustRef(value.trustBundleRef), + certificateIdentities: normalizeCertificateIdentities( + value.certificateIdentities, + ), + lifecycleState: "active", + }); +} + +function normalizeRef(value) { + if (typeof value !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value)) { + throw new TypeError("device_edge_channel_registration_ref_invalid"); + } + return value; +} + +function normalizeTrustRef(value) { + if (typeof value !== "string" || !/^edge-trust:[a-z][a-z0-9-]{1,62}$/.test(value)) { + throw new TypeError("device_edge_channel_trust_bundle_ref_invalid"); + } + return value; +} + +function isPublicIpv4(value) { + const octets = value.split(".").map(Number); + if (octets.length !== 4 || octets.some((item) => + !Number.isInteger(item) || item < 0 || item > 255 + )) return false; + const [a, b, c] = octets; + return a >= 1 && a < 224 + && a !== 10 && a !== 127 + && !(a === 100 && b >= 64 && b <= 127) + && !(a === 169 && b === 254) + && !(a === 172 && b >= 16 && b <= 31) + && !(a === 192 && (b === 0 || b === 168)) + && !(a === 192 && b === 88 && c === 99) + && !(a === 198 && (b === 18 || b === 19 || b === 51)) + && !(a === 203 && b === 0 && c === 113); +} + +function registrationDigest(value) { + return createHash("sha256").update(JSON.stringify(value)).digest("hex"); +} + +function assertClient(value) { + if ( + !value + || typeof value.start !== "function" + || typeof value.stop !== "function" + || typeof value.status !== "function" + ) throw new TypeError("device_edge_channel_client_invalid"); +} + +function normalizeInteger(value, minimum, maximum, fallback) { + const parsed = Number(value ?? fallback); + if (!Number.isSafeInteger(parsed) || parsed < minimum || parsed > maximum) { + throw new TypeError("device_edge_channel_integer_invalid"); + } + return parsed; +} + +function safeErrorCode(error) { + const value = String(error?.message || error || "device_edge_channel_error") + .toLowerCase() + .replaceAll(/[^a-z0-9._:-]/g, "_") + .slice(0, 128); + return /^[a-z][a-z0-9._:-]{2,127}$/.test(value) + ? value + : "device_edge_channel_error"; +} diff --git a/device-plane/services/device-control-core/src/gateway-ingest.mjs b/device-plane/services/device-control-core/src/gateway-ingest.mjs new file mode 100644 index 0000000..3592371 --- /dev/null +++ b/device-plane/services/device-control-core/src/gateway-ingest.mjs @@ -0,0 +1,78 @@ +import { createHash } from "node:crypto"; + +import { + assertSafeProjection, + hashRestrictedIdentifier, + normalizeAdapterAcceptance, + normalizeAdapterMessage, + normalizeDiscoverySignal, + toSafeAdapterMessageView, + toSafeDiscoveryView, +} from "../../../packages/device-protocol-contract/src/index.mjs"; + +export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) { + if (!repository || typeof repository.upsertQuarantineDiscovery !== "function") { + throw new TypeError("device_discovery_repository_required"); + } + if (typeof repository.acceptAdapterMessage !== "function") { + throw new TypeError("device_gateway_message_repository_required"); + } + if (typeof identifierPepper !== "string" || identifierPepper.length < 32) { + throw new TypeError("device_identifier_pepper_invalid"); + } + + return Object.freeze({ + async observeDiscovery(input) { + const signal = normalizeDiscoverySignal(input); + const identifierDigest = hashRestrictedIdentifier( + signal.identifier, + identifierPepper, + ); + const safeView = assertSafeProjection(toSafeDiscoveryView(signal)); + const discovery = await repository.upsertQuarantineDiscovery({ + identifierDigest, + safeView, + sessionRef: signal.sessionRef, + routeRef: signal.routeRef ?? null, + }); + return Object.freeze({ + created: discovery.created === true, + value: assertSafeProjection(discovery.value), + }); + }, + + async acceptMessage(input) { + const message = normalizeAdapterMessage(input); + const identifierDigest = hashRestrictedIdentifier( + message.identifier, + identifierPepper, + ); + const safeView = assertSafeProjection(toSafeAdapterMessageView(message)); + const requestDigest = gatewayMessageRequestDigest({ + edgeRef: safeView.edgeRef, + adapterRef: safeView.adapterRef, + protocolProfileRef: safeView.protocolProfileRef, + protocol: safeView.protocol, + routeRef: safeView.routeRef ?? null, + idempotencyKey: safeView.idempotencyKey, + identifierKind: safeView.identifier.kind, + identifierDigest, + payloadSchemaRef: safeView.payloadSchemaRef, + payload: safeView.payload, + }); + return normalizeAdapterAcceptance( + await repository.acceptAdapterMessage({ + identifierDigest, + requestDigest, + safeView, + }), + ); + }, + }); +} + +function gatewayMessageRequestDigest(value) { + return `sha256:${createHash("sha256") + .update(JSON.stringify(value), "utf8") + .digest("hex")}`; +} diff --git a/device-plane/services/device-control-core/src/infrastructure-management.mjs b/device-plane/services/device-control-core/src/infrastructure-management.mjs index efeafca..6ec8069 100644 --- a/device-plane/services/device-control-core/src/infrastructure-management.mjs +++ b/device-plane/services/device-control-core/src/infrastructure-management.mjs @@ -2,6 +2,9 @@ import { assertIdentifierDigest, assertSafeProjection, } from "../../../packages/device-protocol-contract/src/index.mjs"; +import { + normalizeCertificateIdentities, +} from "../../../packages/device-edge-channel-contract/src/index.mjs"; import { normalizeManagementActor } from "./project-management.mjs"; export const DEVICE_INFRASTRUCTURE_COMMAND_KINDS = Object.freeze([ @@ -166,8 +169,9 @@ export function normalizeInfrastructureManagementCommand(kind, input) { "displayName", "deploymentRef", "lifecycleState", + "channel", ]); - return Object.freeze({ + const normalized = { edgeKey: normalizeKey(input.edgeKey, "device_edge_key_invalid"), displayName: normalizeDisplayText( input.displayName, @@ -183,7 +187,11 @@ export function normalizeInfrastructureManagementCommand(kind, input) { new Set(["provisioning", "active", "suspended", "retired"]), "device_edge_state_invalid", ), - }); + }; + if (input.channel !== undefined) { + normalized.channel = normalizeEdgeChannel(input.channel); + } + return Object.freeze(normalized); } if (kind === "route.ensure") { @@ -339,6 +347,104 @@ function normalizeOptionalOpaqueRef(value, code) { return normalizeOpaqueRef(value, code); } +function normalizeEdgeChannel(input) { + assertPlainObject(input, "device_edge_channel_invalid"); + assertAllowedKeys(input, [ + "endpoint", + "servername", + "generationRef", + "trustBundleRef", + "certificateIdentities", + "lifecycleState", + ]); + const lifecycleState = normalizeEnum( + input.lifecycleState ?? "disabled", + new Set(["disabled", "active", "revoked"]), + "device_edge_channel_state_invalid", + ); + if (lifecycleState === "disabled") { + if (Object.keys(input).some((key) => key !== "lifecycleState")) { + throw new TypeError("device_edge_channel_disabled_configuration_invalid"); + } + return Object.freeze({ + endpoint: null, + servername: null, + generationRef: null, + trustBundleRef: null, + certificateIdentities: Object.freeze([]), + lifecycleState, + }); + } + + const endpoint = normalizeEdgeEndpoint(input.endpoint); + const servername = normalizePattern( + input.servername, + /^[A-Za-z0-9.-]{1,253}$/, + "device_edge_channel_servername_invalid", + ).toLowerCase(); + if (servername !== endpoint.hostname) { + throw new TypeError("device_edge_channel_servername_mismatch"); + } + return Object.freeze({ + endpoint: endpoint.toString(), + servername, + generationRef: normalizeProfileRef( + input.generationRef, + "device_edge_channel_generation_invalid", + ), + trustBundleRef: normalizePattern( + input.trustBundleRef, + /^edge-trust:[a-z][a-z0-9-]{1,62}$/, + "device_edge_channel_trust_bundle_ref_invalid", + ), + certificateIdentities: normalizeCertificateIdentities( + input.certificateIdentities, + ), + lifecycleState, + }); +} + +function normalizeEdgeEndpoint(value) { + let endpoint; + try { + endpoint = new URL(String(value || "")); + } catch { + throw new TypeError("device_edge_channel_endpoint_invalid"); + } + if ( + endpoint.protocol !== "https:" + || endpoint.username + || endpoint.password + || endpoint.pathname !== "/" + || endpoint.search + || endpoint.hash + || endpoint.port !== "8443" + || !isPublicIpv4(endpoint.hostname) + ) { + throw new TypeError("device_edge_channel_endpoint_invalid"); + } + return endpoint; +} + +function isPublicIpv4(value) { + const octets = value.split(".").map(Number); + if ( + octets.length !== 4 + || octets.some((part) => !Number.isInteger(part) || part < 0 || part > 255) + ) return false; + const [a, b, c] = octets; + if (a < 1 || a >= 224) return false; + if (a === 10 || a === 127) return false; + if (a === 100 && b >= 64 && b <= 127) return false; + if (a === 169 && b === 254) return false; + if (a === 172 && b >= 16 && b <= 31) return false; + if (a === 192 && (b === 0 || b === 168)) return false; + if (a === 192 && b === 88 && c === 99) return false; + if (a === 198 && (b === 18 || b === 19 || b === 51)) return false; + if (a === 203 && b === 0 && c === 113) return false; + return true; +} + function normalizePattern(value, pattern, code) { if (typeof value !== "string" || !pattern.test(value)) { throw new TypeError(code); diff --git a/device-plane/services/device-control-core/src/infrastructure-repository.mjs b/device-plane/services/device-control-core/src/infrastructure-repository.mjs index 29405ed..bad8cc7 100644 --- a/device-plane/services/device-control-core/src/infrastructure-repository.mjs +++ b/device-plane/services/device-control-core/src/infrastructure-repository.mjs @@ -304,6 +304,15 @@ async function registerModelProfile(client, actor, command) { async function ensureEdge(client, actor, command) { assertPlatformCatalogAuthority(actor); + const channelProvided = command.channel !== undefined; + const channel = command.channel ?? { + endpoint: null, + servername: null, + generationRef: null, + trustBundleRef: null, + certificateIdentities: [], + lifecycleState: "disabled", + }; const result = await client.query( `insert into device_edges ( id, @@ -311,35 +320,78 @@ async function ensureEdge(client, actor, command) { display_name, deployment_ref, lifecycle_state, + channel_endpoint, + channel_servername, + channel_generation_ref, + channel_trust_bundle_ref, + channel_certificate_identities, + channel_lifecycle_state, created_by_ref - ) values ($1, $2, $3, $4, $5, $6) + ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11, $13) on conflict (edge_key) do update set display_name = excluded.display_name, deployment_ref = excluded.deployment_ref, lifecycle_state = excluded.lifecycle_state, + channel_endpoint = case when $12 then excluded.channel_endpoint + else device_edges.channel_endpoint end, + channel_servername = case when $12 then excluded.channel_servername + else device_edges.channel_servername end, + channel_generation_ref = case when $12 then excluded.channel_generation_ref + else device_edges.channel_generation_ref end, + channel_trust_bundle_ref = case when $12 then excluded.channel_trust_bundle_ref + else device_edges.channel_trust_bundle_ref end, + channel_certificate_identities = case when $12 + then excluded.channel_certificate_identities + else device_edges.channel_certificate_identities end, + channel_lifecycle_state = case when $12 + then excluded.channel_lifecycle_state + else device_edges.channel_lifecycle_state end, updated_at = now() - where - device_edges.lifecycle_state = excluded.lifecycle_state - or ( - device_edges.lifecycle_state = 'provisioning' - and excluded.lifecycle_state in ('active', 'retired') + where ( + device_edges.lifecycle_state = excluded.lifecycle_state + or ( + device_edges.lifecycle_state = 'provisioning' + and excluded.lifecycle_state in ('active', 'retired') + ) + or ( + device_edges.lifecycle_state = 'active' + and excluded.lifecycle_state in ('suspended', 'retired') + ) + or ( + device_edges.lifecycle_state = 'suspended' + and excluded.lifecycle_state in ('active', 'retired') + ) ) - or ( - device_edges.lifecycle_state = 'active' - and excluded.lifecycle_state in ('suspended', 'retired') - ) - or ( - device_edges.lifecycle_state = 'suspended' - and excluded.lifecycle_state in ('active', 'retired') + and ( + not $12 + or device_edges.channel_lifecycle_state = excluded.channel_lifecycle_state + or ( + device_edges.channel_lifecycle_state = 'disabled' + and excluded.channel_lifecycle_state = 'active' + ) + or ( + device_edges.channel_lifecycle_state = 'active' + and excluded.channel_lifecycle_state in ('disabled', 'revoked') + ) ) returning id, edge_key, display_name, deployment_ref, lifecycle_state, - created_at, updated_at, (xmax = 0) as created`, + channel_endpoint, channel_servername, channel_generation_ref, + channel_trust_bundle_ref, channel_certificate_identities, + channel_lifecycle_state, created_at, updated_at, + (xmax = 0) as created`, [ randomUUID(), command.edgeKey, command.displayName, command.deploymentRef, command.lifecycleState, + channel.endpoint, + channel.servername, + channel.generationRef, + channel.trustBundleRef, + JSON.stringify(channel.certificateIdentities), + channel.lifecycleState, + channelProvided, actor.userRef, ], ); @@ -352,6 +404,9 @@ async function ensureEdge(client, actor, command) { edgeKey: row.edge_key, deploymentRef: row.deployment_ref, lifecycleState: row.lifecycle_state, + channelLifecycleState: row.channel_lifecycle_state, + channelGenerationRef: row.channel_generation_ref, + channelTrustBundleRef: row.channel_trust_bundle_ref, }, }); return { @@ -767,6 +822,14 @@ function edgeView(row) { displayName: row.display_name, deploymentRef: row.deployment_ref ?? null, lifecycleState: row.lifecycle_state, + channel: { + lifecycleState: row.channel_lifecycle_state ?? "disabled", + endpoint: row.channel_endpoint ?? null, + servername: row.channel_servername ?? null, + generationRef: row.channel_generation_ref ?? null, + trustBundleRef: row.channel_trust_bundle_ref ?? null, + certificateIdentities: [...(row.channel_certificate_identities ?? [])], + }, createdAt: toIso(row.created_at), updatedAt: toIso(row.updated_at), }; diff --git a/device-plane/services/device-control-core/src/postgres-repository.mjs b/device-plane/services/device-control-core/src/postgres-repository.mjs index 77902ad..e1fbc29 100644 --- a/device-plane/services/device-control-core/src/postgres-repository.mjs +++ b/device-plane/services/device-control-core/src/postgres-repository.mjs @@ -57,6 +57,7 @@ const migrationFiles = [ "010_device_control_resources.sql", "011_device_control_resource_commands.sql", "012_device_gateway_message_receipts.sql", + "013_device_edge_channels.sql", ]; export class PostgresDeviceRepository { @@ -168,6 +169,36 @@ export class PostgresDeviceRepository { ); } + async listActiveEdgeChannelRegistrations(limit = 64) { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 64) { + throw new TypeError("device_edge_channel_registration_limit_invalid"); + } + return this.#executeRead(async (client) => { + const result = await client.query( + `select id, channel_endpoint, channel_servername, + channel_generation_ref, channel_trust_bundle_ref, + channel_certificate_identities, channel_lifecycle_state + from device_edges + where lifecycle_state = 'active' + and channel_lifecycle_state = 'active' + order by id + limit $1`, + [limit], + ); + return result.rows.map((row) => Object.freeze({ + edgeRegistrationId: `edge:${row.id}`, + endpoint: row.channel_endpoint, + servername: row.channel_servername, + channelGeneration: row.channel_generation_ref, + trustBundleRef: row.channel_trust_bundle_ref, + certificateIdentities: Object.freeze( + [...(row.channel_certificate_identities ?? [])], + ), + lifecycleState: row.channel_lifecycle_state, + })); + }); + } + async #executeRead(operation) { const client = await this.pool.connect(); try { diff --git a/device-plane/services/device-control-core/src/server.mjs b/device-plane/services/device-control-core/src/server.mjs index 0c9d50b..c4feac2 100644 --- a/device-plane/services/device-control-core/src/server.mjs +++ b/device-plane/services/device-control-core/src/server.mjs @@ -1,7 +1,17 @@ -import { readFile } from "node:fs/promises"; +import { + createPrivateKey, + createPublicKey, + timingSafeEqual, + X509Certificate, +} from "node:crypto"; +import { lstat, readFile } from "node:fs/promises"; import { createControlCoreApp } from "./app.mjs"; import { resolveDeviceDatabaseUrl } from "./database-config.mjs"; +import { + createDeviceEdgeChannelSupervisor, +} from "./edge-channel-supervisor.mjs"; +import { createDeviceGatewayIngest } from "./gateway-ingest.mjs"; import { PostgresDeviceRepository } from "./postgres-repository.mjs"; const config = await readConfig(); @@ -12,6 +22,24 @@ const repository = new PostgresDeviceRepository({ await repository.migrate(); +const gatewayIngest = config.discoveryIngestEnabled || config.edgeChannelEnabled + ? createDeviceGatewayIngest({ + repository, + identifierPepper: config.identifierPepper, + }) + : null; +const edgeChannelSupervisor = config.edgeChannelEnabled + ? createDeviceEdgeChannelSupervisor({ + repository, + gatewayIngest, + coreIdentity: config.edgeChannelCoreIdentity, + trustRoot: config.edgeChannelTrustRoot, + maxEdges: config.edgeChannelMaxEdges, + reconcileIntervalMs: config.edgeChannelReconcileIntervalMs, + }) + : null; +await edgeChannelSupervisor?.start(); + const server = createControlCoreApp({ repository, gatewayToken: config.gatewayToken, @@ -19,6 +47,10 @@ const server = createControlCoreApp({ discoveryIngestEnabled: config.discoveryIngestEnabled, managementApiEnabled: config.managementApiEnabled, managementToken: config.managementToken, + gatewayIngest, + edgeChannelStatusProvider: edgeChannelSupervisor + ? () => edgeChannelSupervisor.status() + : null, }); server.listen(config.port, config.host, () => { @@ -28,6 +60,7 @@ server.listen(config.port, config.host, () => { port: config.port, discoveryIngest: config.discoveryIngestEnabled, managementApi: config.managementApiEnabled, + edgeChannels: config.edgeChannelEnabled, commandTransport: "disabled", })); }); @@ -37,6 +70,7 @@ process.on("SIGINT", shutdown); async function shutdown() { server.close(async () => { + await edgeChannelSupervisor?.stop(); await repository.close(); process.exit(0); }); @@ -51,6 +85,13 @@ async function readConfig() { process.env.DEVICE_MANAGEMENT_API_ENABLED, false, ); + const edgeChannelEnabled = parseBoolean( + process.env.DEVICE_EDGE_CHANNEL_ENABLED, + false, + ); + const edgeChannelCoreIdentity = edgeChannelEnabled + ? await readCoreIdentity(process.env) + : null; return { host: String(process.env.HOST || "127.0.0.1").trim(), port: parsePort(process.env.PORT, 18120), @@ -61,13 +102,15 @@ async function readConfig() { ), discoveryIngestEnabled, managementApiEnabled, + edgeChannelEnabled, gatewayToken: discoveryIngestEnabled ? await readRequiredSecretFile( process.env.DEVICE_GATEWAY_CORE_TOKEN_FILE, "device_gateway_core_token_file_required", ) : "", - identifierPepper: discoveryIngestEnabled || managementApiEnabled + identifierPepper: + discoveryIngestEnabled || managementApiEnabled || edgeChannelEnabled ? await readRequiredSecretFile( process.env.DEVICE_IDENTIFIER_PEPPER_FILE, "device_identifier_pepper_file_required", @@ -79,9 +122,82 @@ async function readConfig() { "device_management_core_token_file_required", ) : "", + edgeChannelCoreIdentity, + edgeChannelTrustRoot: edgeChannelEnabled + ? await readRequiredDirectory( + process.env.DEVICE_EDGE_CHANNEL_TRUST_ROOT, + "device_edge_channel_trust_root_required", + ) + : "", + edgeChannelMaxEdges: parseBoundedInt( + process.env.DEVICE_EDGE_CHANNEL_MAX_EDGES, + 32, + 1, + 64, + ), + edgeChannelReconcileIntervalMs: parseBoundedInt( + process.env.DEVICE_EDGE_CHANNEL_RECONCILE_INTERVAL_MS, + 15_000, + 1_000, + 300_000, + ), }; } +async function readCoreIdentity(environment) { + const keyPath = requiredValue( + environment.DEVICE_EDGE_CHANNEL_CORE_KEY_FILE, + "device_edge_channel_core_key_file_required", + ); + const certificatePath = requiredValue( + environment.DEVICE_EDGE_CHANNEL_CORE_CERTIFICATE_FILE, + "device_edge_channel_core_certificate_file_required", + ); + const [key, cert] = await Promise.all([ + readBoundedRegularFile(keyPath, 32 * 1024), + readBoundedRegularFile(certificatePath, 32 * 1024), + ]); + const privatePublic = createPublicKey(createPrivateKey(key)).export({ + type: "spki", + format: "der", + }); + const certificatePublic = new X509Certificate(cert).publicKey.export({ + type: "spki", + format: "der", + }); + if ( + privatePublic.length !== certificatePublic.length + || !timingSafeEqual(privatePublic, certificatePublic) + ) { + throw new Error("device_edge_channel_core_identity_mismatch"); + } + return Object.freeze({ + identityRef: "workload:device-control-core", + key, + cert, + }); +} + +async function readBoundedRegularFile(path, maximumBytes) { + const state = await lstat(path); + if ( + state.isSymbolicLink() + || !state.isFile() + || state.size < 1 + || state.size > maximumBytes + ) { + throw new Error("device_edge_channel_core_identity_file_invalid"); + } + return readFile(path); +} + +async function readRequiredDirectory(path, errorCode) { + const normalized = requiredValue(path, errorCode); + const state = await lstat(normalized); + if (state.isSymbolicLink() || !state.isDirectory()) throw new Error(errorCode); + return normalized; +} + async function readRequiredSecretFile(path, errorCode) { const normalized = requiredValue(path, errorCode); const value = (await readFile(normalized, "utf8")).trim(); @@ -112,6 +228,18 @@ function parsePositiveInt(value, fallback) { return parsed; } +function parseBoundedInt(value, fallback, minimum, maximum) { + const parsed = Number(value ?? fallback); + if ( + !Number.isSafeInteger(parsed) + || parsed < minimum + || parsed > maximum + ) { + throw new Error("device_bounded_integer_invalid"); + } + return parsed; +} + function parseBoolean(value, fallback) { if (value === undefined || value === null || value === "") return fallback; const normalized = String(value).trim().toLowerCase(); diff --git a/device-plane/services/device-control-core/test/app.test.mjs b/device-plane/services/device-control-core/test/app.test.mjs index 21f91b6..cf448ad 100644 --- a/device-plane/services/device-control-core/test/app.test.mjs +++ b/device-plane/services/device-control-core/test/app.test.mjs @@ -27,6 +27,12 @@ test("health reports database readiness and disabled command transport", async ( database: "ready", discoveryIngest: "disabled", managementApi: "disabled", + edgeChannels: { + enabled: false, + configured: 0, + accepted: 0, + degraded: 0, + }, commandTransport: "disabled", }); } finally { diff --git a/device-plane/services/device-control-core/test/device-edge-channel-migration.test.mjs b/device-plane/services/device-control-core/test/device-edge-channel-migration.test.mjs new file mode 100644 index 0000000..b82f561 --- /dev/null +++ b/device-plane/services/device-control-core/test/device-edge-channel-migration.test.mjs @@ -0,0 +1,22 @@ +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import test from "node:test"; + +const migrationUrl = new URL( + "../migrations/013_device_edge_channels.sql", + import.meta.url, +); + +test("Edge channel migration extends the canonical Edge without storing keys", async () => { + const sql = await readFile(migrationUrl, "utf8"); + + assert.match(sql, /alter table device_edges/); + assert.match(sql, /channel_endpoint text/); + assert.match(sql, /channel_generation_ref text/); + assert.match(sql, /channel_trust_bundle_ref text/); + assert.match(sql, /channel_certificate_identities jsonb/); + assert.match(sql, /channel_lifecycle_state in \('disabled', 'active', 'revoked'\)/); + assert.match(sql, /where channel_lifecycle_state = 'active'/); + assert.doesNotMatch(sql, /private[_ ]?key/i); + assert.doesNotMatch(sql, /password/i); +}); diff --git a/device-plane/services/device-control-core/test/edge-channel-supervisor.test.mjs b/device-plane/services/device-control-core/test/edge-channel-supervisor.test.mjs new file mode 100644 index 0000000..112fea5 --- /dev/null +++ b/device-plane/services/device-control-core/test/edge-channel-supervisor.test.mjs @@ -0,0 +1,115 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + createDeviceEdgeChannelSupervisor, +} from "../src/edge-channel-supervisor.mjs"; + +test("supervisor reconciles one in-process client per active Edge", async () => { + let registrations = [registration("channel-generation:1")]; + const clients = []; + const supervisor = createDeviceEdgeChannelSupervisor({ + repository: { + listActiveEdgeChannelRegistrations: async () => registrations, + }, + gatewayIngest: { + observeDiscovery: async () => ({}), + acceptMessage: async () => ({}), + }, + coreIdentity: { + identityRef: "workload:device-control-core", + key: "test-key", + cert: "test-cert", + }, + readPeerTrust: async () => "test-edge-certificate", + clientFactory: (options) => { + const state = { started: 0, stopped: 0, options }; + clients.push(state); + return { + async start() { state.started += 1; }, + async stop() { state.stopped += 1; }, + status: () => ({ channel: "accepted", lastErrorCode: null }), + }; + }, + reconcileIntervalMs: 300_000, + }); + + await supervisor.start(); + assert.equal(clients.length, 1); + assert.equal(supervisor.status().accepted, 1); + assert.equal(clients[0].options.registration.channelGeneration, "channel-generation:1"); + + await supervisor.reconcile(); + assert.equal(clients.length, 1); + + registrations = [registration("channel-generation:2")]; + await supervisor.reconcile(); + assert.equal(clients.length, 2); + assert.equal(clients[0].stopped, 1); + assert.equal(clients[1].started, 1); + + registrations = []; + await supervisor.reconcile(); + assert.equal(clients[1].stopped, 1); + assert.equal(supervisor.status().configured, 0); + assert.equal(JSON.stringify(supervisor.status()).includes("155.212"), false); + await supervisor.stop(); +}); + +test("supervisor keeps a failed trust enrollment isolated from other Edges", async () => { + const supervisor = createDeviceEdgeChannelSupervisor({ + repository: { + listActiveEdgeChannelRegistrations: async () => [ + registration("channel-generation:1", "edge:good"), + registration("channel-generation:1", "edge:bad"), + ], + }, + gatewayIngest: { + observeDiscovery: async () => ({}), + acceptMessage: async () => ({}), + }, + coreIdentity: { + identityRef: "workload:device-control-core", + key: "test-key", + cert: "test-cert", + }, + readPeerTrust: async ({ registration: value }) => { + if (value.edgeRegistrationId === "edge:bad") { + throw new Error("device_edge_channel_trust_bundle_identity_mismatch"); + } + return "test-edge-certificate"; + }, + clientFactory: () => ({ + start: async () => undefined, + stop: async () => undefined, + status: () => ({ channel: "accepted", lastErrorCode: null }), + }), + reconcileIntervalMs: 300_000, + }); + + await supervisor.start(); + const status = supervisor.status(); + assert.equal(status.configured, 2); + assert.equal(status.accepted, 1); + assert.equal(status.degraded, 1); + assert.equal(status.commandTransport, "disabled"); + assert.equal(status.edges.find((item) => item.edgeRegistrationId === "edge:bad") + .lastErrorCode, "device_edge_channel_trust_bundle_identity_mismatch"); + await supervisor.stop(); +}); + +function registration(channelGeneration, edgeRegistrationId = "edge:pilot") { + return { + edgeRegistrationId, + endpoint: "https://155.212.211.15:8443/", + servername: "155.212.211.15", + channelGeneration, + trustBundleRef: "edge-trust:moscow-edge", + certificateIdentities: [{ + generationRef: "edge-identity:1", + fingerprint: "AA:".repeat(31) + "AA", + status: "active", + }], + lifecycleState: "active", + }; +} diff --git a/device-plane/services/device-control-core/test/gateway-ingest.test.mjs b/device-plane/services/device-control-core/test/gateway-ingest.test.mjs new file mode 100644 index 0000000..148689f --- /dev/null +++ b/device-plane/services/device-control-core/test/gateway-ingest.test.mjs @@ -0,0 +1,87 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + DEVICE_ADAPTER_MESSAGE_SCHEMA, + DEVICE_DISCOVERY_SIGNAL_SCHEMA, +} from "../../../packages/device-protocol-contract/src/index.mjs"; +import { createDeviceGatewayIngest } from "../src/gateway-ingest.mjs"; + +const identifierPepper = "test-only-identifier-pepper-with-32-bytes"; +const rawImei = "000000000000001"; + +test("shared gateway ingest masks identifiers for HTTP and Edge callers", async () => { + const stored = []; + const ingest = createDeviceGatewayIngest({ + identifierPepper, + repository: { + async upsertQuarantineDiscovery(value) { + stored.push(value); + return { + created: true, + value: { ...value.safeView, discoveryRef: "discovery:test" }, + }; + }, + async acceptAdapterMessage(value) { + stored.push(value); + return { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }; + }, + }, + }); + + const discovery = await ingest.observeDiscovery(discoverySignal()); + const acceptance = await ingest.acceptMessage(adapterMessage()); + + assert.equal(discovery.value.identifier.masked, "***********0001"); + assert.equal(acceptance.status, "accepted"); + assert.match(stored[0].identifierDigest, /^hmac-sha256:[a-f0-9]{64}$/); + assert.match(stored[1].requestDigest, /^sha256:[a-f0-9]{64}$/); + assert.equal(JSON.stringify(stored).includes(rawImei), false); +}); + +function discoverySignal() { + return { + schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA, + sessionRef: "session:test", + modelProfileRef: "arusnavi.b2.internal.v1", + protocol: "INTERNAL", + observedAt: "2026-08-11T12:00:00.000Z", + identifier: { kind: "imei", value: rawImei }, + evidence: { + transport: "tcp", + bytesObserved: 16, + framingStatus: "verified", + specificationRef: "arusnavi.internal.framing.test-v1", + }, + }; +} + +function adapterMessage() { + return { + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: "edge:test", + adapterRef: "arusnavi-b2", + protocolProfileRef: "arusnavi.b2.internal.v1", + protocol: "INTERNAL", + sessionRef: "session:test", + messageRef: "package:1:test", + messageType: "telemetry.package", + sequence: 1, + observedAt: "2026-08-11T12:00:00.000Z", + idempotencyKey: `sha256:${"a".repeat(64)}`, + identifier: { kind: "imei", value: rawImei }, + payloadSchemaRef: "arusnavi.internal.package-metadata.v1", + payload: { + packageNumber: 1, + packetCount: 1, + packageDigest: `sha256:${"b".repeat(64)}`, + }, + }; +} diff --git a/device-plane/services/device-control-core/test/infrastructure-management.test.mjs b/device-plane/services/device-control-core/test/infrastructure-management.test.mjs index da3161e..8102672 100644 --- a/device-plane/services/device-control-core/test/infrastructure-management.test.mjs +++ b/device-plane/services/device-control-core/test/infrastructure-management.test.mjs @@ -143,6 +143,62 @@ test("shared catalog and Edge authority requires the Hub owner ceiling", () => { ); }); +test("normalizes only a pinned Core-initiated public Edge channel", () => { + const command = normalizeInfrastructureManagementCommand("edge.ensure", { + edgeKey: "moscow-edge", + displayName: "Moscow Edge", + deploymentRef: "deployment:device-edge/moscow-1", + lifecycleState: "active", + channel: { + endpoint: "https://155.212.211.15:8443/", + servername: "155.212.211.15", + generationRef: "channel-generation:1", + trustBundleRef: "edge-trust:moscow-edge", + certificateIdentities: [{ + generationRef: "edge-identity:1", + fingerprint: "AA:".repeat(31) + "AA", + status: "active", + }], + lifecycleState: "active", + }, + }); + + assert.equal(command.channel.endpoint, "https://155.212.211.15:8443/"); + assert.equal(command.channel.lifecycleState, "active"); + assert.equal(command.channel.certificateIdentities.length, 1); + for (const endpoint of [ + "https://127.0.0.1:8443/", + "https://192.168.1.1:8443/", + "https://155.212.211.15:9921/", + "http://155.212.211.15:8443/", + ]) { + assert.throws( + () => normalizeInfrastructureManagementCommand("edge.ensure", { + edgeKey: "bad-edge", + displayName: "Bad Edge", + channel: { ...command.channel, endpoint }, + }), + /device_edge_channel_endpoint_invalid/, + ); + } + assert.throws( + () => normalizeInfrastructureManagementCommand("edge.ensure", { + edgeKey: "bad-edge", + displayName: "Bad Edge", + channel: { ...command.channel, servername: "example.invalid" }, + }), + /device_edge_channel_servername_mismatch/, + ); + assert.throws( + () => normalizeInfrastructureManagementCommand("edge.ensure", { + edgeKey: "bad-edge", + displayName: "Bad Edge", + channel: { lifecycleState: "disabled", endpoint: command.channel.endpoint }, + }), + /device_edge_channel_disabled_configuration_invalid/, + ); +}); + function actor(hubRole) { return normalizeManagementActor({ userRef: "user:platform-admin", diff --git a/device-plane/services/device-control-core/test/infrastructure-repository.test.mjs b/device-plane/services/device-control-core/test/infrastructure-repository.test.mjs index aa3c08f..8159d54 100644 --- a/device-plane/services/device-control-core/test/infrastructure-repository.test.mjs +++ b/device-plane/services/device-control-core/test/infrastructure-repository.test.mjs @@ -192,6 +192,48 @@ test("stores enrollment digest but returns and audits only its masked projection assert.equal(client.released, true); }); +test("lists only bounded active Edge channel registrations without secrets", async () => { + const client = scriptedClient([ + step("begin transaction read only"), + step("from device_edges", { + rows: [{ + id: edgeId, + channel_endpoint: "https://155.212.211.15:8443/", + channel_servername: "155.212.211.15", + channel_generation_ref: "channel-generation:1", + channel_trust_bundle_ref: "edge-trust:moscow-edge", + channel_certificate_identities: [{ + generationRef: "edge-identity:1", + fingerprint: "AA:".repeat(31) + "AA", + status: "active", + }], + channel_lifecycle_state: "active", + }], + }), + step("commit"), + ]); + const repository = repositoryWithClient(client); + + const registrations = await repository.listActiveEdgeChannelRegistrations(8); + + assert.deepEqual(registrations[0], { + edgeRegistrationId: `edge:${edgeId}`, + endpoint: "https://155.212.211.15:8443/", + servername: "155.212.211.15", + channelGeneration: "channel-generation:1", + trustBundleRef: "edge-trust:moscow-edge", + certificateIdentities: [{ + generationRef: "edge-identity:1", + fingerprint: "AA:".repeat(31) + "AA", + status: "active", + }], + lifecycleState: "active", + }); + assert.equal(JSON.stringify(registrations).includes("private"), false); + assert.equal(client.remaining(), 0); + assert.equal(client.released, true); +}); + function managementActor(hubRole) { return normalizeManagementActor({ userRef: "user:platform-owner", diff --git a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs index 2de6c24..25881e5 100644 --- a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs +++ b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs @@ -173,6 +173,24 @@ test("rejects a revoked Edge registration before opening a channel", async () => } }); +test("rejects an Edge hello from an unexpected channel generation", async () => { + const edge = createEdgeServer({ + channelGeneration: "generation:edge-unexpected", + }); + const address = await edge.start(); + const core = createCoreClient({ address }); + try { + await core.start(); + await waitFor(() => core.status().protocolFailures >= 1, 2_000); + assert.notEqual(core.status().channel, "accepted"); + assert.match(core.status().lastErrorCode, /channel_generation_mismatch/); + assert.equal(edge.status().channelsAccepted, 0); + } finally { + await core.stop(); + await edge.stop(); + } +}); + test("rejects an authenticated but non-allowlisted Core certificate", async () => { const edge = createEdgeServer(); const address = await edge.start(); @@ -370,7 +388,7 @@ function createEdgeServer(options = {}) { const edgeCertificate = options.edgeCertificate ?? certificates.edge; return createDeviceEdgeChannelServer({ edgeRegistrationId: "edge:pilot-1", - channelGeneration: "generation:pilot-1", + channelGeneration: options.channelGeneration ?? "generation:pilot-1", trustGeneration: options.edgeTrustGeneration ?? "trust-generation:1", host: "127.0.0.1", port: 0, @@ -427,6 +445,7 @@ function createCoreClient(options) { function edgeRegistration(address, certificateIdentities, lifecycleState = "active") { return { edgeRegistrationId: "edge:pilot-1", + channelGeneration: "generation:pilot-1", endpoint: `https://127.0.0.1:${address.port}/`, servername: "localhost", certificateIdentities, diff --git a/device-plane/services/device-gateway-core/src/runtime.mjs b/device-plane/services/device-gateway-core/src/runtime.mjs index 6373980..f07e92f 100644 --- a/device-plane/services/device-gateway-core/src/runtime.mjs +++ b/device-plane/services/device-gateway-core/src/runtime.mjs @@ -224,6 +224,11 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { if (envelope.messageKind !== "channel.hello") { throw new Error("device_gateway_core_channel_hello_required"); } + if ( + envelope.channelGeneration !== connection.registration.channelGeneration + ) { + throw new Error("device_gateway_core_channel_generation_mismatch"); + } if ( envelope.payload?.status !== "ready" || envelope.payload?.transport !== "http2-mtls" @@ -233,7 +238,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { ) { throw new Error("device_gateway_core_channel_hello_invalid"); } - connection.channelGeneration = envelope.channelGeneration; + connection.channelGeneration = connection.registration.channelGeneration; send(connection, "channel.accepted", { status: "accepted", coreIdentity: config.coreIdentity, @@ -549,6 +554,10 @@ function normalizeRegistration(value) { value.edgeRegistrationId, "edge_registration_id", ), + channelGeneration: normalizeRef( + value.channelGeneration, + "channel_generation", + ), endpoint: endpoint.toString(), servername, certificateIdentities: normalizeCertificateIdentities(