cleanup and minor requested changes

backend/src/ee/services/dynamic-secret/providers/kubernetes.ts

@@ -2,7 +2,7 @@ import axios from "axios";
 import https from "https";
 
 import { InternalServerError } from "@app/lib/errors";
-import { ProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
+import { GatewayProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
 import { blockLocalAndPrivateIpAddresses } from "@app/lib/validator";
 import { TKubernetesTokenRequest } from "@app/services/identity-kubernetes-auth/identity-kubernetes-auth-types";
 
@@ -43,7 +43,7 @@ export const KubernetesProvider = ({ gatewayService }: TKubernetesProviderDTO):
         return res;
       },
       {
-        protocol: ProxyProtocol.Tcp,
+        protocol: GatewayProxyProtocol.Tcp,
         targetHost: inputs.targetHost,
         targetPort: inputs.targetPort,
         relayHost,

backend/src/ee/services/dynamic-secret/providers/sql-database.ts

@@ -3,7 +3,7 @@ import handlebars from "handlebars";
 import knex from "knex";
 import { z } from "zod";
 
-import { ProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
+import { GatewayProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
 import { alphaNumericNanoId } from "@app/lib/nanoid";
 import { validateHandlebarTemplate } from "@app/lib/template/validate-handlebars";
 
@@ -185,7 +185,7 @@ export const SqlDatabaseProvider = ({ gatewayService }: TSqlDatabaseProviderDTO)
         await gatewayCallback("localhost", port);
       },
       {
-        protocol: ProxyProtocol.Tcp,
+        protocol: GatewayProxyProtocol.Tcp,
         targetHost: providerInputs.host,
         targetPort: providerInputs.port,
         relayHost,

backend/src/ee/services/dynamic-secret/providers/vertica.ts

@@ -4,7 +4,7 @@ import knex, { Knex } from "knex";
 import { z } from "zod";
 
 import { BadRequestError } from "@app/lib/errors";
-import { ProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
+import { GatewayProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
 import { logger } from "@app/lib/logger";
 import { alphaNumericNanoId } from "@app/lib/nanoid";
 import { validateHandlebarTemplate } from "@app/lib/template/validate-handlebars";
@@ -196,7 +196,7 @@ export const VerticaProvider = ({ gatewayService }: TVerticaProviderDTO): TDynam
         await gatewayCallback("localhost", port);
       },
       {
-        protocol: ProxyProtocol.Tcp,
+        protocol: GatewayProxyProtocol.Tcp,
         targetHost: providerInputs.host,
         targetPort: providerInputs.port,
         relayHost,

backend/src/lib/gateway/gateway.ts

@@ -0,0 +1,411 @@
+/* eslint-disable no-await-in-loop */
+import crypto from "node:crypto";
+import net from "node:net";
+
+import quicDefault, * as quicModule from "@infisical/quic";
+import axios from "axios";
+import https from "https";
+
+import { BadRequestError } from "../errors";
+import { logger } from "../logger";
+import {
+  GatewayProxyProtocol,
+  IGatewayProxyOptions,
+  IGatewayProxyServer,
+  TGatewayTlsOptions,
+  TPingGatewayAndVerifyDTO
+} from "./types";
+
+const DEFAULT_MAX_RETRIES = 3;
+const DEFAULT_RETRY_DELAY = 1000; // 1 second
+
+const quic = quicDefault || quicModule;
+
+const parseSubjectDetails = (data: string) => {
+  const values: Record<string, string> = {};
+  data.split("\n").forEach((el) => {
+    const [key, value] = el.split("=");
+    values[key.trim()] = value.trim();
+  });
+  return values;
+};
+
+const createQuicConnection = async (
+  relayHost: string,
+  relayPort: number,
+  tlsOptions: TGatewayTlsOptions,
+  identityId: string,
+  orgId: string
+) => {
+  const client = await quic.QUICClient.createQUICClient({
+    host: relayHost,
+    port: relayPort,
+    config: {
+      ca: tlsOptions.ca,
+      cert: tlsOptions.cert,
+      key: tlsOptions.key,
+      applicationProtos: ["infisical-gateway"],
+      verifyPeer: true,
+      verifyCallback: async (certs) => {
+        if (!certs || certs.length === 0) return quic.native.CryptoError.CertificateRequired;
+        const serverCertificate = new crypto.X509Certificate(Buffer.from(certs[0]));
+        const caCertificate = new crypto.X509Certificate(tlsOptions.ca);
+        const isValidServerCertificate = serverCertificate.verify(caCertificate.publicKey);
+        if (!isValidServerCertificate) return quic.native.CryptoError.BadCertificate;
+
+        const subjectDetails = parseSubjectDetails(serverCertificate.subject);
+        if (subjectDetails.OU !== "Gateway" || subjectDetails.CN !== identityId || subjectDetails.O !== orgId) {
+          return quic.native.CryptoError.CertificateUnknown;
+        }
+
+        if (new Date() > new Date(serverCertificate.validTo) || new Date() < new Date(serverCertificate.validFrom)) {
+          return quic.native.CryptoError.CertificateExpired;
+        }
+
+        const formatedRelayHost =
+          process.env.NODE_ENV === "development" ? relayHost.replace("host.docker.internal", "127.0.0.1") : relayHost;
+        if (!serverCertificate.checkIP(formatedRelayHost)) return quic.native.CryptoError.BadCertificate;
+      },
+      maxIdleTimeout: 90000,
+      keepAliveIntervalTime: 30000
+    },
+    crypto: {
+      ops: {
+        randomBytes: async (data) => {
+          crypto.getRandomValues(new Uint8Array(data));
+        }
+      }
+    }
+  });
+  return client;
+};
+
+export const pingGatewayAndVerify = async ({
+  relayHost,
+  relayPort,
+  tlsOptions,
+  maxRetries = DEFAULT_MAX_RETRIES,
+  identityId,
+  orgId
+}: TPingGatewayAndVerifyDTO) => {
+  let lastError: Error | null = null;
+  const quicClient = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => {
+    throw new BadRequestError({
+      message: (err as Error)?.message,
+      error: err as Error
+    });
+  });
+
+  for (let attempt = 1; attempt <= maxRetries; attempt += 1) {
+    try {
+      const stream = quicClient.connection.newStream("bidi");
+      const pingWriter = stream.writable.getWriter();
+      await pingWriter.write(Buffer.from("PING\n"));
+      pingWriter.releaseLock();
+
+      // Read PONG response
+      const reader = stream.readable.getReader();
+      const { value, done } = await reader.read();
+
+      if (done) {
+        throw new Error("Gateway closed before receiving PONG");
+      }
+
+      const response = Buffer.from(value).toString();
+
+      if (response !== "PONG\n" && response !== "PONG") {
+        throw new Error(`Failed to Ping. Unexpected response: ${response}`);
+      }
+
+      reader.releaseLock();
+      return;
+    } catch (err) {
+      lastError = err as Error;
+
+      if (attempt < maxRetries) {
+        await new Promise((resolve) => {
+          setTimeout(resolve, DEFAULT_RETRY_DELAY);
+        });
+      }
+    } finally {
+      await quicClient.destroy();
+    }
+  }
+
+  logger.error(lastError);
+  throw new BadRequestError({
+    message: `Failed to ping gateway after ${maxRetries} attempts. Last error: ${lastError?.message}`
+  });
+};
+
+const setupProxyServer = async ({
+  targetPort,
+  targetHost,
+  tlsOptions,
+  relayHost,
+  relayPort,
+  identityId,
+  orgId,
+  protocol = GatewayProxyProtocol.Tcp,
+  httpsAgent
+}: {
+  targetHost: string;
+  targetPort: number;
+  relayPort: number;
+  relayHost: string;
+  tlsOptions: TGatewayTlsOptions;
+  identityId: string;
+  orgId: string;
+  protocol?: GatewayProxyProtocol;
+  httpsAgent?: https.Agent;
+}): Promise<IGatewayProxyServer> => {
+  const quicClient = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => {
+    throw new BadRequestError({
+      error: err as Error
+    });
+  });
+  const proxyErrorMsg = [""];
+
+  return new Promise((resolve, reject) => {
+    const server = net.createServer();
+
+    let streamClosed = false;
+
+    // eslint-disable-next-line @typescript-eslint/no-misused-promises
+    server.on("connection", async (clientConn) => {
+      try {
+        clientConn.setKeepAlive(true, 30000); // 30 seconds
+        clientConn.setNoDelay(true);
+
+        const stream = quicClient.connection.newStream("bidi");
+
+        const forwardWriter = stream.writable.getWriter();
+        let command: string;
+
+        if (protocol === GatewayProxyProtocol.Http) {
+          const targetUrl = `${targetHost}:${targetPort}`; // note(daniel): targetHost MUST include the scheme (https|http)
+          command = `FORWARD-HTTP ${targetUrl}`;
+          logger.debug(`Using HTTP proxy mode: ${command.trim()}`);
+
+          // extract ca certificate from httpsAgent if present
+          if (httpsAgent && targetHost.startsWith("https://")) {
+            const agentOptions = httpsAgent.options;
+            if (agentOptions && agentOptions.ca) {
+              const caCert = Array.isArray(agentOptions.ca) ? agentOptions.ca.join("\n") : agentOptions.ca;
+              const caB64 = Buffer.from(caCert as string).toString("base64");
+              command += ` ca=${caB64}`;
+
+              const rejectUnauthorized = agentOptions.rejectUnauthorized !== false;
+              command += ` verify=${rejectUnauthorized}`;
+
+              logger.debug(`Using HTTP proxy mode [command=${command.trim()}]`);
+            }
+          }
+
+          command += "\n";
+        } else if (protocol === GatewayProxyProtocol.Tcp) {
+          // For TCP mode, send FORWARD-TCP with host:port
+          command = `FORWARD-TCP ${targetHost}:${targetPort}\n`;
+          logger.debug(`Using TCP proxy mode: ${command.trim()}`);
+        } else {
+          throw new BadRequestError({
+            message: `Invalid protocol: ${protocol as string}`
+          });
+        }
+
+        await forwardWriter.write(Buffer.from(command));
+        forwardWriter.releaseLock();
+
+        // Set up bidirectional copy
+        const setupCopy = () => {
+          // Client to QUIC
+          // eslint-disable-next-line
+          (async () => {
+            const writer = stream.writable.getWriter();
+
+            // Create a handler for client data
+            clientConn.on("data", (chunk) => {
+              writer.write(chunk).catch((err) => {
+                proxyErrorMsg.push((err as Error)?.message);
+              });
+            });
+
+            // Handle client connection close
+            clientConn.on("end", () => {
+              if (!streamClosed) {
+                try {
+                  writer.close().catch((err) => {
+                    logger.debug(err, "Error closing writer (already closed)");
+                  });
+                } catch (error) {
+                  logger.debug(error, "Error in writer close");
+                }
+              }
+            });
+
+            clientConn.on("error", (clientConnErr) => {
+              writer.abort(clientConnErr?.message).catch((err) => {
+                proxyErrorMsg.push((err as Error)?.message);
+              });
+            });
+          })();
+
+          // QUIC to Client
+          void (async () => {
+            try {
+              const reader = stream.readable.getReader();
+
+              let reading = true;
+              while (reading) {
+                const { value, done } = await reader.read();
+
+                if (done) {
+                  reading = false;
+                  clientConn.end(); // Close client connection when QUIC stream ends
+                  break;
+                }
+
+                // Write data to TCP client
+                const canContinue = clientConn.write(Buffer.from(value));
+
+                // Handle backpressure
+                if (!canContinue) {
+                  await new Promise((res) => {
+                    clientConn.once("drain", res);
+                  });
+                }
+              }
+            } catch (err) {
+              proxyErrorMsg.push((err as Error)?.message);
+              clientConn.destroy();
+            }
+          })();
+        };
+
+        setupCopy();
+        // Handle connection closure
+        clientConn.on("close", () => {
+          if (!streamClosed) {
+            streamClosed = true;
+            stream.destroy().catch((err) => {
+              logger.debug(err, "Stream already destroyed during close event");
+            });
+          }
+        });
+
+        const cleanup = async () => {
+          try {
+            clientConn?.destroy();
+          } catch (err) {
+            logger.debug(err, "Error destroying client connection");
+          }
+
+          if (!streamClosed) {
+            streamClosed = true;
+            try {
+              await stream.destroy();
+            } catch (err) {
+              logger.debug(err, "Error destroying stream (might be already closed)");
+            }
+          }
+        };
+
+        clientConn.on("error", (clientConnErr) => {
+          logger.error(clientConnErr, "Client socket error");
+          cleanup().catch((err) => {
+            logger.error(err, "Client conn cleanup");
+          });
+        });
+
+        clientConn.on("end", () => {
+          cleanup().catch((err) => {
+            logger.error(err, "Client conn end");
+          });
+        });
+      } catch (err) {
+        logger.error(err, "Failed to establish target connection:");
+        clientConn.end();
+        reject(err);
+      }
+    });
+
+    server.on("error", (err) => {
+      reject(err);
+    });
+
+    server.on("close", () => {
+      quicClient?.destroy().catch((err) => {
+        logger.error(err, "Failed to destroy quic client");
+      });
+    });
+
+    server.listen(0, () => {
+      const address = server.address();
+      if (!address || typeof address === "string") {
+        server.close();
+        reject(new Error("Failed to get server port"));
+        return;
+      }
+
+      logger.info(`Gateway proxy started on port ${address.port} (${protocol} mode)`);
+      resolve({
+        server,
+        port: address.port,
+        cleanup: async () => {
+          try {
+            server.close();
+          } catch (err) {
+            logger.debug(err, "Error closing server");
+          }
+
+          try {
+            await quicClient?.destroy();
+          } catch (err) {
+            logger.debug(err, "Error destroying QUIC client");
+          }
+        },
+        getProxyError: () => proxyErrorMsg.join(",")
+      });
+    });
+  });
+};
+
+export const withGatewayProxy = async <T>(
+  callback: (port: number, httpsAgent?: https.Agent) => Promise<T>,
+  options: IGatewayProxyOptions
+): Promise<T> => {
+  const { relayHost, relayPort, targetHost, targetPort, tlsOptions, identityId, orgId, protocol, httpsAgent } = options;
+
+  // Setup the proxy server
+  const { port, cleanup, getProxyError } = await setupProxyServer({
+    targetHost,
+    targetPort,
+    relayPort,
+    relayHost,
+    tlsOptions,
+    identityId,
+    orgId,
+    protocol,
+    httpsAgent
+  });
+
+  try {
+    // Execute the callback with the allocated port
+    return await callback(port, httpsAgent);
+  } catch (err) {
+    const proxyErrorMessage = getProxyError();
+    if (proxyErrorMessage) {
+      logger.error(new Error(proxyErrorMessage), "Failed to proxy");
+    }
+    logger.error(err, "Failed to do gateway");
+    let errorMessage = proxyErrorMessage || (err as Error)?.message;
+    if (axios.isAxiosError(err) && (err.response?.data as { message?: string })?.message) {
+      errorMessage = (err.response?.data as { message: string }).message;
+    }
+
+    throw new BadRequestError({ message: errorMessage });
+  } finally {
+    // Ensure cleanup happens regardless of success or failure
+    await cleanup();
+  }
+};

backend/src/lib/gateway/index.ts

@@ -1,439 +1,2 @@
-/* eslint-disable no-await-in-loop */
+export { pingGatewayAndVerify, withGatewayProxy } from "./gateway";
-import crypto from "node:crypto";
+export { GatewayHttpProxyActions, GatewayProxyProtocol } from "./types";
-import net from "node:net";
-
-import quicDefault, * as quicModule from "@infisical/quic";
-import axios from "axios";
-import https from "https";
-
-import { BadRequestError } from "../errors";
-import { logger } from "../logger";
-
-const DEFAULT_MAX_RETRIES = 3;
-const DEFAULT_RETRY_DELAY = 1000; // 1 second
-
-const quic = quicDefault || quicModule;
-
-const parseSubjectDetails = (data: string) => {
-  const values: Record<string, string> = {};
-  data.split("\n").forEach((el) => {
-    const [key, value] = el.split("=");
-    values[key.trim()] = value.trim();
-  });
-  return values;
-};
-
-type TTlsOption = { ca: string; cert: string; key: string };
-
-const createQuicConnection = async (
-  relayHost: string,
-  relayPort: number,
-  tlsOptions: TTlsOption,
-  identityId: string,
-  orgId: string
-) => {
-  const client = await quic.QUICClient.createQUICClient({
-    host: relayHost,
-    port: relayPort,
-    config: {
-      ca: tlsOptions.ca,
-      cert: tlsOptions.cert,
-      key: tlsOptions.key,
-      applicationProtos: ["infisical-gateway"],
-      verifyPeer: true,
-      verifyCallback: async (certs) => {
-        if (!certs || certs.length === 0) return quic.native.CryptoError.CertificateRequired;
-        const serverCertificate = new crypto.X509Certificate(Buffer.from(certs[0]));
-        const caCertificate = new crypto.X509Certificate(tlsOptions.ca);
-        const isValidServerCertificate = serverCertificate.verify(caCertificate.publicKey);
-        if (!isValidServerCertificate) return quic.native.CryptoError.BadCertificate;
-
-        const subjectDetails = parseSubjectDetails(serverCertificate.subject);
-        if (subjectDetails.OU !== "Gateway" || subjectDetails.CN !== identityId || subjectDetails.O !== orgId) {
-          return quic.native.CryptoError.CertificateUnknown;
-        }
-
-        if (new Date() > new Date(serverCertificate.validTo) || new Date() < new Date(serverCertificate.validFrom)) {
-          return quic.native.CryptoError.CertificateExpired;
-        }
-
-        const formatedRelayHost =
-          process.env.NODE_ENV === "development" ? relayHost.replace("host.docker.internal", "127.0.0.1") : relayHost;
-        if (!serverCertificate.checkIP(formatedRelayHost)) return quic.native.CryptoError.BadCertificate;
-      },
-      maxIdleTimeout: 90000,
-      keepAliveIntervalTime: 30000
-    },
-    crypto: {
-      ops: {
-        randomBytes: async (data) => {
-          crypto.getRandomValues(new Uint8Array(data));
-        }
-      }
-    }
-  });
-  return client;
-};
-
-type TPingGatewayAndVerifyDTO = {
-  relayHost: string;
-  relayPort: number;
-  tlsOptions: TTlsOption;
-  maxRetries?: number;
-  identityId: string;
-  orgId: string;
-};
-
-export const pingGatewayAndVerify = async ({
-  relayHost,
-  relayPort,
-  tlsOptions,
-  maxRetries = DEFAULT_MAX_RETRIES,
-  identityId,
-  orgId
-}: TPingGatewayAndVerifyDTO) => {
-  let lastError: Error | null = null;
-  const quicClient = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => {
-    throw new BadRequestError({
-      message: (err as Error)?.message,
-      error: err as Error
-    });
-  });
-
-  for (let attempt = 1; attempt <= maxRetries; attempt += 1) {
-    try {
-      const stream = quicClient.connection.newStream("bidi");
-      const pingWriter = stream.writable.getWriter();
-      await pingWriter.write(Buffer.from("PING\n"));
-      pingWriter.releaseLock();
-
-      // Read PONG response
-      const reader = stream.readable.getReader();
-      const { value, done } = await reader.read();
-
-      if (done) {
-        throw new Error("Gateway closed before receiving PONG");
-      }
-
-      const response = Buffer.from(value).toString();
-
-      if (response !== "PONG\n" && response !== "PONG") {
-        throw new Error(`Failed to Ping. Unexpected response: ${response}`);
-      }
-
-      reader.releaseLock();
-      return;
-    } catch (err) {
-      lastError = err as Error;
-
-      if (attempt < maxRetries) {
-        await new Promise((resolve) => {
-          setTimeout(resolve, DEFAULT_RETRY_DELAY);
-        });
-      }
-    } finally {
-      await quicClient.destroy();
-    }
-  }
-
-  logger.error(lastError);
-  throw new BadRequestError({
-    message: `Failed to ping gateway after ${maxRetries} attempts. Last error: ${lastError?.message}`
-  });
-};
-
-interface TProxyServer {
-  server: net.Server;
-  port: number;
-  cleanup: () => Promise<void>;
-  getProxyError: () => string;
-}
-
-export enum ProxyProtocol {
-  Http = "http",
-  Tcp = "tcp"
-}
-
-const setupProxyServer = async ({
-  targetPort,
-  targetHost,
-  tlsOptions,
-  relayHost,
-  relayPort,
-  identityId,
-  orgId,
-  protocol = ProxyProtocol.Tcp,
-  httpsAgent
-}: {
-  targetHost: string;
-  targetPort: number;
-  relayPort: number;
-  relayHost: string;
-  tlsOptions: TTlsOption;
-  identityId: string;
-  orgId: string;
-  protocol?: ProxyProtocol;
-  httpsAgent?: https.Agent;
-}): Promise<TProxyServer> => {
-  const quicClient = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => {
-    throw new BadRequestError({
-      error: err as Error
-    });
-  });
-  const proxyErrorMsg = [""];
-
-  return new Promise((resolve, reject) => {
-    const server = net.createServer();
-
-    let streamClosed = false;
-
-    // eslint-disable-next-line @typescript-eslint/no-misused-promises
-    server.on("connection", async (clientConn) => {
-      try {
-        clientConn.setKeepAlive(true, 30000); // 30 seconds
-        clientConn.setNoDelay(true);
-
-        const stream = quicClient.connection.newStream("bidi");
-
-        const forwardWriter = stream.writable.getWriter();
-        let command: string;
-
-        if (protocol === ProxyProtocol.Http) {
-          const targetUrl = `${targetHost}:${targetPort}`; // note(daniel): targetHost MUST include the scheme (https|http)
-          command = `FORWARD-HTTP ${targetUrl}`;
-          logger.debug(`Using HTTP proxy mode: ${command.trim()}`);
-
-          // extract ca certificate from httpsAgent if present
-          if (httpsAgent && targetHost.startsWith("https://")) {
-            const agentOptions = httpsAgent.options;
-            if (agentOptions && agentOptions.ca) {
-              const caCert = Array.isArray(agentOptions.ca) ? agentOptions.ca.join("\n") : agentOptions.ca;
-              const caB64 = Buffer.from(caCert as string).toString("base64");
-              command += ` ca=${caB64}`;
-
-              const rejectUnauthorized = agentOptions.rejectUnauthorized !== false;
-              command += ` verify=${rejectUnauthorized}`;
-
-              logger.debug(`Using HTTP proxy mode [command=${command.trim()}]`);
-            }
-          }
-
-          command += "\n";
-        } else if (protocol === ProxyProtocol.Tcp) {
-          // For TCP mode, send FORWARD-TCP with host:port
-          command = `FORWARD-TCP ${targetHost}:${targetPort}\n`;
-          logger.debug(`Using TCP proxy mode: ${command.trim()}`);
-        } else {
-          throw new BadRequestError({
-            message: `Invalid protocol: ${protocol as string}`
-          });
-        }
-
-        await forwardWriter.write(Buffer.from(command));
-        forwardWriter.releaseLock();
-
-        // Set up bidirectional copy
-        const setupCopy = () => {
-          // Client to QUIC
-          // eslint-disable-next-line
-          (async () => {
-            const writer = stream.writable.getWriter();
-
-            // Create a handler for client data
-            clientConn.on("data", (chunk) => {
-              writer.write(chunk).catch((err) => {
-                proxyErrorMsg.push((err as Error)?.message);
-              });
-            });
-
-            // Handle client connection close
-            clientConn.on("end", () => {
-              if (!streamClosed) {
-                try {
-                  writer.close().catch((err) => {
-                    logger.debug(err, "Error closing writer (already closed)");
-                  });
-                } catch (error) {
-                  logger.debug(error, "Error in writer close");
-                }
-              }
-            });
-
-            clientConn.on("error", (clientConnErr) => {
-              writer.abort(clientConnErr?.message).catch((err) => {
-                proxyErrorMsg.push((err as Error)?.message);
-              });
-            });
-          })();
-
-          // QUIC to Client
-          void (async () => {
-            try {
-              const reader = stream.readable.getReader();
-
-              let reading = true;
-              while (reading) {
-                const { value, done } = await reader.read();
-
-                if (done) {
-                  reading = false;
-                  clientConn.end(); // Close client connection when QUIC stream ends
-                  break;
-                }
-
-                // Write data to TCP client
-                const canContinue = clientConn.write(Buffer.from(value));
-
-                // Handle backpressure
-                if (!canContinue) {
-                  await new Promise((res) => {
-                    clientConn.once("drain", res);
-                  });
-                }
-              }
-            } catch (err) {
-              proxyErrorMsg.push((err as Error)?.message);
-              clientConn.destroy();
-            }
-          })();
-        };
-
-        setupCopy();
-        // Handle connection closure
-        clientConn.on("close", () => {
-          if (!streamClosed) {
-            streamClosed = true;
-            stream.destroy().catch((err) => {
-              logger.debug(err, "Stream already destroyed during close event");
-            });
-          }
-        });
-
-        const cleanup = async () => {
-          try {
-            clientConn?.destroy();
-          } catch (err) {
-            logger.debug(err, "Error destroying client connection");
-          }
-
-          if (!streamClosed) {
-            streamClosed = true;
-            try {
-              await stream.destroy();
-            } catch (err) {
-              logger.debug(err, "Error destroying stream (might be already closed)");
-            }
-          }
-        };
-
-        clientConn.on("error", (clientConnErr) => {
-          logger.error(clientConnErr, "Client socket error");
-          cleanup().catch((err) => {
-            logger.error(err, "Client conn cleanup");
-          });
-        });
-
-        clientConn.on("end", () => {
-          cleanup().catch((err) => {
-            logger.error(err, "Client conn end");
-          });
-        });
-      } catch (err) {
-        logger.error(err, "Failed to establish target connection:");
-        clientConn.end();
-        reject(err);
-      }
-    });
-
-    server.on("error", (err) => {
-      reject(err);
-    });
-
-    server.on("close", () => {
-      quicClient?.destroy().catch((err) => {
-        logger.error(err, "Failed to destroy quic client");
-      });
-    });
-
-    server.listen(0, () => {
-      const address = server.address();
-      if (!address || typeof address === "string") {
-        server.close();
-        reject(new Error("Failed to get server port"));
-        return;
-      }
-
-      logger.info(`Gateway proxy started on port ${address.port} (${protocol} mode)`);
-      resolve({
-        server,
-        port: address.port,
-        cleanup: async () => {
-          try {
-            server.close();
-          } catch (err) {
-            logger.debug(err, "Error closing server");
-          }
-
-          try {
-            await quicClient?.destroy();
-          } catch (err) {
-            logger.debug(err, "Error destroying QUIC client");
-          }
-        },
-        getProxyError: () => proxyErrorMsg.join(",")
-      });
-    });
-  });
-};
-
-interface ProxyOptions {
-  targetHost: string;
-  targetPort: number;
-  relayHost: string;
-  relayPort: number;
-  tlsOptions: TTlsOption;
-  identityId: string;
-  orgId: string;
-  protocol: ProxyProtocol;
-  httpsAgent?: https.Agent;
-}
-
-export const withGatewayProxy = async <T>(
-  callback: (port: number, httpsAgent?: https.Agent) => Promise<T>,
-  options: ProxyOptions
-): Promise<T> => {
-  const { relayHost, relayPort, targetHost, targetPort, tlsOptions, identityId, orgId, protocol, httpsAgent } = options;
-
-  // Setup the proxy server
-  const { port, cleanup, getProxyError } = await setupProxyServer({
-    targetHost,
-    targetPort,
-    relayPort,
-    relayHost,
-    tlsOptions,
-    identityId,
-    orgId,
-    protocol,
-    httpsAgent
-  });
-
-  try {
-    // Execute the callback with the allocated port
-    return await callback(port, httpsAgent);
-  } catch (err) {
-    const proxyErrorMessage = getProxyError();
-    if (proxyErrorMessage) {
-      logger.error(new Error(proxyErrorMessage), "Failed to proxy");
-    }
-    logger.error(err, "Failed to do gateway");
-    let errorMessage = proxyErrorMessage || (err as Error)?.message;
-    if (axios.isAxiosError(err) && (err.response?.data as { message?: string })?.message) {
-      errorMessage = (err.response?.data as { message: string }).message;
-    }
-
-    throw new BadRequestError({ message: errorMessage });
-  } finally {
-    // Ensure cleanup happens regardless of success or failure
-    await cleanup();
-  }
-};

backend/src/lib/gateway/types.ts

@@ -0,0 +1,42 @@
+import net from "node:net";
+
+import https from "https";
+
+export type TGatewayTlsOptions = { ca: string; cert: string; key: string };
+
+export enum GatewayProxyProtocol {
+  Http = "http",
+  Tcp = "tcp"
+}
+
+export enum GatewayHttpProxyActions {
+  InjectGatewayK8sServiceAccountToken = "inject-k8s-sa-auth-token"
+}
+
+export interface IGatewayProxyOptions {
+  targetHost: string;
+  targetPort: number;
+  relayHost: string;
+  relayPort: number;
+  tlsOptions: TGatewayTlsOptions;
+  identityId: string;
+  orgId: string;
+  protocol: GatewayProxyProtocol;
+  httpsAgent?: https.Agent;
+}
+
+export type TPingGatewayAndVerifyDTO = {
+  relayHost: string;
+  relayPort: number;
+  tlsOptions: TGatewayTlsOptions;
+  maxRetries?: number;
+  identityId: string;
+  orgId: string;
+};
+
+export interface IGatewayProxyServer {
+  server: net.Server;
+  port: number;
+  cleanup: () => Promise<void>;
+  getProxyError: () => string;
+}

backend/src/services/identity-kubernetes-auth/identity-kubernetes-auth-service.ts

@@ -20,7 +20,7 @@ import {
 import { TPermissionServiceFactory } from "@app/ee/services/permission/permission-service";
 import { getConfig } from "@app/lib/config/env";
 import { BadRequestError, NotFoundError, PermissionBoundaryError, UnauthorizedError } from "@app/lib/errors";
-import { ProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
+import { GatewayHttpProxyActions, GatewayProxyProtocol, withGatewayProxy } from "@app/lib/gateway";
 import { extractIPDetails, isValidIpOrCidr } from "@app/lib/ip";
 import { logger } from "@app/lib/logger";
 
@@ -92,7 +92,7 @@ export const identityKubernetesAuthServiceFactory = ({
         return res;
       },
       {
-        protocol: inputs.reviewTokenThroughGateway ? ProxyProtocol.Http : ProxyProtocol.Tcp,
+        protocol: inputs.reviewTokenThroughGateway ? GatewayProxyProtocol.Http : GatewayProxyProtocol.Tcp,
         targetHost: inputs.targetHost,
         targetPort: inputs.targetPort,
         relayHost,
@@ -239,7 +239,7 @@ export const identityKubernetesAuthServiceFactory = ({
           {
             headers: {
               "Content-Type": "application/json",
-              "x-infisical-action": "inject-k8s-sa-auth-token"
+              "x-infisical-action": GatewayHttpProxyActions.InjectGatewayK8sServiceAccountToken
             },
             signal: AbortSignal.timeout(10000),
             timeout: 10000,