Webhook ба signature

import crypto from "node:crypto";

// rawBody: Buffer эсвэл string, header: X-2FA-Signature
export function verifySignature(rawBody, header, secret, toleranceSec = 300) {
  const parts = Object.fromEntries(
    (header ?? "").split(",").map((kv) => kv.trim().split("="))
  );
  const t = Number(parts.t);
  if (!Number.isInteger(t) || !parts.v1) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - t) > toleranceSec) return false;

  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${t}.`)
    .update(rawBody)
    .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 жишээ
// app.post("/webhooks/2fa", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verifySignature(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 webhook

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

func VerifySignature(rawBody []byte, header, secret string, tolerance time.Duration) bool {
	var ts, sig string
	for _, part := range strings.Split(header, ",") {
		k, v, ok := strings.Cut(strings.TrimSpace(part), "=")
		if !ok {
			continue
		}
		switch k {
		case "t":
			ts = v
		case "v1":
			sig = v
		}
	}
	t, err := strconv.ParseInt(ts, 10, 64)
	if err != nil || sig == "" {
		return false
	}
	if math.Abs(float64(time.Now().Unix()-t)) > tolerance.Seconds() {
		return false
	}

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

	// hmac.Equal нь тогтмол хугацаанд харьцуулна
	return hmac.Equal([]byte(expected), []byte(sig))
}

func Handler(secret string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			http.Error(w, "bad body", http.StatusBadRequest)
			return
		}
		if !VerifySignature(body, r.Header.Get("X-2FA-Signature"), secret, 5*time.Minute) {
			http.Error(w, "invalid signature", http.StatusBadRequest)
			return
		}
		// json.Unmarshal(body, &event) ... event.ID-аар давхардлыг шалгана
		w.WriteHeader(http.StatusOK)
	}
}

<?php

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

    $expected = hash_hmac('sha256', $parts['t'] . '.' . $rawBody, $secret);

    // hash_equals нь тогтмол хугацаанд харьцуулна
    return hash_equals($expected, $parts['v1']);
}

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

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

$event = json_decode($rawBody, true);
// $event['id']-аар давхардлыг шалгана
http_response_code(200);

import hashlib
import hmac
import os
import time

from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["TWOFA_WEBHOOK_SECRET"]


def verify_signature(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    parts = dict(
        kv.strip().split("=", 1) for kv in (header or "").split(",") if "=" in kv
    )
    ts, sig = parts.get("t"), parts.get("v1")
    if not ts or not sig or not ts.isdigit():
        return False
    if abs(int(time.time()) - int(ts)) > tolerance:
        return False

    expected = hmac.new(
        secret.encode(), ts.encode() + b"." + raw_body, hashlib.sha256
    ).hexdigest()

    # compare_digest нь тогтмол хугацаанд харьцуулна
    return hmac.compare_digest(expected, sig)


@app.post("/webhooks/2fa")
def webhook():
    raw = request.get_data()
    if not verify_signature(raw, request.headers.get("X-2FA-Signature", ""), SECRET):
        abort(400)
    event = request.get_json()
    # event["id"]-аар давхардлыг шалгана
    return "", 200