feat: route code queue egress through provider gateway

src/components/provider-gateway/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@unidesk/provider-gateway",
-  "version": "0.2.17",
+  "version": "0.2.18",
   "private": true,
   "type": "module",
   "scripts": {

src/components/provider-gateway/src/egress-proxy.ts

@@ -0,0 +1,268 @@
+import { createServer, type Server, type Socket } from "node:net";
+import type {
+  CoreEgressTcpCloseMessage,
+  CoreEgressTcpDataMessage,
+  CoreEgressTcpOpenedMessage,
+  JsonValue,
+  ProviderEgressTcpCloseMessage,
+  ProviderEgressTcpDataMessage,
+  ProviderEgressTcpOpenMessage,
+} from "../../shared/src/index";
+
+type Logger = (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue) => void;
+
+type EgressToCoreMessage = ProviderEgressTcpOpenMessage | ProviderEgressTcpDataMessage | ProviderEgressTcpCloseMessage;
+type EgressFromCoreMessage = CoreEgressTcpOpenedMessage | CoreEgressTcpDataMessage | CoreEgressTcpCloseMessage;
+
+interface ProviderEgressProxyOptions {
+  providerId: string;
+  listenHost: string;
+  listenPort: number;
+  sendToCore: (message: EgressToCoreMessage) => boolean;
+  isCoreConnected: () => boolean;
+  logger: Logger;
+}
+
+interface Tunnel {
+  id: string;
+  client: Socket;
+  method: string;
+  opened: boolean;
+  pending: Buffer[];
+  closed: boolean;
+}
+
+export interface ProviderEgressProxyHandle {
+  status: () => Record<string, JsonValue>;
+  handleCoreMessage: (message: EgressFromCoreMessage) => boolean;
+  closeAll: (error?: string) => void;
+  close: () => void;
+}
+
+function nowIso(): string {
+  return new Date().toISOString();
+}
+
+function safeConnectionId(): string {
+  return `egress_${Date.now()}_${Math.random().toString(16).slice(2)}`;
+}
+
+function parseConnectTarget(target: string): { host: string; port: number } | null {
+  const trimmed = target.trim();
+  if (trimmed.startsWith("[")) {
+    const end = trimmed.indexOf("]");
+    if (end <= 1 || trimmed[end + 1] !== ":") return null;
+    const port = Number(trimmed.slice(end + 2));
+    return Number.isInteger(port) ? { host: trimmed.slice(1, end), port } : null;
+  }
+  const index = trimmed.lastIndexOf(":");
+  if (index <= 0) return null;
+  const port = Number(trimmed.slice(index + 1));
+  return Number.isInteger(port) ? { host: trimmed.slice(0, index), port } : null;
+}
+
+function parseHttpProxyRequest(headerText: string): { method: string; target: string; version: string } | null {
+  const line = headerText.split(/\r?\n/u)[0] ?? "";
+  const match = /^([A-Z]+)\s+(\S+)\s+(HTTP\/\d(?:\.\d)?)$/u.exec(line);
+  if (match === null) return null;
+  return { method: match[1] ?? "", target: match[2] ?? "", version: match[3] ?? "HTTP/1.1" };
+}
+
+function httpRequestTarget(rawTarget: string): { host: string; port: number; rewrittenTarget: string } | null {
+  try {
+    const url = new URL(rawTarget);
+    if (url.protocol !== "http:") return null;
+    return {
+      host: url.hostname,
+      port: url.port.length > 0 ? Number(url.port) : 80,
+      rewrittenTarget: `${url.pathname}${url.search}`,
+    };
+  } catch {
+    return null;
+  }
+}
+
+function rewriteAbsoluteHttpRequest(header: Buffer, rewrittenTarget: string): Buffer {
+  const text = header.toString("latin1");
+  const lines = text.split(/\r\n/u);
+  const first = lines[0] ?? "";
+  const match = /^([A-Z]+)\s+\S+\s+(HTTP\/\d(?:\.\d)?)$/u.exec(first);
+  if (match !== null) lines[0] = `${match[1]} ${rewrittenTarget} ${match[2]}`;
+  return Buffer.from(lines.join("\r\n"), "latin1");
+}
+
+function response(status: string, body: string, contentType = "text/plain; charset=utf-8"): string {
+  return [
+    `HTTP/1.1 ${status}`,
+    `Content-Type: ${contentType}`,
+    "Cache-Control: no-store",
+    "Connection: close",
+    `Content-Length: ${Buffer.byteLength(body)}`,
+    "",
+    body,
+  ].join("\r\n");
+}
+
+function jsonResponse(value: JsonValue): string {
+  return response("200 OK", `${JSON.stringify(value)}\n`, "application/json; charset=utf-8");
+}
+
+function proxyUrlFor(host: string, port: number): string {
+  return `http://${host}:${port}`;
+}
+
+export function startProviderEgressProxy(options: ProviderEgressProxyOptions): ProviderEgressProxyHandle {
+  const proxyUrl = proxyUrlFor(options.listenHost, options.listenPort);
+  const tunnels = new Map<string, Tunnel>();
+  let closed = false;
+
+  const send = (message: EgressToCoreMessage): boolean => options.sendToCore(message);
+
+  const status = (): Record<string, JsonValue> => ({
+    enabled: true,
+    providerId: options.providerId,
+    connected: options.isCoreConnected(),
+    proxyUrl,
+    listenHost: options.listenHost,
+    listenPort: options.listenPort,
+    activeTunnels: tunnels.size,
+    channel: "provider-gateway",
+  });
+
+  const closeTunnel = (id: string, error?: string): void => {
+    const tunnel = tunnels.get(id);
+    if (tunnel === undefined) return;
+    tunnels.delete(id);
+    tunnel.closed = true;
+    if (!tunnel.client.destroyed) tunnel.client.destroy();
+    send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: id, at: nowIso() });
+    if (error !== undefined) options.logger("warn", "egress_proxy_tunnel_closed", { connectionId: id, error });
+  };
+
+  const handleCoreMessage = (message: EgressFromCoreMessage): boolean => {
+    const tunnel = tunnels.get(message.connectionId);
+    if (message.type === "egress_tcp_opened") {
+      if (tunnel === undefined || tunnel.closed) return true;
+      tunnel.opened = true;
+      if (tunnel.method === "CONNECT") {
+        tunnel.client.write("HTTP/1.1 200 Connection Established\r\nProxy-Agent: UniDesk-ProviderGateway\r\n\r\n");
+      }
+      for (const chunk of tunnel.pending.splice(0)) {
+        send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: tunnel.id, data: chunk.toString("base64"), encoding: "base64", at: nowIso() });
+      }
+      return true;
+    }
+    if (message.type === "egress_tcp_data") {
+      if (tunnel === undefined || tunnel.client.destroyed) return true;
+      tunnel.client.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
+      return true;
+    }
+    if (message.type === "egress_tcp_close") {
+      if (tunnel !== undefined) {
+        tunnels.delete(message.connectionId);
+        tunnel.closed = true;
+        if (!tunnel.client.destroyed) tunnel.client.destroy();
+      }
+      if (message.error !== undefined && message.error.length > 0) {
+        options.logger("warn", "egress_proxy_remote_close", { connectionId: message.connectionId, error: message.error });
+      }
+      return true;
+    }
+    return false;
+  };
+
+  const server: Server = createServer((client) => {
+    let header = Buffer.alloc(0);
+    let settled = false;
+    const fail = (statusCode: string, text: string): void => {
+      if (!client.destroyed) client.end(response(statusCode, text));
+    };
+    client.on("data", (chunk) => {
+      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+      if (settled) return;
+      header = Buffer.concat([header, buffer]);
+      const headerEnd = header.indexOf("\r\n\r\n");
+      if (headerEnd < 0) {
+        if (header.length > 64 * 1024) {
+          settled = true;
+          fail("431 Request Header Fields Too Large", "proxy request header too large\n");
+        }
+        return;
+      }
+      settled = true;
+      const head = header.subarray(0, headerEnd + 4);
+      const rest = header.subarray(headerEnd + 4);
+      const parsed = parseHttpProxyRequest(head.toString("latin1"));
+      if (parsed === null) {
+        fail("400 Bad Request", "invalid proxy request\n");
+        return;
+      }
+      if (parsed.method === "GET" && (parsed.target === "/health" || parsed.target === "/__unidesk/egress-proxy/health")) {
+        client.end(jsonResponse({ ok: true, service: "provider-gateway-egress-proxy", ...status() }));
+        return;
+      }
+      if (!options.isCoreConnected()) {
+        fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
+        return;
+      }
+      const id = safeConnectionId();
+      let firstPayload: Buffer | null = null;
+      let target: { host: string; port: number } | null = null;
+      if (parsed.method === "CONNECT") {
+        target = parseConnectTarget(parsed.target);
+        firstPayload = rest.length > 0 ? rest : null;
+      } else {
+        const httpTarget = httpRequestTarget(parsed.target);
+        if (httpTarget !== null) {
+          target = { host: httpTarget.host, port: httpTarget.port };
+          firstPayload = Buffer.concat([rewriteAbsoluteHttpRequest(head, httpTarget.rewrittenTarget), rest]);
+        }
+      }
+      if (target === null || target.port <= 0 || target.port > 65_535) {
+        fail("400 Bad Request", "unsupported proxy target\n");
+        return;
+      }
+      const tunnel: Tunnel = { id, client, method: parsed.method, opened: false, pending: [], closed: false };
+      tunnels.set(id, tunnel);
+      client.on("data", (nextChunk) => {
+        const nextBuffer = Buffer.isBuffer(nextChunk) ? nextChunk : Buffer.from(nextChunk);
+        if (!tunnel.opened) {
+          tunnel.pending.push(nextBuffer);
+          return;
+        }
+        send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: id, data: nextBuffer.toString("base64"), encoding: "base64", at: nowIso() });
+      });
+      client.on("close", () => closeTunnel(id));
+      client.on("error", (error) => closeTunnel(id, error.message));
+      const opened = send({ type: "egress_tcp_open", providerId: options.providerId, connectionId: id, host: target.host, port: target.port, at: nowIso() });
+      if (!opened) {
+        tunnels.delete(id);
+        tunnel.closed = true;
+        fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
+        return;
+      }
+      if (firstPayload !== null) tunnel.pending.push(firstPayload);
+    });
+    client.on("error", () => undefined);
+  });
+
+  server.listen(options.listenPort, options.listenHost, () => {
+    options.logger("info", "egress_proxy_listening", status());
+  });
+  server.on("error", (error) => {
+    options.logger("error", "egress_proxy_listen_failed", { proxyUrl, error: error.message });
+  });
+
+  return {
+    status,
+    handleCoreMessage,
+    closeAll: (error?: string) => {
+      for (const id of Array.from(tunnels.keys())) closeTunnel(id, error);
+    },
+    close: () => {
+      closed = true;
+      for (const id of Array.from(tunnels.keys())) closeTunnel(id, "egress proxy stopping");
+      server.close();
+    },
+  };
+}

src/components/provider-gateway/src/index.ts

@@ -1,6 +1,9 @@
 import { spawnSync } from "node:child_process";
 import { existsSync, readFileSync, readdirSync } from "node:fs";
 import {
+  type CoreEgressTcpCloseMessage,
+  type CoreEgressTcpDataMessage,
+  type CoreEgressTcpOpenedMessage,
   type CoreDispatchMessage,
   type CoreHostSshCloseMessage,
   type CoreHostSshEofMessage,
@@ -20,6 +23,7 @@ import {
   parseJsonObject,
 } from "../../shared/src/index";
 import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../shared/src/rotating-jsonl";
+import { startProviderEgressProxy, type ProviderEgressProxyHandle } from "./egress-proxy";
 
 interface RuntimeConfig {
   serverUrl: string;
@@ -45,6 +49,9 @@ interface RuntimeConfig {
   hostSshKey: string | null;
   hostRemoteCwd: string | null;
   hostLoginShell: string | null;
+  egressProxyEnabled: boolean;
+  egressProxyListenHost: string;
+  egressProxyPort: number;
   logFile: string;
 }
 
@@ -114,6 +121,7 @@ interface DockerContainerInspectDetails {
 const hostSshSessions = new Map<string, HostSshSession>();
 const microserviceHttpCache = new Map<string, MicroserviceHttpCacheEntry>();
 const microserviceHttpInFlight = new Map<string, Promise<JsonValue>>();
+let providerEgressProxy: ProviderEgressProxyHandle | null = null;
 const gatewayMetadata = readGatewayMetadata();
 const defaultMasterServer = "http://74.48.78.17/";
 const defaultProviderToken = "unidesk-dev-token-change-me";
@@ -196,6 +204,15 @@ function readOptionalNumberEnv(name: string, ...aliases: string[]): number | nul
   return parsed;
 }
 
+function readBooleanEnv(name: string, fallback: boolean, ...aliases: string[]): boolean {
+  const raw = readEnv(name, ...aliases);
+  if (raw === null || raw.trim().length === 0) return fallback;
+  const value = raw.trim().toLowerCase();
+  if (value === "1" || value === "true" || value === "yes" || value === "on") return true;
+  if (value === "0" || value === "false" || value === "no" || value === "off") return false;
+  return fallback;
+}
+
 function providerServerUrlFromMaster(rawMaster: string): string {
   const raw = rawMaster.trim() || defaultMasterServer;
   const withScheme = /^[a-z][a-z0-9+.-]*:\/\//iu.test(raw) ? raw : `http://${raw}`;
@@ -462,6 +479,9 @@ function readConfig(): RuntimeConfig {
     hostSshKey: readOptionalStringEnv("HOST_SSH_KEY") ?? (mountedHostSshKey ? defaultHostSshKey : null),
     hostRemoteCwd: readOptionalStringEnv("HOST_REMOTE_CWD") ?? defaultHostRemoteCwd(hostSshUser),
     hostLoginShell: readOptionalStringEnv("HOST_LOGIN_SHELL") ?? (hostSshUser === null ? null : "/bin/bash"),
+    egressProxyEnabled: readBooleanEnv("PROVIDER_EGRESS_PROXY_ENABLED", true, "UNIDESK_PROVIDER_EGRESS_PROXY_ENABLED"),
+    egressProxyListenHost: readOptionalStringEnv("PROVIDER_EGRESS_PROXY_LISTEN_HOST", "UNIDESK_PROVIDER_EGRESS_PROXY_LISTEN_HOST") ?? "0.0.0.0",
+    egressProxyPort: readNumberEnv("PROVIDER_EGRESS_PROXY_PORT", 18789, "UNIDESK_PROVIDER_EGRESS_PROXY_PORT"),
     logFile: readOptionalStringEnv("LOG_FILE") ?? `/var/log/unidesk/provider-gateway-${providerId}.jsonl`,
   };
 }
@@ -497,6 +517,7 @@ function withToken(rawUrl: string, token: string): string {
 function currentLabels(): ProviderLabels {
   const hostSshConfigured = isHostSshConfigured();
   const containerGuard = refreshSelfContainerGuard();
+  const egressProxyStatus = providerEgressProxy?.status() ?? null;
   return {
     ...config.labels,
     dockerSocketPresent: existsSync(config.dockerSocketPath),
@@ -509,6 +530,10 @@ function currentLabels(): ProviderLabels {
     providerGatewayStartedAt: startedAt.toISOString(),
     providerGatewayUpgradePolicy: "always-enabled",
     providerGatewayMicroserviceHttpCache: true,
+    providerGatewayEgressProxy: config.egressProxyEnabled,
+    providerGatewayEgressProxyPort: config.egressProxyPort,
+    providerGatewayEgressProxyConnected: egressProxyStatus === null ? false : egressProxyStatus.connected === true,
+    providerGatewayEgressProxyActiveTunnels: egressProxyStatus === null ? 0 : egressProxyStatus.activeTunnels ?? 0,
     providerGatewayDockerRestartGuard: true,
     providerGatewayContainerId: containerGuard.containerId,
     providerGatewayContainerName: containerGuard.containerName,
@@ -529,9 +554,16 @@ function sendJson(value: unknown): void {
   socket.send(JSON.stringify(value));
 }
 
+function sendJsonOk(value: unknown): boolean {
+  if (!socket || socket.readyState !== WebSocket.OPEN) return false;
+  socket.send(JSON.stringify(value));
+  return true;
+}
+
 function sendRegister(): void {
   const capabilities = ["heartbeat", "system.status", "docker.status", "docker.ps", "provider.upgrade", "microservice.http", "microservice.http.cache", "echo"];
   if (isHostSshConfigured()) capabilities.push("host.ssh");
+  if (config.egressProxyEnabled) capabilities.push("network.egress-proxy");
   sendJson({
     type: "register",
     providerId: config.providerId,
@@ -2043,6 +2075,14 @@ function handleMessage(raw: MessageEvent<string>): void {
       closeHostSshSession(parsed as CoreHostSshCloseMessage);
       return;
     }
+    if (
+      parsed.type === "egress_tcp_opened" ||
+      parsed.type === "egress_tcp_data" ||
+      parsed.type === "egress_tcp_close"
+    ) {
+      providerEgressProxy?.handleCoreMessage(parsed as CoreEgressTcpOpenedMessage | CoreEgressTcpDataMessage | CoreEgressTcpCloseMessage);
+      return;
+    }
     logger("debug", "core_message", parsed as JsonValue);
   } catch (error) {
     logger("error", "core_message_parse_failed", { error: String(error) });
@@ -2135,6 +2175,7 @@ function connect(): void {
   socket.addEventListener("close", (event) => {
     logger("warn", "connect_close", { code: event.code, reason: event.reason });
     clearConnectionTimers();
+    providerEgressProxy?.closeAll("provider-gateway core socket closed");
     scheduleReconnect();
   });
   socket.addEventListener("error", () => {
@@ -2153,8 +2194,20 @@ process.on("SIGTERM", () => {
     session.proc.kill("SIGTERM");
   }
   hostSshSessions.clear();
+  providerEgressProxy?.close();
   socket?.close(1000, "provider shutdown");
   process.exit(0);
 });
 
+if (config.egressProxyEnabled) {
+  providerEgressProxy = startProviderEgressProxy({
+    providerId: config.providerId,
+    listenHost: config.egressProxyListenHost,
+    listenPort: config.egressProxyPort,
+    isCoreConnected: () => socket?.readyState === WebSocket.OPEN,
+    sendToCore: (message) => sendJsonOk(message),
+    logger,
+  });
+}
+
 connect();