CaptchaAI 回调签名验证：Webhook 安全指南

pingback 回调地址一旦泄露，任何人都能伪造 GET 请求塞入假 token——回调端点默认不带身份校验。本文给出四种可叠加的防护方案，均附 Python 和 JavaScript 示例。

pingback 回调是怎么工作的


1. You submit task:
   POST https://ocr.captchaai.com/in.php
     ?key=YOUR_API_KEY
     &method=userrecaptcha
     &googlekey=SITE_KEY
     &pageurl=https://example.com
     &pingback=https://your-server.com/captcha/callback

2. CaptchaAI solves the CAPTCHA

3. CaptchaAI sends result to your endpoint:
   GET https://your-server.com/captcha/callback?id=TASK_ID&code=SOLUTION_TOKEN

第 3 步是关键：这是未经身份验证的请求，服务器无法凭直觉判断来源。下面四种方案分别拦截不同攻击面。

方案一：任务 ID 校验（最低门槛，务必先上）

提交任务时把返回的 request（任务 ID）记进内存或 Redis 的待处理集合
回调到达时核对一次，未知 ID 一律拒绝，通过后从集合删掉

Python（Flask）

import os
import threading
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# Thread-safe set of pending task IDs
pending_tasks = set()
pending_lock = threading.Lock()
results = {}

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def submit_captcha(sitekey, pageurl):
    """Submit CAPTCHA and register the task ID."""
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": "https://your-server.com/captcha/callback",
        "json": 1
    })
    data = resp.json()

    if data.get("status") == 1:
        task_id = data["request"]
        with pending_lock:
            pending_tasks.add(task_id)
        return task_id
    return None


@app.route("/captcha/callback")
def captcha_callback():
    task_id = request.args.get("id")
    solution = request.args.get("code")

    # Validate: only accept known task IDs
    with pending_lock:
        if task_id not in pending_tasks:
            return jsonify({"error": "unknown task"}), 403
        pending_tasks.discard(task_id)

    results[task_id] = solution
    return "OK", 200

JavaScript（Express）：

const express = require("express");
const axios = require("axios");

const app = express();
const API_KEY = process.env.CAPTCHAAI_API_KEY;

const pendingTasks = new Set();
const results = new Map();

async function submitCaptcha(sitekey, pageurl) {
  const resp = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: {
      key: API_KEY,
      method: "userrecaptcha",
      googlekey: sitekey,
      pageurl: pageurl,
      pingback: "https://your-server.com/captcha/callback",
      json: 1,
    },
  });

  if (resp.data.status === 1) {
    const taskId = resp.data.request;
    pendingTasks.add(taskId);
    return taskId;
  }
  return null;
}

app.get("/captcha/callback", (req, res) => {
  const taskId = req.query.id;
  const solution = req.query.code;

  // Validate: only accept known task IDs
  if (!pendingTasks.has(taskId)) {
    return res.status(403).json({ error: "unknown task" });
  }

  pendingTasks.delete(taskId);
  results.set(taskId, solution);
  res.sendStatus(200);
});

app.listen(3000);

方案二：HMAC 签名，让回调地址无法被猜出

只校验任务 ID 还不够——ID 格式可能被硬猜出来
更稳做法：给回调 URL 加一个只有你能算出的签名，服务器重新计算比对，对不上就拒绝

Python

import hashlib
import hmac
import os

CALLBACK_SECRET = os.environ["CALLBACK_SECRET"]  # Random 32+ character string


def generate_callback_url(task_id):
    """Generate callback URL with HMAC signature."""
    signature = hmac.new(
        CALLBACK_SECRET.encode(),
        task_id.encode(),
        hashlib.sha256
    ).hexdigest()

    return f"https://your-server.com/captcha/callback?token={signature}"


@app.route("/captcha/callback")
def captcha_callback():
    task_id = request.args.get("id")
    token = request.args.get("token")
    solution = request.args.get("code")

    # Verify HMAC signature
    expected = hmac.new(
        CALLBACK_SECRET.encode(),
        task_id.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(token, expected):
        return jsonify({"error": "invalid signature"}), 403

    results[task_id] = solution
    return "OK", 200

JavaScript：

const crypto = require("crypto");

const CALLBACK_SECRET = process.env.CALLBACK_SECRET;

function generateCallbackUrl(taskId) {
  const signature = crypto
    .createHmac("sha256", CALLBACK_SECRET)
    .update(taskId)
    .digest("hex");

  return `https://your-server.com/captcha/callback?token=${signature}`;
}

app.get("/captcha/callback", (req, res) => {
  const taskId = req.query.id;
  const token = req.query.token;
  const solution = req.query.code;

  // Verify HMAC signature
  const expected = crypto
    .createHmac("sha256", CALLBACK_SECRET)
    .update(taskId)
    .digest("hex");

  if (!crypto.timingSafeEqual(Buffer.from(token), Buffer.from(expected))) {
    return res.status(403).json({ error: "invalid signature" });
  }

  results.set(taskId, solution);
  res.sendStatus(200);
});

提交任务时把签名拼进 pingback 参数：pingback=https://your-server.com/captcha/callback?token=HMAC_SIGNATURE
CALLBACK_SECRET 只放服务器端环境变量，不要写进前端代码或提交到仓库

方案三：IP 白名单，只放行 CaptchaAI 的服务器

只放行 CaptchaAI 服务器来源的请求，其余拒绝
IP 段可能变化，建议定期核实

Python（Flask）

# CaptchaAI callback source IPs (verify current IPs with CaptchaAI support)
ALLOWED_IPS = {"138.201.XX.XX", "148.251.XX.XX"}  # Replace with actual IPs


@app.before_request
def check_ip():
    if request.path.startswith("/captcha/callback"):
        client_ip = request.remote_addr
        if client_ip not in ALLOWED_IPS:
            return jsonify({"error": "forbidden"}), 403

JavaScript（Express）：

const ALLOWED_IPS = new Set(["138.201.XX.XX", "148.251.XX.XX"]);

app.use("/captcha/callback", (req, res, next) => {
  const clientIp = req.ip || req.connection.remoteAddress;
  if (!ALLOWED_IPS.has(clientIp)) {
    return res.status(403).json({ error: "forbidden" });
  }
  next();
});

注意： 具体 IP 段请找 CaptchaAI 支持人员确认。部署在阿里云 SLB、腾讯云 CLB 后面时，request.remote_addr 拿到的是代理内网 IP，需改读 X-Forwarded-For，否则白名单不会命中。

防重放攻击：一次性使用 + 时间戳校验

前三种方案都挡不住“重放”——截获一个有效回调，再发一次仍会被接受
修复方式：给回调加上时间戳有效期，每个任务 ID 只允许处理一次

Python

import time

CALLBACK_TTL = 300  # Reject callbacks older than 5 minutes
used_callbacks = set()


@app.route("/captcha/callback")
def captcha_callback():
    task_id = request.args.get("id")
    timestamp = request.args.get("ts")
    solution = request.args.get("code")

    # Check timestamp freshness
    if timestamp:
        age = time.time() - float(timestamp)
        if age > CALLBACK_TTL or age < 0:
            return jsonify({"error": "expired"}), 403

    # One-time use
    if task_id in used_callbacks:
        return jsonify({"error": "already processed"}), 409

    used_callbacks.add(task_id)
    results[task_id] = solution
    return "OK", 200

四层防护速查表

层级	防住什么	落地方式
任务 ID 校验	随机/未知任务注入	记录待处理 ID，回调时核对
HMAC 签名	URL 被猜测、伪造回调	密钥签名，服务端重新计算比对
IP 白名单	非授权服务器请求	只放行 CaptchaAI 来源 IP
防重放	有效回调被重复提交	一次性使用 + 时间戳校验
HTTPS	窃听、中间人攻击	回调端点全程走 TLS

常见故障排查

所有回调都被拒绝——IP 白名单未覆盖当前 IP：核实最新 IP，检查反向代理是否透传来源 IP
HMAC 校验总失败——提交与回调的任务 ID 不一致：确认用的是 in.php 返回的原始 request 值
同一回调处理两次——并发请求竞态条件：改用原子集合操作，或数据库唯一约束
回调经常超时——端点同步处理耗时过长：先返回 200，再异步处理业务逻辑

常见问题

本地开发环境怎么测试 pingback 回调？

localhost 收不到公网请求，可用 ngrok、cpolar 等内网穿透工具生成临时公网地址替换 pingback 参数。上线前记得换回正式域名。

四种防护策略要同时全部启用吗？

任务 ID 校验（方案一）是底线；对外暴露的端点再加 HMAC 签名（方案二）。若 CaptchaAI 提供稳定来源 IP，IP 白名单（方案三）值得加上；涉及计费等敏感业务，防重放基本必选。

CALLBACK_SECRET 泄露了该怎么处理？

立刻换一个新的随机字符串（32 位以上）。换密钥不影响历史任务，只是旧的、未回调的 URL 会失效，需重新提交。

回调端点临时挂掉，结果会丢失吗？

不会。结果始终可通过轮询接口 res.php 拿到——建议做一个兜底任务，定期扫描超时未收到回调的任务 ID 并主动查询。

下一步

给回调端点加上签名校验——获取 API Key，从任务 ID 校验开始，逐步补齐。

继续阅读：

CaptchaAI Webhook 安全：验证回调签名

pingback 回调是怎么工作的

方案一：任务 ID 校验（最低门槛，务必先上）

Python（Flask）

方案二：HMAC 签名，让回调地址无法被猜出

Python

方案三：IP 白名单，只放行 CaptchaAI 的服务器

Python（Flask）

防重放攻击：一次性使用 + 时间戳校验

Python

四层防护速查表

常见故障排查

常见问题

本地开发环境怎么测试 pingback 回调？

四种防护策略要同时全部启用吗？

CALLBACK_SECRET 泄露了该怎么处理？

回调端点临时挂掉，结果会丢失吗？

相关文章

下一步

CaptchaAI IP白名单和API密钥安全

CaptchaAI 回调 URL 设置：完整的 Webhook 指南

CaptchaAI 回调 URL 错误处理：重试和死信模式

CaptchaAI Pingback 和任务通知模式

CAPTCHA 管道状态的 Discord Webhook 警报

AWS Lambda + CaptchaAI：无服务器验证码解决