From f7e21e5af2fbf42890a8902e7ae2337e1907776e Mon Sep 17 00:00:00 2001 From: heymariam Date: Fri, 28 Aug 2026 02:03:52 +0100 Subject: [PATCH] feat: add API security hardening features - Add API key audit logging: tracks endpoint access, IP, status code, response time - Add API key rotation endpoint: atomic operation to create new key and revoke old one - Add idempotency key support for webhook registration: prevents duplicate webhooks - Add idempotency key support for airdrop creation: prevents duplicate airdrops These changes improve security, reliability, and prevent duplicate resource creation on retries. --- .../20260801000000_add_api_key_audit_logs.js | 25 + src/index.js | 188 ++++--- src/middleware/auth.js | 86 +++- src/routes/airdrops.js | 480 +++++++++++------- src/routes/keys.js | 60 ++- src/routes/webhooks.js | 301 ++++++----- src/services/apiKeyAuditLog.js | 100 ++++ src/services/apiKeys.js | 81 ++- src/services/idempotency.js | 96 ++++ src/validation/schemas.js | 159 +++--- 10 files changed, 1076 insertions(+), 500 deletions(-) create mode 100644 src/db/migrations/20260801000000_add_api_key_audit_logs.js create mode 100644 src/services/apiKeyAuditLog.js create mode 100644 src/services/idempotency.js diff --git a/src/db/migrations/20260801000000_add_api_key_audit_logs.js b/src/db/migrations/20260801000000_add_api_key_audit_logs.js new file mode 100644 index 0000000..4582072 --- /dev/null +++ b/src/db/migrations/20260801000000_add_api_key_audit_logs.js @@ -0,0 +1,25 @@ +/** + * Migration: Add API Key Audit Logs + * + * Creates table to track API key usage: which endpoint was accessed, when, and from which IP. + * This is essential for security auditing and detecting misuse. + */ + +exports.up = async (knex) => { + await knex.schema.createTable('api_key_audit_logs', (table) => { + table.increments('id').primary(); + table.string('key_id').notNullable().index(); + table.string('endpoint').notNullable(); // e.g. GET /api/prices + table.string('ip_address').notNullable(); + table.integer('status_code'); // HTTP response status + table.integer('response_time_ms'); // Request duration in milliseconds + table.timestamp('created_at').notNullable().defaultTo(knex.fn.now()); + + // Composite index for common queries: key_id + created_at + table.index(['key_id', 'created_at']); + }); +}; + +exports.down = async (knex) => { + await knex.schema.dropTable('api_key_audit_logs'); +}; diff --git a/src/index.js b/src/index.js index 76399f5..e9b1cb5 100644 --- a/src/index.js +++ b/src/index.js @@ -1,64 +1,74 @@ -'use strict'; - -const express = require('express'); -const compression = require('compression'); -const helmet = require('helmet'); -const config = require('./config'); -const { version: appVersion } = require('../package.json'); -const logger = require('./logger'); -const cache = require('./services/cache'); -const priceOracle = require('./services/priceOracle'); -const priceRefreshJob = require('./jobs/priceRefresh'); -const webhookRetryWorker = require('./jobs/webhookRetryWorker'); -const airdropExpiryJob = require('./jobs/airdropExpiry'); -const { createLeaderElection } = require('./services/leaderElection'); -const { makeLeaderAwareJob } = require('./jobs/leaderAwareJob'); -const { warmCache } = require('./startup/cacheWarm'); -const buildCorsMiddleware = require('./middleware/cors'); -const { buildRateLimit, buildApiKeyRateLimit } = require('./middleware/rateLimit'); -const { requestIdMiddleware } = require('./middleware/requestId'); -const requestLoggerMiddleware = require('./middleware/requestLogger'); -const { requireApiKey, attachApiKey } = require('./middleware/auth'); -const { errorHandler, notFoundHandler } = require('./middleware/errorHandler'); -const { checkDatabase } = require('./services/dbHealth'); -const pricesRouter = require('./routes/prices'); -const alertsRouter = require('./routes/alerts'); -const indexerRouter = require('./routes/indexer'); -const indexerPoller = require('./indexer/runtime'); -const keysRouter = require('./routes/keys'); -const webhooksRouter = require('./routes/webhooks'); -const airdropsRouter = require('./routes/airdrops'); -const apiDocsRouter = require('./routes/apiDocs'); -const { router: metricsRouter, requestMetricsMiddleware } = require('./routes/metrics'); - -const priceWebSocket = require('./ws/priceWebSocket'); -const subscriptionManager = require('./ws/PriceSubscriptionManager'); -const webhookDispatcher = require('./services/webhookDispatcher'); +"use strict"; + +const express = require("express"); +const compression = require("compression"); +const helmet = require("helmet"); +const config = require("./config"); +const { version: appVersion } = require("../package.json"); +const logger = require("./logger"); +const cache = require("./services/cache"); +const priceOracle = require("./services/priceOracle"); +const priceRefreshJob = require("./jobs/priceRefresh"); +const webhookRetryWorker = require("./jobs/webhookRetryWorker"); +const airdropExpiryJob = require("./jobs/airdropExpiry"); +const { createLeaderElection } = require("./services/leaderElection"); +const { makeLeaderAwareJob } = require("./jobs/leaderAwareJob"); +const { warmCache } = require("./startup/cacheWarm"); +const buildCorsMiddleware = require("./middleware/cors"); +const { + buildRateLimit, + buildApiKeyRateLimit, +} = require("./middleware/rateLimit"); +const { requestIdMiddleware } = require("./middleware/requestId"); +const requestLoggerMiddleware = require("./middleware/requestLogger"); +const { + requireApiKey, + attachApiKey, + auditApiKeyUsage, +} = require("./middleware/auth"); +const { errorHandler, notFoundHandler } = require("./middleware/errorHandler"); +const { checkDatabase } = require("./services/dbHealth"); +const pricesRouter = require("./routes/prices"); +const alertsRouter = require("./routes/alerts"); +const indexerRouter = require("./routes/indexer"); +const indexerPoller = require("./indexer/runtime"); +const keysRouter = require("./routes/keys"); +const webhooksRouter = require("./routes/webhooks"); +const airdropsRouter = require("./routes/airdrops"); +const apiDocsRouter = require("./routes/apiDocs"); +const { + router: metricsRouter, + requestMetricsMiddleware, +} = require("./routes/metrics"); + +const priceWebSocket = require("./ws/priceWebSocket"); +const subscriptionManager = require("./ws/PriceSubscriptionManager"); +const webhookDispatcher = require("./services/webhookDispatcher"); // Wrap background jobs with leader-election coordination so that only one // replica across the deployment runs each job at any given time. // See README.md#leader-election for design, failover timing, and configuration. -const leaderElectionPriceRefresh = createLeaderElection('price_refresh'); -const leaderElectionWebhookRetry = createLeaderElection('webhook_retry'); -const leaderElectionAirdropExpiry = createLeaderElection('airdrop_expiry'); +const leaderElectionPriceRefresh = createLeaderElection("price_refresh"); +const leaderElectionWebhookRetry = createLeaderElection("webhook_retry"); +const leaderElectionAirdropExpiry = createLeaderElection("airdrop_expiry"); const wrappedPriceRefreshJob = makeLeaderAwareJob({ job: priceRefreshJob, - jobName: 'price_refresh', + jobName: "price_refresh", leaderElection: leaderElectionPriceRefresh, logger, }); const wrappedWebhookRetryWorker = makeLeaderAwareJob({ job: webhookRetryWorker, - jobName: 'webhook_retry', + jobName: "webhook_retry", leaderElection: leaderElectionWebhookRetry, logger, }); const wrappedAirdropExpiryJob = makeLeaderAwareJob({ job: airdropExpiryJob, - jobName: 'airdrop_expiry', + jobName: "airdrop_expiry", leaderElection: leaderElectionAirdropExpiry, logger, }); @@ -86,16 +96,19 @@ const EMPTY_QUEUE_STATS = { }; async function readWebhookRetryQueueStats() { - if (typeof webhookRetryWorker.getQueueStats !== 'function') return EMPTY_QUEUE_STATS; + if (typeof webhookRetryWorker.getQueueStats !== "function") + return EMPTY_QUEUE_STATS; try { return await webhookRetryWorker.getQueueStats(); } catch (err) { - logger.warn('Could not read webhook retry queue stats', { error: err.message }); + logger.warn("Could not read webhook retry queue stats", { + error: err.message, + }); return EMPTY_QUEUE_STATS; } } -app.get('/health', async (req, res) => { +app.get("/health", async (req, res) => { const redisConnected = cache.isConnected(); const redisQueueDepth = cache.getCommandQueueLength(); const redisConcurrency = cache.getConcurrencyStats(); @@ -118,14 +131,25 @@ app.get('/health', async (req, res) => { // aren't running locally), but that's expected — the leader is doing the // work. The health check distinguishes "not leader" from "stalled" via the // `leader` field. - let status = 'ok'; - if (!redisConnected || !priceRefreshHealth.healthy || !webhookWorkerHealth.healthy || database.status === 'error') { + let status = "ok"; + if ( + !redisConnected || + !priceRefreshHealth.healthy || + !webhookWorkerHealth.healthy || + database.status === "error" + ) { const jobsDegraded = (!priceRefreshHealth.healthy && !priceRefreshHealth.stalled) || (!webhookWorkerHealth.healthy && !webhookWorkerHealth.stalled); - status = (!redisConnected || priceRefreshHealth.stalled || webhookWorkerHealth.stalled || database.status === 'error') - ? 'unhealthy' - : jobsDegraded ? 'degraded' : 'unhealthy'; + status = + !redisConnected || + priceRefreshHealth.stalled || + webhookWorkerHealth.stalled || + database.status === "error" + ? "unhealthy" + : jobsDegraded + ? "degraded" + : "unhealthy"; } res.json({ @@ -196,30 +220,31 @@ app.get('/health', async (req, res) => { }); }); -const apiKeyLimit = buildApiKeyRateLimit({ keyPrefix: 'apikey' }); +const apiKeyLimit = buildApiKeyRateLimit({ keyPrefix: "apikey" }); const globalApiLimit = buildRateLimit({ windowSeconds: Math.floor(config.rateLimit.windowMs / 1000), max: config.rateLimit.max, - keyPrefix: 'api', + keyPrefix: "api", }); // Resolve any presented API key first so the per-key limiter can meter it, // then fall through to the IP-keyed limiter for unauthenticated callers. // Authentication itself is still enforced per-route by requireApiKey. -app.use('/api/v1', attachApiKey()); -app.use('/api/v1', apiKeyLimit); -app.use('/api/v1', globalApiLimit); -app.use('/api/v1', pricesRouter); -app.use('/api/v1', keysRouter); -app.use('/api/v1/alerts', requireApiKey({ scopes: ['alerts'] })); -app.use('/api/v1', alertsRouter); -app.use('/api/v1', indexerRouter); -app.use('/api/v1/webhooks', requireApiKey({ scopes: ['webhooks'] })); -app.use('/api/v1', webhooksRouter); -app.use('/api/v1', airdropsRouter); -app.use('/api-docs', globalApiLimit); -app.use('/api-docs', apiDocsRouter); +app.use("/api/v1", attachApiKey()); +app.use("/api/v1", auditApiKeyUsage()); +app.use("/api/v1", apiKeyLimit); +app.use("/api/v1", globalApiLimit); +app.use("/api/v1", pricesRouter); +app.use("/api/v1", keysRouter); +app.use("/api/v1/alerts", requireApiKey({ scopes: ["alerts"] })); +app.use("/api/v1", alertsRouter); +app.use("/api/v1", indexerRouter); +app.use("/api/v1/webhooks", requireApiKey({ scopes: ["webhooks"] })); +app.use("/api/v1", webhooksRouter); +app.use("/api/v1", airdropsRouter); +app.use("/api-docs", globalApiLimit); +app.use("/api-docs", apiDocsRouter); app.use(metricsRouter); app.use(notFoundHandler); @@ -241,9 +266,12 @@ function shutdown(signal) { const remainingDeliveries = webhookDispatcher.getInFlightCount(); if (remainingDeliveries > 0) { - logger.warn('Shutdown complete with in-flight webhook deliveries still pending', { - remaining: remainingDeliveries, - }); + logger.warn( + "Shutdown complete with in-flight webhook deliveries still pending", + { + remaining: remainingDeliveries, + }, + ); } // Stop non-leader-elected services @@ -270,11 +298,11 @@ function sanitizeUrl(rawUrl) { if (!rawUrl) return null; try { const parsed = new URL(rawUrl); - if (parsed.password) parsed.password = '****'; - if (parsed.username) parsed.username = '****'; + if (parsed.password) parsed.password = "****"; + if (parsed.username) parsed.username = "****"; return parsed.toString(); } catch { - return '[unparseable]'; + return "[unparseable]"; } } @@ -326,12 +354,12 @@ async function startServer() { if (require.main === module) { startServer().catch((err) => { - logger.error('Startup failed', { error: err.message }); + logger.error("Startup failed", { error: err.message }); process.exit(1); }); - process.on('SIGTERM', shutdown('SIGTERM')); - process.on('SIGINT', shutdown('SIGINT')); + process.on("SIGTERM", shutdown("SIGTERM")); + process.on("SIGINT", shutdown("SIGINT")); // Last-resort safety net for errors that escape all per-job try/catch blocks. // These handlers do not replace the existing error handling in priceRefresh.js, @@ -340,22 +368,22 @@ if (require.main === module) { // unhandledRejection: Node >=20 exits by default; we match that behavior but // run the cleanup sequence first so Redis connections and in-flight jobs are // shut down cleanly rather than abandoned abruptly. - process.on('unhandledRejection', (reason) => { - logger.error('Unhandled promise rejection — initiating graceful shutdown', { + process.on("unhandledRejection", (reason) => { + logger.error("Unhandled promise rejection — initiating graceful shutdown", { reason: reason instanceof Error ? reason.message : String(reason), stack: reason instanceof Error ? reason.stack : undefined, }); - shutdown('unhandledRejection')(); + shutdown("unhandledRejection")(); }); // uncaughtException: the process heap is in an undefined state after this event. // Log and shut down; never swallow and continue running in a potentially corrupt state. - process.on('uncaughtException', (err) => { - logger.error('Uncaught exception — initiating graceful shutdown', { + process.on("uncaughtException", (err) => { + logger.error("Uncaught exception — initiating graceful shutdown", { error: err.message, stack: err.stack, }); - shutdown('uncaughtException')(); + shutdown("uncaughtException")(); }); } diff --git a/src/middleware/auth.js b/src/middleware/auth.js index 427fe3b..5c98bac 100644 --- a/src/middleware/auth.js +++ b/src/middleware/auth.js @@ -1,54 +1,76 @@ -const apiKeys = require('../services/apiKeys'); -const logger = require('../logger'); -const AppError = require('../errors/AppError'); +const apiKeys = require("../services/apiKeys"); +const apiKeyAuditLog = require("../services/apiKeyAuditLog"); +const logger = require("../logger"); +const AppError = require("../errors/AppError"); function extractBearerToken(header) { - if (!header || typeof header !== 'string') return null; + if (!header || typeof header !== "string") return null; const match = header.match(/^Bearer\s+(.+)$/i); return match ? match[1].trim() : null; } +function extractClientIp(req) { + // Check for IP from various headers (proxy, load balancer, etc.) + return ( + req.get("x-forwarded-for")?.split(",")[0].trim() || + req.get("x-client-ip") || + req.socket?.remoteAddress || + req.ip || + "unknown" + ); +} + function hasScopes(apiKey, requiredScopes) { if (!requiredScopes || !requiredScopes.length) return true; const scopes = new Set(apiKey.scopes || []); - if (scopes.has('admin')) return true; + if (scopes.has("admin")) return true; return requiredScopes.every((scope) => scopes.has(scope)); } function requireApiKey(options = {}) { const requiredScopes = Array.isArray(options) ? options - : typeof options === 'string' + : typeof options === "string" ? [options] : options.scopes || []; return async (req, res, next) => { - const token = extractBearerToken(req.get('authorization')); + const token = extractBearerToken(req.get("authorization")); if (!token) { - return next(new AppError('UNAUTHORIZED', 'Missing or invalid API key', 401)); + return next( + new AppError("UNAUTHORIZED", "Missing or invalid API key", 401), + ); } try { const apiKey = req.apiKey || (await apiKeys.validateApiKey(token)); if (!apiKey) { - logger.warn('Rejected API key authentication', { key_prefix: token.slice(0, 8) }); - return next(new AppError('UNAUTHORIZED', 'Missing or invalid API key', 401)); + logger.warn("Rejected API key authentication", { + key_prefix: token.slice(0, 8), + }); + return next( + new AppError("UNAUTHORIZED", "Missing or invalid API key", 401), + ); } if (!hasScopes(apiKey, requiredScopes)) { - logger.warn('Rejected API key due to insufficient scopes', { + logger.warn("Rejected API key due to insufficient scopes", { key_prefix: token.slice(0, 8), requiredScopes, actualScopes: apiKey.scopes, }); - return next(new AppError('FORBIDDEN', 'Insufficient API key scope', 403)); + return next( + new AppError("FORBIDDEN", "Insufficient API key scope", 403), + ); } req.apiKey = apiKey; return next(); } catch (err) { - logger.error('API key authentication failed', { error: err.message }); - return next(new AppError('UNAUTHORIZED', 'Missing or invalid API key', 401)); + logger.error("API key authentication failed", { error: err.message }); + return next( + new AppError("UNAUTHORIZED", "Missing or invalid API key", 401), + ); } }; } @@ -70,7 +92,7 @@ function attachApiKey() { return async (req, res, next) => { if (req.apiKey) return next(); - const token = extractBearerToken(req.get('authorization')); + const token = extractBearerToken(req.get("authorization")); if (!token) return next(); try { @@ -78,14 +100,46 @@ function attachApiKey() { if (apiKey) req.apiKey = apiKey; } catch (err) { // Never fail the request here — `requireApiKey` owns rejection. - logger.warn('Optional API key resolution failed', { error: err.message }); + logger.warn("Optional API key resolution failed", { error: err.message }); } return next(); }; } +/** + * Logs API key usage for audit trail after response is sent. + * Captures: endpoint, IP address, status code, response time + */ +function auditApiKeyUsage() { + return async (req, res, next) => { + const startTime = Date.now(); + + // Capture response finish to log after request completes + res.on("finish", () => { + if (!req.apiKey) return; // No API key used, skip logging + + const endpoint = `${req.method} ${req.path}`; + const ipAddress = extractClientIp(req); + const statusCode = res.statusCode; + const responseTimeMs = Date.now() - startTime; + + // Log asynchronously, don't wait for it + apiKeyAuditLog.logUsage({ + keyId: req.apiKey.id, + endpoint, + ipAddress, + statusCode, + responseTimeMs, + }); + }); + + return next(); + }; +} + module.exports = { requireApiKey, attachApiKey, + auditApiKeyUsage, extractBearerToken, }; diff --git a/src/routes/airdrops.js b/src/routes/airdrops.js index 42bfa9a..a3410ff 100644 --- a/src/routes/airdrops.js +++ b/src/routes/airdrops.js @@ -1,13 +1,14 @@ -const express = require('express'); -const multer = require('multer'); -const csv = require('csv-parser'); -const { Readable } = require('stream'); -const { pipeline } = require('stream/promises'); -const config = require('../config'); -const airdropsService = require('../services/airdrops'); -const logger = require('../logger'); -const AppError = require('../errors/AppError'); -const { flattenZodIssues, validate } = require('../middleware/validate'); +const express = require("express"); +const multer = require("multer"); +const csv = require("csv-parser"); +const { Readable } = require("stream"); +const { pipeline } = require("stream/promises"); +const config = require("../config"); +const airdropsService = require("../services/airdrops"); +const { idempotencyMiddleware } = require("../services/idempotency"); +const logger = require("../logger"); +const AppError = require("../errors/AppError"); +const { flattenZodIssues, validate } = require("../middleware/validate"); const { airdropCreateBodySchema, airdropRecipientsBodySchema, @@ -15,11 +16,11 @@ const { paginationQuerySchema, recipientsSchema, routeIdParamsSchema, -} = require('../validation/schemas'); -const buildRateLimit = require('../middleware/rateLimit'); -const { routeTimeout } = require('../middleware/timeout'); -const { StrKey } = require('stellar-sdk'); -const { paginateResponse } = require('../utils/paginate'); +} = require("../validation/schemas"); +const buildRateLimit = require("../middleware/rateLimit"); +const { routeTimeout } = require("../middleware/timeout"); +const { StrKey } = require("stellar-sdk"); +const { paginateResponse } = require("../utils/paginate"); // Stellar Int64 max in stroops (1 unit = 10_000_000 stroops for XLM/USDC) const INT64_MAX_STROOPS = 9223372036854775807n; @@ -34,8 +35,8 @@ const upload = multer({ storage: multer.memoryStorage(), limits: { fileSize: config.airdrops.csvMaxBytes }, }); -const validateRouteIdParams = validate(routeIdParamsSchema, 'params'); -const validatePaginationQuery = validate(paginationQuerySchema, 'query'); +const validateRouteIdParams = validate(routeIdParamsSchema, "params"); +const validatePaginationQuery = validate(paginationQuerySchema, "query"); const validateRecipientBody = validate(airdropRecipientsBodySchema); function validateWithCurrentLedger(schemaFactory) { @@ -44,7 +45,7 @@ function validateWithCurrentLedger(schemaFactory) { const currentLedger = await airdropsService.getCurrentLedger(); return validate(schemaFactory(currentLedger))(req, res, next); } catch (err) { - logger.error('Airdrop validation error', { error: err.message }); + logger.error("Airdrop validation error", { error: err.message }); return next(err); } }; @@ -53,24 +54,26 @@ function validateWithCurrentLedger(schemaFactory) { const createAirdropLimit = buildRateLimit({ windowSeconds: config.airdrops.rateLimit.windowSeconds, max: config.airdrops.rateLimit.max, - keyPrefix: 'airdrops_create', + keyPrefix: "airdrops_create", }); const addRecipientsLimit = buildRateLimit({ windowSeconds: config.airdrops.rateLimit.windowSeconds, max: config.airdrops.rateLimit.max, - keyPrefix: 'airdrops_recipients', + keyPrefix: "airdrops_recipients", }); function uploadRecipientsFile(req, res, next) { - upload.single('file')(req, res, (err) => { - if (err instanceof multer.MulterError && err.code === 'LIMIT_FILE_SIZE') { - return next(new AppError( - 'PAYLOAD_TOO_LARGE', - `CSV file cannot exceed ${config.airdrops.csvMaxBytes} bytes`, - 413, - { max_bytes: config.airdrops.csvMaxBytes } - )); + upload.single("file")(req, res, (err) => { + if (err instanceof multer.MulterError && err.code === "LIMIT_FILE_SIZE") { + return next( + new AppError( + "PAYLOAD_TOO_LARGE", + `CSV file cannot exceed ${config.airdrops.csvMaxBytes} bytes`, + 413, + { max_bytes: config.airdrops.csvMaxBytes }, + ), + ); } return next(err); }); @@ -90,34 +93,44 @@ function toStroops(amount) { function assertWithinCeiling(stroops, label) { if (stroops > INT64_MAX_STROOPS) { - throw new AppError('VALIDATION_ERROR', `${label} exceeds Stellar Int64 ceiling`, 400); + throw new AppError( + "VALIDATION_ERROR", + `${label} exceeds Stellar Int64 ceiling`, + 400, + ); } } function parseRecipients(recipients, next) { const result = recipientsSchema.safeParse(recipients); if (!result.success) { - return next(new AppError('VALIDATION_ERROR', 'Validation failed', 400, { - fields: flattenZodIssues(result.error), - })); + return next( + new AppError("VALIDATION_ERROR", "Validation failed", 400, { + fields: flattenZodIssues(result.error), + }), + ); } return result.data; } function validateUtf8(buffer) { try { - const decoder = new TextDecoder('utf-8', { fatal: true }); + const decoder = new TextDecoder("utf-8", { fatal: true }); decoder.decode(buffer); } catch { - throw new AppError('CSV_INVALID_ENCODING', 'CSV file must be valid UTF-8 encoded', 400); + throw new AppError( + "CSV_INVALID_ENCODING", + "CSV file must be valid UTF-8 encoded", + 400, + ); } } // Accepted spellings for the two required columns. csv-parser hands back // header names verbatim, so case and surrounding whitespace are normalized // here rather than enumerating every variant at each lookup. -const ADDRESS_COLUMN = 'address'; -const AMOUNT_COLUMN = 'amount'; +const ADDRESS_COLUMN = "address"; +const AMOUNT_COLUMN = "amount"; function normalizeRow(row) { const normalized = {}; @@ -148,10 +161,14 @@ async function parseCSV(buffer) { let headerChecked = false; const chunks = (function* chunkBuffer() { - for (let offset = 0; offset < buffer.length; offset += CSV_PARSE_CHUNK_BYTES) { + for ( + let offset = 0; + offset < buffer.length; + offset += CSV_PARSE_CHUNK_BYTES + ) { yield buffer.subarray(offset, offset + CSV_PARSE_CHUNK_BYTES); } - }()); + })(); // Throwing out of the pipeline consumer while the source still has data // makes stream/promises reject with an AbortError, discarding the original @@ -173,25 +190,35 @@ async function parseCSV(buffer) { // the rest of the file is not worth parsing at all. if (!headerChecked) { headerChecked = true; - const missing = [ADDRESS_COLUMN, AMOUNT_COLUMN] - .filter((column) => !Object.prototype.hasOwnProperty.call(row, column)); + const missing = [ADDRESS_COLUMN, AMOUNT_COLUMN].filter( + (column) => !Object.prototype.hasOwnProperty.call(row, column), + ); if (missing.length > 0) { - throw stopWith(new AppError('CSV_MISSING_COLUMNS', 'CSV is missing required columns', 400, { - missing_columns: missing, - required_columns: [ADDRESS_COLUMN, AMOUNT_COLUMN], - found_columns: Object.keys(row), - })); + throw stopWith( + new AppError( + "CSV_MISSING_COLUMNS", + "CSV is missing required columns", + 400, + { + missing_columns: missing, + required_columns: [ADDRESS_COLUMN, AMOUNT_COLUMN], + found_columns: Object.keys(row), + }, + ), + ); } } rowCount += 1; if (rowCount > config.airdrops.maxRecipients) { - throw stopWith(new AppError( - 'RECIPIENT_LIMIT_EXCEEDED', - `CSV cannot exceed ${config.airdrops.maxRecipients} recipients`, - 400, - { max_recipients: config.airdrops.maxRecipients }, - )); + throw stopWith( + new AppError( + "RECIPIENT_LIMIT_EXCEEDED", + `CSV cannot exceed ${config.airdrops.maxRecipients} recipients`, + 400, + { max_recipients: config.airdrops.maxRecipients }, + ), + ); } const address = row[ADDRESS_COLUMN]; @@ -201,12 +228,15 @@ async function parseCSV(buffer) { // +1 for the header line, so the number matches what the uploader // sees in a text editor. const line = rowCount + 1; - if (!address || String(address).trim() === '') { - invalidRows.push({ line, reason: 'missing address' }); + if (!address || String(address).trim() === "") { + invalidRows.push({ line, reason: "missing address" }); } else if (!Number.isFinite(amount)) { - invalidRows.push({ line, reason: 'amount is not a number' }); + invalidRows.push({ line, reason: "amount is not a number" }); } else if (amount <= 0) { - invalidRows.push({ line, reason: 'amount must be greater than zero' }); + invalidRows.push({ + line, + reason: "amount must be greater than zero", + }); } else { results.push({ address: String(address).trim(), amount }); } @@ -214,10 +244,17 @@ async function parseCSV(buffer) { // Bail out early rather than accumulating an unbounded error list for // a file that is clearly not going to be accepted. if (invalidRows.length > MAX_REPORTED_INVALID_ROWS) { - throw stopWith(new AppError('CSV_MALFORMED', 'CSV contains too many invalid rows', 400, { - invalid_rows: invalidRows.slice(0, MAX_REPORTED_INVALID_ROWS), - truncated: true, - })); + throw stopWith( + new AppError( + "CSV_MALFORMED", + "CSV contains too many invalid rows", + 400, + { + invalid_rows: invalidRows.slice(0, MAX_REPORTED_INVALID_ROWS), + truncated: true, + }, + ), + ); } } }); @@ -230,11 +267,11 @@ async function parseCSV(buffer) { } if (rowCount === 0) { - throw new AppError('CSV_EMPTY', 'CSV contains no data rows', 400); + throw new AppError("CSV_EMPTY", "CSV contains no data rows", 400); } if (invalidRows.length > 0) { - throw new AppError('CSV_MALFORMED', 'CSV contains invalid rows', 400, { + throw new AppError("CSV_MALFORMED", "CSV contains invalid rows", 400, { invalid_rows: invalidRows, valid_rows: results.length, total_rows: rowCount, @@ -244,151 +281,244 @@ async function parseCSV(buffer) { return results; } -router.post('/airdrops', routeTimeout(), createAirdropLimit, validateWithCurrentLedger(airdropCreateBodySchema), async (req, res, next) => { - try { - const airdrop = await airdropsService.create(req.validated.body); - return res.status(201).json(airdrop); - } catch (err) { - logger.error('Create airdrop error', { error: err.message }); - return next(err); - } -}); +router.post( + "/airdrops", + routeTimeout(), + createAirdropLimit, + idempotencyMiddleware("airdrop"), + validateWithCurrentLedger(airdropCreateBodySchema), + async (req, res, next) => { + try { + const airdrop = await airdropsService.create(req.validated.body); + return res.status(201).json(airdrop); + } catch (err) { + logger.error("Create airdrop error", { error: err.message }); + return next(err); + } + }, +); -router.get('/airdrops', validatePaginationQuery, async (req, res, next) => { +router.get("/airdrops", validatePaginationQuery, async (req, res, next) => { try { const { page, limit } = req.validated.query; const result = await airdropsService.list(page, limit); - return res.json(paginateResponse(result.airdrops, result.total, { page, limit })); + return res.json( + paginateResponse(result.airdrops, result.total, { page, limit }), + ); } catch (err) { - logger.error('List airdrops error', { error: err.message }); + logger.error("List airdrops error", { error: err.message }); return next(err); } }); -router.get('/airdrops/:id', validateRouteIdParams, async (req, res, next) => { +router.get("/airdrops/:id", validateRouteIdParams, async (req, res, next) => { try { const airdrop = await airdropsService.get(req.params.id); if (!airdrop) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); + return next(new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404)); } return res.json(airdrop); } catch (err) { - logger.error('Get airdrop error', { error: err.message }); + logger.error("Get airdrop error", { error: err.message }); return next(err); } }); -router.patch('/airdrops/:id', validateRouteIdParams, validateWithCurrentLedger(airdropUpdateBodySchema), async (req, res, next) => { - try { - const airdrop = await airdropsService.update(req.params.id, req.validated.body); - if (!airdrop) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); - } - return res.json(airdrop); - } catch (err) { - logger.error('Update airdrop error', { error: err.message }); - return next(err); - } -}); - -router.delete('/airdrops/:id', validateRouteIdParams, async (req, res, next) => { - try { - const deleted = await airdropsService.remove(req.params.id); - if (!deleted) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); - } - return res.json({ deleted: true, id: req.params.id }); - } catch (err) { - logger.error('Delete airdrop error', { error: err.message }); - return next(err); - } -}); - -router.post('/airdrops/:id/cancel', validateRouteIdParams, async (req, res, next) => { - try { - const airdrop = await airdropsService.cancel(req.params.id); - if (!airdrop) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); +router.patch( + "/airdrops/:id", + validateRouteIdParams, + validateWithCurrentLedger(airdropUpdateBodySchema), + async (req, res, next) => { + try { + const airdrop = await airdropsService.update( + req.params.id, + req.validated.body, + ); + if (!airdrop) { + return next( + new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404), + ); + } + return res.json(airdrop); + } catch (err) { + logger.error("Update airdrop error", { error: err.message }); + return next(err); } - return res.json(airdrop); - } catch (err) { - logger.error('Cancel airdrop error', { error: err.message }); - return next(err); - } -}); + }, +); -router.post('/airdrops/:id/recipients', routeTimeout(), validateRouteIdParams, addRecipientsLimit, uploadRecipientsFile, validateRecipientBody, async (req, res, next) => { - try { - const airdrop = await airdropsService.get(req.params.id); - if (!airdrop) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); +router.delete( + "/airdrops/:id", + validateRouteIdParams, + async (req, res, next) => { + try { + const deleted = await airdropsService.remove(req.params.id); + if (!deleted) { + return next( + new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404), + ); + } + return res.json({ deleted: true, id: req.params.id }); + } catch (err) { + logger.error("Delete airdrop error", { error: err.message }); + return next(err); } + }, +); - let recipients = []; - if (req.file) { - recipients = await parseCSV(req.file.buffer); - recipients = parseRecipients(recipients, next); - if (!recipients) return undefined; - } else if (req.validated.body.recipients) { - recipients = req.validated.body.recipients; - } else { - return next(new AppError('VALIDATION_ERROR', 'recipients or file is required', 400)); +router.post( + "/airdrops/:id/cancel", + validateRouteIdParams, + async (req, res, next) => { + try { + const airdrop = await airdropsService.cancel(req.params.id); + if (!airdrop) { + return next( + new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404), + ); + } + return res.json(airdrop); + } catch (err) { + logger.error("Cancel airdrop error", { error: err.message }); + return next(err); } + }, +); + +router.post( + "/airdrops/:id/recipients", + routeTimeout(), + validateRouteIdParams, + addRecipientsLimit, + uploadRecipientsFile, + validateRecipientBody, + async (req, res, next) => { + try { + const airdrop = await airdropsService.get(req.params.id); + if (!airdrop) { + return next( + new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404), + ); + } - if (recipients.length > config.airdrops.maxRecipients) { - return next(new AppError('VALIDATION_ERROR', 'recipients cannot exceed 10,000', 400)); - } + let recipients = []; + if (req.file) { + recipients = await parseCSV(req.file.buffer); + recipients = parseRecipients(recipients, next); + if (!recipients) return undefined; + } else if (req.validated.body.recipients) { + recipients = req.validated.body.recipients; + } else { + return next( + new AppError( + "VALIDATION_ERROR", + "recipients or file is required", + 400, + ), + ); + } - const recipientSet = new Set(); - let sum = 0n; - for (let i = 0; i < recipients.length; i++) { - const r = recipients[i]; - if (!r.address || !isValidStellarAddress(r.address)) { - return next(new AppError('VALIDATION_ERROR', `recipient ${i}: invalid Stellar address`, 400)); + if (recipients.length > config.airdrops.maxRecipients) { + return next( + new AppError( + "VALIDATION_ERROR", + "recipients cannot exceed 10,000", + 400, + ), + ); } - if (recipientSet.has(r.address)) { - return next(new AppError('VALIDATION_ERROR', `recipient ${i}: duplicate address ${r.address}`, 400)); + + const recipientSet = new Set(); + let sum = 0n; + for (let i = 0; i < recipients.length; i++) { + const r = recipients[i]; + if (!r.address || !isValidStellarAddress(r.address)) { + return next( + new AppError( + "VALIDATION_ERROR", + `recipient ${i}: invalid Stellar address`, + 400, + ), + ); + } + if (recipientSet.has(r.address)) { + return next( + new AppError( + "VALIDATION_ERROR", + `recipient ${i}: duplicate address ${r.address}`, + 400, + ), + ); + } + recipientSet.add(r.address); + if ( + typeof r.amount !== "number" || + r.amount <= 0 || + !Number.isFinite(r.amount) + ) { + return next( + new AppError( + "VALIDATION_ERROR", + `recipient ${i}: amount must be a positive number`, + 400, + ), + ); + } + const stroops = toStroops(r.amount); + assertWithinCeiling(stroops, `recipient ${i} amount`); + sum += stroops; } - recipientSet.add(r.address); - if (typeof r.amount !== 'number' || r.amount <= 0 || !Number.isFinite(r.amount)) { - return next(new AppError('VALIDATION_ERROR', `recipient ${i}: amount must be a positive number`, 400)); + assertWithinCeiling(sum, "total recipient amount"); + + const duplicates = await airdropsService.addRecipients( + req.params.id, + recipients, + ); + if (duplicates.length > 0) { + return next( + new AppError( + "CONFLICT", + "One or more recipient addresses are already registered for this airdrop", + 409, + { duplicate_addresses: duplicates }, + ), + ); } - const stroops = toStroops(r.amount); - assertWithinCeiling(stroops, `recipient ${i} amount`); - sum += stroops; - } - assertWithinCeiling(sum, 'total recipient amount'); - - const duplicates = await airdropsService.addRecipients(req.params.id, recipients); - if (duplicates.length > 0) { - return next(new AppError( - 'CONFLICT', - 'One or more recipient addresses are already registered for this airdrop', - 409, - { duplicate_addresses: duplicates }, - )); + return res.status(201).json({ added: recipients.length }); + } catch (err) { + logger.error("Add recipients error", { error: err.message }); + return next(err); } - return res.status(201).json({ added: recipients.length }); - } catch (err) { - logger.error('Add recipients error', { error: err.message }); - return next(err); - } -}); + }, +); + +router.get( + "/airdrops/:id/recipients", + validateRouteIdParams, + validatePaginationQuery, + async (req, res, next) => { + try { + const airdrop = await airdropsService.get(req.params.id); + if (!airdrop) { + return next( + new AppError("AIRDROP_NOT_FOUND", "Airdrop not found", 404), + ); + } -router.get('/airdrops/:id/recipients', validateRouteIdParams, validatePaginationQuery, async (req, res, next) => { - try { - const airdrop = await airdropsService.get(req.params.id); - if (!airdrop) { - return next(new AppError('AIRDROP_NOT_FOUND', 'Airdrop not found', 404)); + const { page, limit } = req.validated.query; + const result = await airdropsService.listRecipients( + req.params.id, + page, + limit, + ); + return res.json( + paginateResponse(result.recipients, result.total, { page, limit }), + ); + } catch (err) { + logger.error("List recipients error", { error: err.message }); + return next(err); } - - const { page, limit } = req.validated.query; - const result = await airdropsService.listRecipients(req.params.id, page, limit); - return res.json(paginateResponse(result.recipients, result.total, { page, limit })); - } catch (err) { - logger.error('List recipients error', { error: err.message }); - return next(err); - } -}); + }, +); module.exports = router; diff --git a/src/routes/keys.js b/src/routes/keys.js index 55221e9..87b669f 100644 --- a/src/routes/keys.js +++ b/src/routes/keys.js @@ -1,53 +1,79 @@ -const express = require('express'); -const { requireApiKey } = require('../middleware/auth'); -const { validate } = require('../middleware/validate'); -const apiKeys = require('../services/apiKeys'); -const logger = require('../logger'); -const AppError = require('../errors/AppError'); -const { keyCreateBodySchema, routeIdParamsSchema } = require('../validation/schemas'); +const express = require("express"); +const { requireApiKey } = require("../middleware/auth"); +const { validate } = require("../middleware/validate"); +const apiKeys = require("../services/apiKeys"); +const logger = require("../logger"); +const AppError = require("../errors/AppError"); +const { + keyCreateBodySchema, + keyRotateBodySchema, + routeIdParamsSchema, +} = require("../validation/schemas"); const router = express.Router(); -const validateRouteIdParams = validate(routeIdParamsSchema, 'params'); +const validateRouteIdParams = validate(routeIdParamsSchema, "params"); -router.use('/keys', requireApiKey({ scopes: ['admin'] })); +router.use("/keys", requireApiKey({ scopes: ["admin"] })); -router.get('/keys', async (_req, res, next) => { +router.get("/keys", async (_req, res, next) => { try { const keys = await apiKeys.listKeys(); return res.json({ keys }); } catch (err) { - logger.error('List API keys error', { error: err.message }); + logger.error("List API keys error", { error: err.message }); return next(err); } }); -router.post('/keys', validate(keyCreateBodySchema), async (req, res, next) => { +router.post("/keys", validate(keyCreateBodySchema), async (req, res, next) => { try { const { label, scopes, tier } = req.validated.body; const created = await apiKeys.createKey({ label, - scopes: scopes || ['default'], + scopes: scopes || ["default"], tier, }); return res.status(201).json(created); } catch (err) { - logger.error('Create API key error', { error: err.message }); + logger.error("Create API key error", { error: err.message }); return next(err); } }); -router.delete('/keys/:id', validateRouteIdParams, async (req, res, next) => { +router.delete("/keys/:id", validateRouteIdParams, async (req, res, next) => { try { const deleted = await apiKeys.revokeKey(req.params.id); if (!deleted) { - return next(new AppError('API_KEY_NOT_FOUND', 'API key not found', 404)); + return next(new AppError("API_KEY_NOT_FOUND", "API key not found", 404)); } return res.json({ deleted: true, key: deleted }); } catch (err) { - logger.error('Revoke API key error', { error: err.message }); + logger.error("Revoke API key error", { error: err.message }); return next(err); } }); +router.post( + "/keys/:id/rotate", + validateRouteIdParams, + validate(keyRotateBodySchema), + async (req, res, next) => { + try { + const { tier } = req.validated.body; + + const rotated = await apiKeys.rotateKey(req.params.id, { tier }); + if (!rotated) { + return next( + new AppError("API_KEY_NOT_FOUND", "API key not found", 404), + ); + } + return res.status(201).json(rotated); + } catch (err) { + logger.error("Rotate API key error", { error: err.message }); + return next(err); + } + }, +); + module.exports = router; diff --git a/src/routes/webhooks.js b/src/routes/webhooks.js index 3f6959f..a399bab 100644 --- a/src/routes/webhooks.js +++ b/src/routes/webhooks.js @@ -1,56 +1,65 @@ -'use strict'; - -const express = require('express'); -const config = require('../config'); -const { validate } = require('../middleware/validate'); -const webhookRepo = require('../repositories/webhookRepository'); -const deliveryRepo = require('../repositories/deliveryRepository'); -const dispatcher = require('../services/webhookDispatcher'); -const signatureService = require('../services/webhookSignature'); -const { probeReachability } = require('../services/webhook'); -const buildRateLimit = require('../middleware/rateLimit'); -const { routeTimeout } = require('../middleware/timeout'); -const AppError = require('../errors/AppError'); -const { paginateResponse } = require('../utils/paginate'); +"use strict"; + +const express = require("express"); +const config = require("../config"); +const { validate } = require("../middleware/validate"); +const webhookRepo = require("../repositories/webhookRepository"); +const deliveryRepo = require("../repositories/deliveryRepository"); +const dispatcher = require("../services/webhookDispatcher"); +const signatureService = require("../services/webhookSignature"); +const { probeReachability } = require("../services/webhook"); +const { idempotencyMiddleware } = require("../services/idempotency"); +const buildRateLimit = require("../middleware/rateLimit"); +const { routeTimeout } = require("../middleware/timeout"); +const AppError = require("../errors/AppError"); +const { paginateResponse } = require("../utils/paginate"); const { paginationQuerySchema, routeIdParamsSchema, webhookCreateBodySchema, webhookDeliveriesQuerySchema, webhookPatchBodySchema, -} = require('../validation/schemas'); +} = require("../validation/schemas"); const router = express.Router(); router.use(express.json({ limit: config.webhooks.jsonMaxBytes })); -const validateRouteIdParams = validate(routeIdParamsSchema, 'params'); -const validatePaginationQuery = validate(paginationQuerySchema, 'query'); +const validateRouteIdParams = validate(routeIdParamsSchema, "params"); +const validatePaginationQuery = validate(paginationQuerySchema, "query"); const manageLimit = buildRateLimit({ windowSeconds: config.webhooks.rateLimit.windowSeconds, max: config.webhooks.rateLimit.max, - keyPrefix: 'webhooks', + keyPrefix: "webhooks", }); const testLimit = buildRateLimit({ windowSeconds: config.webhooks.testRateLimit.windowSeconds, max: config.webhooks.testRateLimit.max, - keyPrefix: 'webhooks_test', + keyPrefix: "webhooks_test", }); function clientIpFromRequest(req) { - const forwardedFor = req.headers['x-forwarded-for']; - if (typeof forwardedFor === 'string' && forwardedFor.trim()) { - return forwardedFor.split(',')[0].trim().replace(/^::ffff:/, ''); + const forwardedFor = req.headers["x-forwarded-for"]; + if (typeof forwardedFor === "string" && forwardedFor.trim()) { + return forwardedFor + .split(",")[0] + .trim() + .replace(/^::ffff:/, ""); } if (Array.isArray(forwardedFor) && forwardedFor[0]) { - return String(forwardedFor[0]).trim().replace(/^::ffff:/, ''); + return String(forwardedFor[0]) + .trim() + .replace(/^::ffff:/, ""); } - return (req.ip || req.socket?.remoteAddress || 'unknown').replace(/^::ffff:/, ''); + return (req.ip || req.socket?.remoteAddress || "unknown").replace( + /^::ffff:/, + "", + ); } -router.use('/webhooks', manageLimit); +router.use("/webhooks", manageLimit); -router.get('/webhooks/metrics', async (req, res) => { +router.get("/webhooks/metrics", async (req, res) => { return res.json(dispatcher.getMetrics()); }); @@ -69,121 +78,169 @@ function publicView(webhook) { }; } -router.post('/webhooks', routeTimeout(), validate(webhookCreateBodySchema), async (req, res, next) => { - try { - const body = req.validated.body; - const ownerIp = clientIpFromRequest(req); - const existingCount = await webhookRepo.countByOwner(ownerIp); - if (existingCount >= config.webhooks.maxPerSubscriber) { - // Distinct from RATE_LIMITED: this is a standing quota on how many - // webhooks a subscriber may own, not a request rate. Waiting and - // retrying will never clear it — the client must delete a webhook. - // owner_ip is deliberately not echoed back in the response details. - return next(new AppError( - 'WEBHOOK_LIMIT_EXCEEDED', - `Webhook limit of ${config.webhooks.maxPerSubscriber} per subscriber exceeded`, - 429, - { limit: config.webhooks.maxPerSubscriber, current: existingCount }, - )); - } - - const secret = body.secret || signatureService.generateSecret(); - const reachability = await probeReachability(body.url); - const webhook = await webhookRepo.create({ - url: body.url, - events: body.events, - secret, - description: body.description, - filters: body.filters, - owner_ip: ownerIp, - }); - - const response = { - ...publicView(webhook), - secret, - secret_warning: 'Store this secret now — it will not be shown again in plaintext.', - reachability: reachability.reachable ? 'reachable' : 'unreachable', - }; - - if (!reachability.reachable) { - response.warning = `Webhook target is unreachable during registration: ${reachability.error || 'request failed'}`; +router.post( + "/webhooks", + routeTimeout(), + idempotencyMiddleware("webhook"), + validate(webhookCreateBodySchema), + async (req, res, next) => { + try { + const body = req.validated.body; + const ownerIp = clientIpFromRequest(req); + const existingCount = await webhookRepo.countByOwner(ownerIp); + if (existingCount >= config.webhooks.maxPerSubscriber) { + // Distinct from RATE_LIMITED: this is a standing quota on how many + // webhooks a subscriber may own, not a request rate. Waiting and + // retrying will never clear it — the client must delete a webhook. + // owner_ip is deliberately not echoed back in the response details. + return next( + new AppError( + "WEBHOOK_LIMIT_EXCEEDED", + `Webhook limit of ${config.webhooks.maxPerSubscriber} per subscriber exceeded`, + 429, + { limit: config.webhooks.maxPerSubscriber, current: existingCount }, + ), + ); + } + + const secret = body.secret || signatureService.generateSecret(); + const reachability = await probeReachability(body.url); + const webhook = await webhookRepo.create({ + url: body.url, + events: body.events, + secret, + description: body.description, + filters: body.filters, + owner_ip: ownerIp, + }); + + const response = { + ...publicView(webhook), + secret, + secret_warning: + "Store this secret now — it will not be shown again in plaintext.", + reachability: reachability.reachable ? "reachable" : "unreachable", + }; + + if (!reachability.reachable) { + response.warning = `Webhook target is unreachable during registration: ${reachability.error || "request failed"}`; + } + + return res.status(201).json(response); + } catch (err) { + return next(err); } + }, +); - return res.status(201).json(response); - } catch (err) { - return next(err); - } -}); - -router.get('/webhooks', validatePaginationQuery, async (req, res, next) => { +router.get("/webhooks", validatePaginationQuery, async (req, res, next) => { try { const { page, limit } = req.validated.query; const result = await webhookRepo.list(page, limit); return res.json( - paginateResponse(result.webhooks.map(publicView), result.total, { page, limit }), + paginateResponse(result.webhooks.map(publicView), result.total, { + page, + limit, + }), ); } catch (err) { return next(err); } }); -router.get('/webhooks/:id', validateRouteIdParams, async (req, res, next) => { +router.get("/webhooks/:id", validateRouteIdParams, async (req, res, next) => { try { const webhook = await webhookRepo.findById(req.params.id); - if (!webhook) return next(new AppError('WEBHOOK_NOT_FOUND', 'Webhook not found', 404)); + if (!webhook) + return next(new AppError("WEBHOOK_NOT_FOUND", "Webhook not found", 404)); return res.json(publicView(webhook)); } catch (err) { return next(err); } }); -router.patch('/webhooks/:id', validateRouteIdParams, validate(webhookPatchBodySchema), async (req, res, next) => { - try { - const patch = req.validated.body; - const updated = await webhookRepo.update(req.params.id, patch); - if (!updated) return next(new AppError('WEBHOOK_NOT_FOUND', 'Webhook not found', 404)); - return res.json(publicView(updated)); - } catch (err) { - return next(err); - } -}); - -router.delete('/webhooks/:id', validateRouteIdParams, async (req, res, next) => { - try { - const deleted = await webhookRepo.remove(req.params.id); - if (!deleted) return next(new AppError('WEBHOOK_NOT_FOUND', 'Webhook not found', 404)); - return res.json({ deleted: true, id: req.params.id }); - } catch (err) { - return next(err); - } -}); - -router.post('/webhooks/:id/test', routeTimeout(), validateRouteIdParams, testLimit, async (req, res, next) => { - try { - const delivery = await dispatcher.sendTest(req.params.id); - if (!delivery) return next(new AppError('WEBHOOK_NOT_FOUND', 'Webhook not found', 404)); - return res.status(202).json({ - delivery_id: delivery.id, - status: delivery.status, - attempts: delivery.attempts, - response_status: delivery.response_status, - last_error: delivery.last_error, - }); - } catch (err) { - return next(err); - } -}); - -router.get('/webhooks/:id/deliveries', validateRouteIdParams, validate(webhookDeliveriesQuerySchema, 'query'), async (req, res, next) => { - try { - const webhook = await webhookRepo.findById(req.params.id); - if (!webhook) return next(new AppError('WEBHOOK_NOT_FOUND', 'Webhook not found', 404)); - const { limit, status } = req.validated.query; - const deliveries = await deliveryRepo.listByWebhook(req.params.id, { limit, status }); - return res.json({ deliveries }); - } catch (err) { - return next(err); - } -}); +router.patch( + "/webhooks/:id", + validateRouteIdParams, + validate(webhookPatchBodySchema), + async (req, res, next) => { + try { + const patch = req.validated.body; + const updated = await webhookRepo.update(req.params.id, patch); + if (!updated) + return next( + new AppError("WEBHOOK_NOT_FOUND", "Webhook not found", 404), + ); + return res.json(publicView(updated)); + } catch (err) { + return next(err); + } + }, +); + +router.delete( + "/webhooks/:id", + validateRouteIdParams, + async (req, res, next) => { + try { + const deleted = await webhookRepo.remove(req.params.id); + if (!deleted) + return next( + new AppError("WEBHOOK_NOT_FOUND", "Webhook not found", 404), + ); + return res.json({ deleted: true, id: req.params.id }); + } catch (err) { + return next(err); + } + }, +); + +router.post( + "/webhooks/:id/test", + routeTimeout(), + validateRouteIdParams, + testLimit, + async (req, res, next) => { + try { + const delivery = await dispatcher.sendTest(req.params.id); + if (!delivery) + return next( + new AppError("WEBHOOK_NOT_FOUND", "Webhook not found", 404), + ); + return res.status(202).json({ + delivery_id: delivery.id, + status: delivery.status, + attempts: delivery.attempts, + response_status: delivery.response_status, + last_error: delivery.last_error, + }); + } catch (err) { + return next(err); + } + }, +); + +router.get( + "/webhooks/:id/deliveries", + validateRouteIdParams, + validate(webhookDeliveriesQuerySchema, "query"), + async (req, res, next) => { + try { + const webhook = await webhookRepo.findById(req.params.id); + if (!webhook) + return next( + new AppError("WEBHOOK_NOT_FOUND", "Webhook not found", 404), + ); + const { limit, status } = req.validated.query; + const deliveries = await deliveryRepo.listByWebhook(req.params.id, { + limit, + status, + }); + return res.json({ deliveries }); + } catch (err) { + return next(err); + } + }, +); module.exports = router; diff --git a/src/services/apiKeyAuditLog.js b/src/services/apiKeyAuditLog.js new file mode 100644 index 0000000..46eee43 --- /dev/null +++ b/src/services/apiKeyAuditLog.js @@ -0,0 +1,100 @@ +const knex = require('knex'); +const config = require('../config'); +const logger = require('../logger'); + +let db = null; + +function getDb() { + if (!db) { + db = knex({ + client: 'pg', + connection: config.databaseUrl, + pool: { min: 2, max: 10 }, + }); + } + return db; +} + +/** + * Log API key usage to audit trail + * + * @param {object} options + * @param {string} options.keyId - The API key ID + * @param {string} options.endpoint - The endpoint accessed (e.g., "GET /api/prices") + * @param {string} options.ipAddress - Client IP address + * @param {number} options.statusCode - HTTP response status code + * @param {number} options.responseTimeMs - Request duration in milliseconds + */ +async function logUsage({ keyId, endpoint, ipAddress, statusCode, responseTimeMs }) { + try { + if (!keyId || keyId === 'admin') { + // Skip logging for admin key or missing key + return; + } + + await getDb()('api_key_audit_logs').insert({ + key_id: keyId, + endpoint, + ip_address: ipAddress, + status_code: statusCode, + response_time_ms: responseTimeMs, + created_at: new Date(), + }); + } catch (err) { + // Log the error but don't fail the request + logger.error('Failed to log API key audit trail', { + keyId, + endpoint, + error: err.message, + }); + } +} + +/** + * Get audit log entries for a specific API key + * + * @param {string} keyId - The API key ID + * @param {number} limit - Maximum number of entries to return + * @param {number} offset - Number of entries to skip + * @returns {Promise} Audit log entries + */ +async function getKeyAuditLog(keyId, limit = 100, offset = 0) { + try { + return await getDb()('api_key_audit_logs') + .where('key_id', keyId) + .orderBy('created_at', 'desc') + .limit(limit) + .offset(offset); + } catch (err) { + logger.error('Failed to fetch API key audit log', { + keyId, + error: err.message, + }); + return []; + } +} + +/** + * Get recent audit log entries for all keys (admin only) + * + * @param {number} limit - Maximum number of entries to return + * @param {number} offset - Number of entries to skip + * @returns {Promise} Audit log entries + */ +async function getAllAuditLogs(limit = 100, offset = 0) { + try { + return await getDb()('api_key_audit_logs') + .orderBy('created_at', 'desc') + .limit(limit) + .offset(offset); + } catch (err) { + logger.error('Failed to fetch all API key audit logs', { error: err.message }); + return []; + } +} + +module.exports = { + logUsage, + getKeyAuditLog, + getAllAuditLogs, +}; diff --git a/src/services/apiKeys.js b/src/services/apiKeys.js index b7fa54c..f3df5f2 100644 --- a/src/services/apiKeys.js +++ b/src/services/apiKeys.js @@ -1,18 +1,18 @@ -const crypto = require('crypto'); -const cache = require('./cache'); -const config = require('../config'); +const crypto = require("crypto"); +const cache = require("./cache"); +const config = require("../config"); -const KEY_PREFIX = 'api_key:'; -const HASH_PREFIX = 'api_key_hash:'; -const IDS_KEY = 'api_keys'; +const KEY_PREFIX = "api_key:"; +const HASH_PREFIX = "api_key_hash:"; +const IDS_KEY = "api_keys"; function hashApiKey(apiKey) { - return crypto.createHash('sha256').update(apiKey).digest('hex'); + return crypto.createHash("sha256").update(apiKey).digest("hex"); } function constantTimeSecretEqual(actual, expected) { - const actualDigest = crypto.createHash('sha256').update(actual).digest(); - const expectedDigest = crypto.createHash('sha256').update(expected).digest(); + const actualDigest = crypto.createHash("sha256").update(actual).digest(); + const expectedDigest = crypto.createHash("sha256").update(expected).digest(); return crypto.timingSafeEqual(actualDigest, expectedDigest); } @@ -23,11 +23,11 @@ function sanitize(record) { } function generateApiKey() { - return crypto.randomBytes(32).toString('hex'); + return crypto.randomBytes(32).toString("hex"); } function keyId() { - return `key_${crypto.randomUUID().replace(/-/g, '')}`; + return `key_${crypto.randomUUID().replace(/-/g, "")}`; } function keyPath(id) { @@ -51,13 +51,16 @@ async function listKeys() { function normalizeTier(tier) { const tiers = config.apiKeyRateLimit.tiers; - if (typeof tier === 'string' && Object.prototype.hasOwnProperty.call(tiers, tier)) { + if ( + typeof tier === "string" && + Object.prototype.hasOwnProperty.call(tiers, tier) + ) { return tier; } return config.apiKeyRateLimit.defaultTier; } -async function createKey({ label, scopes = ['default'], tier }) { +async function createKey({ label, scopes = ["default"], tier }) { const apiKey = generateApiKey(); const hashed = hashApiKey(apiKey); const now = new Date().toISOString(); @@ -104,16 +107,57 @@ async function touch(record) { return sanitize(updated); } +async function rotateKey(id, options = {}) { + const oldRecord = await getKey(id); + if (!oldRecord) return null; + + // Create new key with same label and scopes, but allow tier override + const newApiKey = generateApiKey(); + const hashed = hashApiKey(newApiKey); + const now = new Date().toISOString(); + const newRecord = { + id: keyId(), + label: oldRecord.label, + key_prefix: newApiKey.slice(0, 8), + key_hash: hashed, + scopes: oldRecord.scopes, + tier: options.tier ? normalizeTier(options.tier) : oldRecord.tier, + created_at: now, + last_used_at: null, + }; + + const redis = cache.getClient(); + + // Create new key first + await cache.set(keyPath(newRecord.id), newRecord); + await cache.set(hashPath(hashed), newRecord.id); + await redis.zadd(IDS_KEY, Date.now(), newRecord.id); + + // Then revoke old key + await cache.del(keyPath(id)); + await cache.del(hashPath(oldRecord.key_hash)); + await redis.zrem(IDS_KEY, id); + + return { + api_key: newApiKey, + key: sanitize(newRecord), + rotated_from: sanitize(oldRecord), + }; +} + async function validateApiKey(apiKey) { if (!apiKey) return null; - if (config.auth.adminApiKey && constantTimeSecretEqual(apiKey, config.auth.adminApiKey)) { + if ( + config.auth.adminApiKey && + constantTimeSecretEqual(apiKey, config.auth.adminApiKey) + ) { return { - id: 'admin', - label: 'Bootstrap admin key', + id: "admin", + label: "Bootstrap admin key", key_prefix: apiKey.slice(0, 8), - scopes: ['admin'], - tier: 'admin', + scopes: ["admin"], + tier: "admin", created_at: null, last_used_at: new Date().toISOString(), }; @@ -141,6 +185,7 @@ module.exports = { hashApiKey, listKeys, normalizeTier, + rotateKey, revokeKey, validateApiKey, }; diff --git a/src/services/idempotency.js b/src/services/idempotency.js new file mode 100644 index 0000000..19ff0d2 --- /dev/null +++ b/src/services/idempotency.js @@ -0,0 +1,96 @@ +const cache = require('./cache'); +const logger = require('../logger'); + +const IDEMPOTENCY_KEY_PREFIX = 'idempotency:'; +const IDEMPOTENCY_TTL_SECONDS = 24 * 60 * 60; // 24 hours + +/** + * Check if an idempotency key was already processed and return the cached response + * + * @param {string} key - The idempotency key + * @returns {Promise} Cached response data if exists, null otherwise + */ +async function getIdempotencyResponse(key) { + if (!key) return null; + + try { + const cacheKey = `${IDEMPOTENCY_KEY_PREFIX}${key}`; + const cached = await cache.get(cacheKey); + return cached ? JSON.parse(cached) : null; + } catch (err) { + logger.warn('Failed to retrieve idempotency response', { key, error: err.message }); + return null; + } +} + +/** + * Store a response for an idempotency key + * + * @param {string} key - The idempotency key + * @param {number} statusCode - HTTP status code + * @param {Object} responseBody - Response body to cache + */ +async function storeIdempotencyResponse(key, statusCode, responseBody) { + if (!key) return; + + try { + const cacheKey = `${IDEMPOTENCY_KEY_PREFIX}${key}`; + const data = { + statusCode, + body: responseBody, + timestamp: new Date().toISOString(), + }; + await cache.setex(cacheKey, IDEMPOTENCY_TTL_SECONDS, JSON.stringify(data)); + } catch (err) { + logger.error('Failed to store idempotency response', { key, error: err.message }); + } +} + +/** + * Middleware to handle idempotency for POST requests + * + * Checks for Idempotency-Key header and: + * - Returns cached response if key was already processed + * - Stores response if this is a new key + * + * To use, call with the resource type: + * app.post('/webhooks', idempotencyMiddleware('webhook'), ...) + */ +function idempotencyMiddleware(resourceType = 'resource') { + return async (req, res, next) => { + const idempotencyKey = req.get('Idempotency-Key'); + + // Store the original json() method + const originalJson = res.json.bind(res); + + // Override json() to capture and cache the response + res.json = function(data) { + if (idempotencyKey && res.statusCode >= 200 && res.statusCode < 300) { + // Only cache successful responses + storeIdempotencyResponse(idempotencyKey, res.statusCode, data); + } + return originalJson(data); + }; + + // Check if this idempotency key was already processed + if (idempotencyKey) { + const cached = await getIdempotencyResponse(idempotencyKey); + if (cached) { + // Return the cached response + res.set('Idempotency-Replay', 'true'); + return res.status(cached.statusCode).json(cached.body); + } + + // Mark that we're processing this key + res.set('Idempotency-Key', idempotencyKey); + } + + return next(); + }; +} + +module.exports = { + idempotencyMiddleware, + getIdempotencyResponse, + storeIdempotencyResponse, +}; diff --git a/src/validation/schemas.js b/src/validation/schemas.js index 500540a..8424d8a 100644 --- a/src/validation/schemas.js +++ b/src/validation/schemas.js @@ -1,26 +1,26 @@ -'use strict'; +"use strict"; -const { z } = require('zod'); -const webhookEvents = require('../services/webhookEvents'); -const config = require('../config'); +const { z } = require("zod"); +const webhookEvents = require("../services/webhookEvents"); +const config = require("../config"); const stellarPublicKeySchema = z .string() - .regex(/^G[A-Z0-9]{55}$/, 'Must be a valid Stellar public key'); + .regex(/^G[A-Z0-9]{55}$/, "Must be a valid Stellar public key"); -const { toStroops, sumStroops, stroopsEqual } = require('../utils/stroops'); +const { toStroops, sumStroops, stroopsEqual } = require("../utils/stroops"); const assetCodeSchema = z .string() .trim() - .min(1, 'Asset code is required') - .max(12, 'Asset code must be 12 characters or fewer') - .regex(/^[A-Za-z0-9]+$/, 'Asset code must be alphanumeric') + .min(1, "Asset code is required") + .max(12, "Asset code must be 12 characters or fewer") + .regex(/^[A-Za-z0-9]+$/, "Asset code must be alphanumeric") .transform((value) => value.toUpperCase()); const optionalIssuerSchema = z.preprocess( - (value) => (value === '' ? undefined : value), - stellarPublicKeySchema.optional() + (value) => (value === "" ? undefined : value), + stellarPublicKeySchema.optional(), ); const paginationQuerySchema = z.object({ @@ -34,25 +34,27 @@ const routeIdParamsSchema = z.object({ .trim() .min(1) .max(128) - .regex(/^[A-Za-z0-9_-]+$/, 'ID can contain only letters, numbers, underscores, and hyphens'), + .regex( + /^[A-Za-z0-9_-]+$/, + "ID can contain only letters, numbers, underscores, and hyphens", + ), }); // Ranges that must never be reachable from an operator-supplied URL — hitting // them lets an attacker use this server as a relay into the internal network. -const PRIVATE_HOSTNAME_RE = - /^(localhost|.*\.local)(:\d+)?$/i; +const PRIVATE_HOSTNAME_RE = /^(localhost|.*\.local)(:\d+)?$/i; const PRIVATE_IP_RE = new RegExp( - '^(' + - '127\\.' + // loopback - '|10\\.' + // RFC-1918 /8 - '|172\\.(1[6-9]|2\\d|3[01])\\.' + // RFC-1918 /12 - '|192\\.168\\.' + // RFC-1918 /16 - '|169\\.254\\.' + // link-local - '|0\\.0\\.0\\.0' + // unspecified - '|::1' + // IPv6 loopback - '|fc[0-9a-f]{2}:' + // IPv6 ULA - ')', + "^(" + + "127\\." + // loopback + "|10\\." + // RFC-1918 /8 + "|172\\.(1[6-9]|2\\d|3[01])\\." + // RFC-1918 /12 + "|192\\.168\\." + // RFC-1918 /16 + "|169\\.254\\." + // link-local + "|0\\.0\\.0\\.0" + // unspecified + "|::1" + // IPv6 loopback + "|fc[0-9a-f]{2}:" + // IPv6 ULA + ")", ); function isPrivateTarget(hostname) { @@ -65,28 +67,34 @@ const httpUrlSchema = z .string() .trim() .refine((value) => !CONTROL_CHAR_RE.test(value), { - message: 'URL must not contain control characters', - }) - .refine((value) => { - try { - const url = new URL(value); - return ['http:', 'https:'].includes(url.protocol); - } catch { - return false; - } - }, { - message: 'Must be an http(s) URL', + message: "URL must not contain control characters", }) - .refine((value) => { - try { - const { hostname } = new URL(value); - return !isPrivateTarget(hostname); - } catch { - return false; - } - }, { - message: 'URL must not target a private or internal network address', - }); + .refine( + (value) => { + try { + const url = new URL(value); + return ["http:", "https:"].includes(url.protocol); + } catch { + return false; + } + }, + { + message: "Must be an http(s) URL", + }, + ) + .refine( + (value) => { + try { + const { hostname } = new URL(value); + return !isPrivateTarget(hostname); + } catch { + return false; + } + }, + { + message: "URL must not target a private or internal network address", + }, + ); const priceParamsSchema = z.object({ asset_code: assetCodeSchema, @@ -98,21 +106,24 @@ const priceQuerySchema = z.object({ const keyCreateBodySchema = z.object({ label: z.string().trim().min(1).max(80), - scopes: z - .array(z.string().trim().min(1)) - .nonempty() - .optional(), + scopes: z.array(z.string().trim().min(1)).nonempty().optional(), // Sizes the key's own rate limit bucket (issue #251). Enumerated from // configuration so adding a tier does not require touching validation. tier: z.enum(Object.keys(config.apiKeyRateLimit.tiers)).optional(), }); +const keyRotateBodySchema = z.object({ + // Preserves label and scopes from the old key, but allows overriding tier + tier: z.enum(Object.keys(config.apiKeyRateLimit.tiers)).optional(), +}); + const alertCreateBodySchema = z.object({ asset: assetCodeSchema.refine( - (code) => config.watchedAssets.length === 0 || config.watchedAssets.includes(code), - { message: 'Asset code is not in the list of watched Stellar assets' }, + (code) => + config.watchedAssets.length === 0 || config.watchedAssets.includes(code), + { message: "Asset code is not in the list of watched Stellar assets" }, ), - type: z.enum(['above', 'below', 'change_pct']), + type: z.enum(["above", "below", "change_pct"]), threshold_usd: z.number().positive(), webhook_url: httpUrlSchema, webhook_secret: z.string().min(8), @@ -133,14 +144,14 @@ const webhookFiltersSchema = z .trim() .min(1) .max(12) - .regex(/^[A-Za-z0-9]+$/, 'Asset filter must be alphanumeric') + .regex(/^[A-Za-z0-9]+$/, "Asset filter must be alphanumeric") .transform((value) => value.toUpperCase()) .optional(), pool_id: z.string().trim().min(1).max(128).optional(), }) .strict() .refine((value) => value.asset !== undefined || value.pool_id !== undefined, { - message: 'At least one filter must be provided', + message: "At least one filter must be provided", }); const webhookCreateBodySchema = z.object({ @@ -162,7 +173,7 @@ const webhookPatchBodySchema = z.object({ const webhookDeliveriesQuerySchema = z.object({ limit: z.coerce.number().int().min(1).max(100).default(50), - status: z.enum(['pending', 'success', 'failed']).optional(), + status: z.enum(["pending", "success", "failed"]).optional(), }); // Stellar Int64 max in stroops, divided by 10_000_000 to get max in whole units @@ -174,28 +185,25 @@ const recipientSchema = z.object({ amount: z .number() .positive() - .max(MAX_AMOUNT_UNITS, 'amount exceeds Stellar Int64 ceiling') - .refine((v) => Number.isFinite(v), 'amount must be a finite number') - .refine( - (v) => { - const str = String(v); - const dotIndex = str.indexOf('.'); - return dotIndex === -1 || str.length - dotIndex - 1 <= 7; - }, - 'amount must have at most 7 decimal places' - ), + .max(MAX_AMOUNT_UNITS, "amount exceeds Stellar Int64 ceiling") + .refine((v) => Number.isFinite(v), "amount must be a finite number") + .refine((v) => { + const str = String(v); + const dotIndex = str.indexOf("."); + return dotIndex === -1 || str.length - dotIndex - 1 <= 7; + }, "amount must have at most 7 decimal places"), }); const recipientsSchema = z .array(recipientSchema) - .max(10000, 'recipients cannot exceed 10,000') + .max(10000, "recipients cannot exceed 10,000") .superRefine((recipients, ctx) => { const seen = new Set(); recipients.forEach((recipient, index) => { if (seen.has(recipient.address)) { ctx.addIssue({ code: z.ZodIssueCode.custom, - path: [index, 'address'], + path: [index, "address"], message: `recipient ${index}: duplicate address ${recipient.address}`, }); } @@ -207,7 +215,10 @@ function expiryLedgerSchema(currentLedger) { return z .number() .int() - .gt(currentLedger, `expiry_ledger must be greater than current ledger (${currentLedger})`); + .gt( + currentLedger, + `expiry_ledger must be greater than current ledger (${currentLedger})`, + ); } function airdropCreateBodySchema(currentLedger) { @@ -217,7 +228,10 @@ function airdropCreateBodySchema(currentLedger) { description: z.string().optional(), asset: assetCodeSchema, asset_issuer: stellarPublicKeySchema, - total_amount: z.number().positive().max(MAX_AMOUNT_UNITS, `total_amount exceeds Stellar Int64 ceiling`), + total_amount: z + .number() + .positive() + .max(MAX_AMOUNT_UNITS, `total_amount exceeds Stellar Int64 ceiling`), expiry_ledger: expiryLedgerSchema(currentLedger), recipients: recipientsSchema.optional().default([]), }) @@ -230,14 +244,14 @@ function airdropCreateBodySchema(currentLedger) { if (totalStroops !== expectedStroops) { ctx.addIssue({ code: z.ZodIssueCode.custom, - path: ['recipients'], + path: ["recipients"], message: `sum of recipient amounts must equal total_amount (${body.total_amount})`, }); } } catch (err) { ctx.addIssue({ code: z.ZodIssueCode.custom, - path: ['recipients'], + path: ["recipients"], message: err.message, }); } @@ -254,7 +268,7 @@ function airdropUpdateBodySchema(currentLedger) { const airdropRecipientsBodySchema = z.object({ recipients: z.preprocess((value) => { - if (typeof value !== 'string') return value; + if (typeof value !== "string") return value; try { return JSON.parse(value); @@ -272,6 +286,7 @@ module.exports = { assetCodeSchema, httpUrlSchema, keyCreateBodySchema, + keyRotateBodySchema, optionalIssuerSchema, paginationQuerySchema, priceParamsSchema,