أحداث Webhooks وحمولاتها والتحقق من التواقيع

{
  "id": "evt_3f2b8c1d9a7e4f60b5c2d8e1a4f7b9c3",
  "type": "message.received",
  "created_at": "2026-10-21T09:30:00.000Z",
  "payload_version": "1",
  "data": {
    "messageId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "conversationId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "contactId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "direction": "INBOUND",
    "type": "TEXT",
    "metaMessageId": "wamid.HBgLMTU1NTEyMzQ1NjcVAgASGBQzQUY"
  }
}

{
  "id": "evt_3f2b8c1d9a7e4f60b5c2d8e1a4f7b9c3",
  "type": "message.received",
  "occurred_at": "2026-10-21T09:30:00.000Z",
  "version": "2026-08-11",
  "correlation_id": null,
  "data": {
    "messageId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "conversationId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "contactId": "clxxxxxxxxxxxxxxxxxxxxxxxx",
    "direction": "INBOUND",
    "type": "TEXT",
    "metaMessageId": "wamid.HBgLMTU1NTEyMzQ1NjcVAgASGBQzQUY"
  }
}

import { createHmac, timingSafeEqual } from "node:crypto";

// rawBody: the request body as a Buffer, before any JSON parsing.
// secrets: the secrets you accept, for example [current] or [current, previous].
export function isValidFlowellaSignature(rawBody, headers, secrets) {
  const received = [
    headers["x-flowella-signature"],
    headers["x-flowella-signature-previous"],
  ].filter(Boolean);

  return secrets.some((secret) => {
    const expected = createHmac("sha256", secret).update(rawBody).digest();
    return received.some((value) => {
      const given = Buffer.from(value, "hex");
      return given.length === expected.length && timingSafeEqual(given, expected);
    });
  });
}

// Express: keep the raw body for this route.
// app.post("/flowella", express.raw({ type: "application/json" }), (req, res) => {
//   if (!isValidFlowellaSignature(req.body, req.headers, [process.env.FLOWELLA_WEBHOOK_SECRET])) {
//     return res.sendStatus(401);
//   }
//   const event = JSON.parse(req.body.toString("utf8"));
//   res.sendStatus(200); // reply first, then process the event in the background
// });

import hashlib
import hmac


def is_valid_flowella_signature(raw_body: bytes, headers, secrets: list[str]) -> bool:
    """raw_body is the request body exactly as received, before JSON parsing."""
    received = [
        value.lower()
        for value in (
            headers.get("X-Flowella-Signature"),
            headers.get("X-Flowella-Signature-Previous"),
        )
        if value
    ]
    for secret in secrets:
        expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, value) for value in received):
            return True
    return False