NODEDC_PLATFORM/services/dc-amd-connector/server.mjs

import { createServer } from "node:http";
import { readFile } from "node:fs/promises";
import { connect } from "node:net";
import { timingSafeEqual } from "node:crypto";

const allowedHosts = new Set([
  "api.cesium.com",
  "assets.ion.cesium.com",
  "dev.virtualearth.net",
  "ecn.t0.tiles.virtualearth.net",
  "ecn.t1.tiles.virtualearth.net",
  "ecn.t2.tiles.virtualearth.net",
  "ecn.t3.tiles.virtualearth.net",
]);
const config = await readConfig();

const server = createServer((request, response) => {
  if (!isAuthorized(request.headers["proxy-authorization"])) return writeJson(response, 401, { ok: false, error: "connector_unauthorized" }, { "proxy-authenticate": "Bearer" });
  if (request.method === "GET" && request.url === "/healthz") return writeJson(response, 200, { ok: true, service: "dc-amd-connector", mode: "restricted-connect", allowedHosts: allowedHosts.size });
  return writeJson(response, 405, { ok: false, error: "connect_only" });
});

server.on("connect", (request, clientSocket, head) => {
  if (!isAuthorized(request.headers["proxy-authorization"])) return rejectTunnel(clientSocket, 407, "Proxy Authentication Required", { "Proxy-Authenticate": "Bearer" });

  let target;
  try {
    target = parseConnectTarget(request.url || "");
  } catch (error) {
    log("warn", "connect_rejected", { reason: error.message });
    return rejectTunnel(clientSocket, 403, "Forbidden");
  }

  const upstream = connect({ host: target.host, port: target.port });
  let settled = false;
  const timeout = setTimeout(() => upstream.destroy(new Error("upstream_connect_timeout")), config.connectTimeoutMs);
  clientSocket.setTimeout(config.idleTimeoutMs, () => clientSocket.destroy());
  upstream.setTimeout(config.idleTimeoutMs, () => upstream.destroy());

  upstream.once("connect", () => {
    settled = true;
    clearTimeout(timeout);
    clientSocket.write("HTTP/1.1 200 Connection Established\r\nProxy-Agent: dc-amd-connector\r\n\r\n");
    if (head?.length) upstream.write(head);
    clientSocket.pipe(upstream);
    upstream.pipe(clientSocket);
    log("info", "connect_established", { host: target.host, port: target.port });
  });
  upstream.once("error", (error) => {
    clearTimeout(timeout);
    if (!settled) {
      log("warn", "connect_failed", { host: target.host, port: target.port, reason: sanitizeError(error) });
      rejectTunnel(clientSocket, 502, "Bad Gateway");
    }
  });
  clientSocket.once("error", () => upstream.destroy());
  clientSocket.once("close", () => upstream.destroy());
  upstream.once("close", () => clientSocket.destroy());
});

server.on("clientError", (_error, socket) => rejectTunnel(socket, 400, "Bad Request"));
server.listen(config.port, "0.0.0.0", () => log("info", "connector_started", { port: config.port, allowedHosts: allowedHosts.size }));

for (const signal of ["SIGINT", "SIGTERM"]) process.on(signal, () => server.close(() => process.exit(0)));

async function readConfig() {
  const port = parsePort(process.env.PORT, 8791);
  const tokenFile = String(process.env.AMD_CONNECTOR_ACCESS_TOKEN_FILE || "/run/dc-amd-secrets/connector-access").trim();
  const token = String(await readFile(tokenFile, "utf8")).trim();
  if (!/^[A-Za-z0-9_-]{48,256}$/.test(token)) throw new Error("connector_access_token_invalid");
  return {
    port,
    token: Buffer.from(token, "utf8"),
    connectTimeoutMs: parseDuration(process.env.AMD_CONNECTOR_CONNECT_TIMEOUT_SECONDS, 20),
    idleTimeoutMs: parseDuration(process.env.AMD_CONNECTOR_IDLE_TIMEOUT_SECONDS, 90),
  };
}

function parsePort(raw, fallback) {
  const value = Number(String(raw || fallback).trim());
  if (!Number.isInteger(value) || value < 1024 || value > 65535) throw new Error("connector_port_invalid");
  return value;
}

function parseDuration(raw, fallbackSeconds) {
  const seconds = Number(String(raw || fallbackSeconds).trim());
  if (!Number.isInteger(seconds) || seconds < 1 || seconds > 600) throw new Error("connector_timeout_invalid");
  return seconds * 1000;
}

function isAuthorized(rawHeader) {
  const match = /^Bearer\s+([A-Za-z0-9_-]{48,256})$/i.exec(String(rawHeader || "").trim());
  if (!match) return false;
  const candidate = Buffer.from(match[1], "utf8");
  return candidate.length === config.token.length && timingSafeEqual(candidate, config.token);
}

function parseConnectTarget(raw) {
  const match = /^([A-Za-z0-9.-]{1,253}):(443)$/.exec(String(raw || "").trim());
  if (!match) throw new Error("connect_target_invalid");
  const host = match[1].toLowerCase();
  if (!allowedHosts.has(host)) throw new Error("connect_target_not_allowed");
  return { host, port: Number(match[2]) };
}

function rejectTunnel(socket, status, message, headers = {}) {
  if (!socket || socket.destroyed) return;
  const extra = Object.entries(headers).map(([name, value]) => `${name}: ${value}\r\n`).join("");
  socket.end(`HTTP/1.1 ${status} ${message}\r\n${extra}Connection: close\r\nContent-Length: 0\r\n\r\n`);
}

function writeJson(response, status, body, headers = {}) {
  const payload = JSON.stringify(body);
  response.writeHead(status, {
    "content-type": "application/json; charset=utf-8",
    "content-length": Buffer.byteLength(payload),
    "cache-control": "no-store",
    ...headers,
  });
  response.end(payload);
}

function sanitizeError(error) {
  const message = String(error?.message || "connector_error");
  return message.replace(/[^A-Za-z0-9_.:-]/g, "_").slice(0, 120);
}

function log(level, event, fields = {}) {
  console.log(JSON.stringify({ ts: new Date().toISOString(), level, event, ...fields }));
}