Alert-Webhooks: Team-Alerts an Ihren Endpoint

{
  "alert_id": "alt_9f2c4e1a7b3d5c80",
  "created_at": 1791884460,
  "data": {
    "event": "session_in_review",
    "label": "Session",
    "value": "#1042"
  },
  "organization": {
    "id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
    "name": "Example Org"
  },
  "severity": "warning",
  "subtitle": "A KYC or KYB verification was completed and flagged for manual review.",
  "title": "Verification in review",
  "url": "https://business.didit.me/console/3fa85f64-5717-4562-b3fc-2c963f66afa6/9b2d1c4e-8f3a-4e6b-a1d2-5c7e9f0b3a14/verifications?session-id=1b4e28ba-2fa1-41d2-883f-0016d3cca427",
  "webhook_type": "alert.verification.in_review"
}

const express = require("express");
const crypto = require("crypto");

const app = express();
// The full secret Didit showed when you added the destination, including "whsec_".
const ALERT_SECRET = process.env.DIDIT_ALERT_SECRET;
const TOLERANCE_SECONDS = 300;

// express.raw keeps the body as the exact bytes Didit sent.
app.post("/api/didit/alerts", express.raw({ type: "application/json" }), (req, res) => {
  const timestamp = req.get("X-Timestamp") || "";
  const signature = Buffer.from(req.get("X-Signature-V2") || "", "utf8");

  const now = Math.floor(Date.now() / 1000);
  if (!/^\d+$/.test(timestamp) || Math.abs(now - Number(timestamp)) > TOLERANCE_SECONDS) {
    return res.status(400).send("Stale or missing timestamp");
  }

  const expected = Buffer.from(
    crypto.createHmac("sha256", ALERT_SECRET).update(`${timestamp}.`).update(req.body).digest("hex"),
    "utf8",
  );
  if (signature.length !== expected.length || !crypto.timingSafeEqual(signature, expected)) {
    return res.status(400).send("Invalid signature");
  }

  const alert = JSON.parse(req.body.toString("utf8"));
  res.sendStatus(204); // answer within 5 seconds, then process
  handleAlert(alert); // skip an alert_id you already handled
});

import hashlib
import hmac
import json
import os
import time

from flask import Flask, request

app = Flask(__name__)
# The full secret Didit showed when you added the destination, including "whsec_".
ALERT_SECRET = os.environ["DIDIT_ALERT_SECRET"].encode("utf-8")
TOLERANCE_SECONDS = 300


@app.post("/api/didit/alerts")
def didit_alert():
    raw_body = request.get_data()  # the exact bytes Didit sent
    timestamp = request.headers.get("X-Timestamp", "")
    signature = request.headers.get("X-Signature-V2", "")

    try:
        too_old = abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS
    except ValueError:
        too_old = True
    if too_old:
        return "Stale or missing timestamp", 400

    expected = hmac.new(ALERT_SECRET, timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected.encode(), signature.encode()):
        return "Invalid signature", 400

    alert = json.loads(raw_body)
    handle_alert(alert)  # skip an alert_id you already handled; keep this fast
    return "", 204