diff --git a/backend/env.example.txt b/backend/env.example.txt index 0367ffb4..1ce2e60f 100644 --- a/backend/env.example.txt +++ b/backend/env.example.txt @@ -104,3 +104,12 @@ VERIFY_CONCURRENCY_RETRY_AFTER_SECONDS=5 # Require cryptographic signatures for cancel (production always requires). REQUIRE_SIGNATURES=false + +# Signed seller webhooks. Run db:migrate before enabling PostgreSQL delivery. +# MVP endpoints/outbox are volatile; use PostgreSQL for durable delivery. +# Generate a stable server-only encryption key with: openssl rand -hex 32 +# Share the key across replicas; never expose it with NEXT_PUBLIC. +WEBHOOKS_ENABLED=false +WEBHOOK_ENCRYPTION_KEY= +WEBHOOK_MAX_ATTEMPTS=8 +WEBHOOK_DISABLE_AFTER_FAILURES=10 diff --git a/backend/env.mvp.example b/backend/env.mvp.example index 69594946..57388839 100644 --- a/backend/env.mvp.example +++ b/backend/env.mvp.example @@ -100,3 +100,12 @@ REQUIRE_SIGNATURES=false # Redis connection for verify-cache (optional). Rate limiters use in-memory. # Falls back to in-memory cache if not configured. REDIS_URL= + +# Signed seller webhooks. Run db:migrate before enabling PostgreSQL delivery. +# MVP endpoints/outbox are volatile; use PostgreSQL for durable delivery. +# Generate a stable server-only encryption key with: openssl rand -hex 32 +# Share the key across replicas; never expose it with NEXT_PUBLIC. +WEBHOOKS_ENABLED=false +WEBHOOK_ENCRYPTION_KEY= +WEBHOOK_MAX_ATTEMPTS=8 +WEBHOOK_DISABLE_AFTER_FAILURES=10 diff --git a/backend/src/observability/log-events.ts b/backend/src/observability/log-events.ts index 44e54de0..adc3c7de 100644 --- a/backend/src/observability/log-events.ts +++ b/backend/src/observability/log-events.ts @@ -50,6 +50,7 @@ const FAILURE_OPERATIONS = [ 'stellar.streamPayment', 'stellar.payments', 'stellar.submit', 'cache.connect', 'cache.get', 'cache.set', 'cache.clear', 'cache.verify', 'proof.invariants', 'proof.json', 'proof.pdf', 'proof.handoff', + 'webhook.worker', 'webhook.manage', ] as const; type FailureOperation = typeof FAILURE_OPERATIONS[number]; diff --git a/backend/src/routes/invoice.handlers.ts b/backend/src/routes/invoice.handlers.ts index e9d7e125..bc838fc4 100644 --- a/backend/src/routes/invoice.handlers.ts +++ b/backend/src/routes/invoice.handlers.ts @@ -137,6 +137,22 @@ export function createInvoiceHandlers(options: InvoiceHandlerOptions): InvoiceHa .set(invoiceId, txHash, httpStatus, body) .catch(() => { emitOperationalFailure('cache.set'); }); + // A rejected verification is acknowledged/cached only after its audit/outbox + // transaction commits. Otherwise the caller can retry instead of losing an event. + const recordRejection = async (res: Response, id: string, txHash: string, code: string): Promise => { + try { + await storage.logPaymentEvent(id, + code === 'AMOUNT_TOO_LOW' || code === 'AMOUNT_MISMATCH' ? 'PARTIAL_PAYMENT' : 'PAYMENT_REJECTED', + { code, txHash, source: 'manual-verify' }); + return true; + } catch { + emitOperationalFailure('invoice.events'); + res.status(503).json({ success: false, code: 'PAYMENT_EVENT_UNAVAILABLE', + error: 'The payment event could not be recorded. Please retry verification.' }); + return false; + } + }; + const frontendUrl = () => options.frontendUrl || process.env.FRONTEND_URL || 'http://localhost:3000'; @@ -689,21 +705,8 @@ export function createInvoiceHandlers(options: InvoiceHandlerOptions): InvoiceHa durationMs: Date.now() - startedAt, }); const body = verificationFailureBody(verification.code, verification.error); + if (!await recordRejection(res, id, hashCheck.value, verification.code)) return; await cacheResult(id, hashCheck.value, 400, body); - // Issue #515: a rejected verify lands on the seller's audit feed with - // the same taxonomy the monitor uses, so "still PENDING" answers - // itself without the payer having to say so. - await storage.logPaymentEvent( - id, - verification.code === 'AMOUNT_TOO_LOW' || verification.code === 'AMOUNT_MISMATCH' - ? 'PARTIAL_PAYMENT' - : 'PAYMENT_REJECTED', - { - code: verification.code, - txHash: hashCheck.value, - source: 'manual-verify', - } - ).catch(() => undefined); return sendVerificationFailure(res, 400, verification.code, verification.error); } @@ -774,6 +777,7 @@ export function createInvoiceHandlers(options: InvoiceHandlerOptions): InvoiceHa durationMs: Date.now() - startedAt, }); const body = verificationFailureBody(error.code, messageForCode(error.code)); + if (!await recordRejection(res, id, hashCheck.value, error.code)) return; await cacheResult(id, hashCheck.value, 409, body); res.status(409).json(body); return; diff --git a/backend/src/routes/invoice.routes.ts b/backend/src/routes/invoice.routes.ts index 8b062058..4ab3c163 100644 --- a/backend/src/routes/invoice.routes.ts +++ b/backend/src/routes/invoice.routes.ts @@ -13,6 +13,7 @@ import { } from '../middleware/rate-limit'; import { createInvoiceCeilingMiddleware } from '../middleware/invoice-ceiling'; import { createVerifyCacheMiddleware, verificationCache } from '../middleware/verify-cache'; +import { createWebhookRouter } from './webhook.routes'; export interface InvoiceRouterOptions extends InvoiceHandlerOptions { enableRateLimiting?: boolean; @@ -48,6 +49,7 @@ export function createInvoiceRouter(options: InvoiceRouterOptions): Router { const handlers = createInvoiceHandlers(options); const router = Router(); router.use(requestCorrelationMiddleware); + if (options.storage.webhooks) router.use(createWebhookRouter(options.storage.webhooks)); const enableRateLimiting = options.enableRateLimiting ?? diff --git a/backend/src/routes/webhook.routes.ts b/backend/src/routes/webhook.routes.ts new file mode 100644 index 00000000..0bc97036 --- /dev/null +++ b/backend/src/routes/webhook.routes.ts @@ -0,0 +1,256 @@ +import { randomUUID } from 'node:crypto'; +import { Router, type Request, type Response, type RequestHandler } from 'express'; +import { + WEBHOOK_EVENT_TYPES, + webhookProofMessage, + type WebhookAction, + type WebhookEventType, +} from '../../../shared/webhooks'; +import { createRateLimiter, getClientIp } from '../middleware/rate-limit'; +import { emitOperationalFailure } from '../observability/log-events'; +import { + WebhookStoreError, + publicWebhookDelivery, + publicWebhookEndpoint, + type WebhookEndpoint, + type WebhookStorage, +} from '../storage/webhook-storage'; +import { newWebhookSecret, webhookEncryptionKey } from '../services/webhook-crypto'; +import { + validateWebhookUrl, + WebhookTransportError, + type WebhookResolver, +} from '../services/webhook-transport'; +import { verifySellerSignature } from '../utils/signature-verification'; + +export interface WebhookRouterOptions { + encryptionKey?: Buffer; + enabled?: boolean; + clock?: () => Date; + resolve?: WebhookResolver; +} + +const UUID = /^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$/; +const eventTypes: ReadonlySet = new Set(WEBHOOK_EVENT_TYPES); + +function fail(res: Response, status: number, code: string, message: string): void { + res.status(status).json({ success: false, code, error: message }); +} + +export function createWebhookRouter( + storage: WebhookStorage, + options: WebhookRouterOptions = {} +): Router { + const router = Router(); + const clock = options.clock ?? (() => new Date()); + const enabled = options.enabled ?? process.env.WEBHOOKS_ENABLED === 'true'; + const encryptionKey = options.encryptionKey ?? (enabled ? webhookEncryptionKey() : undefined); + if (encryptionKey && encryptionKey.length !== 32) { + throw new Error('Webhook encryption requires a 32-byte key'); + } + + router.use('/webhooks', (_req: Request, res: Response, next) => { + res.setHeader('Cache-Control', 'no-store'); + next(); + }, createRateLimiter({ + windowMs: 60_000, + max: 60, + keyGenerator: (req) => 'manage_webhooks:' + getClientIp(req), + code: 'WEBHOOK_RATE_LIMIT_EXCEEDED', + message: 'Too many webhook management requests. Please retry later.', + })); + + function handler(action: WebhookAction): RequestHandler { + return async (req: Request, res: Response): Promise => { + try { + const body = req.body && typeof req.body === 'object' && !Array.isArray(req.body) + ? req.body as Record + : {}; + const endpointId = typeof req.params.id === 'string' ? req.params.id : undefined; + if ( + action !== 'list' && + action !== 'register' && + (!endpointId || !UUID.test(endpointId)) + ) { + fail(res, 400, 'INVALID_ENDPOINT_ID', 'Provide a valid webhook endpoint ID.'); + return; + } + + const { sellerPublicKey, signature, timestamp, nonce } = body; + if ( + typeof sellerPublicKey !== 'string' || + typeof signature !== 'string' || + timestamp === undefined || + typeof nonce !== 'string' + ) { + fail(res, 401, 'WEBHOOK_PROOF_REQUIRED', 'A signed seller wallet proof is required.'); + return; + } + + const now = clock(); + if ( + sellerPublicKey.length !== 56 || + signature.length === 0 || + signature.length > 256 || + typeof timestamp !== 'number' || + !Number.isSafeInteger(timestamp) || + !UUID.test(nonce) || + now.getTime() - timestamp * 1000 >= 300_000 || + timestamp * 1000 - now.getTime() > 30_000 + ) { + fail(res, 401, 'INVALID_WEBHOOK_PROOF', 'The wallet proof is invalid or has expired.'); + return; + } + + let url: string | undefined; + let events: WebhookEventType[] | undefined; + if (action === 'register') { + if ( + typeof body.url !== 'string' || + body.url.length === 0 || + body.url.length > 2048 || + !Array.isArray(body.events) || + body.events.length < 1 || + body.events.length > WEBHOOK_EVENT_TYPES.length || + !body.events.every((event): event is WebhookEventType => + typeof event === 'string' && eventTypes.has(event)) || + new Set(body.events).size !== body.events.length + ) { + fail(res, 400, 'INVALID_WEBHOOK', 'Provide an HTTPS URL and one or more distinct supported events.'); + return; + } + url = body.url; + events = body.events; + } + + const message = webhookProofMessage({ + sellerPublicKey, + action, + endpointId, + url, + events, + timestamp, + nonce, + }); + if (!verifySellerSignature(sellerPublicKey, signature, [message])) { + fail(res, 401, 'INVALID_WEBHOOK_PROOF', 'The wallet proof is invalid or has expired.'); + return; + } + + if (action !== 'list' && action !== 'remove' && (!enabled || !encryptionKey)) { + fail(res, 503, 'WEBHOOKS_DISABLED', 'Webhook delivery is not enabled on this server.'); + return; + } + const consumed = await storage.consumeProof( + sellerPublicKey, nonce, new Date((timestamp + 300) * 1000), now + ); + if (!consumed) { + fail(res, 409, 'WEBHOOK_PROOF_REPLAYED', 'This wallet proof has already been used.'); + return; + } + + if (action === 'list') { + const [endpoints, deliveries] = await Promise.all([ + storage.listEndpoints(sellerPublicKey), + storage.listDeliveries(sellerPublicKey), + ]); + res.json({ + success: true, + data: { + endpoints: endpoints.map(publicWebhookEndpoint), + deliveries: deliveries.map(publicWebhookDelivery), + }, + }); + return; + } + + if (action === 'register') { + const validated = await validateWebhookUrl(url!, options.resolve); + const { secret, ...secrets } = newWebhookSecret(encryptionKey!); + const endpoint: WebhookEndpoint = { + id: randomUUID(), + sellerPublicKey, + url: validated.url.href, + events: events!, + enabled: true, + failureCount: 0, + createdAt: now.toISOString(), + ...secrets, + }; + await storage.register(endpoint); + res.status(201).json({ + success: true, + data: { endpoint: publicWebhookEndpoint(endpoint), secret }, + }); + return; + } + + if (action === 'remove') { + if (!await storage.remove(sellerPublicKey, endpointId!, now)) { + fail(res, 404, 'ENDPOINT_NOT_FOUND', 'Webhook endpoint not found.'); + return; + } + res.json({ success: true, data: { id: endpointId, removed: true } }); + return; + } + + if (action === 'rotate') { + const { secret, ...secrets } = newWebhookSecret(encryptionKey!); + const endpoint = await storage.rotate( + sellerPublicKey, endpointId!, secrets, now, 86_400_000 + ); + if (!endpoint) { + fail(res, 404, 'ENDPOINT_NOT_FOUND', 'Webhook endpoint not found.'); + return; + } + res.json({ + success: true, + data: { + endpoint: publicWebhookEndpoint(endpoint), + secret, + previousSecretExpiresAt: endpoint.previousSecretExpiresAt, + }, + }); + return; + } + + const eventId = randomUUID(); + if (!await storage.enqueueTest(sellerPublicKey, endpointId!, eventId, now)) { + fail(res, 404, 'ENDPOINT_NOT_FOUND', 'Webhook endpoint not found.'); + return; + } + res.status(202).json({ success: true, data: { eventId } }); + } catch (error) { + if (error instanceof WebhookStoreError) { + switch (error.code) { + case 'ENDPOINT_LIMIT': + fail(res, 400, error.code, 'The webhook endpoint limit has been reached.'); + return; + case 'ENDPOINT_NOT_FOUND': + fail(res, 404, error.code, 'Webhook endpoint not found.'); + return; + case 'ROTATION_IN_PROGRESS': + fail(res, 409, error.code, 'The previous signing secret is still in its rotation window.'); + return; + case 'TEST_RATE_LIMIT': + fail(res, 429, error.code, 'Wait before sending another test event.'); + return; + } + } + if (error instanceof WebhookTransportError) { + fail(res, 400, 'INVALID_WEBHOOK', 'The endpoint must resolve to a public HTTPS address without credentials or a fragment.'); + return; + } + emitOperationalFailure('webhook.manage'); + fail(res, 503, 'WEBHOOK_UNAVAILABLE', 'Webhook management is temporarily unavailable.'); + } + }; + } + + router.post('/webhooks', handler('register')); + router.post('/webhooks/list', handler('list')); + router.post('/webhooks/:id/remove', handler('remove')); + router.post('/webhooks/:id/rotate', handler('rotate')); + router.post('/webhooks/:id/test', handler('test')); + return router; +} diff --git a/backend/src/server-dual.ts b/backend/src/server-dual.ts index 4021decd..6873cc18 100644 --- a/backend/src/server-dual.ts +++ b/backend/src/server-dual.ts @@ -38,6 +38,7 @@ import { configuredFrontendOrigins, configuredStorageMode, corsOptions } from '. import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; import type { InvoiceStorage } from './storage/invoice-storage'; dotenv.config(); @@ -74,6 +75,7 @@ paymentMonitorService.configure({ const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; app.use(cors(corsOptions())); app.use(requestCorrelationMiddleware); @@ -165,6 +167,8 @@ async function initialize(): Promise { * configured one. Importing this module never starts a server on its own. */ export function startServer(port: number | string = PORT) { + const worker = startConfiguredWebhookWorker(storage.webhooks); + webhookWorker = worker; const server = app.listen(port, async () => { await initialize(); const modeLabel = storageMode === 'postgres' ? 'PostgreSQL' : 'In-Memory (MVP)'; @@ -179,6 +183,8 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); return server; } @@ -188,6 +194,7 @@ if (/server-dual(\.[cm]?[jt]s)?$/.test(entryPoint)) { } process.on('SIGTERM', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); if (storageMode === 'postgres') { await pool.end(); @@ -196,6 +203,7 @@ process.on('SIGTERM', async () => { }); process.on('SIGINT', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); if (storageMode === 'postgres') { await pool.end(); diff --git a/backend/src/server-mvp.ts b/backend/src/server-mvp.ts index ea5555d0..c43bbaf0 100644 --- a/backend/src/server-mvp.ts +++ b/backend/src/server-mvp.ts @@ -21,6 +21,7 @@ import { configuredFrontendOrigins, corsOptions } from './config/runtime'; import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; dotenv.config(); @@ -37,6 +38,7 @@ paymentMonitorService.configure({ const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; // Middleware app.use(cors(corsOptions())); @@ -119,6 +121,8 @@ app.use((req: Request, res: Response) => { * configured one, and so importing this module never starts a server. */ export function startServer(port: number | string = PORT) { + const worker = startConfiguredWebhookWorker(memoryInvoiceStorage.webhooks); + webhookWorker = worker; if (SELLER_PUBLIC_KEY) { try { paymentMonitorService.start(); @@ -127,7 +131,7 @@ export function startServer(port: number | string = PORT) { } } - return app.listen(port, () => { + const server = app.listen(port, () => { console.log('\n๐Ÿš€ Quittance Backend (MVP Mode)'); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”'); console.log(`โœ… Server running on port ${port}`); @@ -138,6 +142,9 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); + return server; } const entryPoint = process.argv[1] ?? ''; @@ -145,12 +152,14 @@ if (/server-mvp(\.[cm]?[jt]s)?$/.test(entryPoint)) { startServer(); } -process.on('SIGTERM', () => { +process.on('SIGTERM', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); process.exit(0); }); -process.on('SIGINT', () => { +process.on('SIGINT', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); process.exit(0); }); diff --git a/backend/src/server.ts b/backend/src/server.ts index 78dfbd54..07fefcaa 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -13,12 +13,14 @@ import postgresInvoiceStorage from './storage/postgres-invoice-storage'; import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; dotenv.config(); const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; app.use(cors(corsOptions())); app.use(requestCorrelationMiddleware); @@ -86,7 +88,9 @@ async function initialize() { } export function startServer(port: number | string = PORT) { - return app.listen(port, async () => { + const worker = startConfiguredWebhookWorker(postgresInvoiceStorage.webhooks); + webhookWorker = worker; + const server = app.listen(port, async () => { await initialize(); console.log('\n๐Ÿš€ Quittance Backend (Postgres Mode)'); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”'); @@ -98,6 +102,9 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); + return server; } const entryPoint = process.argv[1] ?? ''; @@ -106,6 +113,7 @@ if (/server(\.[cm]?[jt]s)?$/.test(entryPoint)) { } process.on('SIGTERM', async () => { + await webhookWorker?.stop(); console.log('Shutting down...'); paymentMonitorService.stop(); await pool.end(); @@ -113,6 +121,7 @@ process.on('SIGTERM', async () => { }); process.on('SIGINT', async () => { + await webhookWorker?.stop(); console.log('Shutting down...'); paymentMonitorService.stop(); await pool.end(); diff --git a/backend/src/services/invoice-memory.service.ts b/backend/src/services/invoice-memory.service.ts index e5d7fb03..ba48703d 100644 --- a/backend/src/services/invoice-memory.service.ts +++ b/backend/src/services/invoice-memory.service.ts @@ -32,6 +32,10 @@ export class InvoiceMemoryService { private readonly nextId: () => string = generatePublicInvoiceId ) {} + get webhooks() { + return this.storage.webhooks; + } + async createInvoice(input: CreateInvoiceInput): Promise { if (!input.sellerPublicKey) { throw new Error('Seller public key is required'); diff --git a/backend/src/services/invoice.service.ts b/backend/src/services/invoice.service.ts index cf6a86d9..2ca80d9b 100644 --- a/backend/src/services/invoice.service.ts +++ b/backend/src/services/invoice.service.ts @@ -1,3 +1,5 @@ +import type { Pool } from 'pg'; +import { PostgresWebhookStorage } from '../storage/postgres-webhook-storage'; import { emitOperationalFailure } from '../observability/log-events'; import { v4 as uuidv4 } from 'uuid'; import { InvoiceIdCollisionError, MemoCollisionError } from '../domain/payment-attribution'; @@ -64,7 +66,13 @@ export interface Invoice { } export class InvoiceService { - constructor(private readonly db: Queryable = pool) {} + readonly webhooks?: PostgresWebhookStorage; + + constructor(private readonly db: Queryable = pool) { + if (typeof (db as Partial).connect === 'function') { + this.webhooks = new PostgresWebhookStorage(db as Pool); + } + } /** * Create a new invoice for the seller wallet supplied by the request diff --git a/backend/src/services/webhook-crypto.ts b/backend/src/services/webhook-crypto.ts new file mode 100644 index 00000000..8a8f47b9 --- /dev/null +++ b/backend/src/services/webhook-crypto.ts @@ -0,0 +1,171 @@ +import { + createCipheriv, + createDecipheriv, + createHash, + createHmac, + randomBytes, + timingSafeEqual, +} from 'node:crypto'; + +export interface WebhookSecrets { + secretHash: string; + secretEncrypted: string; + previousSecretHash?: string; + previousSecretEncrypted?: string; + previousSecretExpiresAt?: string; +} + +const SECRET_AAD = Buffer.from('quittance-webhook-secret-v1', 'utf8'); + +export function webhookEncryptionKey( + value: string | undefined = process.env.WEBHOOK_ENCRYPTION_KEY +): Buffer { + if (!value || !/^[a-fA-F0-9]{64}$/.test(value)) { + throw new Error('WEBHOOK_ENCRYPTION_KEY must contain exactly 64 hexadecimal characters'); + } + return Buffer.from(value, 'hex'); +} + +function assertKey(key: Buffer): void { + if (!Buffer.isBuffer(key) || key.length !== 32) { + throw new Error('Webhook encryption requires a 32-byte key'); + } +} + +function secretHash(secret: string): string { + return createHash('sha256').update(secret, 'utf8').digest('hex'); +} + +export function sealWebhookSecret(secret: string, key: Buffer): WebhookSecrets { + assertKey(key); + const iv = randomBytes(12); + const cipher = createCipheriv('aes-256-gcm', key, iv); + cipher.setAAD(SECRET_AAD); + const ciphertext = Buffer.concat([cipher.update(secret, 'utf8'), cipher.final()]); + return { + secretHash: secretHash(secret), + secretEncrypted: [ + 'v1', + iv.toString('base64'), + cipher.getAuthTag().toString('base64'), + ciphertext.toString('base64'), + ].join('.'), + }; +} + +export function newWebhookSecret(key: Buffer): WebhookSecrets & { secret: string } { + const secret = 'whsec_' + randomBytes(32).toString('base64url'); + return { ...sealWebhookSecret(secret, key), secret }; +} + +export function openWebhookSecret( + encrypted: string, + expectedHash: string, + key: Buffer +): string { + assertKey(key); + try { + if (encrypted.length > 8192 || !/^[a-fA-F0-9]{64}$/.test(expectedHash)) { + throw new Error('Invalid signing secret'); + } + const parts = encrypted.split('.'); + if (parts.length !== 4 || parts[0] !== 'v1') { + throw new Error('Invalid signing secret'); + } + const iv = Buffer.from(parts[1], 'base64'); + const tag = Buffer.from(parts[2], 'base64'); + const ciphertext = Buffer.from(parts[3], 'base64'); + if (iv.length !== 12 || tag.length !== 16 || ciphertext.length === 0) { + throw new Error('Invalid signing secret'); + } + const decipher = createDecipheriv('aes-256-gcm', key, iv); + decipher.setAAD(SECRET_AAD); + decipher.setAuthTag(tag); + const secret = Buffer.concat([decipher.update(ciphertext), decipher.final()]).toString('utf8'); + const actualHash = Buffer.from(secretHash(secret), 'hex'); + if (!timingSafeEqual(actualHash, Buffer.from(expectedHash, 'hex'))) { + throw new Error('Invalid signing secret'); + } + return secret; + } catch { + throw new Error('Invalid webhook signing secret'); + } +} + +export function webhookSignature( + secret: string, + timestamp: number, + body: string | Buffer +): string { + return createHmac('sha256', secret) + .update(String(timestamp) + '.', 'utf8') + .update(body) + .digest('hex'); +} + +export function webhookSignatureHeader( + secrets: WebhookSecrets, + key: Buffer, + timestamp: number, + body: string | Buffer +): string { + if (!Number.isSafeInteger(timestamp) || timestamp < 0) { + throw new Error('Invalid webhook timestamp'); + } + const current = openWebhookSecret(secrets.secretEncrypted, secrets.secretHash, key); + const signatures = ['t=' + timestamp, 'v1=' + webhookSignature(current, timestamp, body)]; + if ( + secrets.previousSecretEncrypted && + secrets.previousSecretHash && + secrets.previousSecretExpiresAt && + Date.parse(secrets.previousSecretExpiresAt) > timestamp * 1000 + ) { + const previous = openWebhookSecret( + secrets.previousSecretEncrypted, + secrets.previousSecretHash, + key + ); + signatures.push('v1=' + webhookSignature(previous, timestamp, body)); + } + return signatures.join(','); +} + +/** Verify the original request bytes before parsing JSON or processing an event. */ +export function verifyWebhookSignature( + secret: string, + header: string, + rawBody: string | Buffer, + nowMs: number = Date.now() +): boolean { + if (typeof header !== 'string' || header.length > 1024 || !Number.isFinite(nowMs)) { + return false; + } + let rawTimestamp: string | undefined; + const candidates: Buffer[] = []; + for (const part of header.split(',')) { + const entry = part.trim(); + if (entry.startsWith('t=')) { + if (rawTimestamp !== undefined || !/^t=[0-9]{1,12}$/.test(entry)) return false; + rawTimestamp = entry.slice(2); + } else if (/^v1=[a-fA-F0-9]{64}$/.test(entry)) { + candidates.push(Buffer.from(entry.slice(3), 'hex')); + } else { + return false; + } + } + if (rawTimestamp === undefined || candidates.length === 0) return false; + const timestamp = Number(rawTimestamp); + if ( + !Number.isSafeInteger(timestamp) || + String(timestamp) !== rawTimestamp || + Math.abs(nowMs - timestamp * 1000) >= 300_000 + ) { + return false; + } + const expected = Buffer.from(webhookSignature(secret, timestamp, rawBody), 'hex'); + let matches = false; + for (const candidate of candidates) { + matches = timingSafeEqual(expected, candidate) || matches; + } + return matches; +} diff --git a/backend/src/services/webhook-transport.ts b/backend/src/services/webhook-transport.ts new file mode 100644 index 00000000..48299033 --- /dev/null +++ b/backend/src/services/webhook-transport.ts @@ -0,0 +1,207 @@ +import { lookup as dnsLookup } from 'node:dns/promises'; +import { request as httpsRequest, type RequestOptions } from 'node:https'; +import { BlockList, isIP } from 'node:net'; + +export type WebhookTransportErrorCode = + | 'UNSAFE_URL' + | 'UNSAFE_ADDRESS' + | 'DNS_FAILED' + | 'TIMEOUT' + | 'TRANSPORT_FAILED'; + +export class WebhookTransportError extends Error { + constructor(public readonly code: WebhookTransportErrorCode) { + super('Webhook transport failed: ' + code); + this.name = 'WebhookTransportError'; + } +} + +export type WebhookResolver = ( + hostname: string +) => Promise; + +export type WebhookTransport = ( + url: string, + body: string, + headers: Record +) => Promise<{ statusCode: number }>; + +const deniedAddresses = new BlockList(); +for (const [address, prefix] of [ + ['0.0.0.0', 8], + ['10.0.0.0', 8], + ['100.64.0.0', 10], + ['127.0.0.0', 8], + ['169.254.0.0', 16], + ['172.16.0.0', 12], + ['192.0.0.0', 24], + ['192.0.2.0', 24], + ['192.88.99.0', 24], + ['192.168.0.0', 16], + ['198.18.0.0', 15], + ['198.51.100.0', 24], + ['203.0.113.0', 24], + ['224.0.0.0', 4], + ['240.0.0.0', 4], +] as const) { + deniedAddresses.addSubnet(address, prefix, 'ipv4'); +} +for (const [address, prefix] of [ + ['::', 96], + ['::1', 128], + ['64:ff9b::', 96], + ['64:ff9b:1::', 48], + ['100::', 64], + ['2001::', 32], + ['2001:2::', 48], + ['2001:10::', 28], + ['2001:20::', 28], + ['2001:db8::', 32], + ['2002::', 16], + ['3fff::', 20], + ['fc00::', 7], + ['fe80::', 10], + ['fec0::', 10], + ['ff00::', 8], +] as const) { + deniedAddresses.addSubnet(address, prefix, 'ipv6'); +} +const publicV6Addresses = new BlockList(); +publicV6Addresses.addSubnet('2000::', 3, 'ipv6'); +publicV6Addresses.addSubnet('::ffff:0:0', 96, 'ipv6'); + +/** Node's BlockList also applies the IPv4 rules to IPv4-mapped IPv6 addresses. */ +export function isPublicWebhookAddress(address: string): boolean { + if (address.includes('%')) return false; + const family = isIP(address); + if (family === 4) return !deniedAddresses.check(address, 'ipv4'); + if (family !== 6) return false; + return publicV6Addresses.check(address, 'ipv6') && + !deniedAddresses.check(address, 'ipv6'); +} + +const defaultResolver: WebhookResolver = (hostname) => + dnsLookup(hostname, { all: true, verbatim: true }); + +export async function validateWebhookUrl( + raw: string, + resolve: WebhookResolver = defaultResolver +): Promise<{ url: URL; address: string; family: number }> { + let url: URL; + try { + if (typeof raw !== 'string' || raw.length > 2048 || raw !== raw.trim()) { + throw new Error('Invalid URL'); + } + url = new URL(raw); + if (url.protocol !== 'https:' || url.username || url.password || url.hash) { + throw new Error('Invalid URL'); + } + } catch { + throw new WebhookTransportError('UNSAFE_URL'); + } + + const hostname = url.hostname.replace(/^\[|\]$/g, ''); + const literalFamily = isIP(hostname); + let addresses: readonly { address: string; family: number }[]; + if (literalFamily) { + addresses = [{ address: hostname, family: literalFamily }]; + } else { + let timer: ReturnType | undefined; + try { + addresses = await Promise.race([ + resolve(hostname), + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new WebhookTransportError('DNS_FAILED')), 2000); + }), + ]); + } catch { + throw new WebhookTransportError('DNS_FAILED'); + } finally { + if (timer) clearTimeout(timer); + } + } + + if (!Array.isArray(addresses) || addresses.length === 0) { + throw new WebhookTransportError('DNS_FAILED'); + } + for (const entry of addresses) { + if ( + (entry.family !== 4 && entry.family !== 6) || + isIP(entry.address) !== entry.family || + !isPublicWebhookAddress(entry.address) + ) { + throw new WebhookTransportError('UNSAFE_ADDRESS'); + } + } + return { url, address: addresses[0].address, family: addresses[0].family }; +} + +export function createWebhookTransport( + options: { resolve?: WebhookResolver; timeoutMs?: number } = {} +): WebhookTransport { + const timeoutMs = options.timeoutMs ?? 5000; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { + throw new Error('Webhook transport timeout must be positive'); + } + + return async (rawUrl, body, headers) => { + // Resolve again at send time and pin this result through the connection. + const { url, address, family } = await validateWebhookUrl(rawUrl, options.resolve); + const pinnedLookup: NonNullable = (_hostname, lookupOptions, callback) => { + if (lookupOptions.all) callback(null, [{ address, family }]); + else callback(null, address, family); + }; + return new Promise<{ statusCode: number }>((resolve, reject) => { + let settled = false; + let timer: ReturnType | undefined; + const finish = (error?: WebhookTransportError, statusCode?: number) => { + if (settled) return; + settled = true; + if (timer) clearTimeout(timer); + if (error) reject(error); + else resolve({ statusCode: statusCode ?? 0 }); + }; + + const request = httpsRequest(url, { + method: 'POST', + agent: false, + lookup: pinnedLookup, + maxHeaderSize: 16 * 1024, + headers: { + ...headers, + 'Content-Type': 'application/json', + 'Content-Length': String(Buffer.byteLength(body, 'utf8')), + Connection: 'close', + }, + }, (response) => { + // Native HTTPS does not follow redirects. Read and discard the response + // without retaining or logging potentially sensitive receiver content. + response.once('error', () => { + request.destroy(); + finish(new WebhookTransportError('TRANSPORT_FAILED')); + }); + response.once('aborted', () => { + request.destroy(); + finish(new WebhookTransportError('TRANSPORT_FAILED')); + }); + response.once('end', () => finish(undefined, response.statusCode ?? 0)); + response.resume(); + }); + + request.once('error', (error) => { + finish(error instanceof WebhookTransportError + ? error + : new WebhookTransportError('TRANSPORT_FAILED')); + }); + const timeout = () => { + const error = new WebhookTransportError('TIMEOUT'); + request.destroy(error); + finish(error); + }; + request.setTimeout(timeoutMs, timeout); + // A wall-clock limit also stops a receiver that keeps trickling bytes. + timer = setTimeout(timeout, timeoutMs); + request.end(body); + }); + }; +} diff --git a/backend/src/services/webhook-worker.ts b/backend/src/services/webhook-worker.ts new file mode 100644 index 00000000..7817ce24 --- /dev/null +++ b/backend/src/services/webhook-worker.ts @@ -0,0 +1,177 @@ +import { emitOperationalFailure } from '../observability/log-events'; +import type { WebhookStorage, WebhookAttemptResult } from '../storage/webhook-storage'; +import { monitorBackoffMs } from '../utils/monitor-retry-backoff'; +import { redactWebhookPayload } from '../utils/payment-event-redaction'; +import { webhookEncryptionKey, webhookSignatureHeader } from './webhook-crypto'; +import { + createWebhookTransport, + WebhookTransportError, + type WebhookTransport, +} from './webhook-transport'; + +export interface WebhookWorkerOptions { + encryptionKey: Buffer; + transport?: WebhookTransport; + maxAttempts?: number; + disableAfterFailures?: number; + clock?: () => Date; + random?: () => number; + pollMs?: number; +} + +function boundedInteger(value: number, name: string, min: number, max: number): number { + if (!Number.isSafeInteger(value) || value < min || value > max) { + throw new Error(name + ' must be an integer from ' + min + ' through ' + max); + } + return value; +} + +export class WebhookWorker { + private readonly transport: WebhookTransport; + private readonly maxAttempts: number; + private readonly disableAfterFailures: number; + private readonly clock: () => Date; + private readonly random: () => number; + private readonly pollMs: number; + private timer?: ReturnType; + private inFlight?: Promise; + private running = false; + + constructor( + private readonly storage: WebhookStorage, + private readonly options: WebhookWorkerOptions + ) { + if (!Buffer.isBuffer(options.encryptionKey) || options.encryptionKey.length !== 32) { + throw new Error('Webhook encryption requires a 32-byte key'); + } + this.transport = options.transport ?? createWebhookTransport(); + this.maxAttempts = boundedInteger(options.maxAttempts ?? 8, 'maxAttempts', 1, 20); + this.disableAfterFailures = boundedInteger( + options.disableAfterFailures ?? 10, 'disableAfterFailures', 1, 100 + ); + this.clock = options.clock ?? (() => new Date()); + this.random = options.random ?? Math.random; + this.pollMs = boundedInteger(options.pollMs ?? 1000, 'pollMs', 1, 60_000); + } + + /** The adapter holds the delivery and endpoint locks until this attempt commits. */ + async runOnce(now: Date = this.clock()): Promise { + return this.storage.processNext(now, async (delivery, endpoint): Promise => { + const attempt = delivery.attempt + 1; + let lastResponseCode: number | undefined; + let lastErrorCode: string | undefined; + try { + const body = JSON.stringify(redactWebhookPayload(delivery.payload)); + const timestamp = Math.floor(this.clock().getTime() / 1000); + const signature = webhookSignatureHeader( + endpoint, this.options.encryptionKey, timestamp, body + ); + const response = await this.transport(endpoint.url, body, { + 'X-Quittance-Signature': signature, + 'X-Quittance-Event-Id': delivery.eventId, + 'X-Quittance-Event-Type': delivery.eventType, + 'User-Agent': 'Quittance-Webhooks/1', + }); + if (Number.isInteger(response.statusCode)) { + lastResponseCode = response.statusCode; + } + if ( + !Number.isInteger(response.statusCode) || + response.statusCode < 200 || + response.statusCode >= 300 + ) { + lastErrorCode = 'HTTP_ERROR'; + } + } catch (error) { + lastErrorCode = error instanceof WebhookTransportError + ? error.code + : 'DELIVERY_FAILED'; + } + + const completed = this.clock(); + const completedAt = completed.toISOString(); + if (!lastErrorCode) { + return { + attempt, + status: 'delivered', + nextAttemptAt: completedAt, + completedAt, + lastResponseCode, + failureCount: 0, + disableEndpoint: false, + }; + } + + const failureCount = endpoint.failureCount + 1; + const unsafe = lastErrorCode === 'UNSAFE_URL' || lastErrorCode === 'UNSAFE_ADDRESS'; + const disableEndpoint = unsafe || failureCount >= this.disableAfterFailures; + const dead = disableEndpoint || attempt >= this.maxAttempts; + const random = this.random(); + const jitter = Number.isFinite(random) ? Math.min(1, Math.max(0, random)) : 0; + const delay = Math.round(monitorBackoffMs(attempt - 1) * (1 + jitter * 0.25)); + return { + attempt, + status: dead ? 'dead' : 'pending', + nextAttemptAt: dead ? completedAt : new Date(completed.getTime() + delay).toISOString(), + completedAt: dead ? completedAt : undefined, + lastResponseCode, + lastErrorCode, + failureCount, + disableEndpoint, + }; + }); + } + + private tick(): void { + if (!this.running || this.inFlight) return; + this.inFlight = (async () => { + try { + for (let count = 0; count < 25 && this.running; count += 1) { + if (!await this.runOnce()) break; + } + } catch { + emitOperationalFailure('webhook.worker'); + } + })().finally(() => { + this.inFlight = undefined; + }); + } + + start(): void { + if (this.running) return; + this.running = true; + this.timer = setInterval(() => this.tick(), this.pollMs); + this.timer.unref(); + this.tick(); + } + + async stop(): Promise { + this.running = false; + if (this.timer) clearInterval(this.timer); + this.timer = undefined; + await this.inFlight; + } +} + +function configuredInteger(name: string, fallback: number, min: number, max: number): number { + const raw = process.env[name]; + if (raw === undefined) return fallback; + if (!/^[0-9]+$/.test(raw)) { + throw new Error(name + ' must be an integer from ' + min + ' through ' + max); + } + return boundedInteger(Number(raw), name, min, max); +} + +export function startConfiguredWebhookWorker( + storage: WebhookStorage | undefined +): WebhookWorker | undefined { + if (process.env.WEBHOOKS_ENABLED !== 'true') return undefined; + if (!storage) throw new Error('WEBHOOKS_ENABLED requires webhook storage'); + const worker = new WebhookWorker(storage, { + encryptionKey: webhookEncryptionKey(), + maxAttempts: configuredInteger('WEBHOOK_MAX_ATTEMPTS', 8, 1, 20), + disableAfterFailures: configuredInteger('WEBHOOK_DISABLE_AFTER_FAILURES', 10, 1, 100), + }); + worker.start(); + return worker; +} diff --git a/backend/src/storage/invoice-storage.ts b/backend/src/storage/invoice-storage.ts index 97349d54..1b6e05cc 100644 --- a/backend/src/storage/invoice-storage.ts +++ b/backend/src/storage/invoice-storage.ts @@ -1,3 +1,4 @@ +import type { WebhookStorage } from './webhook-storage'; import { CreateInvoiceInput } from '../utils/validation'; import type { LatePaymentWarningCode, @@ -82,6 +83,8 @@ export interface MarkAsPaidOptions { export interface InvoiceStorage { /** Reported by /api/health so a running server tells you which backend it uses. */ readonly mode: string; + /** Seller webhooks share this adapter's persistence and invoice transactions. */ + readonly webhooks?: WebhookStorage; createInvoice(input: CreateInvoiceInput): Promise; getInvoiceById(id: string): Promise; diff --git a/backend/src/storage/memory-invoice-storage.ts b/backend/src/storage/memory-invoice-storage.ts index e7136e90..56c34265 100644 --- a/backend/src/storage/memory-invoice-storage.ts +++ b/backend/src/storage/memory-invoice-storage.ts @@ -8,6 +8,10 @@ export class MemoryInvoiceStorage implements InvoiceStorage { constructor(private readonly service: InvoiceMemoryService = new InvoiceMemoryService()) {} + get webhooks() { + return this.service.webhooks; + } + async createInvoice(input: CreateInvoiceInput): Promise { return this.service.createInvoice(input); } diff --git a/backend/src/storage/memory-storage.ts b/backend/src/storage/memory-storage.ts index 12930883..3d20d741 100644 --- a/backend/src/storage/memory-storage.ts +++ b/backend/src/storage/memory-storage.ts @@ -1,4 +1,6 @@ import { v4 as uuidv4 } from 'uuid'; +import { MemoryWebhookStorage } from './memory-webhook-storage'; +import { invoiceWebhookPayload } from './webhook-storage'; import { calculateInvoiceStats } from './invoice-stats'; import type { InvoiceStats } from './invoice-stats'; import { isPendingInvoiceExpired } from '../domain/invoice-expiry'; @@ -31,6 +33,8 @@ class MemoryStorage { private readonly paymentClaims = new PaymentClaimIndex(); private paymentEvents: MemoryPaymentEvent[] = []; + constructor(readonly webhooks: MemoryWebhookStorage = new MemoryWebhookStorage()) {} + createInvoice(data: Partial): Invoice { const invoice: Invoice = { id: data.id || uuidv4(), @@ -65,6 +69,10 @@ class MemoryStorage { throw new InvoiceIdCollisionError(invoice.id); } + const enqueue = this.webhooks.prepareEvent( + invoice.sellerPublicKey, + invoiceWebhookPayload('invoice.created', invoice, uuidv4(), invoice.createdAt) + ); this.invoices.set(invoice.id, invoice); this.invoicesByMemo.set(invoice.memo, invoice.id); if (invoice.idempotencyKey) { @@ -74,6 +82,7 @@ class MemoryStorage { ); } + enqueue(); return invoice; } @@ -116,11 +125,56 @@ class MemoryStorage { if (!invoice) return undefined; const updated = { ...invoice, ...updates }; - this.invoices.set(id, updated); - + const commit = this.prepareInvoiceChange(invoice, updated); + commit(); return updated; } + /** + * Prepare the audit and delivery rows before changing invoice state. The + * returned synchronous commit contains no validation, randomness, or I/O. + */ + private prepareInvoiceChange( + previous: Invoice, + updated: Invoice, + paidData?: Record + ): () => void { + if ( + previous.status === updated.status || + !['PAID', 'CANCELLED', 'EXPIRED'].includes(updated.status) + ) { + return () => { this.invoices.set(updated.id, updated); }; + } + const eventTypes = { + PAID: ['invoice.paid', 'PAYMENT_CONFIRMED'], + CANCELLED: ['invoice.cancelled', 'INVOICE_CANCELLED'], + EXPIRED: ['invoice.expired', 'INVOICE_EXPIRED'], + } as const; + const [webhookType, auditType] = eventTypes[updated.status as keyof typeof eventTypes]; + const at = updated.status === 'PAID' + ? updated.paidAt ?? new Date() + : updated.status === 'CANCELLED' + ? updated.cancelledAt ?? new Date() + : new Date(); + const eventId = uuidv4(); + const payload = invoiceWebhookPayload(webhookType, updated, eventId, at); + const enqueue = this.webhooks.prepareEvent(updated.sellerPublicKey, payload); + const audit: MemoryPaymentEvent = { + id: eventId, + invoiceId: updated.id, + eventType: auditType, + eventData: updated.status === 'PAID' + ? paidData ?? {} + : { priorStatus: previous.status, status: updated.status }, + createdAt: at, + }; + return () => { + this.invoices.set(updated.id, updated); + this.paymentEvents.push(audit); + enqueue(); + }; + } + // Cancel invoice cancelInvoice(id: string, sellerPublicKey?: string): Invoice | undefined { this.markExpiredInvoices(); @@ -155,17 +209,8 @@ class MemoryStorage { // Settlement time must come from the ledger close time; never invent one. const settlement = settlementFieldsForInvoice(invoice, options.settledAt); - // One transaction settles one invoice. The claim below reads and records in - // the same synchronous step, so a second caller holding the same hash gets a - // decision here rather than a second PAID transition. A replay against this - // same invoice falls back to the "already processed" contract above. - const decision = this.paymentClaims.claim(txHash, id, now); - if (decision.kind === 'conflict') { - throw new PaymentClaimError(txHash, id, decision.claim.invoiceId); - } - if (decision.kind === 'replay') return undefined; - - const updated = this.updateInvoice(id, { + const updated: Invoice = { + ...invoice, status: 'PAID', paymentTxHash: txHash, payerPublicKey, @@ -177,19 +222,25 @@ class MemoryStorage { settlementContext: settlement.settlementContext, priorStatus: settlement.priorStatus, latePaymentWarningCode: settlement.latePaymentWarningCode, + }; + const commit = this.prepareInvoiceChange(invoice, updated, { + txHash, + payerPublicKey, + settledAt: settlement.settledAt.toISOString(), + settlementContext: settlement.settlementContext, + priorStatus: settlement.priorStatus, + latePaymentWarningCode: settlement.latePaymentWarningCode, + ...(options.destinationMuxedId ? { destinationMuxedId: options.destinationMuxedId } : {}), }); - if (updated) { - this.logPaymentEvent(id, 'PAYMENT_CONFIRMED', { - txHash, - payerPublicKey, - settledAt: settlement.settledAt.toISOString(), - settlementContext: settlement.settlementContext, - priorStatus: settlement.priorStatus, - latePaymentWarningCode: settlement.latePaymentWarningCode, - ...(options.destinationMuxedId ? { destinationMuxedId: options.destinationMuxedId } : {}), - }); + // Prepare everything that can reject before claiming the transaction. + // One synchronous commit then records the claim, invoice, audit and outbox. + const decision = this.paymentClaims.claim(txHash, id, now); + if (decision.kind === 'conflict') { + throw new PaymentClaimError(txHash, id, decision.claim.invoiceId); } + if (decision.kind === 'replay') return undefined; + commit(); return updated; } @@ -214,14 +265,15 @@ class MemoryStorage { // Mark expired invoices markExpiredInvoices(now: Date = new Date()): number { - let count = 0; - + const commits: Array<() => void> = []; this.invoices.forEach((invoice) => { if (isPendingInvoiceExpired(invoice, now)) { - invoice.status = 'EXPIRED'; - count++; + commits.push(this.prepareInvoiceChange(invoice, { ...invoice, status: 'EXPIRED' })); } }); + // A failed preparation leaves the entire sweep unchanged. + for (const commit of commits) commit(); + const count = commits.length; if (count > 0) { console.log(`โฐ Marked ${count} invoices as expired`); @@ -234,13 +286,22 @@ class MemoryStorage { * Records a payment lifecycle audit event in memory. */ logPaymentEvent(invoiceId: string, eventType: string, eventData: any): void { - this.paymentEvents.push({ + const event: MemoryPaymentEvent = { id: uuidv4(), invoiceId, eventType, eventData, createdAt: new Date(), - }); + }; + const invoice = this.invoices.get(invoiceId); + const enqueue = invoice && (eventType === 'PAYMENT_REJECTED' || eventType === 'PARTIAL_PAYMENT') + ? this.webhooks.prepareEvent( + invoice.sellerPublicKey, + invoiceWebhookPayload('payment.rejected', invoice, event.id, event.createdAt, eventData) + ) + : undefined; + this.paymentEvents.push(event); + enqueue?.(); } /** @@ -260,6 +321,7 @@ class MemoryStorage { this.invoicesByIdempotencyKey.clear(); this.paymentClaims.clear(); this.paymentEvents = []; + this.webhooks.clear(); console.log('๐Ÿ—‘๏ธ Memory storage cleared'); } diff --git a/backend/src/storage/memory-webhook-storage.ts b/backend/src/storage/memory-webhook-storage.ts new file mode 100644 index 00000000..0ddeffa4 --- /dev/null +++ b/backend/src/storage/memory-webhook-storage.ts @@ -0,0 +1,183 @@ +import type { WebhookPayload } from '../../../shared/webhooks'; +import type { WebhookSecrets } from '../services/webhook-crypto'; +import { redactWebhookPayload } from '../utils/payment-event-redaction'; +import { + type DeliverWebhook, + type WebhookDelivery, + type WebhookEndpoint, + type WebhookStorage, + WebhookStoreError, + newWebhookDelivery, + testWebhookPayload, +} from './webhook-storage'; + +const copyEndpoint = (endpoint: WebhookEndpoint): WebhookEndpoint => ({ + ...endpoint, events: [...endpoint.events], +}); +const copyDelivery = (delivery: WebhookDelivery): WebhookDelivery => ({ + ...delivery, payload: redactWebhookPayload(delivery.payload), +}); + +/** Synchronous preparation/commit keeps invoices and their outbox atomic. */ +export class MemoryWebhookStorage implements WebhookStorage { + private readonly endpoints = new Map(); + private readonly deliveries = new Map(); + private readonly proofs = new Map(); + private readonly lockedEndpoints = new Set(); + + prepareEvent(sellerPublicKey: string, payload: WebhookPayload): () => void { + const clean = redactWebhookPayload(payload); + const prepared = [...this.endpoints.values()] + .filter(endpoint => + endpoint.sellerPublicKey === sellerPublicKey && + endpoint.enabled && !endpoint.deletedAt && endpoint.events.includes(clean.type) + ) + .map(endpoint => newWebhookDelivery(endpoint.id, clean)); + return () => { + for (const delivery of prepared) this.deliveries.set(delivery.id, delivery); + }; + } + + async register(endpoint: WebhookEndpoint): Promise { + const owned = [...this.endpoints.values()].filter( + value => value.sellerPublicKey === endpoint.sellerPublicKey && !value.deletedAt + ); + if (owned.length >= 5) throw new WebhookStoreError('ENDPOINT_LIMIT'); + if (this.endpoints.has(endpoint.id)) throw new Error('Webhook endpoint ID already exists'); + this.endpoints.set(endpoint.id, copyEndpoint(endpoint)); + } + + async listEndpoints(sellerPublicKey: string): Promise { + return [...this.endpoints.values()] + .filter(endpoint => endpoint.sellerPublicKey === sellerPublicKey && !endpoint.deletedAt) + .sort((a, b) => a.createdAt.localeCompare(b.createdAt) || a.id.localeCompare(b.id)) + .map(copyEndpoint); + } + + async listDeliveries(sellerPublicKey: string): Promise { + return [...this.deliveries.values()] + .filter(delivery => this.endpoints.get(delivery.endpointId)?.sellerPublicKey === sellerPublicKey) + .sort((a, b) => b.createdAt.localeCompare(a.createdAt) || b.id.localeCompare(a.id)) + .slice(0, 50) + .map(copyDelivery); + } + + async remove(sellerPublicKey: string, id: string, now: Date): Promise { + const endpoint = this.endpoints.get(id); + if (!endpoint || endpoint.sellerPublicKey !== sellerPublicKey || endpoint.deletedAt) return false; + const at = now.toISOString(); + endpoint.enabled = false; + endpoint.disabledAt = at; + endpoint.deletedAt = at; + for (const delivery of this.deliveries.values()) { + if (delivery.endpointId === id && delivery.status === 'pending') { + delivery.status = 'cancelled'; + delivery.completedAt = at; + delivery.lastErrorCode = 'ENDPOINT_DISABLED'; + } + } + return true; + } + + async rotate( + sellerPublicKey: string, + id: string, + secrets: WebhookSecrets, + now: Date, + overlapMs: number + ): Promise { + const endpoint = this.endpoints.get(id); + if (!endpoint || endpoint.sellerPublicKey !== sellerPublicKey || endpoint.deletedAt) return undefined; + if (endpoint.previousSecretExpiresAt && new Date(endpoint.previousSecretExpiresAt).getTime() > now.getTime()) { + throw new WebhookStoreError('ROTATION_IN_PROGRESS'); + } + const rotated: WebhookEndpoint = { + ...endpoint, + previousSecretHash: endpoint.secretHash, + previousSecretEncrypted: endpoint.secretEncrypted, + previousSecretExpiresAt: new Date(now.getTime() + overlapMs).toISOString(), + secretHash: secrets.secretHash, + secretEncrypted: secrets.secretEncrypted, + }; + this.endpoints.set(id, rotated); + return copyEndpoint(rotated); + } + + async enqueueTest(sellerPublicKey: string, id: string, eventId: string, now: Date): Promise { + const endpoint = this.endpoints.get(id); + if (!endpoint || endpoint.sellerPublicKey !== sellerPublicKey || endpoint.deletedAt || !endpoint.enabled) return false; + const cutoff = now.getTime() - 60_000; + if ([...this.deliveries.values()].some(delivery => + delivery.endpointId === id && delivery.payload.test === true && + new Date(delivery.createdAt).getTime() > cutoff + )) { + throw new WebhookStoreError('TEST_RATE_LIMIT'); + } + const delivery = newWebhookDelivery(id, testWebhookPayload(eventId, now)); + this.deliveries.set(delivery.id, delivery); + return true; + } + + async consumeProof(sellerPublicKey: string, nonce: string, expiresAt: Date, now: Date): Promise { + const timestamp = now.getTime(); + for (const [key, expiry] of this.proofs) { + if (expiry <= timestamp) this.proofs.delete(key); + } + if (expiresAt.getTime() <= timestamp) return false; + const key = sellerPublicKey + ':' + nonce; + if (this.proofs.has(key)) return false; + this.proofs.set(key, expiresAt.getTime()); + return true; + } + + async processNext(now: Date, deliver: DeliverWebhook): Promise { + const next = [...this.deliveries.values()] + .filter(delivery => { + const endpoint = this.endpoints.get(delivery.endpointId); + return delivery.status === 'pending' && + new Date(delivery.nextAttemptAt).getTime() <= now.getTime() && + endpoint?.enabled && !endpoint.deletedAt && !this.lockedEndpoints.has(endpoint.id); + }) + .sort((a, b) => a.nextAttemptAt.localeCompare(b.nextAttemptAt) || a.id.localeCompare(b.id))[0]; + if (!next) return false; + const endpoint = this.endpoints.get(next.endpointId)!; + this.lockedEndpoints.add(endpoint.id); + try { + const result = await deliver(copyDelivery(next), copyEndpoint(endpoint)); + // Removal is synchronous and can happen while delivery is in flight. + // Preserve it rather than restoring the endpoint snapshot passed to I/O. + const current = this.endpoints.get(endpoint.id)!; + const removed = Boolean(current.deletedAt); + Object.assign(next, { + attempt: result.attempt, + status: removed ? 'cancelled' : result.status, + nextAttemptAt: result.nextAttemptAt, + lastResponseCode: result.lastResponseCode, + lastErrorCode: removed ? 'ENDPOINT_DISABLED' : result.lastErrorCode, + completedAt: removed ? current.deletedAt : result.completedAt, + }); + current.failureCount = result.failureCount; + if (result.disableEndpoint && !removed) { + current.enabled = false; + current.disabledAt = now.toISOString(); + for (const delivery of this.deliveries.values()) { + if (delivery.endpointId === current.id && delivery.status === 'pending') { + delivery.status = 'dead'; + delivery.completedAt = now.toISOString(); + delivery.lastErrorCode = 'ENDPOINT_DISABLED'; + } + } + } + return true; + } finally { + this.lockedEndpoints.delete(endpoint.id); + } + } + + clear(): void { + this.endpoints.clear(); + this.deliveries.clear(); + this.proofs.clear(); + this.lockedEndpoints.clear(); + } +} diff --git a/backend/src/storage/postgres-invoice-storage.ts b/backend/src/storage/postgres-invoice-storage.ts index 8c4d6081..1d467970 100644 --- a/backend/src/storage/postgres-invoice-storage.ts +++ b/backend/src/storage/postgres-invoice-storage.ts @@ -19,6 +19,10 @@ export class PostgresInvoiceStorage implements InvoiceStorage { constructor(private readonly service: InvoiceService = invoiceService) {} + get webhooks() { + return this.service.webhooks; + } + async createInvoice(input: CreateInvoiceInput): Promise { return this.service.createInvoice(input); } diff --git a/backend/src/storage/postgres-webhook-storage.ts b/backend/src/storage/postgres-webhook-storage.ts new file mode 100644 index 00000000..6e5eba5e --- /dev/null +++ b/backend/src/storage/postgres-webhook-storage.ts @@ -0,0 +1,277 @@ +import type { Pool, PoolClient } from 'pg'; +import type { WebhookSecrets } from '../services/webhook-crypto'; +import { redactWebhookPayload } from '../utils/payment-event-redaction'; +import { + type DeliverWebhook, + type WebhookDelivery, + type WebhookEndpoint, + type WebhookStorage, + WebhookStoreError, + newWebhookDelivery, + testWebhookPayload, +} from './webhook-storage'; + +const iso = (value: string | Date): string => new Date(value).toISOString(); +const optionalIso = (value: string | Date | null | undefined): string | undefined => + value == null ? undefined : iso(value); + +function mapEndpoint(row: any): WebhookEndpoint { + return { + id: row.id, + sellerPublicKey: row.seller_public_key, + url: row.url, + events: [...row.events], + secretHash: row.secret_hash, + secretEncrypted: row.secret_encrypted, + previousSecretHash: row.previous_secret_hash ?? undefined, + previousSecretEncrypted: row.previous_secret_encrypted ?? undefined, + previousSecretExpiresAt: optionalIso(row.previous_secret_expires_at), + enabled: row.enabled, + failureCount: Number(row.failure_count), + createdAt: iso(row.created_at), + disabledAt: optionalIso(row.disabled_at), + deletedAt: optionalIso(row.deleted_at), + }; +} + +function mapDelivery(row: any): WebhookDelivery { + return { + id: row.id, + eventId: row.event_id, + endpointId: row.endpoint_id, + eventType: row.event_type, + payload: redactWebhookPayload(row.payload), + attempt: Number(row.attempt), + status: row.status, + nextAttemptAt: iso(row.next_attempt_at), + lastResponseCode: row.last_response_code ?? undefined, + lastErrorCode: row.last_error_code ?? undefined, + createdAt: iso(row.created_at), + completedAt: optionalIso(row.completed_at), + }; +} + +/** + * Invoice triggers enqueue inside the invoice transaction. Delivery holds both + * the endpoint and delivery row locks across bounded I/O, so another process + * cannot post the same row or rotate/remove its signing key mid-attempt. + */ +export class PostgresWebhookStorage implements WebhookStorage { + constructor(private readonly pool: Pool) {} + + private async transaction(work: (client: PoolClient) => Promise): Promise { + const client = await this.pool.connect(); + let failedRollback = false; + try { + await client.query('BEGIN'); + const result = await work(client); + await client.query('COMMIT'); + return result; + } catch (error) { + try { + await client.query('ROLLBACK'); + } catch { + failedRollback = true; + } + throw error; + } finally { + client.release(failedRollback); + } + } + + async register(endpoint: WebhookEndpoint): Promise { + await this.transaction(async client => { + // Serialize each seller's count-and-insert across application processes. + await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 584))', [endpoint.sellerPublicKey]); + const count = await client.query( + 'SELECT COUNT(*)::integer AS count FROM webhook_endpoints WHERE seller_public_key = $1 AND deleted_at IS NULL', + [endpoint.sellerPublicKey] + ); + if (Number(count.rows[0].count) >= 5) throw new WebhookStoreError('ENDPOINT_LIMIT'); + await client.query( + `INSERT INTO webhook_endpoints ( + id, seller_public_key, url, events, secret_hash, secret_encrypted, + previous_secret_hash, previous_secret_encrypted, previous_secret_expires_at, + enabled, failure_count, created_at, disabled_at, deleted_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, + [ + endpoint.id, endpoint.sellerPublicKey, endpoint.url, endpoint.events, + endpoint.secretHash, endpoint.secretEncrypted, endpoint.previousSecretHash ?? null, + endpoint.previousSecretEncrypted ?? null, endpoint.previousSecretExpiresAt ?? null, + endpoint.enabled, endpoint.failureCount, endpoint.createdAt, + endpoint.disabledAt ?? null, endpoint.deletedAt ?? null, + ] + ); + }); + } + + async listEndpoints(sellerPublicKey: string): Promise { + const result = await this.pool.query( + `SELECT * FROM webhook_endpoints + WHERE seller_public_key = $1 AND deleted_at IS NULL + ORDER BY created_at ASC, id ASC`, + [sellerPublicKey] + ); + return result.rows.map(mapEndpoint); + } + + async listDeliveries(sellerPublicKey: string): Promise { + const result = await this.pool.query( + `SELECT d.* FROM webhook_deliveries d + JOIN webhook_endpoints e ON e.id = d.endpoint_id + WHERE e.seller_public_key = $1 + ORDER BY d.created_at DESC, d.id DESC LIMIT 50`, + [sellerPublicKey] + ); + return result.rows.map(mapDelivery); + } + + async remove(sellerPublicKey: string, id: string, now: Date): Promise { + return this.transaction(async client => { + const endpoint = await client.query( + `UPDATE webhook_endpoints + SET enabled = false, disabled_at = $3, deleted_at = $3 + WHERE id = $2 AND seller_public_key = $1 AND deleted_at IS NULL RETURNING id`, + [sellerPublicKey, id, now] + ); + if (!endpoint.rows.length) return false; + await client.query( + `UPDATE webhook_deliveries + SET status = 'cancelled', completed_at = $2, last_error_code = 'ENDPOINT_DISABLED' + WHERE endpoint_id = $1 AND status = 'pending'`, + [id, now] + ); + return true; + }); + } + + async rotate( + sellerPublicKey: string, + id: string, + secrets: WebhookSecrets, + now: Date, + overlapMs: number + ): Promise { + return this.transaction(async client => { + const found = await client.query( + `SELECT * FROM webhook_endpoints + WHERE seller_public_key = $1 AND id = $2 AND deleted_at IS NULL FOR UPDATE`, + [sellerPublicKey, id] + ); + if (!found.rows.length) return undefined; + const current = mapEndpoint(found.rows[0]); + if (current.previousSecretExpiresAt && new Date(current.previousSecretExpiresAt).getTime() > now.getTime()) { + throw new WebhookStoreError('ROTATION_IN_PROGRESS'); + } + const result = await client.query( + `UPDATE webhook_endpoints + SET previous_secret_hash = secret_hash, + previous_secret_encrypted = secret_encrypted, + previous_secret_expires_at = $5, + secret_hash = $3, secret_encrypted = $4 + WHERE id = $2 AND seller_public_key = $1 RETURNING *`, + [sellerPublicKey, id, secrets.secretHash, secrets.secretEncrypted, new Date(now.getTime() + overlapMs)] + ); + return mapEndpoint(result.rows[0]); + }); + } + + async enqueueTest(sellerPublicKey: string, id: string, eventId: string, now: Date): Promise { + return this.transaction(async client => { + const endpoint = await client.query( + `SELECT id FROM webhook_endpoints + WHERE seller_public_key = $1 AND id = $2 AND enabled AND deleted_at IS NULL FOR UPDATE`, + [sellerPublicKey, id] + ); + if (!endpoint.rows.length) return false; + const recent = await client.query( + `SELECT id FROM webhook_deliveries + WHERE endpoint_id = $1 AND payload->>'test' = 'true' + AND created_at > $2::timestamptz - INTERVAL '1 minute' LIMIT 1`, + [id, now] + ); + if (recent.rows.length) throw new WebhookStoreError('TEST_RATE_LIMIT'); + const delivery = newWebhookDelivery(id, testWebhookPayload(eventId, now)); + await client.query( + `INSERT INTO webhook_deliveries + (id, endpoint_id, event_id, event_type, payload, attempt, next_attempt_at, status, created_at) + VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7,$8,$9)`, + [ + delivery.id, delivery.endpointId, delivery.eventId, delivery.eventType, + JSON.stringify(delivery.payload), delivery.attempt, delivery.nextAttemptAt, + delivery.status, delivery.createdAt, + ] + ); + return true; + }); + } + + async consumeProof(sellerPublicKey: string, nonce: string, expiresAt: Date, now: Date): Promise { + if (expiresAt.getTime() <= now.getTime()) return false; + return this.transaction(async client => { + await client.query('DELETE FROM webhook_proofs WHERE expires_at <= $1', [now]); + const result = await client.query( + `INSERT INTO webhook_proofs (seller_public_key, nonce, expires_at) VALUES ($1,$2,$3) + ON CONFLICT (seller_public_key, nonce) DO NOTHING RETURNING nonce`, + [sellerPublicKey, nonce, expiresAt] + ); + return result.rows.length > 0; + }); + } + + async processNext(now: Date, deliver: DeliverWebhook): Promise { + return this.transaction(async client => { + // Retire pending rows for endpoints disabled or removed outside a worker. + await client.query( + `UPDATE webhook_deliveries d + SET status = CASE WHEN e.deleted_at IS NOT NULL THEN 'cancelled' ELSE 'dead' END, + completed_at = $1, last_error_code = 'ENDPOINT_DISABLED' + FROM webhook_endpoints e + WHERE d.endpoint_id = e.id AND d.status = 'pending' + AND (NOT e.enabled OR e.deleted_at IS NOT NULL)`, + [now] + ); + const result = await client.query( + `SELECT d.*, to_jsonb(e) AS endpoint + FROM webhook_deliveries d JOIN webhook_endpoints e ON e.id = d.endpoint_id + WHERE d.status = 'pending' AND d.next_attempt_at <= $1 + AND e.enabled AND e.deleted_at IS NULL + ORDER BY d.next_attempt_at ASC, d.id ASC + LIMIT 1 FOR UPDATE OF d, e SKIP LOCKED`, + [now] + ); + if (!result.rows.length) return false; + const row = result.rows[0]; + const delivery = mapDelivery(row); + const endpoint = mapEndpoint(row.endpoint); + const attempt = await deliver(delivery, endpoint); + await client.query( + `UPDATE webhook_deliveries + SET attempt = $2, status = $3, next_attempt_at = $4, + last_response_code = $5, last_error_code = $6, completed_at = $7 + WHERE id = $1`, + [ + delivery.id, attempt.attempt, attempt.status, attempt.nextAttemptAt, + attempt.lastResponseCode ?? null, attempt.lastErrorCode ?? null, attempt.completedAt ?? null, + ] + ); + await client.query( + `UPDATE webhook_endpoints + SET failure_count = $2, + enabled = CASE WHEN $3 THEN false ELSE enabled END, + disabled_at = CASE WHEN $3 THEN COALESCE(disabled_at, $4) ELSE disabled_at END + WHERE id = $1`, + [endpoint.id, attempt.failureCount, attempt.disableEndpoint, now] + ); + if (attempt.disableEndpoint) { + await client.query( + `UPDATE webhook_deliveries + SET status = 'dead', completed_at = $2, last_error_code = 'ENDPOINT_DISABLED' + WHERE endpoint_id = $1 AND status = 'pending'`, + [endpoint.id, now] + ); + } + return true; + }); + } +} diff --git a/backend/src/storage/webhook-storage.ts b/backend/src/storage/webhook-storage.ts new file mode 100644 index 00000000..f44b6465 --- /dev/null +++ b/backend/src/storage/webhook-storage.ts @@ -0,0 +1,150 @@ +import { randomUUID } from 'node:crypto'; +import type { WebhookEventType, WebhookPayload } from '../../../shared/webhooks'; +import type { WebhookSecrets } from '../services/webhook-crypto'; +import type { StoredInvoice } from './invoice-storage'; +import { canonicalAmount } from '../utils/safe-amount-compare'; +import { redactWebhookPayload } from '../utils/payment-event-redaction'; + +export interface WebhookEndpoint extends WebhookSecrets { + id: string; + sellerPublicKey: string; + url: string; + events: WebhookEventType[]; + enabled: boolean; + failureCount: number; + createdAt: string; + disabledAt?: string; + deletedAt?: string; +} + +export interface WebhookDelivery { + id: string; + eventId: string; + endpointId: string; + eventType: WebhookEventType; + payload: WebhookPayload; + attempt: number; + status: 'pending' | 'delivered' | 'dead' | 'cancelled'; + nextAttemptAt: string; + lastResponseCode?: number; + lastErrorCode?: string; + createdAt: string; + completedAt?: string; +} + +export type WebhookAttemptResult = Pick< + WebhookDelivery, + 'attempt' | 'status' | 'nextAttemptAt' | 'lastResponseCode' | 'lastErrorCode' | 'completedAt' +> & { failureCount: number; disableEndpoint: boolean }; + +export type DeliverWebhook = ( + delivery: WebhookDelivery, + endpoint: WebhookEndpoint +) => Promise; + +export class WebhookStoreError extends Error { + constructor( + readonly code: 'ENDPOINT_LIMIT' | 'ENDPOINT_NOT_FOUND' | 'ROTATION_IN_PROGRESS' | 'TEST_RATE_LIMIT' + ) { + super(code); + this.name = 'WebhookStoreError'; + } +} + +export interface WebhookStorage { + register(endpoint: WebhookEndpoint): Promise; + listEndpoints(sellerPublicKey: string): Promise; + listDeliveries(sellerPublicKey: string): Promise; + remove(sellerPublicKey: string, id: string, now: Date): Promise; + rotate( + sellerPublicKey: string, + id: string, + secrets: WebhookSecrets, + now: Date, + overlapMs: number + ): Promise; + enqueueTest(sellerPublicKey: string, id: string, eventId: string, now: Date): Promise; + consumeProof(sellerPublicKey: string, nonce: string, expiresAt: Date, now: Date): Promise; + processNext(now: Date, deliver: DeliverWebhook): Promise; +} + +export function publicWebhookEndpoint(endpoint: WebhookEndpoint) { + return { + id: endpoint.id, + url: endpoint.url, + events: [...endpoint.events], + enabled: endpoint.enabled, + failureCount: endpoint.failureCount, + createdAt: endpoint.createdAt, + ...(endpoint.disabledAt ? { disabledAt: endpoint.disabledAt } : {}), + }; +} + +export function publicWebhookDelivery(delivery: WebhookDelivery) { + return { + id: delivery.id, + eventId: delivery.eventId, + endpointId: delivery.endpointId, + eventType: delivery.eventType, + attempt: delivery.attempt, + status: delivery.status, + nextAttemptAt: delivery.nextAttemptAt, + createdAt: delivery.createdAt, + ...(delivery.lastResponseCode !== undefined ? { lastResponseCode: delivery.lastResponseCode } : {}), + ...(delivery.lastErrorCode ? { lastErrorCode: delivery.lastErrorCode } : {}), + ...(delivery.completedAt ? { completedAt: delivery.completedAt } : {}), + }; +} + +export function invoiceWebhookPayload( + type: WebhookEventType, + invoice: StoredInvoice, + eventId: string = randomUUID(), + at = new Date(), + payment?: Record +): WebhookPayload { + return redactWebhookPayload({ + version: 1, + id: eventId, + type, + createdAt: at.toISOString(), + invoice: { + id: invoice.id, + amount: canonicalAmount(invoice.amount), + assetCode: invoice.assetCode, + assetIssuer: invoice.assetIssuer, + status: invoice.status, + paymentTxHash: invoice.paymentTxHash, + settledAt: invoice.settledAt?.toISOString(), + settlementContext: invoice.settlementContext, + priorStatus: invoice.priorStatus, + latePaymentWarningCode: invoice.latePaymentWarningCode, + expiresAt: invoice.expiresAt.toISOString(), + }, + payment, + }); +} + +export function testWebhookPayload(id: string, at: Date): WebhookPayload { + return redactWebhookPayload({ + version: 1, + id, + type: 'invoice.created', + createdAt: at.toISOString(), + test: true, + }); +} + +export function newWebhookDelivery(endpointId: string, payload: WebhookPayload): WebhookDelivery { + return { + id: randomUUID(), + eventId: payload.id, + endpointId, + eventType: payload.type, + payload, + attempt: 0, + status: 'pending', + nextAttemptAt: payload.createdAt, + createdAt: payload.createdAt, + }; +} diff --git a/backend/src/utils/payment-event-redaction.ts b/backend/src/utils/payment-event-redaction.ts index 0c1bf95f..e1263b8b 100644 --- a/backend/src/utils/payment-event-redaction.ts +++ b/backend/src/utils/payment-event-redaction.ts @@ -23,3 +23,86 @@ export function redactPaymentEventData( } return clean; } + + +/** + * Webhooks cross a seller-controlled network boundary. Build a new object from + * approved fields, rather than relying on the internal audit feed's denylist. + */ +export function redactWebhookPayload(input: unknown): import('../../../shared/webhooks').WebhookPayload { + const object = (value: unknown): Record | undefined => + value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined; + const iso = (value: unknown): string | undefined => { + if (typeof value !== 'string') return undefined; + const parsed = new Date(value); + return Number.isFinite(parsed.getTime()) ? parsed.toISOString() : undefined; + }; + const data = object(input); + const eventTypes = [ + 'invoice.created', 'invoice.paid', 'invoice.cancelled', 'invoice.expired', 'payment.rejected', + ]; + const createdAt = iso(data?.createdAt); + if ( + !data || data.version !== 1 || typeof data.id !== 'string' || !data.id || + typeof data.type !== 'string' || !eventTypes.includes(data.type) || !createdAt + ) { + throw new Error('Invalid webhook payload'); + } + const output: Record = { + version: 1, + id: data.id, + type: data.type, + createdAt, + }; + const source = object(data.invoice); + if (source) { + const expiresAt = iso(source.expiresAt); + if ( + typeof source.id !== 'string' || !source.id || + typeof source.amount !== 'string' || + typeof source.assetCode !== 'string' || + typeof source.status !== 'string' || + !['PENDING', 'PAID', 'CANCELLED', 'EXPIRED'].includes(source.status) || + !expiresAt + ) { + throw new Error('Invalid webhook invoice'); + } + const invoice: Record = { + id: source.id, + amount: source.amount, + assetCode: source.assetCode, + status: source.status, + expiresAt, + }; + if (typeof source.assetIssuer === 'string') invoice.assetIssuer = source.assetIssuer; + if (typeof source.paymentTxHash === 'string' && /^[a-fA-F0-9]{64}$/.test(source.paymentTxHash)) { + invoice.paymentTxHash = source.paymentTxHash; + } + const settledAt = iso(source.settledAt); + if (settledAt) invoice.settledAt = settledAt; + for (const key of ['settlementContext', 'priorStatus', 'latePaymentWarningCode']) { + const value = source[key]; + if (typeof value === 'string' && /^[A-Z_]{1,60}$/.test(value)) { + invoice[key] = value; + } + } + output.invoice = invoice; + } else if (data.test !== true) { + throw new Error('Webhook invoice is required'); + } + const sourcePayment = object(data.payment); + if (sourcePayment) { + const payment: Record = {}; + if (typeof sourcePayment.code === 'string' && /^[A-Z0-9_]{1,100}$/.test(sourcePayment.code)) { + payment.code = sourcePayment.code; + } + if (typeof sourcePayment.txHash === 'string' && /^[a-fA-F0-9]{64}$/.test(sourcePayment.txHash)) { + payment.txHash = sourcePayment.txHash; + } + if (Object.keys(payment).length > 0) output.payment = payment; + } + if (data.test === true) output.test = true; + return output as unknown as import('../../../shared/webhooks').WebhookPayload; +} diff --git a/backend/tests/fixtures/webhook-postgres-worker.ts b/backend/tests/fixtures/webhook-postgres-worker.ts new file mode 100644 index 00000000..fe902fed --- /dev/null +++ b/backend/tests/fixtures/webhook-postgres-worker.ts @@ -0,0 +1,112 @@ +import { Pool } from 'pg'; +import { PostgresWebhookStorage } from '../../src/storage/postgres-webhook-storage'; + +// One real database transaction per child. The parent controls only when a +// claimed delivery may finish; no webhook request or payload crosses IPC. +let pool: Pool | undefined; +let closing: Promise | undefined; +let running = false; +let finished = false; +let releaseHold: (() => void) | undefined; + +function send(message: Record): Promise { + return new Promise((resolve, reject) => { + if (!process.send || !process.connected) { + reject(Object.assign(new Error('Worker requires IPC'), { code: 'IPC_UNAVAILABLE' })); + return; + } + process.send(message, error => error ? reject(error) : resolve()); + }); +} + +function closePool(): Promise { + return closing ??= pool ? pool.end() : Promise.resolve(); +} + +async function fail(error: unknown): Promise { + if (finished) return; + finished = true; + process.exitCode = 1; + const code = (error as { code?: unknown } | null)?.code; + // Database messages can contain connection details or row values. Report + // only a stable explanation and the diagnostic code, never the raw error. + await send({ + type: 'error', + message: 'PostgreSQL webhook worker failed', + ...(typeof code === 'string' && /^[A-Z0-9_]{1,40}$/.test(code) ? { code } : {}), + }).catch(() => {}); + await closePool().catch(() => {}); + if (process.connected) process.disconnect(); +} + +async function main(): Promise { + const schema = process.env.Q584_PG_SCHEMA; + const connectionString = process.env.DATABASE_URL; + if (!schema || !/^q584_webhooks_[a-f0-9]+$/.test(schema) || !connectionString) { + throw Object.assign(new Error('Invalid worker configuration'), { code: 'INVALID_WORKER_CONFIG' }); + } + pool = new Pool({ + connectionString, + max: 1, + options: `-c search_path=${schema}`, + application_name: `q584-worker-${process.pid}`, + }); + pool.on('error', error => { void fail(error); }); + const storage = new PostgresWebhookStorage(pool); + + async function run(command: Record): Promise { + const mode = command.mode; + const rawNow = command.now; + if (typeof rawNow !== 'string' || (mode !== 'hold' && mode !== 'finish')) { + throw Object.assign(new Error('Invalid run command'), { code: 'INVALID_WORKER_COMMAND' }); + } + const now = new Date(rawNow); + if (!Number.isFinite(now.getTime())) { + throw Object.assign(new Error('Invalid run time'), { code: 'INVALID_WORKER_COMMAND' }); + } + const timestamp = now.toISOString(); + const processed = await storage.processNext(now, async (delivery, _endpoint) => { + // Register release before announcing the claim: the parent may reply + // immediately, before the send callback has completed in this child. + const held = mode === 'hold' + ? new Promise(resolve => { releaseHold = resolve; }) + : undefined; + await send({ + type: 'claimed', pid: process.pid, id: delivery.id, + eventId: delivery.eventId, attempt: delivery.attempt, + }); + if (held) await held; + releaseHold = undefined; + return { + attempt: delivery.attempt + 1, + status: 'delivered', + nextAttemptAt: timestamp, + lastResponseCode: 200, + completedAt: timestamp, + failureCount: 0, + disableEndpoint: false, + }; + }); + // processNext returns only after COMMIT. Close the connection before the + // receipt so a parent can inspect durable state without a lingering pool. + await closePool(); + await send({ type: 'done', pid: process.pid, processed }); + finished = true; + if (process.connected) process.disconnect(); + } + + process.on('message', message => { + if (finished || !message || typeof message !== 'object') return; + const command = message as Record; + if (command.type === 'release') { + releaseHold?.(); + return; + } + if (command.type !== 'run' || running) return; + running = true; + void run(command).catch(fail); + }); + await send({ type: 'ready', pid: process.pid }); +} + +void main().catch(fail); diff --git a/backend/tests/invoice-cancel-payment-race.test.ts b/backend/tests/invoice-cancel-payment-race.test.ts index 376b21f0..3da90d87 100644 --- a/backend/tests/invoice-cancel-payment-race.test.ts +++ b/backend/tests/invoice-cancel-payment-race.test.ts @@ -595,9 +595,10 @@ describe('cancel versus payment monitor attribution on memory storage', () => { assert.equal(stored?.paymentTxHash, undefined); const events = await invoiceService.getPaymentEvents(invoice.id); - assert.equal(events.length, 1); - assert.equal(events[0].eventType, 'PARTIAL_PAYMENT'); - assert.equal(events[0].eventData.code, 'AMOUNT_TOO_LOW'); + assert.equal(events.length, 2); + assert.equal(events[0].eventType, 'INVOICE_CANCELLED'); + assert.equal(events[1].eventType, 'PARTIAL_PAYMENT'); + assert.equal(events[1].eventData.code, 'AMOUNT_TOO_LOW'); }); }); diff --git a/backend/tests/stellar-service.test.ts b/backend/tests/stellar-service.test.ts index 772a74e5..5c20ebd4 100644 --- a/backend/tests/stellar-service.test.ts +++ b/backend/tests/stellar-service.test.ts @@ -1,6 +1,25 @@ import assert from 'node:assert/strict'; -import { describe, it } from 'node:test'; -import stellarService from '../src/services/stellar.service'; +import { after, before, describe, it } from 'node:test'; +import http from 'node:http'; +import type { AddressInfo } from 'node:net'; + +let stellarService: typeof import('../src/services/stellar.service')['default']; +let horizon: http.Server; +before(async () => { + // The CI network is intentionally offline. A missing transaction is a real + // HTTP404, not a connection refusal; exercise that SDK path locally. + horizon = http.createServer((_req, res) => { + res.writeHead(404, { 'content-type': 'application/problem+json' }); + res.end(JSON.stringify({ status: 404, title: 'Resource Missing' })); + }); + await new Promise(resolve => horizon.listen(0, '127.0.0.1', resolve)); + process.env.STELLAR_HORIZON_URL = `http://127.0.0.1:${(horizon.address() as AddressInfo).port}`; + ({ default: stellarService } = await import('../src/services/stellar.service')); +}); +after(async () => { + horizon.closeAllConnections(); + await new Promise(resolve => horizon.close(() => resolve())); +}); describe('StellarService - Horizon Read Consolidation', () => { it('exposes loadAccount, getBalance, verifyPayment, getTransaction, streamPayments, getRecentPayments, sendPayment', () => { diff --git a/backend/tests/webhook-delivery.test.ts b/backend/tests/webhook-delivery.test.ts new file mode 100644 index 00000000..b9d797c9 --- /dev/null +++ b/backend/tests/webhook-delivery.test.ts @@ -0,0 +1,289 @@ +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import { randomUUID } from 'node:crypto'; +import { createServer } from 'node:http'; +import { test } from 'node:test'; +import express from 'express'; +import { Keypair } from '@stellar/stellar-sdk'; +import { WEBHOOK_EVENT_TYPES, webhookProofMessage, type WebhookAction, type WebhookPayload } from '../../shared/webhooks.ts'; +import { MemoryStorage } from '../src/storage/memory-storage.ts'; +import { MemoryWebhookStorage } from '../src/storage/memory-webhook-storage.ts'; +import type { WebhookEndpoint } from '../src/storage/webhook-storage.ts'; +import { createWebhookRouter } from '../src/routes/webhook.routes.ts'; +import { newWebhookSecret, openWebhookSecret, sealWebhookSecret, verifyWebhookSignature, webhookSignature, webhookSignatureHeader } from '../src/services/webhook-crypto.ts'; +import { createWebhookTransport, isPublicWebhookAddress, validateWebhookUrl } from '../src/services/webhook-transport.ts'; +import { WebhookWorker } from '../src/services/webhook-worker.ts'; + +// Public test key, never deployment configuration. +const KEY = Buffer.alloc(32, 0x42); +const seller = Keypair.random(); +function endpoint(overrides: Partial = {}): WebhookEndpoint { + return { id: randomUUID(), sellerPublicKey: seller.publicKey(), url: 'https://receiver.example/webhooks', + events: [...WEBHOOK_EVENT_TYPES], enabled: true, failureCount: 0, createdAt: new Date().toISOString(), + ...sealWebhookSecret('whsec_public_test', KEY), ...overrides }; +} +function invoice(store: MemoryStorage) { + return store.createInvoice({ id: randomUUID(), memo: 'PRIVATE_MEMO_' + randomUUID(), + sellerPublicKey: seller.publicKey(), amount: 12.5, assetCode: 'XLM', + customerName: 'Private Customer', customerEmail: 'customer@private.example', + sellerEmail: 'seller@private.example', description: 'PRIVATE_DESCRIPTION', + metadata: { nested: { email: 'nested@private.example' } }, expiresAt: new Date(Date.now() + 86_400_000) }); +} + +test('documented HMAC vector verifies raw bytes, rejects stale timestamps and tampering', () => { + const secret = 'whsec_test_0123456789abcdef'; + const body = '{"id":"evt_1","type":"invoice.paid"}'; + const timestamp = 1735689600; + const expected = '7489caff2d92322967b02be31d80ec2a16eb05ba167bbd9719f630fc84eedd07'; + assert.equal(webhookSignature(secret, timestamp, body), expected); + const header = 't=' + timestamp + ',v1=' + expected; + assert.equal(verifyWebhookSignature(secret, header, Buffer.from(body), timestamp * 1000), true); + assert.equal(verifyWebhookSignature(secret, header, body + ' ', timestamp * 1000), false); + assert.equal(verifyWebhookSignature(secret, header, body, (timestamp + 301) * 1000), false); + assert.equal(verifyWebhookSignature(secret, 't=' + timestamp + ',' + header, body, timestamp * 1000), false); +}); + +test('encrypted rotation signs with both keys only within the promised overlap', async () => { + const storage = new MemoryWebhookStorage(); + const old = newWebhookSecret(KEY); + const original = endpoint({ secretHash: old.secretHash, secretEncrypted: old.secretEncrypted }); + await storage.register(original); + assert.equal(JSON.stringify(await storage.listEndpoints(seller.publicKey())).includes(old.secret), false); + assert.equal(openWebhookSecret(old.secretEncrypted, old.secretHash, KEY), old.secret); + const fresh = newWebhookSecret(KEY); + const now = new Date('2030-01-01T00:00:00Z'); + const rotated = await storage.rotate(seller.publicKey(), original.id, fresh, now, 86_400_000); + assert.ok(rotated); + const timestamp = now.getTime() / 1000; + const header = webhookSignatureHeader(rotated, KEY, timestamp, '{}'); + assert.equal(verifyWebhookSignature(old.secret, header, '{}', now.getTime()), true); + assert.equal(verifyWebhookSignature(fresh.secret, header, '{}', now.getTime()), true); + await assert.rejects(storage.rotate(seller.publicKey(), original.id, newWebhookSecret(KEY), now, 86_400_000), /ROTATION_IN_PROGRESS/); + const later = timestamp + 86_401; + const laterHeader = webhookSignatureHeader(rotated, KEY, later, '{}'); + assert.equal(verifyWebhookSignature(old.secret, laterHeader, '{}', later * 1000), false); + assert.equal(verifyWebhookSignature(fresh.secret, laterHeader, '{}', later * 1000), true); +}); + +test('registration and delivery SSRF checks cover literals, mapped addresses and changed DNS', async () => { + for (const address of ['127.0.0.1', '10.1.2.3', '169.254.169.254', '172.16.1.2', '192.168.1.2', '::1', '::ffff:127.0.0.1', 'fd00::1', 'fe80::1']) { + assert.equal(isPublicWebhookAddress(address), false, address); + await assert.rejects(validateWebhookUrl('https://' + (address.includes(':') ? '[' + address + ']' : address) + '/hook'), /UNSAFE_ADDRESS/); + } + await assert.rejects(validateWebhookUrl('https://2130706433/hook'), /UNSAFE_ADDRESS/); + await assert.rejects(validateWebhookUrl('http://public.example/hook'), /UNSAFE_URL/); + await assert.rejects(validateWebhookUrl('https://name:password@public.example/hook'), /UNSAFE_URL/); + assert.equal(isPublicWebhookAddress('93.184.216.34'), true); + assert.equal(isPublicWebhookAddress('2606:4700:4700::1111'), true); + const url = 'https://receiver.example/hook'; + await validateWebhookUrl(url, async () => [{ address: '93.184.216.34', family: 4 }]); + for (const address of ['127.0.0.1', '10.0.0.1', '169.254.169.254']) { + const send = createWebhookTransport({ resolve: async () => [{ address, family: 4 }] }); + await assert.rejects(send(url, '{}', {}), /UNSAFE_ADDRESS/); + } + await assert.rejects(validateWebhookUrl(url, async () => [ + { address: '93.184.216.34', family: 4 }, { address: '10.0.0.1', family: 4 }, + ]), /UNSAFE_ADDRESS/); +}); + +test('memory lifecycle, audit and payment claims commit with the outbox and redact private fields', async () => { + class FaultingOutbox extends MemoryWebhookStorage { + fail = false; + override prepareEvent(owner: string, payload: WebhookPayload): () => void { + if (this.fail) throw new Error('forced outbox failure'); + return super.prepareEvent(owner, payload); + } + } + const outbox = new FaultingOutbox(); + await outbox.register(endpoint()); + const store = new MemoryStorage(outbox); + outbox.fail = true; + assert.throws(() => invoice(store), /forced outbox failure/); + assert.equal(store.countInvoices(), 0); + assert.equal((await outbox.listDeliveries(seller.publicKey())).length, 0); + outbox.fail = false; + const paid = invoice(store), cancelled = invoice(store), expired = invoice(store); + const hash = 'a'.repeat(64), close = new Date(); + outbox.fail = true; + assert.throws(() => store.markAsPaid(paid.id, hash, seller.publicKey(), undefined, { settledAt: close }), /forced outbox failure/); + assert.throws(() => store.cancelInvoice(cancelled.id, seller.publicKey()), /forced outbox failure/); + assert.throws(() => store.markExpiredInvoices(new Date(Date.now() + 172_800_000)), /forced outbox failure/); + assert.throws(() => store.logPaymentEvent(paid.id, 'PAYMENT_REJECTED', { code: 'AMOUNT_TOO_LOW' }), /forced outbox failure/); + assert.equal(store.getPaymentClaim(hash), undefined); + assert.deepEqual(store.getAllInvoices().map(row => row.status), ['PENDING', 'PENDING', 'PENDING']); + assert.equal(store.getPaymentEvents().length, 0); + assert.equal((await outbox.listDeliveries(seller.publicKey())).length, 3); + outbox.fail = false; + store.markAsPaid(paid.id, hash, seller.publicKey(), { payerEmail: 'payer@private.example' }, { settledAt: close }); + assert.equal(store.markAsPaid(paid.id, hash, seller.publicKey(), undefined, { settledAt: close }), undefined); + store.cancelInvoice(cancelled.id, seller.publicKey()); + assert.equal(store.cancelInvoice(cancelled.id, seller.publicKey()), undefined); + assert.equal(store.markExpiredInvoices(new Date(Date.now() + 172_800_000)), 1); + assert.equal(store.markExpiredInvoices(new Date(Date.now() + 172_800_000)), 0); + store.logPaymentEvent(paid.id, 'PAYMENT_REJECTED', { code: 'AMOUNT_TOO_LOW', txHash: 'b'.repeat(64), + memo: 'PRIVATE_MEMO', email: 'private@example.test', error: 'PRIVATE_ERROR', nested: [{ payerEmail: 'private@example.test' }] }); + store.logPaymentEvent(paid.id, 'PARTIAL_PAYMENT', { code: 'AMOUNT_MISMATCH' }); + const deliveries = await outbox.listDeliveries(seller.publicKey()); + assert.deepEqual(deliveries.map(row => row.eventType).sort(), [ + 'invoice.created', 'invoice.created', 'invoice.created', 'invoice.paid', 'invoice.cancelled', 'invoice.expired', 'payment.rejected', 'payment.rejected', + ].sort()); + assert.equal(store.getPaymentEvents().length, 5); + const text = JSON.stringify(deliveries.map(row => row.payload)); + for (const value of ['PRIVATE_', 'private.example', 'private@example', 'payerPublicKey', 'metadata', 'memo', 'customerName']) assert.equal(text.includes(value), false, value); + assert.deepEqual(deliveries.find(row => row.payload.payment?.txHash)?.payload.payment, { code: 'AMOUNT_TOO_LOW', txHash: 'b'.repeat(64) }); +}); + +test('actual HTTP receiver 500/500/200 produces three attempts, growing delays and stable event body/id', async t => { + const received: { body: string; signature: string; eventId: string }[] = []; + const server = createServer(async (req, res) => { + const parts: Buffer[] = []; + for await (const part of req) parts.push(Buffer.from(part)); + received.push({ body: Buffer.concat(parts).toString(), signature: String(req.headers['x-quittance-signature']), eventId: String(req.headers['x-quittance-event-id']) }); + res.writeHead(received.length < 3 ? 500 : 200).end(); + }); + server.listen(0, '127.0.0.1'); await once(server, 'listening'); + t.after(() => new Promise(resolve => server.close(() => resolve()))); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const outbox = new MemoryWebhookStorage(), target = endpoint(); + await outbox.register(target); + let now = new Date(); + const eventId = randomUUID(); + await outbox.enqueueTest(seller.publicKey(), target.id, eventId, now); + // Explicit test transport. Production transport has no loopback exemption. + const worker = new WebhookWorker(outbox, { encryptionKey: KEY, clock: () => now, random: () => 0.5, + transport: async (_url, body, headers) => { + const response = await fetch('http://127.0.0.1:' + address.port, { method: 'POST', body, headers }); + await response.arrayBuffer(); return { statusCode: response.status }; + }, + }); + const delays: number[] = []; + for (let attempt = 1; attempt <= 3; attempt++) { + const started = now.getTime(); + assert.equal(await worker.runOnce(), true); + const [row] = await outbox.listDeliveries(seller.publicKey()); + assert.equal(row.attempt, attempt); + assert.equal(verifyWebhookSignature('whsec_public_test', received.at(-1)!.signature, received.at(-1)!.body, now.getTime()), true); + if (attempt < 3) { + assert.equal(row.status, 'pending'); assert.equal(await worker.runOnce(), false); + delays.push(Date.parse(row.nextAttemptAt) - started); now = new Date(row.nextAttemptAt); + } else { assert.equal(row.status, 'delivered'); assert.equal(row.lastResponseCode, 200); } + } + assert.deepEqual(delays, [1125, 2250]); + assert.equal(new Set(received.map(row => row.body)).size, 1); + assert.deepEqual(received.map(row => row.eventId), [eventId, eventId, eventId]); + assert.equal(await worker.runOnce(), false); +}); + +test('concurrent memory workers exclude an in-flight row and disable after repeated failures', async () => { + const outbox = new MemoryWebhookStorage(), target = endpoint(); + await outbox.register(target); + let now = new Date(); + await outbox.enqueueTest(seller.publicKey(), target.id, randomUUID(), now); + let entered!: () => void, finish!: () => void; + const enteredPromise = new Promise(resolve => { entered = resolve; }); + const finishedPromise = new Promise(resolve => { finish = resolve; }); + const worker = new WebhookWorker(outbox, { encryptionKey: KEY, clock: () => now, random: () => 0, + disableAfterFailures: 2, transport: async () => { entered(); await finishedPromise; return { statusCode: 500 }; } }); + const inFlight = worker.runOnce(); await enteredPromise; + assert.equal(await worker.runOnce(), false); + finish(); await inFlight; + now = new Date((await outbox.listDeliveries(seller.publicKey()))[0].nextAttemptAt); + assert.equal(await worker.runOnce(), true); + const [delivery] = await outbox.listDeliveries(seller.publicKey()); + assert.equal(delivery.status, 'dead'); assert.equal(delivery.attempt, 2); + assert.equal((await outbox.listEndpoints(seller.publicKey()))[0].enabled, false); +}); + +test('mounted seller API binds one-use proofs to operations and returns secrets only on creation/rotation', async t => { + const outbox = new MemoryWebhookStorage(), now = new Date(), app = express(); + app.use(express.json()); + app.use('/api', createWebhookRouter(outbox, { enabled: true, encryptionKey: KEY, clock: () => now, + resolve: async () => [{ address: '93.184.216.34', family: 4 }] })); + const server = app.listen(0, '127.0.0.1'); await once(server, 'listening'); + t.after(() => new Promise(resolve => server.close(() => resolve()))); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const post = async (path: string, body: any) => { + const response = await fetch('http://127.0.0.1:' + address.port + '/api' + path, { method: 'POST', + headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) }); + return { status: response.status, body: await response.json() as any, cache: response.headers.get('cache-control') }; + }; + const proof = (action: WebhookAction, fields: any = {}, key = seller, age = 0) => { + const input = { action, sellerPublicKey: key.publicKey(), timestamp: Math.floor(now.getTime() / 1000) - age, nonce: randomUUID(), ...fields }; + const signature = key.sign(Buffer.from(webhookProofMessage(input))).toString('base64'); + const { action: _action, endpointId: _id, ...body } = input; + return { ...body, signature }; + }; + assert.equal((await post('/webhooks/list', { sellerPublicKey: seller.publicKey() })).status, 401); + const registration = proof('register', { url: 'https://receiver.example/hook', events: ['invoice.paid'] }); + const registered = await post('/webhooks', registration); + assert.equal(registered.status, 201); assert.equal(registered.cache, 'no-store'); + const { endpoint: created, secret } = registered.body.data; + assert.ok(secret.startsWith('whsec_')); + assert.equal((await post('/webhooks', registration)).status, 409); + const tampered = proof('register', { url: 'https://receiver.example/hook', events: ['invoice.paid'] }); + tampered.url = 'https://different.example/hook'; + assert.equal((await post('/webhooks', tampered)).status, 401); + assert.equal((await post('/webhooks/list', proof('list', {}, seller, 300))).status, 401); + const list = await post('/webhooks/list', proof('list')); + assert.equal(list.status, 200); + for (const field of [secret, 'secretHash', 'secretEncrypted']) assert.equal(JSON.stringify(list.body).includes(field), false); + const other = Keypair.random(); + assert.equal((await post('/webhooks/' + created.id + '/remove', proof('remove', { endpointId: created.id }, other))).status, 404); + assert.equal((await post('/webhooks/list', proof('list', {}, other))).body.data.endpoints.length, 0); + for (const url of ['https://127.0.0.1', 'https://10.0.0.1', 'https://169.254.169.254']) { + assert.equal((await post('/webhooks', proof('register', { url, events: ['invoice.paid'] }))).status, 400); + } + assert.equal((await post('/webhooks/' + created.id + '/test', proof('test', { endpointId: created.id }))).status, 202); + const rotated = await post('/webhooks/' + created.id + '/rotate', proof('rotate', { endpointId: created.id })); + assert.equal(rotated.status, 200); assert.notEqual(rotated.body.data.secret, secret); + assert.equal((await post('/webhooks/' + created.id + '/remove', proof('remove', { endpointId: created.id }))).status, 200); + assert.equal((await outbox.listDeliveries(seller.publicKey()))[0].status, 'cancelled'); +}); + +test('a failed rejection outbox returns 503 without caching; retry records the event before the verdict', async t => { + const { InvoiceMemoryService } = await import('../src/services/invoice-memory.service.ts'); + const { MemoryInvoiceStorage } = await import('../src/storage/memory-invoice-storage.ts'); + const { createInvoiceRouter } = await import('../src/routes/invoice.routes.ts'); + class FaultingOutbox extends MemoryWebhookStorage { + fail = false; + override prepareEvent(owner: string, payload: WebhookPayload): () => void { + if (this.fail) throw new Error('forced rejection outbox failure'); + return super.prepareEvent(owner, payload); + } + } + const outbox = new FaultingOutbox(); + await outbox.register(endpoint({ events: ['payment.rejected'] })); + const raw = new MemoryStorage(outbox), created = invoice(raw), hash = 'c'.repeat(64); + const storage = new MemoryInvoiceStorage(new InvoiceMemoryService(raw)); + let cached = 0; + const app = express(); app.use(express.json()); + app.use('/api', createInvoiceRouter({ storage, enableRateLimiting: false, enableVerifyCache: false, + enableCeilingCheck: false, enableConcurrencyLock: false, + verifyCache: { set: async () => { cached++; } } as any, + stellar: { getTransaction: async () => ({ + transaction: { hash, successful: true, memo_type: 'text', memo: created.memo, created_at: new Date().toISOString() }, + operations: [{ type: 'payment', from: seller.publicKey(), to: seller.publicKey(), + asset_type: 'native', amount: '0.1000000', transaction_successful: true }], + }) }, + })); + const server = app.listen(0, '127.0.0.1'); await once(server, 'listening'); + t.after(() => new Promise(resolve => server.close(() => resolve()))); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const verify = async () => { + const response = await fetch('http://127.0.0.1:' + address.port + '/api/invoices/' + created.id + '/verify', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ txHash: hash }), + }); + return { status: response.status, body: await response.json() as any }; + }; + outbox.fail = true; + const failed = await verify(); + assert.equal(failed.status, 503); assert.equal(failed.body.code, 'PAYMENT_EVENT_UNAVAILABLE'); + assert.equal(cached, 0); assert.equal(raw.getPaymentEvents().length, 0); + outbox.fail = false; + const retried = await verify(); + assert.equal(retried.status, 400); assert.equal(retried.body.code, 'AMOUNT_TOO_LOW'); + assert.equal(cached, 1); assert.equal(raw.getPaymentEvents().length, 1); + const [delivery] = await outbox.listDeliveries(seller.publicKey()); + assert.equal(delivery.eventId, raw.getPaymentEvents()[0].id); + assert.equal(delivery.payload.payment?.code, 'AMOUNT_TOO_LOW'); +}); diff --git a/backend/tests/webhook-postgres.integration.test.ts b/backend/tests/webhook-postgres.integration.test.ts new file mode 100644 index 00000000..afefb61e --- /dev/null +++ b/backend/tests/webhook-postgres.integration.test.ts @@ -0,0 +1,316 @@ +import assert from 'node:assert/strict'; +import { after, before, beforeEach, describe, it, type TestContext } from 'node:test'; +import { randomUUID } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import path from 'node:path'; +import { fork, type ChildProcess } from 'node:child_process'; +import { setTimeout as delay } from 'node:timers/promises'; +import { Pool } from 'pg'; +import { InvoiceService } from '../src/services/invoice.service'; +import { PostgresWebhookStorage } from '../src/storage/postgres-webhook-storage'; +import { sealWebhookSecret } from '../src/services/webhook-crypto'; +import { WEBHOOK_EVENT_TYPES, type WebhookEventType } from '../../shared/webhooks'; +import type { WebhookEndpoint } from '../src/storage/webhook-storage'; + +const DATABASE_URL = process.env.DATABASE_URL; +const SELLER = 'GAYF33NNNMI2Z6VNRFXQ64D4E4SF77PM46NW3ZUZEEU5X7FCHAZCMHKU'; +const PAYER = 'GCT7D6S5VTFGEURS6ZYIO33YZRPQMA3LNWB4GEOHDFDXZGWTA4EPIM5E'; +const TX_HASH = 'a'.repeat(64); +const WORKER = path.join(__dirname, 'fixtures/webhook-postgres-worker.ts'); + +interface WorkerMessage { + type: string; + pid: number; + id?: string; + eventId?: string; + attempt?: number; + processed?: boolean; + message?: string; +} + +/** IPC barriers establish the overlap; no timing guess decides whether workers raced. */ +function spawnWorker(t: TestContext, schema: string) { + const child: ChildProcess = fork(WORKER, [], { + execArgv: ['--import', 'tsx'], + env: { ...process.env, Q584_PG_SCHEMA: schema }, + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + }); + const queued: WorkerMessage[] = []; + const waiting: Array<{ type: string; resolve: (message: WorkerMessage) => void; reject: (error: Error) => void }> = []; + let stderr = ''; + let exit: { code: number | null; signal: NodeJS.Signals | null } | undefined; + let failure: Error | undefined; + let resolveExit: (value: { code: number | null; signal: NodeJS.Signals | null }) => void; + const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => { resolveExit = resolve; }); + child.stderr!.on('data', (chunk) => { stderr = (stderr + chunk.toString()).slice(-4000); }); + child.on('message', (value) => { + const message = value as WorkerMessage; + if (message.type === 'error') { + failure = new Error(message.message || 'Webhook worker failed'); + for (const waiter of waiting.splice(0)) waiter.reject(failure); + return; + } + const index = waiting.findIndex((item) => item.type === message.type); + if (index === -1) queued.push(message); + else waiting.splice(index, 1)[0].resolve(message); + }); + child.on('error', (error) => { + failure = error; + for (const waiter of waiting.splice(0)) waiter.reject(error); + resolveExit({ code: null, signal: null }); + }); + child.on('exit', (code, signal) => { + exit = { code, signal }; + for (const waiter of waiting.splice(0)) waiter.reject(new Error(`Worker exited before ${waiter.type}: ${code ?? signal}; ${stderr}`)); + resolveExit(exit); + }); + t.after(async () => { + if (!exit && child.pid) child.kill('SIGKILL'); + await exited; + }); + return { + child, exited, + send(message: Record) { child.send(message); }, + wait(type: string): Promise { + const index = queued.findIndex((item) => item.type === type); + if (index !== -1) return Promise.resolve(queued.splice(index, 1)[0]); + if (failure) return Promise.reject(failure); + if (exit) return Promise.reject(new Error(`Worker already exited before ${type}: ${stderr}`)); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const index = waiting.indexOf(waiter); + if (index !== -1) waiting.splice(index, 1); + reject(new Error(`Timed out waiting for worker ${type}: ${stderr}`)); + }, 10_000); + const waiter = { + type, + resolve(message: WorkerMessage) { clearTimeout(timer); resolve(message); }, + reject(error: Error) { clearTimeout(timer); reject(error); }, + }; + waiting.push(waiter); + }); + }, + }; +} + +describe('PostgreSQL webhook outbox acceptance', { + skip: DATABASE_URL ? false : 'DATABASE_URL is not set', + concurrency: false, +}, () => { + const schema = `q584_webhooks_${randomUUID().replaceAll('-', '')}`; + let admin: Pool; + let database: Pool; + let invoices: InvoiceService; + let webhooks: PostgresWebhookStorage; + + before(async () => { + admin = new Pool({ connectionString: DATABASE_URL }); + await admin.query(`CREATE SCHEMA "${schema}"`); + database = new Pool({ connectionString: DATABASE_URL, options: `-c search_path=${schema},public` }); + const sql = readFileSync(path.join(__dirname, '../../db/schema.sql'), 'utf8'); + await database.query(sql); + await database.query(sql); + invoices = new InvoiceService(database); + webhooks = new PostgresWebhookStorage(database); + }); + + beforeEach(async () => { + await database.query('DROP TRIGGER IF EXISTS test_reject_outbox ON webhook_deliveries'); + await database.query('TRUNCATE webhook_deliveries, webhook_endpoints, webhook_proofs, payment_events, transactions, invoices CASCADE'); + }); + + after(async () => { + await database?.end(); + if (admin) { + await admin.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + await admin.end(); + } + }); + + async function endpoint(events: readonly WebhookEventType[] = WEBHOOK_EVENT_TYPES) { + const value: WebhookEndpoint = { + id: randomUUID(), sellerPublicKey: SELLER, url: 'https://merchant.example/webhook', + events: [...events], enabled: true, failureCount: 0, createdAt: new Date().toISOString(), + ...sealWebhookSecret('whsec_integration_test_only', Buffer.alloc(32, 7)), + }; + await webhooks.register(value); + return value; + } + const createInvoice = (idempotencyKey?: string) => invoices.createInvoice({ + sellerPublicKey: SELLER, amount: 10, assetCode: 'XLM', idempotencyKey, + }); + async function deliveries(invoiceId: string) { + const result = await database.query( + `SELECT * FROM webhook_deliveries WHERE payload->'invoice'->>'id'=$1 ORDER BY created_at,id`, [invoiceId]); + return result.rows; + } + async function failOutboxInserts() { + await database.query(`CREATE OR REPLACE FUNCTION test_reject_outbox() RETURNS TRIGGER LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'Q584_TEST_OUTBOX_FAILURE'; RETURN NEW; END; $$; + CREATE TRIGGER test_reject_outbox BEFORE INSERT ON webhook_deliveries + FOR EACH ROW EXECUTE FUNCTION test_reject_outbox();`); + } + + it('enqueues each creation and paid/cancelled/expired transition exactly once', async () => { + await endpoint(); + const paid = await createInvoice('same-create'); + const cancelled = await createInvoice(); + const expired = await createInvoice(); + assert.equal((await createInvoice('same-create')).id, paid.id); + await invoices.markAsPaid(paid.id, TX_HASH, PAYER, undefined, { settledAt: new Date() }); + await invoices.cancelInvoice(cancelled.id, SELLER); + await database.query(`UPDATE invoices SET expires_at=now()-interval '1 second' WHERE id=$1`, [expired.id]); + assert.equal(await invoices.markExpiredInvoices(), 1); + assert.equal(await invoices.markExpiredInvoices(), 0); + await assert.rejects(invoices.markAsPaid(paid.id, TX_HASH, PAYER, undefined, { settledAt: new Date() })); + await assert.rejects(invoices.cancelInvoice(cancelled.id, SELLER)); + await database.query('UPDATE invoices SET status=status'); + + for (const [invoice, kind, audit] of [ + [paid, 'invoice.paid', 'PAYMENT_CONFIRMED'], + [cancelled, 'invoice.cancelled', 'INVOICE_CANCELLED'], + [expired, 'invoice.expired', 'INVOICE_EXPIRED'], + ] as const) { + const rows = await deliveries(invoice.id); + assert.deepEqual(rows.map((row) => row.event_type).sort(), ['invoice.created', kind].sort()); + assert.equal(new Set(rows.map((row) => row.event_id)).size, 2); + for (const row of rows) { + assert.equal(row.payload.id, row.event_id); + assert.equal(row.payload.invoice.id, invoice.id); + assert.equal(row.payload.invoice.amount, '10.0000000'); + } + const events = await invoices.getPaymentEvents(invoice.id); + assert.equal(events.filter((row) => row.eventType === audit).length, 1); + } + }); + + it('rolls back invoice creation when its outbox insert fails', async () => { + await endpoint(); + await failOutboxInserts(); + await assert.rejects(createInvoice('failed-create'), /Q584_TEST_OUTBOX_FAILURE/); + for (const table of ['invoices', 'payment_events', 'webhook_deliveries']) { + assert.equal(Number((await database.query(`SELECT count(*) FROM ${table}`)).rows[0].count), 0); + } + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + const invoice = await createInvoice('failed-create'); + assert.equal((await deliveries(invoice.id)).length, 1); + }); + + for (const transition of ['paid', 'cancelled', 'expired'] as const) { + it(`rolls back ${transition} invoice and audit state when its outbox insert fails`, async () => { + await endpoint(); + const invoice = await createInvoice(); + if (transition === 'expired') { + await database.query(`UPDATE invoices SET expires_at=now()-interval '1 second' WHERE id=$1`, [invoice.id]); + } + const mutate = () => transition === 'paid' + ? invoices.markAsPaid(invoice.id, TX_HASH, PAYER, undefined, { settledAt: new Date() }) + : transition === 'cancelled' + ? invoices.cancelInvoice(invoice.id, SELLER) + : invoices.markExpiredInvoices(); + await failOutboxInserts(); + await assert.rejects(mutate(), /Q584_TEST_OUTBOX_FAILURE/); + // Direct read intentionally avoids InvoiceService's expiry-on-read sweep. + const row = (await database.query('SELECT * FROM invoices WHERE id=$1', [invoice.id])).rows[0]; + assert.equal(row.status, 'PENDING'); + assert.equal(row.payment_tx_hash, null); + assert.equal(row.paid_at, null); + assert.equal(row.cancelled_at, null); + assert.equal(row.settled_at, null); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 0); + assert.deepEqual((await deliveries(invoice.id)).map((row) => row.event_type), ['invoice.created']); + + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + await mutate(); + assert.equal((await deliveries(invoice.id)).filter((row) => row.event_type === `invoice.${transition}`).length, 1); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 1); + }); + } + + it('commits a payment rejection audit and outbox together, with the same event id', async () => { + await endpoint(); + const invoice = await createInvoice(); + const rejection = { code: 'ASSET_MISMATCH', txHash: TX_HASH, customerEmail: 'private@example.invalid', memo: 'private-memo' }; + await failOutboxInserts(); + await assert.rejects(invoices.logPaymentEvent(invoice.id, 'PAYMENT_REJECTED', rejection), /Q584_TEST_OUTBOX_FAILURE/); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 0); + assert.equal((await deliveries(invoice.id)).filter((row) => row.event_type === 'payment.rejected').length, 0); + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + await invoices.logPaymentEvent(invoice.id, 'PAYMENT_REJECTED', rejection); + const audit = (await invoices.getPaymentEvents(invoice.id))[0]; + const outbox = (await deliveries(invoice.id)).filter((row) => row.event_type === 'payment.rejected'); + assert.equal(outbox.length, 1); + assert.equal(outbox[0].event_id, audit.id); + assert.deepEqual(outbox[0].payload.payment, { code: rejection.code, txHash: TX_HASH }); + assert.doesNotMatch(JSON.stringify(outbox[0].payload), /private@example|private-memo/); + assert.equal((await invoices.getInvoiceById(invoice.id))?.status, 'PENDING'); + }); + + it('lets only one of two separate worker processes own a due delivery', { timeout: 30_000 }, async (t) => { + await endpoint(['invoice.created']); + const invoice = await createInvoice(); + const [row] = await deliveries(invoice.id); + const first = spawnWorker(t, schema); + await first.wait('ready'); + first.send({ type: 'run', mode: 'hold', now: new Date().toISOString() }); + const claim = await first.wait('claimed'); + assert.equal(claim.id, row.id); + + const second = spawnWorker(t, schema); + await second.wait('ready'); + assert.notEqual(first.child.pid, second.child.pid); + second.send({ type: 'run', mode: 'finish', now: new Date().toISOString() }); + const refused = await second.wait('done'); + assert.equal(refused.processed, false, 'second worker must finish without claiming the held row'); + assert.equal((await second.exited).code, 0); + assert.equal((await deliveries(invoice.id))[0].attempt, 0, 'first worker has not committed its delivery'); + + first.send({ type: 'release' }); + assert.equal((await first.wait('done')).processed, true); + assert.equal((await first.exited).code, 0); + const [delivered] = await deliveries(invoice.id); + assert.equal(delivered.id, row.id); + assert.equal(delivered.event_id, row.event_id); + assert.equal(delivered.status, 'delivered'); + assert.equal(delivered.attempt, 1); + }); + + it('retains the pending event after a worker crash and delivers it on restart', { timeout: 30_000 }, async (t) => { + await endpoint(['invoice.created']); + const invoice = await createInvoice(); + const [row] = await deliveries(invoice.id); + const first = spawnWorker(t, schema); + await first.wait('ready'); + first.send({ type: 'run', mode: 'hold', now: new Date().toISOString() }); + assert.equal((await first.wait('claimed')).id, row.id); + first.child.kill('SIGKILL'); + assert.equal((await first.exited).signal, 'SIGKILL'); + // Wait for PostgreSQL to observe that exact terminated client's disconnect. + // This is cleanup synchronization, not a claimed latency benchmark. + const deadline = Date.now() + 5000; + for (;;) { + const active = await admin.query('SELECT count(*) FROM pg_stat_activity WHERE application_name=$1', [`q584-worker-${first.child.pid}`]); + if (Number(active.rows[0].count) === 0) break; + assert.ok(Date.now() < deadline, 'database must release the crashed worker connection'); + await delay(20); + } + const [pending] = await deliveries(invoice.id); + assert.equal(pending.id, row.id); + assert.equal(pending.event_id, row.event_id); + assert.equal(pending.status, 'pending'); + assert.equal(pending.attempt, 0); + + const restarted = spawnWorker(t, schema); + await restarted.wait('ready'); + restarted.send({ type: 'run', mode: 'finish', now: new Date().toISOString() }); + const claim = await restarted.wait('claimed'); + assert.equal(claim.eventId, row.event_id); + assert.equal(claim.attempt, 0); + assert.equal((await restarted.wait('done')).processed, true); + assert.equal((await restarted.exited).code, 0); + const [delivered] = await deliveries(invoice.id); + assert.equal(delivered.status, 'delivered'); + assert.equal(delivered.attempt, 1); + assert.equal((await deliveries(invoice.id)).length, 1); + }); +}); diff --git a/db/schema.sql b/db/schema.sql index bf44e527..8e43a519 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -152,7 +152,7 @@ BEGIN 'Resolve conflicts before re-running the migration.'; END IF; - CREATE UNIQUE INDEX invoices_payment_tx_hash_unique + CREATE UNIQUE INDEX IF NOT EXISTS invoices_payment_tx_hash_unique ON invoices(payment_tx_hash) WHERE payment_tx_hash IS NOT NULL; END IF; @@ -192,3 +192,182 @@ ALTER TABLE invoices ADD COLUMN IF NOT EXISTS idempotency_key VARCHAR(255); CREATE UNIQUE INDEX IF NOT EXISTS idx_invoices_seller_idempotency ON invoices (seller_public_key, idempotency_key) WHERE idempotency_key IS NOT NULL; + + +-- Issue #584: seller webhooks and a transactional, durable delivery outbox. +CREATE TABLE IF NOT EXISTS webhook_endpoints ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + seller_public_key VARCHAR(56) NOT NULL, + url TEXT NOT NULL CHECK (char_length(url) <= 2048), + events TEXT[] NOT NULL CHECK ( + cardinality(events) BETWEEN 1 AND 5 AND + events <@ ARRAY[ + 'invoice.created', 'invoice.paid', 'invoice.cancelled', + 'invoice.expired', 'payment.rejected' + ]::TEXT[] + ), + secret_hash VARCHAR(64) NOT NULL, + secret_encrypted TEXT NOT NULL, + previous_secret_hash VARCHAR(64), + previous_secret_encrypted TEXT, + previous_secret_expires_at TIMESTAMPTZ, + enabled BOOLEAN NOT NULL DEFAULT true, + failure_count INTEGER NOT NULL DEFAULT 0 CHECK (failure_count >= 0), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + disabled_at TIMESTAMPTZ, + deleted_at TIMESTAMPTZ +); +CREATE INDEX IF NOT EXISTS idx_webhook_endpoints_seller + ON webhook_endpoints (seller_public_key, created_at, id); + +CREATE TABLE IF NOT EXISTS webhook_deliveries ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + endpoint_id UUID NOT NULL REFERENCES webhook_endpoints(id), + event_id UUID NOT NULL, + event_type TEXT NOT NULL CHECK (event_type IN ( + 'invoice.created', 'invoice.paid', 'invoice.cancelled', + 'invoice.expired', 'payment.rejected' + )), + payload JSONB NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0 CHECK (attempt >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'delivered', 'dead', 'cancelled')), + last_response_code INTEGER, + last_error_code TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + UNIQUE (endpoint_id, event_id) +); +CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_due + ON webhook_deliveries (next_attempt_at, id) WHERE status = 'pending'; +CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_history + ON webhook_deliveries (endpoint_id, created_at DESC, id DESC); + +CREATE TABLE IF NOT EXISTS webhook_proofs ( + seller_public_key VARCHAR(56) NOT NULL, + nonce UUID NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (seller_public_key, nonce) +); +CREATE INDEX IF NOT EXISTS idx_webhook_proofs_expiry ON webhook_proofs (expires_at); + +-- Keep the SQL boundary a whitelist too: no memo, contact, description, raw +-- rejection text, or arbitrary event-data fields can enter the outbox. +CREATE OR REPLACE FUNCTION invoice_webhook_payload( + i invoices, + kind TEXT, + event_id UUID, + at TIMESTAMPTZ, + payment JSONB DEFAULT NULL +) RETURNS JSONB LANGUAGE SQL STABLE AS $$ + SELECT jsonb_strip_nulls(jsonb_build_object( + 'version', 1, + 'id', event_id, + 'type', kind, + 'createdAt', at, + 'invoice', jsonb_strip_nulls(jsonb_build_object( + 'id', (i).id, + 'amount', (i).amount::text, + 'assetCode', (i).asset_code, + 'assetIssuer', (i).asset_issuer, + 'status', (i).status, + 'paymentTxHash', CASE + WHEN (i).payment_tx_hash ~ '^[a-fA-F0-9]{64}$' THEN (i).payment_tx_hash ELSE NULL END, + 'settledAt', (i).settled_at, + 'settlementContext', CASE + WHEN (i).settlement_context ~ '^[A-Z_]{1,60}$' THEN (i).settlement_context ELSE NULL END, + 'priorStatus', CASE + WHEN (i).prior_status ~ '^[A-Z_]{1,60}$' THEN (i).prior_status ELSE NULL END, + 'latePaymentWarningCode', CASE + WHEN (i).late_payment_warning_code ~ '^[A-Z_]{1,60}$' THEN (i).late_payment_warning_code ELSE NULL END, + 'expiresAt', (i).expires_at + )), + 'payment', CASE WHEN payment IS NULL THEN NULL ELSE jsonb_strip_nulls(jsonb_build_object( + 'code', CASE WHEN payment->>'code' ~ '^[A-Z0-9_]{1,100}$' THEN payment->>'code' ELSE NULL END, + 'txHash', CASE WHEN payment->>'txHash' ~ '^[a-fA-F0-9]{64}$' THEN payment->>'txHash' ELSE NULL END + )) END + )); +$$; + +-- AFTER triggers are part of the same INSERT/UPDATE statement and transaction. +-- A failed audit or outbox insert rolls back the invoice change, including the +-- markAsPaid CTE and lazy expiry sweeps. No HTTP request runs in this transaction. +CREATE OR REPLACE FUNCTION invoice_webhook_outbox() RETURNS TRIGGER LANGUAGE plpgsql AS $$ +DECLARE + kind TEXT; + v_event_id UUID := gen_random_uuid(); + happened_at TIMESTAMPTZ := clock_timestamp(); + event_payload JSONB; +BEGIN + IF TG_OP = 'INSERT' THEN + kind := 'invoice.created'; + ELSE + IF OLD.status IS NOT DISTINCT FROM NEW.status THEN + RETURN NEW; + END IF; + CASE NEW.status + WHEN 'PAID' THEN kind := 'invoice.paid'; + WHEN 'CANCELLED' THEN kind := 'invoice.cancelled'; + WHEN 'EXPIRED' THEN kind := 'invoice.expired'; + ELSE RETURN NEW; + END CASE; + IF NEW.status IN ('CANCELLED', 'EXPIRED') THEN + INSERT INTO payment_events (id, invoice_id, event_type, event_data, created_at) + VALUES ( + v_event_id, + NEW.id, + CASE WHEN NEW.status = 'CANCELLED' THEN 'INVOICE_CANCELLED' ELSE 'INVOICE_EXPIRED' END, + jsonb_build_object('priorStatus', OLD.status, 'status', NEW.status), + happened_at + ); + END IF; + END IF; + event_payload := invoice_webhook_payload(NEW, kind, v_event_id, happened_at); + INSERT INTO webhook_deliveries ( + endpoint_id, event_id, event_type, payload, next_attempt_at, created_at + ) + SELECT e.id, v_event_id, kind, event_payload, happened_at, happened_at + FROM webhook_endpoints e + WHERE e.seller_public_key = NEW.seller_public_key + AND e.enabled AND e.deleted_at IS NULL AND kind = ANY(e.events) + ON CONFLICT (endpoint_id, event_id) DO NOTHING; + RETURN NEW; +END; +$$; +DROP TRIGGER IF EXISTS invoice_webhook_outbox_trigger ON invoices; +CREATE TRIGGER invoice_webhook_outbox_trigger + AFTER INSERT OR UPDATE OF status ON invoices + FOR EACH ROW EXECUTE FUNCTION invoice_webhook_outbox(); + +CREATE OR REPLACE FUNCTION rejection_webhook_outbox() RETURNS TRIGGER LANGUAGE plpgsql AS $$ +DECLARE + invoice_row invoices; + event_payload JSONB; + happened_at TIMESTAMPTZ := COALESCE(NEW.created_at::timestamptz, clock_timestamp()); +BEGIN + IF NEW.event_type NOT IN ('PAYMENT_REJECTED', 'PARTIAL_PAYMENT') THEN + RETURN NEW; + END IF; + SELECT * INTO invoice_row FROM invoices WHERE id = NEW.invoice_id; + IF NOT FOUND THEN + RETURN NEW; + END IF; + event_payload := invoice_webhook_payload( + invoice_row, 'payment.rejected', NEW.id, happened_at, NEW.event_data + ); + INSERT INTO webhook_deliveries ( + endpoint_id, event_id, event_type, payload, next_attempt_at, created_at + ) + SELECT e.id, NEW.id, 'payment.rejected', event_payload, happened_at, happened_at + FROM webhook_endpoints e + WHERE e.seller_public_key = invoice_row.seller_public_key + AND e.enabled AND e.deleted_at IS NULL AND 'payment.rejected' = ANY(e.events) + ON CONFLICT (endpoint_id, event_id) DO NOTHING; + RETURN NEW; +END; +$$; +DROP TRIGGER IF EXISTS rejection_webhook_outbox_trigger ON payment_events; +CREATE TRIGGER rejection_webhook_outbox_trigger + AFTER INSERT ON payment_events + FOR EACH ROW EXECUTE FUNCTION rejection_webhook_outbox(); diff --git a/docs/WEBHOOKS.md b/docs/WEBHOOKS.md new file mode 100644 index 00000000..9ef24f39 --- /dev/null +++ b/docs/WEBHOOKS.md @@ -0,0 +1,206 @@ +# Seller webhooks + +Connect invoice events to an accounting, fulfillment or notification service from +**Dashboard โ†’ Webhooks**. Each action asks Freighter to sign the exact operation. +Opening the dashboard does not open a wallet prompt. + +Supported events: `invoice.created`, `invoice.paid`, `invoice.cancelled`, +`invoice.expired`, `payment.rejected`. Rejected and partial-payment audit records +both emit `payment.rejected`. A test delivery uses `invoice.created` with +`test: true`, bypasses event filters, and contains no fabricated invoice ID. + +## Enable + +Run `npm run db:migrate` in `backend` before enabling PostgreSQL delivery. Set: + +```dotenv +WEBHOOKS_ENABLED=true +WEBHOOK_ENCRYPTION_KEY=<64 hex characters from openssl rand -hex 32> +WEBHOOK_MAX_ATTEMPTS=8 +WEBHOOK_DISABLE_AFTER_FAILURES=10 +``` + +Keep the encryption key server-only, stable across restarts, and shared by replicas. +Store it in the deployment secret manager. Changing it without re-encrypting stored +endpoint secrets makes those keys unreadable. It is separate from each seller's +generated `whsec_...` signing secret. + +All three server entrypoints mount the same API and start/stop the worker with the +server. Imports alone do not start a worker. With delivery disabled, sellers can +still list/remove endpoints, but register/rotate/test return 503. + +**Use PostgreSQL for durability.** The memory MVP has equivalent in-process +behavior; its invoices, endpoints, replay protection and outbox are lost on restart. + +## Dashboard and signed API + +An account may register five endpoints. URLs must use HTTPS and resolve only to +public addresses, including in development. Credentials and fragments are refused. +The panel supports add, remove, test, rotate, and manual refresh of the latest 50 +deliveries. Tests are limited to one per endpoint per minute. Removed endpoints +retain history and cancel pending work; a request already in flight may complete. + +The signing secret appears once on registration/rotation. Copy it into the receiver +and dismiss it. List/history never return plaintext, hashes or ciphertext. The +browser keeps the returned secret only in component memory and discards it on +wallet/network changes, including changes while a signing request is in flight. + +Rotation keeps the previous key valid for 24 hours. Deliveries carry both signatures +during that window. Further rotation is refused until the overlap expires so it +cannot invalidate the promised old-key window. To resume a disabled endpoint, +repair the receiver, remove the endpoint and register it again. + +All paths below are POSTs under `/api`, and all require a wallet proof in the body: + +| Path | Action | Additional signed fields | Successful data | +| --- | --- | --- | --- | +| `/webhooks` | register | URL and event filters | endpoint, one-time secret | +| `/webhooks/list` | list | none | endpoints, recent deliveries | +| `/webhooks/:id/remove` | remove | endpoint ID | ID, removed=true | +| `/webhooks/:id/rotate` | rotate | endpoint ID | endpoint, secret, previousSecretExpiresAt | +| `/webhooks/:id/test` | test | endpoint ID | eventId | + +Common body fields are `sellerPublicKey`, integer Unix-seconds `timestamp`, +UUID `nonce`, and base64 Ed25519 `signature`. Sign the UTF-8 bytes produced by +`webhookProofMessage` in `shared/webhooks.ts`, using the same Freighter +`signBlob` and backend `verifySellerSignature` mechanism as cancellation. + +The canonical message binds a versioned domain, seller, action, endpoint ID, exact +URL, sorted filters, timestamp and nonce. Normalize the URL before signing and send +those exact bytes. Proofs must be less than five minutes old and no more than 30 +seconds in the future. Each nonce is consumed once in storage; PostgreSQL shares +replay protection and endpoint registration limits across API instances. Responses +use `Cache-Control: no-store`. + +## Receiver signature and deduplication + +Each POST includes: + +```text +Content-Type: application/json +X-Quittance-Event-Id: +X-Quittance-Event-Type: invoice.paid +X-Quittance-Signature: t=,v1=[,v1=] +``` + +Compute HMAC-SHA256 with the literal UTF-8 `whsec_...` secret over +`timestamp + "." + original_body_bytes`. Do not decode the secret or parse and +re-serialize JSON before verification. Each retry has a fresh signed timestamp +and the same event ID/body. + +The exported `verifyWebhookSignature` helper implements this contract. An +equivalent Node receiver: + +```js +const { createHmac, timingSafeEqual } = require('node:crypto'); + +function verify(secret, header, rawBody, now = Math.floor(Date.now() / 1000)) { + if (typeof header !== 'string' || header.length > 1024) return false; + const fields = header.split(',').map(value => value.trim()); + const times = fields.filter(value => value.startsWith('t=')); + if (times.length !== 1 || !/^t=\d{1,12}$/.test(times[0])) return false; + const timestamp = Number(times[0].slice(2)); + if (Math.abs(now - timestamp) > 300) return false; + const expected = createHmac('sha256', secret) + .update(timestamp + '.').update(rawBody).digest(); + return fields.some(value => /^v1=[a-f0-9]{64}$/.test(value) && + timingSafeEqual(expected, Buffer.from(value.slice(3), 'hex'))); +} +``` + +For an Express receiver, use `express.raw({ type: 'application/json' })` for +this route before the JSON parser. Verify first, then parse. Atomically record +`payload.id` with the receiving business change, then acknowledge with 2xx. +Return 2xx for an already-processed ID without repeating the business change. + +Do not rely on arrival order. Concurrent invoice transactions can become visible +in a different order; use ledger settlement time and the current invoice state +when reconciling. + +### Signature vector + +```text +secret: whsec_test_0123456789abcdef +timestamp: 1735689600 +body: {"id":"evt_1","type":"invoice.paid"} +HMAC: 7489caff2d92322967b02be31d80ec2a16eb05ba167bbd9719f630fc84eedd07 +``` + +There is no newline after the body. This small body is a cryptographic vector, +not a full event. Set the verifier's test clock to that timestamp; a live receiver +correctly rejects the old timestamp. + +## Payload privacy + +Events contain `version: 1`, `id`, `type`, `createdAt` and an invoice projection: +ID, decimal-string amount, asset code/issuer, status and expiry. Paid events may add +transaction hash, ledger settlement time and late-payment classification. +Rejections may add only a stable code and transaction hash in `payment`. + +SQL constructs this explicit field whitelist. Memory writes and every delivery +attempt pass through the stricter webhook projection in +`payment-event-redaction.ts`. Names/emails, payer identity, memo, description, +metadata, raw errors and unknown nested fields are excluded. Receiver bodies are +discarded and never retained/logged. Endpoint secrets have a SHA-256 hash plus +AES-256-GCM ciphertext; a hash alone cannot produce HMAC signatures. + +## Atomic outbox and retries + +PostgreSQL AFTER triggers enqueue in the same transaction as creation, the existing +paid UPDATE/audit CTE, cancellation, expiry, and rejection audit insertion. +Cancellation/expiry now write payment audit events too. An outbox insert failure +rolls back invoice and audit changes. Repeated writes of the same status produce +no additional lifecycle event. Subscribed endpoints each receive one row under +the unique `(endpoint_id, event_id)` constraint. + +Semantic/claim rejection responses are cached only after their audit/outbox write +commits. Failed event storage returns a fixed 503 and leaves verification retryable. + +Workers use `FOR UPDATE OF d,e SKIP LOCKED` and retain row/endpoint locks until +the bounded HTTP attempt and result commit finish. Other processes skip those +rows. Endpoint locking serializes its failure counter. A process crash rolls back +the claim and makes the same event available again. Memory storage performs +fallible event preparation before changing invoices, audits or transaction claims. + +Delivery is **at least once**. If a receiver accepts an event and the worker dies +before recording success, another worker sends the same ID again. Database locks +cannot stop business work already started by a remote receiver after a crash; +receiver-side deduplication is required. + +Each attempt repeats DNS checks and pins an approved address while retaining the +original TLS hostname verification. Any private, loopback or link-local answer +is rejected. Redirects are never followed. DNS has a two-second deadline and the +TLS/HTTP request a five-second deadline. An unsafe URL/address immediately +dead-letters the event and disables its endpoint. + +Other non-2xx/transport failures use the existing monitor backoff: 1, 2, 4, 8, 16, +30, 30 seconds, plus 0โ€“25% jitter. The default maximum is eight attempts. Ten +consecutive failed attempts across an endpoint's deliveries disable it; success +resets the counter. Remaining pending events become dead letters. Configuration +bounds are 1โ€“20 attempts and 1โ€“100 consecutive failures. + +The worker polls every second and processes at most 25 rows per pass, without +overlapping its own passes. History exposes attempt, next retry, status, last +response code and safe failure code. Existing structured logging records +`webhook.worker`/`webhook.manage` failures without URLs, secrets or bodies. +Rows are retained for diagnosis; this change does not silently delete old history. + +## Acceptance + +```sh +cd backend +npm run typecheck +node --import tsx --test tests/webhook-delivery.test.ts +DATABASE_URL=postgresql://... node --import tsx --test tests/webhook-postgres.integration.test.ts +``` + +Delivery acceptance includes the vector, rotation, registration/send SSRF checks, +memory rollback, actual HTTP 500/500/200 delivery, in-process exclusion, signed +management, and the rejection storage/cache boundary. The test receiver explicitly +injects loopback transport; there is no production private-address bypass. Its +controlled clock records 1125 ms and 2250 ms retry schedules without waiting. + +PostgreSQL acceptance uses an isolated schema and separate OS processes for atomic +rollback, exclusive claims, and SIGKILL/restart recovery. It explicitly skips when +DATABASE_URL is absent. Frontend `tests/seller-webhooks.test.js` covers signing, +the full panel, wallet/network switches and stale secret responses. diff --git a/frontend/app/dashboard/page.tsx b/frontend/app/dashboard/page.tsx index 16567bb2..842f45fb 100644 --- a/frontend/app/dashboard/page.tsx +++ b/frontend/app/dashboard/page.tsx @@ -5,6 +5,7 @@ import { invoiceApi } from '@/lib/api'; import InvoiceCard from '@/components/InvoiceCard'; import WalletConnect from '@/components/WalletConnect'; import UserProfile from '@/components/UserProfile'; +import SellerWebhooksPanel from '@/components/SellerWebhooksPanel'; import FreighterInstallPrompt from '@/components/FreighterInstallPrompt'; import AssetLogo from '@/components/AssetLogo'; import { useWalletStore } from '@/lib/store'; @@ -510,6 +511,7 @@ export default function DashboardPage() { )} + )} diff --git a/frontend/components/SellerWebhooksPanel.tsx b/frontend/components/SellerWebhooksPanel.tsx new file mode 100644 index 00000000..9c7db293 --- /dev/null +++ b/frontend/components/SellerWebhooksPanel.tsx @@ -0,0 +1,315 @@ +'use client'; + +import { useEffect, useRef, useState, type FormEvent } from 'react'; +import { WEBHOOK_EVENT_TYPES, type WebhookEventType } from '@shared/webhooks'; +import { useWalletStore } from '@/lib/store'; +import { EXPECTED_WALLET_NETWORK } from '@/lib/stellar'; +import { walletSessionGate } from '@/lib/wallet-session'; +import { + webhookApi, + webhookWalletContext, + type WebhookEndpoint, + type WebhookList, + type WebhookRequestContext, + type WebhookSecretResult, +} from '@/lib/webhook-api'; + +const EVENT_LABELS: Record = { + 'invoice.created': 'Invoice created', + 'invoice.paid': 'Invoice paid', + 'invoice.cancelled': 'Invoice cancelled', + 'invoice.expired': 'Invoice expired', + 'payment.rejected': 'Payment rejected', +}; + +const STATUS_LABELS = { + pending: 'Pending', + delivered: 'Delivered', + dead: 'Failed permanently', + cancelled: 'Cancelled', +}; + +function dateLabel(value?: string | null): string { + if (!value) return 'โ€”'; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? 'โ€”' : date.toLocaleString(); +} + +/** The keyed subtree clears secrets and responses on any wallet/network change. */ +export default function SellerWebhooksPanel() { + const wallet = useWalletStore(); + const gate = walletSessionGate(wallet, EXPECTED_WALLET_NETWORK); + if (!gate.ready || !wallet.publicKey) return null; + const context = webhookWalletContext(wallet); + return ; +} + +function SellerWebhooksContent({ sellerPublicKey, walletContext }: Omit) { + const [data, setData] = useState(null); + const [hasLoaded, setHasLoaded] = useState(false); + const [url, setUrl] = useState(''); + const [events, setEvents] = useState([...WEBHOOK_EVENT_TYPES]); + const [busy, setBusy] = useState(null); + const [error, setError] = useState(null); + const [notice, setNotice] = useState(null); + const [secret, setSecret] = useState(null); + const [copied, setCopied] = useState(false); + const [removeId, setRemoveId] = useState(null); + const active = useRef(true); + const inFlight = useRef(false); + const abort = useRef(null); + + useEffect(() => { + active.current = true; + return () => { + active.current = false; + abort.current?.abort(); + }; + }, []); + + async function run(label: string, action: (context: WebhookRequestContext) => Promise<() => void>) { + if (inFlight.current) return; + inFlight.current = true; + const controller = new AbortController(); + abort.current = controller; + setBusy(label); + setError(null); + setNotice(null); + try { + const commit = await action({ sellerPublicKey, walletContext, signal: controller.signal }); + if (active.current && !controller.signal.aborted && + webhookWalletContext(useWalletStore.getState()) === walletContext) commit(); + } catch (caught) { + if (active.current && !controller.signal.aborted) { + setError(caught instanceof Error ? caught.message : 'The webhook action could not be completed.'); + } + } finally { + inFlight.current = false; + if (active.current) setBusy(null); + } + } + + function upsertEndpoint(endpoint: WebhookEndpoint) { + setData((previous) => ({ + endpoints: [endpoint, ...(previous?.endpoints ?? []).filter((item) => item.id !== endpoint.id)], + deliveries: previous?.deliveries ?? [], + })); + } + + function revealSecret(result: WebhookSecretResult) { + upsertEndpoint(result.endpoint); + setSecret(result); + setCopied(false); + } + + function load() { + void run('Loading endpoints', async (context) => { + const result = await webhookApi.list(context); + return () => { + setData(result); + setHasLoaded(true); + setRemoveId(null); + setNotice('Endpoints and recent deliveries are up to date.'); + }; + }); + } + + function register(event: FormEvent) { + event.preventDefault(); + if (secret) return; + void run('Adding endpoint', async (context) => { + const result = await webhookApi.register(context, url, events); + return () => { + revealSecret(result); + setUrl(''); + setNotice('Endpoint added. Save its signing secret below.'); + }; + }); + } + + function rotate(endpointId: string) { + if (secret) return; + void run('Rotating signing secret', async (context) => { + const result = await webhookApi.rotate(context, endpointId); + return () => { + revealSecret(result); + setNotice('Signing secret rotated. Update the secret in your receiver.'); + }; + }); + } + + function remove(endpointId: string) { + void run('Removing endpoint', async (context) => { + await webhookApi.remove(context, endpointId); + return () => { + setData((previous) => previous ? { + endpoints: previous.endpoints.filter((endpoint) => endpoint.id !== endpointId), + // A refresh obtains the final delivery status from the server. + deliveries: previous.deliveries.filter((delivery) => delivery.endpointId !== endpointId), + } : null); + if (secret?.endpoint.id === endpointId) setSecret(null); + setRemoveId(null); + setNotice('Endpoint removed. Refresh to load the latest delivery history.'); + }; + }); + } + + function sendTest(endpointId: string) { + void run('Queuing test delivery', async (context) => { + const result = await webhookApi.test(context, endpointId); + return () => setNotice(`Test event queued: ${result.eventId}. Refresh shortly to check its delivery.`); + }); + } + + async function copySecret() { + if (!secret) return; + try { + await navigator.clipboard.writeText(secret.secret); + if (active.current && webhookWalletContext(useWalletStore.getState()) === walletContext) setCopied(true); + } catch { + if (active.current) setError('Could not copy automatically. Select and copy the signing secret below.'); + } + } + + return ( +
+
+
+

Webhooks

+

Receive signed invoice updates in your own system.

+

Approve a Freighter wallet signature for each action.

+
+ +
+ +

+ {busy ? `${busy}. Check Freighter if a signature is requested.` : notice} +

+ {error &&

{error}

} + + {secret && ( +
+

Copy and save this signing secret now

+

It is shown once. Store it securely on the server that receives your webhooks.

+

{secret.endpoint.url}

+ +