Webhooks

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

// header: "t=1792578565,v1=5257a8...", with a second v1 for a day after a
// secret rotation. Any one v1 that matches will do.
export const verify = (body: string, header: string, secret: string) => {
	const parts = header.split(',').map(part => part.trim().split('='))
	const t = Number(parts.find(([name]) => name === 't')?.[1])
	const signatures = parts.filter(([name]) => name === 'v1').map(([, value]) => value ?? '')

	if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > 300) {
		return false
	}

	const expected = Buffer.from(createHmac('sha256', secret).update(`${t}.${body}`).digest('hex'))

	return signatures.some(signature => {
		const given = Buffer.from(signature)
		return given.length === expected.length && timingSafeEqual(given, expected)
	})
}

import { verifyWebhook } from '@zeno-id/server'

// Takes several v1 and checks the time; true or false.
const valid = verifyWebhook(rawBody, request.headers.get('fingerprint-signature'), process.env.FP_WEBHOOK_SECRET)

import hashlib
import hmac
import time

def verify(body: bytes, header: str, secret: str) -> bool:
    """header: "t=1792578565,v1=5257a8...", maybe with a second v1."""
    parts = [part.strip().partition("=") for part in header.split(",")]
    timestamps = [value for name, _, value in parts if name == "t"]
    signatures = [value for name, _, value in parts if name == "v1"]
    if not timestamps or not timestamps[0].isdigit():
        return False
    t = timestamps[0]
    if abs(time.time() - int(t)) > 300:
        return False
    # The secret as it's shown, whsec_ included, in UTF-8: no base64 decoding.
    expected = hmac.new(secret.encode(), f"{t}.".encode() + body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)

function verifyWebhook(string $rawBody, string $header, array $secrets): bool
{
    $t = null;
    $signatures = [];

    // Not parse_str or a name => value map: both v1 entries share a name,
    // and a map keeps only one of them.
    foreach (explode(',', $header) as $part) {
        [$name, $value] = array_pad(explode('=', trim($part), 2), 2, '');

        if ($name === 't' && $t === null) {
            $t = $value;
        } elseif ($name === 'v1') {
            $signatures[] = $value;
        }
    }

    if ($t === null || !ctype_digit($t) || abs(time() - (int) $t) > 300) {
        return false;
    }

    foreach ($secrets as $secret) {
        // The secret as it's shown, whsec_ included: no base64 decoding.
        $expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);

        foreach ($signatures as $signature) {
            if (hash_equals($expected, $signature)) {
                return true;
            }
        }
    }

    return false;
}

// Verify checks a Fingerprint-Signature header against the raw body: t within
// 5 minutes, and any v1 matching any of the secrets.
func Verify(body []byte, header string, secrets ...string) bool {
	var t string
	var signatures [][]byte

	for _, part := range strings.Split(header, ",") {
		name, value, _ := strings.Cut(strings.TrimSpace(part), "=")
		switch name {
		case "t":
			if t == "" {
				t = value
			}
		case "v1":
			if signature, err := hex.DecodeString(value); err == nil {
				signatures = append(signatures, signature)
			}
		}
	}

	seconds, err := strconv.ParseInt(t, 10, 64)
	if err != nil || time.Since(time.Unix(seconds, 0)).Abs() > 5*time.Minute {
		return false
	}

	for _, secret := range secrets {
		mac := hmac.New(sha256.New, []byte(secret)) // whsec_... as it's shown
		mac.Write([]byte(t + "."))
		mac.Write(body)
		expected := mac.Sum(nil)

		for _, signature := range signatures {
			if hmac.Equal(signature, expected) {
				return true
			}
		}
	}

	return false
}

// Express: the raw body as a string, for this route only.
app.post('/webhooks/fingerprint', express.text({ type: 'application/json' }), (request, response) => {
	if (!verify(request.body, request.get('fingerprint-signature') ?? '', process.env.FP_WEBHOOK_SECRET!)) {
		return response.sendStatus(400)
	}

	const event = JSON.parse(request.body)
	response.sendStatus(204)
})

// Next.js App Router (app/webhooks/fingerprint/route.ts), Hono, Bun or any
// fetch-style handler: read the text before parsing.
export const POST = async (request: Request) => {
	const body = await request.text()

	if (!verify(body, request.headers.get('fingerprint-signature') ?? '', process.env.FP_WEBHOOK_SECRET!)) {
		return new Response(null, { status: 400 })
	}

	const event = JSON.parse(body)
	return new Response(null, { status: 204 })
}

# Flask
@app.post("/webhooks/fingerprint")
def fingerprint_webhook():
    body = request.get_data()  # the raw bytes, before request.json
    if not verify(body, request.headers.get("Fingerprint-Signature", ""), os.environ["FP_WEBHOOK_SECRET"]):
        abort(400)
    event = json.loads(body)
    return "", 204

# Django
@csrf_exempt
@require_POST
def fingerprint_webhook(request):
    body = request.body  # the raw bytes
    if not verify(body, request.headers.get("Fingerprint-Signature", ""), settings.FP_WEBHOOK_SECRET):
        return HttpResponseBadRequest()
    event = json.loads(body)
    return HttpResponse(status=204)

# FastAPI or Starlette: take the Request, not a pydantic body.
@app.post("/webhooks/fingerprint", status_code=204)
async def fingerprint_webhook(request: Request):
    body = await request.body()  # the raw bytes
    if not verify(body, request.headers.get("fingerprint-signature", ""), os.environ["FP_WEBHOOK_SECRET"]):
        raise HTTPException(status_code=400)
    event = json.loads(body)

// routes/web.php
Route::post('/webhooks/fingerprint', function (Request $request) {
    // getContent() is the raw body; $request->all() or ->json() is parsed,
    // and encoding it again won't give the same bytes.
    $raw = $request->getContent();
    $secrets = [config('services.fingerprint.webhook_secret')];

    if (!verifyWebhook($raw, (string) $request->header('Fingerprint-Signature'), $secrets)) {
        abort(400);
    }

    $event = json_decode($raw, true, flags: JSON_THROW_ON_ERROR);

    return response()->noContent();
});

// bootstrap/app.php (Laravel 11 and up): the web routes' CSRF check answers
// 419 to every delivery, so leave this route out of it, or put the route in
// routes/api.php, which has none.
->withMiddleware(function (Middleware $middleware) {
    $middleware->validateCsrfTokens(except: ['webhooks/fingerprint']);
})

// Plain PHP: the raw bytes, before anything parses them.
// In Symfony it's $request->getContent().
$raw = file_get_contents('php://input');
$header = $_SERVER['HTTP_FINGERPRINT_SIGNATURE'] ?? '';

if (!verifyWebhook($raw, $header, [getenv('FP_WEBHOOK_SECRET')])) {
    http_response_code(400);
    exit;
}

$event = json_decode($raw, true, flags: JSON_THROW_ON_ERROR);
http_response_code(204);

http.HandleFunc("POST /webhooks/fingerprint", func(w http.ResponseWriter, r *http.Request) {
	// The raw bytes, before any JSON decoding.
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
	if err != nil || !Verify(body, r.Header.Get("Fingerprint-Signature"), os.Getenv("FP_WEBHOOK_SECRET")) {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	var event map[string]any
	if err := json.Unmarshal(body, &event); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	w.WriteHeader(http.StatusNoContent)
})