feat(device-plane): harden edge channel rotation

This commit is contained in:
Codex
2026-08-11 19:37:32 +03:00
parent 50b9179fe4
commit 1124c15216
7 changed files with 365 additions and 34 deletions
@@ -22,6 +22,8 @@
"edgeCertificatePinRequired": true, "edgeCertificatePinRequired": true,
"coreCertificateAllowlistRequired": true, "coreCertificateAllowlistRequired": true,
"unknownOrRevokedIdentity": "reject", "unknownOrRevokedIdentity": "reject",
"rotation": "one-active-plus-one-staged-generation",
"retiredFingerprint": "reject",
"privateKeysInSource": false, "privateKeysInSource": false,
"privateKeysInArtifact": false "privateKeysInArtifact": false
}, },
@@ -42,14 +44,17 @@
"delivery": "at-least-once-with-core-idempotency" "delivery": "at-least-once-with-core-idempotency"
}, },
"sourceAcceptance": { "sourceAcceptance": {
"devicePlaneTestsPassed": 189, "devicePlaneTestsPassed": 193,
"tls13MutualAuthenticationTested": true, "tls13MutualAuthenticationTested": true,
"keepaliveTested": true, "keepaliveTested": true,
"disconnectReconnectTested": true, "disconnectReconnectTested": true,
"certificateRotationOverlapAndRetirementTested": true,
"idempotentReplayTested": true, "idempotentReplayTested": true,
"unknownAndRevokedIdentityTested": true, "unknownAndRevokedIdentityTested": true,
"oversizedEnvelopeTested": true, "oversizedEnvelopeTested": true,
"coreUnavailableRejectionTested": true, "coreUnavailableRejectionTested": true,
"crossSessionProgressAndPerSessionOrderingTested": true,
"boundedAcceptanceWindowTested": true,
"existingCoreImageBuild": "passed-no-cache", "existingCoreImageBuild": "passed-no-cache",
"existingGatewayImageBuild": "passed-no-cache" "existingGatewayImageBuild": "passed-no-cache"
}, },
@@ -216,6 +216,45 @@ export function normalizeCertificateFingerprint(value) {
return compact.match(/.{2}/g).join(":"); return compact.match(/.{2}/g).join(":");
} }
export function normalizeCertificateIdentities(value) {
if (!Array.isArray(value) || value.length < 1 || value.length > 2) {
throw new TypeError("device_edge_channel_certificate_identities_invalid");
}
const generations = new Set();
const fingerprints = new Set();
let activeCount = 0;
const identities = value.map((identity) => {
assertPlainObject(identity, "device_edge_channel_certificate_identity");
rejectUnexpectedKeys(
identity,
new Set(["generationRef", "fingerprint", "status"]),
);
if (!["active", "staged"].includes(identity.status)) {
throw new TypeError("device_edge_channel_certificate_identity_status_invalid");
}
if (identity.status === "active") activeCount += 1;
const generationRef = normalizeRef(
identity.generationRef,
"certificate_generation_ref",
);
const fingerprint = normalizeCertificateFingerprint(identity.fingerprint);
if (generations.has(generationRef) || fingerprints.has(fingerprint)) {
throw new TypeError("device_edge_channel_certificate_identity_duplicate");
}
generations.add(generationRef);
fingerprints.add(fingerprint);
return Object.freeze({
generationRef,
fingerprint,
status: identity.status,
});
});
if (activeCount !== 1) {
throw new TypeError("device_edge_channel_active_certificate_identity_invalid");
}
return Object.freeze(identities);
}
export function nextReconnectDelay(attempt, options = {}) { export function nextReconnectDelay(attempt, options = {}) {
const normalizedAttempt = Number(attempt); const normalizedAttempt = Number(attempt);
if (!Number.isSafeInteger(normalizedAttempt) || normalizedAttempt < 0) { if (!Number.isSafeInteger(normalizedAttempt) || normalizedAttempt < 0) {
@@ -6,6 +6,7 @@ import {
createChannelEnvelopeDecoder, createChannelEnvelopeDecoder,
encodeChannelEnvelope, encodeChannelEnvelope,
nextReconnectDelay, nextReconnectDelay,
normalizeCertificateIdentities,
normalizeChannelEnvelope, normalizeChannelEnvelope,
} from "../src/index.mjs"; } from "../src/index.mjs";
@@ -75,3 +76,31 @@ test("uses bounded jittered exponential reconnect delays", () => {
random: () => 0, random: () => 0,
}), 15_000); }), 15_000);
}); });
test("allows exactly one active and at most one staged certificate generation", () => {
const activeFingerprint = "AA:".repeat(31) + "AA";
const stagedFingerprint = "BB:".repeat(31) + "BB";
const identities = normalizeCertificateIdentities([
{
generationRef: "trust-generation:1",
fingerprint: activeFingerprint,
status: "active",
},
{
generationRef: "trust-generation:2",
fingerprint: stagedFingerprint,
status: "staged",
},
]);
assert.equal(identities.length, 2);
assert.equal(identities[0].status, "active");
assert.equal(identities[1].status, "staged");
assert.throws(() => normalizeCertificateIdentities([
{ ...identities[0], status: "staged" },
identities[1],
]), /active_certificate_identity_invalid/);
assert.throws(() => normalizeCertificateIdentities([
identities[0],
{ ...identities[1], status: "active" },
]), /active_certificate_identity_invalid/);
});
@@ -74,6 +74,11 @@ test("records source acceptance without opening an Edge or tracker port", async
assert.equal(acceptance.transport.tls, "TLSv1.3-mutual-authentication"); assert.equal(acceptance.transport.tls, "TLSv1.3-mutual-authentication");
assert.equal(acceptance.identity.privateKeysInSource, false); assert.equal(acceptance.identity.privateKeysInSource, false);
assert.equal(acceptance.identity.privateKeysInArtifact, false); assert.equal(acceptance.identity.privateKeysInArtifact, false);
assert.equal(
acceptance.identity.rotation,
"one-active-plus-one-staged-generation",
);
assert.equal(acceptance.identity.retiredFingerprint, "reject");
assert.equal(acceptance.runtime.mutationInThisTransition, false); assert.equal(acceptance.runtime.mutationInThisTransition, false);
assert.equal(acceptance.runtime.edgePort8443Published, false); assert.equal(acceptance.runtime.edgePort8443Published, false);
assert.equal(acceptance.runtime.trackerPort9921Published, false); assert.equal(acceptance.runtime.trackerPort9921Published, false);
@@ -1,4 +1,4 @@
import { randomUUID } from "node:crypto"; import { randomUUID, X509Certificate } from "node:crypto";
import { createSecureServer } from "node:http2"; import { createSecureServer } from "node:http2";
import { import {
@@ -20,6 +20,7 @@ const CHANNEL_PROFILE_REF = "channel.control.v1";
export function createDeviceEdgeChannelServer(options = {}) { export function createDeviceEdgeChannelServer(options = {}) {
const config = normalizeConfig(options); const config = normalizeConfig(options);
let trust = config.trust;
const pending = new Map(); const pending = new Map();
let active = null; let active = null;
let started = false; let started = false;
@@ -32,9 +33,9 @@ export function createDeviceEdgeChannelServer(options = {}) {
let totalProtocolFailures = 0; let totalProtocolFailures = 0;
const server = createSecureServer({ const server = createSecureServer({
key: config.tls.key, key: trust.key,
cert: config.tls.cert, cert: trust.cert,
ca: config.tls.ca, ca: trust.ca,
minVersion: "TLSv1.3", minVersion: "TLSv1.3",
maxVersion: "TLSv1.3", maxVersion: "TLSv1.3",
allowHTTP1: false, allowHTTP1: false,
@@ -88,6 +89,7 @@ export function createDeviceEdgeChannelServer(options = {}) {
send(state, "channel.hello", { send(state, "channel.hello", {
status: "ready", status: "ready",
transport: "http2-mtls", transport: "http2-mtls",
trustGeneration: trust.generationRef,
commandTransport: "disabled", commandTransport: "disabled",
}, { }, {
trackerSessionId: CHANNEL_TRACKER_SESSION_ID, trackerSessionId: CHANNEL_TRACKER_SESSION_ID,
@@ -179,6 +181,8 @@ export function createDeviceEdgeChannelServer(options = {}) {
channel: active?.accepted ? "accepted" : active ? "negotiating" : "absent", channel: active?.accepted ? "accepted" : active ? "negotiating" : "absent",
edgeRegistrationId: config.edgeRegistrationId, edgeRegistrationId: config.edgeRegistrationId,
channelGeneration: config.channelGeneration, channelGeneration: config.channelGeneration,
trustGeneration: trust.generationRef,
edgeCertificateFingerprint: trust.certificateFingerprint,
pendingAcceptances: pending.size, pendingAcceptances: pending.size,
channelsAccepted: totalChannelsAccepted, channelsAccepted: totalChannelsAccepted,
channelsRejected: totalChannelsRejected, channelsRejected: totalChannelsRejected,
@@ -190,6 +194,30 @@ export function createDeviceEdgeChannelServer(options = {}) {
commandTransport: "disabled", commandTransport: "disabled",
}); });
}, },
rotateTrust(next) {
const nextTrust = normalizeTrust(next);
if (nextTrust.generationRef === trust.generationRef) {
throw new TypeError("device_edge_channel_trust_generation_unchanged");
}
server.setSecureContext({
key: nextTrust.key,
cert: nextTrust.cert,
ca: nextTrust.ca,
minVersion: "TLSv1.3",
maxVersion: "TLSv1.3",
});
trust = nextTrust;
const current = active;
if (current) {
current.stream.close();
closeActive(current);
}
return Object.freeze({
trustGeneration: trust.generationRef,
edgeCertificateFingerprint: trust.certificateFingerprint,
channel: "reconnect-required",
});
},
disconnectActiveChannel() { disconnectActiveChannel() {
active?.stream.close(); active?.stream.close();
}, },
@@ -328,7 +356,7 @@ export function createDeviceEdgeChannelServer(options = {}) {
} catch { } catch {
return false; return false;
} }
return config.tls.allowedCoreFingerprints.has(fingerprint); return trust.allowedCoreFingerprints.has(fingerprint);
} }
function protocolFailure(state) { function protocolFailure(state) {
@@ -368,7 +396,10 @@ function normalizeConfig(options) {
options.channelGeneration, options.channelGeneration,
"channel_generation", "channel_generation",
); );
const tls = normalizeTls(options.tls); const trust = normalizeTrust({
...options.tls,
generationRef: options.trustGeneration,
});
const keepaliveMs = normalizeDuration(options.keepaliveMs, 10, const keepaliveMs = normalizeDuration(options.keepaliveMs, 10,
DEVICE_EDGE_CHANNEL_LIMITS.keepaliveMs, "keepalive"); DEVICE_EDGE_CHANNEL_LIMITS.keepaliveMs, "keepalive");
const deadPeerMs = normalizeDuration(options.deadPeerMs, keepaliveMs * 2, const deadPeerMs = normalizeDuration(options.deadPeerMs, keepaliveMs * 2,
@@ -379,7 +410,7 @@ function normalizeConfig(options) {
return Object.freeze({ return Object.freeze({
edgeRegistrationId, edgeRegistrationId,
channelGeneration, channelGeneration,
tls, trust,
host: normalizeHost(options.host ?? "127.0.0.1"), host: normalizeHost(options.host ?? "127.0.0.1"),
port: normalizePort(options.port ?? 8443), port: normalizePort(options.port ?? 8443),
keepaliveMs, keepaliveMs,
@@ -411,7 +442,7 @@ function normalizeConfig(options) {
}); });
} }
function normalizeTls(value) { function normalizeTrust(value) {
if (!value || typeof value !== "object") { if (!value || typeof value !== "object") {
throw new TypeError("device_edge_channel_tls_invalid"); throw new TypeError("device_edge_channel_tls_invalid");
} }
@@ -420,16 +451,38 @@ function normalizeTls(value) {
throw new TypeError(`device_edge_channel_tls_${key}_invalid`); throw new TypeError(`device_edge_channel_tls_${key}_invalid`);
} }
} }
if (!Array.isArray(value.allowedCoreFingerprints) || value.allowedCoreFingerprints.length < 1) { if (
!Array.isArray(value.allowedCoreFingerprints)
|| value.allowedCoreFingerprints.length < 1
|| value.allowedCoreFingerprints.length > 2
) {
throw new TypeError("device_edge_channel_core_identity_allowlist_invalid"); throw new TypeError("device_edge_channel_core_identity_allowlist_invalid");
} }
const generationRef = normalizeRef(
value.generationRef,
"trust_generation",
);
let certificateFingerprint;
try {
certificateFingerprint = normalizeCertificateFingerprint(
new X509Certificate(value.cert).fingerprint256,
);
} catch {
throw new TypeError("device_edge_channel_tls_cert_invalid");
}
const allowedCoreFingerprints = new Set(
value.allowedCoreFingerprints.map(normalizeCertificateFingerprint),
);
if (allowedCoreFingerprints.size !== value.allowedCoreFingerprints.length) {
throw new TypeError("device_edge_channel_core_identity_allowlist_duplicate");
}
return Object.freeze({ return Object.freeze({
generationRef,
key: value.key, key: value.key,
cert: value.cert, cert: value.cert,
ca: value.ca, ca: value.ca,
allowedCoreFingerprints: new Set( certificateFingerprint,
value.allowedCoreFingerprints.map(normalizeCertificateFingerprint), allowedCoreFingerprints,
),
}); });
} }
@@ -94,6 +94,66 @@ test("keeps the channel alive and reconnects without losing idempotency", async
} }
}); });
test("rotates the Edge certificate through staged overlap and rejects retired identity", async () => {
const edge = createEdgeServer();
const address = await edge.start();
let registration = edgeRegistration(address, [
{
generationRef: "trust-generation:1",
fingerprint: certificates.edge.fingerprint,
status: "active",
},
{
generationRef: "trust-generation:2",
fingerprint: certificates.edgeNext.fingerprint,
status: "staged",
},
]);
const core = createCoreClient({
address,
registrationProvider: async () => registration,
});
try {
await core.start();
await core.waitForReady(2_000);
await waitFor(() => edge.status().channel === "accepted", 2_000);
assert.equal(core.status().edgeTrustGeneration, "trust-generation:1");
edge.rotateTrust({
generationRef: "trust-generation:2",
key: certificates.edgeNext.key,
cert: certificates.edgeNext.cert,
ca: certificates.ca,
allowedCoreFingerprints: [certificates.core.fingerprint],
});
await waitFor(() => core.status().connectionAttempts >= 2
&& core.status().channel === "accepted"
&& core.status().edgeTrustGeneration === "trust-generation:2"
&& edge.status().channel === "accepted", 2_000);
registration = edgeRegistration(address, [{
generationRef: "trust-generation:2",
fingerprint: certificates.edgeNext.fingerprint,
status: "active",
}]);
assert.equal((await edge.submitAdapterMessage(adapterMessage())).status, "accepted");
const failuresBeforeRollback = core.status().protocolFailures;
edge.rotateTrust({
generationRef: "trust-generation:3",
key: certificates.edge.key,
cert: certificates.edge.cert,
ca: certificates.ca,
allowedCoreFingerprints: [certificates.core.fingerprint],
});
await waitFor(() => core.status().protocolFailures > failuresBeforeRollback, 2_000);
assert.notEqual(core.status().channel, "accepted");
} finally {
await core.stop();
await edge.stop();
}
});
test("rejects a revoked Edge registration before opening a channel", async () => { test("rejects a revoked Edge registration before opening a channel", async () => {
const edge = createEdgeServer(); const edge = createEdgeServer();
const address = await edge.start(); const address = await edge.start();
@@ -167,6 +227,92 @@ test("returns a conclusive rejection when Core cannot accept a package", async (
} }
}); });
test("isolates tracker session ordering while allowing cross-session progress", async () => {
let releaseSlow;
const slowGate = new Promise((resolve) => {
releaseSlow = resolve;
});
const calls = [];
const pair = await startPair({
acceptMessage: async (message) => {
calls.push(message.sessionRef);
if (message.sessionRef === "session:slow") await slowGate;
return acceptanceFor(message, false);
},
});
try {
let slowResolved = false;
const slow = pair.edge.submitAdapterMessage(adapterMessage({
sessionRef: "session:slow",
messageRef: "message:slow-1",
idempotencyKey: `sha256:${"c".repeat(64)}`,
})).then((value) => {
slowResolved = true;
return value;
});
await waitFor(() => calls.includes("session:slow"), 500);
const fast = await pair.edge.submitAdapterMessage(adapterMessage({
sessionRef: "session:fast",
messageRef: "message:fast-1",
idempotencyKey: `sha256:${"d".repeat(64)}`,
}));
assert.equal(fast.status, "accepted");
assert.equal(slowResolved, false);
releaseSlow();
assert.equal((await slow).status, "accepted");
} finally {
releaseSlow?.();
await stopPair(pair);
}
});
test("preserves per-session order and applies a bounded acceptance window", async () => {
let releaseFirst;
const firstGate = new Promise((resolve) => {
releaseFirst = resolve;
});
const calls = [];
const pair = await startPair({
maxPendingAcceptances: 2,
acceptMessage: async (message) => {
calls.push(message.messageRef);
if (message.messageRef === "message:ordered-1") await firstGate;
return acceptanceFor(message, false);
},
});
try {
const first = pair.edge.submitAdapterMessage(adapterMessage({
sessionRef: "session:ordered",
messageRef: "message:ordered-1",
idempotencyKey: `sha256:${"e".repeat(64)}`,
sequence: 1,
}));
await waitFor(() => calls.length === 1, 500);
const second = pair.edge.submitAdapterMessage(adapterMessage({
sessionRef: "session:ordered",
messageRef: "message:ordered-2",
idempotencyKey: `sha256:${"f".repeat(64)}`,
sequence: 2,
}));
await delay(30);
assert.deepEqual(calls, ["message:ordered-1"]);
await assert.rejects(
pair.edge.submitAdapterMessage(adapterMessage({
sessionRef: "session:overflow",
messageRef: "message:overflow-1",
idempotencyKey: `sha256:${"1".repeat(64)}`,
})),
/acceptance_window_full/,
);
releaseFirst();
await Promise.all([first, second]);
assert.deepEqual(calls, ["message:ordered-1", "message:ordered-2"]);
} finally {
releaseFirst?.();
await stopPair(pair);
}
});
test("closes the logical session on an unknown message kind", async () => { test("closes the logical session on an unknown message kind", async () => {
const edge = createEdgeServer(); const edge = createEdgeServer();
const address = await edge.start(); const address = await edge.start();
@@ -221,34 +367,41 @@ async function startPair(options = {}) {
} }
function createEdgeServer(options = {}) { function createEdgeServer(options = {}) {
const edgeCertificate = options.edgeCertificate ?? certificates.edge;
return createDeviceEdgeChannelServer({ return createDeviceEdgeChannelServer({
edgeRegistrationId: "edge:pilot-1", edgeRegistrationId: "edge:pilot-1",
channelGeneration: "generation:pilot-1", channelGeneration: "generation:pilot-1",
trustGeneration: options.edgeTrustGeneration ?? "trust-generation:1",
host: "127.0.0.1", host: "127.0.0.1",
port: 0, port: 0,
tls: { tls: {
key: certificates.edge.key, key: edgeCertificate.key,
cert: certificates.edge.cert, cert: edgeCertificate.cert,
ca: certificates.ca, ca: certificates.ca,
allowedCoreFingerprints: [certificates.core.fingerprint], allowedCoreFingerprints: options.allowedCoreFingerprints
?? [certificates.core.fingerprint],
}, },
keepaliveMs: options.keepaliveMs ?? 50, keepaliveMs: options.keepaliveMs ?? 50,
deadPeerMs: options.deadPeerMs ?? 150, deadPeerMs: options.deadPeerMs ?? 150,
acceptanceTimeoutMs: 500, acceptanceTimeoutMs: 500,
maxPendingAcceptances: options.maxPendingAcceptances,
}); });
} }
function createCoreClient(options) { function createCoreClient(options) {
const clientCertificate = options.clientCertificate ?? certificates.core; const clientCertificate = options.clientCertificate ?? certificates.core;
return createDeviceGatewayCoreChannelClient({ const registration = edgeRegistration(options.address,
registration: { options.edgeCertificateIdentities ?? [{
edgeRegistrationId: "edge:pilot-1", generationRef: "trust-generation:1",
endpoint: `https://127.0.0.1:${options.address.port}/`, fingerprint: options.expectedEdgeFingerprint
servername: "localhost",
certificateFingerprint: options.expectedEdgeFingerprint
?? certificates.edge.fingerprint, ?? certificates.edge.fingerprint,
lifecycleState: options.lifecycleState ?? "active", status: "active",
}, }],
options.lifecycleState ?? "active");
return createDeviceGatewayCoreChannelClient({
...(options.registrationProvider
? { registrationProvider: options.registrationProvider }
: { registration }),
tls: { tls: {
key: clientCertificate.key, key: clientCertificate.key,
cert: clientCertificate.cert, cert: clientCertificate.cert,
@@ -271,6 +424,16 @@ function createCoreClient(options) {
}); });
} }
function edgeRegistration(address, certificateIdentities, lifecycleState = "active") {
return {
edgeRegistrationId: "edge:pilot-1",
endpoint: `https://127.0.0.1:${address.port}/`,
servername: "localhost",
certificateIdentities,
lifecycleState,
};
}
async function stopPair(pair) { async function stopPair(pair) {
await pair.core.stop(); await pair.core.stop();
await pair.edge.stop(); await pair.edge.stop();
@@ -293,7 +456,7 @@ function discoverySignal() {
}; };
} }
function adapterMessage() { function adapterMessage(overrides = {}) {
return { return {
schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA, schemaVersion: DEVICE_ADAPTER_MESSAGE_SCHEMA,
edgeRef: "edge:pilot-1", edgeRef: "edge:pilot-1",
@@ -314,6 +477,7 @@ function adapterMessage() {
byteLength: 11, byteLength: 11,
packageDigest: `sha256:${"b".repeat(64)}`, packageDigest: `sha256:${"b".repeat(64)}`,
}, },
...overrides,
}; };
} }
@@ -340,6 +504,10 @@ async function generateCertificateFixture(directory) {
"subjectAltName=DNS:localhost,IP:127.0.0.1", "subjectAltName=DNS:localhost,IP:127.0.0.1",
"extendedKeyUsage=serverAuth", "extendedKeyUsage=serverAuth",
]); ]);
await issueCertificate(directory, "edge-next", "localhost", [
"subjectAltName=DNS:localhost,IP:127.0.0.1",
"extendedKeyUsage=serverAuth",
]);
await issueCertificate(directory, "core", "nodedc-device-gateway-core", [ await issueCertificate(directory, "core", "nodedc-device-gateway-core", [
"extendedKeyUsage=clientAuth", "extendedKeyUsage=clientAuth",
]); ]);
@@ -350,6 +518,7 @@ async function generateCertificateFixture(directory) {
return { return {
ca, ca,
edge: await readCertificate(directory, "edge"), edge: await readCertificate(directory, "edge"),
edgeNext: await readCertificate(directory, "edge-next"),
core: await readCertificate(directory, "core"), core: await readCertificate(directory, "core"),
intruder: await readCertificate(directory, "intruder"), intruder: await readCertificate(directory, "intruder"),
}; };
@@ -8,6 +8,7 @@ import {
createChannelEnvelopeDecoder, createChannelEnvelopeDecoder,
encodeChannelEnvelope, encodeChannelEnvelope,
nextReconnectDelay, nextReconnectDelay,
normalizeCertificateIdentities,
normalizeCertificateFingerprint, normalizeCertificateFingerprint,
} from "../../../packages/device-edge-channel-contract/src/index.mjs"; } from "../../../packages/device-edge-channel-contract/src/index.mjs";
import { import {
@@ -80,6 +81,10 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
channel: state?.ready ? "accepted" : state ? "connecting" : "absent", channel: state?.ready ? "accepted" : state ? "connecting" : "absent",
edgeRegistrationId: state?.registration?.edgeRegistrationId ?? null, edgeRegistrationId: state?.registration?.edgeRegistrationId ?? null,
channelGeneration: state?.channelGeneration ?? null, channelGeneration: state?.channelGeneration ?? null,
edgeTrustGeneration: state?.observedEdgeIdentity?.generationRef ?? null,
edgeCertificateFingerprint:
state?.observedEdgeIdentity?.fingerprint ?? null,
activeTrackerSessionChains: state?.sessionChains.size ?? 0,
connectionAttempts: totalConnectionAttempts, connectionAttempts: totalConnectionAttempts,
channelsAccepted: totalChannelsAccepted, channelsAccepted: totalChannelsAccepted,
reconnects: totalReconnects, reconnects: totalReconnects,
@@ -117,7 +122,9 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
direction: "edge-to-core", direction: "edge-to-core",
maxEnvelopeBytes: config.maxEnvelopeBytes, maxEnvelopeBytes: config.maxEnvelopeBytes,
}), }),
sessionChains: new Map(),
channelGeneration: null, channelGeneration: null,
observedEdgeIdentity: null,
edgeSequence: 0, edgeSequence: 0,
coreSequence: 0, coreSequence: 0,
lastEdgeActivityAt: config.clock(), lastEdgeActivityAt: config.clock(),
@@ -220,6 +227,8 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
if ( if (
envelope.payload?.status !== "ready" envelope.payload?.status !== "ready"
|| envelope.payload?.transport !== "http2-mtls" || envelope.payload?.transport !== "http2-mtls"
|| envelope.payload?.trustGeneration
!== connection.observedEdgeIdentity?.generationRef
|| envelope.payload?.commandTransport !== "disabled" || envelope.payload?.commandTransport !== "disabled"
) { ) {
throw new Error("device_gateway_core_channel_hello_invalid"); throw new Error("device_gateway_core_channel_hello_invalid");
@@ -242,17 +251,34 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
return; return;
} }
if (envelope.messageKind === "channel.heartbeat") return; if (envelope.messageKind === "channel.heartbeat") return;
if (envelope.messageKind === "discovery.observed") { if (["discovery.observed", "adapter.message"].includes(envelope.messageKind)) {
await acceptDiscovery(connection, envelope); scheduleTrackerEvent(connection, envelope);
return;
}
if (envelope.messageKind === "adapter.message") {
await acceptAdapterMessage(connection, envelope);
return; return;
} }
throw new Error("device_gateway_core_edge_message_unhandled"); throw new Error("device_gateway_core_edge_message_unhandled");
} }
function scheduleTrackerEvent(connection, envelope) {
if (envelope.trackerSessionId === CHANNEL_TRACKER_SESSION_ID) {
throw new Error("device_gateway_core_tracker_session_invalid");
}
const previous = connection.sessionChains.get(envelope.trackerSessionId);
if (!previous && connection.sessionChains.size >= 128) {
throw new Error("device_gateway_core_tracker_session_limit_reached");
}
const work = (previous ?? Promise.resolve())
.then(() => envelope.messageKind === "discovery.observed"
? acceptDiscovery(connection, envelope)
: acceptAdapterMessage(connection, envelope))
.catch((error) => failConnection(connection, error))
.finally(() => {
if (connection.sessionChains.get(envelope.trackerSessionId) === work) {
connection.sessionChains.delete(envelope.trackerSessionId);
}
});
connection.sessionChains.set(envelope.trackerSessionId, work);
}
async function acceptDiscovery(connection, envelope) { async function acceptDiscovery(connection, envelope) {
try { try {
const signal = normalizeDiscoverySignal(envelope.payload?.signal); const signal = normalizeDiscoverySignal(envelope.payload?.signal);
@@ -352,9 +378,13 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
const observed = normalizeCertificateFingerprint( const observed = normalizeCertificateFingerprint(
socket.getPeerCertificate()?.fingerprint256, socket.getPeerCertificate()?.fingerprint256,
); );
if (observed !== connection.registration.certificateFingerprint) { const identity = connection.registration.certificateIdentities.find(
(candidate) => candidate.fingerprint === observed,
);
if (!identity) {
throw new Error("device_gateway_core_edge_identity_mismatch"); throw new Error("device_gateway_core_edge_identity_mismatch");
} }
connection.observedEdgeIdentity = identity;
} }
function failConnection(connection, error) { function failConnection(connection, error) {
@@ -369,6 +399,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) {
connection.closed = true; connection.closed = true;
clearInterval(connection.heartbeatTimer); clearInterval(connection.heartbeatTimer);
connection.heartbeatTimer = null; connection.heartbeatTimer = null;
connection.sessionChains.clear();
try { try {
connection.request?.close(); connection.request?.close();
} catch {} } catch {}
@@ -520,8 +551,8 @@ function normalizeRegistration(value) {
), ),
endpoint: endpoint.toString(), endpoint: endpoint.toString(),
servername, servername,
certificateFingerprint: normalizeCertificateFingerprint( certificateIdentities: normalizeCertificateIdentities(
value.certificateFingerprint, value.certificateIdentities,
), ),
lifecycleState: value.lifecycleState, lifecycleState: value.lifecycleState,
}); });