diff --git a/backend/src/ee/services/dynamic-secret/providers/kubernetes.ts b/backend/src/ee/services/dynamic-secret/providers/kubernetes.ts index eeea7e4be..130e0fa92 100644 --- a/backend/src/ee/services/dynamic-secret/providers/kubernetes.ts +++ b/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, diff --git a/backend/src/ee/services/dynamic-secret/providers/sql-database.ts b/backend/src/ee/services/dynamic-secret/providers/sql-database.ts index f5299687b..39f8dd6de 100644 --- a/backend/src/ee/services/dynamic-secret/providers/sql-database.ts +++ b/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, diff --git a/backend/src/ee/services/dynamic-secret/providers/vertica.ts b/backend/src/ee/services/dynamic-secret/providers/vertica.ts index 046b73017..e361ab329 100644 --- a/backend/src/ee/services/dynamic-secret/providers/vertica.ts +++ b/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, diff --git a/backend/src/lib/gateway/gateway.ts b/backend/src/lib/gateway/gateway.ts new file mode 100644 index 000000000..179c29fc8 --- /dev/null +++ b/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 = {}; + 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 => { + 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 ( + callback: (port: number, httpsAgent?: https.Agent) => Promise, + options: IGatewayProxyOptions +): Promise => { + 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(); + } +}; diff --git a/backend/src/lib/gateway/index.ts b/backend/src/lib/gateway/index.ts index 9ec6a471e..9292473e5 100644 --- a/backend/src/lib/gateway/index.ts +++ b/backend/src/lib/gateway/index.ts @@ -1,439 +1,2 @@ -/* 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"; - -const DEFAULT_MAX_RETRIES = 3; -const DEFAULT_RETRY_DELAY = 1000; // 1 second - -const quic = quicDefault || quicModule; - -const parseSubjectDetails = (data: string) => { - const values: Record = {}; - 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; - 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 => { - 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 ( - callback: (port: number, httpsAgent?: https.Agent) => Promise, - options: ProxyOptions -): Promise => { - 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(); - } -}; +export { pingGatewayAndVerify, withGatewayProxy } from "./gateway"; +export { GatewayHttpProxyActions, GatewayProxyProtocol } from "./types"; diff --git a/backend/src/lib/gateway/types.ts b/backend/src/lib/gateway/types.ts new file mode 100644 index 000000000..5d0ac8237 --- /dev/null +++ b/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; + getProxyError: () => string; +} diff --git a/backend/src/services/identity-kubernetes-auth/identity-kubernetes-auth-service.ts b/backend/src/services/identity-kubernetes-auth/identity-kubernetes-auth-service.ts index 4eb45ba14..3501c8d28 100644 --- a/backend/src/services/identity-kubernetes-auth/identity-kubernetes-auth-service.ts +++ b/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,