mirror of
https://github.com/awatertrevi/infisical.git
synced 2026-10-14 20:51:50 +00:00
Merge pull request #3731 from Infisical/daniel/gateway-auth-methods
feat(identities/kubernetes-auth): gateway as token reviewer
This commit is contained in:
@@ -0,0 +1,23 @@
|
||||
import { Knex } from "knex";
|
||||
|
||||
import { TableName } from "../schemas";
|
||||
|
||||
export async function up(knex: Knex): Promise<void> {
|
||||
const hasTokenReviewModeColumn = await knex.schema.hasColumn(TableName.IdentityKubernetesAuth, "tokenReviewMode");
|
||||
|
||||
if (!hasTokenReviewModeColumn) {
|
||||
await knex.schema.alterTable(TableName.IdentityKubernetesAuth, (table) => {
|
||||
table.string("tokenReviewMode").notNullable().defaultTo("api");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export async function down(knex: Knex): Promise<void> {
|
||||
const hasTokenReviewModeColumn = await knex.schema.hasColumn(TableName.IdentityKubernetesAuth, "tokenReviewMode");
|
||||
|
||||
if (hasTokenReviewModeColumn) {
|
||||
await knex.schema.alterTable(TableName.IdentityKubernetesAuth, (table) => {
|
||||
table.dropColumn("tokenReviewMode");
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -31,7 +31,8 @@ export const IdentityKubernetesAuthsSchema = z.object({
|
||||
encryptedKubernetesTokenReviewerJwt: zodBuffer.nullable().optional(),
|
||||
encryptedKubernetesCaCertificate: zodBuffer.nullable().optional(),
|
||||
gatewayId: z.string().uuid().nullable().optional(),
|
||||
accessTokenPeriod: z.coerce.number().default(0)
|
||||
accessTokenPeriod: z.coerce.number().default(0),
|
||||
tokenReviewMode: z.string().default("api")
|
||||
});
|
||||
|
||||
export type TIdentityKubernetesAuths = z.infer<typeof IdentityKubernetesAuthsSchema>;
|
||||
|
||||
@@ -2,7 +2,7 @@ import axios from "axios";
|
||||
import https from "https";
|
||||
|
||||
import { InternalServerError } from "@app/lib/errors";
|
||||
import { 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,6 +43,7 @@ export const KubernetesProvider = ({ gatewayService }: TKubernetesProviderDTO):
|
||||
return res;
|
||||
},
|
||||
{
|
||||
protocol: GatewayProxyProtocol.Tcp,
|
||||
targetHost: inputs.targetHost,
|
||||
targetPort: inputs.targetPort,
|
||||
relayHost,
|
||||
|
||||
@@ -3,7 +3,7 @@ import handlebars from "handlebars";
|
||||
import knex from "knex";
|
||||
import { z } from "zod";
|
||||
|
||||
import { 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,6 +185,7 @@ export const SqlDatabaseProvider = ({ gatewayService }: TSqlDatabaseProviderDTO)
|
||||
await gatewayCallback("localhost", port);
|
||||
},
|
||||
{
|
||||
protocol: GatewayProxyProtocol.Tcp,
|
||||
targetHost: providerInputs.host,
|
||||
targetPort: providerInputs.port,
|
||||
relayHost,
|
||||
|
||||
@@ -4,7 +4,7 @@ import knex, { Knex } from "knex";
|
||||
import { z } from "zod";
|
||||
|
||||
import { BadRequestError } from "@app/lib/errors";
|
||||
import { 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,6 +196,7 @@ export const VerticaProvider = ({ gatewayService }: TVerticaProviderDTO): TDynam
|
||||
await gatewayCallback("localhost", port);
|
||||
},
|
||||
{
|
||||
protocol: GatewayProxyProtocol.Tcp,
|
||||
targetHost: providerInputs.host,
|
||||
targetPort: providerInputs.port,
|
||||
relayHost,
|
||||
|
||||
@@ -400,6 +400,8 @@ export const KUBERNETES_AUTH = {
|
||||
caCert: "The PEM-encoded CA cert for the Kubernetes API server.",
|
||||
tokenReviewerJwt:
|
||||
"Optional JWT token for accessing Kubernetes TokenReview API. If provided, this long-lived token will be used to validate service account tokens during authentication. If omitted, the client's own JWT will be used instead, which requires the client to have the system:auth-delegator ClusterRole binding.",
|
||||
tokenReviewMode:
|
||||
"The mode to use for token review. Must be one of: 'api', 'gateway'. If gateway is selected, the gateway must be deployed in Kubernetes, and the gateway must have the system:auth-delegator ClusterRole binding.",
|
||||
allowedNamespaces:
|
||||
"The comma-separated list of trusted namespaces that service accounts must belong to authenticate with Infisical.",
|
||||
allowedNames: "The comma-separated list of trusted service account names that can authenticate with Infisical.",
|
||||
@@ -417,6 +419,8 @@ export const KUBERNETES_AUTH = {
|
||||
caCert: "The new PEM-encoded CA cert for the Kubernetes API server.",
|
||||
tokenReviewerJwt:
|
||||
"Optional JWT token for accessing Kubernetes TokenReview API. If provided, this long-lived token will be used to validate service account tokens during authentication. If omitted, the client's own JWT will be used instead, which requires the client to have the system:auth-delegator ClusterRole binding.",
|
||||
tokenReviewMode:
|
||||
"The mode to use for token review. Must be one of: 'api', 'gateway'. If gateway is selected, the gateway must be deployed in Kubernetes, and the gateway must have the system:auth-delegator ClusterRole binding.",
|
||||
allowedNamespaces:
|
||||
"The new comma-separated list of trusted namespaces that service accounts must belong to authenticate with Infisical.",
|
||||
allowedNames: "The new comma-separated list of trusted service account names that can authenticate with Infisical.",
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
};
|
||||
@@ -1,392 +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 { 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;
|
||||
}
|
||||
|
||||
const setupProxyServer = async ({
|
||||
targetPort,
|
||||
targetHost,
|
||||
tlsOptions,
|
||||
relayHost,
|
||||
relayPort,
|
||||
identityId,
|
||||
orgId
|
||||
}: {
|
||||
targetHost: string;
|
||||
targetPort: number;
|
||||
relayPort: number;
|
||||
relayHost: string;
|
||||
tlsOptions: TTlsOption;
|
||||
identityId: string;
|
||||
orgId: string;
|
||||
}): 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");
|
||||
// Send FORWARD-TCP command
|
||||
const forwardWriter = stream.writable.getWriter();
|
||||
await forwardWriter.write(Buffer.from(`FORWARD-TCP ${targetHost}:${targetPort}\n`));
|
||||
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");
|
||||
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;
|
||||
}
|
||||
|
||||
export const withGatewayProxy = async <T>(
|
||||
callback: (port: number) => Promise<T>,
|
||||
options: ProxyOptions
|
||||
): Promise<T> => {
|
||||
const { relayHost, relayPort, targetHost, targetPort, tlsOptions, identityId, orgId } = options;
|
||||
|
||||
// Setup the proxy server
|
||||
const { port, cleanup, getProxyError } = await setupProxyServer({
|
||||
targetHost,
|
||||
targetPort,
|
||||
relayPort,
|
||||
relayHost,
|
||||
tlsOptions,
|
||||
identityId,
|
||||
orgId
|
||||
});
|
||||
|
||||
try {
|
||||
// Execute the callback with the allocated port
|
||||
return await callback(port);
|
||||
} 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";
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import { readLimit, writeLimit } from "@app/server/config/rateLimiter";
|
||||
import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
|
||||
import { AuthMode } from "@app/services/auth/auth-type";
|
||||
import { TIdentityTrustedIp } from "@app/services/identity/identity-types";
|
||||
import { IdentityKubernetesAuthTokenReviewMode } from "@app/services/identity-kubernetes-auth/identity-kubernetes-auth-types";
|
||||
import { isSuperAdmin } from "@app/services/super-admin/super-admin-fns";
|
||||
|
||||
const IdentityKubernetesAuthResponseSchema = IdentityKubernetesAuthsSchema.pick({
|
||||
@@ -18,6 +19,7 @@ const IdentityKubernetesAuthResponseSchema = IdentityKubernetesAuthsSchema.pick(
|
||||
accessTokenTrustedIps: true,
|
||||
createdAt: true,
|
||||
updatedAt: true,
|
||||
tokenReviewMode: true,
|
||||
identityId: true,
|
||||
kubernetesHost: true,
|
||||
allowedNamespaces: true,
|
||||
@@ -124,6 +126,10 @@ export const registerIdentityKubernetesRouter = async (server: FastifyZodProvide
|
||||
),
|
||||
caCert: z.string().trim().default("").describe(KUBERNETES_AUTH.ATTACH.caCert),
|
||||
tokenReviewerJwt: z.string().trim().optional().describe(KUBERNETES_AUTH.ATTACH.tokenReviewerJwt),
|
||||
tokenReviewMode: z
|
||||
.nativeEnum(IdentityKubernetesAuthTokenReviewMode)
|
||||
.default(IdentityKubernetesAuthTokenReviewMode.Api)
|
||||
.describe(KUBERNETES_AUTH.ATTACH.tokenReviewMode),
|
||||
allowedNamespaces: z.string().describe(KUBERNETES_AUTH.ATTACH.allowedNamespaces), // TODO: validation
|
||||
allowedNames: z.string().describe(KUBERNETES_AUTH.ATTACH.allowedNames),
|
||||
allowedAudience: z.string().describe(KUBERNETES_AUTH.ATTACH.allowedAudience),
|
||||
@@ -157,10 +163,22 @@ export const registerIdentityKubernetesRouter = async (server: FastifyZodProvide
|
||||
.default(0)
|
||||
.describe(KUBERNETES_AUTH.ATTACH.accessTokenNumUsesLimit)
|
||||
})
|
||||
.refine(
|
||||
(val) => val.accessTokenTTL <= val.accessTokenMaxTTL,
|
||||
"Access Token TTL cannot be greater than Access Token Max TTL."
|
||||
),
|
||||
.superRefine((data, ctx) => {
|
||||
if (data.tokenReviewMode === IdentityKubernetesAuthTokenReviewMode.Gateway && !data.gatewayId) {
|
||||
ctx.addIssue({
|
||||
path: ["gatewayId"],
|
||||
code: z.ZodIssueCode.custom,
|
||||
message: "When token review mode is set to Gateway, a gateway must be selected"
|
||||
});
|
||||
}
|
||||
if (data.accessTokenTTL > data.accessTokenMaxTTL) {
|
||||
ctx.addIssue({
|
||||
path: ["accessTokenTTL"],
|
||||
code: z.ZodIssueCode.custom,
|
||||
message: "Access Token TTL cannot be greater than Access Token Max TTL."
|
||||
});
|
||||
}
|
||||
}),
|
||||
response: {
|
||||
200: z.object({
|
||||
identityKubernetesAuth: IdentityKubernetesAuthResponseSchema
|
||||
@@ -247,6 +265,10 @@ export const registerIdentityKubernetesRouter = async (server: FastifyZodProvide
|
||||
),
|
||||
caCert: z.string().trim().optional().describe(KUBERNETES_AUTH.UPDATE.caCert),
|
||||
tokenReviewerJwt: z.string().trim().nullable().optional().describe(KUBERNETES_AUTH.UPDATE.tokenReviewerJwt),
|
||||
tokenReviewMode: z
|
||||
.nativeEnum(IdentityKubernetesAuthTokenReviewMode)
|
||||
.optional()
|
||||
.describe(KUBERNETES_AUTH.UPDATE.tokenReviewMode),
|
||||
allowedNamespaces: z.string().optional().describe(KUBERNETES_AUTH.UPDATE.allowedNamespaces), // TODO: validation
|
||||
allowedNames: z.string().optional().describe(KUBERNETES_AUTH.UPDATE.allowedNames),
|
||||
allowedAudience: z.string().optional().describe(KUBERNETES_AUTH.UPDATE.allowedAudience),
|
||||
@@ -280,10 +302,26 @@ export const registerIdentityKubernetesRouter = async (server: FastifyZodProvide
|
||||
.optional()
|
||||
.describe(KUBERNETES_AUTH.UPDATE.accessTokenMaxTTL)
|
||||
})
|
||||
.refine(
|
||||
(val) => (val.accessTokenMaxTTL && val.accessTokenTTL ? val.accessTokenTTL <= val.accessTokenMaxTTL : true),
|
||||
"Access Token TTL cannot be greater than Access Token Max TTL."
|
||||
),
|
||||
.superRefine((data, ctx) => {
|
||||
if (
|
||||
data.tokenReviewMode &&
|
||||
data.tokenReviewMode === IdentityKubernetesAuthTokenReviewMode.Gateway &&
|
||||
!data.gatewayId
|
||||
) {
|
||||
ctx.addIssue({
|
||||
path: ["gatewayId"],
|
||||
code: z.ZodIssueCode.custom,
|
||||
message: "When token review mode is set to Gateway, a gateway must be selected"
|
||||
});
|
||||
}
|
||||
if (data.accessTokenMaxTTL && data.accessTokenTTL ? data.accessTokenTTL > data.accessTokenMaxTTL : false) {
|
||||
ctx.addIssue({
|
||||
path: ["accessTokenTTL"],
|
||||
code: z.ZodIssueCode.custom,
|
||||
message: "Access Token TTL cannot be greater than Access Token Max TTL."
|
||||
});
|
||||
}
|
||||
}),
|
||||
response: {
|
||||
200: z.object({
|
||||
identityKubernetesAuth: IdentityKubernetesAuthResponseSchema
|
||||
|
||||
+154
-30
@@ -20,8 +20,9 @@ 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 { 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";
|
||||
|
||||
import { ActorType, AuthTokenType } from "../auth/auth-type";
|
||||
import { TIdentityOrgDALFactory } from "../identity/identity-org-dal";
|
||||
@@ -33,6 +34,7 @@ import { validateIdentityUpdateForSuperAdminPrivileges } from "../super-admin/su
|
||||
import { TIdentityKubernetesAuthDALFactory } from "./identity-kubernetes-auth-dal";
|
||||
import { extractK8sUsername } from "./identity-kubernetes-auth-fns";
|
||||
import {
|
||||
IdentityKubernetesAuthTokenReviewMode,
|
||||
TAttachKubernetesAuthDTO,
|
||||
TCreateTokenReviewResponse,
|
||||
TGetKubernetesAuthDTO,
|
||||
@@ -72,19 +74,25 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
gatewayId: string;
|
||||
targetHost: string;
|
||||
targetPort: number;
|
||||
caCert?: string;
|
||||
reviewTokenThroughGateway: boolean;
|
||||
},
|
||||
gatewayCallback: (host: string, port: number) => Promise<T>
|
||||
gatewayCallback: (host: string, port: number, httpsAgent?: https.Agent) => Promise<T>
|
||||
): Promise<T> => {
|
||||
const relayDetails = await gatewayService.fnGetGatewayClientTlsByGatewayId(inputs.gatewayId);
|
||||
const [relayHost, relayPort] = relayDetails.relayAddress.split(":");
|
||||
|
||||
const callbackResult = await withGatewayProxy(
|
||||
async (port) => {
|
||||
// Needs to be https protocol or the kubernetes API server will fail with "Client sent an HTTP request to an HTTPS server"
|
||||
const res = await gatewayCallback("https://localhost", port);
|
||||
async (port, httpsAgent) => {
|
||||
const res = await gatewayCallback(
|
||||
inputs.reviewTokenThroughGateway ? "http://localhost" : "https://localhost",
|
||||
port,
|
||||
httpsAgent
|
||||
);
|
||||
return res;
|
||||
},
|
||||
{
|
||||
protocol: inputs.reviewTokenThroughGateway ? GatewayProxyProtocol.Http : GatewayProxyProtocol.Tcp,
|
||||
targetHost: inputs.targetHost,
|
||||
targetPort: inputs.targetPort,
|
||||
relayHost,
|
||||
@@ -95,7 +103,12 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
ca: relayDetails.certChain,
|
||||
cert: relayDetails.certificate,
|
||||
key: relayDetails.privateKey.toString()
|
||||
}
|
||||
},
|
||||
// we always pass this, because its needed for both tcp and http protocol
|
||||
httpsAgent: new https.Agent({
|
||||
ca: inputs.caCert,
|
||||
rejectUnauthorized: Boolean(inputs.caCert)
|
||||
})
|
||||
}
|
||||
);
|
||||
|
||||
@@ -129,22 +142,29 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
caCert = decryptor({ cipherTextBlob: identityKubernetesAuth.encryptedKubernetesCaCertificate }).toString();
|
||||
}
|
||||
|
||||
let tokenReviewerJwt = "";
|
||||
if (identityKubernetesAuth.encryptedKubernetesTokenReviewerJwt) {
|
||||
tokenReviewerJwt = decryptor({
|
||||
cipherTextBlob: identityKubernetesAuth.encryptedKubernetesTokenReviewerJwt
|
||||
}).toString();
|
||||
} else {
|
||||
// if no token reviewer is provided means the incoming token has to act as reviewer
|
||||
tokenReviewerJwt = serviceAccountJwt;
|
||||
}
|
||||
const tokenReviewCallbackRaw = async (host: string = identityKubernetesAuth.kubernetesHost, port?: number) => {
|
||||
logger.info({ host, port }, "tokenReviewCallbackRaw: Processing kubernetes token review using raw API");
|
||||
let tokenReviewerJwt = "";
|
||||
if (identityKubernetesAuth.encryptedKubernetesTokenReviewerJwt) {
|
||||
tokenReviewerJwt = decryptor({
|
||||
cipherTextBlob: identityKubernetesAuth.encryptedKubernetesTokenReviewerJwt
|
||||
}).toString();
|
||||
} else {
|
||||
// if no token reviewer is provided means the incoming token has to act as reviewer
|
||||
tokenReviewerJwt = serviceAccountJwt;
|
||||
}
|
||||
|
||||
let { kubernetesHost } = identityKubernetesAuth;
|
||||
if (kubernetesHost.startsWith("https://") || kubernetesHost.startsWith("http://")) {
|
||||
kubernetesHost = new RE2("^https?:\\/\\/").replace(kubernetesHost, "");
|
||||
}
|
||||
let servername = identityKubernetesAuth.kubernetesHost;
|
||||
if (servername.startsWith("https://") || servername.startsWith("http://")) {
|
||||
servername = new RE2("^https?:\\/\\/").replace(servername, "");
|
||||
}
|
||||
|
||||
// get the last colon index, if it has a port, remove it, including the colon
|
||||
const lastColonIndex = servername.lastIndexOf(":");
|
||||
if (lastColonIndex !== -1) {
|
||||
servername = servername.substring(0, lastColonIndex);
|
||||
}
|
||||
|
||||
const tokenReviewCallback = async (host: string = identityKubernetesAuth.kubernetesHost, port?: number) => {
|
||||
const baseUrl = port ? `${host}:${port}` : host;
|
||||
|
||||
const res = await axios
|
||||
@@ -165,11 +185,10 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
},
|
||||
signal: AbortSignal.timeout(10000),
|
||||
timeout: 10000,
|
||||
// if ca cert, rejectUnauthorized: true
|
||||
httpsAgent: new https.Agent({
|
||||
ca: caCert,
|
||||
rejectUnauthorized: Boolean(caCert),
|
||||
servername: kubernetesHost
|
||||
servername
|
||||
})
|
||||
}
|
||||
)
|
||||
@@ -192,18 +211,119 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
return res.data;
|
||||
};
|
||||
|
||||
const [k8sHost, k8sPort] = kubernetesHost.split(":");
|
||||
const tokenReviewCallbackThroughGateway = async (
|
||||
host: string = identityKubernetesAuth.kubernetesHost,
|
||||
port?: number,
|
||||
httpsAgent?: https.Agent
|
||||
) => {
|
||||
logger.info(
|
||||
{
|
||||
host,
|
||||
port
|
||||
},
|
||||
"tokenReviewCallbackThroughGateway: Processing kubernetes token review using gateway"
|
||||
);
|
||||
|
||||
const data = identityKubernetesAuth.gatewayId
|
||||
? await $gatewayProxyWrapper(
|
||||
const baseUrl = port ? `${host}:${port}` : host;
|
||||
|
||||
const res = await axios
|
||||
.post<TCreateTokenReviewResponse>(
|
||||
`${baseUrl}/apis/authentication.k8s.io/v1/tokenreviews`,
|
||||
{
|
||||
gatewayId: identityKubernetesAuth.gatewayId,
|
||||
targetHost: k8sHost,
|
||||
targetPort: k8sPort ? Number(k8sPort) : 443
|
||||
apiVersion: "authentication.k8s.io/v1",
|
||||
kind: "TokenReview",
|
||||
spec: {
|
||||
token: serviceAccountJwt,
|
||||
...(identityKubernetesAuth.allowedAudience ? { audiences: [identityKubernetesAuth.allowedAudience] } : {})
|
||||
}
|
||||
},
|
||||
tokenReviewCallback
|
||||
{
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"x-infisical-action": GatewayHttpProxyActions.InjectGatewayK8sServiceAccountToken
|
||||
},
|
||||
signal: AbortSignal.timeout(10000),
|
||||
timeout: 10000,
|
||||
...(httpsAgent ? { httpsAgent } : {})
|
||||
}
|
||||
)
|
||||
: await tokenReviewCallback();
|
||||
.catch((err) => {
|
||||
if (err instanceof AxiosError) {
|
||||
if (err.response) {
|
||||
let { message } = err?.response?.data as unknown as { message?: string };
|
||||
|
||||
if (!message && typeof err.response.data === "string") {
|
||||
message = err.response.data;
|
||||
}
|
||||
|
||||
if (message) {
|
||||
throw new UnauthorizedError({
|
||||
message,
|
||||
name: "KubernetesTokenReviewRequestError"
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
throw err;
|
||||
});
|
||||
|
||||
return res.data;
|
||||
};
|
||||
|
||||
let data: TCreateTokenReviewResponse | undefined;
|
||||
|
||||
if (identityKubernetesAuth.tokenReviewMode === IdentityKubernetesAuthTokenReviewMode.Gateway) {
|
||||
const { kubernetesHost } = identityKubernetesAuth;
|
||||
const lastColonIndex = kubernetesHost.lastIndexOf(":");
|
||||
const k8sHost = kubernetesHost.substring(0, lastColonIndex);
|
||||
const k8sPort = kubernetesHost.substring(lastColonIndex + 1);
|
||||
|
||||
if (!identityKubernetesAuth.gatewayId) {
|
||||
throw new BadRequestError({
|
||||
message: "Gateway ID is required when token review mode is set to Gateway"
|
||||
});
|
||||
}
|
||||
|
||||
data = await $gatewayProxyWrapper(
|
||||
{
|
||||
gatewayId: identityKubernetesAuth.gatewayId,
|
||||
targetHost: k8sHost, // note(daniel): must include the protocol (https|http)
|
||||
targetPort: k8sPort ? Number(k8sPort) : 443,
|
||||
caCert,
|
||||
reviewTokenThroughGateway: true
|
||||
},
|
||||
tokenReviewCallbackThroughGateway
|
||||
);
|
||||
} else if (identityKubernetesAuth.tokenReviewMode === IdentityKubernetesAuthTokenReviewMode.Api) {
|
||||
let { kubernetesHost } = identityKubernetesAuth;
|
||||
if (kubernetesHost.startsWith("https://") || kubernetesHost.startsWith("http://")) {
|
||||
kubernetesHost = new RE2("^https?:\\/\\/").replace(kubernetesHost, "");
|
||||
}
|
||||
|
||||
const [k8sHost, k8sPort] = kubernetesHost.split(":");
|
||||
|
||||
data = identityKubernetesAuth.gatewayId
|
||||
? await $gatewayProxyWrapper(
|
||||
{
|
||||
gatewayId: identityKubernetesAuth.gatewayId,
|
||||
targetHost: k8sHost,
|
||||
targetPort: k8sPort ? Number(k8sPort) : 443,
|
||||
reviewTokenThroughGateway: false
|
||||
},
|
||||
tokenReviewCallbackRaw
|
||||
)
|
||||
: await tokenReviewCallbackRaw();
|
||||
} else {
|
||||
throw new BadRequestError({
|
||||
message: `Invalid token review mode: ${identityKubernetesAuth.tokenReviewMode}`
|
||||
});
|
||||
}
|
||||
|
||||
if (!data) {
|
||||
throw new BadRequestError({
|
||||
message: "Failed to review token"
|
||||
});
|
||||
}
|
||||
|
||||
if ("error" in data.status)
|
||||
throw new UnauthorizedError({ message: data.status.error, name: "KubernetesTokenReviewError" });
|
||||
@@ -298,6 +418,7 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
kubernetesHost,
|
||||
caCert,
|
||||
tokenReviewerJwt,
|
||||
tokenReviewMode,
|
||||
allowedNamespaces,
|
||||
allowedNames,
|
||||
allowedAudience,
|
||||
@@ -384,6 +505,7 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
{
|
||||
identityId: identityMembershipOrg.identityId,
|
||||
kubernetesHost,
|
||||
tokenReviewMode,
|
||||
allowedNamespaces,
|
||||
allowedNames,
|
||||
allowedAudience,
|
||||
@@ -410,6 +532,7 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
kubernetesHost,
|
||||
caCert,
|
||||
tokenReviewerJwt,
|
||||
tokenReviewMode,
|
||||
allowedNamespaces,
|
||||
allowedNames,
|
||||
allowedAudience,
|
||||
@@ -492,6 +615,7 @@ export const identityKubernetesAuthServiceFactory = ({
|
||||
|
||||
const updateQuery: TIdentityKubernetesAuthsUpdate = {
|
||||
kubernetesHost,
|
||||
tokenReviewMode,
|
||||
allowedNamespaces,
|
||||
allowedNames,
|
||||
allowedAudience,
|
||||
|
||||
@@ -5,11 +5,17 @@ export type TLoginKubernetesAuthDTO = {
|
||||
jwt: string;
|
||||
};
|
||||
|
||||
export enum IdentityKubernetesAuthTokenReviewMode {
|
||||
Api = "api",
|
||||
Gateway = "gateway"
|
||||
}
|
||||
|
||||
export type TAttachKubernetesAuthDTO = {
|
||||
identityId: string;
|
||||
kubernetesHost: string;
|
||||
caCert: string;
|
||||
tokenReviewerJwt?: string;
|
||||
tokenReviewMode: IdentityKubernetesAuthTokenReviewMode;
|
||||
allowedNamespaces: string;
|
||||
allowedNames: string;
|
||||
allowedAudience: string;
|
||||
@@ -26,6 +32,7 @@ export type TUpdateKubernetesAuthDTO = {
|
||||
kubernetesHost?: string;
|
||||
caCert?: string;
|
||||
tokenReviewerJwt?: string | null;
|
||||
tokenReviewMode?: IdentityKubernetesAuthTokenReviewMode;
|
||||
allowedNamespaces?: string;
|
||||
allowedNames?: string;
|
||||
allowedAudience?: string;
|
||||
|
||||
Reference in New Issue
Block a user