Vérifier les signatures

import crypto from 'node:crypto'
import express from 'express'
 
const app = express()
const SECRET = process.env.SKOUP_WEBHOOK_SECRET // whsec_…
const TOLERANCE_SECONDS = 300
 
function verifySkoupSignature(rawBody, header, secret) {
  const parts = header.split(',').map((part) => part.split('=', 2))
  const timestamp = parts.find(([key]) => key === 't')?.[1]
  const signatures = parts.filter(([key]) => key === 'v1').map(([, value]) => value)
 
  if (!timestamp || signatures.length === 0) return false
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) return false
 
  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${rawBody}`)
    .digest()
 
  return signatures.some((signature) => {
    const received = Buffer.from(signature, 'hex')
    return received.length === expected.length && crypto.timingSafeEqual(received, expected)
  })
}
 
// express.raw keeps the body as a Buffer: never use express.json() on this route.
app.post('/skoup', express.raw({ type: 'application/json' }), (req, res) => {
  const header = req.get('X-Skoup-Signature') ?? ''
 
  if (!verifySkoupSignature(req.body.toString('utf8'), header, SECRET)) {
    return res.status(400).send('Invalid signature')
  }
 
  const event = JSON.parse(req.body.toString('utf8'))
  queue.push(event) // process asynchronously
  res.sendStatus(200)
})

<?php
 
$secret = getenv('SKOUP_WEBHOOK_SECRET'); // whsec_…
$tolerance = 300;
 
$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_SKOUP_SIGNATURE'] ?? '';
 
$timestamp = null;
$signatures = [];
 
foreach (explode(',', $header) as $part) {
    [$key, $value] = array_pad(explode('=', $part, 2), 2, null);
 
    if ($key === 't') {
        $timestamp = $value;
    } elseif ($key === 'v1') {
        $signatures[] = $value;
    }
}
 
if ($timestamp === null || abs(time() - (int) $timestamp) > $tolerance) {
    http_response_code(400);
    exit('Invalid timestamp');
}
 
$expected = hash_hmac('sha256', $timestamp.'.'.$rawBody, $secret);
 
$valid = false;
foreach ($signatures as $signature) {
    if (hash_equals($expected, $signature)) {
        $valid = true;
        break;
    }
}
 
if (! $valid) {
    http_response_code(400);
    exit('Invalid signature');
}
 
$event = json_decode($rawBody, true);
// Queue $event for processing, then answer fast.
http_response_code(200);

import hashlib
import hmac
import os
import time
 
from flask import Flask, abort, request
 
app = Flask(__name__)
SECRET = os.environ["SKOUP_WEBHOOK_SECRET"].encode()  # whsec_…
TOLERANCE_SECONDS = 300
 
 
def verify_skoup_signature(raw_body: bytes, header: str) -> bool:
    timestamp = None
    signatures = []
    for part in header.split(","):
        key, _, value = part.partition("=")
        if key == "t":
            timestamp = value
        elif key == "v1":
            signatures.append(value)
 
    if timestamp is None or abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False
 
    expected = hmac.new(SECRET, timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)
 
 
@app.post("/skoup")
def skoup_webhook():
    raw_body = request.get_data()  # raw bytes, before any JSON parsing
    if not verify_skoup_signature(raw_body, request.headers.get("X-Skoup-Signature", "")):
        abort(400)
 
    event = request.get_json()
    enqueue(event)  # process asynchronously
    return "", 200

package main
 
import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"io"
	"math"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)
 
const tolerance = 5 * time.Minute
 
func verifySkoupSignature(rawBody []byte, header, secret string) bool {
	var timestamp string
	var signatures []string
 
	for _, part := range strings.Split(header, ",") {
		key, value, _ := strings.Cut(part, "=")
		switch key {
		case "t":
			timestamp = value
		case "v1":
			signatures = append(signatures, value)
		}
	}
 
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil || math.Abs(time.Since(time.Unix(ts, 0)).Seconds()) > tolerance.Seconds() {
		return false
	}
 
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(timestamp + "."))
	mac.Write(rawBody)
	expected := mac.Sum(nil)
 
	for _, signature := range signatures {
		received, err := hex.DecodeString(signature)
		if err == nil && hmac.Equal(received, expected) {
			return true
		}
	}
 
	return false
}
 
func skoupWebhook(w http.ResponseWriter, r *http.Request) {
	rawBody, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "cannot read body", http.StatusBadRequest)
		return
	}
 
	if !verifySkoupSignature(rawBody, r.Header.Get("X-Skoup-Signature"), os.Getenv("SKOUP_WEBHOOK_SECRET")) {
		http.Error(w, "invalid signature", http.StatusBadRequest)
		return
	}
 
	go process(rawBody) // process asynchronously
	w.WriteHeader(http.StatusOK)
}