379 lines
11 KiB
JavaScript
379 lines
11 KiB
JavaScript
const crypto = require("crypto");
|
|
const express = require("express");
|
|
const { createLogger } = require("./logger");
|
|
|
|
const webhookLog = createLogger("core:webhooks", { category: "integration" });
|
|
|
|
function createWebhookService({ limit = "256kb" } = {}) {
|
|
const endpoints = new Map();
|
|
const endpointKeysByPlugin = new Map();
|
|
const router = express.Router();
|
|
|
|
router.use(express.raw({ type: "*/*", limit }));
|
|
router.all("/:namespace/:slug", async (req, res) => {
|
|
const namespace = normalizeSegment(req.params.namespace);
|
|
const slug = normalizeSegment(req.params.slug);
|
|
const endpoint = namespace && slug
|
|
? endpoints.get(`${namespace}/${slug}`)
|
|
: null;
|
|
if (!endpoint) {
|
|
return res.status(404).json({ error: "Webhook endpoint not found." });
|
|
}
|
|
if (!endpoint.methods.includes(req.method.toUpperCase())) {
|
|
res.set("Allow", endpoint.methods.join(", "));
|
|
return res.status(405).json({ error: "Method not allowed." });
|
|
}
|
|
|
|
const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body || "");
|
|
const rawBodyText = rawBody.toString("utf8");
|
|
let parsedJson = null;
|
|
let jsonError = null;
|
|
if (rawBody.length) {
|
|
try {
|
|
parsedJson = JSON.parse(rawBodyText);
|
|
} catch (error) {
|
|
jsonError = error;
|
|
}
|
|
}
|
|
const context = {
|
|
req,
|
|
headers: { ...req.headers },
|
|
method: req.method,
|
|
namespace,
|
|
slug,
|
|
rawBody,
|
|
rawBodyText,
|
|
parsedJson,
|
|
jsonError,
|
|
receivedAt: Date.now(),
|
|
remoteAddress: req.ip || req.socket?.remoteAddress || null,
|
|
userAgent: req.get("user-agent") || null
|
|
};
|
|
|
|
try {
|
|
if (endpoint.verify) {
|
|
const verification = await endpoint.verify(context);
|
|
if (verification === false || verification?.ok === false) {
|
|
return res.status(Number(verification?.status) || 401).json({
|
|
error: verification?.message || "Webhook verification failed."
|
|
});
|
|
}
|
|
context.verification = verification || { ok: true };
|
|
}
|
|
return sendHandlerResult(res, await endpoint.handler(context));
|
|
} catch (error) {
|
|
webhookLog.error("Webhook handler failed", {
|
|
pluginId: endpoint.pluginId,
|
|
endpointId: endpoint.endpointId,
|
|
namespace,
|
|
slug,
|
|
message: error?.message || String(error),
|
|
stack: error?.stack || ""
|
|
}, { event: "inbound_webhook_failed" });
|
|
if (!res.headersSent) {
|
|
return res.status(500).json({ error: "Webhook processing failed." });
|
|
}
|
|
return null;
|
|
}
|
|
});
|
|
router.use((error, req, res, _next) => {
|
|
const status = error?.type === "entity.too.large" ? 413 : 400;
|
|
webhookLog.warn("Webhook request rejected", {
|
|
path: req.path,
|
|
status,
|
|
message: error?.message || String(error)
|
|
}, { event: "inbound_webhook_rejected" });
|
|
res.status(status).json({
|
|
error: status === 413 ? "Webhook payload is too large." : "Invalid webhook request."
|
|
});
|
|
});
|
|
|
|
function registerInbound({
|
|
pluginId,
|
|
namespace,
|
|
endpointId,
|
|
slug,
|
|
description,
|
|
handler,
|
|
verify,
|
|
options = {}
|
|
}) {
|
|
const safePluginId = requireValue(pluginId, "pluginId");
|
|
const safeEndpointId = requireValue(endpointId, "endpointId");
|
|
const safeNamespace = requireSegment(namespace, "namespace");
|
|
const safeSlug = requireSegment(slug, "slug");
|
|
if (typeof handler !== "function") {
|
|
throw new Error("Webhook handler must be a function.");
|
|
}
|
|
const key = `${safeNamespace}/${safeSlug}`;
|
|
if (endpoints.has(key)) {
|
|
throw new Error(`Webhook endpoint already registered: ${key}`);
|
|
}
|
|
unregisterInbound({ pluginId: safePluginId, endpointId: safeEndpointId });
|
|
const endpoint = {
|
|
pluginId: safePluginId,
|
|
endpointId: safeEndpointId,
|
|
namespace: safeNamespace,
|
|
slug: safeSlug,
|
|
description: (description || "").toString(),
|
|
handler,
|
|
verify: typeof verify === "function" ? verify : null,
|
|
methods: normalizeMethods(options.methods),
|
|
options
|
|
};
|
|
endpoints.set(key, endpoint);
|
|
if (!endpointKeysByPlugin.has(safePluginId)) {
|
|
endpointKeysByPlugin.set(safePluginId, new Map());
|
|
}
|
|
endpointKeysByPlugin.get(safePluginId).set(safeEndpointId, key);
|
|
webhookLog.info("Webhook endpoint registered", {
|
|
pluginId: safePluginId,
|
|
endpointId: safeEndpointId,
|
|
path: `/webhooks/${key}`
|
|
}, { event: "webhook_registered" });
|
|
return { namespace: safeNamespace, slug: safeSlug, path: `/webhooks/${key}` };
|
|
}
|
|
|
|
function unregisterInbound({ pluginId, endpointId }) {
|
|
const pluginEndpoints = endpointKeysByPlugin.get((pluginId || "").toString());
|
|
const key = pluginEndpoints?.get((endpointId || "").toString());
|
|
if (!key) {
|
|
return false;
|
|
}
|
|
endpoints.delete(key);
|
|
pluginEndpoints.delete((endpointId || "").toString());
|
|
if (!pluginEndpoints.size) {
|
|
endpointKeysByPlugin.delete((pluginId || "").toString());
|
|
}
|
|
webhookLog.debug("Webhook endpoint unregistered", { pluginId, endpointId }, { event: "webhook_unregistered" });
|
|
return true;
|
|
}
|
|
|
|
function buildPublicUrl({ namespace, slug, req, baseUrl }) {
|
|
const safeNamespace = requireSegment(namespace, "namespace");
|
|
const safeSlug = requireSegment(slug, "slug");
|
|
const origin = (baseUrl || requestOrigin(req) || "").replace(/\/+$/, "");
|
|
const routePath = `/webhooks/${safeNamespace}/${safeSlug}`;
|
|
return origin ? `${origin}${routePath}` : routePath;
|
|
}
|
|
|
|
return {
|
|
router,
|
|
registerInbound,
|
|
unregisterInbound,
|
|
buildPublicUrl,
|
|
generateSlug,
|
|
send: sendWebhook,
|
|
sendJson: ({ payload, ...options }) => sendWebhook({ ...options, json: payload }),
|
|
isTimestampWithinWindow,
|
|
getRegisteredEndpoints: () =>
|
|
Array.from(endpoints.values()).map(({ handler, verify, ...entry }) => entry)
|
|
};
|
|
}
|
|
|
|
async function sendWebhook({
|
|
pluginId,
|
|
url,
|
|
method = "POST",
|
|
headers = {},
|
|
json,
|
|
body,
|
|
timeoutMs = 10000,
|
|
retries = 0,
|
|
sign
|
|
}) {
|
|
if (!url) {
|
|
throw new Error("Webhook URL is required.");
|
|
}
|
|
if (json !== undefined && body !== undefined) {
|
|
throw new Error("Provide either json or body, not both.");
|
|
}
|
|
const requestHeaders = { ...headers };
|
|
let requestBody = body;
|
|
if (json !== undefined) {
|
|
requestBody = JSON.stringify(json);
|
|
if (!hasHeader(requestHeaders, "content-type")) {
|
|
requestHeaders["Content-Type"] = "application/json";
|
|
}
|
|
}
|
|
if (typeof sign === "function") {
|
|
const signedHeaders = await sign({
|
|
method,
|
|
url,
|
|
headers: { ...requestHeaders },
|
|
body: requestBody
|
|
});
|
|
if (signedHeaders && typeof signedHeaders === "object") {
|
|
Object.assign(requestHeaders, signedHeaders);
|
|
}
|
|
}
|
|
|
|
const attempts = Math.max(1, Number(retries) + 1);
|
|
let lastError = null;
|
|
let lastDurationMs = 0;
|
|
for (let attempt = 1; attempt <= attempts; attempt += 1) {
|
|
const startedAt = Date.now();
|
|
const controller = new AbortController();
|
|
const timer = setTimeout(() => controller.abort(), Math.max(1, Number(timeoutMs)));
|
|
try {
|
|
const response = await fetch(url, {
|
|
method,
|
|
headers: requestHeaders,
|
|
body: requestBody,
|
|
signal: controller.signal
|
|
});
|
|
const result = {
|
|
status: response.status,
|
|
headers: Object.fromEntries(response.headers.entries()),
|
|
body: await response.text(),
|
|
durationMs: Date.now() - startedAt,
|
|
success: response.ok
|
|
};
|
|
if (response.ok || attempt === attempts) {
|
|
if (!response.ok) {
|
|
webhookLog.warn("Outbound webhook returned an error", {
|
|
pluginId: pluginId || null,
|
|
url: redactUrl(url),
|
|
status: response.status,
|
|
attempt
|
|
}, { event: "outbound_webhook_error" });
|
|
}
|
|
return result;
|
|
}
|
|
} catch (error) {
|
|
lastError = error;
|
|
lastDurationMs = Date.now() - startedAt;
|
|
if (attempt === attempts) {
|
|
webhookLog.error("Outbound webhook failed", {
|
|
pluginId: pluginId || null,
|
|
url: redactUrl(url),
|
|
attempt,
|
|
message: error?.message || String(error)
|
|
}, { event: "outbound_webhook_failed" });
|
|
}
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
return {
|
|
status: 0,
|
|
headers: {},
|
|
body: "",
|
|
durationMs: lastDurationMs,
|
|
success: false,
|
|
error: lastError?.message || "Webhook request failed."
|
|
};
|
|
}
|
|
|
|
function generateSlug({ identifier, uuid = crypto.randomUUID() }) {
|
|
return `${sanitizeIdentifier(identifier)}-${requireUuid(uuid)}`;
|
|
}
|
|
|
|
function sanitizeIdentifier(value) {
|
|
const safe = (value || "")
|
|
.toString()
|
|
.trim()
|
|
.toLowerCase()
|
|
.replace(/\s+/g, "-")
|
|
.replace(/[^a-z0-9_-]+/g, "")
|
|
.replace(/-+/g, "-")
|
|
.replace(/^[-_]+|[-_]+$/g, "");
|
|
if (!safe) {
|
|
throw new Error("Webhook identifier must contain URL-safe letters or numbers.");
|
|
}
|
|
return safe;
|
|
}
|
|
|
|
function requireUuid(value) {
|
|
const uuid = (value || "").toString().toLowerCase();
|
|
if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(uuid)) {
|
|
throw new Error("Webhook UUID is invalid.");
|
|
}
|
|
return uuid;
|
|
}
|
|
|
|
function requireSegment(value, label) {
|
|
const normalized = normalizeSegment(value);
|
|
if (!normalized) {
|
|
throw new Error(`Webhook ${label} is invalid.`);
|
|
}
|
|
return normalized;
|
|
}
|
|
|
|
function normalizeSegment(value) {
|
|
const raw = (value || "").toString().trim().toLowerCase();
|
|
if (!raw || raw.includes("/") || raw.includes("\\") || raw.includes("..")) {
|
|
return null;
|
|
}
|
|
if (/%2f|%5c/i.test(raw)) {
|
|
return null;
|
|
}
|
|
return /^[a-z0-9][a-z0-9_-]*$/.test(raw) ? raw : null;
|
|
}
|
|
|
|
function requireValue(value, label) {
|
|
const safe = (value || "").toString().trim();
|
|
if (!safe) {
|
|
throw new Error(`Webhook ${label} is required.`);
|
|
}
|
|
return safe;
|
|
}
|
|
|
|
function normalizeMethods(methods) {
|
|
const source = Array.isArray(methods) && methods.length ? methods : ["POST"];
|
|
return Array.from(new Set(source.map((method) => method.toString().toUpperCase())));
|
|
}
|
|
|
|
function sendHandlerResult(res, result) {
|
|
const status = Number(result?.status) || 204;
|
|
if (result?.headers && typeof result.headers === "object") {
|
|
res.set(result.headers);
|
|
}
|
|
if (result?.body === undefined || result?.body === null) {
|
|
return res.status(status).end();
|
|
}
|
|
if (Buffer.isBuffer(result.body) || typeof result.body === "string") {
|
|
return res.status(status).send(result.body);
|
|
}
|
|
return res.status(status).json(result.body);
|
|
}
|
|
|
|
function requestOrigin(req) {
|
|
if (!req) {
|
|
return "";
|
|
}
|
|
const forwardedProto = req.get?.("x-forwarded-proto");
|
|
const protocol = forwardedProto ? forwardedProto.split(",")[0].trim() : req.protocol;
|
|
const host = req.get?.("host");
|
|
return protocol && host ? `${protocol}://${host}` : "";
|
|
}
|
|
|
|
function isTimestampWithinWindow(timestamp, windowSeconds = 300, nowMs = Date.now()) {
|
|
if (!/^\d+$/.test((timestamp || "").toString())) {
|
|
return false;
|
|
}
|
|
const timestampMs = Number(timestamp) * 1000;
|
|
return Number.isFinite(timestampMs) &&
|
|
Math.abs(nowMs - timestampMs) <= Math.max(0, Number(windowSeconds)) * 1000;
|
|
}
|
|
|
|
function hasHeader(headers, name) {
|
|
return Object.keys(headers).some((key) => key.toLowerCase() === name.toLowerCase());
|
|
}
|
|
|
|
function redactUrl(value) {
|
|
try {
|
|
const url = new URL(value);
|
|
return `${url.protocol}//${url.host}${url.pathname}`;
|
|
} catch {
|
|
return "[invalid-url]";
|
|
}
|
|
}
|
|
|
|
module.exports = {
|
|
createWebhookService,
|
|
generateSlug,
|
|
sanitizeIdentifier,
|
|
isTimestampWithinWindow
|
|
};
|