SILVER/NODEDC_PLATFORM
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(device-plane): enable B2 discovery ingress

Browse Source
This commit is contained in:
Codex
2026-07-26 00:52:31 +03:00
parent 3c538ad98c
commit 2b1795509b
19 changed files with 2375 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
device-plane/services/device-gateway/src/core-client.mjs
+75
View File
@@ -0,0 +1,75 @@
export function createCoreDiscoveryClient({
coreUrl,
gatewayToken,
timeoutMs = 5000,
fetchImpl = fetch,
} = {}) {
const endpoint = normalizeCoreEndpoint(coreUrl);
if (typeof gatewayToken !== "string" || gatewayToken.length < 32) {
throw new TypeError("device_gateway_core_token_invalid");
}
const normalizedTimeout = Number(timeoutMs);
if (
!Number.isSafeInteger(normalizedTimeout)
|| normalizedTimeout < 100
|| normalizedTimeout > 30_000
) {
throw new TypeError("device_gateway_core_timeout_invalid");
}
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) {
let url;
try {
url = new URL(String(value || ""));
} catch {
throw new TypeError("device_gateway_core_url_invalid");
}
if (url.protocol !== "http:" || url.username || url.password) {
throw new TypeError("device_gateway_core_url_invalid");
}
if (url.pathname !== "/" || url.search || url.hash) {
throw new TypeError("device_gateway_core_url_invalid");
}
url.pathname = "/internal/v1/device-discoveries:observe";
return url.toString();
}
async function readBoundedJson(response, maxBytes) {
const text = await response.text();
if (Buffer.byteLength(text, "utf8") > maxBytes) {
throw new Error("device_gateway_core_response_too_large");
}
try {
return JSON.parse(text);
} catch {
throw new Error("device_gateway_core_response_invalid");
}
}
device-plane/services/device-gateway/src/runtime.mjs
+201 -47
View File
@@ -4,18 +4,32 @@ import { createServer as createTcpServer } from "node:net";
import {
ARUSNAVI_B2_MODEL_PROFILE,
inspectUnverifiedInitialBytes,
buildB2HeaderAcknowledgement,
buildB2PackageAcknowledgement,
tryParseB2Header2,
tryParseB2Package,
} from "../../../packages/arusnavi-b2-adapter/src/index.mjs";
import {
DEVICE_DISCOVERY_SIGNAL_SCHEMA,
} from "../../../packages/device-protocol-contract/src/index.mjs";
export function createDeviceGatewayRuntime(options = {}) {
const config = normalizeConfig(options);
const sessions = new Map();
const sessionsByAddress = new Map();
const connectionWindows = new Map();
let totalAccepted = 0;
let totalRejected = 0;
let totalEvidence = 0;
let totalDiscoveries = 0;
let totalPackagesAcknowledged = 0;
const tcpServer = createTcpServer((socket) => {
if (sessions.size >= config.maxConcurrentSessions) {
const remoteAddress = normalizeRemoteAddress(socket.remoteAddress);
if (
sessions.size >= config.maxConcurrentSessions
|| currentAddressSessions(remoteAddress) >= config.maxSessionsPerAddress
|| !consumeConnectionPermit(remoteAddress)
) {
totalRejected += 1;
socket.destroy();
return;
@@ -24,40 +38,42 @@ export function createDeviceGatewayRuntime(options = {}) {
const sessionRef = `session:${randomUUID()}`;
const session = {
sessionRef,
bytes: [],
byteLength: 0,
evidenceRecorded: false,
remoteAddress,
buffer: Buffer.alloc(0),
state: "awaiting-header",
processing: false,
rejected: false,
closed: false,
};
sessions.set(socket, session);
incrementAddressSessions(remoteAddress);
totalAccepted += 1;
socket.setNoDelay(true);
socket.setTimeout(config.sessionTimeoutMs);
socket.on("data", (chunk) => {
if (session.evidenceRecorded) return;
session.byteLength += chunk.length;
if (session.byteLength > config.maxInitialBytes) {
totalRejected += 1;
socket.destroy();
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) {
rejectSession(socket, session);
return;
}
session.bytes.push(chunk);
const evidence = inspectUnverifiedInitialBytes(
Buffer.concat(session.bytes, session.byteLength),
);
session.evidenceRecorded = true;
totalEvidence += 1;
config.onEvidence?.({
sessionRef,
evidence,
});
// Until exact official framing is implemented, the gateway never sends
// acknowledgement or command bytes and never guesses an identifier.
socket.end();
if (session.processing) return;
session.processing = true;
void processSession(socket, session)
.catch(() => rejectSession(socket, session))
.finally(() => {
session.processing = false;
if (!session.closed && !session.rejected) socket.resume();
});
});
socket.on("timeout", () => socket.destroy());
socket.on("close", () => sessions.delete(socket));
socket.on("error", () => sessions.delete(socket));
socket.on("timeout", () => rejectSession(socket, session));
socket.on("close", () => closeSession(socket, session));
socket.on("error", () => closeSession(socket, session));
});
const healthServer = createHttpServer((request, response) => {
@@ -74,14 +90,17 @@ export function createDeviceGatewayRuntime(options = {}) {
service: "nodedc-device-gateway",
protocolProfile: ARUSNAVI_B2_MODEL_PROFILE.profileRef,
framing: ARUSNAVI_B2_MODEL_PROFILE.framing.status,
tcpListener: config.listenEnabled ? "internal-test-only" : "disabled",
publicIngress: "disabled",
tcpListener: config.listenEnabled ? "discovery-only" : "disabled",
publicIngress: config.publicIngressEnabled
? "discovery-only"
: "disabled",
commandTransport: "disabled",
sessions: {
active: sessions.size,
accepted: totalAccepted,
rejected: totalRejected,
evidenceRecorded: totalEvidence,
discoveries: totalDiscoveries,
packagesAcknowledged: totalPackagesAcknowledged,
},
})}\n`);
});
@@ -98,7 +117,10 @@ export function createDeviceGatewayRuntime(options = {}) {
};
},
async stop() {
for (const socket of sessions.keys()) socket.destroy();
for (const [socket, session] of sessions) {
session.closed = true;
socket.destroy();
}
await Promise.all([
closeServer(healthServer),
config.listenEnabled ? closeServer(tcpServer) : Promise.resolve(),
@@ -109,25 +131,120 @@ export function createDeviceGatewayRuntime(options = {}) {
activeSessions: sessions.size,
totalAccepted,
totalRejected,
totalEvidence,
totalDiscoveries,
totalPackagesAcknowledged,
commandTransport: "disabled",
publicIngress: "disabled",
publicIngress: config.publicIngressEnabled
? "discovery-only"
: "disabled",
};
},
};
async function processSession(socket, session) {
while (!session.closed && !session.rejected) {
if (session.state === "awaiting-header") {
const parsed = tryParseB2Header2(session.buffer);
if (parsed.status === "incomplete") return;
const observedAt = config.now().toISOString();
await config.onDiscovery?.({
schemaVersion: DEVICE_DISCOVERY_SIGNAL_SCHEMA,
sessionRef: session.sessionRef,
modelProfileRef: ARUSNAVI_B2_MODEL_PROFILE.profileRef,
protocol: ARUSNAVI_B2_MODEL_PROFILE.protocol,
observedAt,
identifier: {
kind: parsed.identifier.kind,
value: parsed.identifier.value,
},
evidence: parsed.evidence,
});
session.buffer = session.buffer.subarray(parsed.bytesConsumed);
session.state = "packages";
totalDiscoveries += 1;
socket.write(buildB2HeaderAcknowledgement(
Math.floor(new Date(observedAt).getTime() / 1000),
));
continue;
}
const parsed = tryParseB2Package(session.buffer);
if (parsed.status === "incomplete") return;
session.buffer = session.buffer.subarray(parsed.bytesConsumed);
totalPackagesAcknowledged += 1;
socket.write(buildB2PackageAcknowledgement(parsed.packageNumber));
}
}
function currentAddressSessions(remoteAddress) {
return sessionsByAddress.get(remoteAddress) || 0;
}
function incrementAddressSessions(remoteAddress) {
sessionsByAddress.set(
remoteAddress,
currentAddressSessions(remoteAddress) + 1,
);
}
function decrementAddressSessions(remoteAddress) {
const current = currentAddressSessions(remoteAddress);
if (current <= 1) {
sessionsByAddress.delete(remoteAddress);
} else {
sessionsByAddress.set(remoteAddress, current - 1);
}
}
function consumeConnectionPermit(remoteAddress) {
const nowMs = config.now().getTime();
const current = connectionWindows.get(remoteAddress);
if (!current || nowMs - current.startedAt >= 60_000) {
connectionWindows.set(remoteAddress, {
startedAt: nowMs,
count: 1,
});
return true;
}
if (current.count >= config.maxConnectionsPerMinutePerAddress) {
return false;
}
current.count += 1;
return true;
}
function rejectSession(socket, session) {
if (!session.rejected) {
session.rejected = true;
totalRejected += 1;
}
socket.destroy();
}
function closeSession(socket, session) {
if (session.closed) return;
session.closed = true;
sessions.delete(socket);
decrementAddressSessions(session.remoteAddress);
}
}
function normalizeConfig(input) {
const listenEnabled = input.listenEnabled === true;
const maxInitialBytes = parseInteger(
input.maxInitialBytes,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxInitialBytes,
1,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxInitialBytes,
"device_gateway_initial_bytes_invalid",
);
const publicIngressEnabled = input.publicIngressEnabled === true;
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");
}
return {
listenEnabled,
publicIngressEnabled,
healthHost: normalizeHealthHost(input.healthHost, "127.0.0.1"),
healthPort: parseInteger(
input.healthPort,
@@ -136,7 +253,11 @@ function normalizeConfig(input) {
65535,
"device_gateway_health_port_invalid",
),
tcpHost: normalizeTcpHost(input.tcpHost, "127.0.0.1"),
tcpHost: normalizeTcpHost(
input.tcpHost,
publicIngressEnabled ? "0.0.0.0" : "127.0.0.1",
publicIngressEnabled,
),
tcpPort: parseInteger(
input.tcpPort,
9921,
@@ -144,7 +265,13 @@ function normalizeConfig(input) {
65535,
"device_gateway_tcp_port_invalid",
),
maxInitialBytes,
maxBufferedBytes: parseInteger(
input.maxBufferedBytes,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes,
1024,
ARUSNAVI_B2_MODEL_PROFILE.framing.maxBufferedBytes,
"device_gateway_buffer_limit_invalid",
),
maxConcurrentSessions: parseInteger(
input.maxConcurrentSessions,
100,
@@ -152,6 +279,20 @@ function normalizeConfig(input) {
10000,
"device_gateway_session_limit_invalid",
),
maxSessionsPerAddress: parseInteger(
input.maxSessionsPerAddress,
10,
1,
1000,
"device_gateway_address_session_limit_invalid",
),
maxConnectionsPerMinutePerAddress: parseInteger(
input.maxConnectionsPerMinutePerAddress,
30,
1,
10000,
"device_gateway_address_rate_limit_invalid",
),
sessionTimeoutMs: parseInteger(
input.sessionTimeoutMs,
10000,
@@ -159,9 +300,10 @@ function normalizeConfig(input) {
60000,
"device_gateway_session_timeout_invalid",
),
onEvidence: typeof input.onEvidence === "function"
? input.onEvidence
onDiscovery: typeof input.onDiscovery === "function"
? input.onDiscovery
: undefined,
now: typeof input.now === "function" ? input.now : () => new Date(),
};
}
@@ -173,14 +315,26 @@ function normalizeHealthHost(value, fallback) {
return normalized;
}
function normalizeTcpHost(value, fallback) {
function normalizeTcpHost(value, fallback, publicIngressEnabled) {
const normalized = String(value || fallback).trim();
if (!["127.0.0.1", "::1"].includes(normalized)) {
throw new TypeError("device_gateway_baseline_loopback_only");
const allowed = publicIngressEnabled
? ["0.0.0.0", "::"]
: ["127.0.0.1", "::1"];
if (!allowed.includes(normalized)) {
throw new TypeError(
publicIngressEnabled
? "device_gateway_public_ingress_host_invalid"
: "device_gateway_baseline_loopback_only",
);
}
return normalized;
}
function normalizeRemoteAddress(value) {
const normalized = String(value || "unknown").trim();
return normalized.slice(0, 64) || "unknown";
}
function parseInteger(value, fallback, minimum, maximum, errorCode) {
const parsed = Number(value ?? fallback);
if (
device-plane/services/device-gateway/src/server.mjs
+101 -18
View File
@@ -1,23 +1,30 @@
import { readFile } from "node:fs/promises";
import { createCoreDiscoveryClient } from "./core-client.mjs";
import { createDeviceGatewayRuntime } from "./runtime.mjs";
const listenEnabled = parseBoolean(
process.env.DEVICE_GATEWAY_LISTEN_ENABLED,
false,
);
const config = await readConfig();
const onDiscovery = config.listenEnabled
? createCoreDiscoveryClient({
coreUrl: config.coreUrl,
gatewayToken: config.gatewayToken,
timeoutMs: config.coreTimeoutMs,
})
: undefined;
const runtime = createDeviceGatewayRuntime({
listenEnabled,
healthHost: String(process.env.DEVICE_GATEWAY_HEALTH_HOST || "127.0.0.1"),
healthPort: parsePort(process.env.DEVICE_GATEWAY_HEALTH_PORT, 18121),
tcpHost: String(process.env.DEVICE_GATEWAY_TCP_HOST || "127.0.0.1"),
tcpPort: parsePort(process.env.DEVICE_GATEWAY_TCP_PORT, 9921),
maxConcurrentSessions: parsePositiveInt(
process.env.DEVICE_GATEWAY_MAX_SESSIONS,
100,
),
sessionTimeoutMs: parsePositiveInt(
process.env.DEVICE_GATEWAY_SESSION_TIMEOUT_MS,
10000,
),
listenEnabled: config.listenEnabled,
publicIngressEnabled: config.publicIngressEnabled,
healthHost: config.healthHost,
healthPort: config.healthPort,
tcpHost: config.tcpHost,
tcpPort: config.tcpPort,
maxBufferedBytes: config.maxBufferedBytes,
maxConcurrentSessions: config.maxConcurrentSessions,
maxSessionsPerAddress: config.maxSessionsPerAddress,
maxConnectionsPerMinutePerAddress:
config.maxConnectionsPerMinutePerAddress,
sessionTimeoutMs: config.sessionTimeoutMs,
onDiscovery,
});
const addresses = await runtime.start();
@@ -25,7 +32,9 @@ console.log(JSON.stringify({
event: "device_gateway_started",
health: addresses.healthAddress,
tcp: addresses.tcpAddress,
publicIngress: "disabled",
publicIngress: config.publicIngressEnabled
? "discovery-only"
: "disabled",
commandTransport: "disabled",
}));
@@ -37,6 +46,80 @@ async function shutdown() {
process.exit(0);
}
async function readConfig() {
const listenEnabled = parseBoolean(
process.env.DEVICE_GATEWAY_LISTEN_ENABLED,
false,
);
const publicIngressEnabled = parseBoolean(
process.env.DEVICE_GATEWAY_PUBLIC_INGRESS_ENABLED,
false,
);
return {
listenEnabled,
publicIngressEnabled,
healthHost: String(
process.env.DEVICE_GATEWAY_HEALTH_HOST || "127.0.0.1",
),
healthPort: parsePort(process.env.DEVICE_GATEWAY_HEALTH_PORT, 18121),
tcpHost: String(
process.env.DEVICE_GATEWAY_TCP_HOST
|| (publicIngressEnabled ? "0.0.0.0" : "127.0.0.1"),
),
tcpPort: parsePort(process.env.DEVICE_GATEWAY_TCP_PORT, 9921),
maxBufferedBytes: parsePositiveInt(
process.env.DEVICE_GATEWAY_MAX_BUFFERED_BYTES,
65536,
),
maxConcurrentSessions: parsePositiveInt(
process.env.DEVICE_GATEWAY_MAX_SESSIONS,
100,
),
maxSessionsPerAddress: parsePositiveInt(
process.env.DEVICE_GATEWAY_MAX_SESSIONS_PER_ADDRESS,
10,
),
maxConnectionsPerMinutePerAddress: parsePositiveInt(
process.env.DEVICE_GATEWAY_MAX_CONNECTIONS_PER_MINUTE_PER_ADDRESS,
30,
),
sessionTimeoutMs: parsePositiveInt(
process.env.DEVICE_GATEWAY_SESSION_TIMEOUT_MS,
10000,
),
coreUrl: listenEnabled
? requiredValue(
process.env.DEVICE_GATEWAY_CORE_URL,
"device_gateway_core_url_required",
)
: "",
gatewayToken: listenEnabled
? await readRequiredSecretFile(
process.env.DEVICE_GATEWAY_CORE_TOKEN_FILE,
"device_gateway_core_token_file_required",
)
: "",
coreTimeoutMs: parsePositiveInt(
process.env.DEVICE_GATEWAY_CORE_TIMEOUT_MS,
5000,
),
};
}
async function readRequiredSecretFile(path, errorCode) {
const normalized = requiredValue(path, errorCode);
const value = (await readFile(normalized, "utf8")).trim();
if (value.length < 32) throw new Error(errorCode);
return value;
}
function requiredValue(value, errorCode) {
if (typeof value !== "string" || value.trim() === "") {
throw new Error(errorCode);
}
return value.trim();
}
function parsePort(value, fallback) {
const parsed = Number(value || fallback);
if (!Number.isSafeInteger(parsed) || parsed < 1 || parsed > 65535) {