Fix edge gateway relay command draining

2026-07-02 12:41:34 +02:00
parent 62c1393f62
commit 243d68ab59
9 changed files with 1170 additions and 38 deletions
@@ -0,0 +1,453 @@
+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 <path>   PHP agent artifact to execute.
+                       Default: ${DEFAULT_AGENT_PATH}
+  --php-image <image>   Docker PHP image with curl, sqlite3, and pdo_sqlite.
+                       Default: ${DEFAULT_PHP_IMAGE}
+  --timeout-ms <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);
+  });
+}