实战教程

CaptchaAI Webhook 安全:验证回调签名

当您使用 CaptchaAI 的回调 URL 功能 (pingback) 时,您的服务器会公开接收 CAPTCHA 解决方案的 HTTP 端点。如果没有验证,任何发现该 URL 的人都可以发送虚假解决方案。本教程介绍如何保护回调端点。

回调流程


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 步是未经身份验证的请求。您需要验证它确实来自 CaptchaAI。

验证策略1:任务ID验证

最简单的方法 - 只接受您实际提交的任务 ID 的回调结果。

Python(烧瓶)

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(快速)

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);

验证策略2:HMAC签名令牌

将秘密令牌添加到攻击者无法猜测的回调 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);
});

提交时使用生成的URL:pingback=https://your-server.com/captcha/callback?token=HMAC_SIGNATURE

验证策略 3:IP 白名单

将您的回调端点限制为 CaptchaAI 的服务器 IP。

Python(烧瓶)

# 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(快速)

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();
});

注意: 请联系 CaptchaAI 支持人员获取当前回调源 IP 列表。如果您位于反向代理后面,请确保 X-Forwarded-For 标头配置正确。

重放攻击预防

甚至可以重放有效的回调。添加时间戳检查和一次性强制执行:

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验证 Random/unknown任务注入 存储待处理的 ID,拒绝未知的 ID
HMAC签名 URL猜测、伪造回调 使用密钥对回调 URL 进行签名
IP 许可名单 来自未经授权的服务器的请求 白名单 CaptchaAI IP
预防重播 重新提交有效回调 一次性使用+时间戳验证
HTTPS 窃听、中间人 回调端点上的 TLS

故障排除

问题 原因 处理方式
所有回调均被拒绝 IP 允许列表不包括 CaptchaAI IP 在支持下验证当前 IP;检查反向代理标头
HMAC验证失败 提交和回调之间的任务 ID 不匹配 确保您使用 in.php 返回的确切任务 ID
处理重复的回调 并发回调的竞争条件 使用原子集操作或数据库唯一约束
回调超时 端点响应时间过长 异步处理——立即接受,在后台处理

常问问题

我应该同时使用所有四种验证策略吗?

至少使用任务 ID 验证(策略 1)。为面向公众的端点添加 HMAC 签名(策略 2)。如果 CaptchaAI 发布稳定的回调 IP,则 IP 白名单(策略 3)是理想的选择。预防重播对于财务或敏感工作流程至关重要。

如果 CaptchaAI 发送结果时我的回调端点已关闭,会发生什么情况?

该解决方案仍然可以通过轮询端点 (res.php) 获得。实现回退,轮询在超时期限内未收到回调的任何任务。

我可以使用双向 TLS (mTLS) 进行回调身份验证吗?

理论上是的,但是 CaptchaAI 的回调系统使用标准 HTTPS GET 请求。HMAC 签名提供等效的身份验证,无需证书管理。

相关文章

下一步

保护您的 CaptchaAI 回调端点 -获取您的 API 密钥并实现签名验证。

相关指南:

该文章已禁用评论。