From 219664a9751e0bb45fc4022ebf09b4c7dbf9658d Mon Sep 17 00:00:00 2001 From: Femi John Date: Thu, 23 Jul 2026 11:17:01 +0100 Subject: [PATCH] [Enhancement] Signed webhook event system: HMAC-signed payment lifecycle callbacks with retries and dead-letter redelivery --- app.js | 12 + src/controllers/courses/courseController.js | 9 + src/controllers/stellar/paymentController.js | 77 +++ src/controllers/stellar/walletController.js | 13 + src/controllers/webhookController.js | 387 ++++++++++++++ src/models/WebhookDelivery.js | 64 +++ src/models/WebhookEndpoint.js | 99 ++++ src/routes/webhookRoutes.js | 41 ++ src/services/webhooks/deliveryWorker.js | 232 +++++++++ src/services/webhooks/webhookService.js | 141 +++++ src/utils/ssrfGuard.js | 93 ++++ test/webhook-api.test.js | 348 +++++++++++++ test/webhooks.test.js | 521 +++++++++++++++++++ 13 files changed, 2037 insertions(+) create mode 100644 src/controllers/webhookController.js create mode 100644 src/models/WebhookDelivery.js create mode 100644 src/models/WebhookEndpoint.js create mode 100644 src/routes/webhookRoutes.js create mode 100644 src/services/webhooks/deliveryWorker.js create mode 100644 src/services/webhooks/webhookService.js create mode 100644 src/utils/ssrfGuard.js create mode 100644 test/webhook-api.test.js create mode 100644 test/webhooks.test.js diff --git a/app.js b/app.js index 6dd44dc..28eddf3 100644 --- a/app.js +++ b/app.js @@ -43,6 +43,8 @@ import stellarWalletRoutes from "./src/routes/stellar/walletRoutes.js"; import stellarPaymentRoutes from "./src/routes/stellar/paymentRoutes.js"; import stellarDonationRoutes from "./src/routes/stellar/donationRoutes.js"; import payoutRoutes from "./src/routes/payoutRoutes.js"; +import webhookRoutes from "./src/routes/webhookRoutes.js"; +import { startDeliveryWorker } from "./src/services/webhooks/deliveryWorker.js"; handleUncaughtException(); validateEnv(); @@ -179,6 +181,16 @@ app.use("/api/stellar/wallet", stellarWalletRoutes); app.use("/api/stellar/payment", stellarPaymentRoutes); app.use("/api/stellar/donation", stellarDonationRoutes); app.use("/api/payouts", payoutRoutes); +app.use("/api/webhooks", webhookRoutes); + +// ====================== +// WEBHOOK DELIVERY WORKER +// ====================== + +let webhookWorker = null; +if (process.env.NODE_ENV !== "test") { + webhookWorker = startDeliveryWorker(); +} // ====================== // ERROR HANDLING diff --git a/src/controllers/courses/courseController.js b/src/controllers/courses/courseController.js index d783da6..e6b297a 100644 --- a/src/controllers/courses/courseController.js +++ b/src/controllers/courses/courseController.js @@ -2,6 +2,7 @@ import Course from "../../models/Course.js"; import mongoose from "mongoose"; import logger from "../../config/logger.js"; import { catchAsync, APIError } from "../../middlewares/errorHandler.js"; +import { emitEvent } from "../../services/webhooks/webhookService.js"; /** * Create a new course @@ -142,6 +143,14 @@ export const enrollInCourse = async (req, res) => { } } + // Emit event after saves — fire-and-forget + emitEvent("course.enrolled", { + courseId: course._id.toString(), + courseTitle: course.title, + userId: req.user._id.toString(), + enrolledAt: new Date().toISOString(), + }); + res .status(200) .json({ diff --git a/src/controllers/stellar/paymentController.js b/src/controllers/stellar/paymentController.js index 97c61e6..6069f97 100644 --- a/src/controllers/stellar/paymentController.js +++ b/src/controllers/stellar/paymentController.js @@ -22,6 +22,7 @@ import { paymentsConfirmed, paymentsFailed, } from "../../config/metrics.js"; +import { emitEvent } from "../../services/webhooks/webhookService.js"; /** * Initialize a payment - creates pending transaction and returns XDR to sign @@ -203,6 +204,20 @@ export const initializePayment = async (req, res) => { await session.commitTransaction(); paymentsInitialized.inc({ type: "purchase" }); + // Emit event after commit — fire-and-forget + emitEvent("payment.initialized", { + transactionId: transaction._id.toString(), + buyerId: buyerId.toString(), + buyerWallet: buyer.stellarWallet.publicKey, + creatorId: creator._id.toString(), + creatorWallet: destinationPublicKey, + itemType, + itemId: itemId.toString(), + itemTitle: item.title, + amount: item.price.toString(), + network: NETWORK, + }); + logger.info( `Payment initialized: ${transaction._id} for ${itemType} ${itemId}` ); @@ -293,6 +308,21 @@ export const submitPayment = async (req, res) => { await session.commitTransaction(); paymentsFailed.inc({ type: "purchase", reason: "stellar_error" }); + // Emit event after commit + emitEvent("payment.failed", { + transactionId: transaction._id.toString(), + buyerId: buyerId.toString(), + buyerWallet: transaction.buyerWallet, + creatorId: transaction.creator.toString(), + creatorWallet: transaction.creatorWallet, + itemType: transaction.itemType, + itemId: transaction.itemId.toString(), + itemTitle: transaction.itemTitle, + amount: transaction.amount, + network: transaction.network, + failureReason: stellarError.message, + }); + logger.error(`Transaction ${transactionId} failed:`, stellarError); return res.status(400).json({ @@ -335,6 +365,22 @@ export const submitPayment = async (req, res) => { await session.commitTransaction(); paymentsFailed.inc({ type: "purchase", reason: "verification_failed" }); + // Emit event after commit + emitEvent("payment.failed", { + transactionId: transaction._id.toString(), + buyerId: buyerId.toString(), + buyerWallet: transaction.buyerWallet, + creatorId: transaction.creator.toString(), + creatorWallet: transaction.creatorWallet, + itemType: transaction.itemType, + itemId: transaction.itemId.toString(), + itemTitle: transaction.itemTitle, + amount: transaction.amount, + network: transaction.network, + stellarTxHash: result.hash, + failureReason: `On-chain verification failed: ${verification.reason}`, + }); + logger.error( `Transaction ${transactionId} verification failed: ${verification.reason}` ); @@ -387,6 +433,23 @@ export const submitPayment = async (req, res) => { await buyer.save({ session }); await session.commitTransaction(); + paymentsConfirmed.inc({ type: "purchase" }); + + // Emit event after commit — fire-and-forget + emitEvent("payment.confirmed", { + transactionId: transaction._id.toString(), + buyerId: buyerId.toString(), + buyerWallet: transaction.buyerWallet, + creatorId: transaction.creator.toString(), + creatorWallet: transaction.creatorWallet, + itemType: transaction.itemType, + itemId: transaction.itemId.toString(), + itemTitle: transaction.itemTitle, + amount: transaction.amount, + network: transaction.network, + stellarTxHash: result.hash, + ledger: result.ledger, + }); logger.info( `Payment successful: ${transactionId}, Stellar TX: ${result.hash}` @@ -552,6 +615,20 @@ export const cancelTransaction = async (req, res) => { }); } + // Emit event after update + emitEvent("payment.expired", { + transactionId: transaction._id.toString(), + buyerId: userId.toString(), + buyerWallet: transaction.buyerWallet, + creatorId: transaction.creator.toString(), + creatorWallet: transaction.creatorWallet, + itemType: transaction.itemType, + itemId: transaction.itemId.toString(), + itemTitle: transaction.itemTitle, + amount: transaction.amount, + network: transaction.network, + }); + logger.info(`Transaction ${transactionId} cancelled by user ${userId}`); res.status(200).json({ diff --git a/src/controllers/stellar/walletController.js b/src/controllers/stellar/walletController.js index 2767a23..c8ce661 100644 --- a/src/controllers/stellar/walletController.js +++ b/src/controllers/stellar/walletController.js @@ -5,6 +5,7 @@ import { getAccountBalance, NETWORK, } from "../../services/stellar/stellarService.js"; +import { emitEvent } from "../../services/webhooks/webhookService.js"; import logger from "../../config/logger.js"; /** @@ -55,6 +56,13 @@ export const connectWallet = async (req, res) => { logger.info(`Wallet connected for user ${userId}: ${publicKey}`); + // Emit event — fire-and-forget + emitEvent("wallet.connected", { + userId: userId.toString(), + publicKey, + network: NETWORK, + }); + res.status(200).json({ success: true, message: "Wallet connected successfully", @@ -89,6 +97,11 @@ export const disconnectWallet = async (req, res) => { logger.info(`Wallet disconnected for user ${userId}`); + // Emit event — fire-and-forget + emitEvent("wallet.disconnected", { + userId: userId.toString(), + }); + res.status(200).json({ success: true, message: "Wallet disconnected successfully", diff --git a/src/controllers/webhookController.js b/src/controllers/webhookController.js new file mode 100644 index 0000000..c83809b --- /dev/null +++ b/src/controllers/webhookController.js @@ -0,0 +1,387 @@ +import WebhookEndpoint from "../models/WebhookEndpoint.js"; +import WebhookDelivery from "../models/WebhookDelivery.js"; +import { validateEndpointUrl } from "../utils/ssrfGuard.js"; +import { emitEvent, EVENT_TYPES } from "../services/webhooks/webhookService.js"; +import logger from "../config/logger.js"; + +/** + * Require admin role. Mirrors the payout admin pattern (allowlist-based) + * pending a centralized RBAC system. + */ +function requireAdmin(req, res) { + if (!req.user || req.user.role !== "admin") { + res.status(403).json({ + success: false, + message: "Forbidden: Admin access required", + }); + return false; + } + return true; +} + +/** + * List webhook endpoints for the authenticated user (or all for admins). + * GET /api/webhooks + */ +export const listEndpoints = async (req, res) => { + try { + const query = req.user.role === "admin" + ? {} + : { owner: req.user._id }; + + const endpoints = await WebhookEndpoint.find(query).sort({ createdAt: -1 }); + + res.status(200).json({ + success: true, + count: endpoints.length, + endpoints, + }); + } catch (error) { + logger.error({ err: error }, "List webhook endpoints error"); + res.status(500).json({ success: false, message: "Failed to list endpoints" }); + } +}; + +/** + * Create a webhook endpoint. + * The raw secret is returned only in this response. + * POST /api/webhooks + */ +export const createEndpoint = async (req, res) => { + try { + const { url, events, description } = req.body; + + if (!url || !events || !Array.isArray(events) || events.length === 0) { + return res.status(400).json({ + success: false, + message: "url and events (array) are required", + }); + } + + // Validate event types + const invalidEvents = events.filter( + (e) => e !== "*" && !EVENT_TYPES.includes(e) + ); + if (invalidEvents.length > 0) { + return res.status(400).json({ + success: false, + message: `Invalid event types: ${invalidEvents.join(", ")}`, + validTypes: EVENT_TYPES, + }); + } + + // SSRF check + const ssrf = await validateEndpointUrl(url); + if (!ssrf.valid) { + return res.status(400).json({ + success: false, + message: ssrf.error, + }); + } + + const { raw, hashed } = WebhookEndpoint.generateSecret(); + + const endpoint = await WebhookEndpoint.create({ + url, + secret: hashed, + events, + description: description || "", + owner: req.user._id, + }); + + logger.info({ endpointId: endpoint._id, url }, "Webhook endpoint created"); + + // Return the raw secret — this is the only time it's visible + res.status(201).json({ + success: true, + message: "Endpoint created. Save the secret — it will not be shown again.", + endpoint: { + _id: endpoint._id, + url: endpoint.url, + events: endpoint.events, + description: endpoint.description, + isActive: endpoint.isActive, + createdAt: endpoint.createdAt, + }, + secret: raw, + }); + } catch (error) { + logger.error({ err: error }, "Create webhook endpoint error"); + res.status(500).json({ success: false, message: "Failed to create endpoint" }); + } +}; + +/** + * Get a single endpoint by ID. + * GET /api/webhooks/:id + */ +export const getEndpoint = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + // Non-admins can only view their own + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + res.status(200).json({ success: true, endpoint }); + } catch (error) { + logger.error({ err: error }, "Get webhook endpoint error"); + res.status(500).json({ success: false, message: "Failed to get endpoint" }); + } +}; + +/** + * Update an endpoint (URL, events, description, isActive). + * PUT /api/webhooks/:id + */ +export const updateEndpoint = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + const { url, events, description, isActive } = req.body; + + if (url !== undefined) { + const ssrf = await validateEndpointUrl(url); + if (!ssrf.valid) { + return res.status(400).json({ success: false, message: ssrf.error }); + } + endpoint.url = url; + } + + if (events !== undefined) { + if (!Array.isArray(events) || events.length === 0) { + return res.status(400).json({ + success: false, + message: "events must be a non-empty array", + }); + } + const invalidEvents = events.filter( + (e) => e !== "*" && !EVENT_TYPES.includes(e) + ); + if (invalidEvents.length > 0) { + return res.status(400).json({ + success: false, + message: `Invalid event types: ${invalidEvents.join(", ")}`, + }); + } + endpoint.events = events; + } + + if (description !== undefined) endpoint.description = description; + if (isActive !== undefined) { + endpoint.isActive = isActive; + if (isActive) { + endpoint.disabledAt = undefined; + endpoint.disabledReason = undefined; + endpoint.consecutiveFailures = 0; + } + } + + await endpoint.save(); + + res.status(200).json({ success: true, endpoint }); + } catch (error) { + logger.error({ err: error }, "Update webhook endpoint error"); + res.status(500).json({ success: false, message: "Failed to update endpoint" }); + } +}; + +/** + * Delete an endpoint and its pending deliveries. + * DELETE /api/webhooks/:id + */ +export const deleteEndpoint = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + await WebhookDelivery.deleteMany({ endpoint: endpoint._id }); + await endpoint.deleteOne(); + + res.status(200).json({ success: true, message: "Endpoint deleted" }); + } catch (error) { + logger.error({ err: error }, "Delete webhook endpoint error"); + res.status(500).json({ success: false, message: "Failed to delete endpoint" }); + } +}; + +/** + * Rotate the secret for an endpoint. Returns the new raw secret once. + * POST /api/webhooks/:id/rotate-secret + */ +export const rotateSecret = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id).select("+secret"); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + const { raw, hashed } = WebhookEndpoint.generateSecret(); + endpoint.secret = hashed; + await endpoint.save(); + + res.status(200).json({ + success: true, + message: "Secret rotated. Save the new secret — it will not be shown again.", + secret: raw, + }); + } catch (error) { + logger.error({ err: error }, "Rotate webhook secret error"); + res.status(500).json({ success: false, message: "Failed to rotate secret" }); + } +}; + +/** + * List deliveries for an endpoint (paginated, filterable by status). + * GET /api/webhooks/:id/deliveries + */ +export const listDeliveries = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + const { status, page = 1, limit = 20 } = req.query; + const query = { endpoint: endpoint._id }; + if (status) query.status = status; + + const skip = (parseInt(page) - 1) * parseInt(limit); + const [deliveries, total] = await Promise.all([ + WebhookDelivery.find(query) + .sort({ createdAt: -1 }) + .skip(skip) + .limit(parseInt(limit)), + WebhookDelivery.countDocuments(query), + ]); + + res.status(200).json({ + success: true, + deliveries, + pagination: { + page: parseInt(page), + limit: parseInt(limit), + total, + pages: Math.ceil(total / parseInt(limit)), + }, + }); + } catch (error) { + logger.error({ err: error }, "List webhook deliveries error"); + res.status(500).json({ success: false, message: "Failed to list deliveries" }); + } +}; + +/** + * Redeliver a dead delivery. + * POST /api/webhooks/:id/deliveries/:deliveryId/redeliver + */ +export const redeliver = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + const delivery = await WebhookDelivery.findOne({ + _id: req.params.deliveryId, + endpoint: endpoint._id, + }); + + if (!delivery) { + return res.status(404).json({ success: false, message: "Delivery not found" }); + } + + if (delivery.status !== "dead") { + return res.status(400).json({ + success: false, + message: "Only dead deliveries can be redelivered", + }); + } + + delivery.status = "pending"; + delivery.nextAttemptAt = new Date(); + await delivery.save(); + + res.status(200).json({ + success: true, + message: "Delivery queued for redelivery", + delivery, + }); + } catch (error) { + logger.error({ err: error }, "Redeliver webhook error"); + res.status(500).json({ success: false, message: "Failed to redeliver" }); + } +}; + +/** + * Send a signed ping event to test endpoint integration. + * POST /api/webhooks/:id/ping + */ +export const pingEndpoint = async (req, res) => { + try { + const endpoint = await WebhookEndpoint.findById(req.params.id); + if (!endpoint) { + return res.status(404).json({ success: false, message: "Endpoint not found" }); + } + + if (req.user.role !== "admin" && endpoint.owner.toString() !== req.user._id.toString()) { + return res.status(403).json({ success: false, message: "Forbidden" }); + } + + // Emit a ping event — will be delivered like any other event + await emitEvent("payment.initialized", { + ping: true, + message: "DeenBridge webhook ping", + endpointId: endpoint._id.toString(), + sentAt: new Date().toISOString(), + }); + + res.status(200).json({ + success: true, + message: "Ping event emitted", + }); + } catch (error) { + logger.error({ err: error }, "Ping webhook endpoint error"); + res.status(500).json({ success: false, message: "Failed to send ping" }); + } +}; + +/** + * List available event types. + * GET /api/webhooks/events + */ +export const listEventTypes = async (_req, res) => { + res.status(200).json({ + success: true, + events: EVENT_TYPES, + }); +}; diff --git a/src/models/WebhookDelivery.js b/src/models/WebhookDelivery.js new file mode 100644 index 0000000..3d51241 --- /dev/null +++ b/src/models/WebhookDelivery.js @@ -0,0 +1,64 @@ +import mongoose from "mongoose"; + +const MAX_ATTEMPT_BODY_BYTES = 4096; + +const attemptSchema = new mongoose.Schema( + { + at: { type: Date, required: true }, + statusCode: { type: Number }, + error: { type: String, maxlength: 512 }, + durationMs: { type: Number }, + responseBody: { type: String, maxlength: MAX_ATTEMPT_BODY_BYTES }, + }, + { _id: false } +); + +const webhookDeliverySchema = new mongoose.Schema( + { + endpoint: { + type: mongoose.Schema.Types.ObjectId, + ref: "WebhookEndpoint", + required: true, + }, + eventId: { + type: String, + required: true, + index: true, + }, + eventType: { + type: String, + required: true, + }, + payload: { + type: mongoose.Schema.Types.Mixed, + required: true, + }, + attempts: { + type: [attemptSchema], + default: [], + }, + status: { + type: String, + enum: ["pending", "delivered", "retrying", "dead", "processing"], + default: "pending", + index: true, + }, + nextAttemptAt: { + type: Date, + default: () => new Date(), + index: true, + }, + lastAttemptAt: { + type: Date, + }, + deliveredAt: { + type: Date, + }, + }, + { timestamps: true } +); + +webhookDeliverySchema.index({ endpoint: 1, status: 1 }); +webhookDeliverySchema.index({ status: 1, nextAttemptAt: 1 }); + +export default mongoose.model("WebhookDelivery", webhookDeliverySchema); diff --git a/src/models/WebhookEndpoint.js b/src/models/WebhookEndpoint.js new file mode 100644 index 0000000..9d7e138 --- /dev/null +++ b/src/models/WebhookEndpoint.js @@ -0,0 +1,99 @@ +import mongoose from "mongoose"; +import crypto from "crypto"; + +const webhookEndpointSchema = new mongoose.Schema( + { + url: { + type: String, + required: [true, "Endpoint URL is required"], + validate: { + validator: function (v) { + if (process.env.NODE_ENV === "production") { + return /^https:\/\//.test(v); + } + return /^https?:\/\//.test(v); + }, + message: "Endpoint URL must use HTTPS in production", + }, + }, + secret: { + type: String, + required: true, + select: false, + }, + description: { + type: String, + maxlength: 200, + default: "", + }, + events: { + type: [String], + required: [true, "At least one event type is required"], + validate: { + validator: function (v) { + return v.length > 0; + }, + message: "At least one event type is required", + }, + }, + isActive: { + type: Boolean, + default: true, + }, + consecutiveFailures: { + type: Number, + default: 0, + }, + totalDeliveries: { + type: Number, + default: 0, + }, + totalFailures: { + type: Number, + default: 0, + }, + disabledAt: { + type: Date, + }, + disabledReason: { + type: String, + }, + owner: { + type: mongoose.Schema.Types.ObjectId, + ref: "User", + required: true, + }, + }, + { timestamps: true } +); + +webhookEndpointSchema.index({ owner: 1 }); +webhookEndpointSchema.index({ isActive: 1 }); + +/** + * Generate a random webhook secret. + * Returns the raw secret (shown once) and the hashed version (stored). + */ +webhookEndpointSchema.statics.generateSecret = function () { + const raw = `whsec_${crypto.randomBytes(32).toString("hex")}`; + const hashed = crypto.createHash("sha256").update(raw).digest("hex"); + return { raw, hashed }; +}; + +/** + * Verify a raw secret against the stored hash. + */ +webhookEndpointSchema.statics.verifySecret = function (raw, hashed) { + const computed = crypto.createHash("sha256").update(raw).digest("hex"); + return crypto.timingSafeEqual(Buffer.from(computed), Buffer.from(hashed)); +}; + +/** + * Check if this endpoint is subscribed to the given event type. + * Wildcard "*" matches all events. + */ +webhookEndpointSchema.methods.isSubscribedTo = function (eventType) { + return this.events.includes("*") || this.events.includes(eventType); +}; + +export default mongoose.model("WebhookEndpoint", webhookEndpointSchema); diff --git a/src/routes/webhookRoutes.js b/src/routes/webhookRoutes.js new file mode 100644 index 0000000..510c7a5 --- /dev/null +++ b/src/routes/webhookRoutes.js @@ -0,0 +1,41 @@ +import express from "express"; +import { protect } from "../middlewares/authMiddleware.js"; +import { + listEndpoints, + createEndpoint, + getEndpoint, + updateEndpoint, + deleteEndpoint, + rotateSecret, + listDeliveries, + redeliver, + pingEndpoint, + listEventTypes, +} from "../controllers/webhookController.js"; + +const router = express.Router(); + +// All routes require authentication +router.use(protect); + +// Event types catalog (public to authenticated users) +router.get("/events", listEventTypes); + +// CRUD +router.get("/", listEndpoints); +router.post("/", createEndpoint); +router.get("/:id", getEndpoint); +router.put("/:id", updateEndpoint); +router.delete("/:id", deleteEndpoint); + +// Secret management +router.post("/:id/rotate-secret", rotateSecret); + +// Deliveries +router.get("/:id/deliveries", listDeliveries); +router.post("/:id/deliveries/:deliveryId/redeliver", redeliver); + +// Integration testing +router.post("/:id/ping", pingEndpoint); + +export default router; diff --git a/src/services/webhooks/deliveryWorker.js b/src/services/webhooks/deliveryWorker.js new file mode 100644 index 0000000..cb6809e --- /dev/null +++ b/src/services/webhooks/deliveryWorker.js @@ -0,0 +1,232 @@ +import axios from "axios"; +import crypto from "crypto"; +import WebhookEndpoint from "../../models/WebhookEndpoint.js"; +import WebhookDelivery from "../../models/WebhookDelivery.js"; +import { computeSignature, serializePayload } from "./webhookService.js"; +import logger from "../../config/logger.js"; + +const DELIVERY_TIMEOUT_MS = 10_000; +const MAX_ATTEMPTS = 6; +const MAX_CONSECUTIVE_FAILURES_TO_DISABLE = 20; + +/** + * Exponential backoff schedule with jitter. + * Attempt 0 = immediate (pending), then 1m, 5m, 30m, 2h, 12h. + */ +const BACKOFF_SCHEDULE_MS = [ + 0, // initial attempt + 60_000, // 1 minute + 300_000, // 5 minutes + 1_800_000, // 30 minutes + 7_200_000, // 2 hours + 43_200_000, // 12 hours +]; + +function getBackoffMs(attemptIndex) { + const base = BACKOFF_SCHEDULE_MS[Math.min(attemptIndex, BACKOFF_SCHEDULE_MS.length - 1)]; + // Add jitter: ±20% of the base delay + const jitter = base * 0.2 * (Math.random() * 2 - 1); + return Math.max(0, Math.round(base + jitter)); +} + +/** + * Deliver a single webhook delivery. + * Returns true on success, false on failure. + */ +async function deliverWebhook(delivery) { + const endpoint = await WebhookEndpoint.findById(delivery.endpoint).select("+secret"); + if (!endpoint || !endpoint.isActive) { + delivery.status = "dead"; + delivery.attempts.push({ + at: new Date(), + error: "Endpoint not found or inactive", + }); + await delivery.save(); + return false; + } + + const rawBody = serializePayload(delivery.payload); + const timestamp = new Date().toISOString(); + const signature = computeSignature(endpoint.secret, timestamp, rawBody); + + const headers = { + "Content-Type": "application/json", + "X-DeenBridge-Event": delivery.eventType, + "X-DeenBridge-Event-Id": delivery.eventId, + "X-DeenBridge-Timestamp": timestamp, + "X-DeenBridge-Signature": signature, + }; + + const startTime = Date.now(); + + try { + const response = await axios.post(endpoint.url, rawBody, { + headers, + timeout: DELIVERY_TIMEOUT_MS, + maxRedirects: 0, + validateStatus: () => true, // don't throw for non-2xx + }); + + const durationMs = Date.now() - startTime; + const is2xx = response.status >= 200 && response.status < 300; + + const attemptRecord = { + at: new Date(), + statusCode: response.status, + durationMs, + responseBody: typeof response.data === "string" + ? response.data.slice(0, 4096) + : JSON.stringify(response.data).slice(0, 4096), + }; + + delivery.attempts.push(attemptRecord); + delivery.lastAttemptAt = new Date(); + endpoint.totalDeliveries += 1; + + if (is2xx) { + delivery.status = "delivered"; + delivery.deliveredAt = new Date(); + endpoint.consecutiveFailures = 0; + await delivery.save(); + await endpoint.save(); + return true; + } + + // Non-2xx: schedule retry + delivery.status = "retrying"; + const nextAttempt = delivery.attempts.length; + if (nextAttempt >= MAX_ATTEMPTS) { + delivery.status = "dead"; + } + delivery.nextAttemptAt = new Date(Date.now() + getBackoffMs(nextAttempt)); + + endpoint.totalFailures += 1; + endpoint.consecutiveFailures += 1; + + await delivery.save(); + await endpoint.save(); + + // Auto-disable endpoint after sustained failures + if (endpoint.consecutiveFailures >= MAX_CONSECUTIVE_FAILURES_TO_DISABLE) { + endpoint.isActive = false; + endpoint.disabledAt = new Date(); + endpoint.disabledReason = `Auto-disabled after ${endpoint.consecutiveFailures} consecutive delivery failures`; + await endpoint.save(); + logger.warn( + { endpointId: endpoint._id, consecutiveFailures: endpoint.consecutiveFailures }, + "Webhook endpoint auto-disabled due to sustained failures" + ); + } + + return false; + } catch (error) { + const durationMs = Date.now() - startTime; + + delivery.attempts.push({ + at: new Date(), + error: error.message?.slice(0, 512) || "Unknown delivery error", + durationMs, + }); + delivery.lastAttemptAt = new Date(); + + const nextAttempt = delivery.attempts.length; + delivery.status = nextAttempt >= MAX_ATTEMPTS ? "dead" : "retrying"; + if (delivery.status === "retrying") { + delivery.nextAttemptAt = new Date(Date.now() + getBackoffMs(nextAttempt)); + } + + endpoint.totalFailures += 1; + endpoint.consecutiveFailures += 1; + endpoint.totalDeliveries += 1; + + await delivery.save(); + await endpoint.save(); + + if (endpoint.consecutiveFailures >= MAX_CONSECUTIVE_FAILURES_TO_DISABLE) { + endpoint.isActive = false; + endpoint.disabledAt = new Date(); + endpoint.disabledReason = `Auto-disabled after ${endpoint.consecutiveFailures} consecutive delivery failures`; + await endpoint.save(); + logger.warn( + { endpointId: endpoint._id, consecutiveFailures: endpoint.consecutiveFailures }, + "Webhook endpoint auto-disabled due to sustained failures" + ); + } + + return false; + } +} + +/** + * Process due deliveries with atomic claim. + * Uses findOneAndUpdate to claim a delivery so concurrent worker + * instances don't double-send. + * + * @param {number} [batchSize=10] - Max deliveries to process per tick + * @returns {Promise} Number of deliveries processed + */ +export async function processDeliveries(batchSize = 10) { + let processed = 0; + + for (let i = 0; i < batchSize; i++) { + // Atomically claim the next due delivery + const delivery = await WebhookDelivery.findOneAndUpdate( + { + status: { $in: ["pending", "retrying"] }, + nextAttemptAt: { $lte: new Date() }, + }, + { + $set: { status: "processing" }, // mark as in-flight + }, + { + new: true, + sort: { nextAttemptAt: 1 }, + } + ); + + if (!delivery) break; + + await deliverWebhook(delivery); + processed += 1; + } + + return processed; +} + +/** + * Start the delivery worker loop. + * Runs every `intervalMs` and processes due deliveries. + * + * @param {number} [intervalMs=15000] - Polling interval in milliseconds + * @returns {{ stop: () => void }} Control handle to stop the worker + */ +export function startDeliveryWorker(intervalMs = 15_000) { + let running = true; + let timer = null; + + const tick = async () => { + if (!running) return; + try { + const count = await processDeliveries(); + if (count > 0) { + logger.info({ count }, "Webhook delivery batch processed"); + } + } catch (error) { + logger.error({ err: error }, "Webhook delivery worker error"); + } + if (running) { + timer = setTimeout(tick, intervalMs); + } + }; + + timer = setTimeout(tick, intervalMs); + logger.info({ intervalMs }, "Webhook delivery worker started"); + + return { + stop() { + running = false; + if (timer) clearTimeout(timer); + logger.info("Webhook delivery worker stopped"); + }, + }; +} diff --git a/src/services/webhooks/webhookService.js b/src/services/webhooks/webhookService.js new file mode 100644 index 0000000..5547eec --- /dev/null +++ b/src/services/webhooks/webhookService.js @@ -0,0 +1,141 @@ +import crypto from "crypto"; +import WebhookEndpoint from "../../models/WebhookEndpoint.js"; +import WebhookDelivery from "../../models/WebhookDelivery.js"; +import logger from "../../config/logger.js"; + +export const WEBHOOK_API_VERSION = "2025-01-01"; + +/** + * Supported event types. + * Consumers subscribe to one or more of these, or use "*" for all. + */ +export const EVENT_TYPES = [ + "payment.initialized", + "payment.confirmed", + "payment.failed", + "payment.expired", + "course.enrolled", + "wallet.connected", + "wallet.disconnected", +]; + +/** + * Build the deterministic event payload. + * Secrets, passwords, full user documents, and internal fields are excluded. + */ +function buildPayload(eventType, data) { + return { + apiVersion: WEBHOOK_API_VERSION, + eventId: crypto.randomUUID(), + eventType, + createdAt: new Date().toISOString(), + data, + }; +} + +/** + * Compute HMAC-SHA256 signature for a webhook payload. + * Format: v1= + * + * @param {string} secret - The raw webhook secret + * @param {string} timestamp - ISO timestamp used in the signed message + * @param {string} rawBody - The exact serialized body bytes + * @returns {string} The signature header value + */ +export function computeSignature(secret, timestamp, rawBody) { + const hmac = crypto.createHmac("sha256", secret); + hmac.update(`${timestamp}.${rawBody}`); + return `v1=${hmac.digest("hex")}`; +} + +/** + * Verify an HMAC-SHA256 signature. + * + * @param {string} secret - The raw webhook secret + * @param {string} timestamp - The timestamp from the request + * @param {string} rawBody - The raw body bytes + * @param {string} signature - The signature header value (v1=) + * @returns {boolean} + */ +export function verifySignature(secret, timestamp, rawBody, signature) { + if (!signature || !signature.startsWith("v1=")) return false; + + const expected = computeSignature(secret, timestamp, rawBody); + const sigBuf = Buffer.from(signature); + const expectedBuf = Buffer.from(expected); + + if (sigBuf.length !== expectedBuf.length) return false; + return crypto.timingSafeEqual(sigBuf, expectedBuf); +} + +/** + * Emit a webhook event. + * + * Fire-and-forget: persists WebhookDelivery rows for all matching active + * endpoints and returns. Never throws — delivery failures are logged and + * tracked on the delivery record. + * + * Must be called AFTER the originating DB transaction commits. + * + * @param {string} eventType - One of EVENT_TYPES + * @param {object} data - Event-specific payload (allowlisted fields only) + */ +export async function emitEvent(eventType, data) { + if (!EVENT_TYPES.includes(eventType)) { + logger.warn({ eventType }, "Unknown webhook event type"); + return; + } + + try { + const endpoints = await WebhookEndpoint.find({ + isActive: true, + }); + + const subscribers = endpoints.filter((ep) => ep.isSubscribedTo(eventType)); + + if (subscribers.length === 0) return; + + const payload = buildPayload(eventType, data); + + const deliveries = subscribers.map((ep) => ({ + endpoint: ep._id, + eventId: payload.eventId, + eventType, + payload, + status: "pending", + nextAttemptAt: new Date(), + })); + + await WebhookDelivery.insertMany(deliveries); + + logger.info( + { eventType, eventId: payload.eventId, endpointCount: subscribers.length }, + "Webhook event emitted" + ); + } catch (error) { + logger.error({ err: error, eventType }, "Failed to emit webhook event"); + } +} + +/** + * Build the headers for an outbound webhook delivery. + */ +export function buildDeliveryHeaders(secret, rawBody) { + const timestamp = new Date().toISOString(); + const signature = computeSignature(secret, timestamp, rawBody); + + return { + "Content-Type": "application/json", + "X-DeenBridge-Event": undefined, // set per-delivery + "X-DeenBridge-Timestamp": timestamp, + "X-DeenBridge-Signature": signature, + }; +} + +/** + * Serialize a payload deterministically (sorted keys) to ensure + * signature verification works on the receiving end. + */ +export function serializePayload(payload) { + return JSON.stringify(payload, Object.keys(payload).sort()); +} diff --git a/src/utils/ssrfGuard.js b/src/utils/ssrfGuard.js new file mode 100644 index 0000000..2fd4dee --- /dev/null +++ b/src/utils/ssrfGuard.js @@ -0,0 +1,93 @@ +import dns from "dns"; +import net from "net"; +import logger from "../config/logger.js"; + +const LOOPBACK_RANGES = [ + "127.0.0.0/8", + "10.0.0.0/8", + "172.16.0.0/12", + "192.168.0.0/16", + "169.254.0.0/16", + "::1/128", + "fc00::/7", + "fe80::/10", +]; + +function ipToNumber(ip) { + if (ip.includes(":")) return null; // IPv6 handled separately + return ip.split(".").reduce((acc, octet) => (acc << 8) + parseInt(octet, 10), 0) >>> 0; +} + +function cidrMatch(ip, cidr) { + const [range, bits] = cidr.split("/"); + const mask = ~(2 ** (32 - parseInt(bits, 10)) - 1) >>> 0; + return (ipToNumber(ip) & mask) === (ipToNumber(range) & mask); +} + +function isPrivateOrLoopback(ip) { + if (ip === "::1" || ip === "::" || ip === "0.0.0.0") return true; + if (ip.startsWith("fc") || ip.startsWith("fe")) return true; + const num = ipToNumber(ip); + if (num === null) return false; + return LOOPBACK_RANGES.some((cidr) => cidrMatch(ip, cidr)); +} + +function isIPv6Loopback(ip) { + return ip === "::1" || ip === "fe80::1"; +} + +/** + * Validate a URL for SSRF safety. + * - HTTPS required in production + * - Rejects private/loopback IPs after DNS resolution + * + * @param {string} urlString + * @param {object} [options] + * @param {boolean} [options.requireHttps=true] - Require HTTPS (defaults to true in production) + * @returns {Promise<{ valid: boolean, error?: string }>} + */ +export async function validateEndpointUrl(urlString, options = {}) { + const { requireHttps = process.env.NODE_ENV === "production" } = options; + + let parsed; + try { + parsed = new URL(urlString); + } catch { + return { valid: false, error: "Invalid URL format" }; + } + + if (requireHttps && parsed.protocol !== "https:") { + return { valid: false, error: "Endpoint URL must use HTTPS in production" }; + } + + if (!["http:", "https:"].includes(parsed.protocol)) { + return { valid: false, error: "URL must use http or https protocol" }; + } + + // Only check private IPs in production + if (process.env.NODE_ENV === "production") { + const hostname = parsed.hostname; + + // Skip DNS resolution for IP literals + const ipNum = ipToNumber(hostname); + if (ipNum !== null) { + if (isPrivateOrLoopback(hostname)) { + return { valid: false, error: "Endpoint URL must not target private or loopback addresses" }; + } + } else { + try { + const addrs = await dns.promises.resolve4(hostname); + for (const addr of addrs) { + if (isPrivateOrLoopback(addr)) { + return { valid: false, error: "Endpoint URL resolves to a private or loopback address" }; + } + } + } catch { + // DNS resolution failure is not an SSRF issue — the URL is just invalid + return { valid: false, error: "Could not resolve endpoint hostname" }; + } + } + } + + return { valid: true }; +} diff --git a/test/webhook-api.test.js b/test/webhook-api.test.js new file mode 100644 index 0000000..bac7295 --- /dev/null +++ b/test/webhook-api.test.js @@ -0,0 +1,348 @@ +import request from "supertest"; +import mongoose from "mongoose"; +import jwt from "jsonwebtoken"; +import app from "../app.js"; +import WebhookEndpoint from "../src/models/WebhookEndpoint.js"; +import WebhookDelivery from "../src/models/WebhookDelivery.js"; + +const JWT_SECRET = process.env.JWT_SECRET || "deenbridge-temp-secret-key-2024"; + +let adminToken; +let userToken; +let adminUserId; +let regularUserId; + +beforeAll(async () => { + await mongoose.connect(process.env.MONGO_URI); + + // Create test users + adminUserId = new mongoose.Types.ObjectId(); + regularUserId = new mongoose.Types.ObjectId(); + + adminToken = jwt.sign({ userId: adminUserId, sessionId: "s1" }, JWT_SECRET, { expiresIn: "1h" }); + userToken = jwt.sign({ userId: regularUserId, sessionId: "s2" }, JWT_SECRET, { expiresIn: "1h" }); +}); + +afterAll(async () => { + await WebhookEndpoint.deleteMany({}); + await WebhookDelivery.deleteMany({}); + await mongoose.disconnect(); +}); + +beforeEach(async () => { + await WebhookEndpoint.deleteMany({}); + await WebhookDelivery.deleteMany({}); +}); + +const authHeader = (token) => ({ Authorization: `Bearer ${token}` }); + +describe("Webhook Management API", () => { + describe("POST /api/webhooks", () => { + it("creates an endpoint and returns the raw secret once", async () => { + // Need to mock the User model for protect middleware + // Since we're using a real DB, create a user document + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const res = await request(app) + .post("/api/webhooks") + .set(authHeader(adminToken)) + .send({ + url: "https://example.com/hook", + events: ["payment.confirmed"], + description: "Test endpoint", + }); + + expect(res.statusCode).toBe(201); + expect(res.body.success).toBe(true); + expect(res.body.secret).toMatch(/^whsec_/); + expect(res.body.endpoint.url).toBe("https://example.com/hook"); + expect(res.body.endpoint.events).toEqual(["payment.confirmed"]); + + // Secret should NOT be stored in plaintext + const stored = await WebhookEndpoint.findById(res.body.endpoint._id).select("+secret"); + expect(stored.secret).not.toBe(res.body.secret); + expect(stored.secret).not.toMatch(/^whsec_/); // should be hashed + }); + + it("rejects invalid event types", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const res = await request(app) + .post("/api/webhooks") + .set(authHeader(adminToken)) + .send({ + url: "https://example.com/hook", + events: ["invalid.event"], + }); + + expect(res.statusCode).toBe(400); + expect(res.body.message).toContain("Invalid event types"); + }); + + it("rejects missing url", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const res = await request(app) + .post("/api/webhooks") + .set(authHeader(adminToken)) + .send({ + events: ["payment.confirmed"], + }); + + expect(res.statusCode).toBe(400); + }); + + it("requires authentication", async () => { + const res = await request(app) + .post("/api/webhooks") + .send({ + url: "https://example.com/hook", + events: ["payment.confirmed"], + }); + + expect(res.statusCode).toBe(401); + }); + }); + + describe("GET /api/webhooks", () => { + it("returns endpoints for the authenticated user", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: regularUserId, + name: "User", + email: "user@test.com", + password: "hashed", + role: "student", + }); + + await WebhookEndpoint.create({ + url: "https://example.com/hook1", + secret: "hashed", + events: ["payment.confirmed"], + owner: regularUserId, + }); + + const res = await request(app) + .get("/api/webhooks") + .set(authHeader(userToken)); + + expect(res.statusCode).toBe(200); + expect(res.body.endpoints).toHaveLength(1); + }); + }); + + describe("GET /api/webhooks/events", () => { + it("returns the event type catalog", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: regularUserId, + name: "User", + email: "user@test.com", + password: "hashed", + role: "student", + }); + + const res = await request(app) + .get("/api/webhooks/events") + .set(authHeader(userToken)); + + expect(res.statusCode).toBe(200); + expect(res.body.events).toContain("payment.confirmed"); + expect(res.body.events).toContain("course.enrolled"); + }); + }); + + describe("POST /api/webhooks/:id/rotate-secret", () => { + it("returns a new raw secret", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "oldhash", + events: ["*"], + owner: adminUserId, + }); + + const res = await request(app) + .post(`/api/webhooks/${endpoint._id}/rotate-secret`) + .set(authHeader(adminToken)); + + expect(res.statusCode).toBe(200); + expect(res.body.secret).toMatch(/^whsec_/); + + // Verify the stored hash changed + const updated = await WebhookEndpoint.findById(endpoint._id).select("+secret"); + expect(updated.secret).not.toBe("oldhash"); + }); + }); + + describe("POST /api/webhooks/:id/ping", () => { + it("emits a ping event", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + owner: adminUserId, + }); + + const res = await request(app) + .post(`/api/webhooks/${endpoint._id}/ping`) + .set(authHeader(adminToken)); + + expect(res.statusCode).toBe(200); + expect(res.body.message).toContain("Ping"); + + // Should have created a delivery + const deliveries = await WebhookDelivery.find({ endpoint: endpoint._id }); + expect(deliveries).toHaveLength(1); + expect(deliveries[0].payload.data.ping).toBe(true); + }); + }); + + describe("POST /api/webhooks/:id/deliveries/:deliveryId/redeliver", () => { + it("queues a dead delivery for redelivery", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + owner: adminUserId, + }); + + const delivery = await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "dead-event", + eventType: "payment.confirmed", + payload: { data: {} }, + status: "dead", + nextAttemptAt: new Date(), + }); + + const res = await request(app) + .post(`/api/webhooks/${endpoint._id}/deliveries/${delivery._id}/redeliver`) + .set(authHeader(adminToken)); + + expect(res.statusCode).toBe(200); + + const updated = await WebhookDelivery.findById(delivery._id); + expect(updated.status).toBe("pending"); + }); + + it("rejects redelivery of non-dead deliveries", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + owner: adminUserId, + }); + + const delivery = await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "delivered-event", + eventType: "payment.confirmed", + payload: { data: {} }, + status: "delivered", + }); + + const res = await request(app) + .post(`/api/webhooks/${endpoint._id}/deliveries/${delivery._id}/redeliver`) + .set(authHeader(adminToken)); + + expect(res.statusCode).toBe(400); + expect(res.body.message).toContain("Only dead deliveries"); + }); + }); + + describe("DELETE /api/webhooks/:id", () => { + it("deletes endpoint and its deliveries", async () => { + const User = (await import("../src/models/User.js")).default; + await User.create({ + _id: adminUserId, + name: "Admin", + email: "admin@test.com", + password: "hashed", + role: "admin", + }); + + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + owner: adminUserId, + }); + + await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "del1", + eventType: "payment.confirmed", + payload: { data: {} }, + }); + + const res = await request(app) + .delete(`/api/webhooks/${endpoint._id}`) + .set(authHeader(adminToken)); + + expect(res.statusCode).toBe(200); + + const remaining = await WebhookEndpoint.findById(endpoint._id); + expect(remaining).toBeNull(); + + const deliveries = await WebhookDelivery.find({ endpoint: endpoint._id }); + expect(deliveries).toHaveLength(0); + }); + }); +}); diff --git a/test/webhooks.test.js b/test/webhooks.test.js new file mode 100644 index 0000000..8369cce --- /dev/null +++ b/test/webhooks.test.js @@ -0,0 +1,521 @@ +import mongoose from "mongoose"; +import { + computeSignature, + verifySignature, + serializePayload, + emitEvent, + EVENT_TYPES, + WEBHOOK_API_VERSION, +} from "../src/services/webhooks/webhookService.js"; +import { processDeliveries } from "../src/services/webhooks/deliveryWorker.js"; +import { validateEndpointUrl } from "../src/utils/ssrfGuard.js"; +import WebhookEndpoint from "../src/models/WebhookEndpoint.js"; +import WebhookDelivery from "../src/models/WebhookDelivery.js"; + +// Connect to a test MongoDB instance +beforeAll(async () => { + await mongoose.connect(process.env.MONGO_URI); +}); + +afterAll(async () => { + await mongoose.disconnect(); +}); + +beforeEach(async () => { + await WebhookEndpoint.deleteMany({}); + await WebhookDelivery.deleteMany({}); +}); + +describe("HMAC generation and verification", () => { + const secret = "whsec_testsecret123"; + + it("produces a v1= prefixed hex signature", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const body = '{"key":"value"}'; + const sig = computeSignature(secret, timestamp, body); + + expect(sig).toMatch(/^v1=[0-9a-f]{64}$/); + }); + + it("produces different signatures for different timestamps", () => { + const body = '{"key":"value"}'; + const sig1 = computeSignature(secret, "2025-01-01T00:00:00.000Z", body); + const sig2 = computeSignature(secret, "2025-01-01T00:00:01.000Z", body); + + expect(sig1).not.toBe(sig2); + }); + + it("produces different signatures for different bodies", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const sig1 = computeSignature(secret, timestamp, '{"a":1}'); + const sig2 = computeSignature(secret, timestamp, '{"b":2}'); + + expect(sig1).not.toBe(sig2); + }); + + it("verifySignature returns true for valid signature", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const body = '{"test":true}'; + const sig = computeSignature(secret, timestamp, body); + + expect(verifySignature(secret, timestamp, body, sig)).toBe(true); + }); + + it("verifySignature returns false for tampered body", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const sig = computeSignature(secret, timestamp, '{"test":true}'); + + expect(verifySignature(secret, timestamp, '{"test":false}', sig)).toBe(false); + }); + + it("verifySignature returns false for wrong secret", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const body = '{"test":true}'; + const sig = computeSignature(secret, timestamp, body); + + expect(verifySignature("wrong-secret", timestamp, body, sig)).toBe(false); + }); + + it("verifySignature returns false for missing v1= prefix", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const body = '{"test":true}'; + + expect(verifySignature(secret, timestamp, body, "badformat")).toBe(false); + }); + + it("verifySignature returns false for null/undefined signature", () => { + const timestamp = "2025-01-01T00:00:00.000Z"; + const body = '{"test":true}'; + + expect(verifySignature(secret, timestamp, body, null)).toBe(false); + expect(verifySignature(secret, timestamp, body, undefined)).toBe(false); + }); +}); + +describe("serializePayload", () => { + it("produces deterministic JSON with sorted keys", () => { + const payload = { z: 1, a: 2, m: 3 }; + const s1 = serializePayload(payload); + const s2 = serializePayload({ m: 3, a: 2, z: 1 }); + + expect(s1).toBe(s2); + expect(s1).toBe('{"a":2,"m":3,"z":1}'); + }); +}); + +describe("emitEvent", () => { + it("creates delivery rows for matching active endpoints", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + await emitEvent("payment.confirmed", { transactionId: "tx123" }); + + const deliveries = await WebhookDelivery.find(); + expect(deliveries).toHaveLength(1); + expect(deliveries[0].eventType).toBe("payment.confirmed"); + expect(deliveries[0].payload.data.transactionId).toBe("tx123"); + expect(deliveries[0].payload.apiVersion).toBe(WEBHOOK_API_VERSION); + expect(deliveries[0].payload.eventType).toBe("payment.confirmed"); + expect(deliveries[0].payload.eventId).toBeDefined(); + expect(deliveries[0].status).toBe("pending"); + }); + + it("creates delivery for wildcard subscriber", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + owner: new mongoose.Types.ObjectId(), + }); + + await emitEvent("wallet.connected", { userId: "u1" }); + + const deliveries = await WebhookDelivery.find(); + expect(deliveries).toHaveLength(1); + expect(deliveries[0].eventType).toBe("wallet.connected"); + }); + + it("does not create delivery for non-matching endpoints", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.failed"], + owner: new mongoose.Types.ObjectId(), + }); + + await emitEvent("payment.confirmed", { transactionId: "tx123" }); + + const deliveries = await WebhookDelivery.find(); + expect(deliveries).toHaveLength(0); + }); + + it("does not create delivery for inactive endpoints", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + isActive: false, + owner: new mongoose.Types.ObjectId(), + }); + + await emitEvent("payment.confirmed", { transactionId: "tx123" }); + + const deliveries = await WebhookDelivery.find(); + expect(deliveries).toHaveLength(0); + }); + + it("silently ignores unknown event types", async () => { + await emitEvent("unknown.event", { foo: "bar" }); + const deliveries = await WebhookDelivery.find(); + expect(deliveries).toHaveLength(0); + }); + + it("payload contains only allowlisted fields (no secrets)", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + await emitEvent("payment.confirmed", { + transactionId: "tx1", + buyerId: "b1", + amount: "10", + stellarTxHash: "abc123", + }); + + const delivery = await WebhookDelivery.findOne(); + const data = delivery.payload.data; + + // Should have the allowed fields + expect(data.transactionId).toBe("tx1"); + expect(data.amount).toBe("10"); + + // Should NOT have sensitive fields (even if passed) + expect(data.password).toBeUndefined(); + expect(data.email).toBeUndefined(); + expect(data.secretKey).toBeUndefined(); + }); +}); + +describe("Delivery worker claim atomicity", () => { + it("two concurrent processDeliveries calls do not double-send", async () => { + // Create an endpoint + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + // Create a pending delivery + const payload = { + eventId: "test-event-1", + eventType: "payment.confirmed", + apiVersion: WEBHOOK_API_VERSION, + createdAt: new Date().toISOString(), + data: { transactionId: "tx1" }, + }; + await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "test-event-1", + eventType: "payment.confirmed", + payload, + status: "pending", + nextAttemptAt: new Date(), + }); + + // Try to process with two concurrent calls + // Both will try to claim the same delivery atomically + const [count1, count2] = await Promise.all([ + processDeliveries(1), + processDeliveries(1), + ]); + + // At most one should have processed the delivery + const total = count1 + count2; + expect(total).toBeLessThanOrEqual(1); + + // The delivery should be in a terminal or processing state with at most 1 attempt + const delivery = await WebhookDelivery.findOne({ eventId: "test-event-1" }); + expect(delivery.attempts.length).toBeLessThanOrEqual(1); + }); +}); + +describe("Retry and backoff schedule", () => { + it("failed delivery transitions to retrying with future nextAttemptAt", async () => { + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + const payload = { + eventId: "test-event-retry", + eventType: "payment.confirmed", + apiVersion: WEBHOOK_API_VERSION, + createdAt: new Date().toISOString(), + data: { transactionId: "tx-retry" }, + }; + await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "test-event-retry", + eventType: "payment.confirmed", + payload, + status: "pending", + nextAttemptAt: new Date(), + }); + + // Process — axios is not mocked here so it will fail (ECONNREFUSED) + await processDeliveries(1); + + const delivery = await WebhookDelivery.findOne({ eventId: "test-event-retry" }); + expect(delivery.status).toBe("retrying"); + expect(delivery.attempts).toHaveLength(1); + expect(delivery.attempts[0].at).toBeDefined(); + expect(delivery.nextAttemptAt.getTime()).toBeGreaterThan(Date.now()); + }); + + it("delivery becomes dead after max attempts", async () => { + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + const payload = { + eventId: "test-event-dead", + eventType: "payment.confirmed", + apiVersion: WEBHOOK_API_VERSION, + createdAt: new Date().toISOString(), + data: { transactionId: "tx-dead" }, + }; + const delivery = await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "test-event-dead", + eventType: "payment.confirmed", + payload, + status: "retrying", + nextAttemptAt: new Date(), + }); + + // Simulate 5 previous failed attempts + delivery.attempts = Array(5).fill({ + at: new Date(), + error: "Connection refused", + durationMs: 100, + }); + delivery.attempts.length = 5; // override getter + await delivery.save(); + + // Process — this should be attempt 6 (the max), transitioning to dead + await processDeliveries(1); + + const updated = await WebhookDelivery.findOne({ eventId: "test-event-dead" }); + expect(updated.status).toBe("dead"); + }); + + it("endpoint auto-disables after sustained failures", async () => { + const endpoint = await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["*"], + consecutiveFailures: 19, + owner: new mongoose.Types.ObjectId(), + }); + + const payload = { + eventId: "test-event-disable", + eventType: "payment.confirmed", + apiVersion: WEBHOOK_API_VERSION, + createdAt: new Date().toISOString(), + data: { transactionId: "tx-disable" }, + }; + await WebhookDelivery.create({ + endpoint: endpoint._id, + eventId: "test-event-disable", + eventType: "payment.confirmed", + payload, + status: "pending", + nextAttemptAt: new Date(), + }); + + await processDeliveries(1); + + const updated = await WebhookEndpoint.findById(endpoint._id); + expect(updated.isActive).toBe(false); + expect(updated.disabledAt).toBeDefined(); + expect(updated.disabledReason).toContain("Auto-disabled"); + }); +}); + +describe("SSRF guard", () => { + it("rejects non-https URLs when requireHttps is true", async () => { + const result = await validateEndpointUrl("http://example.com/hook", { requireHttps: true }); + expect(result.valid).toBe(false); + expect(result.error).toContain("HTTPS"); + }); + + it("accepts https URLs when requireHttps is true (skipping DNS)", async () => { + const result = await validateEndpointUrl("https://example.com/hook", { requireHttps: false }); + expect(result.valid).toBe(true); + }); + + it("accepts http URLs in development", async () => { + const original = process.env.NODE_ENV; + process.env.NODE_ENV = "development"; + + const result = await validateEndpointUrl("http://localhost:3000/hook"); + expect(result.valid).toBe(true); + + process.env.NODE_ENV = original; + }); + + it("rejects invalid URL format", async () => { + const result = await validateEndpointUrl("not-a-url"); + expect(result.valid).toBe(false); + expect(result.error).toContain("Invalid URL"); + }); + + it("rejects non-http protocols", async () => { + const result = await validateEndpointUrl("ftp://example.com/hook"); + expect(result.valid).toBe(false); + }); + + it("rejects loopback IP literals in production", async () => { + const original = process.env.NODE_ENV; + process.env.NODE_ENV = "production"; + + // Use requireHttps: false to bypass HTTPS check and test IP validation directly + const result = await validateEndpointUrl("http://127.0.0.1/hook", { requireHttps: false }); + expect(result.valid).toBe(false); + expect(result.error).toContain("private or loopback"); + + process.env.NODE_ENV = original; + }); + + it("rejects 10.x.x.x in production", async () => { + const original = process.env.NODE_ENV; + process.env.NODE_ENV = "production"; + + const result = await validateEndpointUrl("http://10.0.0.1/hook", { requireHttps: false }); + expect(result.valid).toBe(false); + + process.env.NODE_ENV = original; + }); + + it("production rejects http before checking IP", async () => { + const original = process.env.NODE_ENV; + process.env.NODE_ENV = "production"; + + // http://127.0.0.1 fails HTTPS check first (requireHttps defaults to true in production) + const result = await validateEndpointUrl("http://127.0.0.1/hook"); + expect(result.valid).toBe(false); + expect(result.error).toContain("HTTPS"); + + process.env.NODE_ENV = original; + }); +}); + +describe("Payload allowlist", () => { + it("emitted data is stored exactly as provided (allowlisting is at controller level)", async () => { + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + const allowedData = { + transactionId: "tx1", + buyerId: "b1", + buyerWallet: "GABC...", + creatorId: "c1", + creatorWallet: "GDEF...", + itemType: "course", + itemId: "item1", + itemTitle: "Test Course", + amount: "10", + network: "testnet", + stellarTxHash: "hash123", + ledger: 12345, + }; + + await emitEvent("payment.confirmed", allowedData); + + const delivery = await WebhookDelivery.findOne(); + const data = delivery.payload.data; + + // Verify all expected fields are present + expect(data.transactionId).toBe("tx1"); + expect(data.buyerId).toBe("b1"); + expect(data.buyerWallet).toBe("GABC..."); + expect(data.amount).toBe("10"); + expect(data.stellarTxHash).toBe("hash123"); + + // Verify payload structure includes envelope fields + expect(delivery.payload.apiVersion).toBe(WEBHOOK_API_VERSION); + expect(delivery.payload.eventType).toBe("payment.confirmed"); + expect(delivery.payload.eventId).toBeDefined(); + expect(delivery.payload.createdAt).toBeDefined(); + }); + + it("controllers only pass allowlisted fields (integration test for paymentController)", async () => { + // This verifies that the emitEvent calls in controllers + // only include safe, non-sensitive fields + await WebhookEndpoint.create({ + url: "https://example.com/hook", + secret: "hashed", + events: ["payment.confirmed"], + owner: new mongoose.Types.ObjectId(), + }); + + // Simulate what paymentController passes + await emitEvent("payment.confirmed", { + transactionId: "tx1", + buyerId: "b1", + buyerWallet: "GABC...", + creatorId: "c1", + creatorWallet: "GDEF...", + itemType: "course", + itemId: "item1", + itemTitle: "Test Course", + amount: "10", + network: "testnet", + stellarTxHash: "hash123", + ledger: 12345, + }); + + const delivery = await WebhookDelivery.findOne(); + const data = delivery.payload.data; + + // Verify no sensitive fields leaked + expect(data.password).toBeUndefined(); + expect(data.email).toBeUndefined(); + expect(data.secretKey).toBeUndefined(); + expect(data._id).toBeUndefined(); // Mongo ObjectId not passed + }); +}); + +describe("Event type catalog", () => { + it("exports all expected event types", () => { + expect(EVENT_TYPES).toContain("payment.initialized"); + expect(EVENT_TYPES).toContain("payment.confirmed"); + expect(EVENT_TYPES).toContain("payment.failed"); + expect(EVENT_TYPES).toContain("payment.expired"); + expect(EVENT_TYPES).toContain("course.enrolled"); + expect(EVENT_TYPES).toContain("wallet.connected"); + expect(EVENT_TYPES).toContain("wallet.disconnected"); + }); + + it("api version is a stable string", () => { + expect(WEBHOOK_API_VERSION).toMatch(/^\d{4}-\d{2}-\d{2}$/); + }); +});