diff --git a/device-plane/package-lock.json b/device-plane/package-lock.json index 9f5d91e..f6466c2 100644 --- a/device-plane/package-lock.json +++ b/device-plane/package-lock.json @@ -19,6 +19,14 @@ "resolved": "packages/arusnavi-b2-adapter", "link": true }, + "node_modules/@nodedc/device-adapter-catalog": { + "resolved": "packages/device-adapter-catalog", + "link": true + }, + "node_modules/@nodedc/device-adapter-runtime": { + "resolved": "packages/device-adapter-runtime", + "link": true + }, "node_modules/@nodedc/device-control-core": { "resolved": "services/device-control-core", "link": true @@ -188,6 +196,20 @@ "node": ">=20" } }, + "packages/device-adapter-catalog": { + "name": "@nodedc/device-adapter-catalog", + "version": "0.1.0", + "engines": { + "node": ">=20" + } + }, + "packages/device-adapter-runtime": { + "name": "@nodedc/device-adapter-runtime", + "version": "0.1.0", + "engines": { + "node": ">=20" + } + }, "packages/device-protocol-contract": { "name": "@nodedc/device-protocol-contract", "version": "0.1.0", diff --git a/device-plane/packages/arusnavi-b2-adapter/src/index.mjs b/device-plane/packages/arusnavi-b2-adapter/src/index.mjs index 79dd9ab..e74db69 100644 --- a/device-plane/packages/arusnavi-b2-adapter/src/index.mjs +++ b/device-plane/packages/arusnavi-b2-adapter/src/index.mjs @@ -1,3 +1,10 @@ +import { createHash } from "node:crypto"; + +import { + DEVICE_ADAPTER_CONTRACT_VERSION, + defineDeviceAdapter, +} from "../../device-adapter-runtime/src/index.mjs"; + export const ARUSNAVI_INTERNAL_SPECIFICATION_REF = "arusnavi.internal.protocol-sheet.gid-12.v1"; @@ -53,6 +60,26 @@ export const ARUSNAVI_B2_MODEL_PROFILE = deepFreeze({ }, }); +export const ARUSNAVI_B2_ADAPTER = defineDeviceAdapter({ + contractVersion: DEVICE_ADAPTER_CONTRACT_VERSION, + adapterRef: "arusnavi-b2", + runtimePackageRef: "@nodedc/arusnavi-b2-adapter", + profiles: [ARUSNAVI_B2_MODEL_PROFILE], + createSession({ profileRef } = {}) { + if (profileRef !== ARUSNAVI_B2_MODEL_PROFILE.profileRef) { + throw new TypeError("b2_adapter_profile_unsupported"); + } + return { + parseHeader: tryParseB2Header2, + buildHeaderAcknowledgement: buildB2HeaderAcknowledgement, + parseMessage: tryParseB2Package, + buildMessageAcknowledgement(message) { + return buildB2PackageAcknowledgement(message.packageNumber); + }, + }; + }, +}); + export function tryParseB2Header2(input) { assertBuffer(input, "b2_header_buffer_required"); if (input.length < HEADER2_LENGTH) { @@ -148,6 +175,16 @@ export function tryParseB2Package(input) { bytesConsumed: offset + 1, packageNumber, packetCount, + messageType: "telemetry.package", + payloadSchemaRef: "arusnavi.internal.package-metadata.v1", + payload: Object.freeze({ + packageNumber, + packetCount, + byteLength: offset + 1, + packageDigest: `sha256:${createHash("sha256") + .update(input.subarray(0, offset + 1)) + .digest("hex")}`, + }), }); } if (input.length - offset < 3) { diff --git a/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs b/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs index 90cea35..5eada75 100644 --- a/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs +++ b/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs @@ -66,11 +66,22 @@ test("parses and acknowledges the official package example", () => { tryParseB2Package(specificationPackage.subarray(0, -1)).status, "incomplete", ); - assert.deepEqual(tryParseB2Package(specificationPackage), { - status: "complete", - bytesConsumed: specificationPackage.length, + const parsed = tryParseB2Package(specificationPackage); + assert.equal(parsed.status, "complete"); + assert.equal(parsed.bytesConsumed, specificationPackage.length); + assert.equal(parsed.packageNumber, 1); + assert.equal(parsed.packetCount, 1); + assert.equal(parsed.messageType, "telemetry.package"); + assert.equal( + parsed.payloadSchemaRef, + "arusnavi.internal.package-metadata.v1", + ); + assert.deepEqual(parsed.payload, { packageNumber: 1, packetCount: 1, + byteLength: specificationPackage.length, + packageDigest: + "sha256:bba7205d2f613ac3dfdb7bccdee292b3837e1c8d3d1254ee68bba2dda3853e11", }); assert.equal( buildB2PackageAcknowledgement(1).toString("hex").toUpperCase(), @@ -93,12 +104,14 @@ test("uses packet lengths and checksum instead of scanning for 0x5D", () => { checksum, 0x5d, ]); - assert.deepEqual(tryParseB2Package(packageBytes), { - status: "complete", - bytesConsumed: packageBytes.length, - packageNumber: 2, - packetCount: 1, - }); + const parsed = tryParseB2Package(packageBytes); + assert.equal(parsed.status, "complete"); + assert.equal(parsed.bytesConsumed, packageBytes.length); + assert.equal(parsed.packageNumber, 2); + assert.equal(parsed.packetCount, 1); + assert.equal(parsed.messageType, "telemetry.package"); + assert.equal(parsed.payload.byteLength, packageBytes.length); + assert.match(parsed.payload.packageDigest, /^sha256:[a-f0-9]{64}$/); }); test("fails closed on unsupported headers and malformed packages", () => { diff --git a/device-plane/packages/device-adapter-catalog/package.json b/device-plane/packages/device-adapter-catalog/package.json new file mode 100644 index 0000000..c7bddf9 --- /dev/null +++ b/device-plane/packages/device-adapter-catalog/package.json @@ -0,0 +1,15 @@ +{ + "name": "@nodedc/device-adapter-catalog", + "version": "0.1.0", + "private": true, + "type": "module", + "exports": { + ".": "./src/index.mjs" + }, + "scripts": { + "test": "node --test test/*.test.mjs" + }, + "engines": { + "node": ">=20" + } +} diff --git a/device-plane/packages/device-adapter-catalog/src/index.mjs b/device-plane/packages/device-adapter-catalog/src/index.mjs new file mode 100644 index 0000000..138cf35 --- /dev/null +++ b/device-plane/packages/device-adapter-catalog/src/index.mjs @@ -0,0 +1,13 @@ +import { + createDeviceAdapterRegistry, +} from "../../device-adapter-runtime/src/index.mjs"; +import { + ARUSNAVI_B2_ADAPTER, +} from "../../arusnavi-b2-adapter/src/index.mjs"; + +export const DEVICE_ADAPTER_CATALOG = Object.freeze({ + defaultProfileRef: "arusnavi.b2.internal.v1", + registry: createDeviceAdapterRegistry({ + adapters: [ARUSNAVI_B2_ADAPTER], + }), +}); diff --git a/device-plane/packages/device-adapter-catalog/test/catalog.test.mjs b/device-plane/packages/device-adapter-catalog/test/catalog.test.mjs new file mode 100644 index 0000000..8e3d897 --- /dev/null +++ b/device-plane/packages/device-adapter-catalog/test/catalog.test.mjs @@ -0,0 +1,17 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { DEVICE_ADAPTER_CATALOG } from "../src/index.mjs"; + +test("allowlists ARUSNAVI B2 as the first adapter without making Gateway vendor-specific", () => { + assert.deepEqual(DEVICE_ADAPTER_CATALOG.registry.adapterRefs, ["arusnavi-b2"]); + assert.deepEqual( + DEVICE_ADAPTER_CATALOG.registry.profileRefs, + ["arusnavi.b2.internal.v1"], + ); + const registration = DEVICE_ADAPTER_CATALOG.registry.resolveProfile( + DEVICE_ADAPTER_CATALOG.defaultProfileRef, + ); + assert.equal(registration.adapter.runtimePackageRef, "@nodedc/arusnavi-b2-adapter"); + assert.equal(registration.profile.model, "B2"); +}); diff --git a/device-plane/packages/device-adapter-runtime/package.json b/device-plane/packages/device-adapter-runtime/package.json new file mode 100644 index 0000000..eadc701 --- /dev/null +++ b/device-plane/packages/device-adapter-runtime/package.json @@ -0,0 +1,15 @@ +{ + "name": "@nodedc/device-adapter-runtime", + "version": "0.1.0", + "private": true, + "type": "module", + "exports": { + ".": "./src/index.mjs" + }, + "scripts": { + "test": "node --test test/*.test.mjs" + }, + "engines": { + "node": ">=20" + } +} diff --git a/device-plane/packages/device-adapter-runtime/src/index.mjs b/device-plane/packages/device-adapter-runtime/src/index.mjs new file mode 100644 index 0000000..9c99dc2 --- /dev/null +++ b/device-plane/packages/device-adapter-runtime/src/index.mjs @@ -0,0 +1,151 @@ +export const DEVICE_ADAPTER_CONTRACT_VERSION = + "nodedc.device-adapter.v1"; + +const ADAPTER_REF_RE = /^[a-z][a-z0-9-]{1,62}$/; +const PROFILE_REF_RE = /^[a-z][a-z0-9._-]{2,127}$/; +const RUNTIME_PACKAGE_REF_RE = /^@[a-z0-9-]+\/[a-z0-9-]+$/; + +export function defineDeviceAdapter(input) { + assertPlainObject(input, "device_adapter"); + if (input.contractVersion !== DEVICE_ADAPTER_CONTRACT_VERSION) { + throw new TypeError("device_adapter_contract_version_invalid"); + } + const adapterRef = normalizeRef( + input.adapterRef, + ADAPTER_REF_RE, + "device_adapter_ref_invalid", + ); + const runtimePackageRef = normalizeRef( + input.runtimePackageRef, + RUNTIME_PACKAGE_REF_RE, + "device_adapter_runtime_package_ref_invalid", + ); + if (!Array.isArray(input.profiles) || input.profiles.length === 0) { + throw new TypeError("device_adapter_profiles_required"); + } + const profiles = input.profiles.map((profile) => + normalizeProfile(profile, adapterRef) + ); + if (new Set(profiles.map((profile) => profile.profileRef)).size !== profiles.length) { + throw new TypeError("device_adapter_profile_duplicate"); + } + if (typeof input.createSession !== "function") { + throw new TypeError("device_adapter_session_factory_required"); + } + + return deepFreeze({ + contractVersion: DEVICE_ADAPTER_CONTRACT_VERSION, + adapterRef, + runtimePackageRef, + profiles, + createSession: input.createSession, + }); +} + +export function createDeviceAdapterRegistry({ adapters = [] } = {}) { + if (!Array.isArray(adapters)) { + throw new TypeError("device_adapter_registry_adapters_invalid"); + } + const byAdapterRef = new Map(); + const byProfileRef = new Map(); + for (const adapter of adapters) { + const normalized = defineDeviceAdapter(adapter); + if (byAdapterRef.has(normalized.adapterRef)) { + throw new TypeError("device_adapter_registry_adapter_duplicate"); + } + byAdapterRef.set(normalized.adapterRef, normalized); + for (const profile of normalized.profiles) { + if (byProfileRef.has(profile.profileRef)) { + throw new TypeError("device_adapter_registry_profile_duplicate"); + } + byProfileRef.set(profile.profileRef, Object.freeze({ + adapter: normalized, + profile, + })); + } + } + + return Object.freeze({ + adapterRefs: Object.freeze([...byAdapterRef.keys()].sort()), + profileRefs: Object.freeze([...byProfileRef.keys()].sort()), + getAdapter(adapterRef) { + const normalized = normalizeRef( + adapterRef, + ADAPTER_REF_RE, + "device_adapter_ref_invalid", + ); + const adapter = byAdapterRef.get(normalized); + if (!adapter) throw new TypeError("device_adapter_not_allowlisted"); + return adapter; + }, + resolveProfile(profileRef) { + const normalized = normalizeRef( + profileRef, + PROFILE_REF_RE, + "device_adapter_profile_ref_invalid", + ); + const registration = byProfileRef.get(normalized); + if (!registration) { + throw new TypeError("device_adapter_profile_not_allowlisted"); + } + return registration; + }, + }); +} + +export function assertDeviceAdapterSession(session) { + assertPlainObject(session, "device_adapter_session"); + for (const method of [ + "parseHeader", + "buildHeaderAcknowledgement", + "parseMessage", + "buildMessageAcknowledgement", + ]) { + if (typeof session[method] !== "function") { + throw new TypeError(`device_adapter_session_method_missing:${method}`); + } + } + return session; +} + +function normalizeProfile(profile, adapterRef) { + assertPlainObject(profile, "device_adapter_profile"); + const profileRef = normalizeRef( + profile.profileRef, + PROFILE_REF_RE, + "device_adapter_profile_ref_invalid", + ); + if (profile.adapterRef != null && profile.adapterRef !== adapterRef) { + throw new TypeError("device_adapter_profile_adapter_mismatch"); + } + const maxBufferedBytes = Number(profile?.framing?.maxBufferedBytes); + if ( + !Number.isSafeInteger(maxBufferedBytes) + || maxBufferedBytes < 1024 + || maxBufferedBytes > 256 * 1024 + ) { + throw new TypeError("device_adapter_profile_buffer_limit_invalid"); + } + return deepFreeze({ ...profile, profileRef, adapterRef }); +} + +function normalizeRef(value, pattern, errorCode) { + if (typeof value !== "string" || !pattern.test(value)) { + throw new TypeError(errorCode); + } + return value; +} + +function assertPlainObject(value, label) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError(`${label}_invalid`); + } +} + +function deepFreeze(value) { + if (!value || typeof value !== "object" || Object.isFrozen(value)) { + return value; + } + Object.values(value).forEach(deepFreeze); + return Object.freeze(value); +} diff --git a/device-plane/packages/device-adapter-runtime/test/registry.test.mjs b/device-plane/packages/device-adapter-runtime/test/registry.test.mjs new file mode 100644 index 0000000..eef97d9 --- /dev/null +++ b/device-plane/packages/device-adapter-runtime/test/registry.test.mjs @@ -0,0 +1,60 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + DEVICE_ADAPTER_CONTRACT_VERSION, + createDeviceAdapterRegistry, + defineDeviceAdapter, +} from "../src/index.mjs"; + +function adapter(adapterRef = "generic-tracker", profileRef = "generic.tracker.v1") { + return { + contractVersion: DEVICE_ADAPTER_CONTRACT_VERSION, + adapterRef, + runtimePackageRef: `@nodedc/${adapterRef}-adapter`, + profiles: [{ + profileRef, + framing: { maxBufferedBytes: 64 * 1024 }, + }], + createSession: () => ({}), + }; +} + +test("resolves only explicitly allowlisted adapter profiles", () => { + const registry = createDeviceAdapterRegistry({ + adapters: [adapter()], + }); + assert.deepEqual(registry.adapterRefs, ["generic-tracker"]); + assert.deepEqual(registry.profileRefs, ["generic.tracker.v1"]); + assert.equal( + registry.resolveProfile("generic.tracker.v1").adapter.adapterRef, + "generic-tracker", + ); + assert.throws( + () => registry.resolveProfile("unknown.tracker.v1"), + /device_adapter_profile_not_allowlisted/, + ); +}); + +test("rejects duplicate adapters and cross-adapter profile collisions", () => { + assert.throws( + () => createDeviceAdapterRegistry({ adapters: [adapter(), adapter()] }), + /device_adapter_registry_adapter_duplicate/, + ); + assert.throws( + () => createDeviceAdapterRegistry({ + adapters: [ + adapter("generic-tracker", "shared.profile.v1"), + adapter("other-tracker", "shared.profile.v1"), + ], + }), + /device_adapter_registry_profile_duplicate/, + ); +}); + +test("freezes adapter metadata but keeps the session factory callable", () => { + const defined = defineDeviceAdapter(adapter()); + assert.equal(Object.isFrozen(defined), true); + assert.equal(Object.isFrozen(defined.profiles[0]), true); + assert.equal(typeof defined.createSession, "function"); +}); diff --git a/device-plane/packages/device-protocol-contract/src/index.mjs b/device-plane/packages/device-protocol-contract/src/index.mjs index 2f5cca9..8c14696 100644 --- a/device-plane/packages/device-protocol-contract/src/index.mjs +++ b/device-plane/packages/device-protocol-contract/src/index.mjs @@ -6,6 +6,12 @@ export const DEVICE_DISCOVERY_VIEW_SCHEMA = "nodedc.device.discovery-view.v1"; export const DEVICE_PLANE_BINDING_SCHEMA = "nodedc.device-plane-control.binding.v1"; +export const DEVICE_ADAPTER_MESSAGE_SCHEMA = + "nodedc.device-adapter-message.v1"; +export const DEVICE_ADAPTER_MESSAGE_VIEW_SCHEMA = + "nodedc.device-adapter-message-view.v1"; +export const DEVICE_ADAPTER_ACCEPTANCE_SCHEMA = + "nodedc.device-adapter-acceptance.v1"; export const DEVICE_LIFECYCLE_STATES = Object.freeze([ "quarantine", @@ -24,7 +30,9 @@ export const DEVICE_BINDING_CAPABILITIES = Object.freeze([ const OPAQUE_REF_RE = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; const IMEI_RE = /^\d{15}$/; +const IDENTIFIER_VALUE_RE = /^[A-Za-z0-9][A-Za-z0-9._:+\/-]{3,127}$/; const DIGEST_RE = /^hmac-sha256:[a-f0-9]{64}$/; +const SHA256_DIGEST_RE = /^sha256:[a-f0-9]{64}$/; const IDENTIFIER_KIND_RE = /^[a-z][a-z0-9._:-]{1,63}$/; const forbiddenKeyFragments = Object.freeze([ "password", @@ -75,6 +83,129 @@ export function normalizeDiscoverySignal(input) { }); } +export function normalizeAdapterMessage(input, { maxBytes = 1024 * 1024 } = {}) { + assertPlainObject(input, "device_adapter_message"); + const allowedKeys = new Set([ + "schemaVersion", + "edgeRef", + "adapterRef", + "protocolProfileRef", + "protocol", + "sessionRef", + "routeRef", + "messageRef", + "messageType", + "sequence", + "observedAt", + "idempotencyKey", + "identifier", + "payloadSchemaRef", + "payload", + ]); + rejectUnexpectedKeys(input, allowedKeys, "device_adapter_message_field_unexpected"); + rejectForbiddenKeys(input); + if (input.schemaVersion !== DEVICE_ADAPTER_MESSAGE_SCHEMA) { + throw new TypeError("device_adapter_message_schema_invalid"); + } + const normalizedMaxBytes = normalizeByteLimit(maxBytes); + const serializedBytes = Buffer.byteLength(JSON.stringify(input), "utf8"); + if (serializedBytes > normalizedMaxBytes) { + throw new TypeError("device_adapter_message_too_large"); + } + const sequence = Number(input.sequence); + if (!Number.isSafeInteger(sequence) || sequence < 1) { + throw new TypeError("device_adapter_message_sequence_invalid"); + } + assertPlainObject(input.payload, "device_adapter_message_payload"); + assertJsonValue(input.payload, 0); + + return deepFreeze({ + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: normalizeOpaqueRef(input.edgeRef, "edge_ref"), + adapterRef: normalizeAdapterRef(input.adapterRef), + protocolProfileRef: normalizeOpaqueRef( + input.protocolProfileRef, + "protocol_profile_ref", + ), + protocol: normalizeUpperToken(input.protocol, "protocol"), + sessionRef: normalizeOpaqueRef(input.sessionRef, "session_ref"), + ...(input.routeRef == null + ? {} + : { routeRef: normalizeEntityRef(input.routeRef, "route", "route_ref") }), + messageRef: normalizeOpaqueRef(input.messageRef, "message_ref"), + messageType: normalizeLowerToken(input.messageType, "message_type"), + sequence, + observedAt: normalizeTimestamp(input.observedAt, "observed_at"), + idempotencyKey: normalizeSha256Digest( + input.idempotencyKey, + "idempotency_key", + ), + identifier: normalizeRestrictedIdentifier(input.identifier), + payloadSchemaRef: normalizeOpaqueRef( + input.payloadSchemaRef, + "payload_schema_ref", + ), + payload: cloneJsonValue(input.payload), + }); +} + +export function toSafeAdapterMessageView(message) { + const normalized = normalizeAdapterMessage(message); + return deepFreeze({ + schemaVersion: DEVICE_ADAPTER_MESSAGE_VIEW_SCHEMA, + edgeRef: normalized.edgeRef, + adapterRef: normalized.adapterRef, + protocolProfileRef: normalized.protocolProfileRef, + protocol: normalized.protocol, + sessionRef: normalized.sessionRef, + ...(normalized.routeRef ? { routeRef: normalized.routeRef } : {}), + messageRef: normalized.messageRef, + messageType: normalized.messageType, + sequence: normalized.sequence, + observedAt: normalized.observedAt, + idempotencyKey: normalized.idempotencyKey, + identifier: normalizeRestrictedIdentifierProjection({ + kind: normalized.identifier.kind, + masked: maskRestrictedIdentifier(normalized.identifier), + }), + payloadSchemaRef: normalized.payloadSchemaRef, + payload: normalized.payload, + }); +} + +export function normalizeAdapterAcceptance(input) { + assertPlainObject(input, "device_adapter_acceptance"); + const allowedKeys = new Set([ + "schemaVersion", + "acceptanceRef", + "idempotencyKey", + "status", + "replayed", + "acceptedAt", + ]); + rejectUnexpectedKeys(input, allowedKeys, "device_adapter_acceptance_field_unexpected"); + if (input.schemaVersion !== DEVICE_ADAPTER_ACCEPTANCE_SCHEMA) { + throw new TypeError("device_adapter_acceptance_schema_invalid"); + } + if (input.status !== "accepted") { + throw new TypeError("device_adapter_acceptance_status_invalid"); + } + if (typeof input.replayed !== "boolean") { + throw new TypeError("device_adapter_acceptance_replayed_invalid"); + } + return Object.freeze({ + schemaVersion: DEVICE_ADAPTER_ACCEPTANCE_SCHEMA, + acceptanceRef: normalizeOpaqueRef(input.acceptanceRef, "acceptance_ref"), + idempotencyKey: normalizeSha256Digest( + input.idempotencyKey, + "idempotency_key", + ), + status: "accepted", + replayed: input.replayed, + acceptedAt: normalizeTimestamp(input.acceptedAt, "accepted_at"), + }); +} + export function toSafeDiscoveryView(signal, options = {}) { const normalized = normalizeDiscoverySignal(signal); const discoveryRef = options.discoveryRef @@ -204,13 +335,24 @@ export function assertSafeProjection(value) { export function normalizeRestrictedIdentifier(input) { assertPlainObject(input, "restricted_identifier"); - if (input.kind !== "imei") { - throw new TypeError("restricted_identifier_kind_unsupported"); + rejectUnexpectedKeys( + input, + new Set(["kind", "value"]), + "restricted_identifier_field_unexpected", + ); + if (typeof input.kind !== "string" || !IDENTIFIER_KIND_RE.test(input.kind)) { + throw new TypeError("restricted_identifier_kind_invalid"); } - if (typeof input.value !== "string" || !IMEI_RE.test(input.value)) { + if (input.kind === "imei" && !IMEI_RE.test(input.value)) { throw new TypeError("restricted_identifier_imei_invalid"); } - return Object.freeze({ kind: "imei", value: input.value }); + if ( + typeof input.value !== "string" + || !IDENTIFIER_VALUE_RE.test(input.value) + ) { + throw new TypeError("restricted_identifier_value_invalid"); + } + return Object.freeze({ kind: input.kind, value: input.value }); } export function maskRestrictedIdentifier(identifier) { @@ -218,7 +360,12 @@ export function maskRestrictedIdentifier(identifier) { if (normalized.kind === "imei") { return `***********${normalized.value.slice(-4)}`; } - throw new TypeError("restricted_identifier_kind_unsupported"); + const visible = normalized.value.slice(-4); + const maskedLength = Math.min( + 124, + Math.max(4, normalized.value.length - visible.length), + ); + return `${"*".repeat(maskedLength)}${visible}`; } function normalizeDiscoveryEvidence(input) { @@ -265,6 +412,73 @@ function rejectForbiddenKeys(value, path = "$") { } } +function rejectUnexpectedKeys(input, allowedKeys, errorCode) { + for (const key of Object.keys(input)) { + if (!allowedKeys.has(key)) throw new TypeError(`${errorCode}:${key}`); + } +} + +function normalizeLowerToken(value, label) { + if (typeof value !== "string" || !/^[a-z][a-z0-9._-]{1,127}$/.test(value)) { + throw new TypeError(`${label}_invalid`); + } + return value; +} + +function normalizeAdapterRef(value) { + if (typeof value !== "string" || !/^[a-z][a-z0-9-]{1,62}$/.test(value)) { + throw new TypeError("adapter_ref_invalid"); + } + return value; +} + +function normalizeSha256Digest(value, label) { + if (typeof value !== "string" || !SHA256_DIGEST_RE.test(value)) { + throw new TypeError(`${label}_invalid`); + } + return value; +} + +function normalizeByteLimit(value) { + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 1024 || parsed > 1024 * 1024) { + throw new TypeError("device_adapter_message_limit_invalid"); + } + return parsed; +} + +function assertJsonValue(value, depth) { + if (depth > 12) throw new TypeError("device_adapter_message_payload_too_deep"); + if (value === null || typeof value === "boolean" || typeof value === "string") { + if (typeof value === "string" && value.length > 64 * 1024) { + throw new TypeError("device_adapter_message_payload_string_too_large"); + } + return; + } + if (typeof value === "number") { + if (!Number.isFinite(value)) { + throw new TypeError("device_adapter_message_payload_number_invalid"); + } + return; + } + if (Array.isArray(value)) { + if (value.length > 4096) { + throw new TypeError("device_adapter_message_payload_array_too_large"); + } + value.forEach((item) => assertJsonValue(item, depth + 1)); + return; + } + assertPlainObject(value, "device_adapter_message_payload"); + if (Object.keys(value).length > 1024) { + throw new TypeError("device_adapter_message_payload_object_too_large"); + } + for (const child of Object.values(value)) assertJsonValue(child, depth + 1); +} + +function cloneJsonValue(value) { + return JSON.parse(JSON.stringify(value)); +} + function normalizeOpaqueRef(value, label) { if (typeof value !== "string" || !OPAQUE_REF_RE.test(value)) { throw new TypeError(`${label}_invalid`); @@ -306,3 +520,11 @@ function assertPlainObject(value, label) { throw new TypeError(`${label}_invalid`); } } + +function deepFreeze(value) { + if (!value || typeof value !== "object" || Object.isFrozen(value)) { + return value; + } + Object.values(value).forEach(deepFreeze); + return Object.freeze(value); +} diff --git a/device-plane/packages/device-protocol-contract/test/contract.test.mjs b/device-plane/packages/device-protocol-contract/test/contract.test.mjs index 00b5ef9..286fc9b 100644 --- a/device-plane/packages/device-protocol-contract/test/contract.test.mjs +++ b/device-plane/packages/device-protocol-contract/test/contract.test.mjs @@ -2,16 +2,23 @@ import assert from "node:assert/strict"; import test from "node:test"; import { + DEVICE_ADAPTER_ACCEPTANCE_SCHEMA, + DEVICE_ADAPTER_MESSAGE_SCHEMA, DEVICE_DISCOVERY_SIGNAL_SCHEMA, DEVICE_PLANE_BINDING_SCHEMA, assertIdentifierDigest, assertSafeProjection, hashRestrictedIdentifier, + maskRestrictedIdentifier, normalizeDevicePlaneBinding, + normalizeAdapterAcceptance, + normalizeAdapterMessage, normalizeDiscoverySignal, + normalizeRestrictedIdentifier, normalizeRestrictedIdentifierProjection, normalizeRestrictedIdentifierRecord, toSafeDiscoveryView, + toSafeAdapterMessageView, } from "../src/index.mjs"; const fakeImei = "000000000000001"; @@ -110,6 +117,25 @@ test("restricted identifier records keep digest internal and expose only a mask" ); }); +test("restricted identifiers support future adapter-defined hardware ids", () => { + const identifier = normalizeRestrictedIdentifier({ + kind: "serial", + value: "SN-TRACKER-0001", + }); + assert.deepEqual(identifier, { + kind: "serial", + value: "SN-TRACKER-0001", + }); + assert.equal(maskRestrictedIdentifier(identifier), "***********0001"); + assert.match( + hashRestrictedIdentifier( + identifier, + "test-only-pepper-with-at-least-32-bytes", + ), + /^hmac-sha256:[a-f0-9]{64}$/, + ); +}); + test("rejects unverified framing and command-shaped discovery input", () => { assert.throws( () => normalizeDiscoverySignal({ @@ -147,3 +173,93 @@ test("normalizes an opaque Foundry control binding without device data", () => { assert.deepEqual(binding.capabilities, ["inspect", "observe"]); assertSafeProjection(binding); }); + +test("normalizes a bounded typed adapter message and masks its identity", () => { + const message = { + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: "edge:robot2b-vps-001", + adapterRef: "arusnavi-b2", + protocolProfileRef: "arusnavi.b2.internal.v1", + protocol: "INTERNAL", + sessionRef: "session:test-001", + routeRef: "route:11111111-1111-4111-8111-111111111111", + messageRef: "package:1:abc123", + messageType: "telemetry.package", + sequence: 1, + observedAt: "2026-08-11T12:00:00.000Z", + idempotencyKey: `sha256:${"a".repeat(64)}`, + identifier: { kind: "imei", value: fakeImei }, + payloadSchemaRef: "arusnavi.internal.package-metadata.v1", + payload: { + packageNumber: 1, + packetCount: 1, + packageDigest: `sha256:${"b".repeat(64)}`, + }, + }; + const normalized = normalizeAdapterMessage(message); + const safe = toSafeAdapterMessageView(normalized); + assert.equal(normalized.identifier.value, fakeImei); + assert.equal(safe.identifier.masked, "***********0001"); + assert.equal(JSON.stringify(safe).includes(fakeImei), false); + assertSafeProjection(safe); +}); + +test("rejects oversized, untyped and secret-shaped adapter messages", () => { + const base = { + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: "edge:test", + adapterRef: "generic-tracker", + protocolProfileRef: "generic.tracker.v1", + protocol: "GENERIC", + sessionRef: "session:test", + messageRef: "message:1", + messageType: "telemetry.sample", + sequence: 1, + observedAt: "2026-08-11T12:00:00.000Z", + idempotencyKey: `sha256:${"a".repeat(64)}`, + identifier: { kind: "imei", value: fakeImei }, + payloadSchemaRef: "generic.telemetry.v1", + payload: { value: 1 }, + }; + assert.throws( + () => normalizeAdapterMessage({ ...base, payload: "raw" }), + /device_adapter_message_payload_invalid/, + ); + assert.throws( + () => normalizeAdapterMessage({ + ...base, + payload: { devicePassword: "forbidden" }, + }), + /forbidden_device_field/, + ); + assert.throws( + () => normalizeAdapterMessage({ + ...base, + payload: { value: "x".repeat(4096) }, + }, { maxBytes: 1024 }), + /device_adapter_message_too_large/, + ); + assert.throws( + () => normalizeAdapterMessage({ + ...base, + idempotencyKey: "message-not-a-digest", + }), + /idempotency_key_invalid/, + ); +}); + +test("accepts only an explicit durable Core acceptance contract", () => { + const acceptance = normalizeAdapterAcceptance({ + schemaVersion: DEVICE_ADAPTER_ACCEPTANCE_SCHEMA, + acceptanceRef: "acceptance:test-001", + idempotencyKey: `sha256:${"a".repeat(64)}`, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }); + assert.equal(acceptance.status, "accepted"); + assert.throws( + () => normalizeAdapterAcceptance({ ...acceptance, status: "queued" }), + /device_adapter_acceptance_status_invalid/, + ); +}); diff --git a/device-plane/services/device-control-core/Dockerfile b/device-plane/services/device-control-core/Dockerfile index f8f9a78..c0d85c3 100644 --- a/device-plane/services/device-control-core/Dockerfile +++ b/device-plane/services/device-control-core/Dockerfile @@ -4,7 +4,9 @@ WORKDIR /app COPY package.json package-lock.json ./ COPY packages/device-protocol-contract ./packages/device-protocol-contract -COPY packages/arusnavi-b2-adapter ./packages/arusnavi-b2-adapter +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 diff --git a/device-plane/services/device-control-core/migrations/012_device_gateway_message_receipts.sql b/device-plane/services/device-control-core/migrations/012_device_gateway_message_receipts.sql new file mode 100644 index 0000000..9ddd587 --- /dev/null +++ b/device-plane/services/device-control-core/migrations/012_device_gateway_message_receipts.sql @@ -0,0 +1,66 @@ +begin; + +create table if not exists device_gateway_message_receipts ( + id uuid primary key, + idempotency_key text not null + check (idempotency_key ~ '^sha256:[a-f0-9]{64}$'), + request_digest text not null + check (request_digest ~ '^sha256:[a-f0-9]{64}$'), + edge_ref text not null + check (length(btrim(edge_ref)) between 3 and 128), + adapter_ref text not null + check (adapter_ref ~ '^[a-z][a-z0-9-]{1,62}$'), + protocol_profile_ref text not null + references device_model_profiles(profile_ref), + protocol text not null + check (protocol ~ '^[A-Z][A-Z0-9_]{0,31}$'), + route_id uuid references device_routes(id), + project_id uuid references device_projects(id), + session_ref text not null + check (length(btrim(session_ref)) between 3 and 128), + message_ref text not null + check (length(btrim(message_ref)) between 3 and 128), + message_type text not null + check (message_type ~ '^[a-z][a-z0-9._-]{1,127}$'), + sequence bigint not null check (sequence > 0), + identifier_kind text not null + check (identifier_kind ~ '^[a-z][a-z0-9._:-]{1,63}$'), + identifier_digest text not null + check (identifier_digest ~ '^hmac-sha256:[a-f0-9]{64}$'), + identifier_masked text not null + check (length(identifier_masked) between 5 and 128), + payload_schema_ref text not null + check (length(btrim(payload_schema_ref)) between 3 and 128), + payload jsonb not null, + observed_at timestamptz not null, + accepted_at timestamptz not null default now(), + unique (idempotency_key), + unique (edge_ref, session_ref, message_ref), + foreign key (route_id, project_id) + references device_routes(id, project_id), + check ( + (route_id is null and project_id is null) + or (route_id is not null and project_id is not null) + ) +); + +create index if not exists device_gateway_message_receipts_route_time_idx + on device_gateway_message_receipts (route_id, accepted_at desc) + where route_id is not null; + +create index if not exists device_gateway_message_receipts_identity_time_idx + on device_gateway_message_receipts ( + identifier_kind, + identifier_digest, + accepted_at desc + ); + +drop trigger if exists device_gateway_message_receipts_immutable_guard + on device_gateway_message_receipts; +create trigger device_gateway_message_receipts_immutable_guard +before update or delete or truncate +on device_gateway_message_receipts +for each statement +execute function reject_device_immutable_record_mutation(); + +commit; diff --git a/device-plane/services/device-control-core/src/app.mjs b/device-plane/services/device-control-core/src/app.mjs index 9c3efcd..29632f1 100644 --- a/device-plane/services/device-control-core/src/app.mjs +++ b/device-plane/services/device-control-core/src/app.mjs @@ -5,9 +5,12 @@ import { assertSafeProjection, hashRestrictedIdentifier, maskRestrictedIdentifier, + normalizeAdapterAcceptance, + normalizeAdapterMessage, normalizeDiscoverySignal, normalizeRestrictedIdentifier, toSafeDiscoveryView, + toSafeAdapterMessageView, } from "../../../packages/device-protocol-contract/src/index.mjs"; import { normalizeManagementActor, @@ -64,6 +67,9 @@ export function createControlCoreApp({ if (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 gatewayToken !== "string" || gatewayToken.length < 32) { throw new TypeError("device_gateway_token_invalid"); } @@ -249,6 +255,55 @@ export function createControlCoreApp({ }); } + if ( + request.method === "POST" + && requestUrl.pathname === "/internal/v1/gateway/messages:accept" + ) { + if (!discoveryIngestEnabled) { + return writeJson(response, 404, { + ok: false, + error: "device_gateway_message_ingest_disabled", + }); + } + if (!matchesBearer(request.headers.authorization, gatewayToken)) { + return writeJson(response, 401, { + ok: false, + error: "device_gateway_auth_required", + }); + } + + 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, + }), + ); + return writeJson(response, acceptance.replayed ? 200 : 201, { + ok: true, + acceptance, + }); + } + return writeJson(response, 404, { ok: false, error: "device_control_core_route_not_found", @@ -366,6 +421,12 @@ 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/gateway-message-repository.mjs b/device-plane/services/device-control-core/src/gateway-message-repository.mjs new file mode 100644 index 0000000..df3d57e --- /dev/null +++ b/device-plane/services/device-control-core/src/gateway-message-repository.mjs @@ -0,0 +1,163 @@ +import { randomUUID } from "node:crypto"; + +export async function acceptGatewayMessage({ + pool, + identifierDigest, + requestDigest, + safeView, +}) { + const routeId = safeView.routeRef == null + ? null + : parseEntityRef(safeView.routeRef, "route"); + const client = await pool.connect(); + try { + await client.query("begin"); + const route = routeId == null + ? null + : await findActiveRoute(client, routeId, safeView); + const id = randomUUID(); + const inserted = await client.query( + `insert into device_gateway_message_receipts ( + id, + idempotency_key, + request_digest, + edge_ref, + adapter_ref, + protocol_profile_ref, + protocol, + route_id, + project_id, + session_ref, + message_ref, + message_type, + sequence, + identifier_kind, + identifier_digest, + identifier_masked, + payload_schema_ref, + payload, + observed_at + ) values ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, + $11, $12, $13, $14, $15, $16, $17, $18::jsonb, $19 + ) + on conflict (idempotency_key) do nothing + returning id, idempotency_key, accepted_at`, + [ + id, + safeView.idempotencyKey, + requestDigest, + safeView.edgeRef, + safeView.adapterRef, + safeView.protocolProfileRef, + safeView.protocol, + route?.id ?? null, + route?.project_id ?? null, + safeView.sessionRef, + safeView.messageRef, + safeView.messageType, + safeView.sequence, + safeView.identifier.kind, + identifierDigest, + safeView.identifier.masked, + safeView.payloadSchemaRef, + JSON.stringify(safeView.payload), + safeView.observedAt, + ], + ); + if (inserted.rows[0]) { + await client.query("commit"); + return acceptanceView(inserted.rows[0], false); + } + + const existing = await client.query( + `select id, idempotency_key, request_digest, accepted_at + from device_gateway_message_receipts + where idempotency_key = $1 + for share`, + [safeView.idempotencyKey], + ); + const row = existing.rows[0]; + if (!row) throw domainError("device_gateway_receipt_missing", 409); + if (row.request_digest !== requestDigest) { + throw domainError("device_gateway_idempotency_conflict", 409); + } + await client.query("commit"); + return acceptanceView(row, true); + } catch (error) { + await client.query("rollback").catch(() => undefined); + throw error; + } finally { + client.release(); + } +} + +async function findActiveRoute(client, routeId, safeView) { + const edgeId = parseEntityRef(safeView.edgeRef, "edge"); + const result = await client.query( + `select r.id, r.project_id, r.edge_id, r.model_profile_ref, + r.protocol, r.lifecycle_state, + e.lifecycle_state as edge_lifecycle_state, + p.lifecycle_state as profile_lifecycle_state, + ap.package_key as adapter_ref, + ap.lifecycle_state as adapter_lifecycle_state, + av.lifecycle_state as adapter_version_lifecycle_state + from device_routes r + join device_edges e on e.id = r.edge_id + join device_model_profiles p on p.profile_ref = r.model_profile_ref + join device_adapter_versions av on av.id = p.adapter_version_id + join device_adapter_packages ap on ap.id = av.adapter_package_id + where r.id = $1 + for share`, + [routeId], + ); + const route = result.rows[0]; + if (!route) throw domainError("device_gateway_route_not_found", 404); + if ( + route.lifecycle_state !== "active" + || route.edge_lifecycle_state !== "active" + || route.profile_lifecycle_state !== "active" + || route.adapter_lifecycle_state !== "active" + || route.adapter_version_lifecycle_state !== "active" + ) { + throw domainError("device_gateway_route_not_active", 409); + } + if ( + route.edge_id !== edgeId + || route.model_profile_ref !== safeView.protocolProfileRef + || route.protocol !== safeView.protocol + || route.adapter_ref !== safeView.adapterRef + ) { + throw domainError("device_gateway_route_contract_mismatch", 409); + } + return route; +} + +function acceptanceView(row, replayed) { + return { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: `acceptance:${row.id}`, + idempotencyKey: row.idempotency_key, + status: "accepted", + replayed, + acceptedAt: new Date(row.accepted_at).toISOString(), + }; +} + +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 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 aa5163d..77902ad 100644 --- a/device-plane/services/device-control-core/src/postgres-repository.mjs +++ b/device-plane/services/device-control-core/src/postgres-repository.mjs @@ -5,8 +5,8 @@ import { fileURLToPath } from "node:url"; import pg from "pg"; -import { ARUSNAVI_B2_MODEL_PROFILE } from "../../../packages/arusnavi-b2-adapter/src/index.mjs"; import { observeQuarantineDiscovery } from "./discovery-repository.mjs"; +import { acceptGatewayMessage } from "./gateway-message-repository.mjs"; import { applyControlResourceManagementCommand, authorizeControlResourceManagementReplay, @@ -56,6 +56,7 @@ const migrationFiles = [ "009_device_sensitive_reference_commands.sql", "010_device_control_resources.sql", "011_device_control_resource_commands.sql", + "012_device_gateway_message_receipts.sql", ]; export class PostgresDeviceRepository { @@ -88,30 +89,6 @@ export class PostgresDeviceRepository { ); await this.pool.query(sql); } - await this.pool.query( - `insert into device_model_profiles ( - profile_ref, - schema_version, - vendor, - model, - device_type, - protocol, - profile - ) values ($1, $2, $3, $4, $5, $6, $7::jsonb) - on conflict (profile_ref) do update set - schema_version = excluded.schema_version, - profile = excluded.profile, - updated_at = now()`, - [ - ARUSNAVI_B2_MODEL_PROFILE.profileRef, - ARUSNAVI_B2_MODEL_PROFILE.schemaVersion, - ARUSNAVI_B2_MODEL_PROFILE.vendor, - ARUSNAVI_B2_MODEL_PROFILE.model, - ARUSNAVI_B2_MODEL_PROFILE.deviceType, - ARUSNAVI_B2_MODEL_PROFILE.protocol, - JSON.stringify(ARUSNAVI_B2_MODEL_PROFILE), - ], - ); } async health() { @@ -126,6 +103,13 @@ export class PostgresDeviceRepository { }); } + async acceptAdapterMessage(input) { + return acceptGatewayMessage({ + pool: this.pool, + ...input, + }); + } + async executeManagementCommand({ idempotencyKey, commandKind, 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 39bd90f..21f91b6 100644 --- a/device-plane/services/device-control-core/test/app.test.mjs +++ b/device-plane/services/device-control-core/test/app.test.mjs @@ -2,6 +2,7 @@ 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 { createControlCoreApp } from "../src/app.mjs"; @@ -362,6 +363,9 @@ test("authenticated ingest stores only digest and returns a masked view", async }, }; }, + acceptAdapterMessage: async () => { + throw new Error("must_not_accept_message"); + }, }, }); try { @@ -400,6 +404,54 @@ test("authenticated ingest stores only digest and returns a masked view", async } }); +test("gateway message endpoint returns acceptance only after repository commit", async () => { + let stored; + const runtime = await startTestServer({ + discoveryIngestEnabled: true, + gatewayToken, + identifierPepper, + repository: { + health: async () => "ready", + upsertQuarantineDiscovery: async () => { + throw new Error("must_not_observe_discovery"); + }, + acceptAdapterMessage: async (value) => { + stored = value; + return { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test-001", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }; + }, + }, + }); + try { + const response = await fetch( + `${runtime.baseUrl}/internal/v1/gateway/messages:accept`, + { + method: "POST", + headers: { + Authorization: `Bearer ${gatewayToken}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(fakeAdapterMessage()), + }, + ); + assert.equal(response.status, 201); + const body = await response.json(); + assert.equal(body.acceptance.status, "accepted"); + assert.match(stored.identifierDigest, /^hmac-sha256:[a-f0-9]{64}$/); + assert.match(stored.requestDigest, /^sha256:[a-f0-9]{64}$/); + assert.equal(stored.safeView.identifier.masked, "***********0001"); + assert.equal(JSON.stringify(stored).includes(fakeImei), false); + } finally { + await runtime.close(); + } +}); + function fakeSignal() { return { schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA, @@ -417,6 +469,29 @@ function fakeSignal() { }; } +function fakeAdapterMessage() { + return { + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: "edge:test-001", + adapterRef: "arusnavi-b2", + protocolProfileRef: "arusnavi.b2.internal.v1", + protocol: "INTERNAL", + sessionRef: "session:test-001", + 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: fakeImei }, + payloadSchemaRef: "arusnavi.internal.package-metadata.v1", + payload: { + packageNumber: 1, + packetCount: 1, + packageDigest: `sha256:${"b".repeat(64)}`, + }, + }; +} + function managementHeaders({ includeIdempotency = true } = {}) { return { Authorization: `Bearer ${managementToken}`, diff --git a/device-plane/services/device-control-core/test/gateway-message-migration.test.mjs b/device-plane/services/device-control-core/test/gateway-message-migration.test.mjs new file mode 100644 index 0000000..86c2f57 --- /dev/null +++ b/device-plane/services/device-control-core/test/gateway-message-migration.test.mjs @@ -0,0 +1,39 @@ +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import test from "node:test"; + +const migrationUrl = new URL( + "../migrations/012_device_gateway_message_receipts.sql", + import.meta.url, +); +const repositoryUrl = new URL("../src/postgres-repository.mjs", import.meta.url); + +test("gateway receipts persist only typed bounded Core acceptance evidence", async () => { + const sql = await readFile(migrationUrl, "utf8"); + + assert.match(sql, /create table if not exists device_gateway_message_receipts/); + assert.match(sql, /unique \(idempotency_key\)/); + assert.match(sql, /unique \(edge_ref, session_ref, message_ref\)/); + assert.match(sql, /identifier_digest text not null/); + assert.match(sql, /identifier_masked text not null/); + assert.match(sql, /payload_schema_ref text not null/); + assert.match(sql, /payload jsonb not null/); + assert.match(sql, /device_gateway_message_receipts_immutable_guard/); + assert.doesNotMatch(sql, /raw_packet|raw_identifier|password|token|secret/i); + assert.doesNotMatch(sql, /insert\s+into|arusnavi|gelios|\bb2\b|imei/i); +}); + +test("gateway receipt migration follows the generic control resource schema", async () => { + const repository = await readFile(repositoryUrl, "utf8"); + const controlResourceIndex = repository.indexOf( + "011_device_control_resource_commands.sql", + ); + const gatewayReceiptIndex = repository.indexOf( + "012_device_gateway_message_receipts.sql", + ); + + assert.notEqual(controlResourceIndex, -1); + assert.notEqual(gatewayReceiptIndex, -1); + assert.ok(controlResourceIndex < gatewayReceiptIndex); + assert.doesNotMatch(repository, /arusnavi-b2-adapter/); +}); diff --git a/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs b/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs new file mode 100644 index 0000000..a1e5044 --- /dev/null +++ b/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs @@ -0,0 +1,165 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { acceptGatewayMessage } from "../src/gateway-message-repository.mjs"; + +const acceptedAt = new Date("2026-08-11T12:00:00.000Z"); +const idempotencyKey = `sha256:${"a".repeat(64)}`; +const requestDigest = `sha256:${"b".repeat(64)}`; + +test("commits a gateway receipt before returning Core acceptance", async () => { + const client = scriptedClient([ + step("begin"), + step("insert into device_gateway_message_receipts", { + rows: [{ + id: "11111111-1111-4111-8111-111111111111", + idempotency_key: idempotencyKey, + accepted_at: acceptedAt, + }], + }), + step("commit"), + ]); + + const result = await acceptGatewayMessage(messageInput(client)); + + assert.equal(result.status, "accepted"); + assert.equal(result.replayed, false); + assert.equal(result.idempotencyKey, idempotencyKey); + assert.equal(result.acceptedAt, acceptedAt.toISOString()); + assert.equal(client.remaining(), 0); + assert.equal(client.released, true); +}); + +test("replays one durable receipt for the same normalized request", async () => { + const client = scriptedClient([ + step("begin"), + step("insert into device_gateway_message_receipts", { rows: [] }), + step("from device_gateway_message_receipts", { + rows: [{ + id: "11111111-1111-4111-8111-111111111111", + idempotency_key: idempotencyKey, + request_digest: requestDigest, + accepted_at: acceptedAt, + }], + }), + step("commit"), + ]); + + const result = await acceptGatewayMessage(messageInput(client)); + + assert.equal(result.status, "accepted"); + assert.equal(result.replayed, true); + assert.equal(client.remaining(), 0); + assert.equal(client.released, true); +}); + +test("rejects idempotency reuse with different content", async () => { + const client = scriptedClient([ + step("begin"), + step("insert into device_gateway_message_receipts", { rows: [] }), + step("from device_gateway_message_receipts", { + rows: [{ + id: "11111111-1111-4111-8111-111111111111", + idempotency_key: idempotencyKey, + request_digest: `sha256:${"c".repeat(64)}`, + accepted_at: acceptedAt, + }], + }), + step("rollback"), + ]); + + await assert.rejects( + acceptGatewayMessage(messageInput(client)), + /device_gateway_idempotency_conflict/, + ); + assert.equal(client.remaining(), 0); + assert.equal(client.released, true); +}); + +test("fails closed when a route does not match its Edge contract", async () => { + const routeId = "22222222-2222-4222-8222-222222222222"; + const client = scriptedClient([ + step("begin"), + step("from device_routes r", { + rows: [{ + id: routeId, + project_id: "33333333-3333-4333-8333-333333333333", + edge_id: "44444444-4444-4444-8444-444444444444", + model_profile_ref: "generic.model.protocol.v1", + protocol: "GENERIC_TCP", + lifecycle_state: "active", + edge_lifecycle_state: "active", + profile_lifecycle_state: "active", + adapter_ref: "generic-adapter", + adapter_lifecycle_state: "active", + adapter_version_lifecycle_state: "active", + }], + }), + step("rollback"), + ]); + const input = messageInput(client); + input.safeView.routeRef = `route:${routeId}`; + input.safeView.edgeRef = "edge:55555555-5555-4555-8555-555555555555"; + + await assert.rejects( + acceptGatewayMessage(input), + /device_gateway_route_contract_mismatch/, + ); + assert.equal(client.remaining(), 0); + assert.equal(client.released, true); +}); + +function messageInput(client) { + return { + pool: { + connect: async () => client, + }, + identifierDigest: `hmac-sha256:${"d".repeat(64)}`, + requestDigest, + safeView: { + edgeRef: "edge:test-001", + adapterRef: "generic-adapter", + protocolProfileRef: "generic.model.protocol.v1", + protocol: "GENERIC_TCP", + sessionRef: "session:test-001", + messageRef: "message:test-001", + messageType: "telemetry.sample", + sequence: 1, + idempotencyKey, + identifier: { + kind: "serial", + masked: "********0001", + }, + payloadSchemaRef: "generic.telemetry.v1", + payload: { value: 1 }, + observedAt: "2026-08-11T12:00:00.000Z", + }, + }; +} + +function step(includes, result = { rows: [] }) { + return { includes, result }; +} + +function scriptedClient(steps) { + const queue = [...steps]; + return { + released: false, + async query(sql) { + const next = queue.shift(); + assert.ok(next, `Unexpected query: ${sql}`); + assert.match(String(sql), new RegExp(escapeRegExp(next.includes), "i")); + return next.result; + }, + release() { + this.released = true; + }, + remaining() { + return queue.length; + }, + }; +} + +function escapeRegExp(value) { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} diff --git a/device-plane/services/device-gateway/Dockerfile b/device-plane/services/device-gateway/Dockerfile index 3b4cdac..539e468 100644 --- a/device-plane/services/device-gateway/Dockerfile +++ b/device-plane/services/device-gateway/Dockerfile @@ -4,6 +4,8 @@ WORKDIR /app COPY package.json package-lock.json ./ COPY packages/device-protocol-contract ./packages/device-protocol-contract +COPY packages/device-adapter-runtime ./packages/device-adapter-runtime +COPY packages/device-adapter-catalog ./packages/device-adapter-catalog COPY packages/arusnavi-b2-adapter ./packages/arusnavi-b2-adapter COPY services/device-gateway ./services/device-gateway COPY services/device-control-core/package.json ./services/device-control-core/package.json diff --git a/device-plane/services/device-gateway/src/core-client.mjs b/device-plane/services/device-gateway/src/core-client.mjs index b08d4f9..6d75ebb 100644 --- a/device-plane/services/device-gateway/src/core-client.mjs +++ b/device-plane/services/device-gateway/src/core-client.mjs @@ -1,10 +1,77 @@ +import { + normalizeAdapterAcceptance, +} from "../../../packages/device-protocol-contract/src/index.mjs"; + +export function createCoreGatewayClient(options = {}) { + const { + coreUrl, + gatewayToken, + timeoutMs = 5000, + fetchImpl = fetch, + } = options; + const baseUrl = normalizeCoreBaseUrl(coreUrl); + validateClientOptions({ gatewayToken, timeoutMs, fetchImpl }); + const discoveryEndpoint = endpointUrl( + baseUrl, + "/internal/v1/device-discoveries:observe", + ); + const messageEndpoint = endpointUrl( + baseUrl, + "/internal/v1/gateway/messages:accept", + ); + + return Object.freeze({ + async observeDiscovery(signal) { + const body = await postJson({ + endpoint: discoveryEndpoint, + gatewayToken, + timeoutMs, + fetchImpl, + value: signal, + maxResponseBytes: 32 * 1024, + }); + if ( + !body.discovery + || !["quarantine", "claimed"].includes(body.discovery.lifecycleState) + || body.discovery.commandTransport !== "disabled" + ) { + throw new Error("device_gateway_core_ingest_contract_invalid"); + } + return body.discovery; + }, + async acceptMessage(message) { + const body = await postJson({ + endpoint: messageEndpoint, + gatewayToken, + timeoutMs, + fetchImpl, + value: message, + maxResponseBytes: 16 * 1024, + }); + const acceptance = normalizeAdapterAcceptance(body.acceptance); + if (acceptance.idempotencyKey !== message?.idempotencyKey) { + throw new Error("device_gateway_core_acceptance_mismatch"); + } + return acceptance; + }, + }); +} + export function createCoreDiscoveryClient({ coreUrl, gatewayToken, timeoutMs = 5000, fetchImpl = fetch, } = {}) { - const endpoint = normalizeCoreEndpoint(coreUrl); + return createCoreGatewayClient({ + coreUrl, + gatewayToken, + timeoutMs, + fetchImpl, + }).observeDiscovery; +} + +function validateClientOptions({ gatewayToken, timeoutMs, fetchImpl }) { if (typeof gatewayToken !== "string" || gatewayToken.length < 32) { throw new TypeError("device_gateway_core_token_invalid"); } @@ -19,33 +86,33 @@ export function createCoreDiscoveryClient({ if (typeof fetchImpl !== "function") { throw new TypeError("device_gateway_core_fetch_invalid"); } - - return async function observeDiscovery(signal) { - const response = await fetchImpl(endpoint, { - method: "POST", - headers: { - Authorization: `Bearer ${gatewayToken}`, - "Content-Type": "application/json", - }, - body: JSON.stringify(signal), - signal: AbortSignal.timeout(normalizedTimeout), - }); - const body = await readBoundedJson(response, 32 * 1024); - if (!response.ok || body?.ok !== true) { - throw new Error("device_gateway_core_ingest_failed"); - } - if ( - !body.discovery - || body.discovery.lifecycleState !== "quarantine" - || body.discovery.commandTransport !== "disabled" - ) { - throw new Error("device_gateway_core_ingest_contract_invalid"); - } - return body.discovery; - }; } -function normalizeCoreEndpoint(value) { +async function postJson({ + endpoint, + gatewayToken, + timeoutMs, + fetchImpl, + value, + maxResponseBytes, +}) { + const response = await fetchImpl(endpoint, { + method: "POST", + headers: { + Authorization: `Bearer ${gatewayToken}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(value), + signal: AbortSignal.timeout(Number(timeoutMs)), + }); + const body = await readBoundedJson(response, maxResponseBytes); + if (!response.ok || body?.ok !== true) { + throw new Error("device_gateway_core_ingest_failed"); + } + return body; +} + +function normalizeCoreBaseUrl(value) { let url; try { url = new URL(String(value || "")); @@ -58,7 +125,12 @@ function normalizeCoreEndpoint(value) { if (url.pathname !== "/" || url.search || url.hash) { throw new TypeError("device_gateway_core_url_invalid"); } - url.pathname = "/internal/v1/device-discoveries:observe"; + return url; +} + +function endpointUrl(baseUrl, pathname) { + const url = new URL(baseUrl); + url.pathname = pathname; return url.toString(); } diff --git a/device-plane/services/device-gateway/src/runtime.mjs b/device-plane/services/device-gateway/src/runtime.mjs index 455caff..d4dc70c 100644 --- a/device-plane/services/device-gateway/src/runtime.mjs +++ b/device-plane/services/device-gateway/src/runtime.mjs @@ -1,16 +1,15 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { createServer as createHttpServer } from "node:http"; import { createServer as createTcpServer } from "node:net"; import { - ARUSNAVI_B2_MODEL_PROFILE, - buildB2HeaderAcknowledgement, - buildB2PackageAcknowledgement, - tryParseB2Header2, - tryParseB2Package, -} from "../../../packages/arusnavi-b2-adapter/src/index.mjs"; + assertDeviceAdapterSession, +} from "../../../packages/device-adapter-runtime/src/index.mjs"; import { + DEVICE_ADAPTER_MESSAGE_SCHEMA, DEVICE_DISCOVERY_SIGNAL_SCHEMA, + normalizeAdapterAcceptance, + normalizeAdapterMessage, } from "../../../packages/device-protocol-contract/src/index.mjs"; export function createDeviceGatewayRuntime(options = {}) { @@ -21,7 +20,9 @@ export function createDeviceGatewayRuntime(options = {}) { let totalAccepted = 0; let totalRejected = 0; let totalDiscoveries = 0; + let totalMessagesAccepted = 0; let totalPackagesAcknowledged = 0; + let totalBufferedBytes = 0; const tcpServer = createTcpServer((socket) => { const remoteAddress = normalizeRemoteAddress(socket.remoteAddress); @@ -40,6 +41,13 @@ export function createDeviceGatewayRuntime(options = {}) { sessionRef, remoteAddress, buffer: Buffer.alloc(0), + adapterSession: assertDeviceAdapterSession( + config.adapter.createSession({ + profileRef: config.profile.profileRef, + }), + ), + identifier: null, + sequence: 0, state: "awaiting-header", processing: false, rejected: false, @@ -54,11 +62,11 @@ export function createDeviceGatewayRuntime(options = {}) { socket.on("data", (chunk) => { if (session.closed || session.rejected) return; socket.pause(); - session.buffer = Buffer.concat( - [session.buffer, chunk], - session.buffer.length + chunk.length, - ); - if (session.buffer.length > config.maxBufferedBytes) { + appendBuffer(session, chunk); + if ( + session.buffer.length > config.maxBufferedBytes + || totalBufferedBytes > config.maxAggregateBufferedBytes + ) { rejectSession(socket, session); return; } @@ -88,11 +96,12 @@ export function createDeviceGatewayRuntime(options = {}) { return response.end(`${JSON.stringify({ ok: true, service: "nodedc-device-gateway", - protocolProfile: ARUSNAVI_B2_MODEL_PROFILE.profileRef, - framing: ARUSNAVI_B2_MODEL_PROFILE.framing.status, - tcpListener: config.listenEnabled ? "discovery-only" : "disabled", + adapter: config.adapter?.adapterRef ?? "disabled", + protocolProfile: config.profile?.profileRef ?? "disabled", + framing: config.profile?.framing?.status ?? "disabled", + tcpListener: config.listenEnabled ? "telemetry-ingest" : "disabled", publicIngress: config.publicIngressEnabled - ? "discovery-only" + ? "telemetry-ingest" : "disabled", commandTransport: "disabled", sessions: { @@ -100,7 +109,9 @@ export function createDeviceGatewayRuntime(options = {}) { accepted: totalAccepted, rejected: totalRejected, discoveries: totalDiscoveries, + messagesAccepted: totalMessagesAccepted, packagesAcknowledged: totalPackagesAcknowledged, + bufferedBytes: totalBufferedBytes, }, })}\n`); }); @@ -118,7 +129,7 @@ export function createDeviceGatewayRuntime(options = {}) { }, async stop() { for (const [socket, session] of sessions) { - session.closed = true; + closeSession(socket, session); socket.destroy(); } await Promise.all([ @@ -132,10 +143,12 @@ export function createDeviceGatewayRuntime(options = {}) { totalAccepted, totalRejected, totalDiscoveries, + totalMessagesAccepted, totalPackagesAcknowledged, + totalBufferedBytes, commandTransport: "disabled", publicIngress: config.publicIngressEnabled - ? "discovery-only" + ? "telemetry-ingest" : "disabled", }; }, @@ -144,36 +157,74 @@ export function createDeviceGatewayRuntime(options = {}) { async function processSession(socket, session) { while (!session.closed && !session.rejected) { if (session.state === "awaiting-header") { - const parsed = tryParseB2Header2(session.buffer); + const parsed = session.adapterSession.parseHeader(session.buffer); if (parsed.status === "incomplete") return; const observedAt = config.now().toISOString(); - await config.onDiscovery?.({ + const discovery = { schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA, sessionRef: session.sessionRef, - modelProfileRef: ARUSNAVI_B2_MODEL_PROFILE.profileRef, - protocol: ARUSNAVI_B2_MODEL_PROFILE.protocol, + ...(config.routeRef ? { routeRef: config.routeRef } : {}), + modelProfileRef: config.profile.profileRef, + protocol: config.profile.protocol, observedAt, identifier: { kind: parsed.identifier.kind, value: parsed.identifier.value, }, evidence: parsed.evidence, - }); - session.buffer = session.buffer.subarray(parsed.bytesConsumed); + }; + const acceptedDiscovery = await config.onDiscovery(discovery); + assertDiscoveryAccepted(acceptedDiscovery); + session.identifier = discovery.identifier; + consumeBuffer(session, parsed.bytesConsumed); session.state = "packages"; totalDiscoveries += 1; - socket.write(buildB2HeaderAcknowledgement( + await writeWithBackpressure(socket, session.adapterSession.buildHeaderAcknowledgement( Math.floor(new Date(observedAt).getTime() / 1000), )); continue; } - const parsed = tryParseB2Package(session.buffer); + const parsed = session.adapterSession.parseMessage(session.buffer); if (parsed.status === "incomplete") return; - session.buffer = session.buffer.subarray(parsed.bytesConsumed); + const observedAt = config.now().toISOString(); + session.sequence += 1; + const message = normalizeAdapterMessage({ + schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, + edgeRef: config.edgeRef, + adapterRef: config.adapter.adapterRef, + protocolProfileRef: config.profile.profileRef, + protocol: config.profile.protocol, + sessionRef: session.sessionRef, + ...(config.routeRef ? { routeRef: config.routeRef } : {}), + messageRef: messageRef(parsed), + messageType: parsed.messageType, + sequence: session.sequence, + observedAt, + idempotencyKey: messageIdempotencyKey({ + adapterRef: config.adapter.adapterRef, + profileRef: config.profile.profileRef, + identifier: session.identifier, + payload: parsed.payload, + }), + identifier: session.identifier, + payloadSchemaRef: parsed.payloadSchemaRef, + payload: parsed.payload, + }); + const acceptance = normalizeAdapterAcceptance( + await config.onMessage(message), + ); + if (acceptance.idempotencyKey !== message.idempotencyKey) { + throw new TypeError("device_gateway_core_acceptance_mismatch"); + } + consumeBuffer(session, parsed.bytesConsumed); + totalMessagesAccepted += 1; totalPackagesAcknowledged += 1; - socket.write(buildB2PackageAcknowledgement(parsed.packageNumber)); + await writeWithBackpressure( + socket, + session.adapterSession.buildMessageAcknowledgement(parsed), + ); } } @@ -199,8 +250,16 @@ export function createDeviceGatewayRuntime(options = {}) { function consumeConnectionPermit(remoteAddress) { const nowMs = config.now().getTime(); + for (const [address, window] of connectionWindows) { + if (nowMs - window.startedAt >= 60_000) { + connectionWindows.delete(address); + } + } const current = connectionWindows.get(remoteAddress); if (!current || nowMs - current.startedAt >= 60_000) { + if (connectionWindows.size >= config.maxTrackedSourceAddresses) { + return false; + } connectionWindows.set(remoteAddress, { startedAt: nowMs, count: 1, @@ -225,9 +284,24 @@ export function createDeviceGatewayRuntime(options = {}) { function closeSession(socket, session) { if (session.closed) return; session.closed = true; + totalBufferedBytes -= session.buffer.length; + session.buffer = Buffer.alloc(0); sessions.delete(socket); decrementAddressSessions(session.remoteAddress); } + + function appendBuffer(session, chunk) { + session.buffer = Buffer.concat( + [session.buffer, chunk], + session.buffer.length + chunk.length, + ); + totalBufferedBytes += chunk.length; + } + + function consumeBuffer(session, bytesConsumed) { + session.buffer = session.buffer.subarray(bytesConsumed); + totalBufferedBytes -= bytesConsumed; + } } function normalizeConfig(input) { @@ -236,15 +310,30 @@ function normalizeConfig(input) { if (publicIngressEnabled && !listenEnabled) { throw new TypeError("device_gateway_public_ingress_listener_required"); } - if ( - publicIngressEnabled - && typeof input.onDiscovery !== "function" - ) { - throw new TypeError("device_gateway_discovery_sink_required"); + if (publicIngressEnabled && input.coreChannelAuthenticated !== true) { + throw new TypeError("device_gateway_authenticated_core_channel_required"); } + if ( + listenEnabled + && (typeof input.onDiscovery !== "function" + || typeof input.onMessage !== "function") + ) { + throw new TypeError("device_gateway_core_acceptance_sink_required"); + } + const registration = listenEnabled + ? resolveAdapterRegistration(input.adapterRegistry, input.protocolProfileRef) + : null; return { listenEnabled, publicIngressEnabled, + adapter: registration?.adapter ?? null, + profile: registration?.profile ?? null, + edgeRef: listenEnabled + ? normalizeOpaqueRef(input.edgeRef, "device_gateway_edge_ref_invalid") + : "edge:disabled", + routeRef: input.routeRef == null || input.routeRef === "" + ? undefined + : normalizeRouteRef(input.routeRef), healthHost: normalizeHealthHost(input.healthHost, "127.0.0.1"), healthPort: parseInteger( input.healthPort, @@ -267,32 +356,46 @@ function normalizeConfig(input) { ), maxBufferedBytes: parseInteger( input.maxBufferedBytes, - ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes, + registration?.profile?.framing?.maxBufferedBytes ?? 256 * 1024, 1024, - ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes, + registration?.profile?.framing?.maxBufferedBytes ?? 256 * 1024, "device_gateway_buffer_limit_invalid", ), + maxAggregateBufferedBytes: parseInteger( + input.maxAggregateBufferedBytes, + 32 * 1024 * 1024, + 1024, + 32 * 1024 * 1024, + "device_gateway_aggregate_buffer_limit_invalid", + ), maxConcurrentSessions: parseInteger( input.maxConcurrentSessions, - 100, + 128, 1, 10000, "device_gateway_session_limit_invalid", ), maxSessionsPerAddress: parseInteger( input.maxSessionsPerAddress, - 10, + 16, 1, 1000, "device_gateway_address_session_limit_invalid", ), maxConnectionsPerMinutePerAddress: parseInteger( input.maxConnectionsPerMinutePerAddress, - 30, + 60, 1, 10000, "device_gateway_address_rate_limit_invalid", ), + maxTrackedSourceAddresses: parseInteger( + input.maxTrackedSourceAddresses, + 2048, + 1, + 65536, + "device_gateway_source_tracking_limit_invalid", + ), sessionTimeoutMs: parseInteger( input.sessionTimeoutMs, 10000, @@ -303,10 +406,83 @@ function normalizeConfig(input) { onDiscovery: typeof input.onDiscovery === "function" ? input.onDiscovery : undefined, + onMessage: typeof input.onMessage === "function" + ? input.onMessage + : undefined, now: typeof input.now === "function" ? input.now : () => new Date(), }; } +function resolveAdapterRegistration(registry, profileRef) { + if (!registry || typeof registry.resolveProfile !== "function") { + throw new TypeError("device_gateway_adapter_registry_required"); + } + return registry.resolveProfile(profileRef); +} + +function normalizeOpaqueRef(value, errorCode) { + if ( + typeof value !== "string" + || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value) + ) { + throw new TypeError(errorCode); + } + return value; +} + +function normalizeRouteRef(value) { + if ( + typeof value !== "string" + || !/^route:[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value) + ) { + throw new TypeError("device_gateway_route_ref_invalid"); + } + return value.toLowerCase(); +} + +function assertDiscoveryAccepted(value) { + if ( + !value + || typeof value !== "object" + || !["quarantine", "claimed"].includes(value.lifecycleState) + ) { + throw new TypeError("device_gateway_core_discovery_not_accepted"); + } +} + +function messageRef(parsed) { + const digest = String(parsed?.payload?.packageDigest || ""); + if (!/^sha256:[a-f0-9]{64}$/.test(digest)) { + throw new TypeError("device_gateway_adapter_message_digest_invalid"); + } + return `package:${parsed.packageNumber}:${digest.slice("sha256:".length)}`; +} + +function messageIdempotencyKey({ adapterRef, profileRef, identifier, payload }) { + return `sha256:${createHash("sha256") + .update(JSON.stringify({ adapterRef, profileRef, identifier, payload }), "utf8") + .digest("hex")}`; +} + +function writeWithBackpressure(socket, bytes) { + if (!Buffer.isBuffer(bytes) || bytes.length === 0 || bytes.length > 4096) { + throw new TypeError("device_gateway_adapter_ack_invalid"); + } + if (socket.write(bytes)) return Promise.resolve(); + return new Promise((resolve, reject) => { + const onDrain = () => { + socket.off("error", onError); + resolve(); + }; + const onError = (error) => { + socket.off("drain", onDrain); + reject(error); + }; + socket.once("drain", onDrain); + socket.once("error", onError); + }); +} + function normalizeHealthHost(value, fallback) { const normalized = String(value || fallback).trim(); if (!["127.0.0.1", "::1", "0.0.0.0", "::"].includes(normalized)) { diff --git a/device-plane/services/device-gateway/src/server.mjs b/device-plane/services/device-gateway/src/server.mjs index 9eb97bb..dd0d65a 100644 --- a/device-plane/services/device-gateway/src/server.mjs +++ b/device-plane/services/device-gateway/src/server.mjs @@ -1,11 +1,14 @@ import { readFile } from "node:fs/promises"; -import { createCoreDiscoveryClient } from "./core-client.mjs"; +import { + DEVICE_ADAPTER_CATALOG, +} from "../../../packages/device-adapter-catalog/src/index.mjs"; +import { createCoreGatewayClient } from "./core-client.mjs"; import { createDeviceGatewayRuntime } from "./runtime.mjs"; const config = await readConfig(); -const onDiscovery = config.listenEnabled - ? createCoreDiscoveryClient({ +const coreClient = config.listenEnabled + ? createCoreGatewayClient({ coreUrl: config.coreUrl, gatewayToken: config.gatewayToken, timeoutMs: config.coreTimeoutMs, @@ -14,17 +17,25 @@ const onDiscovery = config.listenEnabled const runtime = createDeviceGatewayRuntime({ listenEnabled: config.listenEnabled, publicIngressEnabled: config.publicIngressEnabled, + coreChannelAuthenticated: false, + adapterRegistry: DEVICE_ADAPTER_CATALOG.registry, + protocolProfileRef: config.protocolProfileRef, + edgeRef: config.edgeRef, + routeRef: config.routeRef, healthHost: config.healthHost, healthPort: config.healthPort, tcpHost: config.tcpHost, tcpPort: config.tcpPort, maxBufferedBytes: config.maxBufferedBytes, + maxAggregateBufferedBytes: config.maxAggregateBufferedBytes, maxConcurrentSessions: config.maxConcurrentSessions, maxSessionsPerAddress: config.maxSessionsPerAddress, maxConnectionsPerMinutePerAddress: config.maxConnectionsPerMinutePerAddress, + maxTrackedSourceAddresses: config.maxTrackedSourceAddresses, sessionTimeoutMs: config.sessionTimeoutMs, - onDiscovery, + onDiscovery: coreClient?.observeDiscovery, + onMessage: coreClient?.acceptMessage, }); const addresses = await runtime.start(); @@ -33,7 +44,7 @@ console.log(JSON.stringify({ health: addresses.healthAddress, tcp: addresses.tcpAddress, publicIngress: config.publicIngressEnabled - ? "discovery-only" + ? "telemetry-ingest" : "disabled", commandTransport: "disabled", })); @@ -58,6 +69,17 @@ async function readConfig() { return { listenEnabled, publicIngressEnabled, + protocolProfileRef: String( + process.env.DEVICE_GATEWAY_PROTOCOL_PROFILE_REF + || DEVICE_ADAPTER_CATALOG.defaultProfileRef, + ), + edgeRef: listenEnabled + ? requiredValue( + process.env.DEVICE_GATEWAY_EDGE_REF, + "device_gateway_edge_ref_required", + ) + : "edge:disabled", + routeRef: String(process.env.DEVICE_GATEWAY_ROUTE_REF || ""), healthHost: String( process.env.DEVICE_GATEWAY_HEALTH_HOST || "127.0.0.1", ), @@ -71,17 +93,25 @@ async function readConfig() { process.env.DEVICE_GATEWAY_MAX_BUFFERED_BYTES, 65536, ), + maxAggregateBufferedBytes: parsePositiveInt( + process.env.DEVICE_GATEWAY_MAX_AGGREGATE_BUFFERED_BYTES, + 32 * 1024 * 1024, + ), maxConcurrentSessions: parsePositiveInt( process.env.DEVICE_GATEWAY_MAX_SESSIONS, - 100, + 128, ), maxSessionsPerAddress: parsePositiveInt( process.env.DEVICE_GATEWAY_MAX_SESSIONS_PER_ADDRESS, - 10, + 16, ), maxConnectionsPerMinutePerAddress: parsePositiveInt( process.env.DEVICE_GATEWAY_MAX_CONNECTIONS_PER_MINUTE_PER_ADDRESS, - 30, + 60, + ), + maxTrackedSourceAddresses: parsePositiveInt( + process.env.DEVICE_GATEWAY_MAX_TRACKED_SOURCE_ADDRESSES, + 2048, ), sessionTimeoutMs: parsePositiveInt( process.env.DEVICE_GATEWAY_SESSION_TIMEOUT_MS, diff --git a/device-plane/services/device-gateway/test/core-client.test.mjs b/device-plane/services/device-gateway/test/core-client.test.mjs index e94960f..cba6b1c 100644 --- a/device-plane/services/device-gateway/test/core-client.test.mjs +++ b/device-plane/services/device-gateway/test/core-client.test.mjs @@ -1,7 +1,10 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { createCoreDiscoveryClient } from "../src/core-client.mjs"; +import { + createCoreDiscoveryClient, + createCoreGatewayClient, +} from "../src/core-client.mjs"; const gatewayToken = "test-only-gateway-token-with-32-bytes"; @@ -41,14 +44,14 @@ test("posts a discovery through the authenticated internal Core boundary", async assert.equal(discovery.lifecycleState, "quarantine"); }); -test("fails closed when Core does not return a quarantine view", async () => { +test("fails closed when Core does not return an accepted discovery view", async () => { const observe = createCoreDiscoveryClient({ coreUrl: "http://device-control-core:18120", gatewayToken, fetchImpl: async () => new Response(JSON.stringify({ ok: true, discovery: { - lifecycleState: "claimed", + lifecycleState: "observed", commandTransport: "disabled", }, }), { status: 200 }), @@ -58,3 +61,58 @@ test("fails closed when Core does not return a quarantine view", async () => { /device_gateway_core_ingest_contract_invalid/, ); }); + +test("accepts a package only through the explicit Core acceptance contract", async () => { + let captured; + const client = createCoreGatewayClient({ + coreUrl: "http://device-control-core:18120", + gatewayToken, + fetchImpl: async (url, options) => { + captured = { url, options }; + const message = JSON.parse(options.body); + return new Response(JSON.stringify({ + ok: true, + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test-001", + idempotencyKey: message.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, + }), { status: 201 }); + }, + }); + const message = { idempotencyKey: `sha256:${"a".repeat(64)}` }; + const acceptance = await client.acceptMessage(message); + assert.equal( + captured.url, + "http://device-control-core:18120/internal/v1/gateway/messages:accept", + ); + assert.equal(acceptance.status, "accepted"); + assert.equal(acceptance.idempotencyKey, message.idempotencyKey); +}); + +test("rejects a mismatched or non-durable Core package response", async () => { + const client = createCoreGatewayClient({ + coreUrl: "http://device-control-core:18120", + gatewayToken, + fetchImpl: async () => new Response(JSON.stringify({ + ok: true, + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test-001", + idempotencyKey: `sha256:${"b".repeat(64)}`, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, + }), { status: 201 }), + }); + await assert.rejects( + () => client.acceptMessage({ + idempotencyKey: `sha256:${"a".repeat(64)}`, + }), + /device_gateway_core_acceptance_mismatch/, + ); +}); diff --git a/device-plane/services/device-gateway/test/discovery-integration.test.mjs b/device-plane/services/device-gateway/test/discovery-integration.test.mjs index f431c14..4af836c 100644 --- a/device-plane/services/device-gateway/test/discovery-integration.test.mjs +++ b/device-plane/services/device-gateway/test/discovery-integration.test.mjs @@ -2,10 +2,13 @@ import assert from "node:assert/strict"; import { connect } from "node:net"; import test from "node:test"; +import { + DEVICE_ADAPTER_CATALOG, +} from "../../../packages/device-adapter-catalog/src/index.mjs"; import { createControlCoreApp, } from "../../device-control-core/src/app.mjs"; -import { createCoreDiscoveryClient } from "../src/core-client.mjs"; +import { createCoreGatewayClient } from "../src/core-client.mjs"; import { createDeviceGatewayRuntime } from "../src/runtime.mjs"; const gatewayToken = "test-only-gateway-token-with-32-bytes"; @@ -21,6 +24,7 @@ const specificationPackage = Buffer.from( test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", async () => { let stored; + let storedMessage; const core = createControlCoreApp({ discoveryIngestEnabled: true, gatewayToken, @@ -37,11 +41,22 @@ test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", as }, }; }, + acceptAdapterMessage: async (value) => { + storedMessage = value; + return { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:integration-001", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }; + }, }, }); await listen(core); const coreAddress = core.address(); - const observe = createCoreDiscoveryClient({ + const observe = createCoreGatewayClient({ coreUrl: `http://127.0.0.1:${coreAddress.port}`, gatewayToken, }); @@ -51,8 +66,13 @@ test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", as tcpPort: 0, listenEnabled: true, publicIngressEnabled: true, + coreChannelAuthenticated: true, + adapterRegistry: DEVICE_ADAPTER_CATALOG.registry, + protocolProfileRef: DEVICE_ADAPTER_CATALOG.defaultProfileRef, + edgeRef: "edge:integration-001", now: () => new Date(0x52db95de * 1000), - onDiscovery: observe, + onDiscovery: observe.observeDiscovery, + onMessage: observe.acceptMessage, }); const addresses = await gateway.start(); try { @@ -73,6 +93,12 @@ test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", as JSON.stringify(stored).includes("865209039777769"), false, ); + assert.equal(storedMessage.safeView.messageType, "telemetry.package"); + assert.equal(storedMessage.safeView.identifier.masked, "***********7769"); + assert.equal( + JSON.stringify(storedMessage).includes("865209039777769"), + false, + ); } finally { await gateway.stop(); await close(core); diff --git a/device-plane/services/device-gateway/test/runtime.test.mjs b/device-plane/services/device-gateway/test/runtime.test.mjs index f0a390a..7581d2e 100644 --- a/device-plane/services/device-gateway/test/runtime.test.mjs +++ b/device-plane/services/device-gateway/test/runtime.test.mjs @@ -2,6 +2,9 @@ import assert from "node:assert/strict"; import { connect } from "node:net"; import test from "node:test"; +import { + DEVICE_ADAPTER_CATALOG, +} from "../../../packages/device-adapter-catalog/src/index.mjs"; import { createDeviceGatewayRuntime } from "../src/runtime.mjs"; const specificationHeader = Buffer.from( @@ -34,7 +37,7 @@ test("baseline health exposes no public ingress and no command transport", async } }); -test("discovery-only ingress persists HEADER2 before acknowledging packages", async () => { +test("telemetry ingress persists HEADER2 before acknowledging packages", async () => { const captured = []; const runtime = createDeviceGatewayRuntime({ healthPort: 0, @@ -42,8 +45,14 @@ test("discovery-only ingress persists HEADER2 before acknowledging packages", as tcpPort: 0, listenEnabled: true, publicIngressEnabled: true, + coreChannelAuthenticated: true, now: () => new Date(0x52db95de * 1000), - onDiscovery: async (value) => captured.push(value), + ...gatewayAdapterOptions(), + onDiscovery: async (value) => { + captured.push(value); + return { lifecycleState: "quarantine" }; + }, + onMessage: async (message) => acceptanceFor(message), }); const addresses = await runtime.start(); const client = await connectAndCollect(addresses.tcpAddress.port); @@ -70,17 +79,18 @@ test("discovery-only ingress persists HEADER2 before acknowledging packages", as "7B00017D", ); assert.equal(runtime.status().totalDiscoveries, 1); + assert.equal(runtime.status().totalMessagesAccepted, 1); assert.equal(runtime.status().totalPackagesAcknowledged, 1); assert.equal(runtime.status().commandTransport, "disabled"); - assert.equal(runtime.status().publicIngress, "discovery-only"); + assert.equal(runtime.status().publicIngress, "telemetry-ingest"); const response = await fetch( `http://127.0.0.1:${addresses.healthAddress.port}/healthz`, ); const body = await response.json(); assert.equal(body.framing, "verified-read-only"); - assert.equal(body.tcpListener, "discovery-only"); - assert.equal(body.publicIngress, "discovery-only"); + assert.equal(body.tcpListener, "telemetry-ingest"); + assert.equal(body.publicIngress, "telemetry-ingest"); assert.equal(body.commandTransport, "disabled"); } finally { client.socket.destroy(); @@ -94,7 +104,12 @@ test("does not acknowledge malformed or unverified initial bytes", async () => { healthPort: 0, tcpPort: 0, listenEnabled: true, - onDiscovery: async (value) => captured.push(value), + ...gatewayAdapterOptions(), + onDiscovery: async (value) => { + captured.push(value); + return { lifecycleState: "quarantine" }; + }, + onMessage: async (message) => acceptanceFor(message), }); const addresses = await runtime.start(); try { @@ -110,14 +125,137 @@ test("does not acknowledge malformed or unverified initial bytes", async () => { } }); -test("public ingress requires an authenticated discovery sink", () => { +test("does not acknowledge a header when Core rejects discovery", async () => { + const runtime = createDeviceGatewayRuntime({ + healthPort: 0, + tcpPort: 0, + listenEnabled: true, + ...gatewayAdapterOptions(), + onDiscovery: async () => { + throw new Error("core_unavailable"); + }, + onMessage: async (message) => acceptanceFor(message), + }); + const addresses = await runtime.start(); + try { + const received = await sendAndCollect( + addresses.tcpAddress.port, + specificationHeader, + ); + assert.equal(received.length, 0); + assert.equal(runtime.status().totalDiscoveries, 0); + assert.equal(runtime.status().totalRejected, 1); + } finally { + await runtime.stop(); + } +}); + +test("does not acknowledge a package when durable Core acceptance fails", async () => { + const runtime = createDeviceGatewayRuntime({ + healthPort: 0, + tcpPort: 0, + listenEnabled: true, + now: () => new Date(0x52db95de * 1000), + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "quarantine" }), + onMessage: async () => { + throw new Error("core_commit_failed"); + }, + }); + const addresses = await runtime.start(); + try { + const received = await sendAndCollect( + addresses.tcpAddress.port, + Buffer.concat([specificationHeader, specificationPackage]), + ); + assert.equal( + received.toString("hex").toUpperCase(), + "7B0400A0DE95DB527D", + ); + assert.equal(runtime.status().totalMessagesAccepted, 0); + assert.equal(runtime.status().totalPackagesAcknowledged, 0); + assert.equal(runtime.status().totalRejected, 1); + } finally { + await runtime.stop(); + } +}); + +test("acknowledges an idempotent Core replay as accepted delivery", async () => { + const runtime = createDeviceGatewayRuntime({ + healthPort: 0, + tcpPort: 0, + listenEnabled: true, + now: () => new Date(0x52db95de * 1000), + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "quarantine" }), + onMessage: async (message) => acceptanceFor(message, true), + }); + const addresses = await runtime.start(); + try { + const received = await sendAndCollect( + addresses.tcpAddress.port, + Buffer.concat([specificationHeader, specificationPackage]), + ); + assert.equal( + received.toString("hex").toUpperCase(), + "7B0400A0DE95DB527D7B00017D", + ); + assert.equal(runtime.status().totalMessagesAccepted, 1); + assert.equal(runtime.status().totalPackagesAcknowledged, 1); + } finally { + await runtime.stop(); + } +}); + +test("closes an oversized tracker buffer and releases its aggregate budget", async () => { + const runtime = createDeviceGatewayRuntime({ + healthPort: 0, + tcpPort: 0, + listenEnabled: true, + maxBufferedBytes: 1024, + maxAggregateBufferedBytes: 1024, + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "quarantine" }), + onMessage: async (message) => acceptanceFor(message), + }); + const addresses = await runtime.start(); + try { + const received = await sendAndCollect( + addresses.tcpAddress.port, + Buffer.alloc(1025, 0x01), + ); + assert.equal(received.length, 0); + assert.equal(runtime.status().totalRejected, 1); + assert.equal(runtime.status().totalBufferedBytes, 0); + assert.equal(runtime.status().activeSessions, 0); + } finally { + await runtime.stop(); + } +}); + +test("public ingress requires both discovery and durable message acceptance sinks", () => { + assert.throws( + () => createDeviceGatewayRuntime({ + listenEnabled: true, + publicIngressEnabled: true, + coreChannelAuthenticated: true, + tcpHost: "0.0.0.0", + }), + /device_gateway_core_acceptance_sink_required/, + ); +}); + +test("public ingress cannot start on the legacy bearer HTTP Core client", () => { assert.throws( () => createDeviceGatewayRuntime({ listenEnabled: true, publicIngressEnabled: true, tcpHost: "0.0.0.0", + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "quarantine" }), + onMessage: async (message) => acceptanceFor(message), }), - /device_gateway_discovery_sink_required/, + /device_gateway_authenticated_core_channel_required/, ); }); @@ -126,6 +264,9 @@ test("baseline rejects non-loopback binding", () => { () => createDeviceGatewayRuntime({ listenEnabled: true, tcpHost: "0.0.0.0", + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "quarantine" }), + onMessage: async (message) => acceptanceFor(message), }), /device_gateway_baseline_loopback_only/, ); @@ -192,3 +333,22 @@ function sendAndCollect(port, payload) { socket.on("error", reject); }); } + +function gatewayAdapterOptions() { + return { + adapterRegistry: DEVICE_ADAPTER_CATALOG.registry, + protocolProfileRef: DEVICE_ADAPTER_CATALOG.defaultProfileRef, + edgeRef: "edge:test-001", + }; +} + +function acceptanceFor(message, replayed = false) { + return { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test-001", + idempotencyKey: message.idempotencyKey, + status: "accepted", + replayed, + acceptedAt: "2026-08-11T12:00:00.000Z", + }; +}