Webhooks prüfen

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

const TOLERANCE_SECONDS = 300;

// rawBody: der Körper der Anfrage unverändert (Buffer oder String), nicht das geparste JSON.
// signatureHeader: der Wert von Zenit-Signature, secret: das komplette Webhook-Secret einschließlich zhs_.
export function verifyZenitSignature(
  rawBody,
  signatureHeader,
  secret,
  nowSeconds = Math.floor(Date.now() / 1000),
) {
  const parts = {};
  for (const part of String(signatureHeader ?? '').split(',')) {
    const index = part.indexOf('=');
    if (index > 0) parts[part.slice(0, index).trim()] = part.slice(index + 1).trim();
  }

  const timestamp = Number(parts.t);
  if (!Number.isInteger(timestamp) || !parts.v1) return false;
  if (Math.abs(nowSeconds - timestamp) > TOLERANCE_SECONDS) return false;

  const expected = createHmac('sha256', secret)
    .update(parts.t + '.')
    .update(rawBody)
    .digest();
  const received = Buffer.from(parts.v1, 'hex');

  return received.length === expected.length && timingSafeEqual(received, expected);
}

import express from 'express';

const app = express();

// express.raw liefert den Körper als Buffer, bevor ihn etwas als JSON liest.
app.post('/zenit/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  // Kein Buffer heißt: Der Körper wurde schon anders gelesen oder hatte einen anderen Content-Type.
  if (!Buffer.isBuffer(req.body)) return res.sendStatus(400);

  const valid = verifyZenitSignature(
    req.body,
    req.get('Zenit-Signature'),
    process.env.ZENIT_WEBHOOK_SECRET,
  );
  if (!valid) return res.sendStatus(400);

  const event = JSON.parse(req.body.toString('utf8'));
  // Zuerst prüfen, ob event.id schon verarbeitet ist, dann verarbeiten.
  res.sendStatus(204);
});

import hashlib
import hmac
import time

TOLERANCE_SECONDS = 300


def verify_zenit_signature(raw_body: bytes, signature_header: str, secret: str, now=None) -> bool:
    # raw_body: der Körper der Anfrage unverändert, nicht das geparste JSON.
    # signature_header: der Wert von Zenit-Signature, secret: das komplette Webhook-Secret einschließlich zhs_.
    parts = {}
    for part in (signature_header or "").split(","):
        key, separator, value = part.partition("=")
        if separator:
            parts[key.strip()] = value.strip()

    try:
        timestamp = int(parts["t"])
        received = parts["v1"]
    except (KeyError, ValueError):
        return False

    current = time.time() if now is None else now
    if abs(current - timestamp) > TOLERANCE_SECONDS:
        return False

    signed = parts["t"].encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

    return hmac.compare_digest(expected.encode(), received.encode())

import os

from flask import Flask, request

app = Flask(__name__)


@app.post("/zenit/webhook")
def zenit_webhook():
    # get_data liefert den Körper unverändert als bytes.
    valid = verify_zenit_signature(
        request.get_data(),
        request.headers.get("Zenit-Signature", ""),
        os.environ["ZENIT_WEBHOOK_SECRET"],
    )
    if not valid:
        return "", 400

    event = request.get_json()
    # Zuerst prüfen, ob event["id"] schon verarbeitet ist, dann verarbeiten.
    return "", 204