Webhooks & signatures

import crypto from "node:crypto";

export function verify2faSignature(rawBody, header, secret) {
  const parts = Object.fromEntries(
    (header ?? "").split(",").map((p) => p.trim().split("=")),
  );
  const t = Number(parts.t);
  if (!t || !parts.v1) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - t) > 300) return false;

  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${t}.`)
    .update(rawBody) // Buffer or string, exactly as received
    .digest("hex");

  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(parts.v1, "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// Express: use express.raw() so req.body is the raw Buffer.
// app.post("/webhooks/2fa", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verify2faSignature(req.body, req.get("X-2FA-Signature"), process.env.TWOFA_WEBHOOK_SECRET)) {
//     return res.sendStatus(400);
//   }
//   const event = JSON.parse(req.body);
//   res.sendStatus(200);
// });

package twofa

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

func VerifySignature(rawBody []byte, header, secret string) bool {
	var t int64
	var v1 string
	for _, part := range strings.Split(header, ",") {
		kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch kv[0] {
		case "t":
			t, _ = strconv.ParseInt(kv[1], 10, 64)
		case "v1":
			v1 = kv[1]
		}
	}
	if t == 0 || v1 == "" {
		return false
	}
	if math.Abs(float64(time.Now().Unix()-t)) > 300 {
		return false
	}

	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(strconv.FormatInt(t, 10) + "."))
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))

	return hmac.Equal([]byte(expected), []byte(v1))
}

// In your handler: body, _ := io.ReadAll(r.Body)
// if !VerifySignature(body, r.Header.Get("X-2FA-Signature"), secret) { w.WriteHeader(400); return }

<?php

function verify_2fa_signature(string $rawBody, string $header, string $secret): bool
{
    $parts = [];
    foreach (explode(',', $header) as $part) {
        $kv = explode('=', trim($part), 2);
        if (count($kv) === 2) {
            $parts[$kv[0]] = $kv[1];
        }
    }
    $t = isset($parts['t']) ? (int) $parts['t'] : 0;
    $v1 = $parts['v1'] ?? '';
    if ($t === 0 || $v1 === '') {
        return false;
    }
    if (abs(time() - $t) > 300) {
        return false;
    }

    $expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);
    return hash_equals($expected, $v1);
}

$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_2FA_SIGNATURE'] ?? '';

if (!verify_2fa_signature($rawBody, $header, getenv('TWOFA_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}

$event = json_decode($rawBody, true);
http_response_code(200);

import hashlib
import hmac
import time


def verify_2fa_signature(raw_body: bytes, header: str, secret: str) -> bool:
    parts = {}
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if value:
            parts[key] = value
    try:
        t = int(parts.get("t", "0"))
    except ValueError:
        return False
    v1 = parts.get("v1", "")
    if not t or not v1:
        return False
    if abs(int(time.time()) - t) > 300:
        return False

    expected = hmac.new(
        secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, v1)


# Flask:
# @app.post("/webhooks/2fa")
# def webhook():
#     if not verify_2fa_signature(request.get_data(), request.headers.get("X-2FA-Signature", ""), SECRET):
#         abort(400)
#     event = request.get_json()
#     return "", 200