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 index dfacaba..38bc08e 100644 --- a/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs +++ b/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs @@ -108,8 +108,14 @@ export function createDeviceEdgeChannelSupervisor(options = {}) { ca, }, coreIdentity: config.coreIdentity.identityRef, - observeDiscovery: config.gatewayIngest.observeDiscovery, - acceptMessage: config.gatewayIngest.acceptMessage, + observeDiscovery: (signal) => config.gatewayIngest.observeDiscovery( + signal, + { authenticatedEdgeRef: registration.edgeRegistrationId }, + ), + acceptMessage: (message) => config.gatewayIngest.acceptMessage( + message, + { authenticatedEdgeRef: registration.edgeRegistrationId }, + ), commandTransport: config.typedCommandRuntime ? "typed-service-ping-v1" : "disabled", diff --git a/device-plane/services/device-control-core/src/gateway-ingest.mjs b/device-plane/services/device-control-core/src/gateway-ingest.mjs index 890dacd..6e4e158 100644 --- a/device-plane/services/device-control-core/src/gateway-ingest.mjs +++ b/device-plane/services/device-control-core/src/gateway-ingest.mjs @@ -22,12 +22,26 @@ export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) } return Object.freeze({ - async observeDiscovery(input) { - const signal = normalizeDiscoverySignal(input); + async observeDiscovery(input, context = {}) { + const receivedSignal = normalizeDiscoverySignal(input); const identifierDigest = hashRestrictedIdentifier( - signal.identifier, + receivedSignal.identifier, identifierPepper, ); + const routeRef = await resolveAuthenticatedRoute(repository, { + edgeRef: context.authenticatedEdgeRef, + modelProfileRef: receivedSignal.modelProfileRef, + protocol: receivedSignal.protocol, + identifierKind: receivedSignal.identifier.kind, + identifierDigest, + observedAt: receivedSignal.observedAt, + }); + const signal = routeRef === undefined + ? receivedSignal + : normalizeDiscoverySignal({ + ...withoutKeys(receivedSignal, ["routeRef"]), + ...(routeRef ? { routeRef } : {}), + }); const safeView = assertSafeProjection(toSafeDiscoveryView(signal)); const discovery = await repository.upsertQuarantineDiscovery({ identifierDigest, @@ -42,12 +56,27 @@ export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) }); }, - async acceptMessage(input) { - const message = normalizeAdapterMessage(input); + async acceptMessage(input, context = {}) { + const receivedMessage = normalizeAdapterMessage(input); const identifierDigest = hashRestrictedIdentifier( - message.identifier, + receivedMessage.identifier, identifierPepper, ); + const routeRef = await resolveAuthenticatedRoute(repository, { + edgeRef: context.authenticatedEdgeRef, + modelProfileRef: receivedMessage.protocolProfileRef, + protocol: receivedMessage.protocol, + identifierKind: receivedMessage.identifier.kind, + identifierDigest, + observedAt: receivedMessage.observedAt, + }); + const message = routeRef === undefined + ? receivedMessage + : normalizeAdapterMessage({ + ...withoutKeys(receivedMessage, ["edgeRef", "routeRef"]), + edgeRef: context.authenticatedEdgeRef, + ...(routeRef ? { routeRef } : {}), + }); const safeView = assertSafeProjection(toSafeAdapterMessageView(message)); const requestDigest = gatewayMessageRequestDigest({ edgeRef: safeView.edgeRef, @@ -74,6 +103,21 @@ export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) }); } +async function resolveAuthenticatedRoute(repository, input) { + if (input.edgeRef == null) return undefined; + if (typeof repository.resolveInboundRoute !== "function") { + throw new TypeError("device_inbound_route_repository_required"); + } + return repository.resolveInboundRoute(input); +} + +function withoutKeys(value, keys) { + const omitted = new Set(keys); + return Object.fromEntries( + Object.entries(value).filter(([key]) => !omitted.has(key)), + ); +} + function gatewayMessageRequestDigest(value) { return `sha256:${createHash("sha256") .update(JSON.stringify(value), "utf8") diff --git a/device-plane/services/device-control-core/src/inbound-route-repository.mjs b/device-plane/services/device-control-core/src/inbound-route-repository.mjs new file mode 100644 index 0000000..029477b --- /dev/null +++ b/device-plane/services/device-control-core/src/inbound-route-repository.mjs @@ -0,0 +1,111 @@ +export async function resolveInboundRoute(client, input = {}) { + if (!client || typeof client.query !== "function") { + throw new TypeError("device_inbound_route_client_required"); + } + const edgeId = parseEntityRef(input.edgeRef, "edge"); + const modelProfileRef = normalizeOpaqueRef( + input.modelProfileRef, + "model_profile_ref", + ); + const protocol = normalizeUpperToken(input.protocol, "protocol"); + const identifierKind = normalizeLowerToken( + input.identifierKind, + "identifier_kind", + ); + const identifierDigest = normalizeIdentifierDigest(input.identifierDigest); + const observedAt = normalizeTimestamp(input.observedAt, "observed_at"); + + const result = await client.query( + `select r.id + from device_enrollment_intents ei + join device_routes r + on r.id = ei.route_id + and r.project_id = ei.project_id + and r.model_profile_ref = ei.model_profile_ref + join device_edges e on e.id = r.edge_id + where r.edge_id = $1 + and r.model_profile_ref = $2 + and r.protocol = $3 + and r.lifecycle_state = 'active' + and e.lifecycle_state = 'active' + and e.channel_lifecycle_state = 'active' + and ei.expected_identifier_kind = $4 + and ei.expected_identifier_digest = $5 + and ei.lifecycle_state in ('pending', 'observed', 'claimed') + and ( + ei.lifecycle_state = 'claimed' + or ei.expires_at is null + or ei.expires_at > $6 + ) + order by r.id + limit 2`, + [ + edgeId, + modelProfileRef, + protocol, + identifierKind, + identifierDigest, + observedAt, + ], + ); + if (result.rows.length > 1) { + throw domainError("device_inbound_route_ambiguous", 409); + } + return result.rows[0]?.id ? `route:${result.rows[0].id}` : null; +} + +function parseEntityRef(value, prefix) { + if (typeof value !== "string") { + throw new TypeError(`device_${prefix}_ref_invalid`); + } + const match = value.match(new RegExp( + `^${prefix}:([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$`, + "i", + )); + if (!match) throw new TypeError(`device_${prefix}_ref_invalid`); + return match[1].toLowerCase(); +} + +function normalizeOpaqueRef(value, name) { + if ( + typeof value !== "string" + || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value) + ) { + throw new TypeError(`device_inbound_route_${name}_invalid`); + } + return value; +} + +function normalizeUpperToken(value, name) { + if (typeof value !== "string" || !/^[A-Z][A-Z0-9_]{0,31}$/.test(value)) { + throw new TypeError(`device_inbound_route_${name}_invalid`); + } + return value; +} + +function normalizeLowerToken(value, name) { + if (typeof value !== "string" || !/^[a-z][a-z0-9._:-]{1,63}$/.test(value)) { + throw new TypeError(`device_inbound_route_${name}_invalid`); + } + return value; +} + +function normalizeIdentifierDigest(value) { + if (typeof value !== "string" || !/^hmac-sha256:[a-f0-9]{64}$/.test(value)) { + throw new TypeError("device_inbound_route_identifier_digest_invalid"); + } + return value; +} + +function normalizeTimestamp(value, name) { + if (typeof value !== "string" || Number.isNaN(Date.parse(value))) { + throw new TypeError(`device_inbound_route_${name}_invalid`); + } + return value; +} + +function domainError(code, statusCode) { + const error = new Error(code); + error.statusCode = statusCode; + return error; +} 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 a3efb76..c4fcf75 100644 --- a/device-plane/services/device-control-core/src/postgres-repository.mjs +++ b/device-plane/services/device-control-core/src/postgres-repository.mjs @@ -7,6 +7,7 @@ import pg from "pg"; import { observeQuarantineDiscovery } from "./discovery-repository.mjs"; import { acceptGatewayMessage } from "./gateway-message-repository.mjs"; +import { resolveInboundRoute } from "./inbound-route-repository.mjs"; import { applyControlResourceManagementCommand, authorizeControlResourceManagementReplay, @@ -116,6 +117,10 @@ export class PostgresDeviceRepository { }); } + async resolveInboundRoute(input) { + return this.#executeRead((client) => resolveInboundRoute(client, input)); + } + async executeManagementCommand({ idempotencyKey, commandKind, 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 index e2abfab..e970c4f 100644 --- 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 @@ -8,13 +8,14 @@ import { test("supervisor reconciles one in-process client per active Edge", async () => { let registrations = [registration("channel-generation:1")]; const clients = []; + const ingestCalls = []; const supervisor = createDeviceEdgeChannelSupervisor({ repository: { listActiveEdgeChannelRegistrations: async () => registrations, }, gatewayIngest: { - observeDiscovery: async () => ({}), - acceptMessage: async () => ({}), + observeDiscovery: async (...args) => ingestCalls.push(["discovery", ...args]), + acceptMessage: async (...args) => ingestCalls.push(["message", ...args]), }, coreIdentity: { identityRef: "workload:device-control-core", @@ -38,6 +39,20 @@ test("supervisor reconciles one in-process client per active Edge", async () => assert.equal(clients.length, 1); assert.equal(supervisor.status().accepted, 1); assert.equal(clients[0].options.registration.channelGeneration, "channel-generation:1"); + await clients[0].options.observeDiscovery({ signal: true }); + await clients[0].options.acceptMessage({ message: true }); + assert.deepEqual(ingestCalls, [ + [ + "discovery", + { signal: true }, + { authenticatedEdgeRef: "edge:pilot" }, + ], + [ + "message", + { message: true }, + { authenticatedEdgeRef: "edge:pilot" }, + ], + ]); await supervisor.reconcile(); assert.equal(clients.length, 1); 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 index 64bfb75..c851b1c 100644 --- a/device-plane/services/device-control-core/test/gateway-ingest.test.mjs +++ b/device-plane/services/device-control-core/test/gateway-ingest.test.mjs @@ -53,6 +53,55 @@ test("shared gateway ingest masks identifiers for HTTP and Edge callers", async assert.equal(JSON.stringify(stored).includes(rawImei), false); }); +test("authenticated Edge identity resolves the allowlisted project route", async () => { + const stored = []; + const resolutions = []; + const authenticatedEdgeRef = "edge:11111111-1111-4111-8111-111111111111"; + const routeRef = "route:22222222-2222-4222-8222-222222222222"; + const ingest = createDeviceGatewayIngest({ + identifierPepper, + repository: { + async resolveInboundRoute(value) { + resolutions.push(value); + return routeRef; + }, + async upsertQuarantineDiscovery(value) { + stored.push(value); + return { + created: false, + value: { ...value.safeView, discoveryRef: "discovery:test" }, + }; + }, + async acceptAdapterMessage(value) { + stored.push(value); + return { + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, + claimedDeviceRef: null, + }; + }, + }, + }); + + await ingest.observeDiscovery(discoverySignal(), { authenticatedEdgeRef }); + await ingest.acceptMessage(adapterMessage(), { authenticatedEdgeRef }); + + assert.equal(resolutions.length, 2); + assert.equal(resolutions[0].edgeRef, authenticatedEdgeRef); + assert.match(resolutions[0].identifierDigest, /^hmac-sha256:[a-f0-9]{64}$/); + assert.equal(stored[0].safeView.routeRef, routeRef); + assert.equal(stored[1].safeView.routeRef, routeRef); + assert.equal(stored[1].safeView.edgeRef, authenticatedEdgeRef); + assert.notEqual(stored[1].safeView.edgeRef, adapterMessage().edgeRef); + assert.equal(JSON.stringify(resolutions).includes(rawImei), false); +}); + function discoverySignal() { return { schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA, diff --git a/device-plane/services/device-control-core/test/inbound-route-repository.test.mjs b/device-plane/services/device-control-core/test/inbound-route-repository.test.mjs new file mode 100644 index 0000000..1d73a07 --- /dev/null +++ b/device-plane/services/device-control-core/test/inbound-route-repository.test.mjs @@ -0,0 +1,69 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + resolveInboundRoute, +} from "../src/inbound-route-repository.mjs"; + +const edgeRef = "edge:11111111-1111-4111-8111-111111111111"; +const routeId = "22222222-2222-4222-8222-222222222222"; +const identifierDigest = `hmac-sha256:${"a".repeat(64)}`; + +test("inbound route resolution is scoped by authenticated Edge and enrollment", async () => { + const queries = []; + const client = { + async query(sql, values) { + queries.push({ sql, values }); + return { rows: [{ id: routeId }] }; + }, + }; + + const result = await resolveInboundRoute(client, input()); + + assert.equal(result, `route:${routeId}`); + assert.equal(queries.length, 1); + assert.match(queries[0].sql, /device_enrollment_intents/); + assert.match(queries[0].sql, /r\.edge_id = \$1/); + assert.match(queries[0].sql, /ei\.expected_identifier_digest = \$5/); + assert.deepEqual(queries[0].values, [ + edgeRef.slice("edge:".length), + "arusnavi.b2.internal.v1", + "INTERNAL", + "imei", + identifierDigest, + "2026-08-13T09:00:00.000Z", + ]); +}); + +test("inbound route resolution leaves unknown identifiers quarantined", async () => { + const result = await resolveInboundRoute( + { query: async () => ({ rows: [] }) }, + input(), + ); + assert.equal(result, null); +}); + +test("inbound route resolution fails closed on ambiguous ownership", async () => { + await assert.rejects( + () => resolveInboundRoute( + { query: async () => ({ rows: [{ id: routeId }, { id: routeId }] }) }, + input(), + ), + (error) => { + assert.equal(error.message, "device_inbound_route_ambiguous"); + assert.equal(error.statusCode, 409); + return true; + }, + ); +}); + +function input() { + return { + edgeRef, + modelProfileRef: "arusnavi.b2.internal.v1", + protocol: "INTERNAL", + identifierKind: "imei", + identifierDigest, + observedAt: "2026-08-13T09:00:00.000Z", + }; +}