当您使用 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 密钥并实现签名验证。
相关指南: