diff --git a/README.md b/README.md index 3ec8140..4ff9cd3 100644 --- a/README.md +++ b/README.md @@ -354,6 +354,16 @@ If none of the three are set, Vault is skipped. Once any of them is set, the ser On success the loaded key names (never values) are logged. +#### Database credential rotation + +When Vault is configured, the server re-reads the Vault secret every `VAULT_REFRESH_INTERVAL_MS` (default `300000`, i.e. 5 minutes; `0` disables polling). If `DATABASE_URL` or `DATABASE_REPLICA_URL` changed, it reconnects without a restart: + +1. A new Prisma client is built from the new URL and connected **before** anything is swapped. If Vault is unreachable or the database rejects the new credentials, the error is logged, the existing connections stay in use, and the next refresh retries. +2. Once connected, the new client takes over all new queries. +3. The previous client is disconnected after `DATABASE_ROTATION_DRAIN_MS` (default `30000`) so in-flight queries and transactions can finish. Keep the old credentials valid in the database for at least that long after rotating them in Vault. + +Only the database URLs are applied at runtime; other secrets are still read once at startup. + ### OpenTelemetry tracing Tracing is enabled when the OpenTelemetry packages are installed and the following variables are set: diff --git a/src/common/secrets/vault-secrets.loader.ts b/src/common/secrets/vault-secrets.loader.ts index a5fbbab..02ea940 100644 --- a/src/common/secrets/vault-secrets.loader.ts +++ b/src/common/secrets/vault-secrets.loader.ts @@ -49,9 +49,58 @@ export class VaultSecretsError extends Error { export async function loadVaultSecrets(): Promise { const logger = new Logger('VaultSecrets'); + const secrets = await fetchVaultSecrets(); + if (!secrets) { + return; + } + const vaultPath = process.env['VAULT_KV_PATH']!; + + const loaded: string[] = []; + for (const [key, value] of Object.entries(secrets)) { + process.env[key] = value; + loaded.push(key); + } + + const missingRequired = parseCsv(process.env['VAULT_REQUIRED_KEYS']) + .filter((key) => !isNonEmptyString(process.env[key])); + if (missingRequired.length > 0) { + throw new VaultSecretsError( + `Required secret(s) missing after loading Vault path ${vaultPath}: ${missingRequired.join(', ')}. ` + + 'Add them to the Vault secret or set them as environment variables.', + ); + } + + if (loaded.length === 0) { + logger.warn(`Vault path ${vaultPath} returned no secrets`); + } else { + // Log key names only — never values. + logger.log(`Loaded ${loaded.length} secret(s) from Vault path ${vaultPath}: ${loaded.join(', ')}`); + } +} + +/** + * Whether Vault is configured at all (any of VAULT_ADDR / VAULT_TOKEN / + * VAULT_KV_PATH set). Partial configuration counts as configured so that + * fetchVaultSecrets() reports it. + */ +export function isVaultConfigured(): boolean { + return VAULT_CONFIG_VARS.some((name) => isNonEmptyString(process.env[name])); +} + +/** + * Fetch the KV v2 secret from Vault without touching process.env. Returns + * null when Vault is not configured; otherwise the secret's key/value pairs + * (null/undefined values dropped, everything else stringified). Throws a + * VaultSecretsError describing why they could not be read. + * + * Used both at startup (via loadVaultSecrets) and at runtime by + * DatabaseCredentialRotationService to pick up rotated credentials without a + * restart. + */ +export async function fetchVaultSecrets(): Promise | null> { const configured = VAULT_CONFIG_VARS.filter((name) => isNonEmptyString(process.env[name])); if (configured.length === 0) { - return; + return null; } if (configured.length < VAULT_CONFIG_VARS.length) { const missing = VAULT_CONFIG_VARS.filter((name) => !configured.includes(name)); @@ -117,28 +166,12 @@ export async function loadVaultSecrets(): Promise { ); } - const loaded: string[] = []; + const result: Record = {}; for (const [key, value] of Object.entries(secrets)) { if (value === undefined || value === null) { continue; } - process.env[key] = String(value); - loaded.push(key); - } - - const missingRequired = parseCsv(process.env['VAULT_REQUIRED_KEYS']) - .filter((key) => !isNonEmptyString(process.env[key])); - if (missingRequired.length > 0) { - throw new VaultSecretsError( - `Required secret(s) missing after loading Vault path ${vaultPath}: ${missingRequired.join(', ')}. ` + - 'Add them to the Vault secret or set them as environment variables.', - ); - } - - if (loaded.length === 0) { - logger.warn(`Vault path ${vaultPath} returned no secrets`); - } else { - // Log key names only — never values. - logger.log(`Loaded ${loaded.length} secret(s) from Vault path ${vaultPath}: ${loaded.join(', ')}`); + result[key] = String(value); } + return result; } diff --git a/src/config/env.d.ts b/src/config/env.d.ts index 3215a18..f0f8f4b 100644 --- a/src/config/env.d.ts +++ b/src/config/env.d.ts @@ -51,6 +51,11 @@ export interface EnvironmentVariables { VAULT_ADDR?: string; VAULT_TOKEN?: string; VAULT_KV_PATH?: string; + VAULT_TIMEOUT_MS?: string; + VAULT_REQUIRED_KEYS?: string; + // Database credential rotation (see src/prisma/database-credential-rotation.service.ts) + VAULT_REFRESH_INTERVAL_MS?: string; + DATABASE_ROTATION_DRAIN_MS?: string; OTEL_SERVICE_NAME?: string; OTEL_EXPORTER_OTLP_ENDPOINT?: string; OTEL_EXPORTER_OTLP_HEADERS?: string; diff --git a/src/prisma/database-credential-rotation.service.spec.ts b/src/prisma/database-credential-rotation.service.spec.ts new file mode 100644 index 0000000..3c620ce --- /dev/null +++ b/src/prisma/database-credential-rotation.service.spec.ts @@ -0,0 +1,128 @@ +import { Logger } from '@nestjs/common'; +import { DatabaseCredentialRotationService } from './database-credential-rotation.service'; +import { PrismaService } from './prisma.service'; + +describe('DatabaseCredentialRotationService', () => { + const originalEnv = process.env; + const OLD_URL = 'postgresql://old:old-pass@localhost:5432/db'; + const NEW_URL = 'postgresql://new:new-pass@localhost:5432/db'; + + let fetchMock: jest.Mock; + let prisma: { rotateCredentials: jest.Mock }; + let service: DatabaseCredentialRotationService; + + function vaultReturns(data: Record) { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + statusText: 'OK', + json: jest.fn().mockResolvedValue({ data: { data } }), + }); + } + + beforeEach(() => { + jest.useFakeTimers(); + process.env = { + ...originalEnv, + VAULT_ADDR: 'https://vault.example.com', + VAULT_TOKEN: 's.token', + VAULT_KV_PATH: 'secret/data/parashield', + DATABASE_URL: OLD_URL, + JWT_SECRET: 'startup-jwt', + }; + delete process.env.VAULT_REFRESH_INTERVAL_MS; + delete process.env.DATABASE_REPLICA_URL; + fetchMock = jest.fn(); + global.fetch = fetchMock as unknown as typeof fetch; + prisma = { rotateCredentials: jest.fn().mockResolvedValue(true) }; + service = new DatabaseCredentialRotationService(prisma as unknown as PrismaService); + jest.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + }); + + afterEach(() => { + service.onModuleDestroy(); + process.env = originalEnv; + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('applies rotated database credentials from Vault and reconnects', async () => { + vaultReturns({ DATABASE_URL: NEW_URL, JWT_SECRET: 'rotated-jwt' }); + + await expect(service.refresh()).resolves.toBe(true); + + expect(process.env.DATABASE_URL).toBe(NEW_URL); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + // Only database credentials are applied at runtime. + expect(process.env.JWT_SECRET).toBe('startup-jwt'); + }); + + it('does not reconnect when the credentials are unchanged', async () => { + vaultReturns({ DATABASE_URL: OLD_URL }); + + await expect(service.refresh()).resolves.toBe(false); + expect(prisma.rotateCredentials).not.toHaveBeenCalled(); + }); + + it('keeps current credentials when Vault is unreachable', async () => { + fetchMock.mockRejectedValue(new TypeError('fetch failed')); + + await expect(service.refresh()).resolves.toBe(false); + expect(process.env.DATABASE_URL).toBe(OLD_URL); + expect(prisma.rotateCredentials).not.toHaveBeenCalled(); + expect(Logger.prototype.warn).toHaveBeenCalledWith( + expect.stringContaining('keeping current credentials'), + ); + }); + + it('restores the previous credentials when reconnecting fails, so the next refresh retries', async () => { + vaultReturns({ DATABASE_URL: NEW_URL }); + prisma.rotateCredentials.mockRejectedValueOnce(new Error('P1000: Authentication failed')); + + await expect(service.refresh()).resolves.toBe(false); + expect(process.env.DATABASE_URL).toBe(OLD_URL); + + await expect(service.refresh()).resolves.toBe(true); + expect(process.env.DATABASE_URL).toBe(NEW_URL); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(2); + }); + + it('polls Vault on the configured interval', async () => { + process.env.VAULT_REFRESH_INTERVAL_MS = '60000'; + vaultReturns({ DATABASE_URL: NEW_URL }); + service.onModuleInit(); + + await jest.advanceTimersByTimeAsync(59_999); + expect(fetchMock).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(1); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + }); + + it('does not poll when Vault is not configured or polling is disabled', async () => { + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(1); + service.onModuleDestroy(); + + process.env.VAULT_REFRESH_INTERVAL_MS = '0'; + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(0); + + delete process.env.VAULT_REFRESH_INTERVAL_MS; + delete process.env.VAULT_ADDR; + delete process.env.VAULT_TOKEN; + delete process.env.VAULT_KV_PATH; + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(0); + }); + + it('shares one refresh between overlapping calls', async () => { + vaultReturns({ DATABASE_URL: NEW_URL }); + + await Promise.all([service.refresh(), service.refresh()]); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/prisma/database-credential-rotation.service.ts b/src/prisma/database-credential-rotation.service.ts new file mode 100644 index 0000000..bb5f440 --- /dev/null +++ b/src/prisma/database-credential-rotation.service.ts @@ -0,0 +1,107 @@ +import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { fetchVaultSecrets, isVaultConfigured } from '../common/secrets/vault-secrets.loader'; +import { PrismaService } from './prisma.service'; + +// Keys whose rotation PrismaService knows how to apply without a restart. +// Other secrets are read once at startup by their consumers, so they are +// deliberately not overwritten in process.env at runtime. +const DATABASE_CREDENTIAL_KEYS = ['DATABASE_URL', 'DATABASE_REPLICA_URL'] as const; + +const DEFAULT_REFRESH_INTERVAL_MS = 300_000; + +function getRefreshIntervalMs(): number { + const raw = process.env.VAULT_REFRESH_INTERVAL_MS; + if (raw === undefined || raw.trim() === '') return DEFAULT_REFRESH_INTERVAL_MS; + const parsed = Number(raw); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : DEFAULT_REFRESH_INTERVAL_MS; +} + +/** + * Periodically re-reads the Vault secret loaded at startup and, when the + * database credentials in it have changed, applies them to process.env and + * asks PrismaService to reconnect with them. + * + * Only active when Vault is configured. Runs every VAULT_REFRESH_INTERVAL_MS + * (default 5 minutes, 0 disables polling; refresh() can still be called). + * Failures — Vault unreachable, or the new credentials not yet accepted by the + * database — are logged and the current connections are kept, so a rotation + * hiccup never takes the API down; the next tick retries. + */ +@Injectable() +export class DatabaseCredentialRotationService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(DatabaseCredentialRotationService.name); + private timer: NodeJS.Timeout | null = null; + private inFlight: Promise | null = null; + + constructor(private readonly prisma: PrismaService) {} + + onModuleInit(): void { + if (!isVaultConfigured()) return; + + const intervalMs = getRefreshIntervalMs(); + if (intervalMs === 0) { + this.logger.log('Vault credential refresh polling disabled (VAULT_REFRESH_INTERVAL_MS=0)'); + return; + } + this.timer = setInterval(() => void this.refresh(), intervalMs); + this.timer.unref?.(); + this.logger.log(`Polling Vault for database credential rotation every ${intervalMs / 1000}s`); + } + + onModuleDestroy(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** + * Fetch the current secret from Vault and rotate database connections if + * the credentials changed. Overlapping calls share one refresh. Resolves + * true when connections were swapped; never rejects. + */ + refresh(): Promise { + this.inFlight ??= this.doRefresh().finally(() => { + this.inFlight = null; + }); + return this.inFlight; + } + + private async doRefresh(): Promise { + let secrets: Record | null; + try { + secrets = await fetchVaultSecrets(); + } catch (err) { + this.logger.warn( + `Vault credential refresh failed, keeping current credentials: ${(err as Error).message}`, + ); + return false; + } + if (!secrets) return false; + + const changed = DATABASE_CREDENTIAL_KEYS.filter( + (key) => secrets[key] !== undefined && secrets[key] !== process.env[key], + ); + if (changed.length === 0) return false; + + // Log key names only — never values. + this.logger.log(`Rotated credential(s) detected in Vault: ${changed.join(', ')}`); + const previous = Object.fromEntries(changed.map((key) => [key, process.env[key]])); + for (const key of changed) { + process.env[key] = secrets[key]; + } + + try { + return await this.prisma.rotateCredentials(); + } catch { + // PrismaService already logged why. Restore the old values so + // process.env keeps describing the connections actually in use and the + // next tick sees the change again and retries. + for (const key of changed) { + if (previous[key] === undefined) delete process.env[key]; + else process.env[key] = previous[key]; + } + return false; + } + } +} diff --git a/src/prisma/prisma.module.ts b/src/prisma/prisma.module.ts index 11883c5..9725758 100644 --- a/src/prisma/prisma.module.ts +++ b/src/prisma/prisma.module.ts @@ -1,9 +1,10 @@ import { Global, Module } from '@nestjs/common'; import { PrismaService } from './prisma.service'; +import { DatabaseCredentialRotationService } from './database-credential-rotation.service'; @Global() @Module({ - providers: [PrismaService], - exports: [PrismaService], + providers: [PrismaService, DatabaseCredentialRotationService], + exports: [PrismaService, DatabaseCredentialRotationService], }) export class PrismaModule {} diff --git a/src/prisma/prisma.service.spec.ts b/src/prisma/prisma.service.spec.ts index 6c16b64..b3231e1 100644 --- a/src/prisma/prisma.service.spec.ts +++ b/src/prisma/prisma.service.spec.ts @@ -1,3 +1,4 @@ +import { Logger } from '@nestjs/common'; import { PrismaService } from './prisma.service'; describe('PrismaService', () => { diff --git a/src/prisma/prisma.service.ts b/src/prisma/prisma.service.ts index b238a48..38092f4 100644 --- a/src/prisma/prisma.service.ts +++ b/src/prisma/prisma.service.ts @@ -134,19 +134,53 @@ export function sanitizeConnectionError(error: unknown, fallbackUrl?: string): E const DEFAULT_QUERY_CACHE_TTL_SECONDS = 30; +// After a credential rotation the previous clients are kept alive for this +// long so in-flight queries and interactive transactions can finish before +// their pools are closed. Override with DATABASE_ROTATION_DRAIN_MS. +const DEFAULT_ROTATION_DRAIN_MS = 30_000; + +function getRotationDrainMs(): number { + const parsed = parseInt(process.env.DATABASE_ROTATION_DRAIN_MS ?? '', 10); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : DEFAULT_ROTATION_DRAIN_MS; +} + interface CacheEntry { value: unknown; expiresAt: number; } +// Database credentials can rotate at runtime (see rotateCredentials), and a +// PrismaClient's datasource URL is fixed once constructed. PrismaService +// therefore wraps a swappable client instead of extending PrismaClient: the +// interface merge below keeps the full PrismaClient API on PrismaService, and +// the Proxy returned from the constructor forwards everything PrismaService +// doesn't define itself to whichever client is current. +// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging +export interface PrismaService extends PrismaClient {} + @Injectable() -export class PrismaService extends PrismaClient implements OnModuleInit, OnModuleDestroy { +// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging +export class PrismaService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(PrismaService.name); + // Primary client; replaced on credential rotation. + private client: PrismaClient; + // #439 — Read replica client. Instantiated only when DATABASE_REPLICA_URL // is set; otherwise reader falls back to the primary so callers need no // conditional logic and existing code paths are unchanged. - private readonly replicaClient: PrismaClient | null; + private replicaClient: PrismaClient | null; + + // Raw URLs the current clients were built from, compared against + // process.env to detect rotated credentials. + private primaryUrl: string | undefined; + private replicaUrl: string | undefined; + + // Serializes rotations so two refreshes never race on the swap. + private rotation: Promise = Promise.resolve(false); + + // Clients replaced by a rotation, waiting out the drain period. + private readonly retiring = new Map(); // In-process TTL cache for repeated read queries (e.g. product lookups, // config-style tables) that are hit frequently but change rarely. Callers @@ -154,19 +188,31 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul // writes and transactions are never affected. private readonly queryCache = new Map(); - public readonly poolConfig: PrismaPoolConfig; + public poolConfig: PrismaPoolConfig; constructor() { - super({ datasourceUrl: withConnectionPoolParams(process.env.DATABASE_URL) }); + this.primaryUrl = process.env.DATABASE_URL; + this.client = this.createClient(this.primaryUrl); + this.poolConfig = getEffectivePoolConfig(this.primaryUrl); - this.poolConfig = getEffectivePoolConfig(process.env.DATABASE_URL); + this.replicaUrl = process.env.DATABASE_REPLICA_URL; + this.replicaClient = this.replicaUrl ? this.createClient(this.replicaUrl) : null; - const replicaUrl = withConnectionPoolParams(process.env.DATABASE_REPLICA_URL); - this.replicaClient = replicaUrl - ? new PrismaClient({ datasourceUrl: replicaUrl }) - : null; + return new Proxy(this, { + get(target, prop, receiver) { + if (prop in target) { + return Reflect.get(target, prop, receiver); + } + const value = Reflect.get(target.client, prop); + return typeof value === 'function' ? value.bind(target.client) : value; + }, + }); } + /** Builds a client for `url` with the pool parameters applied. */ + protected createClient(url: string | undefined): PrismaClient { + return new PrismaClient({ datasourceUrl: withConnectionPoolParams(url) }); + } /** * Returns the read-replica PrismaClient when DATABASE_REPLICA_URL is @@ -183,6 +229,94 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul return this.replicaClient ?? this; } + /** + * Picks up rotated database credentials without a restart. + * + * Re-reads DATABASE_URL and DATABASE_REPLICA_URL from process.env; for each + * one that changed, a new client is built and connected *before* it is + * swapped in, so a bad or not-yet-active credential never replaces a + * working connection — the error is thrown and the current clients stay in + * place (the next rotation attempt retries). Once swapped, new queries use + * the new credentials, while the replaced clients are disconnected only + * after DATABASE_ROTATION_DRAIN_MS (default 30s) so in-flight queries and + * transactions can complete. + * + * Rotations are serialized. Resolves true when any client was swapped. + */ + rotateCredentials(): Promise { + this.rotation = this.rotation + .catch(() => false) + .then(() => this.swapClientsIfCredentialsChanged()); + return this.rotation; + } + + private async swapClientsIfCredentialsChanged(): Promise { + const primaryUrl = process.env.DATABASE_URL; + const replicaUrl = process.env.DATABASE_REPLICA_URL; + const primaryChanged = primaryUrl !== this.primaryUrl; + const replicaChanged = replicaUrl !== this.replicaUrl; + if (!primaryChanged && !replicaChanged) { + return false; + } + + const nextPrimary = primaryChanged ? this.createClient(primaryUrl) : null; + const nextReplica = replicaChanged && replicaUrl ? this.createClient(replicaUrl) : null; + try { + await nextPrimary?.$connect(); + await nextReplica?.$connect(); + } catch (err) { + await this.disconnectQuietly([nextPrimary, nextReplica]); + const errorMsg = err instanceof Error ? err.message : String(err); + this.logger.error( + `Database credential rotation failed, keeping existing connections: ${errorMsg}`, + ); + throw err; + } + + const retired: PrismaClient[] = []; + const rotated: string[] = []; + if (nextPrimary) { + retired.push(this.client); + this.client = nextPrimary; + this.primaryUrl = primaryUrl; + this.poolConfig = getEffectivePoolConfig(primaryUrl); + rotated.push('primary'); + } + if (replicaChanged) { + if (this.replicaClient) retired.push(this.replicaClient); + this.replicaClient = nextReplica; + this.replicaUrl = replicaUrl; + rotated.push('replica'); + } + + const drainMs = getRotationDrainMs(); + const timer = setTimeout(() => { + this.retiring.delete(timer); + void this.disconnectQuietly(retired); + }, drainMs); + timer.unref?.(); + this.retiring.set(timer, retired); + + this.logger.log( + `Database credentials rotated (${rotated.join(', ')}); ` + + `previous connections close in ${drainMs / 1000}s`, + ); + return true; + } + + private async disconnectQuietly(clients: Array): Promise { + await Promise.all(clients.map(async (client) => { + if (!client) return; + try { + await client.$disconnect(); + } catch (err) { + this.logger.warn( + `Failed to disconnect retired database client: ${err instanceof Error ? err.message : String(err)}`, + ); + } + })); + } + async onModuleInit() { // #696 — Handle SIGTERM / SIGINT signals gracefully process.once('SIGTERM', async () => { @@ -231,6 +365,11 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul } async onModuleDestroy() { + const retiring = [...this.retiring.entries()]; + this.retiring.clear(); + for (const [timer] of retiring) clearTimeout(timer); + await this.disconnectQuietly(retiring.flatMap(([, clients]) => clients)); + await this.$disconnect(); if (this.replicaClient) { await this.replicaClient.$disconnect(); @@ -296,8 +435,8 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul */ async getMetrics(): Promise { try { - if (typeof (this as any).$metrics?.json === 'function') { - return await (this as any).$metrics.json(); + if (typeof (this.client as any).$metrics?.json === 'function') { + return await (this.client as any).$metrics.json(); } } catch (err) { this.logger.warn(`Failed to collect Prisma metrics: ${err instanceof Error ? err.message : String(err)}`);