签名验证

<?php

function verifyWebhookSignature(
    string $webhookPath,
    string $rawBody,
    string $timestamp,
    string $receivedSignature,
    string $publicKeyPath
): bool {
    // 第一步：对原始请求体进行哈希
    $payloadHash = strtolower(bin2hex(hash('sha256', $rawBody, true)));

    // 第二步：构建待验证字符串
    $stringToVerify = "POST:{$webhookPath}:{$payloadHash}:{$timestamp}";

    // 第三步：加载网关的公钥
    $publicKey = openssl_get_publickey('file://' . $publicKeyPath);
    if (!$publicKey) {
        throw new RuntimeException('无法加载公钥');
    }

    // 第四步：验证签名
    $result = openssl_verify(
        $stringToVerify,
        base64_decode($receivedSignature),
        $publicKey,
        OPENSSL_ALGO_SHA256
    );

    return $result === 1;
}

// 在您的 Webhook 处理程序中使用：
$rawBody = file_get_contents('php://input');
$signature = $_SERVER['HTTP_X_SIGNATURE'] ?? '';
$timestamp = $_SERVER['HTTP_X_TIMESTAMP'] ?? '';
$partnerId = $_SERVER['HTTP_X_PARTNER_ID'] ?? '';

// 验证 Partner ID 是否匹配您的客户端密钥
if ($partnerId !== 'your-client-key') {
    http_response_code(401);
    exit('无效的 Partner ID');
}

// 验证签名
$isValid = verifyWebhookSignature(
    '/api/webhooks/payment',   // 您的 Webhook 端点路径
    $rawBody,
    $timestamp,
    $signature,
    '/path/to/gateway-public-key.pem'
);

if (!$isValid) {
    http_response_code(401);
    exit('无效的签名');
}

// 签名验证通过 — 处理 Webhook
$payload = json_decode($rawBody, true);
$event = $payload['event'];
$data = $payload['data'];

// 处理事件...
http_response_code(200);
echo 'OK';

const crypto = require('crypto');
const fs = require('fs');

function verifyWebhookSignature(webhookPath, rawBody, timestamp, receivedSignature, publicKeyPath) {
  // 第一步：对原始请求体进行哈希
  const payloadHash = crypto.createHash('sha256').update(rawBody).digest('hex').toLowerCase();

  // 第二步：构建待验证字符串
  const stringToVerify = `POST:${webhookPath}:${payloadHash}:${timestamp}`;

  // 第三步：加载网关的公钥
  const publicKey = fs.readFileSync(publicKeyPath);

  // 第四步：验证签名
  const verifier = crypto.createVerify('RSA-SHA256');
  verifier.update(stringToVerify);
  return verifier.verify(publicKey, receivedSignature, 'base64');
}

// Express.js 示例：
const express = require('express');
const app = express();

app.post('/api/webhooks/payment', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-signature'];
  const timestamp = req.headers['x-timestamp'];
  const partnerId = req.headers['x-partner-id'];

  // 验证 Partner ID
  if (partnerId !== 'your-client-key') {
    return res.status(401).send('无效的 Partner ID');
  }

  // 验证签名
  const isValid = verifyWebhookSignature(
    '/api/webhooks/payment',
    req.body,              // 使用 express.raw() 时为原始 Buffer
    timestamp,
    signature,
    './gateway-public-key.pem'
  );

  if (!isValid) {
    return res.status(401).send('无效的签名');
  }

  // 签名验证通过 — 处理 Webhook
  const payload = JSON.parse(req.body);
  console.log('事件:', payload.event);
  console.log('数据:', payload.data);

  res.status(200).send('OK');
});

import hashlib
import base64
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

def verify_webhook_signature(
    webhook_path: str,
    raw_body: bytes,
    timestamp: str,
    received_signature: str,
    public_key_path: str
) -> bool:
    # 第一步：对原始请求体进行哈希
    payload_hash = hashlib.sha256(raw_body).hexdigest().lower()

    # 第二步：构建待验证字符串
    string_to_verify = f"POST:{webhook_path}:{payload_hash}:{timestamp}"

    # 第三步：加载网关的公钥
    with open(public_key_path, "rb") as key_file:
        public_key = serialization.load_pem_public_key(key_file.read())

    # 第四步：验证签名
    try:
        public_key.verify(
            base64.b64decode(received_signature),
            string_to_verify.encode(),
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False

# Flask 示例：
from flask import Flask, request

app = Flask(__name__)

@app.route('/api/webhooks/payment', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-SIGNATURE', '')
    timestamp = request.headers.get('X-TIMESTAMP', '')
    partner_id = request.headers.get('X-PARTNER-ID', '')

    if partner_id != 'your-client-key':
        return '无效的 Partner ID', 401

    is_valid = verify_webhook_signature(
        '/api/webhooks/payment',
        request.get_data(),
        timestamp,
        signature,
        './gateway-public-key.pem'
    )

    if not is_valid:
        return '无效的签名', 401

    payload = request.get_json()
    print(f"事件: {payload['event']}")

    return 'OK', 200