Webhook 接收端示例

import crypto from 'node:crypto'
import express from 'express'

const app = express()
app.use(express.json({ limit: '1mb' }))

// 生产环境请使用数据库唯一键；这里只用于演示。
const processedEventIds = new Set()

app.post('/webhooks/qingzhoushan', async (req, res) => {
  const eventId = req.get('X-Webhook-Event-Id') ?? ''
  const timestamp = req.get('X-Webhook-Timestamp') ?? ''
  const signature = req.get('X-Webhook-Signature') ?? ''
  const appSecret = process.env.QINGZHOUSHAN_APP_SECRET ?? ''

  const timestampNumber = Number(timestamp)
  if (!Number.isInteger(timestampNumber) || Math.abs(Date.now() / 1000 - timestampNumber) > 300) {
    return res.status(401).json({ success: false, code: 401, message: 'timestamp expired' })
  }
  if (!eventId || eventId !== req.body?.event_id || !appSecret) {
    return res.status(401).json({ success: false, code: 401, message: 'invalid event' })
  }

  const expected = crypto
    .createHmac('sha256', appSecret)
    .update(`${eventId}\n${timestamp}`, 'utf8')
    .digest('hex')
  const expectedBuffer = Buffer.from(expected, 'hex')
  const providedBuffer = Buffer.from(signature, 'hex')
  const valid = providedBuffer.length === expectedBuffer.length
    && crypto.timingSafeEqual(providedBuffer, expectedBuffer)
  if (!valid) {
    return res.status(401).json({ success: false, code: 401, message: 'invalid signature' })
  }

  if (!processedEventIds.has(eventId)) {
    // 应先通过数据库唯一键保存事件，再交给你自己的队列异步处理。
    processedEventIds.add(eventId)
  }
  return res.status(200).json({ success: true, code: 200, message: 'accepted' })
})

app.listen(3000)

import hashlib
import hmac
import os
import time
from flask import Flask, jsonify, request

app = Flask(__name__)
processed_event_ids = set()  # 生产环境请改为数据库唯一键。

@app.post("/webhooks/qingzhoushan")
def receive_webhook():
    event_id = request.headers.get("X-Webhook-Event-Id", "")
    timestamp = request.headers.get("X-Webhook-Timestamp", "")
    signature = request.headers.get("X-Webhook-Signature", "")
    app_secret = os.environ.get("QINGZHOUSHAN_APP_SECRET", "")
    payload = request.get_json(silent=True) or {}

    try:
        timestamp_number = int(timestamp)
    except ValueError:
        timestamp_number = 0
    if abs(int(time.time()) - timestamp_number) > 300:
        return jsonify(success=False, code=401, message="timestamp expired"), 401
    if not event_id or event_id != payload.get("event_id") or not app_secret:
        return jsonify(success=False, code=401, message="invalid event"), 401

    message = f"{event_id}\n{timestamp}".encode("utf-8")
    expected = hmac.new(app_secret.encode("utf-8"), message, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        return jsonify(success=False, code=401, message="invalid signature"), 401

    if event_id not in processed_event_ids:
        # 应先通过数据库唯一键保存事件，再交给你自己的队列异步处理。
        processed_event_ids.add(event_id)
    return jsonify(success=True, code=200, message="accepted"), 200