JITSU-100: webhook destination optional request signing (HMAC / Ed25519)

libs/destination-functions/__tests__/webhook-destination.test.ts

@@ -0,0 +1,120 @@
+import { createHmac, generateKeyPairSync, verify } from "crypto";
+import { afterAll, afterEach, beforeAll, expect, test } from "vitest";
+import { http, HttpResponse } from "msw";
+import { setupServer } from "msw/node";
+import { AnalyticsServerEvent } from "@jitsu/protocols/analytics";
+import { testJitsuFunction } from "./lib/testing-lib";
+import WebhookDestination from "../src/functions/webhook-destination";
+import { WebhookDestinationConfig } from "../src/meta";
+
+// MSW intercepts the real fetch so we assert on the actual signed request.
+const url = "http://webhook.test.local/hook";
+const secret = "super-secret-key";
+const server = setupServer();
+
+beforeAll(() => server.listen({ onUnhandledRequest: "error" }));
+afterEach(() => server.resetHandlers());
+afterAll(() => server.close());
+
+// Runs the webhook destination once and returns the signature headers + sent body.
+async function sign(
+  config: Partial<WebhookDestinationConfig>
+): Promise<{ signature: string; timestamp: string | null; body: string }> {
+  let captured: { signature: string; timestamp: string | null; body: string } | undefined;
+  server.use(
+    http.post(url, async ({ request }) => {
+      captured = {
+        signature: request.headers.get("Jitsu-Signature") ?? "",
+        timestamp: request.headers.get("Jitsu-Signature-Timestamp"),
+        body: await request.text(),
+      };
+      return HttpResponse.text("ok");
+    })
+  );
+  await testJitsuFunction<WebhookDestinationConfig>({
+    func: WebhookDestination,
+    config: { url, method: "POST", ...config } as WebhookDestinationConfig,
+    events: [{ type: "track", event: "test_event", messageId: "m1" } as AnalyticsServerEvent],
+  });
+  if (!captured) throw new Error("webhook was never called");
+  return captured;
+}
+
+test("hmac: signs timestamp + body and sends a timestamp header by default", async () => {
+  const { signature, timestamp, body } = await sign({ signatureMethod: "hmac", signatureSecret: secret });
+  expect(timestamp).toMatch(/^\d+$/);
+  expect(signature).toBe(createHmac("sha256", secret).update(`${timestamp}.${body}`).digest("hex"));
+});
+
+test("hmac: signs the body only and omits the timestamp header when replay protection is off", async () => {
+  const { signature, timestamp, body } = await sign({
+    signatureMethod: "hmac",
+    signatureSecret: secret,
+    signatureIncludeTimestamp: false,
+  });
+  expect(timestamp).toBeNull();
+  expect(signature).toBe(createHmac("sha256", secret).update(body).digest("hex"));
+});
+
+test("ed25519: signature verifies against the public key", async () => {
+  const { publicKey, privateKey } = generateKeyPairSync("ed25519");
+  const privatePem = privateKey.export({ type: "pkcs8", format: "pem" }).toString();
+
+  const { signature, timestamp, body } = await sign({
+    signatureMethod: "ed25519",
+    signaturePrivateKey: privatePem,
+  });
+
+  expect(timestamp).toMatch(/^\d+$/);
+  const ok = verify(null, Buffer.from(`${timestamp}.${body}`), publicKey, Buffer.from(signature, "hex"));
+  expect(ok).toBe(true);
+});
+
+test("a signing method with no key configured fails without retrying", async () => {
+  server.use(http.post(url, () => HttpResponse.text("ok")));
+  await expect(
+    testJitsuFunction<WebhookDestinationConfig>({
+      func: WebhookDestination,
+      config: { url, method: "POST", signatureMethod: "hmac" } as WebhookDestinationConfig,
+      events: [{ type: "track", event: "test_event", messageId: "m1" } as AnalyticsServerEvent],
+    })
+  ).rejects.toMatchObject({ name: "NoRetryError" });
+});
+
+test("schema rejects an invalid signature header name", () => {
+  expect(
+    WebhookDestinationConfig.safeParse({ url: "https://example.com", signatureHeader: "Bad Header" }).success
+  ).toBe(false);
+  expect(
+    WebhookDestinationConfig.safeParse({ url: "https://example.com", signatureHeader: "X-My-Signature" }).success
+  ).toBe(true);
+});
+
+test("ed25519: a malformed private key fails without retrying", async () => {
+  server.use(http.post(url, () => HttpResponse.text("ok")));
+  await expect(
+    testJitsuFunction<WebhookDestinationConfig>({
+      func: WebhookDestination,
+      config: {
+        url,
+        method: "POST",
+        signatureMethod: "ed25519",
+        signaturePrivateKey: "not-a-real-key",
+      } as WebhookDestinationConfig,
+      events: [{ type: "track", event: "test_event", messageId: "m1" } as AnalyticsServerEvent],
+    })
+  ).rejects.toMatchObject({ name: "NoRetryError" });
+});
+
+test("ed25519: accepts a bare base64 key body with surrounding whitespace", async () => {
+  const { publicKey, privateKey } = generateKeyPairSync("ed25519");
+  const bareBase64 = privateKey.export({ type: "pkcs8", format: "der" }).toString("base64");
+
+  const { signature, timestamp, body } = await sign({
+    signatureMethod: "ed25519",
+    signaturePrivateKey: `\n  ${bareBase64}  \n`,
+  });
+
+  const ok = verify(null, Buffer.from(`${timestamp}.${body}`), publicKey, Buffer.from(signature, "hex"));
+  expect(ok).toBe(true);
+});

libs/destination-functions/package.json

@@ -28,6 +28,7 @@
     "@types/node": "catalog:",
     "@vitest/ui": "catalog:",
     "json5": "^2.1.0",
+    "msw": "^2.15.0",
     "vitest": "catalog:",
     "vite": "^6.4.3"
   },

libs/destination-functions/src/functions/webhook-destination.ts

@@ -1,5 +1,6 @@
+import { createHmac, createPrivateKey, sign } from "crypto";
 import { JitsuFunction } from "@jitsu/protocols/functions";
-import { HTTPError, RetryError } from "@jitsu/functions-lib";
+import { HTTPError, NoRetryError, NoRetryErrorName, RetryError } from "@jitsu/functions-lib";
 import type { AnalyticsServerEvent } from "@jitsu/protocols/analytics";
 import { WebhookDestinationConfig } from "../meta";
 import { MetricsMeta } from "@jitsu/core-functions-lib";
@@ -10,6 +11,21 @@ const bulkerAuthKey = process.env.BULKER_AUTH_KEY;
 
 const macrosPattern = /\{\{\s*([\w.-]+)\s*}}/g;
 
+// Accepts an Ed25519 private key as full PEM or as the bare base64 PKCS#8 body
+// (with any surrounding whitespace), returning a PEM string createPrivateKey parses.
+function toPrivateKeyPem(input: string): string {
+  const trimmed = input.trim();
+  if (trimmed.includes("-----BEGIN")) {
+    return trimmed;
+  }
+  const body =
+    trimmed
+      .replace(/\s+/g, "")
+      .match(/.{1,64}/g)
+      ?.join("\n") ?? trimmed;
+  return `-----BEGIN PRIVATE KEY-----\n${body}\n-----END PRIVATE KEY-----`;
+}
+
 const WebhookDestination: JitsuFunction<AnalyticsServerEvent, WebhookDestinationConfig> = async (event, ctx) => {
   if (ctx["connectionOptions"]?.mode === "batch" && bulkerBase) {
     const metricsMeta: Omit<MetricsMeta, "messageId"> = {
@@ -79,6 +95,38 @@ const WebhookDestination: JitsuFunction<AnalyticsServerEvent, WebhookDestination
         payload = JSON.stringify(event);
       }
       const headers = ctx.props.headers || [];
+      const signatureHeaders: Record<string, string> = {};
+      const method = ctx.props.signatureMethod || "none";
+      if (method !== "none") {
+        const secret = method === "hmac" ? ctx.props.signatureSecret : undefined;
+        const privateKey = method === "ed25519" ? ctx.props.signaturePrivateKey : undefined;
+        if (!secret && !privateKey) {
+          // Method selected but no key — fail loudly instead of silently sending unsigned.
+          throw new NoRetryError(
+            `Webhook signing method "${method}" is enabled but no ${
+              method === "hmac" ? "secret" : "private key"
+            } is configured`
+          );
+        }
+        const headerName = ctx.props.signatureHeader || "Jitsu-Signature";
+        let signedData = payload;
+        let timestamp: string | undefined;
+        if (ctx.props.signatureIncludeTimestamp ?? true) {
+          timestamp = String(Math.floor(Date.now() / 1000));
+          signedData = `${timestamp}.${payload}`;
+        }
+        try {
+          signatureHeaders[headerName] = secret
+            ? createHmac("sha256", secret).update(signedData).digest("hex")
+            : sign(null, Buffer.from(signedData), createPrivateKey(toPrivateKeyPem(privateKey!))).toString("hex");
+        } catch (e: any) {
+          // A malformed key/secret can never succeed on retry — drop instead of retrying forever.
+          throw new NoRetryError(`Webhook signing failed, check the signing key/secret: ${e.message}`);
+        }
+        if (timestamp) {
+          signatureHeaders[`${headerName}-Timestamp`] = timestamp;
+        }
+      }
       const res = await ctx.fetch(ctx.props.url, {
         method: ctx.props.method || "POST",
         body: payload,
@@ -88,6 +136,7 @@ const WebhookDestination: JitsuFunction<AnalyticsServerEvent, WebhookDestination
             const [key, value] = header.split(":");
             return { ...res, [key]: value };
           }, {}),
+          ...signatureHeaders,
         },
       });
       if (!res.ok) {
@@ -103,6 +152,9 @@ const WebhookDestination: JitsuFunction<AnalyticsServerEvent, WebhookDestination
       }
       return event;
     } catch (e: any) {
+      if (e.name === NoRetryErrorName) {
+        throw e;
+      }
       throw new RetryError(e.message);
     }
   }

libs/destination-functions/src/meta.ts

@@ -47,6 +47,41 @@ export const WebhookDestinationConfig = z.object({
     .describe(
       "Payload Template::Template for the webhook payload. The following macros are supported:<ul><li><code>{{ EVENT }}</code> - event json object for stream mode or batches with size=1</li><li><code>{{ EVENTS }}</code> - for batch mode - json array of events</li><li><code>{{ EVENTS_COUNT }}</code> - count of events in batch</li><li><code>{{ NAME }}</code> - event name</li><li><code>{{ env.VAR_NAME }}</code> - value of VAR_NAME environment variable</li></ul>"
     ),
+  signatureMethod: z
+    .enum(["none", "hmac", "ed25519"])
+    .optional()
+    .default("none")
+    .describe(
+      "Signature Method::How outgoing requests are signed so your endpoint can verify them.<ul><li><code>none</code> — requests are not signed.</li><li><code>hmac</code> — symmetric HMAC-SHA256 with a shared secret (like Stripe/GitHub). Simple, but anyone who can verify can also forge, so only use it for endpoints you fully trust.</li><li><code>ed25519</code> — asymmetric signature. Jitsu signs with a private key; your endpoint verifies with the public key and cannot forge messages. Prefer this when many or untrusted endpoints receive the webhook.</li></ul><b>Note:</b> signing applies to <b>stream</b>-mode delivery only. In <b>batch</b> mode events are delivered by Bulker and are <b>not</b> signed — use stream mode if you need signed webhooks."
+    ),
+  signatureSecret: z
+    .string()
+    .optional()
+    .describe(
+      "Signing Secret::Shared secret for <code>HMAC-SHA256</code> signing (symmetric — there is no key pair). Put the <b>same</b> value here and in your receiving endpoint. Generate a random one with <code>openssl rand -hex 32</code> (any high-entropy string of 32+ bytes works). To verify a request, your endpoint recomputes the HMAC over the signed payload with this secret and compares the hex digest to the <code>Jitsu-Signature</code> header using a constant-time comparison. Keep it secret."
+    ),
+  signaturePrivateKey: z
+    .string()
+    .optional()
+    .describe(
+      "Private Key (PEM)::Ed25519 private key used to sign requests. Generate a key pair:<br/><code>openssl genpkey -algorithm ed25519 -out jitsu_private.pem</code><br/><code>openssl pkey -in jitsu_private.pem -pubout -out jitsu_public.pem</code><br/>Paste the contents of <code>jitsu_private.pem</code> here and install <code>jitsu_public.pem</code> on your endpoint. To verify a request, your endpoint checks the <code>Jitsu-Signature</code> header (a hex Ed25519 signature) against the signed payload using the public key. Jitsu holds only the private key — keep it secret; the public key is safe to distribute."
+    ),
+  signatureHeader: z
+    .string()
+    .regex(
+      /^[A-Za-z0-9!#$%&'*+.^_`|~-]+$/,
+      "Must be a valid HTTP header name (letters, digits, and !#$%&'*+-.^_`|~ only)"
+    )
+    .optional()
+    .default("Jitsu-Signature")
+    .describe("Signature Header::Name of the header carrying the signature. Only used when signing is enabled."),
+  signatureIncludeTimestamp: z
+    .boolean()
+    .optional()
+    .default(true)
+    .describe(
+      "Replay Protection::When enabled (recommended), the request timestamp (unix seconds) is folded into the signed payload — the signature covers <code>&lt;timestamp&gt;.&lt;body&gt;</code> — and sent in a companion <code>&lt;Signature Header&gt;-Timestamp</code> header. Your endpoint can then reject requests with an old timestamp to block replays of a captured, still-valid request. When disabled, only the raw body is signed and no timestamp header is sent: simpler to verify, but a captured request stays valid forever. Only used when signing is enabled."
+    ),
 });
 
 export type WebhookDestinationConfig = z.infer<typeof WebhookDestinationConfig>;