fix(device-core): bind edge telemetry to enrolled route

2026-08-13 09:41:14 +03:00
parent fe1c8054be
commit 71fff82c99
7 changed files with 309 additions and 10 deletions
@@ -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",
@@ -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")
@@ -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;
+}
@@ -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,
@@ -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);
@@ -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,
@@ -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",
+  };
+}