Google Cloud Fraud Defense 登场：从验证码到全链路反欺诈的进化

import json
import httpx

# 项目配置 — 替换为你的 Google Cloud 项目 ID 和 API Key
PROJECT_ID = "your-gcp-project-id"
API_KEY = "your-api-key"
SITE_KEY = "your-site-key"

FRAUD_DEFENSE_ENDPOINT = (
    f"https://frauddefense.googleapis.com/v1/projects/{PROJECT_ID}"
    f"/assessments?key={API_KEY}"
)


async def assess_login_risk(session_token: str, user_email: str, client_ip: str) -> dict:
    """
    在登录流程中调用 Cloud Fraud Defense 进行风险评估。
    session_token: 前端 SDK 生成的会话令牌
    user_email: 用户登录邮箱
    client_ip: 请求来源 IP（从 X-Forwarded-For 或连接信息提取）
    返回: 风险评分与建议动作
    """
    payload = {
        "event": {
            "token": session_token,
            "siteKey": SITE_KEY,
            "expectedAction": "LOGIN",
            "userAgent": "server-side-integration",
            "userInfo": {
                "accountId": user_email,
            },
            "clientIp": client_ip,
        },
        # 指定需要评估的欺诈类型
        "assessmentTypes": [
            "FRAUD_ACCOUNT_TAKEOVER",
            "FRAUD_CREDENTIAL_STUFFING",
        ],
    }

    resp = await httpx.post(FRAUD_DEFENSE_ENDPOINT, json=payload, timeout=10)
    result = resp.json()

    # 解析关键字段 — 字段名以官方文档为准，此处为示意结构
    risk_score = result.get("riskAnalysis", {}).get("riskScore", 0)  # 0-1 浮点数
    reasons = result.get("riskAnalysis", {}).get("reasons", [])
    recommended_action = result.get("riskAnalysis", {}).get("recommendedAction", "ALLOW")

    return {
        "risk_score": risk_score,
        "reasons": reasons,
        "recommended_action": recommended_action,
    }


# ── 在 FastAPI 路由中使用 ──────────────────────────────────
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()


@app.post("/api/login")
async def login(request: Request):
    body = await request.json()
    email = body["email"]
    token = body["fraud_token"]  # 前端 Cloud Fraud Defense SDK 生成

    client_ip = request.headers.get("X-Forwarded-For", request.client.host)

    assessment = await assess_login_risk(token, email, client_ip)

    # 根据风险评分决定下一步
    if assessment["recommended_action"] == "BLOCK":
        raise HTTPException(status_code=403, detail="登录行为异常，已被拦截")
    elif assessment["recommended_action"] == "CHALLENGE":
        # 返回前端，触发二次验证（MFA / 验证码）
        return {"status": "challenge_required", "reasons": assessment["reasons"]}
    else:
        # 正常通过，继续执行登录逻辑
        return {"status": "allowed", "risk_score": assessment["risk_score"]}