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

feat(device-edge): add fail-closed ingress admission

Browse Source
This commit is contained in:
Codex
2026-08-04 11:53:18 +03:00
parent 3bb5e6dc27
commit 1eb6c462e3
11 changed files with 335 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
device-plane/services/device-edge-relay/src/runtime.mjs
+92 -2
View File
@@ -11,9 +11,10 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
let totalForwarded = 0;
const tcpServer = createTcpServer({ allowHalfOpen: true }, (socket) => {
const remoteAddress = normalizeRemoteAddress(socket.remoteAddress);
const remoteAddress = config.resolveRemoteAddress(socket.remoteAddress);
if (
sessions.size >= config.maxConcurrentSessions
!allowsSource(remoteAddress)
|| sessions.size >= config.maxConcurrentSessions
|| currentAddressSessions(remoteAddress) >= config.maxSessionsPerAddress
|| !consumeConnectionPermit(remoteAddress)
) {
@@ -28,6 +29,8 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
upstream: null,
closed: false,
forwarded: false,
inboundBytes: 0,
outboundBytes: 0,
};
sessions.set(socket, session);
incrementAddressSessions(remoteAddress);
@@ -39,6 +42,12 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
socket.on("timeout", () => rejectSession(session));
socket.on("close", () => closeSession(session));
socket.on("error", () => rejectSession(session));
socket.on("data", (chunk) => {
session.inboundBytes += chunk.length;
if (session.inboundBytes > config.maxBytesPerDirection) {
rejectSession(session);
}
});
const upstream = connect({
host: config.upstreamHost,
@@ -61,6 +70,12 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
upstream.on("timeout", () => rejectSession(session));
upstream.on("error", () => rejectSession(session));
upstream.on("close", () => closeSession(session));
upstream.on("data", (chunk) => {
session.outboundBytes += chunk.length;
if (session.outboundBytes > config.maxBytesPerDirection) {
rejectSession(session);
}
});
});
const healthServer = createHttpServer((request, response) => {
@@ -79,6 +94,7 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
ingress: config.ingressEnabled ? "relay-only" : "disabled",
protocolInspection: "disabled",
commandTransport: "disabled",
sourceAdmission: config.sourcePolicy,
sessions: {
active: sessions.size,
accepted: totalAccepted,
@@ -115,6 +131,7 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
ingress: config.ingressEnabled ? "relay-only" : "disabled",
protocolInspection: "disabled",
commandTransport: "disabled",
sourceAdmission: config.sourcePolicy,
};
},
};
@@ -141,8 +158,16 @@ export function createDeviceEdgeRelayRuntime(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 });
return true;
}
@@ -151,6 +176,11 @@ export function createDeviceEdgeRelayRuntime(options = {}) {
return true;
}
function allowsSource(remoteAddress) {
if (config.sourcePolicy === "any") return true;
return isPublicIpv4Address(remoteAddress);
}
function rejectSession(session) {
if (!session.closed) totalRejected += 1;
session.socket.destroy();
@@ -219,6 +249,24 @@ function normalizeConfig(input) {
10000,
"device_edge_relay_connection_rate_invalid",
),
maxTrackedSourceAddresses: parseInteger(
input.maxTrackedSourceAddresses,
2048,
1,
65_536,
"device_edge_relay_source_table_limit_invalid",
),
maxBytesPerDirection: parseInteger(
input.maxBytesPerDirection,
262_144,
1_024,
16 * 1024 * 1024,
"device_edge_relay_byte_limit_invalid",
),
sourcePolicy: normalizeSourcePolicy(input.sourcePolicy, ingressEnabled),
resolveRemoteAddress: typeof input.resolveRemoteAddress === "function"
? input.resolveRemoteAddress
: normalizeRemoteAddress,
sessionTimeoutMs: parseInteger(
input.sessionTimeoutMs,
10000,
@@ -269,6 +317,48 @@ function normalizeRemoteAddress(value) {
return normalized.slice(0, 64) || "unknown";
}
function normalizeSourcePolicy(value, ingressEnabled) {
const fallback = ingressEnabled ? "public-ipv4-only" : "any";
const normalized = String(value || fallback).trim().toLowerCase();
if (!["any", "public-ipv4-only"].includes(normalized)) {
throw new TypeError("device_edge_relay_source_policy_invalid");
}
if (ingressEnabled && normalized !== "public-ipv4-only") {
throw new TypeError("device_edge_relay_ingress_source_policy_invalid");
}
return normalized;
}
function isPublicIpv4Address(value) {
const normalized = String(value || "").trim().replace(/^::ffff:/i, "");
const parts = normalized.split(".");
if (parts.length !== 4) return false;
const octets = parts.map((part) => Number(part));
if (octets.some((part) => !Number.isInteger(part) || part < 0 || part > 255)) {
return false;
}
const [first, second, third] = octets;
if (
first === 0
|| first === 10
|| first === 127
|| first >= 224
|| (first === 100 && second >= 64 && second <= 127)
|| (first === 169 && second === 254)
|| (first === 172 && second >= 16 && second <= 31)
|| (first === 192 && second === 0 && third === 0)
|| (first === 192 && second === 0 && third === 2)
|| (first === 192 && second === 88 && third === 99)
|| (first === 192 && second === 168)
|| (first === 198 && (second === 18 || second === 19))
|| (first === 198 && second === 51 && third === 100)
|| (first === 203 && second === 0 && third === 113)
) {
return false;
}
return true;
}
function parseInteger(value, fallback, minimum, maximum, errorCode) {
const parsed = Number(value ?? fallback);
if (!Number.isSafeInteger(parsed) || parsed < minimum || parsed > maximum) {
device-plane/services/device-edge-relay/src/server.mjs
+9
View File
@@ -26,6 +26,15 @@ const runtime = createDeviceEdgeRelayRuntime({
process.env.DEVICE_EDGE_RELAY_MAX_CONNECTIONS_PER_MINUTE_PER_ADDRESS,
30,
),
maxTrackedSourceAddresses: parsePositiveInt(
process.env.DEVICE_EDGE_RELAY_MAX_TRACKED_SOURCE_ADDRESSES,
2048,
),
maxBytesPerDirection: parsePositiveInt(
process.env.DEVICE_EDGE_RELAY_MAX_BYTES_PER_DIRECTION,
262_144,
),
sourcePolicy: process.env.DEVICE_EDGE_RELAY_SOURCE_POLICY,
sessionTimeoutMs: parsePositiveInt(
process.env.DEVICE_EDGE_RELAY_SESSION_TIMEOUT_MS,
10000,
device-plane/services/device-edge-relay/test/ingress-source.test.mjs
+26
View File
@@ -38,6 +38,9 @@ test("single-NIC ingress source has no host publication and a fixed ipvlan", asy
"subnet: 192.168.68.0/22",
"gateway: 192.168.68.1",
"gw_priority: 100",
"DEVICE_EDGE_RELAY_SOURCE_POLICY: public-ipv4-only",
'DEVICE_EDGE_RELAY_MAX_TRACKED_SOURCE_ADDRESSES: "2048"',
'DEVICE_EDGE_RELAY_MAX_BYTES_PER_DIRECTION: "262144"',
]) {
assert.ok(ingress.includes(required), `missing ingress boundary: ${required}`);
}
@@ -82,3 +85,26 @@ test("ingress descriptor keeps address approval and router exposure gated", asyn
assert.equal(descriptor.amneziaHostFullTunnel, "preserved");
assert.equal(descriptor.routerNatFirewall, "separate-manual-gate");
});
test("admission-gate descriptor pins the fail-closed relay boundary", async () => {
const descriptor = JSON.parse(await readFile(
resolve(
devicePlaneRoot,
"deployment/device-edge-admission-gate-v1.json",
),
"utf8",
));
assert.equal(
descriptor.schemaVersion,
"nodedc.device-edge.admission-gate.v1",
);
assert.equal(descriptor.sourceAdmission, "public-ipv4-only");
assert.equal(descriptor.maxTrackedSourceAddresses, 2048);
assert.equal(descriptor.maxBytesPerDirection, 262144);
assert.equal(descriptor.hostPortPublication, "disabled");
assert.equal(descriptor.healthPublication, "disabled");
assert.equal(descriptor.commandTransport, "disabled");
assert.equal(descriptor.gelios, "untouched");
assert.equal(descriptor.routerNatFirewall, "separate-manual-gate");
});
device-plane/services/device-edge-relay/test/runtime.test.mjs
+70 -2
View File
@@ -31,6 +31,7 @@ test("relay is transparent and never emits its own protocol bytes", async () =>
tcpPort: 0,
upstreamHost: "127.0.0.1",
upstreamPort: upstream.port,
resolveRemoteAddress: () => "8.8.8.8",
});
const addresses = await runtime.start();
try {
@@ -47,6 +48,73 @@ test("relay is transparent and never emits its own protocol bytes", async () =>
}
});
test("enabled ingress rejects a non-public source before opening upstream", async () => {
const upstream = await startEchoServer();
const runtime = createDeviceEdgeRelayRuntime({
healthPort: 0,
ingressEnabled: true,
tcpHost: "0.0.0.0",
tcpPort: 0,
upstreamHost: "127.0.0.1",
upstreamPort: upstream.port,
resolveRemoteAddress: () => "127.0.0.1",
});
const addresses = await runtime.start();
try {
const response = await sendAndCollect(
addresses.tcpAddress.port,
Buffer.from("denied"),
);
assert.equal(response.length, 0);
assert.equal(runtime.status().totalAccepted, 0);
assert.equal(runtime.status().totalForwarded, 0);
assert.equal(runtime.status().sourceAdmission, "public-ipv4-only");
} finally {
await runtime.stop();
await closeServer(upstream.server);
}
});
test("relay terminates a byte stream that exceeds its per-direction budget", async () => {
const upstream = await startEchoServer();
const runtime = createDeviceEdgeRelayRuntime({
healthPort: 0,
ingressEnabled: true,
tcpHost: "0.0.0.0",
tcpPort: 0,
upstreamHost: "127.0.0.1",
upstreamPort: upstream.port,
resolveRemoteAddress: () => "8.8.8.8",
maxBytesPerDirection: 1024,
});
const addresses = await runtime.start();
try {
const response = await sendAndCollect(
addresses.tcpAddress.port,
Buffer.alloc(1025, 0x5d),
);
assert.ok(response.length <= 1024);
assert.equal(runtime.status().totalForwarded, 1);
assert.ok(runtime.status().totalRejected >= 1);
} finally {
await runtime.stop();
await closeServer(upstream.server);
}
});
test("production ingress cannot opt out of public IPv4 admission", () => {
assert.throws(
() => createDeviceEdgeRelayRuntime({
ingressEnabled: true,
tcpHost: "0.0.0.0",
upstreamHost: "device-edge-backhaul",
upstreamPort: 19921,
sourcePolicy: "any",
}),
/device_edge_relay_ingress_source_policy_invalid/,
);
});
test("enabled relay requires a concrete private upstream", () => {
assert.throws(
() => createDeviceEdgeRelayRuntime({
@@ -77,14 +145,14 @@ function startEchoServer() {
}
function sendAndCollect(port, payload) {
return new Promise((resolve, reject) => {
return new Promise((resolve) => {
const chunks = [];
const socket = connect({ host: "127.0.0.1", port }, () => {
socket.end(payload);
});
socket.on("data", (chunk) => chunks.push(chunk));
socket.on("close", () => resolve(Buffer.concat(chunks)));
socket.on("error", reject);
socket.on("error", () => {});
});
}