Guide des webhooks

"""Vérifier la signature d'un webhook Tauraco (Standard Webhooks) et le recevoir.

Verify a Tauraco webhook signature (Standard Webhooks) and receive it.

Bibliothèque standard seule (Python 3.9+). Le contenu signé est
``"{webhook-id}.{webhook-timestamp}." + corps brut`` ; la signature vaut
``"v1," + base64(HMAC-SHA256(clé, contenu))``, la clé étant le secret ``whsec_…``
décodé de base64 (préfixe retiré). L'en-tête ``webhook-signature`` peut porter
plusieurs signatures séparées par des espaces. Un secret vide ou illisible est refusé :
une clé HMAC vide permettrait à n'importe qui de signer.

Vecteur de référence (identique dans la page « Guide des webhooks ») :
    secret            whsec_dGF1cmFjby1leGVtcGxlLWRlLXZlY3RldXItMjAyNiE=
    webhook-id        evt_exemple_0001
    webhook-timestamp 1793750400
    corps             {"id":"evt_exemple_0001","type":"webhook.test","data":{"message":"ok"}}
    signature         v1,z25573sJuRUxUk3JkjiZN3OW2KihQFtLvIGhZTHUnuM=

Lancer le récepteur de démonstration (port 8787) :
    TAURACO_WEBHOOK_SECRET='whsec_…' python3 verifier_webhook.py
"""
from __future__ import annotations

import base64
import binascii
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Mapping, Optional

TOLERANCE_SECONDS = 300


def secret_key(secret: str) -> Optional[bytes]:
    """Clé HMAC du secret ``whsec_…`` (base64, préfixe retiré) ; ``None`` si vide ou illisible."""
    if not isinstance(secret, str):
        return None
    try:
        key = base64.b64decode(secret.removeprefix("whsec_").encode("ascii"), validate=True)
    except (binascii.Error, UnicodeEncodeError, ValueError):
        return None
    return key or None  # clé vide : refusée


def verify_signature(secret: str, headers: Mapping[str, str], body: bytes, *,
                     tolerance: int = TOLERANCE_SECONDS, now: Optional[int] = None) -> bool:
    """Vrai si une signature ``v1`` de ``webhook-signature`` correspond et que l'horodatage est récent.

    ``body`` doit être le corps BRUT reçu (octets), jamais un JSON re-sérialisé.
    """
    lowered = {str(name).lower(): str(value) for name, value in headers.items()}
    msg_id = lowered.get("webhook-id")
    raw_timestamp = lowered.get("webhook-timestamp")
    raw_signatures = lowered.get("webhook-signature")
    if not msg_id or not raw_timestamp or not raw_signatures:
        return False
    try:
        timestamp = int(raw_timestamp.strip())
    except ValueError:
        return False
    current = int(time.time()) if now is None else now
    if abs(current - timestamp) > tolerance:
        return False  # rejeu d'un ancien message ou horloge déréglée
    key = secret_key(secret)
    if key is None:
        return False
    signed = f"{msg_id}.{timestamp}.".encode() + body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode("ascii")
    for candidate in raw_signatures.split():
        version, _, value = candidate.partition(",")
        if version == "v1" and hmac.compare_digest(value, expected):
            return True
    return False


class AlreadySeen:
    """Dédoublonnage par ``webhook-id`` : la livraison est « au moins une fois ».

    En production, conserver les identifiants dans une base (contrainte d'unicité),
    pas en mémoire : un rejeu ou une reprise renvoie le même ``webhook-id``.
    """

    def __init__(self) -> None:
        self._ids: set[str] = set()

    def first_time(self, msg_id: str) -> bool:
        if msg_id in self._ids:
            return False
        self._ids.add(msg_id)
        return True


def handle_event(event: dict) -> None:
    """Traitement métier, idempotent (à remplacer)."""
    print(f"{event.get('type')} {event.get('id')} tenant={event.get('tenant_id')}")


def make_handler(secret: str, seen: AlreadySeen) -> type[BaseHTTPRequestHandler]:
    class Receiver(BaseHTTPRequestHandler):
        def do_POST(self) -> None:  # noqa: N802 (nom imposé par http.server)
            length = int(self.headers.get("Content-Length") or 0)
            body = self.rfile.read(length)
            headers = {name: value for name, value in self.headers.items()}
            if not verify_signature(secret, headers, body):
                self.send_response(401)
                self.end_headers()
                return
            # Répondre vite (2xx) ; traiter ensuite. Un doublon reçoit aussi 2xx.
            if seen.first_time(headers.get("webhook-id") or headers.get("Webhook-Id", "")):
                handle_event(json.loads(body))
            self.send_response(204)
            self.end_headers()

        def log_message(self, *_args) -> None:  # pas de corps ni d'en-tête dans les journaux
            return

    return Receiver


if __name__ == "__main__":
    webhook_secret = os.environ["TAURACO_WEBHOOK_SECRET"]  # aucune valeur par défaut
    if secret_key(webhook_secret) is None:
        raise SystemExit("TAURACO_WEBHOOK_SECRET vide ou illisible (whsec_ + base64 attendu)")
    HTTPServer(("127.0.0.1", 8787), make_handler(webhook_secret, AlreadySeen())).serve_forever()

/**
 * Vérifier la signature d'un webhook Tauraco (Standard Webhooks) — Node.js 18+.
 * Verify a Tauraco webhook signature (Standard Webhooks) — Node.js 18+.
 *
 * Contenu signé : `${webhook-id}.${webhook-timestamp}.` + corps BRUT (octets).
 * Signature : "v1," + base64(HMAC-SHA256(base64decode(secret sans "whsec_"), contenu)).
 * Un secret vide ou qui n'est pas du base64 strict est refusé (Buffer.from(…, "base64")
 * accepte n'importe quoi et rendrait une clé vide : n'importe qui pourrait signer).
 *
 * Vecteur de référence (page « Guide des webhooks ») :
 *   secret            whsec_dGF1cmFjby1leGVtcGxlLWRlLXZlY3RldXItMjAyNiE=
 *   webhook-id        evt_exemple_0001
 *   webhook-timestamp 1793750400
 *   corps             {"id":"evt_exemple_0001","type":"webhook.test","data":{"message":"ok"}}
 *   signature         v1,z25573sJuRUxUk3JkjiZN3OW2KihQFtLvIGhZTHUnuM=
 */
import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

const TOLERANCE_SECONDS = 300;
const STRICT_BASE64 = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/;

/** Clé HMAC du secret `whsec_…`, ou `null` si vide ou illisible. */
export function secretKey(secret: string): Buffer | null {
  const raw = secret.replace(/^whsec_/, "");
  if (raw.length === 0 || !STRICT_BASE64.test(raw)) return null;
  const key = Buffer.from(raw, "base64");
  return key.length > 0 ? key : null;
}

export function verifySignature(
  secret: string,
  headers: Record<string, string | string[] | undefined>,
  body: Buffer,
  now: number = Math.floor(Date.now() / 1000),
): boolean {
  const header = (name: string): string | undefined => {
    const value = headers[name] ?? headers[name.toLowerCase()];
    return Array.isArray(value) ? value[0] : value;
  };
  const id = header("webhook-id");
  const timestamp = Number(header("webhook-timestamp"));
  const signatures = header("webhook-signature");
  if (!id || !signatures || !Number.isInteger(timestamp)) return false;
  if (Math.abs(now - timestamp) > TOLERANCE_SECONDS) return false;
  const key = secretKey(secret);
  if (key === null) return false;
  const expected = Buffer.from(
    createHmac("sha256", key)
      .update(Buffer.concat([Buffer.from(`${id}.${timestamp}.`), body]))
      .digest("base64"),
  );
  return signatures.split(" ").some((candidate) => {
    const [version, value] = candidate.split(",", 2);
    if (version !== "v1" || !value) return false; // autre version ou candidate illisible : suivante
    const received = Buffer.from(value); // comparaison du texte base64, en temps constant
    return received.length === expected.length && timingSafeEqual(received, expected);
  });
}

// Récepteur de démonstration : TAURACO_WEBHOOK_SECRET=whsec_… node verifierWebhook.js
const seen = new Set<string>(); // en production : table avec contrainte d'unicité sur webhook-id
if (process.env.TAURACO_WEBHOOK_SECRET !== undefined) {
  const secret = process.env.TAURACO_WEBHOOK_SECRET;
  if (secretKey(secret) === null) throw new Error("TAURACO_WEBHOOK_SECRET vide ou illisible");
  createServer((req, res) => {
    const chunks: Buffer[] = [];
    req.on("data", (chunk: Buffer) => chunks.push(chunk));
    req.on("end", () => {
      const body = Buffer.concat(chunks);
      if (!verifySignature(secret, req.headers, body)) {
        res.writeHead(401).end();
        return;
      }
      res.writeHead(204).end(); // répondre vite ; un doublon reçoit aussi 2xx
      const id = String(req.headers["webhook-id"]);
      if (seen.has(id)) return;
      seen.add(id);
      const event = JSON.parse(body.toString("utf8"));
      console.log(event.type, event.id);
    });
  }).listen(8787, "127.0.0.1");
}

// Vérifier la signature d'un webhook Tauraco (Standard Webhooks) — Java 17+, sans dépendance.
// Verify a Tauraco webhook signature (Standard Webhooks) — Java 17+, no dependency.
// Vecteur de référence (page « Guide des webhooks ») :
//   secret            whsec_dGF1cmFjby1leGVtcGxlLWRlLXZlY3RldXItMjAyNiE=
//   webhook-id        evt_exemple_0001
//   webhook-timestamp 1793750400
//   corps             {"id":"evt_exemple_0001","type":"webhook.test","data":{"message":"ok"}}
//   signature         v1,z25573sJuRUxUk3JkjiZN3OW2KihQFtLvIGhZTHUnuM=
// Un secret vide ou illisible est refusé ; une signature candidate illisible est ignorée
// (on essaie la suivante : l'en-tête peut en porter plusieurs).
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Instant;
import java.util.Base64;
import java.util.Map;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

public final class VerifierWebhook {
    private static final long TOLERANCE_SECONDS = 300;

    private VerifierWebhook() {}

    /** {@code headers} : noms en minuscules ; {@code body} : corps BRUT reçu. */
    public static boolean verifySignature(String secret, Map<String, String> headers, byte[] body, long now) {
        String id = headers.get("webhook-id");
        String rawTimestamp = headers.get("webhook-timestamp");
        String signatures = headers.get("webhook-signature");
        if (id == null || rawTimestamp == null || signatures == null) return false;
        long timestamp;
        try {
            timestamp = Long.parseLong(rawTimestamp.trim());
        } catch (NumberFormatException e) {
            return false;
        }
        if (Math.abs(now - timestamp) > TOLERANCE_SECONDS) return false;
        byte[] key;
        try {
            key = Base64.getDecoder().decode(secret.startsWith("whsec_") ? secret.substring(6) : secret);
        } catch (IllegalArgumentException e) {
            return false; // secret illisible
        }
        if (key.length == 0) return false; // clé vide : n'importe qui pourrait signer
        byte[] expected;
        try {
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(new SecretKeySpec(key, "HmacSHA256"));
            mac.update((id + "." + timestamp + ".").getBytes(StandardCharsets.UTF_8));
            expected = mac.doFinal(body);
        } catch (java.security.GeneralSecurityException e) {
            return false;
        }
        for (String candidate : signatures.split(" ")) {
            String[] parts = candidate.split(",", 2);
            if (parts.length != 2 || !parts[0].equals("v1")) continue;
            byte[] received;
            try {
                received = Base64.getDecoder().decode(parts[1]);
            } catch (IllegalArgumentException e) {
                continue; // candidate illisible : essayer la suivante
            }
            if (MessageDigest.isEqual(received, expected)) return true;
        }
        return false;
    }

    public static boolean verifySignature(String secret, Map<String, String> headers, byte[] body) {
        return verifySignature(secret, headers, body, Instant.now().getEpochSecond());
    }
}

// Vérifier la signature d'un webhook Tauraco (Standard Webhooks) — .NET 6+, sans dépendance.
// Verify a Tauraco webhook signature (Standard Webhooks) — .NET 6+, no dependency.
// Vecteur de référence (page « Guide des webhooks ») :
//   secret            whsec_dGF1cmFjby1leGVtcGxlLWRlLXZlY3RldXItMjAyNiE=
//   webhook-id        evt_exemple_0001
//   webhook-timestamp 1793750400
//   corps             {"id":"evt_exemple_0001","type":"webhook.test","data":{"message":"ok"}}
//   signature         v1,z25573sJuRUxUk3JkjiZN3OW2KihQFtLvIGhZTHUnuM=
// Un secret vide ou illisible est refusé ; une signature candidate illisible est ignorée
// (on essaie la suivante : l'en-tête peut en porter plusieurs).
using System;
using System.Collections.Generic;
using System.Security.Cryptography;
using System.Text;

public static class VerifierWebhook
{
    private const long ToleranceSeconds = 300;

    /// <summary><paramref name="headers"/> : noms en minuscules ; <paramref name="body"/> : corps BRUT.</summary>
    public static bool VerifySignature(string secret, IReadOnlyDictionary<string, string> headers, byte[] body, long? now = null)
    {
        if (!headers.TryGetValue("webhook-id", out var id) ||
            !headers.TryGetValue("webhook-timestamp", out var rawTimestamp) ||
            !headers.TryGetValue("webhook-signature", out var signatures) ||
            !long.TryParse(rawTimestamp.Trim(), out var timestamp))
        {
            return false;
        }
        var current = now ?? DateTimeOffset.UtcNow.ToUnixTimeSeconds();
        if (Math.Abs(current - timestamp) > ToleranceSeconds) return false;
        byte[] key;
        try
        {
            key = Convert.FromBase64String(secret.StartsWith("whsec_") ? secret.Substring(6) : secret);
        }
        catch (FormatException)
        {
            return false; // secret illisible
        }
        if (key.Length == 0) return false; // clé vide : n'importe qui pourrait signer
        var prefix = Encoding.UTF8.GetBytes($"{id}.{timestamp}.");
        var signed = new byte[prefix.Length + body.Length];
        Buffer.BlockCopy(prefix, 0, signed, 0, prefix.Length);
        Buffer.BlockCopy(body, 0, signed, prefix.Length, body.Length);
        using var hmac = new HMACSHA256(key);
        var expected = hmac.ComputeHash(signed);
        foreach (var candidate in signatures.Split(' ', StringSplitOptions.RemoveEmptyEntries))
        {
            var parts = candidate.Split(',', 2);
            if (parts.Length != 2 || parts[0] != "v1") continue;
            byte[] received;
            try
            {
                received = Convert.FromBase64String(parts[1]);
            }
            catch (FormatException)
            {
                continue; // candidate illisible : essayer la suivante
            }
            if (CryptographicOperations.FixedTimeEquals(received, expected)) return true;
        }
        return false;
    }
}