Webhook Alerts: real-time depeg and risk event delivery

const crypto = require("crypto");

// secret = the subscription's secret_key (from creation/rotate-secret)
function isValid(body, secret) {
  // reject missing/malformed signatures before any buffer work
  if (typeof body?.signature !== "string" || !/^[a-f0-9]{64}$/i.test(body.signature)) {
    return false;
  }
  const expected = crypto
    .createHmac("sha256", secret)
    .update(JSON.stringify(body.event)) // compact JSON of the event object
    .digest("hex");
  // decode as hex (32-byte digests), not UTF-8; timingSafeEqual needs equal lengths
  const expectedBuf = Buffer.from(expected, "hex");
  const providedBuf = Buffer.from(body.signature, "hex");
  return expectedBuf.length === providedBuf.length &&
    crypto.timingSafeEqual(expectedBuf, providedBuf);
}

// Express example
app.post("/webhooks/webacy", express.json(), (req, res) => {
  if (!isValid(req.body, process.env.WEBACY_WEBHOOK_SECRET)) {
    return res.status(401).end();
  }
  // ...handle req.body.event...
  res.status(200).end(); // any 2xx acknowledges and stops retries
});

import hmac, hashlib, json

def is_valid(body: dict, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        # compact JSON of the event object; ensure_ascii=False matches our serializer
        json.dumps(body["event"], separators=(",", ":"), ensure_ascii=False).encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, body["signature"])

# Flask example
@app.post("/webhooks/webacy")
def webacy_webhook():
    body = request.get_json()
    if not is_valid(body, os.environ["WEBACY_WEBHOOK_SECRET"]):
        return "", 401
    # ...handle body["event"]...
    return "", 200  # any 2xx acknowledges and stops retries