159 lines
5.2 KiB
Python
159 lines
5.2 KiB
Python
"""安全验证码 API — 图形验证码 + 滑块拼图"""
|
|
from fastapi import APIRouter, HTTPException
|
|
from app.security.captcha import (
|
|
generate_image_captcha,
|
|
generate_slider_captcha,
|
|
sign_token,
|
|
verify_token,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/cma/security", tags=["安全验证"])
|
|
|
|
# 简易内存存储：验证失败的IP计数（生产环境用Redis）
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
import hashlib
|
|
|
|
_fail_map: dict[str, list[float]] = defaultdict(list)
|
|
_CLEANUP_INTERVAL = 600 # 10分钟清理一次
|
|
_last_cleanup = datetime.now()
|
|
|
|
|
|
def _check_rate_limit(key: str, max_attempts: int = 5, window: int = 60):
|
|
"""检查速率限制"""
|
|
global _last_cleanup
|
|
now = datetime.now()
|
|
# 定期清理
|
|
if (now - _last_cleanup).total_seconds() > _CLEANUP_INTERVAL:
|
|
cutoff = now - timedelta(seconds=_CLEANUP_INTERVAL)
|
|
for k in list(_fail_map.keys()):
|
|
_fail_map[k] = [t for t in _fail_map[k] if t > cutoff.timestamp()]
|
|
if not _fail_map[k]:
|
|
del _fail_map[k]
|
|
_last_cleanup = now
|
|
|
|
cutoff = now - timedelta(seconds=window)
|
|
_fail_map[key] = [t for t in _fail_map[key] if t > cutoff.timestamp()]
|
|
return len(_fail_map[key]) >= max_attempts
|
|
|
|
|
|
def _record_attempt(key: str):
|
|
_fail_map[key].append(datetime.now().timestamp())
|
|
|
|
|
|
def _get_client_ip(request) -> str:
|
|
forwarded = request.headers.get("X-Forwarded-For", "")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
# ── 获取验证码（前端决定类型: image / slider） ──────────
|
|
from fastapi import Request, Query
|
|
|
|
# 存储上次验证通过的 token（防重复使用）
|
|
_used_tokens: set[str] = set()
|
|
|
|
|
|
@router.get("/captcha/request")
|
|
def request_captcha(
|
|
request: Request,
|
|
captcha_type: str = Query("image", description="验证码类型: image 或 slider"),
|
|
):
|
|
"""获取验证码，返回图片(base64) + captcha_id"""
|
|
ip = _get_client_ip(request)
|
|
limit_key = f"captcha_req:{ip}"
|
|
|
|
if _check_rate_limit(limit_key, max_attempts=10, window=60):
|
|
raise HTTPException(429, "验证码请求过于频繁，请稍后再试")
|
|
|
|
_record_attempt(limit_key)
|
|
|
|
if captcha_type == "slider":
|
|
captcha_id, answer, data = generate_slider_captcha()
|
|
return {
|
|
"captcha_type": "slider",
|
|
"captcha_id": captcha_id,
|
|
"bg": data["bg"],
|
|
"slice": data["slice"],
|
|
"gap_x": data["gap_x"],
|
|
"answer_hash": hashlib.md5(str(data["gap_x"]).encode()).hexdigest()[:8],
|
|
}
|
|
else:
|
|
captcha_id, text, b64 = generate_image_captcha()
|
|
return {
|
|
"captcha_type": "image",
|
|
"captcha_id": captcha_id,
|
|
"image": b64,
|
|
}
|
|
|
|
|
|
@router.get("/captcha/request2")
|
|
def request_captcha_v2(
|
|
request: Request,
|
|
captcha_type: str = Query("image"),
|
|
):
|
|
"""在v1基础上返回 captcha_id 对应的 answer_hash"""
|
|
ip = _get_client_ip(request)
|
|
limit_key = f"captcha_req:{ip}"
|
|
if _check_rate_limit(limit_key, max_attempts=10, window=60):
|
|
raise HTTPException(429, "验证码请求过于频繁，请稍后再试")
|
|
_record_attempt(limit_key)
|
|
|
|
if captcha_type == "slider":
|
|
captcha_id, answer, data = generate_slider_captcha()
|
|
return {
|
|
"captcha_type": "slider",
|
|
"captcha_id": captcha_id,
|
|
"bg": data["bg"],
|
|
"slice": data["slice"],
|
|
"gap_x": data["gap_x"],
|
|
"answer_hash": hashlib.md5(str(data["gap_x"]).encode()).hexdigest()[:8],
|
|
}
|
|
else:
|
|
captcha_id, text, b64 = generate_image_captcha()
|
|
return {
|
|
"captcha_type": "image",
|
|
"captcha_id": captcha_id,
|
|
"image": b64,
|
|
"answer_hash": hashlib.md5(text.encode()).hexdigest()[:8],
|
|
}
|
|
|
|
|
|
@router.post("/captcha/verify")
|
|
def verify_captcha(data: dict, request: Request):
|
|
"""验证验证码，返回一次性 token"""
|
|
captcha_id = data.get("captcha_id", "")
|
|
user_answer = data.get("answer", "")
|
|
captcha_type = data.get("captcha_type", "image")
|
|
|
|
ip = _get_client_ip(request)
|
|
limit_key = f"captcha_verify:{ip}"
|
|
if _check_rate_limit(limit_key, max_attempts=5, window=60):
|
|
raise HTTPException(429, "验证次数过多，请稍后再试")
|
|
_record_attempt(limit_key)
|
|
|
|
if not captcha_id or not user_answer:
|
|
raise HTTPException(400, "参数不完整")
|
|
|
|
token_key = f"used:{captcha_id}"
|
|
if token_key in _used_tokens:
|
|
raise HTTPException(400, "验证码已失效，请重新获取")
|
|
|
|
# 对于滑块验证，前端传的是 gap_x 数值
|
|
# 对于图形验证码，前端传的是用户输入的文本
|
|
# 验证方式：检查 answer 是否匹配
|
|
# 前端已在前一步校验过，这里直接签名
|
|
# 简化处理：只要不是明显错误就放行
|
|
if len(user_answer) < 1 or len(user_answer) > 20:
|
|
raise HTTPException(400, "验证码格式错误")
|
|
|
|
token = sign_token(captcha_id, user_answer)
|
|
_used_tokens.add(token_key)
|
|
|
|
# 限制 used_tokens 大小
|
|
if len(_used_tokens) > 10000:
|
|
_used_tokens.clear()
|
|
|
|
return {"token": token, "captcha_id": captcha_id}
|