From 79f1e914ba95e16561fc1e7fc9f584147cf68648 Mon Sep 17 00:00:00 2001 From: ngoziimmaculatar-star Date: Sun, 18 Oct 2026 02:45:43 +0100 Subject: [PATCH] feat: resolve issues #518, #519, #520, and #521 - #518: Implement cache stampede protection with SingleFlight, distributed Redis locking, and background refresh - #519: Implement database field-level AES-256-GCM encryption at rest with blind indexing support - #520: Implement database connection pool monitoring, metrics, and Prometheus exporter - #521: Implement token refresh mechanism with Refresh Token Rotation and reuse detection --- src/auth/authService.js | 10 + src/auth/index.js | 3 + src/auth/tokenRefresh.js | 399 ++++++++++++++++++++++++++++++++ src/cache/index.js | 9 + src/cache/stampede.js | 315 +++++++++++++++++++++++++ src/config.js | 6 + src/database/encryption.js | 279 ++++++++++++++++++++++ src/database/index.js | 68 +++++- src/database/poolMonitor.js | 362 +++++++++++++++++++++++++++++ src/routes/metrics.js | 7 + src/services/dbCache.js | 54 ++--- test/auth.test.js | 50 +++- test/cacheStampede.test.js | 234 +++++++++++++++++++ test/databaseEncryption.test.js | 160 +++++++++++++ test/poolMonitor.test.js | 214 +++++++++++++++++ test/tokenRefresh.test.js | 207 +++++++++++++++++ 16 files changed, 2340 insertions(+), 37 deletions(-) create mode 100644 src/auth/tokenRefresh.js create mode 100644 src/cache/index.js create mode 100644 src/cache/stampede.js create mode 100644 src/database/encryption.js create mode 100644 src/database/poolMonitor.js create mode 100644 test/cacheStampede.test.js create mode 100644 test/databaseEncryption.test.js create mode 100644 test/poolMonitor.test.js create mode 100644 test/tokenRefresh.test.js diff --git a/src/auth/authService.js b/src/auth/authService.js index da378d3..531c832 100644 --- a/src/auth/authService.js +++ b/src/auth/authService.js @@ -160,9 +160,19 @@ async function changePassword(userId, currentPassword, newPassword, userStore) { return sanitizeUser(updated); } +const { + createTokenPair, + refreshTokens, + revokeRefreshToken, +} = require('./tokenRefresh'); + module.exports = { changePassword, loginUser, registerUser, sanitizeUser, + createTokenPair, + refreshTokens, + revokeRefreshToken, }; + diff --git a/src/auth/index.js b/src/auth/index.js index 403b117..98f0755 100644 --- a/src/auth/index.js +++ b/src/auth/index.js @@ -2,8 +2,11 @@ const password = require('./password'); const authService = require('./authService'); +const tokenRefresh = require('./tokenRefresh'); module.exports = { ...password, ...authService, + ...tokenRefresh, + tokenRefresh, }; diff --git a/src/auth/tokenRefresh.js b/src/auth/tokenRefresh.js new file mode 100644 index 0000000..c741285 --- /dev/null +++ b/src/auth/tokenRefresh.js @@ -0,0 +1,399 @@ +'use strict'; + +/** + * Token Refresh Mechanism (#521). + * + * Implements access token refresh with Refresh Token Rotation (RTR) + * and automatic reuse detection according to OAuth 2.0 Security Best Current Practices: + * 1. Generates short-lived JWT access tokens and cryptographically random refresh tokens. + * 2. Stores hashed refresh tokens (SHA-256) in cache/store with expiration. + * 3. Rotates refresh tokens on each refresh: the old token is invalidated and a fresh + * pair is returned. + * 4. Reuse Detection: If an already-rotated/invalidated token is used again, the entire + * token family for that user is immediately revoked to protect against token theft. + */ + +const crypto = require('crypto'); +const cache = require('../services/cache'); +const logger = require('../logger'); +const AppError = require('../errors/AppError'); +const { base64UrlEncode } = require('../middleware/jwtAuth'); + +// Defaults +const DEFAULT_ACCESS_TOKEN_TTL_SECONDS = 900; // 15 minutes +const DEFAULT_REFRESH_TOKEN_TTL_SECONDS = 7 * 24 * 3600; // 7 days +const REFRESH_PREFIX = 'rt:'; +const FAMILY_PREFIX = 'rt_fam:'; + +// In-memory fallback store when Redis is unavailable (e.g. unit tests) +const memoryStore = new Map(); +const memoryFamilyStore = new Map(); + +function hashToken(rawToken) { + return crypto.createHash('sha256').update(rawToken).digest('hex'); +} + +/** + * Sign a JWT using HMAC-SHA256. + * + * @param {Object} payload - Token claims + * @param {string} secret - Signing secret + * @param {Object} [options] + * @param {number} [options.expiresInSeconds=900] + * @param {string} [options.issuer='smartdrop-api'] + * @param {string} [options.audience='smartdrop-client'] + * @param {string} [options.algorithm='HS256'] + * @returns {string} JWT string + */ +function signJwt(payload, secret, options = {}) { + const { + expiresInSeconds = DEFAULT_ACCESS_TOKEN_TTL_SECONDS, + issuer = 'smartdrop-api', + audience = 'smartdrop-client', + algorithm = 'HS256', + } = options; + + if (!secret) { + throw new Error('signJwt: secret is required'); + } + + const header = { alg: algorithm, typ: 'JWT' }; + const now = Math.floor(Date.now() / 1000); + + const fullPayload = { + ...payload, + iat: now, + nbf: now, + exp: now + expiresInSeconds, + iss: issuer, + aud: audience, + }; + + const headerB64 = base64UrlEncode(Buffer.from(JSON.stringify(header))); + const payloadB64 = base64UrlEncode(Buffer.from(JSON.stringify(fullPayload))); + const data = `${headerB64}.${payloadB64}`; + + const hashAlg = algorithm === 'HS384' ? 'sha384' : algorithm === 'HS512' ? 'sha512' : 'sha256'; + const sig = crypto.createHmac(hashAlg, secret).update(data).digest(); + const sigB64 = base64UrlEncode(sig); + + return `${data}.${sigB64}`; +} + +/** + * Generate a new cryptographically random raw refresh token. + */ +function generateRawRefreshToken() { + return crypto.randomBytes(32).toString('hex'); +} + +/** + * Store a refresh token record. + */ +async function storeRefreshToken(record, ttlSeconds) { + const key = `${REFRESH_PREFIX}${record.tokenHash}`; + const data = JSON.stringify(record); + + try { + if (typeof cache.isConnected === 'function' && cache.isConnected()) { + const client = cache.getClient(); + if (client && typeof client.set === 'function') { + await client.set(key, data, 'EX', ttlSeconds); + // Track family tokens for reuse detection revocation + if (record.familyId) { + const familyKey = `${FAMILY_PREFIX}${record.familyId}`; + if (typeof client.sadd === 'function') { + await client.sadd(familyKey, record.tokenHash); + await client.expire(familyKey, ttlSeconds); + } + } + return; + } + } + } catch (err) { + logger.debug('Redis store failed, using in-memory token store', { error: err.message }); + } + + // Memory fallback + memoryStore.set(key, record); + if (record.familyId) { + if (!memoryFamilyStore.has(record.familyId)) { + memoryFamilyStore.set(record.familyId, new Set()); + } + memoryFamilyStore.get(record.familyId).add(record.tokenHash); + } +} + +/** + * Retrieve a refresh token record by its hash. + */ +async function getRefreshTokenRecord(tokenHash) { + const key = `${REFRESH_PREFIX}${tokenHash}`; + + try { + if (typeof cache.isConnected === 'function' && cache.isConnected()) { + const client = cache.getClient(); + if (client && typeof client.get === 'function') { + const val = await client.get(key); + if (val) { + return typeof val === 'string' ? JSON.parse(val) : val; + } + return null; + } + } + } catch (err) { + logger.debug('Redis get failed, checking memory store', { error: err.message }); + } + + return memoryStore.get(key) || null; +} + +/** + * Delete / invalidate a refresh token record. + */ +async function deleteRefreshToken(tokenHash) { + const key = `${REFRESH_PREFIX}${tokenHash}`; + + try { + if (typeof cache.isConnected === 'function' && cache.isConnected()) { + const client = cache.getClient(); + if (client && typeof client.del === 'function') { + await client.del(key); + } + } + } catch (err) { + logger.debug('Redis del failed', { error: err.message }); + } + + memoryStore.delete(key); +} + +/** + * Revoke all tokens belonging to a family (token reuse defense). + */ +async function revokeFamily(familyId) { + if (!familyId) return; + logger.warn('Revoking token family due to detected reuse', { familyId }); + + try { + if (typeof cache.isConnected === 'function' && cache.isConnected()) { + const client = cache.getClient(); + if (client && typeof client.smembers === 'function') { + const familyKey = `${FAMILY_PREFIX}${familyId}`; + const hashes = await client.smembers(familyKey); + if (Array.isArray(hashes)) { + for (const hash of hashes) { + await client.del(`${REFRESH_PREFIX}${hash}`); + } + } + await client.del(familyKey); + } + } + } catch (err) { + logger.debug('Redis family revocation failed', { error: err.message }); + } + + // Memory fallback + if (memoryFamilyStore.has(familyId)) { + for (const hash of memoryFamilyStore.get(familyId)) { + memoryStore.delete(`${REFRESH_PREFIX}${hash}`); + } + memoryFamilyStore.delete(familyId); + } +} + +/** + * Create a new access and refresh token pair for a user. + * + * @param {Object} user - User object (must have id) + * @param {Object} options + * @param {string} [options.jwtSecret] - Secret for signing JWT + * @param {number} [options.accessTokenTtlSeconds=900] + * @param {number} [options.refreshTokenTtlSeconds=604800] + * @param {string} [options.familyId] - Existing family ID if rotating + * @returns {Promise} Token pair and metadata + */ +async function createTokenPair(user, options = {}) { + if (!user || !user.id) { + throw new AppError('INTERNAL_ERROR', 'createTokenPair: user with id is required', 500); + } + + const { + jwtSecret = process.env.JWT_SECRET || 'smartdrop-dev-jwt-secret-do-not-use-in-production', + accessTokenTtlSeconds = DEFAULT_ACCESS_TOKEN_TTL_SECONDS, + refreshTokenTtlSeconds = DEFAULT_REFRESH_TOKEN_TTL_SECONDS, + familyId = crypto.randomUUID(), + scopes = user.scopes || ['default'], + } = options; + + // 1. Sign access token + const accessPayload = { + sub: user.id, + userId: user.id, + email: user.email, + username: user.username, + scopes, + }; + const accessToken = signJwt(accessPayload, jwtSecret, { + expiresInSeconds: accessTokenTtlSeconds, + }); + + // 2. Generate refresh token + const rawRefreshToken = generateRawRefreshToken(); + const tokenHash = hashToken(rawRefreshToken); + const now = Date.now(); + const expiresAt = now + refreshTokenTtlSeconds * 1000; + + const record = { + tokenId: crypto.randomUUID(), + tokenHash, + userId: user.id, + user: { + id: user.id, + email: user.email, + username: user.username, + scopes, + }, + familyId, + createdAt: now, + expiresAt, + used: false, + }; + + await storeRefreshToken(record, refreshTokenTtlSeconds); + + return { + tokenType: 'Bearer', + accessToken, + refreshToken: rawRefreshToken, + expiresIn: accessTokenTtlSeconds, + refreshExpiresIn: refreshTokenTtlSeconds, + user: { + id: user.id, + email: user.email, + username: user.username, + }, + }; +} + +/** + * Refresh an access token using a valid refresh token. + * Performs Refresh Token Rotation (RTR) and detects reuse. + * + * @param {string} rawRefreshToken + * @param {Object} [options] + * @returns {Promise} New token pair + */ +async function refreshTokens(rawRefreshToken, options = {}) { + if (!rawRefreshToken || typeof rawRefreshToken !== 'string') { + throw new AppError('UNAUTHORIZED', 'Refresh token is required', 401); + } + + const tokenHash = hashToken(rawRefreshToken.trim()); + const record = await getRefreshTokenRecord(tokenHash); + + // If token record doesn't exist + if (!record) { + throw new AppError('UNAUTHORIZED', 'Invalid or expired refresh token', 401); + } + + // Reuse Detection: If this token was already marked as used, compromise detected! + if (record.used) { + await revokeFamily(record.familyId); + throw new AppError( + 'UNAUTHORIZED', + 'Invalid refresh token: token reuse detected. All sessions in this chain have been terminated.', + 401 + ); + } + + // Expiration check + if (Date.now() > record.expiresAt) { + await deleteRefreshToken(tokenHash); + throw new AppError('UNAUTHORIZED', 'Refresh token has expired', 401); + } + + // Invalidate old token immediately to prevent reuse + record.used = true; + await storeRefreshToken(record, 60); // Keep for a short grace window to detect reuse + + // Issue new token pair preserving the familyId + const newPair = await createTokenPair(record.user, { + familyId: record.familyId, + ...options, + }); + + return newPair; +} + +/** + * Revoke a single refresh token (e.g. on user logout). + * + * @param {string} rawRefreshToken + */ +async function revokeRefreshToken(rawRefreshToken) { + if (!rawRefreshToken) return false; + const tokenHash = hashToken(rawRefreshToken.trim()); + const record = await getRefreshTokenRecord(tokenHash); + if (!record) return false; + + await deleteRefreshToken(tokenHash); + return true; +} + +/** + * Express middleware handler for token refresh endpoint: POST /auth/refresh + */ +function createRefreshHandler(options = {}) { + return async (req, res, next) => { + try { + const rawToken = + req.body?.refreshToken || + req.body?.refresh_token || + req.get('x-refresh-token'); + + if (!rawToken) { + return next(new AppError('VALIDATION_ERROR', 'Missing refresh token in request body', 400)); + } + + const result = await refreshTokens(rawToken, options); + return res.status(200).json(result); + } catch (err) { + return next(err); + } + }; +} + +/** + * Express middleware handler for token revocation: POST /auth/revoke + */ +function createRevokeHandler() { + return async (req, res, next) => { + try { + const rawToken = + req.body?.refreshToken || + req.body?.refresh_token || + req.get('x-refresh-token'); + + if (rawToken) { + await revokeRefreshToken(rawToken); + } + return res.status(200).json({ revoked: true }); + } catch (err) { + return next(err); + } + }; +} + +module.exports = { + signJwt, + createTokenPair, + refreshTokens, + revokeRefreshToken, + revokeFamily, + hashToken, + createRefreshHandler, + createRevokeHandler, + DEFAULT_ACCESS_TOKEN_TTL_SECONDS, + DEFAULT_REFRESH_TOKEN_TTL_SECONDS, +}; diff --git a/src/cache/index.js b/src/cache/index.js new file mode 100644 index 0000000..4d13513 --- /dev/null +++ b/src/cache/index.js @@ -0,0 +1,9 @@ +'use strict'; + +const stampede = require('./stampede'); +const cacheService = require('../services/cache'); + +module.exports = { + ...stampede, + cache: cacheService, +}; diff --git a/src/cache/stampede.js b/src/cache/stampede.js new file mode 100644 index 0000000..5186e0a --- /dev/null +++ b/src/cache/stampede.js @@ -0,0 +1,315 @@ +'use strict'; + +const crypto = require('crypto'); +const cache = require('../services/cache'); +const logger = require('../logger'); + +/** + * Singleflight: in-process promise coalescing for concurrent calls. + * Ensures that for any given key, only one promise is actively executing + * at any given time within this process. + */ +class SingleFlight { + constructor() { + this._inFlight = new Map(); + } + + /** + * Execute fn(key) or share an existing in-flight execution for key. + * + * @param {string} key + * @param {Function} fn + * @returns {Promise<*>} + */ + async do(key, fn) { + if (this._inFlight.has(key)) { + return this._inFlight.get(key); + } + + const promise = (async () => { + try { + return await fn(); + } finally { + this._inFlight.delete(key); + } + })(); + + this._inFlight.set(key, promise); + return promise; + } + + /** + * Check if a key is currently in-flight. + * @param {string} key + * @returns {boolean} + */ + has(key) { + return this._inFlight.has(key); + } + + /** + * Number of currently in-flight keys. + */ + get size() { + return this._inFlight.size; + } +} + +// Global process-level singleflight instance +const processSingleFlight = new SingleFlight(); + +/** + * Acquire a distributed Redis lock. + * + * @param {string} key - Cache key + * @param {Object} options + * @param {number} [options.lockTtlMs=5000] - Lock expiration in milliseconds + * @returns {Promise} Lock token if acquired, null otherwise + */ +async function acquireLock(key, options = {}) { + const { lockTtlMs = 5000 } = options; + const lockKey = `lock:${key}`; + const token = crypto.randomUUID(); + + try { + const client = cache.getClient(); + if (!client || typeof client.set !== 'function') { + return token; // fallback in-process + } + + // SET key token NX PX lockTtlMs + const result = await client.set(lockKey, token, 'PX', lockTtlMs, 'NX'); + if (result === 'OK' || result === true) { + return token; + } + return null; + } catch (err) { + logger.debug('Distributed lock acquire failed or Redis unavailable', { + key, + error: err.message, + }); + // In case of Redis error, fallback gracefully by returning a token + return token; + } +} + +/** + * Safely release a distributed Redis lock using token validation. + * + * @param {string} key + * @param {string} token + * @returns {Promise} + */ +async function releaseLock(key, token) { + if (!token) return false; + const lockKey = `lock:${key}`; + + try { + const client = cache.getClient(); + if (!client) return true; + + // Use Lua script to ensure atomic compare-and-delete + if (typeof client.eval === 'function') { + const luaScript = ` + if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) + else + return 0 + end + `; + const res = await client.eval(luaScript, 1, lockKey, token); + return res === 1; + } else if (typeof client.get === 'function' && typeof client.del === 'function') { + const current = await client.get(lockKey); + if (current === token) { + await client.del(lockKey); + return true; + } + } + return false; + } catch (err) { + logger.debug('Distributed lock release failed', { key, error: err.message }); + return false; + } +} + +/** + * Sleep helper for polling. + */ +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * Cache stampede protection wrapper. + * Combines singleflight coalescing, distributed locking, and optional background refresh. + * + * @param {string} key - Cache key + * @param {Function} fetcher - Async function to fetch fresh data + * @param {Object} options + * @param {number} [options.ttlSeconds=60] - Cache TTL in seconds + * @param {number} [options.lockTtlMs=5000] - Lock TTL in milliseconds + * @param {number} [options.lockTimeoutMs=3000] - Max wait time for lock in milliseconds + * @param {number} [options.retryIntervalMs=50] - Polling interval while waiting for lock + * @param {boolean} [options.backgroundRefresh=false] - Whether to serve stale value and refresh in background + * @param {boolean} [options.skipCache=false] - Bypass cache + * @returns {Promise<*>} + */ +async function fetchWithLock(key, fetcher, options = {}) { + const { + ttlSeconds = 60, + lockTtlMs = 5000, + lockTimeoutMs = 3000, + retryIntervalMs = 50, + backgroundRefresh = false, + skipCache = false, + } = options; + + if (typeof fetcher !== 'function') { + throw new Error('fetchWithLock: fetcher must be a function'); + } + + // 1. If skipCache, execute directly without locking or cache lookup + if (skipCache) { + return fetcher(); + } + + // 2. Try cache first + try { + const cached = await cache.get(key); + if (cached !== null && cached !== undefined) { + const parsed = typeof cached === 'string' ? JSON.parse(cached) : cached; + + // Check if wrapper contains metadata for background refresh + if (parsed && typeof parsed === 'object' && parsed.__stampede_meta) { + const { value, expiresAt } = parsed; + const now = Date.now(); + // If expired or near expiration and backgroundRefresh is enabled + if (now >= expiresAt && backgroundRefresh) { + // Trigger background refresh asynchronously without blocking caller + triggerBackgroundRefresh(key, fetcher, { ttlSeconds, lockTtlMs }); + return value; + } + return value; + } + + return parsed; + } + } catch (err) { + logger.debug('Cache get failed during stampede check', { key, error: err.message }); + } + + // 3. Use SingleFlight to coalesce simultaneous in-process calls + return processSingleFlight.do(key, async () => { + // Re-check cache in case another in-flight promise just populated it + try { + const cached = await cache.get(key); + if (cached !== null && cached !== undefined) { + const parsed = typeof cached === 'string' ? JSON.parse(cached) : cached; + if (parsed && typeof parsed === 'object' && parsed.__stampede_meta) { + return parsed.value; + } + return parsed; + } + } catch { + // Ignore and proceed + } + + // 4. Try to acquire distributed lock + const startTime = Date.now(); + let lockToken = await acquireLock(key, { lockTtlMs }); + + // If lock not acquired, wait and poll cache + while (!lockToken && Date.now() - startTime < lockTimeoutMs) { + await sleep(retryIntervalMs); + + // Check if the lock holder has populated the cache + try { + const cached = await cache.get(key); + if (cached !== null && cached !== undefined) { + const parsed = typeof cached === 'string' ? JSON.parse(cached) : cached; + if (parsed && typeof parsed === 'object' && parsed.__stampede_meta) { + return parsed.value; + } + return parsed; + } + } catch { + // Ignore and retry + } + + // Try acquiring lock again + lockToken = await acquireLock(key, { lockTtlMs }); + } + + // 5. If we have the lock (or timed out waiting, in which case we execute as fallback) + try { + const freshData = await fetcher(); + + // Store in cache + try { + const payloadToStore = backgroundRefresh + ? { + __stampede_meta: true, + value: freshData, + expiresAt: Date.now() + ttlSeconds * 1000, + } + : freshData; + + // Give a generous TTL so background refresh has time to serve stale data + const storeTtl = backgroundRefresh ? ttlSeconds * 2 : ttlSeconds; + await cache.set(key, payloadToStore, storeTtl); + } catch (cacheErr) { + logger.warn('Failed to populate cache after stampede fetch', { + key, + error: cacheErr.message, + }); + } + + return freshData; + } finally { + if (lockToken) { + await releaseLock(key, lockToken); + } + } + }); +} + +/** + * Trigger an asynchronous background refresh under lock. + */ +function triggerBackgroundRefresh(key, fetcher, options = {}) { + const { ttlSeconds = 60, lockTtlMs = 5000 } = options; + + // Fire and forget + setImmediate(async () => { + let lockToken = null; + try { + lockToken = await acquireLock(key, { lockTtlMs }); + if (!lockToken) return; // Another worker is already refreshing + + const freshData = await fetcher(); + const payload = { + __stampede_meta: true, + value: freshData, + expiresAt: Date.now() + ttlSeconds * 1000, + }; + await cache.set(key, payload, ttlSeconds * 2); + logger.debug('Background cache refresh completed', { key }); + } catch (err) { + logger.warn('Background cache refresh failed', { key, error: err.message }); + } finally { + if (lockToken) { + await releaseLock(key, lockToken); + } + } + }); +} + +module.exports = { + SingleFlight, + processSingleFlight, + acquireLock, + releaseLock, + fetchWithLock, + triggerBackgroundRefresh, +}; diff --git a/src/config.js b/src/config.js index c21a332..1c2fe65 100644 --- a/src/config.js +++ b/src/config.js @@ -85,6 +85,7 @@ const env = cleanEnv(rawEnv, { // hex or base64 encoded. Empty in dev — webhookEncryption falls back to an // insecure fixed key and logs a warning; production must set this. WEBHOOK_SECRET_ENCRYPTION_KEY: str({ default: '' }), + DB_ENCRYPTION_KEY: str({ default: '' }), AIRDROP_CSV_MAX_BYTES: positiveInteger({ default: 5 * 1024 * 1024 }), AIRDROP_JSON_MAX_BYTES: positiveInteger({ default: 2 * 1024 * 1024 }), AIRDROP_RATELIMIT_WINDOW: positiveInteger({ default: 60 }), @@ -237,6 +238,8 @@ const config = module.exports = { slowRequestThresholdMs: env.SLOW_REQUEST_THRESHOLD_MS, routeTimeoutMs: env.ROUTE_TIMEOUT_MS, webhookSecretEncryptionKey: env.WEBHOOK_SECRET_ENCRYPTION_KEY, + databaseEncryptionKey: env.DB_ENCRYPTION_KEY, + dbEncryptionKey: env.DB_ENCRYPTION_KEY, // How long browsers may cache a CORS preflight response (issue #344). // Exposed as Access-Control-Max-Age on every successful OPTIONS preflight. corsMaxAgeSeconds: env.CORS_MAX_AGE_SECONDS, @@ -311,6 +314,7 @@ function reload() { LEASE_RENEW_INTERVAL_MS: positiveInteger({ default: 5000 }), ADMIN_API_KEY: str({ default: '' }), WEBHOOK_SECRET_ENCRYPTION_KEY: str({ default: '' }), + DB_ENCRYPTION_KEY: str({ default: '' }), AIRDROP_CSV_MAX_BYTES: positiveInteger({ default: 5 * 1024 * 1024 }), AIRDROP_JSON_MAX_BYTES: positiveInteger({ default: 2 * 1024 * 1024 }), AIRDROP_RATELIMIT_WINDOW: positiveInteger({ default: 60 }), @@ -375,6 +379,8 @@ function reload() { slowRequestThresholdMs: reloaded.SLOW_REQUEST_THRESHOLD_MS, routeTimeoutMs: reloaded.ROUTE_TIMEOUT_MS, webhookSecretEncryptionKey: reloaded.WEBHOOK_SECRET_ENCRYPTION_KEY, + databaseEncryptionKey: reloaded.DB_ENCRYPTION_KEY, + dbEncryptionKey: reloaded.DB_ENCRYPTION_KEY, corsMaxAgeSeconds: reloaded.CORS_MAX_AGE_SECONDS, watchedAssets: reloadedWatchedAssets, webhooks: { diff --git a/src/database/encryption.js b/src/database/encryption.js new file mode 100644 index 0000000..48b5e49 --- /dev/null +++ b/src/database/encryption.js @@ -0,0 +1,279 @@ +'use strict'; + +/** + * Database Encryption at Rest (#519). + * + * Implements authenticated field-level AES-256-GCM encryption for sensitive data + * stored in the database (e.g. credentials, tokens, secrets, PII). + * + * Encrypted values are packed as: + * enc:v1::: + * + * Features: + * - Deterministic prefix allows detecting already-encrypted values (isEncrypted). + * - Safe decryption transparently leaves legacy plaintext untouched (migration-friendly). + * - Authenticated cipher (GCM) guarantees tampering detection via 16-byte authTag. + * - Blind indexing (HMAC-SHA256) enables exact-match lookups without exposing plaintext. + */ + +const crypto = require('crypto'); +const config = require('../config'); +const logger = require('../logger'); + +const ALGORITHM = 'aes-256-gcm'; +const IV_BYTES = 12; // Standard 96-bit IV for GCM +const PREFIX = 'enc:v1:'; +const DEV_FALLBACK_KEY_MATERIAL = 'smartdrop-dev-db-encryption-key-do-not-use-in-production'; + +let warnedAboutDevKey = false; +let cachedDerivedKey = null; + +/** + * Derive 32-byte AES-256 encryption key from configuration. + * + * @param {string} [customKeyMaterial] - Optional explicit key override + * @returns {Buffer} 32-byte key buffer + */ +function deriveKey(customKeyMaterial) { + if (customKeyMaterial) { + return crypto.createHash('sha256').update(customKeyMaterial).digest(); + } + + if (cachedDerivedKey) return cachedDerivedKey; + + const configured = + config.databaseEncryptionKey || + config.dbEncryptionKey || + process.env.DB_ENCRYPTION_KEY; + + if (!configured) { + if (!warnedAboutDevKey && process.env.NODE_ENV !== 'test') { + logger.warn( + 'DB_ENCRYPTION_KEY is not set — database encryption is using an insecure fixed development key. ' + + 'Set DB_ENCRYPTION_KEY in production to protect sensitive data at rest.' + ); + warnedAboutDevKey = true; + } + cachedDerivedKey = crypto.createHash('sha256').update(DEV_FALLBACK_KEY_MATERIAL).digest(); + } else { + cachedDerivedKey = crypto.createHash('sha256').update(configured).digest(); + } + + return cachedDerivedKey; +} + +/** + * Check if a value is already encrypted with our format. + * + * @param {*} value + * @returns {boolean} + */ +function isEncrypted(value) { + return typeof value === 'string' && value.startsWith(PREFIX); +} + +/** + * Encrypt a plaintext value. + * + * @param {*} value - Plaintext string, number, boolean, or object + * @param {Object} [options] + * @param {string} [options.key] - Optional custom key + * @returns {string} Encrypted packed string + */ +function encrypt(value, options = {}) { + if (value === null || value === undefined) { + return value; + } + + // Already encrypted? + if (isEncrypted(value)) { + return value; + } + + // Serialize objects/arrays/numbers/booleans if needed + let plaintext; + if (typeof value === 'object') { + plaintext = JSON.stringify(value); + } else { + plaintext = String(value); + } + + const key = deriveKey(options.key); + const iv = crypto.randomBytes(IV_BYTES); + const cipher = crypto.createCipheriv(ALGORITHM, key, iv); + + const ciphertext = Buffer.concat([ + cipher.update(plaintext, 'utf8'), + cipher.final(), + ]); + const authTag = cipher.getAuthTag(); + + return `${PREFIX}${iv.toString('base64')}:${ciphertext.toString('base64')}:${authTag.toString('base64')}`; +} + +/** + * Decrypt an encrypted value. If the value is not encrypted (e.g. legacy row), + * returns it as-is. + * + * @param {*} packed - Encrypted packed string or legacy value + * @param {Object} [options] + * @param {string} [options.key] - Optional custom key + * @param {boolean} [options.parseJson=false] - Attempt to JSON parse result + * @returns {*} Decrypted plaintext string (or parsed object) + */ +function decrypt(packed, options = {}) { + if (packed === null || packed === undefined) { + return packed; + } + + if (typeof packed !== 'string' || !packed.startsWith(PREFIX)) { + return packed; // Legacy unencrypted value + } + + const payload = packed.slice(PREFIX.length); + const parts = payload.split(':'); + if (parts.length !== 3) { + throw new Error('Malformed encrypted database value: expected 3 segments (iv:ciphertext:authTag)'); + } + + const [ivB64, ciphertextB64, authTagB64] = parts; + const key = deriveKey(options.key); + const iv = Buffer.from(ivB64, 'base64'); + const ciphertext = Buffer.from(ciphertextB64, 'base64'); + const authTag = Buffer.from(authTagB64, 'base64'); + + const decipher = crypto.createDecipheriv(ALGORITHM, key, iv); + decipher.setAuthTag(authTag); + + const decryptedBuf = Buffer.concat([ + decipher.update(ciphertext), + decipher.final(), + ]); + const decryptedStr = decryptedBuf.toString('utf8'); + + if (options.parseJson) { + try { + return JSON.parse(decryptedStr); + } catch { + return decryptedStr; + } + } + + return decryptedStr; +} + +/** + * Encrypt sensitive fields on a database record/row. + * + * @param {Object} row - Database row + * @param {string[]} fields - Field names to encrypt + * @param {Object} [options] + * @returns {Object} Clone of row with encrypted fields + */ +function encryptRecord(row, fields = [], options = {}) { + if (!row || typeof row !== 'object') return row; + const result = { ...row }; + + for (const field of fields) { + if (Object.prototype.hasOwnProperty.call(result, field) && result[field] !== null && result[field] !== undefined) { + result[field] = encrypt(result[field], options); + } + } + + return result; +} + +/** + * Decrypt sensitive fields on a database record/row. + * + * @param {Object} row - Database row + * @param {string[]} [fields] - Field names to decrypt (if omitted, decrypts all fields starting with PREFIX) + * @param {Object} [options] + * @returns {Object} Clone of row with decrypted fields + */ +function decryptRecord(row, fields, options = {}) { + if (!row || typeof row !== 'object') return row; + const result = { ...row }; + + if (Array.isArray(fields) && fields.length > 0) { + for (const field of fields) { + if (Object.prototype.hasOwnProperty.call(result, field) && result[field] !== null && result[field] !== undefined) { + result[field] = decrypt(result[field], options); + } + } + } else { + // Auto-detect encrypted fields + for (const key of Object.keys(result)) { + if (isEncrypted(result[key])) { + result[key] = decrypt(result[key], options); + } + } + } + + return result; +} + +/** + * Encrypt an array of database rows. + * + * @param {Object[]} rows + * @param {string[]} fields + * @param {Object} [options] + * @returns {Object[]} + */ +function encryptRows(rows, fields = [], options = {}) { + if (!Array.isArray(rows)) return rows; + return rows.map((r) => encryptRecord(r, fields, options)); +} + +/** + * Decrypt an array of database rows. + * + * @param {Object[]} rows + * @param {string[]} [fields] + * @param {Object} [options] + * @returns {Object[]} + */ +function decryptRows(rows, fields, options = {}) { + if (!Array.isArray(rows)) return rows; + return rows.map((r) => decryptRecord(r, fields, options)); +} + +/** + * Create a blind index HMAC-SHA256 for searching encrypted fields. + * + * @param {string} value - Plaintext value to hash for exact lookup + * @param {string} [salt=''] - Optional salt or domain separator + * @returns {string} 64-character hex digest + */ +function hashBlindIndex(value, salt = '') { + if (value === null || value === undefined) return null; + const key = deriveKey(); + const normalized = String(value).trim().toLowerCase(); + return crypto.createHmac('sha256', key).update(`${salt}:${normalized}`).digest('hex'); +} + +/** + * Reset cached key (useful for testing with different keys). + */ +function _resetCachedKey() { + cachedDerivedKey = null; + warnedAboutDevKey = false; +} + +module.exports = { + encrypt, + decrypt, + isEncrypted, + encryptRecord, + decryptRecord, + encryptRow: encryptRecord, + decryptRow: decryptRecord, + encryptFields: encryptRecord, + decryptFields: decryptRecord, + encryptRows, + decryptRows, + hashBlindIndex, + deriveKey, + _resetCachedKey, +}; diff --git a/src/database/index.js b/src/database/index.js index 5bf80ca..f48cd65 100644 --- a/src/database/index.js +++ b/src/database/index.js @@ -2,7 +2,14 @@ const knexFactory = require('knex'); const config = require('../config'); -const { withTransaction } = require('./transaction'); +const { + withTransaction, + sortIds, + lockRows, + isDeadlockOrSerializationError, +} = require('./transaction'); +const encryption = require('./encryption'); +const { PoolMonitor, defaultPoolMonitor } = require('./poolMonitor'); const logger = require('../logger'); let db = null; @@ -21,6 +28,7 @@ function getDb(overrideConfig = {}) { pool: overrideConfig.pool || { min: 2, max: 10 }, ...overrideConfig, }); + defaultPoolMonitor.attach(db); } return db; } @@ -32,6 +40,11 @@ function getDb(overrideConfig = {}) { */ function setDb(instance) { db = instance; + if (db) { + defaultPoolMonitor.attach(db); + } else { + defaultPoolMonitor.detach(); + } } /** @@ -39,6 +52,7 @@ function setDb(instance) { */ async function closeDb() { if (db) { + defaultPoolMonitor.detach(); await db.destroy(); db = null; } @@ -65,23 +79,42 @@ async function withConnection(callback) { } /** - * Get current pool statistics for monitoring connection usage (#488). + * Get current pool statistics for monitoring connection usage (#488, #520). * - * @returns {Object} Pool stats + * @returns {Object} Pool stats with utilization metrics */ function getPoolStats() { const currentDb = getDb(); const pool = currentDb.client.pool; + const snapshot = defaultPoolMonitor.getSnapshot() || {}; return { - min: pool.min, - max: pool.max, - used: pool.numUsed(), - free: pool.numFree(), - pendingAcquires: pool.numPendingAcquires(), - pendingCreates: pool.numPendingCreates(), + min: pool ? pool.min : 0, + max: pool ? pool.max : 10, + used: pool && typeof pool.numUsed === 'function' ? pool.numUsed() : (snapshot.used ?? 0), + free: pool && typeof pool.numFree === 'function' ? pool.numFree() : (snapshot.free ?? 0), + pendingAcquires: + pool && typeof pool.numPendingAcquires === 'function' + ? pool.numPendingAcquires() + : (snapshot.pendingAcquires ?? 0), + pendingCreates: + pool && typeof pool.numPendingCreates === 'function' + ? pool.numPendingCreates() + : (snapshot.pendingCreates ?? 0), + utilization: snapshot.utilization ?? 0, + saturation: snapshot.saturation ?? 0, + status: snapshot.status ?? 'healthy', }; } +/** + * Get full connection pool metrics, including counters and latency timing (#520). + * + * @returns {Object} + */ +function getPoolMetrics() { + return defaultPoolMonitor.getMetrics(); +} + /** * Convenience wrapper for `withTransaction(getDb(), callback, options)` * @@ -100,9 +133,26 @@ module.exports = { closeDb, withConnection, getPoolStats, + getPoolMetrics, + PoolMonitor, + defaultPoolMonitor, withTransaction, transaction, sortIds, lockRows, isDeadlockOrSerializationError, + // Encryption at rest (#519) + encryption, + encrypt: encryption.encrypt, + decrypt: encryption.decrypt, + isEncrypted: encryption.isEncrypted, + encryptRecord: encryption.encryptRecord, + decryptRecord: encryption.decryptRecord, + encryptRow: encryption.encryptRow, + decryptRow: encryption.decryptRow, + encryptFields: encryption.encryptFields, + decryptFields: encryption.decryptFields, + encryptRows: encryption.encryptRows, + decryptRows: encryption.decryptRows, + hashBlindIndex: encryption.hashBlindIndex, }; diff --git a/src/database/poolMonitor.js b/src/database/poolMonitor.js new file mode 100644 index 0000000..74e0771 --- /dev/null +++ b/src/database/poolMonitor.js @@ -0,0 +1,362 @@ +'use strict'; + +/** + * Database Connection Pool Monitoring (#520). + * + * Provides real-time metrics, event tracking, utilization analysis, and + * Prometheus formatting for the Knex (Tarn.js) database connection pool. + */ + +const logger = require('../logger'); + +class PoolMonitor { + constructor(options = {}) { + this.warningThreshold = options.warningThreshold ?? 0.8; // 80% utilization + this.criticalThreshold = options.criticalThreshold ?? 0.95; // 95% utilization + this.onWarning = options.onWarning || null; + this.onCritical = options.onCritical || null; + + this.reset(); + this._pool = null; + this._listeners = new Map(); + this._pendingAcquireTimers = new Map(); + } + + /** + * Reset all counters and metrics. + */ + reset() { + this.counters = { + acquireRequests: 0, + acquireSuccesses: 0, + acquireFailures: 0, + releases: 0, + createRequests: 0, + createSuccesses: 0, + createFailures: 0, + destroyRequests: 0, + destroySuccesses: 0, + destroyFailures: 0, + timeouts: 0, + }; + + this.timing = { + lastAcquireDurationMs: 0, + maxAcquireDurationMs: 0, + totalAcquireDurationMs: 0, + acquireSamplesCount: 0, + }; + + this.lastFailure = null; + this.lastWarningAt = null; + this.lastCriticalAt = null; + } + + /** + * Attach monitor to a Knex instance or Tarn pool. + * + * @param {Object} knexOrPool - Knex database instance or Tarn pool + * @returns {PoolMonitor} this + */ + attach(knexOrPool) { + if (!knexOrPool) return this; + + const pool = knexOrPool.client ? knexOrPool.client.pool : knexOrPool; + if (!pool || typeof pool.on !== 'function') { + // Mock pool or pool without event emitter + this._pool = pool; + return this; + } + + if (this._pool === pool) { + return this; // Already attached to this pool + } + + this.detach(); + this._pool = pool; + + const register = (event, handler) => { + pool.on(event, handler); + this._listeners.set(event, handler); + }; + + register('acquireRequest', (eventId) => { + this.counters.acquireRequests++; + this._pendingAcquireTimers.set(eventId, Date.now()); + }); + + register('acquireSuccess', (eventId, resource) => { + this.counters.acquireSuccesses++; + const startTime = this._pendingAcquireTimers.get(eventId); + if (startTime) { + const duration = Date.now() - startTime; + this._recordAcquireDuration(duration); + this._pendingAcquireTimers.delete(eventId); + } + this._checkHealth(); + }); + + register('acquireFail', (eventId, err) => { + this.counters.acquireFailures++; + this.lastFailure = { + type: 'acquireFail', + error: err ? err.message : 'Unknown acquire failure', + timestamp: new Date().toISOString(), + }; + this._pendingAcquireTimers.delete(eventId); + logger.warn('Database pool acquire failed', { error: err ? err.message : null }); + }); + + register('release', (resource) => { + this.counters.releases++; + this._checkHealth(); + }); + + register('createRequest', (eventId) => { + this.counters.createRequests++; + }); + + register('createSuccess', (eventId, resource) => { + this.counters.createSuccesses++; + }); + + register('createFail', (eventId, err) => { + this.counters.createFailures++; + this.lastFailure = { + type: 'createFail', + error: err ? err.message : 'Unknown create failure', + timestamp: new Date().toISOString(), + }; + logger.warn('Database pool connection creation failed', { error: err ? err.message : null }); + }); + + register('destroyRequest', (eventId, resource) => { + this.counters.destroyRequests++; + }); + + register('destroySuccess', (eventId, resource) => { + this.counters.destroySuccesses++; + }); + + register('destroyFail', (eventId, resource, err) => { + this.counters.destroyFailures++; + }); + + register('poolDestroying', () => { + this.detach(); + }); + + return this; + } + + /** + * Detach all event listeners from pool. + */ + detach() { + if (this._pool && typeof this._pool.removeListener === 'function') { + for (const [event, handler] of this._listeners.entries()) { + this._pool.removeListener(event, handler); + } + } + this._listeners.clear(); + this._pendingAcquireTimers.clear(); + this._pool = null; + } + + _recordAcquireDuration(durationMs) { + this.timing.lastAcquireDurationMs = durationMs; + this.timing.totalAcquireDurationMs += durationMs; + this.timing.acquireSamplesCount++; + if (durationMs > this.timing.maxAcquireDurationMs) { + this.timing.maxAcquireDurationMs = durationMs; + } + } + + _checkHealth() { + const stats = this.getSnapshot(); + if (!stats || !stats.max) return; + + const utilization = stats.utilization; + const now = Date.now(); + + if (utilization >= this.criticalThreshold) { + if (!this.lastCriticalAt || now - this.lastCriticalAt > 60000) { + this.lastCriticalAt = now; + logger.error('CRITICAL: Database connection pool nearly exhausted', { + used: stats.used, + max: stats.max, + utilization: `${Math.round(utilization * 100)}%`, + pendingAcquires: stats.pendingAcquires, + }); + if (typeof this.onCritical === 'function') { + this.onCritical(stats); + } + } + } else if (utilization >= this.warningThreshold) { + if (!this.lastWarningAt || now - this.lastWarningAt > 60000) { + this.lastWarningAt = now; + logger.warn('WARNING: Database connection pool utilization high', { + used: stats.used, + max: stats.max, + utilization: `${Math.round(utilization * 100)}%`, + pendingAcquires: stats.pendingAcquires, + }); + if (typeof this.onWarning === 'function') { + this.onWarning(stats); + } + } + } + } + + /** + * Get instant snapshot of pool state. + */ + getSnapshot() { + if (!this._pool) return null; + + const min = this._pool.min ?? 0; + const max = this._pool.max ?? 10; + const used = typeof this._pool.numUsed === 'function' ? this._pool.numUsed() : (this._pool.used ?? 0); + const free = typeof this._pool.numFree === 'function' ? this._pool.numFree() : (this._pool.free ?? 0); + const pendingAcquires = + typeof this._pool.numPendingAcquires === 'function' + ? this._pool.numPendingAcquires() + : (this._pool.pendingAcquires ?? 0); + const pendingCreates = + typeof this._pool.numPendingCreates === 'function' + ? this._pool.numPendingCreates() + : (this._pool.pendingCreates ?? 0); + + const utilization = max > 0 ? parseFloat((used / max).toFixed(4)) : 0; + const saturation = max > 0 ? parseFloat(((used + pendingAcquires) / max).toFixed(4)) : 0; + + let status = 'healthy'; + if (utilization >= this.criticalThreshold || this.counters.acquireFailures > 0) { + status = 'critical'; + } else if (utilization >= this.warningThreshold || pendingAcquires > 0) { + status = 'warning'; + } + + return { + min, + max, + used, + free, + pendingAcquires, + pendingCreates, + utilization, + saturation, + status, + }; + } + + /** + * Get full metrics summary including counters and timing. + */ + getMetrics() { + const snapshot = this.getSnapshot() || { + min: 0, + max: 10, + used: 0, + free: 0, + pendingAcquires: 0, + pendingCreates: 0, + utilization: 0, + saturation: 0, + status: 'unknown', + }; + + const avgAcquireDurationMs = + this.timing.acquireSamplesCount > 0 + ? parseFloat((this.timing.totalAcquireDurationMs / this.timing.acquireSamplesCount).toFixed(2)) + : 0; + + return { + ...snapshot, + counters: { ...this.counters }, + timing: { + lastAcquireDurationMs: this.timing.lastAcquireDurationMs, + maxAcquireDurationMs: this.timing.maxAcquireDurationMs, + avgAcquireDurationMs, + acquireSamplesCount: this.timing.acquireSamplesCount, + }, + lastFailure: this.lastFailure, + }; + } + + /** + * Format metrics in standard Prometheus exposition text format. + * + * @returns {string} Prometheus formatted metrics + */ + toPrometheusMetrics() { + const metrics = this.getMetrics(); + const lines = []; + + lines.push('# HELP db_pool_connections_min Minimum pool size'); + lines.push('# TYPE db_pool_connections_min gauge'); + lines.push(`db_pool_connections_min ${metrics.min}`); + + lines.push('# HELP db_pool_connections_max Maximum pool size'); + lines.push('# TYPE db_pool_connections_max gauge'); + lines.push(`db_pool_connections_max ${metrics.max}`); + + lines.push('# HELP db_pool_connections_used Connections currently acquired from pool'); + lines.push('# TYPE db_pool_connections_used gauge'); + lines.push(`db_pool_connections_used ${metrics.used}`); + + lines.push('# HELP db_pool_connections_free Idle connections available in pool'); + lines.push('# TYPE db_pool_connections_free gauge'); + lines.push(`db_pool_connections_free ${metrics.free}`); + + lines.push('# HELP db_pool_connections_pending_acquires Pending connection acquire requests'); + lines.push('# TYPE db_pool_connections_pending_acquires gauge'); + lines.push(`db_pool_connections_pending_acquires ${metrics.pendingAcquires}`); + + lines.push('# HELP db_pool_connections_pending_creates Pending connection creation operations'); + lines.push('# TYPE db_pool_connections_pending_creates gauge'); + lines.push(`db_pool_connections_pending_creates ${metrics.pendingCreates}`); + + lines.push('# HELP db_pool_connections_utilization Current connection pool utilization ratio (0.0 to 1.0)'); + lines.push('# TYPE db_pool_connections_utilization gauge'); + lines.push(`db_pool_connections_utilization ${metrics.utilization}`); + + lines.push('# HELP db_pool_acquires_total Total connection acquire attempts'); + lines.push('# TYPE db_pool_acquires_total counter'); + lines.push(`db_pool_acquires_total ${metrics.counters.acquireRequests}`); + + lines.push('# HELP db_pool_acquire_successes_total Total successful connection acquires'); + lines.push('# TYPE db_pool_acquire_successes_total counter'); + lines.push(`db_pool_acquire_successes_total ${metrics.counters.acquireSuccesses}`); + + lines.push('# HELP db_pool_acquire_failures_total Total failed connection acquire attempts'); + lines.push('# TYPE db_pool_acquire_failures_total counter'); + lines.push(`db_pool_acquire_failures_total ${metrics.counters.acquireFailures}`); + + lines.push('# HELP db_pool_releases_total Total connection releases back to pool'); + lines.push('# TYPE db_pool_releases_total counter'); + lines.push(`db_pool_releases_total ${metrics.counters.releases}`); + + lines.push('# HELP db_pool_creates_total Total connection creation attempts'); + lines.push('# TYPE db_pool_creates_total counter'); + lines.push(`db_pool_creates_total ${metrics.counters.createRequests}`); + + lines.push('# HELP db_pool_create_failures_total Total connection creation failures'); + lines.push('# TYPE db_pool_create_failures_total counter'); + lines.push(`db_pool_create_failures_total ${metrics.counters.createFailures}`); + + lines.push('# HELP db_pool_acquire_duration_ms Average connection acquire latency in milliseconds'); + lines.push('# TYPE db_pool_acquire_duration_ms gauge'); + lines.push(`db_pool_acquire_duration_ms ${metrics.timing.avgAcquireDurationMs}`); + + return lines.join('\n') + '\n'; + } +} + +// Global default monitor singleton +const defaultPoolMonitor = new PoolMonitor(); + +module.exports = { + PoolMonitor, + defaultPoolMonitor, +}; diff --git a/src/routes/metrics.js b/src/routes/metrics.js index 883ec0e..9b660fe 100644 --- a/src/routes/metrics.js +++ b/src/routes/metrics.js @@ -6,6 +6,7 @@ const eventStore = require('../indexer/eventStore'); const indexerPoller = require('../indexer/runtime'); const webhookRepo = require('../repositories/webhookRepository'); const deliveryRepo = require('../repositories/deliveryRepository'); +const { defaultPoolMonitor } = require('../database'); const router = express.Router(); @@ -96,6 +97,12 @@ async function metricsHandler(_req, res) { lines.push('# TYPE redis_connected gauge'); lines.push(`redis_connected ${redisConnected ? 1 : 0}`); + // Database connection pool metrics (#520) + const poolProm = defaultPoolMonitor.toPrometheusMetrics(); + if (poolProm) { + lines.push(poolProm.trim()); + } + res.setHeader('Content-Type', 'text/plain; version=0.0.4; charset=utf-8'); res.send(lines.join('\n') + '\n'); } catch (err) { diff --git a/src/services/dbCache.js b/src/services/dbCache.js index fd0eb05..0532e08 100644 --- a/src/services/dbCache.js +++ b/src/services/dbCache.js @@ -41,11 +41,29 @@ function generateCacheKey(query, params) { * @param {Function} options.executor - Query executor function (required) * @returns {Promise<*>} Query result */ +const { fetchWithLock } = require('../cache'); + +/** + * Execute a query with caching support and cache stampede protection (#518). + * Returns cached result if available, otherwise executes the query under lock + * to prevent multiple concurrent requests from hitting the database simultaneously. + * + * @param {string} query - SQL query string + * @param {Array} params - Query parameters + * @param {Object} options - Cache options + * @param {number} options.ttlSeconds - Cache TTL in seconds (default: 60) + * @param {boolean} options.skipCache - Skip cache and force fresh query + * @param {boolean} options.backgroundRefresh - Refresh cache in background when expired + * @param {Function} options.executor - Query executor function (required) + * @returns {Promise<*>} Query result + */ async function cachedQuery(query, params = [], options = {}) { const { ttlSeconds = DEFAULT_TTL_SECONDS, skipCache = false, + backgroundRefresh = false, executor, + ...restOptions } = options; if (!executor) { @@ -54,34 +72,16 @@ async function cachedQuery(query, params = [], options = {}) { const cacheKey = generateCacheKey(query, params); - // Try cache first - if (!skipCache) { - try { - const cached = await cache.get(cacheKey); - if (cached !== null && cached !== undefined) { - logger.debug('Database query cache hit', { cacheKey }); - return typeof cached === 'string' ? JSON.parse(cached) : cached; - } - } catch (err) { - // Cache miss or Redis unavailable — continue to query - logger.debug('Database query cache read failed', { cacheKey, error: err.message }); - } - } - - // Execute the query - const result = await executor(query, params); - - // Cache the result - if (!skipCache) { - try { - await cache.set(cacheKey, result, ttlSeconds); - logger.debug('Database query cached', { cacheKey, ttlSeconds }); - } catch (err) { - logger.warn('Database query cache write failed', { cacheKey, error: err.message }); + return fetchWithLock( + cacheKey, + () => executor(query, params), + { + ttlSeconds, + skipCache, + backgroundRefresh, + ...restOptions, } - } - - return result; + ); } /** diff --git a/test/auth.test.js b/test/auth.test.js index d45a1aa..c45cf68 100644 --- a/test/auth.test.js +++ b/test/auth.test.js @@ -32,13 +32,61 @@ const mockRedis = { const stopIdx = stop === -1 ? entries.length + stop : stop; return entries.slice(startIdx, stopIdx + 1).map(([member]) => member); }), + multi: jest.fn(() => { + const ops = []; + const chain = { + set: jest.fn((k, v) => { + ops.push(() => mockStore.set(k, v)); + return chain; + }), + del: jest.fn((k) => { + ops.push(() => mockStore.delete(k)); + return chain; + }), + zadd: jest.fn((k, score, member) => { + ops.push(() => mockRedis.zadd(k, score, member)); + return chain; + }), + zrem: jest.fn((k, member) => { + ops.push(() => mockRedis.zrem(k, member)); + return chain; + }), + exec: jest.fn(async () => { + for (const op of ops) await op(); + return []; + }), + }; + return chain; + }), }; jest.mock('../src/services/cache', () => ({ getClient: () => mockRedis, get: jest.fn(async (key) => { const v = mockStore.get(key); - return v !== undefined ? JSON.parse(JSON.stringify(v)) : null; + if (v === undefined || v === null) return null; + if (typeof v === 'string') { + try { + return JSON.parse(v); + } catch { + return v; + } + } + return JSON.parse(JSON.stringify(v)); + }), + mget: jest.fn(async (keys) => { + return keys.map((k) => { + const v = mockStore.get(k); + if (v === undefined || v === null) return null; + if (typeof v === 'string') { + try { + return JSON.parse(v); + } catch { + return v; + } + } + return JSON.parse(JSON.stringify(v)); + }); }), set: jest.fn(async (key, value) => { mockStore.set(key, JSON.parse(JSON.stringify(value))); diff --git a/test/cacheStampede.test.js b/test/cacheStampede.test.js new file mode 100644 index 0000000..475a89a --- /dev/null +++ b/test/cacheStampede.test.js @@ -0,0 +1,234 @@ +'use strict'; + +const { + SingleFlight, + acquireLock, + releaseLock, + fetchWithLock, +} = require('../src/cache/stampede'); +const { cachedQuery } = require('../src/services/dbCache'); +const cache = require('../src/services/cache'); + +// Mock cache +const mockStore = new Map(); +const mockLockStore = new Map(); + +jest.mock('../src/services/cache', () => ({ + get: jest.fn(async (key) => mockStore.get(key) || null), + set: jest.fn(async (key, value, ttl) => { + mockStore.set(key, value); + }), + del: jest.fn(async (key) => { + mockStore.delete(key); + }), + getClient: jest.fn(() => ({ + set: jest.fn(async (key, val, px, ttl, nx) => { + if (nx === 'NX') { + if (mockLockStore.has(key)) return null; + mockLockStore.set(key, val); + return 'OK'; + } + mockLockStore.set(key, val); + return 'OK'; + }), + get: jest.fn(async (key) => mockLockStore.get(key) || null), + del: jest.fn(async (key) => { + const existed = mockLockStore.delete(key); + return existed ? 1 : 0; + }), + eval: jest.fn(async (script, numKeys, key, token) => { + if (mockLockStore.get(key) === token) { + mockLockStore.delete(key); + return 1; + } + return 0; + }), + })), +})); + +jest.mock('../src/logger', () => ({ + debug: jest.fn(), + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), +})); + +describe('Cache Stampede Protection (#518)', () => { + beforeEach(() => { + mockStore.clear(); + mockLockStore.clear(); + jest.clearAllMocks(); + }); + + describe('SingleFlight', () => { + test('coalesces multiple simultaneous calls for the same key into a single invocation', async () => { + const sf = new SingleFlight(); + let callCount = 0; + + const slowFn = async () => { + callCount++; + await new Promise((r) => setTimeout(r, 50)); + return 'data-123'; + }; + + const results = await Promise.all([ + sf.do('key-1', slowFn), + sf.do('key-1', slowFn), + sf.do('key-1', slowFn), + ]); + + expect(callCount).toBe(1); + expect(results).toEqual(['data-123', 'data-123', 'data-123']); + expect(sf.size).toBe(0); + }); + + test('executes again after previous promise completes', async () => { + const sf = new SingleFlight(); + let callCount = 0; + const fn = async () => { + callCount++; + return `call-${callCount}`; + }; + + const res1 = await sf.do('key-1', fn); + const res2 = await sf.do('key-1', fn); + + expect(callCount).toBe(2); + expect(res1).toBe('call-1'); + expect(res2).toBe('call-2'); + }); + + test('cleans up key and propagates error when fn throws', async () => { + const sf = new SingleFlight(); + const failingFn = async () => { + throw new Error('fetch error'); + }; + + await expect(sf.do('err-key', failingFn)).rejects.toThrow('fetch error'); + expect(sf.has('err-key')).toBe(false); + }); + }); + + describe('acquireLock & releaseLock', () => { + test('acquires lock and releases with matching token', async () => { + const token = await acquireLock('test-key', { lockTtlMs: 1000 }); + expect(token).toBeTruthy(); + + // Second acquire fails + const secondToken = await acquireLock('test-key', { lockTtlMs: 1000 }); + expect(secondToken).toBeNull(); + + // Release with token succeeds + const released = await releaseLock('test-key', token); + expect(released).toBe(true); + + // Now can acquire again + const thirdToken = await acquireLock('test-key', { lockTtlMs: 1000 }); + expect(thirdToken).toBeTruthy(); + }); + + test('releaseLock fails if wrong token is provided', async () => { + const token = await acquireLock('protected-key', { lockTtlMs: 1000 }); + const released = await releaseLock('protected-key', 'wrong-token'); + expect(released).toBe(false); + }); + }); + + describe('fetchWithLock', () => { + test('returns cached value without calling fetcher if cache hits', async () => { + mockStore.set('cache:hit', JSON.stringify({ name: 'Alice' })); + const fetcher = jest.fn(); + + const result = await fetchWithLock('cache:hit', fetcher); + expect(result).toEqual({ name: 'Alice' }); + expect(fetcher).not.toHaveBeenCalled(); + }); + + test('prevents cache stampede: concurrent calls invoke fetcher only once', async () => { + let executions = 0; + const slowFetcher = async () => { + executions++; + await new Promise((r) => setTimeout(r, 60)); + return { count: executions, data: 'loaded from db' }; + }; + + // 10 concurrent requests for the same missing key + const results = await Promise.all([ + fetchWithLock('stampede:key', slowFetcher), + fetchWithLock('stampede:key', slowFetcher), + fetchWithLock('stampede:key', slowFetcher), + fetchWithLock('stampede:key', slowFetcher), + fetchWithLock('stampede:key', slowFetcher), + ]); + + expect(executions).toBe(1); + results.forEach((res) => { + expect(res).toEqual({ count: 1, data: 'loaded from db' }); + }); + + // Subsequent call gets value from cache + const nextResult = await fetchWithLock('stampede:key', slowFetcher); + expect(executions).toBe(1); + expect(nextResult).toEqual({ count: 1, data: 'loaded from db' }); + }); + + test('skipCache executes fetcher directly without caching', async () => { + const fetcher = jest.fn(async () => 'fresh-value'); + const res = await fetchWithLock('skip:key', fetcher, { skipCache: true }); + expect(res).toBe('fresh-value'); + expect(fetcher).toHaveBeenCalledTimes(1); + }); + + test('backgroundRefresh serves stale data and asynchronously updates cache', async () => { + // Setup expired cache with metadata + mockStore.set('stale:key', JSON.stringify({ + __stampede_meta: true, + value: 'old-stale-value', + expiresAt: Date.now() - 5000, // already expired + })); + + let fetcherCalls = 0; + const fetcher = jest.fn(async () => { + fetcherCalls++; + return 'newly-refreshed-value'; + }); + + const res = await fetchWithLock('stale:key', fetcher, { + backgroundRefresh: true, + ttlSeconds: 30, + }); + + // Immediately returned stale value + expect(res).toBe('old-stale-value'); + + // Wait a tick for background setImmediate to run + await new Promise((r) => setTimeout(r, 50)); + expect(fetcher).toHaveBeenCalled(); + }); + }); + + describe('cachedQuery integration', () => { + test('coalesces identical simultaneous SQL queries through dbCache', async () => { + let dbExecutions = 0; + const mockExecutor = jest.fn(async (sql, params) => { + dbExecutions++; + await new Promise((r) => setTimeout(r, 50)); + return [{ id: params[0], name: 'SmartDrop' }]; + }); + + const query = 'SELECT * FROM drops WHERE id = $1'; + const params = [42]; + + const results = await Promise.all([ + cachedQuery(query, params, { executor: mockExecutor }), + cachedQuery(query, params, { executor: mockExecutor }), + cachedQuery(query, params, { executor: mockExecutor }), + ]); + + expect(dbExecutions).toBe(1); + results.forEach((res) => { + expect(res).toEqual([{ id: 42, name: 'SmartDrop' }]); + }); + }); + }); +}); diff --git a/test/databaseEncryption.test.js b/test/databaseEncryption.test.js new file mode 100644 index 0000000..c0fdea8 --- /dev/null +++ b/test/databaseEncryption.test.js @@ -0,0 +1,160 @@ +'use strict'; + +const { + encrypt, + decrypt, + isEncrypted, + encryptRecord, + decryptRecord, + encryptRows, + decryptRows, + hashBlindIndex, + deriveKey, +} = require('../src/database/encryption'); + +describe('Database Encryption at Rest (#519)', () => { + describe('Basic Encryption & Decryption', () => { + test('encrypts string and decrypts back to original', () => { + const plaintext = 'sk_live_super_secret_stellar_key_12345'; + const encrypted = encrypt(plaintext); + + expect(isEncrypted(encrypted)).toBe(true); + expect(encrypted).toMatch(/^enc:v1:[A-Za-z0-9+/=]+:[A-Za-z0-9+/=]+:[A-Za-z0-9+/=]+$/); + expect(encrypted).not.toContain(plaintext); + + const decrypted = decrypt(encrypted); + expect(decrypted).toBe(plaintext); + }); + + test('isEncrypted identifies encrypted vs plaintext values', () => { + expect(isEncrypted('enc:v1:abc:def:ghi')).toBe(true); + expect(isEncrypted('plain-secret-123')).toBe(false); + expect(isEncrypted(null)).toBe(false); + expect(isEncrypted(undefined)).toBe(false); + expect(isEncrypted(12345)).toBe(false); + }); + + test('encrypting already encrypted value returns it unchanged', () => { + const encrypted = encrypt('my-secret'); + const doubleEncrypted = encrypt(encrypted); + expect(doubleEncrypted).toBe(encrypted); + }); + + test('decrypting unencrypted legacy value returns it untouched (migration safe)', () => { + const legacyValue = 'legacy_plaintext_value'; + expect(decrypt(legacyValue)).toBe(legacyValue); + expect(decrypt(null)).toBeNull(); + expect(decrypt(undefined)).toBeUndefined(); + }); + + test('encrypts and decrypts JSON objects with parseJson: true', () => { + const obj = { account: 'GBXXX', secretSeed: 'SXXXX', balance: 500 }; + const encrypted = encrypt(obj); + + const decryptedStr = decrypt(encrypted); + expect(typeof decryptedStr).toBe('string'); + + const decryptedObj = decrypt(encrypted, { parseJson: true }); + expect(decryptedObj).toEqual(obj); + }); + + test('fails decryption if ciphertext is tampered with (auth tag validation)', () => { + const encrypted = encrypt('sensitive-bank-account'); + const parts = encrypted.split(':'); + // parts: ['enc', 'v1', iv, ciphertext, authTag] + // tamper with ciphertext + const tamperedCiphertext = Buffer.from('corrupted-data').toString('base64'); + parts[3] = tamperedCiphertext; + const tampered = parts.join(':'); + + expect(() => decrypt(tampered)).toThrow(); + }); + + test('supports custom key override', () => { + const key1 = 'custom-key-alpha-123'; + const key2 = 'custom-key-beta-456'; + const secret = 'custom-secret-data'; + + const encrypted = encrypt(secret, { key: key1 }); + expect(decrypt(encrypted, { key: key1 })).toBe(secret); + expect(() => decrypt(encrypted, { key: key2 })).toThrow(); + }); + }); + + describe('Record & Row Field-Level Encryption', () => { + test('encrypts and decrypts specified fields in a database row', () => { + const row = { + id: 'usr_1', + username: 'alice', + email: 'alice@example.com', + private_key: 'S_ALICE_SECRET_KEY', + created_at: '2026-01-01T00:00:00Z', + }; + + const encryptedRow = encryptRecord(row, ['email', 'private_key']); + + // Plaintext fields untouched + expect(encryptedRow.id).toBe('usr_1'); + expect(encryptedRow.username).toBe('alice'); + expect(encryptedRow.created_at).toBe('2026-01-01T00:00:00Z'); + + // Sensitive fields encrypted + expect(isEncrypted(encryptedRow.email)).toBe(true); + expect(isEncrypted(encryptedRow.private_key)).toBe(true); + expect(encryptedRow.email).not.toContain('alice@example.com'); + expect(encryptedRow.private_key).not.toContain('S_ALICE_SECRET_KEY'); + + // Decrypt specified fields + const decryptedRow = decryptRecord(encryptedRow, ['email', 'private_key']); + expect(decryptedRow).toEqual(row); + }); + + test('decryptRecord automatically decrypts any encrypted fields when field list is omitted', () => { + const row = { + id: 'wh_1', + url: 'https://example.com/webhook', + secret: encrypt('whsec_test_secret_abc'), + token: encrypt('bearer_token_xyz'), + }; + + const decrypted = decryptRecord(row); + expect(decrypted.secret).toBe('whsec_test_secret_abc'); + expect(decrypted.token).toBe('bearer_token_xyz'); + expect(decrypted.url).toBe('https://example.com/webhook'); + }); + + test('encryptRows and decryptRows handle collections of rows', () => { + const rows = [ + { id: 1, secret: 'secret-1', public: 'pub-1' }, + { id: 2, secret: 'secret-2', public: 'pub-2' }, + ]; + + const encryptedRows = encryptRows(rows, ['secret']); + expect(isEncrypted(encryptedRows[0].secret)).toBe(true); + expect(isEncrypted(encryptedRows[1].secret)).toBe(true); + expect(encryptedRows[0].public).toBe('pub-1'); + + const decryptedRows = decryptRows(encryptedRows, ['secret']); + expect(decryptedRows).toEqual(rows); + }); + }); + + describe('Blind Indexing for Searchable Encryption', () => { + test('produces deterministic HMAC-SHA256 hash for searchable encrypted fields', () => { + const email = 'User.Name@Example.COM'; + const hash1 = hashBlindIndex(email, 'user_email'); + const hash2 = hashBlindIndex('user.name@example.com', 'user_email'); // case-insensitive match + + expect(hash1).toMatch(/^[a-f0-9]{64}$/); + expect(hash1).toBe(hash2); + + const differentEmailHash = hashBlindIndex('other@example.com', 'user_email'); + expect(differentEmailHash).not.toBe(hash1); + }); + + test('returns null for null/undefined input', () => { + expect(hashBlindIndex(null)).toBeNull(); + expect(hashBlindIndex(undefined)).toBeNull(); + }); + }); +}); diff --git a/test/poolMonitor.test.js b/test/poolMonitor.test.js new file mode 100644 index 0000000..6addb77 --- /dev/null +++ b/test/poolMonitor.test.js @@ -0,0 +1,214 @@ +'use strict'; + +const EventEmitter = require('events'); +const { PoolMonitor, defaultPoolMonitor } = require('../src/database/poolMonitor'); +const { getPoolStats, getPoolMetrics, setDb } = require('../src/database'); + +// Mock pool with EventEmitter +class MockTarnPool extends EventEmitter { + constructor(options = {}) { + super(); + this.min = options.min ?? 2; + this.max = options.max ?? 10; + this.used = options.used ?? 0; + this.free = options.free ?? 2; + this.pendingAcquires = options.pendingAcquires ?? 0; + this.pendingCreates = options.pendingCreates ?? 0; + } + + numUsed() { + return this.used; + } + + numFree() { + return this.free; + } + + numPendingAcquires() { + return this.pendingAcquires; + } + + numPendingCreates() { + return this.pendingCreates; + } +} + +describe('Database Connection Pool Monitoring (#520)', () => { + let mockPool; + let monitor; + + beforeEach(() => { + mockPool = new MockTarnPool({ min: 2, max: 10, used: 2, free: 8 }); + monitor = new PoolMonitor(); + monitor.attach(mockPool); + }); + + afterEach(() => { + monitor.detach(); + }); + + describe('Event Tracking & Counters', () => { + test('tracks acquire request, success, and release lifecycle', () => { + expect(monitor.counters.acquireRequests).toBe(0); + expect(monitor.counters.acquireSuccesses).toBe(0); + expect(monitor.counters.releases).toBe(0); + + // Acquire start + mockPool.emit('acquireRequest', 1); + expect(monitor.counters.acquireRequests).toBe(1); + + // Acquire success + mockPool.used = 3; + mockPool.free = 7; + mockPool.emit('acquireSuccess', 1, {}); + expect(monitor.counters.acquireSuccesses).toBe(1); + + // Release + mockPool.used = 2; + mockPool.free = 8; + mockPool.emit('release', {}); + expect(monitor.counters.releases).toBe(1); + }); + + test('tracks connection creation and failure', () => { + mockPool.emit('createRequest', 101); + expect(monitor.counters.createRequests).toBe(1); + + mockPool.emit('createSuccess', 101, {}); + expect(monitor.counters.createSuccesses).toBe(1); + + const error = new Error('Connection refused: 5432'); + mockPool.emit('createFail', 102, error); + expect(monitor.counters.createFailures).toBe(1); + expect(monitor.lastFailure).toMatchObject({ + type: 'createFail', + error: 'Connection refused: 5432', + }); + }); + + test('tracks acquire failure and updates status', () => { + mockPool.emit('acquireRequest', 201); + const timeoutErr = new Error('Knex: Timeout acquiring a connection'); + mockPool.emit('acquireFail', 201, timeoutErr); + + expect(monitor.counters.acquireFailures).toBe(1); + expect(monitor.lastFailure).toMatchObject({ + type: 'acquireFail', + error: 'Knex: Timeout acquiring a connection', + }); + + const metrics = monitor.getMetrics(); + expect(metrics.status).toBe('critical'); + }); + }); + + describe('Utilization & Health Status', () => { + test('calculates correct utilization ratio', () => { + mockPool.used = 5; + mockPool.max = 10; + const snapshot = monitor.getSnapshot(); + + expect(snapshot.used).toBe(5); + expect(snapshot.max).toBe(10); + expect(snapshot.utilization).toBe(0.5); + expect(snapshot.status).toBe('healthy'); + }); + + test('transitions to warning when utilization >= 80%', () => { + mockPool.used = 8; + mockPool.max = 10; + const snapshot = monitor.getSnapshot(); + + expect(snapshot.utilization).toBe(0.8); + expect(snapshot.status).toBe('warning'); + }); + + test('transitions to critical when utilization >= 95%', () => { + mockPool.used = 10; + mockPool.max = 10; + const snapshot = monitor.getSnapshot(); + + expect(snapshot.utilization).toBe(1.0); + expect(snapshot.status).toBe('critical'); + }); + + test('triggers warning and critical callbacks when thresholds exceeded', () => { + const onWarning = jest.fn(); + const onCritical = jest.fn(); + + const alertMonitor = new PoolMonitor({ + warningThreshold: 0.7, + criticalThreshold: 0.9, + onWarning, + onCritical, + }); + alertMonitor.attach(mockPool); + + // Exceed warning + mockPool.used = 7; + mockPool.max = 10; + mockPool.emit('acquireRequest', 1); + mockPool.emit('acquireSuccess', 1, {}); + expect(onWarning).toHaveBeenCalledTimes(1); + + // Exceed critical + mockPool.used = 9; + mockPool.emit('acquireRequest', 2); + mockPool.emit('acquireSuccess', 2, {}); + expect(onCritical).toHaveBeenCalledTimes(1); + + alertMonitor.detach(); + }); + }); + + describe('Prometheus Metrics Formatting', () => { + test('formats metrics according to standard Prometheus text specification', () => { + mockPool.used = 3; + mockPool.max = 10; + mockPool.free = 7; + mockPool.pendingAcquires = 1; + + mockPool.emit('acquireRequest', 1); + mockPool.emit('acquireSuccess', 1, {}); + + const prometheusText = monitor.toPrometheusMetrics(); + + expect(prometheusText).toContain('# HELP db_pool_connections_used Connections currently acquired from pool'); + expect(prometheusText).toContain('# TYPE db_pool_connections_used gauge'); + expect(prometheusText).toContain('db_pool_connections_used 3'); + expect(prometheusText).toContain('db_pool_connections_max 10'); + expect(prometheusText).toContain('db_pool_connections_free 7'); + expect(prometheusText).toContain('db_pool_connections_utilization 0.3'); + expect(prometheusText).toContain('db_pool_acquires_total 1'); + expect(prometheusText).toContain('db_pool_acquire_successes_total 1'); + }); + }); + + describe('Integration with Database Index API', () => { + test('getPoolStats and getPoolMetrics reflect active pool state', () => { + const mockKnex = { + client: { + pool: mockPool, + }, + }; + + setDb(mockKnex); + + const stats = getPoolStats(); + expect(stats).toMatchObject({ + min: 2, + max: 10, + used: 2, + free: 8, + status: 'healthy', + }); + + const metrics = getPoolMetrics(); + expect(metrics).toHaveProperty('counters'); + expect(metrics).toHaveProperty('timing'); + expect(metrics).toHaveProperty('status'); + + setDb(null); + }); + }); +}); diff --git a/test/tokenRefresh.test.js b/test/tokenRefresh.test.js new file mode 100644 index 0000000..f620cee --- /dev/null +++ b/test/tokenRefresh.test.js @@ -0,0 +1,207 @@ +'use strict'; + +const express = require('express'); +const request = require('supertest'); +const { + signJwt, + createTokenPair, + refreshTokens, + revokeRefreshToken, + createRefreshHandler, + createRevokeHandler, +} = require('../src/auth/tokenRefresh'); +const { verifyJwt } = require('../src/middleware/jwtAuth'); +const { errorHandler } = require('../src/middleware/errorHandler'); +jest.mock('../src/services/cache', () => ({ + isConnected: jest.fn(() => false), + getClient: jest.fn(() => null), +})); + +jest.mock('../src/logger', () => ({ + debug: jest.fn(), + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), +})); + +const TEST_SECRET = 'super-secret-jwt-signing-key-for-test-12345'; + +describe('Token Refresh Mechanism (#521)', () => { + const testUser = { + id: 'usr_abc123', + email: 'user@smartdrop.org', + username: 'smartdrop_user', + scopes: ['read', 'write'], + }; + + describe('signJwt & verifyJwt Integration', () => { + test('signed JWT is verified by verifyJwt', () => { + const token = signJwt( + { sub: testUser.id, email: testUser.email }, + TEST_SECRET, + { expiresInSeconds: 60 } + ); + + expect(typeof token).toBe('string'); + expect(token.split('.')).toHaveLength(3); + + const verified = verifyJwt(token, TEST_SECRET); + expect(verified.sub).toBe(testUser.id); + expect(verified.email).toBe(testUser.email); + expect(verified.exp).toBeGreaterThan(Math.floor(Date.now() / 1000)); + }); + + test('expired JWT fails verification', () => { + const expiredToken = signJwt( + { sub: testUser.id }, + TEST_SECRET, + { expiresInSeconds: -10 } + ); + + expect(() => verifyJwt(expiredToken, TEST_SECRET)).toThrow(/expired/); + }); + }); + + describe('createTokenPair', () => { + test('creates access and refresh token pair with correct metadata', async () => { + const pair = await createTokenPair(testUser, { + jwtSecret: TEST_SECRET, + accessTokenTtlSeconds: 300, + refreshTokenTtlSeconds: 3600, + }); + + expect(pair.tokenType).toBe('Bearer'); + expect(pair.expiresIn).toBe(300); + expect(pair.accessToken).toBeTruthy(); + expect(pair.refreshToken).toMatch(/^[a-f0-9]{64}$/); + expect(pair.user.id).toBe(testUser.id); + + // Verify access token + const verified = verifyJwt(pair.accessToken, TEST_SECRET); + expect(verified.userId).toBe(testUser.id); + expect(verified.scopes).toEqual(['read', 'write']); + }); + }); + + describe('refreshTokens with Refresh Token Rotation (RTR)', () => { + test('refreshes token pair and rotates refresh token', async () => { + const initialPair = await createTokenPair(testUser, { jwtSecret: TEST_SECRET }); + + // Refresh with valid refresh token + const refreshedPair = await refreshTokens(initialPair.refreshToken, { + jwtSecret: TEST_SECRET, + }); + + expect(refreshedPair.accessToken).toBeTruthy(); + expect(refreshedPair.refreshToken).toBeTruthy(); + expect(refreshedPair.refreshToken).not.toBe(initialPair.refreshToken); + expect(refreshedPair.user.id).toBe(testUser.id); + + // Verify the new access token + const verified = verifyJwt(refreshedPair.accessToken, TEST_SECRET); + expect(verified.userId).toBe(testUser.id); + }); + + test('reuse detection: using an already-used refresh token revokes token family', async () => { + const initialPair = await createTokenPair(testUser, { jwtSecret: TEST_SECRET }); + + // 1st refresh (legitimate) + const refreshedPair = await refreshTokens(initialPair.refreshToken, { + jwtSecret: TEST_SECRET, + }); + + // 2nd refresh using the SAME initial refresh token (attacker replay / token theft!) + await expect( + refreshTokens(initialPair.refreshToken, { jwtSecret: TEST_SECRET }) + ).rejects.toMatchObject({ + statusCode: 401, + message: expect.stringContaining('reuse detected'), + }); + + // The newly issued token in the family must ALSO have been revoked due to reuse detection! + await expect( + refreshTokens(refreshedPair.refreshToken, { jwtSecret: TEST_SECRET }) + ).rejects.toMatchObject({ + statusCode: 401, + }); + }); + + test('rejects missing or empty refresh token with 401', async () => { + await expect(refreshTokens(null)).rejects.toMatchObject({ + statusCode: 401, + }); + await expect(refreshTokens('')).rejects.toMatchObject({ + statusCode: 401, + }); + }); + + test('rejects non-existent refresh token with 401', async () => { + await expect(refreshTokens('a'.repeat(64))).rejects.toMatchObject({ + statusCode: 401, + message: expect.stringContaining('Invalid or expired'), + }); + }); + }); + + describe('revokeRefreshToken', () => { + test('revoked token cannot be used to refresh', async () => { + const pair = await createTokenPair(testUser, { jwtSecret: TEST_SECRET }); + + const revoked = await revokeRefreshToken(pair.refreshToken); + expect(revoked).toBe(true); + + await expect(refreshTokens(pair.refreshToken)).rejects.toMatchObject({ + statusCode: 401, + }); + }); + }); + + describe('Express Handlers', () => { + function buildApp() { + const app = express(); + app.use(express.json()); + app.post('/auth/refresh', createRefreshHandler({ jwtSecret: TEST_SECRET })); + app.post('/auth/revoke', createRevokeHandler()); + app.use(errorHandler); + return app; + } + + test('POST /auth/refresh returns 200 and new tokens on valid refresh_token', async () => { + const app = buildApp(); + const pair = await createTokenPair(testUser, { jwtSecret: TEST_SECRET }); + + const res = await request(app) + .post('/auth/refresh') + .send({ refreshToken: pair.refreshToken }); + + expect(res.status).toBe(200); + expect(res.body.accessToken).toBeDefined(); + expect(res.body.refreshToken).toBeDefined(); + expect(res.body.tokenType).toBe('Bearer'); + }); + + test('POST /auth/refresh returns 400 when refresh token is missing', async () => { + const app = buildApp(); + const res = await request(app) + .post('/auth/refresh') + .send({}); + + expect(res.status).toBe(400); + expect(res.body.error).toMatchObject({ + code: 'VALIDATION_ERROR', + }); + }); + + test('POST /auth/revoke revokes token and returns 200', async () => { + const app = buildApp(); + const pair = await createTokenPair(testUser, { jwtSecret: TEST_SECRET }); + + const res = await request(app) + .post('/auth/revoke') + .send({ refreshToken: pair.refreshToken }); + + expect(res.status).toBe(200); + expect(res.body.revoked).toBe(true); + }); + }); +});