api/scripts/edge-agent-command-drain-proof.mjs

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);
  });
}