import { spawn } from "node:child_process"; import crypto from "node:crypto"; import fs from "node:fs"; import http from "node:http"; import net from "node:net"; import os from "node:os"; import path from "node:path"; import process from "node:process"; import { fileURLToPath } from "node:url"; const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); const DEFAULT_AGENT_PATH = path.join( repoRoot, "services/nginx/app/resources/edge-gateway-agent/agent.php" ); const DEFAULT_PHP_IMAGE = "php:8.2-cli-bookworm"; const DEFAULT_TIMEOUT_MS = 12000; function parseArgs(argv = process.argv.slice(2)) { const options = { agentPath: DEFAULT_AGENT_PATH, phpImage: DEFAULT_PHP_IMAGE, timeoutMs: DEFAULT_TIMEOUT_MS, keepTemp: false, help: false, }; for (let index = 0; index < argv.length; index += 1) { const arg = argv[index]; const next = argv[index + 1]; switch (arg) { case "--agent-path": options.agentPath = path.resolve(String(next || "").trim()); index += 1; break; case "--php-image": options.phpImage = String(next || "").trim() || DEFAULT_PHP_IMAGE; index += 1; break; case "--timeout-ms": options.timeoutMs = Number.parseInt(String(next || ""), 10) || DEFAULT_TIMEOUT_MS; index += 1; break; case "--keep-temp": options.keepTemp = true; break; case "--help": case "-h": options.help = true; break; default: throw new Error(`Unknown argument: ${arg}`); } } return options; } function printUsage() { process.stdout.write(`Usage: node scripts/edge-agent-command-drain-proof.mjs [options] Verifies that a broker-connected PHP compose edge agent still drains API-queued SET_RELAY_STATE jobs to the LAN worker /relay/switch endpoint. Options: --agent-path PHP agent artifact to execute. Default: ${DEFAULT_AGENT_PATH} --php-image Docker PHP image with curl, sqlite3, and pdo_sqlite. Default: ${DEFAULT_PHP_IMAGE} --timeout-ms Proof timeout. Default: ${DEFAULT_TIMEOUT_MS} --keep-temp Keep the temporary config/runtime directory. --help Show this help text. `); } function readJson(request) { return new Promise((resolve) => { let raw = ""; request.setEncoding("utf8"); request.on("data", (chunk) => { raw += chunk; }); request.on("end", () => { if (raw.trim() === "") { resolve({}); return; } try { resolve(JSON.parse(raw)); } catch { resolve({ __invalid: raw }); } }); }); } function sendJson(response, status, payload) { const body = JSON.stringify(payload); response.writeHead(status, { "content-type": "application/json; charset=utf-8", "content-length": Buffer.byteLength(body), }); response.end(body); } function listen(server) { return new Promise((resolve) => { server.listen(0, "127.0.0.1", () => resolve(server.address().port)); }); } function closeServer(server) { return new Promise((resolve) => { server.close(() => resolve()); }); } function websocketAcceptKey(key) { return crypto .createHash("sha1") .update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`) .digest("base64"); } function createBrokerServer(state) { const sockets = new Set(); const server = net.createServer((socket) => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); let buffer = ""; socket.on("data", (chunk) => { buffer += chunk.toString("binary"); if (state.brokerHandshakeSeen || !buffer.includes("\r\n\r\n")) { return; } const requestText = Buffer.from(buffer, "binary").toString("utf8"); const key = requestText.match(/Sec-WebSocket-Key:\s*(.+)\r\n/i)?.[1]?.trim(); const requestLine = requestText.split("\r\n")[0] || ""; if (!requestLine.includes("/ws/agent?")) { state.failure = new Error(`unexpected broker path: ${requestLine}`); } if (!key) { state.failure = new Error("broker handshake missing Sec-WebSocket-Key"); return; } socket.write([ "HTTP/1.1 101 Switching Protocols", "Upgrade: websocket", "Connection: Upgrade", `Sec-WebSocket-Accept: ${websocketAcceptKey(key)}`, "", "", ].join("\r\n")); state.brokerHandshakeSeen = true; buffer = ""; }); }); return { server, sockets }; } function createWorkerServer(state) { return http.createServer(async (request, response) => { const url = new URL(request.url, "http://127.0.0.1"); const body = await readJson(request); state.requests.push({ service: "worker", method: request.method, path: url.pathname, body }); if (request.method === "GET" && url.pathname === "/health") { sendJson(response, 200, { status: "healthy", timestamp: new Date().toISOString() }); return; } if (request.method === "POST" && url.pathname === "/relay/switch") { state.relaySwitchSeen = true; if (body.local_ip !== "10.123.0.31" || body.channel !== 0 || body.on !== true) { state.failure = new Error(`unexpected relay switch payload: ${JSON.stringify(body)}`); } sendJson(response, 200, { online: true, on: true, output: true, raw: { source: "fake-worker" }, }); return; } sendJson(response, 404, { message: "not found" }); }); } function createApiServer(state, brokerPort, workerPort) { return http.createServer(async (request, response) => { const url = new URL(request.url, "http://127.0.0.1"); const body = await readJson(request); state.requests.push({ service: "api", method: request.method, path: url.pathname, body }); if (request.method === "POST" && url.pathname === "/edge-agent/gateways/42/heartbeat") { sendJson(response, 200, { data: { ok: true, broker_url: `ws://127.0.0.1:${brokerPort}` } }); return; } if (request.method === "POST" && url.pathname === "/edge-agent/gateways/42/selfserve/machine-signal-bindings") { sendJson(response, 200, { data: { monitors: [] } }); return; } if (request.method === "POST" && url.pathname === "/edge-agent/gateways/42/commands/poll") { state.commandPollSeen = true; if (body.wait_seconds !== 0) { state.failure = new Error( `broker-connected command poll should be non-blocking, got wait_seconds=${body.wait_seconds}` ); } if (!state.commandDelivered) { state.commandDelivered = true; sendJson(response, 200, { data: { id: 77, command_type: "SET_RELAY_STATE", payload: { localIp: "10.123.0.31", channel: 0, on: true, relayId: "relay-proof", }, }, }); return; } sendJson(response, 200, { data: null }); return; } if (request.method === "POST" && url.pathname === "/edge-agent/gateways/42/commands/77/result") { state.resultSeen = true; if (body.ok !== true || body.result?.on !== true || body.result?.raw?.source !== "fake-worker") { state.failure = new Error(`unexpected command result: ${JSON.stringify(body)}`); } sendJson(response, 200, { data: { acknowledged: true } }); return; } sendJson(response, 404, { message: "not found", path: url.pathname, workerPort }); }); } function writeConfig(tempDir, apiPort, brokerPort, workerPort) { const containerProofDir = "/proof"; const runtimeDir = `${containerProofDir}/runtime`; const config = { apiUrl: `http://127.0.0.1:${apiPort}`, brokerUrl: `ws://127.0.0.1:${brokerPort}`, gatewayId: 42, agentToken: "agent-token", installDir: containerProofDir, runtimeDir, stateDatabasePath: `${runtimeDir}/gateway-state.sqlite`, workerBaseUrl: `http://127.0.0.1:${workerPort}`, heartbeatIntervalSeconds: 60, operationPollTimeoutSeconds: 20, }; const configPath = path.join(tempDir, "config.json"); fs.writeFileSync(configPath, JSON.stringify(config, null, 2)); return { configPath, containerConfigPath: `${containerProofDir}/config.json` }; } function spawnAgent({ agentPath, phpImage, tempDir, containerConfigPath }) { return spawn("docker", [ "run", "--rm", "--network", "host", "-v", `${agentPath}:/agent.php:ro`, "-v", `${tempDir}:/proof`, phpImage, "php", "/agent.php", "--config", containerConfigPath, ], { stdio: ["ignore", "pipe", "pipe"] }); } async function stopChild(child) { if (child.exitCode !== null || child.signalCode !== null) { return; } child.kill("SIGTERM"); const hardKill = setTimeout(() => { if (child.exitCode === null && child.signalCode === null) { child.kill("SIGKILL"); } }, 1500); await Promise.race([ new Promise((resolve) => child.once("exit", resolve)), new Promise((resolve) => setTimeout(resolve, 2200)), ]); clearTimeout(hardKill); } function evidenceFromState(state, childExited) { return { brokerHandshakeSeen: state.brokerHandshakeSeen, commandPollSeen: state.commandPollSeen, relaySwitchSeen: state.relaySwitchSeen, resultSeen: state.resultSeen, agentStayedRunningUntilProofComplete: !childExited, }; } export async function runProof(options) { if (process.platform !== "linux") { throw new Error("This proof uses Docker --network host and currently expects Linux."); } if (!fs.existsSync(options.agentPath)) { throw new Error(`Agent artifact not found: ${options.agentPath}`); } const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "edge-agent-command-drain-proof-")); fs.mkdirSync(path.join(tempDir, "runtime"), { recursive: true }); const state = { brokerHandshakeSeen: false, commandPollSeen: false, relaySwitchSeen: false, resultSeen: false, commandDelivered: false, failure: null, requests: [], }; const broker = createBrokerServer(state); const workerServer = createWorkerServer(state); let apiServer = null; let child = null; let stdout = ""; let stderr = ""; let childExited = false; try { const brokerPort = await listen(broker.server); const workerPort = await listen(workerServer); apiServer = createApiServer(state, brokerPort, workerPort); const apiPort = await listen(apiServer); const { containerConfigPath } = writeConfig(tempDir, apiPort, brokerPort, workerPort); child = spawnAgent({ ...options, tempDir, containerConfigPath }); child.stdout.on("data", (chunk) => { stdout += chunk.toString(); }); child.stderr.on("data", (chunk) => { stderr += chunk.toString(); }); child.once("exit", () => { childExited = true; }); const deadline = Date.now() + options.timeoutMs; while (Date.now() < deadline && !state.failure && !childExited) { if (state.brokerHandshakeSeen && state.commandPollSeen && state.relaySwitchSeen && state.resultSeen) { break; } await new Promise((resolve) => setTimeout(resolve, 100)); } const evidence = evidenceFromState(state, childExited); if ( state.failure || !state.brokerHandshakeSeen || !state.commandPollSeen || !state.relaySwitchSeen || !state.resultSeen ) { const error = state.failure || new Error("missing proof evidence"); error.evidence = evidence; error.requests = state.requests; error.stdout = stdout.slice(-3000); error.stderr = stderr.slice(-3000); throw error; } return { evidence, agentPath: options.agentPath, phpImage: options.phpImage, tempDir, requestCount: state.requests.length, }; } finally { if (child) { await stopChild(child); } for (const socket of broker.sockets) { socket.destroy(); } await Promise.allSettled([ closeServer(broker.server), closeServer(workerServer), apiServer ? closeServer(apiServer) : Promise.resolve(), ]); if (!options.keepTemp) { fs.rmSync(tempDir, { recursive: true, force: true }); } } } async function main() { const options = parseArgs(); if (options.help) { printUsage(); return; } const result = await runProof(options); process.stdout.write("PASS broker-connected API command poll triggered local relay switch and posted result\n"); process.stdout.write(`${JSON.stringify(result.evidence)}\n`); process.stdout.write(`Agent: ${result.agentPath}\n`); process.stdout.write(`PHP image: ${result.phpImage}\n`); if (options.keepTemp) { process.stdout.write(`Temp dir: ${result.tempDir}\n`); } } if (import.meta.url === `file://${process.argv[1]}`) { main().catch((error) => { process.stderr.write(`FAIL ${error.message}\n`); if (error.evidence) { process.stderr.write(`Evidence: ${JSON.stringify(error.evidence)}\n`); } if (error.requests) { process.stderr.write(`Requests: ${JSON.stringify(error.requests, null, 2)}\n`); } if (error.stdout) { process.stderr.write(`stdout: ${error.stdout}\n`); } if (error.stderr) { process.stderr.write(`stderr: ${error.stderr}\n`); } process.exit(1); }); }