Verify signatures

// Node.js. With Express, mount the route with express.raw({ type: "application/json" })
// so req.body is the raw Buffer, not parsed JSON.
import crypto from "node:crypto";

export function verifyFluxSignature(secret, header, rawBody, toleranceSeconds = 300) {
  if (!header) return false;
  let timestamp = null;
  const signatures = [];
  for (const part of header.split(",")) {
    const i = part.indexOf("=");
    if (i < 0) continue;
    const key = part.slice(0, i).trim();
    const value = part.slice(i + 1).trim();
    if (key === "t" && /^\d+$/.test(value)) timestamp = Number(value);
    else if (key === "v1") signatures.push(value);
  }
  if (timestamp === null || signatures.length === 0) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;

  const expected = Buffer.from(
    crypto.createHmac("sha256", secret).update(`${timestamp}.`).update(rawBody).digest("hex"),
  );
  return signatures.some((s) => {
    const given = Buffer.from(s);
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
}

// app.post("/flux", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verifyFluxSignature(process.env.FLUX_WEBHOOK_SECRET, req.get("Flux-Signature"), req.body)) {
//     return res.sendStatus(400);
//   }
//   const event = JSON.parse(req.body);
//   ...
// });

# Python. Pass the raw body bytes: request.get_data() in Flask, request.body in Django.
import hashlib
import hmac
import time


def verify_flux_signature(secret: str, header: str | None, raw_body: bytes, tolerance: int = 300) -> bool:
    if not header:
        return False
    timestamp, signatures = None, []
    for part in header.split(","):
        key, sep, value = part.partition("=")
        if not sep:
            continue
        key, value = key.strip(), value.strip()
        if key == "t" and value.isdigit():
            timestamp = int(value)
        elif key == "v1":
            signatures.append(value)
    if timestamp is None or not signatures:
        return False
    if abs(time.time() - timestamp) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in signatures)

<?php
// PHP. Raw body: file_get_contents('php://input') ($request->getContent() in Laravel).
// Header: $_SERVER['HTTP_FLUX_SIGNATURE'] ($request->header('Flux-Signature') in Laravel).
function verifyFluxSignature(string $secret, ?string $header, string $rawBody, int $tolerance = 300): bool
{
    if (!$header) {
        return false;
    }
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', $part, 2);
        if (count($pair) !== 2) {
            continue;
        }
        [$key, $value] = array_map('trim', $pair);
        if ($key === 't' && ctype_digit($value)) {
            $timestamp = (int) $value;
        } elseif ($key === 'v1') {
            $signatures[] = $value;
        }
    }
    if ($timestamp === null || !$signatures || abs(time() - $timestamp) > $tolerance) {
        return false;
    }
    $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) {
            return true;
        }
    }
    return false;
}

// C#. In ASP.NET Core, read the raw body before anything parses it:
//   using var reader = new StreamReader(Request.Body, Encoding.UTF8);
//   var raw = await reader.ReadToEndAsync();
//   if (!FluxWebhook.Verify(secret, Request.Headers["Flux-Signature"], raw)) return BadRequest();
using System.Security.Cryptography;
using System.Text;

public static class FluxWebhook
{
    public static bool Verify(string secret, string? header, string rawBody, int toleranceSeconds = 300)
    {
        if (string.IsNullOrWhiteSpace(header)) return false;
        long? timestamp = null;
        var signatures = new List<string>();
        foreach (var part in header.Split(','))
        {
            var kv = part.Split('=', 2);
            if (kv.Length != 2) continue;
            var key = kv[0].Trim();
            var value = kv[1].Trim();
            if (key == "t" && long.TryParse(value, out var t)) timestamp = t;
            else if (key == "v1") signatures.Add(value);
        }
        if (timestamp is null || signatures.Count == 0) return false;
        if (Math.Abs(DateTimeOffset.UtcNow.ToUnixTimeSeconds() - timestamp.Value) > toleranceSeconds) return false;

        using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(secret));
        var hash = hmac.ComputeHash(Encoding.UTF8.GetBytes($"{timestamp}.{rawBody}"));
        var expected = Encoding.ASCII.GetBytes(Convert.ToHexString(hash).ToLowerInvariant());
        return signatures.Any(s => CryptographicOperations.FixedTimeEquals(expected, Encoding.ASCII.GetBytes(s)));
    }
}

// Go. Read the raw body with io.ReadAll(r.Body) before decoding it, then:
//   ok := fluxwebhook.Verify(secret, r.Header.Get("Flux-Signature"), body, 5*time.Minute)
package fluxwebhook

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"strconv"
	"strings"
	"time"
)

func Verify(secret, header string, rawBody []byte, tolerance time.Duration) bool {
	timestamp := int64(-1)
	var signatures []string
	for _, part := range strings.Split(header, ",") {
		key, value, ok := strings.Cut(part, "=")
		if !ok {
			continue
		}
		key, value = strings.TrimSpace(key), strings.TrimSpace(value)
		switch key {
		case "t":
			if t, err := strconv.ParseInt(value, 10, 64); err == nil {
				timestamp = t
			}
		case "v1":
			signatures = append(signatures, value)
		}
	}
	if timestamp < 0 || len(signatures) == 0 {
		return false
	}
	age := time.Since(time.Unix(timestamp, 0))
	if age > tolerance || age < -tolerance {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(strconv.FormatInt(timestamp, 10) + "."))
	mac.Write(rawBody)
	expected := []byte(hex.EncodeToString(mac.Sum(nil)))
	for _, s := range signatures {
		if hmac.Equal(expected, []byte(s)) {
			return true
		}
	}
	return false
}