Webhook

import crypto from "node:crypto";
import express from "express";

const SECRET = process.env.VAL_WEBHOOK_SECRET; // whsec_...
const TOLERANCE_SECONDS = 300;

function verify(rawBody, header) {
  const parts = Object.fromEntries(header.split(",").map((part) => part.split("=")));
  const timestamp = Number(parts.t);
  if (!timestamp || !parts.v1) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > TOLERANCE_SECONDS) return false;

  const expected = crypto.createHmac("sha256", SECRET).update(`${timestamp}.${rawBody}`).digest("hex");
  const a = Buffer.from(expected, "hex");
  const b = Buffer.from(parts.v1, "hex");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

const app = express();

// 必须拿到原始请求体，不能先用 express.json() 解析
app.post("/webhooks/voiceailabs", express.raw({ type: "application/json" }), async (req, res) => {
  const rawBody = req.body.toString("utf8");
  if (!verify(rawBody, req.get("VoiceAILabs-Signature") || "")) {
    return res.status(400).send("invalid signature");
  }

  const event = JSON.parse(rawBody);
  // 先返回 2xx，再异步处理耗时逻辑
  res.sendStatus(204);

  if (await alreadyHandled(event.id)) return;
  if (event.type === "synthesis.completed") {
    await saveAudio(event.data.id, event.data.audio_url);
  }
});

import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["VAL_WEBHOOK_SECRET"].encode()  # whsec_...
TOLERANCE_SECONDS = 300

app = Flask(__name__)


def verify(raw_body: bytes, header: str) -> bool:
    try:
        parts = dict(part.split("=", 1) for part in header.split(","))
        timestamp = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(SECRET, f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


@app.post("/webhooks/voiceailabs")
def handle_webhook():
    raw_body = request.get_data()  # 原始字节
    if not verify(raw_body, request.headers.get("VoiceAILabs-Signature", "")):
        abort(400)

    event = json.loads(raw_body)
    if event["type"] == "synthesis.completed":
        save_audio(event["data"]["id"], event["data"]["audio_url"])
    return "", 204