From 9bfc2a5dd2bb0c423bfa89eb12d985ea6419f370 Mon Sep 17 00:00:00 2001 From: = Date: Tue, 4 Mar 2025 22:10:38 +0530 Subject: [PATCH] feat: updated gateway to quic --- backend/Dockerfile.dev | 58 ++-- backend/package.json | 1 + .../ee/services/gateway/gateway-service.ts | 3 +- backend/src/lib/gateway/index.ts | 318 +++++++++++------- cli/go.mod | 20 +- cli/go.sum | 34 +- cli/packages/cmd/gateway.go | 31 ++ cli/packages/gateway/connection.go | 84 +++-- cli/packages/gateway/gateway.go | 155 ++++----- cli/packages/gateway/relay.go | 186 ++++++++++ 10 files changed, 610 insertions(+), 280 deletions(-) create mode 100644 cli/packages/gateway/relay.go diff --git a/backend/Dockerfile.dev b/backend/Dockerfile.dev index 0d1c16574..adb5157f5 100644 --- a/backend/Dockerfile.dev +++ b/backend/Dockerfile.dev @@ -1,4 +1,4 @@ -FROM node:20-alpine +FROM node:20-slim # ? Setup a test SoftHSM module. In production a real HSM is used. @@ -7,32 +7,32 @@ ARG SOFTHSM2_VERSION=2.5.0 ENV SOFTHSM2_VERSION=${SOFTHSM2_VERSION} \ SOFTHSM2_SOURCES=/tmp/softhsm2 -# install build dependencies including python3 (required for pkcs11js and partially TDS driver) -RUN apk --update add \ - alpine-sdk \ - autoconf \ - automake \ - git \ - libtool \ - openssl-dev \ - python3 \ - make \ - g++ \ - openssh +# Install build dependencies including python3 (required for pkcs11js and partially TDS driver) +RUN apt-get update && apt-get install -y \ + build-essential \ + autoconf \ + automake \ + git \ + libtool \ + libssl-dev \ + python3 \ + make \ + g++ \ + openssh-client \ + curl \ + pkg-config -# install dependencies for TDS driver (required for SAP ASE dynamic secrets) -RUN apk add --no-cache \ +# Install dependencies for TDS driver (required for SAP ASE dynamic secrets) +RUN apt-get install -y \ unixodbc \ - freetds \ unixodbc-dev \ - libc-dev \ - freetds-dev + freetds-dev \ + freetds-bin \ + tdsodbc +RUN printf "[FreeTDS]\nDescription = FreeTDS Driver\nDriver = /usr/lib/x86_64-linux-gnu/odbc/libtdsodbc.so\nSetup = /usr/lib/x86_64-linux-gnu/odbc/libtdsodbc.so\nFileUsage = 1\n" > /etc/odbcinst.ini -RUN printf "[FreeTDS]\nDescription = FreeTDS Driver\nDriver = /usr/lib/libtdsodbc.so\nSetup = /usr/lib/libtdsodbc.so\nFileUsage = 1\n" > /etc/odbcinst.ini - -# build and install SoftHSM2 - +# Build and install SoftHSM2 RUN git clone https://github.com/opendnssec/SoftHSMv2.git ${SOFTHSM2_SOURCES} WORKDIR ${SOFTHSM2_SOURCES} @@ -45,16 +45,18 @@ RUN git checkout ${SOFTHSM2_VERSION} -b ${SOFTHSM2_VERSION} \ WORKDIR /root RUN rm -fr ${SOFTHSM2_SOURCES} -# install pkcs11-tool -RUN apk --update add opensc +# Install pkcs11-tool +RUN apt-get install -y opensc -RUN softhsm2-util --init-token --slot 0 --label "auth-app" --pin 1234 --so-pin 0000 +RUN mkdir -p /etc/softhsm2/tokens && \ + softhsm2-util --init-token --slot 0 --label "auth-app" --pin 1234 --so-pin 0000 # ? App setup -RUN apk add --no-cache bash curl && curl -1sLf \ - 'https://dl.cloudsmith.io/public/infisical/infisical-cli/setup.alpine.sh' | bash \ - && apk add infisical=0.8.1 && apk add --no-cache git +# Install Infisical CLI +RUN curl -1sLf 'https://dl.cloudsmith.io/public/infisical/infisical-cli/setup.deb.sh' | bash && \ + apt-get update && \ + apt-get install -y infisical=0.8.1 WORKDIR /app diff --git a/backend/package.json b/backend/package.json index 907a9fbcc..31b01d1c5 100644 --- a/backend/package.json +++ b/backend/package.json @@ -145,6 +145,7 @@ "@fastify/swagger": "^8.14.0", "@fastify/swagger-ui": "^2.1.0", "@google-cloud/kms": "^4.5.0", + "@infisical/quic": "^1.0.8", "@node-saml/passport-saml": "^4.0.4", "@octokit/auth-app": "^7.1.1", "@octokit/plugin-retry": "^5.0.5", diff --git a/backend/src/ee/services/gateway/gateway-service.ts b/backend/src/ee/services/gateway/gateway-service.ts index 75ef1f0f2..1dd523f09 100644 --- a/backend/src/ee/services/gateway/gateway-service.ts +++ b/backend/src/ee/services/gateway/gateway-service.ts @@ -1,5 +1,4 @@ import crypto from "node:crypto"; - import { ForbiddenError } from "@casl/ability"; import * as x509 from "@peculiar/x509"; import { z } from "zod"; @@ -474,7 +473,7 @@ export const gatewayServiceFactory = ({ relayHost, relayPort: Number(relayPort), tlsOptions: { - key: privateKey, + key: privateKey.toString(), ca: `${gatewayCaCert.toString("pem")}\n${rootCaCert.toString("pem")}`.trim(), cert: clientCert.toString("pem") }, diff --git a/backend/src/lib/gateway/index.ts b/backend/src/lib/gateway/index.ts index eaebd896d..7fe65d67f 100644 --- a/backend/src/lib/gateway/index.ts +++ b/backend/src/lib/gateway/index.ts @@ -1,6 +1,9 @@ /* eslint-disable no-await-in-loop */ +import crypto from "node:crypto"; import net from "node:net"; -import tls from "node:tls"; + +import { QUICClient } from "@infisical/quic"; +import { CryptoError } from "@infisical/quic/dist/native"; import { BadRequestError } from "../errors"; import { logger } from "../logger"; @@ -8,34 +11,71 @@ import { logger } from "../logger"; const DEFAULT_MAX_RETRIES = 3; const DEFAULT_RETRY_DELAY = 1000; // 1 second -const createTLSConnection = (relayHost: string, relayPort: number, tlsOptions: tls.TlsOptions = {}) => { - return new Promise((resolve, reject) => { - // @ts-expect-error this is resolved in next connect - const socket = new tls.TLSSocket(null, { - rejectUnauthorized: true, - ...tlsOptions - }); - - const cleanup = () => { - socket.removeAllListeners(); - socket.end(); - }; - - socket.once("error", (err) => { - cleanup(); - reject(err); - }); - - socket.connect(relayPort, relayHost, () => { - resolve(socket); - }); +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 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 CryptoError.CertificateRequired; + const serverCertificate = new crypto.X509Certificate(Buffer.from(certs[0])); + const caCertificate = new crypto.X509Certificate(tlsOptions.ca); + const isValidServerCertificate = serverCertificate.checkIssued(caCertificate); + if (!isValidServerCertificate) return CryptoError.BadCertificate; + + const subjectDetails = parseSubjectDetails(serverCertificate.subject); + if (subjectDetails.OU !== "Gateway" || subjectDetails.CN !== identityId || subjectDetails.O !== orgId) { + return CryptoError.CertificateUnknown; + } + + if (new Date() > new Date(serverCertificate.validTo) || new Date() < new Date(serverCertificate.validFrom)) { + return CryptoError.CertificateExpired; + } + + const formatedRelayHost = + process.env.NODE_ENV === "development" ? relayHost.replace("host.docker.internal", "127.0.0.1") : relayHost; + if (!serverCertificate.checkIP(formatedRelayHost)) return 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: tls.TlsOptions; + tlsOptions: TTlsOption; maxRetries?: number; identityId: string; orgId: string; @@ -44,56 +84,44 @@ type TPingGatewayAndVerifyDTO = { export const pingGatewayAndVerify = async ({ relayHost, relayPort, - tlsOptions = {}, + tlsOptions, maxRetries = DEFAULT_MAX_RETRIES, identityId, orgId }: TPingGatewayAndVerifyDTO) => { let lastError: Error | null = null; - + const quic = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => { + throw new BadRequestError({ + error: err as Error + }); + }); for (let attempt = 1; attempt <= maxRetries; attempt += 1) { try { - const socket = await createTLSConnection(relayHost, relayPort, tlsOptions); - socket.setTimeout(2000); + const stream = quic.connection.newStream("bidi"); + const pingWriter = stream.writable.getWriter(); + await pingWriter.write(Buffer.from("PING\n")); + pingWriter.releaseLock(); - const pingResult = await new Promise((resolve, reject) => { - socket.once("timeout", () => { - socket.destroy(); - reject(new Error("Timeout")); + // Read PONG response + const reader = stream.readable.getReader(); + const { value, done } = await reader.read(); + + if (done) { + throw new BadRequestError({ + message: "Gateway closed before receiving PONG" }); - socket.once("close", () => { - socket.destroy(); + } + + const response = Buffer.from(value).toString(); + + if (response !== "PONG\n" && response !== "PONG") { + throw new BadRequestError({ + message: `Failed to Ping. Unexpected response: ${response}` }); + } - socket.once("end", () => { - socket.destroy(); - }); - socket.once("error", (err) => { - reject(err); - }); - - socket.write(Buffer.from("PING\n"), () => { - socket.once("data", (data) => { - const response = (data as string).toString(); - const certificate = socket.getPeerCertificate(); - - if (certificate.subject.CN !== identityId || certificate.subject.O !== orgId) { - throw new BadRequestError({ - message: `Invalid gateway. Certificate not found for ${identityId} in organization ${orgId}` - }); - } - - if (response === "PONG") { - resolve(true); - } else { - reject(new Error(`Unexpected response: ${response}`)); - } - }); - }); - }); - - socket.end(); - return pingResult; + reader.releaseLock(); + return; } catch (err) { lastError = err as Error; @@ -102,6 +130,8 @@ export const pingGatewayAndVerify = async ({ setTimeout(resolve, DEFAULT_RETRY_DELAY); }); } + } finally { + await quic.destroy(); } } @@ -114,76 +144,125 @@ export const pingGatewayAndVerify = async ({ interface TProxyServer { server: net.Server; port: number; - cleanup: () => void; + cleanup: () => Promise; } -const setupProxyServer = ({ +const setupProxyServer = async ({ targetPort, targetHost, - tlsOptions = {}, + tlsOptions, relayHost, - relayPort + relayPort, + identityId, + orgId }: { targetHost: string; targetPort: number; relayPort: number; relayHost: string; - tlsOptions: tls.TlsOptions; + tlsOptions: TTlsOption; + identityId: string; + orgId: string; }): Promise => { + const quic = await createQuicConnection(relayHost, relayPort, tlsOptions, identityId, orgId).catch((err) => { + throw new BadRequestError({ + error: err as Error + }); + }); + return new Promise((resolve, reject) => { const server = net.createServer(); // eslint-disable-next-line @typescript-eslint/no-misused-promises - server.on("connection", async (clientSocket) => { + server.on("connection", async (clientConn) => { try { - const targetSocket = await createTLSConnection(relayHost, relayPort, tlsOptions); + clientConn.setKeepAlive(true, 30000); // 30 seconds + clientConn.setNoDelay(true); - targetSocket.write(Buffer.from(`FORWARD-TCP ${targetHost}:${targetPort}\n`), () => { - clientSocket.on("data", (data) => { - const flushed = targetSocket.write(data); - if (!flushed) { - clientSocket.pause(); - targetSocket.once("drain", () => { - clientSocket.resume(); - }); - } - }); + const stream = quic.connection.newStream("bidi"); + // Send FORWARD-TCP command + const forwardWriter = stream.writable.getWriter(); + await forwardWriter.write(Buffer.from(`FORWARD-TCP ${targetHost}:${targetPort}\n`)); + forwardWriter.releaseLock(); + /* eslint-disable @typescript-eslint/no-misused-promises */ + // Set up bidirectional copy + const setupCopy = async () => { + // Client to QUIC + // eslint-disable-next-line + (async () => { + try { + const writer = stream.writable.getWriter(); - targetSocket.on("data", (data) => { - const flushed = clientSocket.write(data as string); - if (!flushed) { - targetSocket.pause(); - clientSocket.once("drain", () => { - targetSocket.resume(); + // Create a handler for client data + clientConn.on("data", async (chunk) => { + await writer.write(chunk); }); + + // Handle client connection close + clientConn.on("end", async () => { + await writer.close(); + }); + + clientConn.on("error", async (err) => { + await writer.abort(err); + }); + } catch (err) { + clientConn.destroy(); } - }); + })(); + + // 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) { + clientConn.destroy(); + } + })(); + }; + await setupCopy(); + // + // Handle connection closure + clientConn.on("close", async () => { + await stream.destroy(); }); - const cleanup = () => { - clientSocket?.unpipe(); - clientSocket?.end(); - targetSocket?.unpipe(); - targetSocket?.end(); + const cleanup = async () => { + clientConn?.destroy(); + await stream.destroy(); }; - clientSocket.on("error", (err) => { + clientConn.on("error", (err) => { logger.error(err, "Client socket error"); - cleanup(); + void cleanup(); reject(err); }); - targetSocket.on("error", (err) => { - logger.error(err, "Target socket error"); - cleanup(); - reject(err); - }); - - clientSocket.on("end", cleanup); - targetSocket.on("end", cleanup); + clientConn.on("end", cleanup); } catch (err) { logger.error(err, "Failed to establish target connection:"); - clientSocket.end(); + clientConn.end(); reject(err); } }); @@ -192,6 +271,12 @@ const setupProxyServer = ({ reject(err); }); + server.on("close", async () => { + await quic?.destroy(); + }); + + /* eslint-enable */ + server.listen(0, () => { const address = server.address(); if (!address || typeof address === "string") { @@ -204,8 +289,9 @@ const setupProxyServer = ({ resolve({ server, port: address.port, - cleanup: () => { + cleanup: async () => { server.close(); + await quic?.destroy(); } }); }); @@ -217,8 +303,7 @@ interface ProxyOptions { targetPort: number; relayHost: string; relayPort: number; - tlsOptions?: tls.TlsOptions; - maxRetries?: number; + tlsOptions: TTlsOption; identityId: string; orgId: string; } @@ -227,30 +312,19 @@ export const withGatewayProxy = async ( callback: (port: number) => Promise, options: ProxyOptions ): Promise => { - const { - relayHost, - relayPort, + const { relayHost, relayPort, targetHost, targetPort, tlsOptions, identityId, orgId } = options; + + // Setup the proxy server + const { port, cleanup } = await setupProxyServer({ targetHost, targetPort, - tlsOptions = {}, - maxRetries = DEFAULT_MAX_RETRIES, - identityId, - orgId - } = options; - - // First, try to ping the gateway - await pingGatewayAndVerify({ - relayHost, relayPort, + relayHost, tlsOptions, - maxRetries, identityId, orgId }); - // Setup the proxy server - const { port, cleanup } = await setupProxyServer({ targetHost, targetPort, relayPort, relayHost, tlsOptions }); - try { // Execute the callback with the allocated port await callback(port); @@ -259,6 +333,6 @@ export const withGatewayProxy = async ( throw new BadRequestError({ message: (err as Error)?.message }); } finally { // Ensure cleanup happens regardless of success or failure - cleanup(); + await cleanup(); } }; diff --git a/cli/go.mod b/cli/go.mod index 7339ec68f..66c4e61f8 100644 --- a/cli/go.mod +++ b/cli/go.mod @@ -1,6 +1,8 @@ module github.com/Infisical/infisical-merge -go 1.21 +go 1.23.0 + +toolchain go1.23.5 require ( github.com/bradleyjkemp/cupaloy/v2 v2.8.0 @@ -21,12 +23,13 @@ require ( github.com/pion/logging v0.2.3 github.com/pion/turn/v4 v4.0.0 github.com/posthog/posthog-go v0.0.0-20221221115252-24dfed35d71a + github.com/quic-go/quic-go v0.50.0 github.com/rs/cors v1.11.0 github.com/rs/zerolog v1.26.1 github.com/spf13/cobra v1.6.1 github.com/spf13/viper v1.8.1 github.com/stretchr/testify v1.9.0 - golang.org/x/crypto v0.33.0 + golang.org/x/crypto v0.35.0 golang.org/x/term v0.29.0 gopkg.in/yaml.v2 v2.4.0 ) @@ -58,13 +61,15 @@ require ( github.com/dvsekhvalnov/jose2go v1.6.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.4.9 // indirect - github.com/go-logr/logr v1.4.1 // indirect + github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/errors v0.20.2 // indirect github.com/go-openapi/strfmt v0.21.3 // indirect + github.com/go-task/slim-sprig/v3 v3.0.0 // indirect github.com/godbus/dbus/v5 v5.1.0 // indirect github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect github.com/golang/protobuf v1.5.4 // indirect + github.com/google/pprof v0.0.0-20250302191652-9094ed2288e7 // indirect github.com/google/s2a-go v0.1.7 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect github.com/googleapis/gax-go/v2 v2.12.5 // indirect @@ -82,6 +87,7 @@ require ( github.com/muesli/mango-pflag v0.1.0 // indirect github.com/muesli/termenv v0.15.2 // indirect github.com/oklog/ulid v1.3.1 // indirect + github.com/onsi/ginkgo/v2 v2.22.2 // indirect github.com/pelletier/go-toml v1.9.3 // indirect github.com/pion/dtls/v3 v3.0.4 // indirect github.com/pion/randutil v0.1.0 // indirect @@ -103,17 +109,21 @@ require ( go.opentelemetry.io/otel v1.24.0 // indirect go.opentelemetry.io/otel/metric v1.24.0 // indirect go.opentelemetry.io/otel/trace v1.24.0 // indirect - golang.org/x/net v0.33.0 // indirect + go.uber.org/mock v0.5.0 // indirect + golang.org/x/exp v0.0.0-20250228200357-dead58393ab7 // indirect + golang.org/x/mod v0.23.0 // indirect + golang.org/x/net v0.35.0 // indirect golang.org/x/oauth2 v0.21.0 // indirect golang.org/x/sync v0.11.0 // indirect golang.org/x/sys v0.30.0 // indirect golang.org/x/text v0.22.0 // indirect golang.org/x/time v0.6.0 // indirect + golang.org/x/tools v0.30.0 // indirect google.golang.org/api v0.188.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20240701130421-f6361c86f094 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240708141625-4ad9e859172b // indirect google.golang.org/grpc v1.64.1 // indirect - google.golang.org/protobuf v1.34.2 // indirect + google.golang.org/protobuf v1.36.1 // indirect gopkg.in/ini.v1 v1.62.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/cli/go.sum b/cli/go.sum index 20215d776..e274c0d38 100644 --- a/cli/go.sum +++ b/cli/go.sum @@ -144,8 +144,8 @@ github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ= -github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= +github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-openapi/errors v0.20.2 h1:dxy7PGTqEh94zj2E3h1cUmQQWiM1+aeCROfAr02EmK8= @@ -154,6 +154,8 @@ github.com/go-openapi/strfmt v0.21.3 h1:xwhj5X6CjXEZZHMWy1zKJxvW9AfHC9pkyUjLvHtK github.com/go-openapi/strfmt v0.21.3/go.mod h1:k+RzNO0Da+k3FrrynSNN8F7n/peCmQQqbbXjtDfvmGg= github.com/go-resty/resty/v2 v2.16.5 h1:hBKqmWrr7uRc3euHVqmh1HTHcKn99Smr7o5spptdhTM= github.com/go-resty/resty/v2 v2.16.5/go.mod h1:hkJtXbA2iKHzJheXYvQ8snQES5ZLGKMwQ07xAwp/fiA= +github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= +github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk= github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= @@ -222,6 +224,8 @@ github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLe github.com/google/pprof v0.0.0-20201203190320-1bf35d6f28c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= github.com/google/pprof v0.0.0-20210122040257-d980be63207e/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= +github.com/google/pprof v0.0.0-20250302191652-9094ed2288e7 h1:+J3r2e8+RsmN3vKfo75g0YSY61ms37qzPglu4p0sGro= +github.com/google/pprof v0.0.0-20250302191652-9094ed2288e7/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144= github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= github.com/google/s2a-go v0.1.7 h1:60BLSyTrOV4/haCDW4zb1guZItoSq8foHCXrAnjBo/o= github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw= @@ -342,6 +346,10 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWb github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= +github.com/onsi/ginkgo/v2 v2.22.2 h1:/3X8Panh8/WwhU/3Ssa6rCKqPLuAkVY2I0RoyDLySlU= +github.com/onsi/ginkgo/v2 v2.22.2/go.mod h1:oeMosUL+8LtarXBHu/c0bx2D/K9zyQ6uX3cTyztHwsk= +github.com/onsi/gomega v1.36.2 h1:koNYke6TVk6ZmnyHrCXba/T/MoLBXFjeC1PtvYgw0A8= +github.com/onsi/gomega v1.36.2/go.mod h1:DdwyADRjrc825LhMEkD76cHR5+pUnjhUN8GlHlRPHzY= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pelletier/go-toml v1.9.3 h1:zeC5b1GviRUyKYd6OJPvBU/mcVDVoL1OhT17FCt5dSQ= github.com/pelletier/go-toml v1.9.3/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c= @@ -369,6 +377,8 @@ github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndr github.com/posthog/posthog-go v0.0.0-20221221115252-24dfed35d71a h1:Ey0XWvrg6u6hyIn1Kd/jCCmL+bMv9El81tvuGBbxZGg= github.com/posthog/posthog-go v0.0.0-20221221115252-24dfed35d71a/go.mod h1:oa2sAs9tGai3VldabTV0eWejt/O4/OOD7azP8GaikqU= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/quic-go/quic-go v0.50.0 h1:3H/ld1pa3CYhkcc20TPIyG1bNsdhn9qZBGN3b9/UyUo= +github.com/quic-go/quic-go v0.50.0/go.mod h1:Vim6OmUvlYdwBhXP9ZVrtGmCMWa3wEqhq3NgYrI8b4E= github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= @@ -461,6 +471,8 @@ go.opentelemetry.io/otel/metric v1.24.0/go.mod h1:VYhLe1rFfxuTXLgj4CBiyz+9WYBA8p go.opentelemetry.io/otel/trace v1.24.0 h1:CsKnnL4dUAr/0llH9FKuc698G04IrpWV0MQA/Y1YELI= go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU= +go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM= go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo= golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= @@ -472,8 +484,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8= golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= -golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus= -golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M= +golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs= +golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -484,6 +496,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4= golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM= golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU= +golang.org/x/exp v0.0.0-20250228200357-dead58393ab7 h1:aWwlzYV971S4BXRS9AmqwDLAD85ouC6X+pocatKY58c= +golang.org/x/exp v0.0.0-20250228200357-dead58393ab7/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk= golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= @@ -509,6 +523,8 @@ golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM= +golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -547,8 +563,8 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= -golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= +golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8= +golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -697,6 +713,8 @@ golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4f golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo= +golang.org/x/tools v0.30.0 h1:BgcpHewrV5AUp2G9MebG4XPFI1E2W41zU1SaqVA9vJY= +golang.org/x/tools v0.30.0/go.mod h1:c347cR/OJfw5TI+GfX7RUPNMdDRRbjvYTS0jPyvsVtY= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -811,8 +829,8 @@ google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGj google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= -google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= -google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk= +google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/cli/packages/cmd/gateway.go b/cli/packages/cmd/gateway.go index a0b664664..760fd95c4 100644 --- a/cli/packages/cmd/gateway.go +++ b/cli/packages/cmd/gateway.go @@ -109,6 +109,33 @@ var gatewayCmd = &cobra.Command{ }, } +var gatewayRelayCmd = &cobra.Command{ + Example: `infisical gateway relay`, + Short: "Used to run infisical gateway relay", + Use: "relay", + DisableFlagsInUseLine: true, + Args: cobra.NoArgs, + Run: func(cmd *cobra.Command, args []string) { + relayConfigFilePath, err := cmd.Flags().GetString("config") + if err != nil { + util.HandleError(err, "Unable to parse flag") + } + + if relayConfigFilePath == "" { + util.HandleError(fmt.Errorf("Missing config file")) + } + + gatewayRelay, err := gateway.NewGatewayRelay(relayConfigFilePath) + if err != nil { + util.HandleError(err, "Failed to initialize gateway") + } + err = gatewayRelay.Run() + if err != nil { + util.HandleError(err, "Failed to start gateway") + } + }, +} + func init() { gatewayCmd.SetHelpFunc(func(command *cobra.Command, strings []string) { command.Flags().MarkHidden("domain") @@ -116,5 +143,9 @@ func init() { }) gatewayCmd.Flags().String("token", "", "Connect with Infisical using machine identity access token") + gatewayRelayCmd.Flags().String("config", "", "Relay config yaml file path") + + gatewayCmd.AddCommand(gatewayRelayCmd) + rootCmd.AddCommand(gatewayCmd) } diff --git a/cli/packages/gateway/connection.go b/cli/packages/gateway/connection.go index c29804506..d8bfebc91 100644 --- a/cli/packages/gateway/connection.go +++ b/cli/packages/gateway/connection.go @@ -3,20 +3,49 @@ package gateway import ( "bufio" "bytes" + "context" "errors" "io" "net" + "strings" "sync" + "github.com/quic-go/quic-go" "github.com/rs/zerolog/log" ) -func handleConnection(conn net.Conn) { - defer conn.Close() - log.Info().Msgf("New connection from: %s", conn.RemoteAddr().String()) +func handleConnection(ctx context.Context, quicConn quic.Connection) { + log.Info().Msgf("New connection from: %s", quicConn.RemoteAddr().String()) + // Use WaitGroup to track all streams + var wg sync.WaitGroup + for { + // Accept the first stream, which we'll use for commands + stream, err := quicConn.AcceptStream(ctx) + if err != nil { + log.Printf("Failed to accept QUIC stream: %v", err) + break + } + wg.Add(1) + go func(stream quic.Stream) { + defer wg.Done() + defer stream.Close() + + handleStream(stream, quicConn) + }(stream) + } + + wg.Wait() + log.Printf("All streams closed for connection: %s", quicConn.RemoteAddr().String()) +} + +func handleStream(stream quic.Stream, quicConn quic.Connection) { + streamID := stream.StreamID() + log.Printf("New stream %d from: %s", streamID, quicConn.RemoteAddr().String()) // Use buffered reader for better handling of fragmented data - reader := bufio.NewReader(conn) + reader := bufio.NewReader(stream) + defer stream.Close() + for { msg, err := reader.ReadBytes('\n') if err != nil { @@ -32,6 +61,7 @@ func handleConnection(conn net.Conn) { switch string(cmd) { case "FORWARD-TCP": + log.Info().Msg("Starting secure connector proxy...") proxyAddress := string(bytes.Split(args, []byte(" "))[0]) destTarget, err := net.Dial("tcp", proxyAddress) if err != nil { @@ -56,10 +86,10 @@ func handleConnection(conn net.Conn) { } } - CopyData(conn, destTarget) + CopyDataFromQuicToTcp(stream, destTarget) return case "PING": - if _, err := conn.Write([]byte("PONG")); err != nil { + if _, err := stream.Write([]byte("PONG\n")); err != nil { log.Error().Msgf("Error writing PONG response: %v", err) } return @@ -74,34 +104,38 @@ type CloseWrite interface { CloseWrite() error } -func CopyData(src, dst net.Conn) { +func CopyDataFromQuicToTcp(quicStream quic.Stream, tcpConn net.Conn) { + // Create a WaitGroup to wait for both copy operations var wg sync.WaitGroup wg.Add(2) - copyAndClose := func(dst, src net.Conn, done chan<- bool) { + // Start copying from QUIC stream to TCP + go func() { defer wg.Done() - _, err := io.Copy(dst, src) - if err != nil && !errors.Is(err, io.EOF) { - log.Error().Msgf("Copy error: %v", err) + if _, err := io.Copy(tcpConn, quicStream); err != nil { + log.Error().Msgf("Error copying quic->postgres: %v", err) } - // Signal we're done writing - done <- true - - // Half close the connection if possible - if c, ok := dst.(CloseWrite); ok { - c.CloseWrite() + if e, ok := tcpConn.(CloseWrite); ok { + log.Debug().Msg("Closing TCP write end") + e.CloseWrite() + } else { + log.Debug().Msg("TCP connection does not support CloseWrite") } - } + }() - done1 := make(chan bool, 1) - done2 := make(chan bool, 1) - - go copyAndClose(dst, src, done1) - go copyAndClose(src, dst, done2) + // Start copying from TCP to QUIC stream + go func() { + defer wg.Done() + if _, err := io.Copy(quicStream, tcpConn); err != nil { + log.Debug().Msgf("Error copying postgres->quic: %v", err) + } + // Close the write side of the QUIC stream + if err := quicStream.Close(); err != nil && !strings.Contains(err.Error(), "close called for canceled stream") { + log.Error().Msgf("Error closing QUIC stream write: %v", err) + } + }() // Wait for both copies to complete - <-done1 - <-done2 wg.Wait() } diff --git a/cli/packages/gateway/gateway.go b/cli/packages/gateway/gateway.go index 5960e4ed5..6758f83b5 100644 --- a/cli/packages/gateway/gateway.go +++ b/cli/packages/gateway/gateway.go @@ -15,6 +15,8 @@ import ( "github.com/pion/logging" "github.com/pion/turn/v4" "github.com/rs/zerolog/log" + + "github.com/quic-go/quic-go" ) type GatewayConfig struct { @@ -56,13 +58,12 @@ func (g *Gateway) ConnectWithRelay() error { if relayPort == "5349" { log.Info().Msgf("Provided relay port %s. Using TLS", relayPort) conn, err = tls.Dial("tcp", relayDetails.TurnServerAddress, &tls.Config{ - InsecureSkipVerify: false, - ServerName: relayAddress, + ServerName: relayAddress, }) } else { log.Info().Msgf("Provided relay port %s. Using non TLS connection.", relayPort) - peerAddr, err := net.ResolveTCPAddr("tcp", relayDetails.TurnServerAddress) - if err != nil { + peerAddr, errPeer := net.ResolveTCPAddr("tcp", relayDetails.TurnServerAddress) + if errPeer != nil { return fmt.Errorf("Failed to parse turn server address: %w", err) } conn, err = net.DialTCP("tcp", nil, peerAddr) @@ -116,20 +117,20 @@ func (g *Gateway) Listen(ctx context.Context) error { // Allocate a relay socket on the TURN server. On success, it // will return a net.PacketConn which represents the remote // socket. - relayNonTlsConn, err := g.client.AllocateTCP() + relayUdpConnection, err := g.client.Allocate() if err != nil { return fmt.Errorf("Failed to allocate relay connection: %w", err) } - log.Info().Msg(relayNonTlsConn.Addr().String()) + log.Info().Msg(relayUdpConnection.LocalAddr().String()) defer func() { - if closeErr := relayNonTlsConn.Close(); closeErr != nil { + if closeErr := relayUdpConnection.Close(); closeErr != nil { log.Error().Msgf("Failed to close connection: %s", closeErr) } }() gatewayCert, err := api.CallExchangeRelayCertV1(g.httpClient, api.ExchangeRelayCertRequestV1{ - RelayAddress: relayNonTlsConn.Addr().String(), + RelayAddress: relayUdpConnection.LocalAddr().String(), }) if err != nil { return err @@ -144,45 +145,58 @@ func (g *Gateway) Listen(ctx context.Context) error { if g.config.InfisicalStaticIp != "" { log.Info().Msgf("Found static ip from Infisical: %s. Creating permission IP lifecycle", g.config.InfisicalStaticIp) - peerAddr, err := net.ResolveTCPAddr("tcp", g.config.InfisicalStaticIp) + peerAddr, err := net.ResolveUDPAddr("udp", g.config.InfisicalStaticIp) if err != nil { return fmt.Errorf("Failed to parse infisical static ip: %w", err) } - g.registerPermissionLifecycle(func() error { - err := relayNonTlsConn.CreatePermissions(peerAddr) - return err - }, shutdownCh) + err = g.client.CreatePermission(peerAddr) + if err != nil { + return fmt.Errorf("Failed to set permission: %w", err) + } } cert, err := tls.X509KeyPair([]byte(gatewayCert.Certificate), []byte(gatewayCert.PrivateKey)) if err != nil { - return fmt.Errorf("failed to parse cert: %s", err) + return fmt.Errorf("failed to parse cert: %w", err) } caCertPool := x509.NewCertPool() caCertPool.AppendCertsFromPEM([]byte(gatewayCert.CertificateChain)) - relayConn := tls.NewListener(relayNonTlsConn, &tls.Config{ + // Setup QUIC server + tlsConfig := &tls.Config{ Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12, ClientCAs: caCertPool, ClientAuth: tls.RequireAndVerifyClientCert, - }) + NextProtos: []string{"infisical-gateway"}, + } + + // Setup QUIC listener on the relayConn + quicConfig := &quic.Config{ + EnableDatagrams: true, + MaxIdleTimeout: 30 * time.Second, + KeepAlivePeriod: 15 * time.Second, + } + + quicListener, err := quic.Listen(relayUdpConnection, tlsConfig, quicConfig) + if err != nil { + return fmt.Errorf("Failed to listen for QUIC: %w", err) + } + defer quicListener.Close() + + log.Printf("Listener started on %s", quicListener.Addr()) errCh := make(chan error, 1) log.Info().Msg("Gateway started successfully") g.registerHeartBeat(errCh, shutdownCh) - g.registerRelayIsActive(relayNonTlsConn.Addr().String(), errCh, shutdownCh) + g.registerRelayIsActive(relayUdpConnection.LocalAddr().String(), tlsConfig, quicConfig, errCh, shutdownCh) // Create a WaitGroup to track active connections var wg sync.WaitGroup go func() { for { - if relayDeadlineConn, ok := relayConn.(*net.TCPListener); ok { - relayDeadlineConn.SetDeadline(time.Now().Add(1 * time.Second)) - } - select { case <-ctx.Done(): return @@ -190,67 +204,42 @@ func (g *Gateway) Listen(ctx context.Context) error { return default: // Accept new relay connection - conn, err := relayConn.Accept() + quicConn, err := quicListener.Accept(context.Background()) if err != nil { - // Check if it's a timeout error (which we expect due to our deadline) - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - continue - } - - if !strings.Contains(err.Error(), "data contains incomplete STUN or TURN frame") { - log.Error().Msgf("Failed to accept connection: %v", err) - } + log.Printf("Failed to accept QUIC connection: %v", err) continue } - tlsConn, ok := conn.(*tls.Conn) - if !ok { - log.Error().Msg("Failed to convert to TLS connection") - conn.Close() - continue - } - - // Set a deadline for the handshake to prevent hanging - tlsConn.SetDeadline(time.Now().Add(10 * time.Second)) - err = tlsConn.Handshake() - // Clear the deadline after handshake - tlsConn.SetDeadline(time.Time{}) - if err != nil { - log.Error().Msgf("TLS handshake failed: %v", err) - conn.Close() - continue - } - - // Get connection state which contains certificate information - state := tlsConn.ConnectionState() - if len(state.PeerCertificates) > 0 { - organizationUnit := state.PeerCertificates[0].Subject.OrganizationalUnit - commonName := state.PeerCertificates[0].Subject.CommonName + tlsState := quicConn.ConnectionState().TLS + if len(tlsState.PeerCertificates) > 0 { + organizationUnit := tlsState.PeerCertificates[0].Subject.OrganizationalUnit + commonName := tlsState.PeerCertificates[0].Subject.CommonName if organizationUnit[0] != "gateway-client" || commonName != "cloud" { - log.Error().Msgf("Client certificate verification failed. Received %s, %s", organizationUnit, commonName) - conn.Close() + errMsg := fmt.Sprintf("Client certificate verification failed. Received %s, %s", organizationUnit, commonName) + log.Error().Msg(errMsg) + quicConn.CloseWithError(1, errMsg) continue } } // Handle the connection in a goroutine wg.Add(1) - go func(c net.Conn) { + go func(c quic.Connection) { defer wg.Done() - defer c.Close() + defer c.CloseWithError(0, "connection closed") // Monitor parent context to close this connection when needed go func() { select { case <-ctx.Done(): - c.Close() // Force close connection when context is canceled + c.CloseWithError(0, "connection closed") // Force close connection when context is canceled case <-shutdownCh: - c.Close() // Force close connection when accepting loop is done + c.CloseWithError(0, "connection closed") // Force close connection when accepting loop is done } }() - handleConnection(c) - }(conn) + handleConnection(ctx, c) + }(quicConn) } } }() @@ -282,7 +271,7 @@ func (g *Gateway) Listen(ctx context.Context) error { } func (g *Gateway) registerHeartBeat(errCh chan error, done chan bool) { - ticker := time.NewTicker(1 * time.Hour) + ticker := time.NewTicker(30 * time.Minute) go func() { time.Sleep(10 * time.Second) @@ -306,26 +295,7 @@ func (g *Gateway) registerHeartBeat(errCh chan error, done chan bool) { }() } -func (g *Gateway) registerPermissionLifecycle(permissionFn func() error, done chan bool) { - ticker := time.NewTicker(3 * time.Minute) - - go func() { - // wait for 5 mins - permissionFn() - log.Printf("Created permission for incoming connections") - for { - select { - case <-done: - ticker.Stop() - return - case <-ticker.C: - permissionFn() - } - } - }() -} - -func (g *Gateway) registerRelayIsActive(serverAddr string, errCh chan error, done chan bool) { +func (g *Gateway) registerRelayIsActive(serverAddr string, tlsConf *tls.Config, quicConf *quic.Config, errCh chan error, done chan bool) { ticker := time.NewTicker(10 * time.Second) go func() { @@ -336,14 +306,19 @@ func (g *Gateway) registerRelayIsActive(serverAddr string, errCh chan error, don ticker.Stop() return case <-ticker.C: - conn, err := net.Dial("tcp", serverAddr) - if err != nil { - errCh <- err - return - } - if conn != nil { - conn.Close() - } + func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) // 3s handshake timeout + defer cancel() + conn, err := quic.DialAddr(ctx, serverAddr, tlsConf, quicConf) + if conn != nil { + conn.CloseWithError(0, "connection closed") + } + // this error means quic connection is alive + if err != nil && !strings.Contains(err.Error(), "tls: failed to verify certificate") { + errCh <- err + return + } + }() } } }() diff --git a/cli/packages/gateway/relay.go b/cli/packages/gateway/relay.go new file mode 100644 index 000000000..8141b087c --- /dev/null +++ b/cli/packages/gateway/relay.go @@ -0,0 +1,186 @@ +package gateway + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "os" + "os/signal" + "runtime" + "strconv" + "syscall" + + "github.com/pion/logging" + "github.com/pion/turn/v4" + "github.com/rs/zerolog/log" + "golang.org/x/sys/unix" + "gopkg.in/yaml.v2" +) + +var ( + errMissingTlsCert = errors.New("Missing TLS files") +) + +type GatewayRelay struct { + Config *GatewayRelayConfig +} + +type GatewayRelayConfig struct { + PublicIP string `json:"public_ip"` + Port int `json:"port"` + Realm string `json:"realm"` + AuthSecret string `json:"auth_secret"` + RelayMinPort uint16 `json:"relay_min_port"` + RelayMaxPort uint16 `json:"relay_max_port"` + TlsCertPath string `json:"tls_cert_path"` + TlsPrivateKeyPath string `json:"tls_private_key_path"` + + tls tls.Certificate + isTlsEnabled bool +} + +func NewGatewayRelay(configFilePath string) (*GatewayRelay, error) { + cfgFile, err := os.ReadFile(configFilePath) + if err != nil { + return nil, err + } + var cfg GatewayRelayConfig + if err := yaml.Unmarshal(cfgFile, &cfg); err != nil { + return nil, err + } + + if cfg.PublicIP == "" { + return nil, fmt.Errorf("Missing public ip") + } + + if cfg.AuthSecret == "" { + return nil, fmt.Errorf("Missing auth secret") + } + + if cfg.Realm == "" { + cfg.Realm = "infisical.org" + } + + if cfg.RelayMinPort == 0 { + cfg.RelayMinPort = 49152 + } + + if cfg.RelayMaxPort == 0 { + cfg.RelayMaxPort = 65535 + } + + if cfg.Port == 0 { + cfg.Port = 3478 + } else if cfg.Port == 5349 { + if cfg.TlsCertPath == "" || cfg.TlsPrivateKeyPath == "" { + return nil, errMissingTlsCert + } + + tlsCertFile, err := os.ReadFile(cfg.TlsCertPath) + if err != nil { + return nil, err + } + tlsPrivateKeyFile, err := os.ReadFile(cfg.TlsPrivateKeyPath) + if err != nil { + return nil, err + } + + cert, err := tls.LoadX509KeyPair(string(tlsCertFile), string(tlsPrivateKeyFile)) + if err != nil { + return nil, err + } + cfg.tls = cert + cfg.isTlsEnabled = true + } + + return &GatewayRelay{ + Config: &cfg, + }, nil +} + +func (g *GatewayRelay) Run() error { + addr, err := net.ResolveTCPAddr("tcp", "0.0.0.0:"+strconv.Itoa(g.Config.Port)) + if err != nil { + return fmt.Errorf("Failed to parse server address: %s", err) + } + + // NewLongTermAuthHandler takes a pion.LeveledLogger. This allows you to intercept messages + // and process them yourself. + logger := logging.NewDefaultLeveledLoggerForScope("lt-creds", logging.LogLevelTrace, os.Stdout) + + // Create `numThreads` UDP listeners to pass into pion/turn + // pion/turn itself doesn't allocate any UDP sockets, but lets the user pass them in + // this allows us to add logging, storage or modify inbound/outbound traffic + // UDP listeners share the same local address:port with setting SO_REUSEPORT and the kernel + // will load-balance received packets per the IP 5-tuple + listenerConfig := &net.ListenConfig{ + Control: func(network, address string, conn syscall.RawConn) error { // nolint: revive + var operr error + if err = conn.Control(func(fd uintptr) { + operr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, unix.SO_REUSEPORT, 1) + }); err != nil { + return err + } + + return operr + }, + } + + publicIP := g.Config.PublicIP + relayAddressGenerator := &turn.RelayAddressGeneratorPortRange{ + RelayAddress: net.ParseIP(publicIP), // Claim that we are listening on IP passed by user + Address: "0.0.0.0", // But actually be listening on every interface + MinPort: g.Config.RelayMinPort, + MaxPort: g.Config.RelayMaxPort, + } + + threadNum := runtime.NumCPU() + listenerConfigs := make([]turn.ListenerConfig, threadNum) + for i := 0; i < threadNum; i++ { + conn, listErr := listenerConfig.Listen(context.Background(), addr.Network(), addr.String()) + if listErr != nil { + return fmt.Errorf("Failed to allocate TCP listener at %s:%s %s", addr.Network(), addr.String(), listErr) + } + + listenerConfigs[i] = turn.ListenerConfig{ + RelayAddressGenerator: relayAddressGenerator, + } + + if g.Config.isTlsEnabled { + listenerConfigs[i].Listener = tls.NewListener(conn, &tls.Config{ + Certificates: []tls.Certificate{g.Config.tls}, + }) + } else { + listenerConfigs[i].Listener = conn + } + + log.Printf("Server %d listening on %s\n", i, conn.Addr().String()) + } + + loggerF := logging.NewDefaultLoggerFactory() + loggerF.DefaultLogLevel = logging.LogLevelDebug + + server, err := turn.NewServer(turn.ServerConfig{ + Realm: g.Config.Realm, + AuthHandler: turn.LongTermTURNRESTAuthHandler(g.Config.AuthSecret, logger), + // PacketConnConfigs is a list of UDP Listeners and the configuration around them + ListenerConfigs: listenerConfigs, + LoggerFactory: loggerF, + }) + + if err != nil { + return fmt.Errorf("Failed to start server: %w", err) + } + + // Block until user sends SIGINT or SIGTERM + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + <-sigs + + if err = server.Close(); err != nil { + return fmt.Errorf("Failed to close server: %w", err) + } + return nil +}