webhook-setup

Webhook 设置

此技能可帮助 AI Agent 构建生产级 Webhook 接收器并配置 Webhook 发送方。Agent 会实现 HTTP 端点，用于接收事件负载、验证加密签名以认证发送方、以幂等方式处理事件以安全应对重试，并按类型将事件路由到相应的处理程序。最终可实现可靠的事件驱动集成，处理重放攻击、乱序投递和服务提供商超时等现实故障情况。

工作流

1. 设计 Webhook 端点： 在稳定且难以猜测的 URL 路径上创建 HTTP POST 端点（例如 /webhooks/stripe、/webhooks/github）。端点必须快速返回 200 OK 响应（对大多数服务提供商而言，应在 5 秒内），以确认收到请求——耗时较长的处理应通过作业队列异步完成。只使用 HTTPS；大多数服务提供商会拒绝普通 HTTP 端点。

2. 实现签名验证： 每个 Webhook 服务提供商都会使用 HMAC-SHA256、RSA 或类似方案对负载签名。在处理任何事件之前，使用服务提供商的签名密钥验证签名。使用恒定时间比较函数比较签名，以防止计时攻击。对于缺少签名或签名无效的请求，立即以 401 Unauthorized 响应拒绝。读取原始请求正文进行验证——解析后的 JSON 可能与签名所依据的字节不同。

3. 解析事件并按类型路由： 解析已验证的负载并提取事件类型（例如 payment_intent.succeeded、push）。使用注册表或 switch 语句将每种事件类型路由到专用的处理函数。以警告级别记录无法识别的事件类型，并返回 200 OK，以防服务提供商无限期重试未处理的事件。

4. 以幂等方式处理事件： 如果服务提供商未及时收到 200 响应，就会重试投递 Webhook，这意味着处理程序可能会多次收到同一事件。在数据库表中存储已处理的事件 ID，并在处理前检查重复项。使用数据库事务，以原子方式将事件标记为已处理并执行其副作用。

5. 添加异步处理和死信队列： 对于需要大量处理的事件（例如发送电子邮件、更新多条记录），应立即确认收到 Webhook，并将事件加入队列以供后台处理。耗尽重试次数的失败事件应移至死信队列（DLQ），以便人工检查。设置对 DLQ 队列深度的监控和告警。

6. 在服务提供商一侧配置 Webhook： 向 Webhook 服务提供商注册端点 URL，选择所需的事件类型（仅订阅最少的一组），并记录签名密钥。使用服务提供商的测试／ping 功能测试 Webhook。在服务提供商控制面板中设置投递失败监控。

支持的技术

• Web 框架： Express.js、Fastify、Flask、FastAPI、Django、Rails、Spring Boot
• 队列系统： Bull/BullMQ（Redis）、Celery（Python）、Sidekiq（Ruby）、SQS、RabbitMQ
• 服务提供商： Stripe、GitHub、Slack、Twilio、SendGrid、Shopify、PayPal、Paddle
• 监控： Svix（Webhook 基础设施）、Hookdeck、ngrok（本地开发）
• 数据库： PostgreSQL、MySQL、Redis（用于幂等性跟踪）

用法

向 Agent 提供 Webhook 服务提供商（Stripe、GitHub 等）、要处理的事件以及服务器框架。Agent 将生成完整的 Webhook 接收器，包含签名验证、事件路由、幂等处理和错误处理。对于本地开发，Agent 可以设置 ngrok 或类似的隧道进行测试。

示例

示例 1：Stripe Webhook 接收器（Express.js）

const express = require("express");
const crypto = require("crypto");
const { Queue } = require("bullmq");

const app = express();
const eventQueue = new Queue("webhook-events", { connection: { host: "localhost" } });

// CRITICAL: Use raw body for signature verification — must be before json parser
app.post("/webhooks/stripe", express.raw({ type: "application/json" }), async (req, res) => {
  const signature = req.headers["stripe-signature"];
  const webhookSecret = process.env.STRIPE_WEBHOOK_SECRET;

  // Step 1: Verify signature
  let event;
  try {
    event = verifyStripeSignature(req.body, signature, webhookSecret);
  } catch (err) {
    console.error(`Signature verification failed: ${err.message}`);
    return res.status(401).json({ error: "Invalid signature" });
  }

  // Step 2: Idempotency check — skip if already processed
  const alreadyProcessed = await checkIdempotency(event.id);
  if (alreadyProcessed) {
    console.log(`Event ${event.id} already processed, skipping`);
    return res.status(200).json({ received: true, deduplicated: true });
  }

  // Step 3: Acknowledge immediately, process async
  try {
    await eventQueue.add(event.type, {
      eventId: event.id,
      type: event.type,
      data: event.data.object,
      created: event.created,
    });
    await markEventReceived(event.id);
    res.status(200).json({ received: true });
  } catch (err) {
    console.error(`Failed to enqueue event ${event.id}:`, err);
    res.status(500).json({ error: "Processing failed" });
  }
});

// Stripe signature verification (manual implementation)
function verifyStripeSignature(payload, signatureHeader, secret) {
  const elements = signatureHeader.split(",").reduce((acc, part) => {
    const [key, value] = part.split("=");
    acc[key.trim()] = value;
    return acc;
  }, {});

  const timestamp = elements["t"];
  const expectedSig = elements["v1"];

  // Protect against replay attacks: reject events older than 5 minutes
  const tolerance = 300; // 5 minutes
  const currentTime = Math.floor(Date.now() / 1000);
  if (currentTime - parseInt(timestamp) > tolerance) {
    throw new Error("Webhook timestamp too old — possible replay attack");
  }

  // Compute expected signature
  const signedPayload = `${timestamp}.${payload}`;
  const computedSig = crypto
    .createHmac("sha256", secret)
    .update(signedPayload)
    .digest("hex");

  // Constant-time comparison to prevent timing attacks
  if (!crypto.timingSafeEqual(Buffer.from(expectedSig), Buffer.from(computedSig))) {
    throw new Error("Signature mismatch");
  }

  return JSON.parse(payload);
}

// Idempotency tracking with PostgreSQL
const { Pool } = require("pg");
const db = new Pool();

async function checkIdempotency(eventId) {
  const result = await db.query(
    "SELECT 1 FROM processed_webhook_events WHERE event_id = $1",
    [eventId]
  );
  return result.rows.length > 0;
}

async function markEventReceived(eventId) {
  await db.query(
    "INSERT INTO processed_webhook_events (event_id, received_at) VALUES ($1, NOW()) ON CONFLICT (event_id) DO NOTHING",
    [eventId]
  );
}

// Event handler workers (BullMQ)
const { Worker } = require("bullmq");

const worker = new Worker("webhook-events", async (job) => {
  const { eventId, type, data } = job.data;

  const handlers = {
    "payment_intent.succeeded": handlePaymentSucceeded,
    "payment_intent.payment_failed": handlePaymentFailed,
    "customer.subscription.deleted": handleSubscriptionCanceled,
    "invoice.payment_failed": handleInvoicePaymentFailed,
  };

  const handler = handlers[type];
  if (!handler) {
    console.log(`Unhandled event type: ${type}`);
    return;
  }

  await handler(data, eventId);
  await db.query("UPDATE processed_webhook_events SET processed_at = NOW() WHERE event_id = $1", [eventId]);
}, { connection: { host: "localhost" } });

async function handlePaymentSucceeded(paymentIntent, eventId) {
  const orderId = paymentIntent.metadata.order_id;
  await db.query("UPDATE orders SET status = 'paid', payment_id = $1 WHERE id = $2", [
    paymentIntent.id, orderId,
  ]);
  console.log(`Order ${orderId} marked as paid (event: ${eventId})`);
}

async function handlePaymentFailed(paymentIntent, eventId) {
  const orderId = paymentIntent.metadata.order_id;
  await db.query("UPDATE orders SET status = 'payment_failed' WHERE id = $1", [orderId]);
  // Trigger customer notification
  console.log(`Payment failed for order ${orderId} (event: ${eventId})`);
}

async function handleSubscriptionCanceled(subscription, eventId) {
  await db.query("UPDATE users SET subscription_status = 'canceled' WHERE stripe_customer_id = $1", [
    subscription.customer,
  ]);
}

async function handleInvoicePaymentFailed(invoice, eventId) {
  console.log(`Invoice ${invoice.id} payment failed, attempt ${invoice.attempt_count}`);
}

app.listen(3000, () => console.log("Webhook receiver listening on port 3000"));

示例 2：用于 CI 触发的 GitHub Webhook（Python/FastAPI）

import hashlib
import hmac
import os
import logging
from datetime import datetime, timezone
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()
logger = logging.getLogger(__name__)

GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]

# In-memory idempotency store (use Redis or PostgreSQL in production)
processed_deliveries: set[str] = set()


def verify_github_signature(payload_body: bytes, signature_header: str | None) -> None:
    """Verify the GitHub webhook signature using HMAC-SHA256."""
    if not signature_header:
        raise HTTPException(status_code=401, detail="Missing X-Hub-Signature-256 header")

    expected_signature = "sha256=" + hmac.new(
        GITHUB_WEBHOOK_SECRET.encode(),
        payload_body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected_signature, signature_header):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")


@app.post("/webhooks/github")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
    # Step 1: Read raw body and verify signature
    body = await request.body()
    signature = request.headers.get("X-Hub-Signature-256")
    verify_github_signature(body, signature)

    # Step 2: Idempotency check using delivery ID
    delivery_id = request.headers.get("X-GitHub-Delivery")
    if delivery_id in processed_deliveries:
        logger.info(f"Duplicate delivery {delivery_id}, skipping")
        return {"status": "duplicate", "delivery_id": delivery_id}

    # Step 3: Parse event and route
    event_type = request.headers.get("X-GitHub-Event")
    payload = await request.json()

    logger.info(f"Received GitHub event: {event_type} (delivery: {delivery_id})")

    # Acknowledge immediately, process in background
    processed_deliveries.add(delivery_id)
    background_tasks.add_task(process_github_event, event_type, payload, delivery_id)

    return {"status": "accepted", "event": event_type, "delivery_id": delivery_id}


async def process_github_event(event_type: str, payload: dict, delivery_id: str):
    """Route and process GitHub webhook events."""
    handlers = {
        "push": handle_push,
        "pull_request": handle_pull_request,
        "issues": handle_issue,
        "ping": handle_ping,
    }

    handler = handlers.get(event_type)
    if not handler:
        logger.warning(f"Unhandled GitHub event type: {event_type}")
        return

    try:
        await handler(payload)
        logger.info(f"Successfully processed {event_type} (delivery: {delivery_id})")
    except Exception as e:
        logger.error(f"Failed to process {event_type} (delivery: {delivery_id}): {e}")
        # In production, move to dead letter queue for retry
        raise


async def handle_push(payload: dict):
    """Trigger CI pipeline on push to main branch."""
    ref = payload.get("ref", "")
    repo = payload["repository"]["full_name"]
    commits = payload.get("commits", [])

    if ref != "refs/heads/main":
        logger.info(f"Ignoring push to {ref} on {repo}")
        return

    commit_messages = [c["message"].split("\n")[0] for c in commits]
    logger.info(f"Push to main on {repo}: {len(commits)} commits")

    # Check if any commits modify CI-relevant files
    changed_files = set()
    for commit in commits:
        changed_files.update(commit.get("added", []))
        changed_files.update(commit.get("modified", []))

    # Trigger appropriate CI jobs
    if any(f.startswith("src/") for f in changed_files):
        await trigger_ci_pipeline(repo, "build-and-test", payload["after"])
    if any(f.startswith("infrastructure/") for f in changed_files):
        await trigger_ci_pipeline(repo, "infrastructure-plan", payload["after"])

    logger.info(f"CI pipelines triggered for {repo}@{payload['after'][:8]}")


async def handle_pull_request(payload: dict):
    """Run checks on pull request events."""
    action = payload["action"]
    pr = payload["pull_request"]
    repo = payload["repository"]["full_name"]

    if action in ("opened", "synchronize"):
        logger.info(f"PR #{pr['number']} {action} on {repo}: {pr['title']}")
        await trigger_ci_pipeline(repo, "pr-checks", pr["head"]["sha"])
    elif action == "closed" and pr.get("merged"):
        logger.info(f"PR #{pr['number']} merged on {repo}")
        await trigger_ci_pipeline(repo, "deploy-staging", pr["merge_commit_sha"])


async def handle_issue(payload: dict):
    """Log issue events."""
    action = payload["action"]
    issue = payload["issue"]
    logger.info(f"Issue #{issue['number']} {action}: {issue['title']}")


async def handle_ping(payload: dict):
    """Respond to GitHub's test ping."""
    logger.info(f"Ping received. Hook ID: {payload.get('hook_id')}, Zen: {payload.get('zen')}")


async def trigger_ci_pipeline(repo: str, pipeline: str, commit_sha: str):
    """Trigger a CI pipeline (stub — replace with actual CI API call)."""
    logger.info(f"Triggering {pipeline} for {repo}@{commit_sha[:8]}")
    # In production: POST to Jenkins, GitHub Actions, CircleCI, etc.

最佳实践

• 快速响应，稍后处理。 在 3-5 秒内返回 200 OK。大多数提供商会在 10-30 秒后超时并重试，导致重复处理。将事件加入后台作业队列，供后续实际处理。
• 处理任何 Webhook 负载前，务必验证签名。 使用提供商的签名密钥和恒定时间比较函数。即使在开发环境中，也绝不要跳过验证。
• 使用事件 ID 实现幂等性。 将提供商的投递 ID／事件 ID 存入数据库，并在处理前检查是否重复。使用 INSERT ... ON CONFLICT DO NOTHING 实现原子去重。
• 使用死信队列存放所有重试后仍处理失败的事件。监控 DLQ 深度并设置警报。提供工具，以便修复 Bug 后重放 DLQ 中的事件。
• 只订阅需要处理的事件。 接收不处理的事件会浪费带宽，并使日志产生噪声。大多数提供商允许你在配置 Webhook 时选择特定事件类型。
• 通过检查事件时间戳防范重放攻击。 拒绝超过 5 分钟的事件（可根据你的容忍度配置）。Stripe、GitHub 和大多数提供商都会在签名标头中包含时间戳，以便进行此项检查。

边缘情况

• 乱序投递： Webhook 可能会乱序到达（例如，invoice.paid 早于 invoice.created）。应使用事件时间戳和资源的当前状态（通过 API 获取）来做出判断，而不是假定事件按顺序投递。
• 部署期间提供商重试： 如果你的服务在部署期间宕机，提供商会重试。确保幂等逻辑能够处理服务恢复后收到的同一事件。使用滚动部署，确保始终至少有一个实例可用。
• 负载大小限制： 某些提供商会发送较大的负载（尤其是包含大量差异的 GitHub 推送事件）。为端点设置适当的请求体大小限制（例如 10MB），并妥善处理 413 Payload Too Large。
• 密钥轮换： 轮换 Webhook 签名密钥时，在过渡期间同时支持新旧密钥。优先使用新密钥验证，失败时再回退到旧密钥。确认所有投递都已使用新密钥后，再移除旧密钥。
• 端点 URL 变更： 如果需要更改 Webhook URL，请先注册新 URL，再注销旧 URL。在提供商确认成功向新 URL 投递之前，让两个端点并行运行。
• 本地测试： 使用 ngrok、Cloudflare Tunnel 或提供商的 CLI（例如 stripe listen --forward-to localhost:3000/webhooks/stripe），将 Webhook 转发到本地开发服务。