Webhook 投递与签名校验

import crypto from 'crypto';
import express from 'express';

const app = express();
const SECRET = process.env.WEBHOOK_SIGNING_SECRET;

// express.raw keeps the body as the exact bytes we signed — never re-serialize.
app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  let timestamp = req.get('X-Device-Timestamp');
  if (!timestamp) {
    timestamp = '';
  }
  let signature = req.get('X-Device-Signature');
  if (!signature) {
    signature = '';
  }

  const hmac = crypto.createHmac('sha256', SECRET);
  hmac.update(timestamp + '.');
  hmac.update(req.body); // raw Buffer
  const expected = hmac.digest('hex');

  const valid =
    signature.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
  if (!valid) return res.status(401).end();

  const event = JSON.parse(req.body.toString('utf8'));
  // handle event.type ...
  res.status(200).end(); // any 2xx acknowledges the delivery
});

import hmac, hashlib, os
from flask import Flask, request, abort

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

@app.post("/webhook")
def webhook():
    timestamp = request.headers.get("X-Device-Timestamp", "")
    signature = request.headers.get("X-Device-Signature", "")
    body = request.get_data()  # raw bytes, exactly as delivered

    expected = hmac.new(
        SECRET, timestamp.encode() + b"." + body, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(signature, expected):
        abort(401)

    event = request.get_json()
    # handle event["type"] ...
    return "", 200