oauth-2-0-setup

OAuth 2.0 设置

此技能可让 AI Agent 为 API 集成实现 OAuth 2.0 身份验证。Agent 会根据使用场景选择合适的授权类型：面向用户的应用使用带 PKCE 的授权码模式，机器间身份验证使用客户端凭据模式，输入受限的设备使用设备码模式。它会在整个流程中处理令牌存储、刷新令牌轮换、通过 state 参数防范 CSRF，以及安全的凭据管理。

工作流

1. 选择合适的授权类型： 根据客户端类型选择 OAuth 2.0 流程。对于有用户参与的 Web 和移动应用，使用带 PKCE 的授权码模式；PKCE 可替代客户端密钥，并防止授权码拦截攻击。对于没有用户上下文的服务器间通信，使用客户端凭据模式。对于无法通过浏览器登录的 CLI 工具或智能电视，使用设备码流程。隐式模式已弃用，不应使用。

2. 向提供商注册应用： 在提供商的开发者控制台（Google、GitHub、Auth0 等）中创建 OAuth 应用。精确配置重定向 URI——URI 不匹配是最常见的设置错误。对于 PKCE 流程，将应用标记为公共客户端。记录客户端 ID、客户端密钥（如适用）、授权端点、令牌端点和作用域。

3. 实现授权请求： 使用必需参数构造授权 URL：client_id、redirect_uri、response_type=code、scope，以及用于防范 CSRF 的加密安全随机 state 参数。对于 PKCE，生成随机的 code_verifier（43-128 个字符），使用 SHA-256 派生 code_challenge，并在请求中加入 code_challenge 和 code_challenge_method=S256。在会话中存储 state 和 code_verifier。

4. 处理回调并交换令牌： 当提供商携带授权码重定向回来时，先验证 state 参数是否与会话中存储的值一致。然后向令牌端点发送 POST 请求，以交换令牌；请求中包含 grant_type=authorization_code、授权码、redirect_uri、client_id，以及 code_verifier（用于 PKCE）。解析响应中的 access_token、refresh_token、expires_in 和 token_type。

5. 安全存储令牌： 切勿将令牌存储在 localStorage 中（存在 XSS 漏洞风险），也不要存储在 URL 参数中（会记录在服务器访问日志中）。Web 应用使用仅限 HTTP 且安全的 Cookie，桌面应用使用系统密钥链，移动应用使用加密存储。尽可能在服务器端存储刷新令牌。记录令牌过期时间戳，以便在过期前主动刷新。

6. 实现令牌刷新和轮换： 每次 API 调用前，检查访问令牌是否已过期或即将过期（60 秒内）。如果是，则使用刷新令牌获取新的访问令牌。处理刷新令牌轮换——当提供商在发放新的访问令牌时同时发放新的刷新令牌，应存储新刷新令牌并使旧令牌失效。如果刷新因 invalid_grant 错误而失败，用户必须重新进行身份验证。

支持的技术

• 提供商： Auth0、Google、GitHub、Microsoft Entra ID、Okta、AWS Cognito、Keycloak
• 服务端框架： Express.js、FastAPI、Django、Spring Security、ASP.NET Core
• 库： passport.js（Node.js）、authlib（Python）、oauthlib（Python）、Spring Security OAuth
• 令牌格式： JWT（自包含）、不透明令牌（需要内省）
• 标准： RFC 6749（OAuth 2.0）、RFC 7636（PKCE）、RFC 8628（设备码）、RFC 9449（DPoP）

用法

向 Agent 提供 OAuth 提供商、应用类型（Web 应用、SPA、CLI、服务器间）和所需作用域。Agent 会选择正确的授权类型，并生成完整实现，包括授权流程、令牌交换、安全存储和刷新逻辑。

示例

示例 1：使用 PKCE 的授权码流程（Node.js/Express）

const express = require("express");
const crypto = require("crypto");
const session = require("express-session");

const app = express();
app.use(session({
  secret: process.env.SESSION_SECRET,
  resave: false,
  saveUninitialized: false,
  cookie: { secure: true, httpOnly: true, sameSite: "lax", maxAge: 3600000 },
}));

const OAUTH_CONFIG = {
  clientId: process.env.OAUTH_CLIENT_ID,
  authorizationEndpoint: "https://accounts.google.com/o/oauth2/v2/auth",
  tokenEndpoint: "https://oauth2.googleapis.com/token",
  redirectUri: "https://myapp.com/auth/callback",
  scopes: ["openid", "email", "profile"],
};

// Generate PKCE code verifier and challenge
function generatePKCE() {
  const verifier = crypto.randomBytes(32).toString("base64url");
  const challenge = crypto
    .createHash("sha256")
    .update(verifier)
    .digest("base64url");
  return { verifier, challenge };
}

// Step 1: Start authorization — redirect user to provider
app.get("/auth/login", (req, res) => {
  const state = crypto.randomBytes(16).toString("hex");
  const { verifier, challenge } = generatePKCE();

  // Store in session for verification on callback
  req.session.oauthState = state;
  req.session.codeVerifier = verifier;

  const params = new URLSearchParams({
    client_id: OAUTH_CONFIG.clientId,
    redirect_uri: OAUTH_CONFIG.redirectUri,
    response_type: "code",
    scope: OAUTH_CONFIG.scopes.join(" "),
    state: state,
    code_challenge: challenge,
    code_challenge_method: "S256",
    access_type: "offline",  // Request refresh token
    prompt: "consent",
  });

  res.redirect(`${OAUTH_CONFIG.authorizationEndpoint}?${params}`);
});

// Step 2: Handle callback — verify state and exchange code for tokens
app.get("/auth/callback", async (req, res) => {
  const { code, state, error } = req.query;

  if (error) {
    console.error(`OAuth error: ${error}`);
    return res.redirect("/auth/error");
  }

  // CSRF protection: verify state matches
  if (state !== req.session.oauthState) {
    console.error("State mismatch — possible CSRF attack");
    return res.status(403).send("Invalid state parameter");
  }

  try {
    const tokenResponse = await fetch(OAUTH_CONFIG.tokenEndpoint, {
      method: "POST",
      headers: { "Content-Type": "application/x-www-form-urlencoded" },
      body: new URLSearchParams({
        grant_type: "authorization_code",
        code: code,
        redirect_uri: OAUTH_CONFIG.redirectUri,
        client_id: OAUTH_CONFIG.clientId,
        code_verifier: req.session.codeVerifier,
      }),
    });

    if (!tokenResponse.ok) {
      const err = await tokenResponse.json();
      throw new Error(`Token exchange failed: ${err.error_description || err.error}`);
    }

    const tokens = await tokenResponse.json();

    // Store tokens securely in session (server-side)
    req.session.accessToken = tokens.access_token;
    req.session.refreshToken = tokens.refresh_token;
    req.session.tokenExpiry = Date.now() + tokens.expires_in * 1000;

    // Clean up PKCE and state from session
    delete req.session.oauthState;
    delete req.session.codeVerifier;

    res.redirect("/dashboard");
  } catch (err) {
    console.error("Token exchange error:", err.message);
    res.redirect("/auth/error");
  }
});

// Token refresh middleware
async function ensureValidToken(req, res, next) {
  if (!req.session.accessToken) {
    return res.redirect("/auth/login");
  }

  // Refresh if token expires within 60 seconds
  if (Date.now() > req.session.tokenExpiry - 60000) {
    try {
      const refreshResponse = await fetch(OAUTH_CONFIG.tokenEndpoint, {
        method: "POST",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: new URLSearchParams({
          grant_type: "refresh_token",
          refresh_token: req.session.refreshToken,
          client_id: OAUTH_CONFIG.clientId,
        }),
      });

      if (!refreshResponse.ok) throw new Error("Refresh failed");

      const tokens = await refreshResponse.json();
      req.session.accessToken = tokens.access_token;
      req.session.tokenExpiry = Date.now() + tokens.expires_in * 1000;
      // Handle refresh token rotation
      if (tokens.refresh_token) {
        req.session.refreshToken = tokens.refresh_token;
      }
    } catch {
      // Refresh token is invalid — user must re-authenticate
      req.session.destroy();
      return res.redirect("/auth/login");
    }
  }
  next();
}

app.get("/dashboard", ensureValidToken, async (req, res) => {
  const profile = await fetch("https://www.googleapis.com/oauth2/v2/userinfo", {
    headers: { Authorization: `Bearer ${req.session.accessToken}` },
  });
  const user = await profile.json();
  res.json({ message: `Welcome, ${user.name}` });
});

app.listen(3000);

示例 2：客户端凭据流程（机器对机器）

import os
import time
import threading
import requests

class ClientCredentialsAuth:
    """OAuth 2.0 client credentials flow for server-to-server auth.

    Manages token lifecycle with thread-safe caching and automatic refresh.
    """

    def __init__(self, token_endpoint: str, client_id: str, client_secret: str,
                 scopes: list[str] | None = None):
        self.token_endpoint = token_endpoint
        self.client_id = client_id
        self.client_secret = client_secret
        self.scopes = scopes or []
        self._access_token: str | None = None
        self._token_expiry: float = 0
        self._lock = threading.Lock()

    def get_access_token(self) -> str:
        """Get a valid access token, refreshing if necessary."""
        with self._lock:
            if self._access_token and time.time() < self._token_expiry - 60:
                return self._access_token
            return self._fetch_new_token()

    def _fetch_new_token(self) -> str:
        response = requests.post(
            self.token_endpoint,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "scope": " ".join(self.scopes),
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10,
        )
        response.raise_for_status()
        token_data = response.json()

        self._access_token = token_data["access_token"]
        self._token_expiry = time.time() + token_data.get("expires_in", 3600)
        return self._access_token

    def authorized_request(self, method: str, url: str, **kwargs) -> requests.Response:
        """Make an HTTP request with automatic bearer token injection."""
        token = self.get_access_token()
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {token}"
        return requests.request(method, url, headers=headers, **kwargs)


# Usage
auth = ClientCredentialsAuth(
    token_endpoint="https://auth.example.com/oauth/token",
    client_id=os.environ["M2M_CLIENT_ID"],
    client_secret=os.environ["M2M_CLIENT_SECRET"],
    scopes=["read:data", "write:data"],
)

# Token is fetched automatically and cached
response = auth.authorized_request("GET", "https://api.example.com/v1/reports")
print(response.json())

# Subsequent calls reuse the cached token until it nears expiry
response = auth.authorized_request("POST", "https://api.example.com/v1/exports", json={
    "format": "csv",
    "date_range": "2024-01-01/2024-12-31",
})
print(f"Export started: {response.json()['export_id']}")

最佳实践

• 始终在授权码流程中使用 PKCE，即使是机密客户端也不例外。它几乎不增加任何成本，却能提供纵深防御，防止授权码被截获。隐式流程已弃用，请勿使用。
• 在处理授权码之前，务必在每次回调中验证 state 参数。这是防御 CSRF 攻击的主要手段。请使用至少 128 位的加密安全随机值。
• 尽可能在服务端存储令牌。对于 Web 应用，请将令牌保存在服务端会话中，并使用 HTTP-only、secure、SameSite Cookie 存储会话 ID。切勿向客户端 JavaScript 暴露访问令牌。
• 实施刷新令牌轮换，确保每个刷新令牌只能使用一次。收到令牌响应中的新刷新令牌后，立即替换旧令牌。如果轮换后的令牌被重复使用，请将其视为已泄露，并撤销该会话的所有令牌。
• 仅请求最低限度的作用域，并在条件允许时使用增量授权。先采用最小权限，仅在用户执行需要额外权限的操作时请求更多作用域。
• 设置较短的访问令牌有效期（敏感 API 为 5-15 分钟），并依靠刷新令牌延长使用期限。如果访问令牌泄露，这可以缩短造成损失的时间窗口。

边缘情况

• **刷新令牌过期：**刷新令牌可能过期或被撤销。如果刷新请求返回 invalid_grant，请销毁会话并将用户重定向到重新进行身份验证的流程。不要无限重试。
• **并发刷新请求：**如果多个线程同时尝试刷新令牌，应仅由其中一个线程执行刷新，其他线程等待。使用互斥锁或 single-flight 模式，防止多个刷新请求相互竞争并使彼此的令牌失效。
• **提供方服务中断：**如果令牌端点无法访问，只要当前访问令牌尚未过期，就继续使用它。使用指数退避机制排队重试令牌刷新，而不是立即失败。
• **时钟偏差：**令牌过期检查依赖准确的时钟。检查过期时间时增加一个缓冲时间（30-60 秒），以应对你的服务端与 OAuth 提供方之间的时钟差异。
• **重定向 URI 不匹配：**OAuth 提供方要求重定向 URI 完全匹配。请确保注册的 URI 完全一致，包括协议、端口和末尾斜杠。不匹配会导致难以排查的静默失败。
• **令牌泄露到日志中：**访问令牌出现在日志、错误消息或 URL 中是常见的安全漏洞。请在所有日志中间件中对令牌进行脱敏。切勿将令牌作为 URL 查询参数传递——只能使用 Authorization 标头。