Webhook 签名验证

signature = "sha256=" + HMAC-SHA256(timestamp + "." + payload, webhook_secret)

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, timestamp, secret) {
  // 检查时间戳容差（5 分钟）
  const currentTime = Date.now();
  if (Math.abs(currentTime - parseInt(timestamp)) > 300000) {
    return false;
  }

  // 计算期望签名
  const data = `${timestamp}.${payload}`;
  const expected = 'sha256=' + crypto
    .createHmac('sha256', secret)
    .update(data)
    .digest('hex');

  // 常量时间比较
  try {
    return crypto.timingSafeEqual(
      Buffer.from(expected),
      Buffer.from(signature)
    );
  } catch {
    return false;
  }
}

// 在 Express 中使用
app.post('/webhooks/kyren',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const isValid = verifyWebhookSignature(
      req.body.toString(),
      req.headers['x-kyren-signature'],
      req.headers['x-kyren-timestamp'],
      process.env.KYREN_WEBHOOK_SECRET
    );

    if (!isValid) {
      return res.status(400).send('Invalid signature');
    }

    // 处理事件...
    res.status(200).send('OK');
  }
);

import hmac
import hashlib
import time

def verify_webhook_signature(payload: str, signature: str, timestamp: str, secret: str) -> bool:
    # 检查时间戳容差（5 分钟）
    current_time = int(time.time() * 1000)
    if abs(current_time - int(timestamp)) > 300000:
        return False

    # 计算期望签名
    data = f"{timestamp}.{payload}"
    expected = "sha256=" + hmac.new(
        secret.encode("utf-8"),
        data.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

    # 常量时间比较
    return hmac.compare_digest(expected, signature)


# 在 Flask 中使用
from flask import Flask, request

app = Flask(__name__)

@app.route("/webhooks/kyren", methods=["POST"])
def webhook():
    payload = request.get_data(as_text=True)
    signature = request.headers.get("X-Kyren-Signature", "")
    timestamp = request.headers.get("X-Kyren-Timestamp", "")

    if not verify_webhook_signature(payload, signature, timestamp, WEBHOOK_SECRET):
        return "Invalid signature", 400

    event = request.get_json()
    # 处理事件...

    return "OK", 200

package main

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

func verifyWebhookSignature(payload, signature, timestamp, secret string) bool {
	// 检查时间戳容差（5 分钟）
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false
	}
	currentTime := time.Now().UnixMilli()
	if math.Abs(float64(currentTime-ts)) > 300000 {
		return false
	}

	// 计算期望签名
	data := fmt.Sprintf("%s.%s", timestamp, payload)
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(data))
	expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))

	// 常量时间比较
	return hmac.Equal([]byte(expected), []byte(signature))
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
	body, _ := io.ReadAll(r.Body)
	payload := string(body)
	signature := r.Header.Get("X-Kyren-Signature")
	timestamp := r.Header.Get("X-Kyren-Timestamp")

	if !verifyWebhookSignature(payload, signature, timestamp, webhookSecret) {
		http.Error(w, "Invalid signature", http.StatusBadRequest)
		return
	}

	// 处理事件...
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("OK"))
}