用户 → 客户端 → 授权服务器 → 用户授权 → 授权码 → 客户端后端 → 访问令牌

# 授权服务器实现
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import RedirectResponse
import secrets
import hashlib

app = FastAPI()

# 存储授权码（生产环境用 Redis）
auth_codes = {}
access_tokens = {}
refresh_tokens = {}

CLIENTS = {
    "client_123": {
        "secret": "secret_456",
        "redirect_uri": "https://app.example.com/callback",
        "scopes": ["read", "write"]
    }
}

@app.get("/authorize")
async def authorize(
    response_type: str,
    client_id: str,
    redirect_uri: str,
    scope: str,
    state: str
):
    # 验证客户端
    if client_id not in CLIENTS:
        raise HTTPException(400, "Invalid client")
    
    client = CLIENTS[client_id]
    if redirect_uri != client["redirect_uri"]:
        raise HTTPException(400, "Invalid redirect_uri")
    
    # 显示授权页面（简化）
    # 用户点击"授权"后：
    code = secrets.token_urlsafe(32)
    auth_codes[code] = {
        "client_id": client_id,
        "user_id": "user_001",
        "scope": scope,
        "expires_at": time.time() + 600  # 10 分钟
    }
    
    return RedirectResponse(
        f"{redirect_uri}?code={code}&state={state}"
    )

@app.post("/token")
async def token(
    grant_type: str,
    code: str = None,
    client_id: str = None,
    client_secret: str = None,
    redirect_uri: str = None
):
    if grant_type != "authorization_code":
        raise HTTPException(400, "Unsupported grant_type")
    
    # 验证客户端凭据
    if client_id not in CLIENTS:
        raise HTTPException(401, "Invalid client")
    if CLIENTS[client_id]["secret"] != client_secret:
        raise HTTPException(401, "Invalid client_secret")
    
    # 验证授权码
    if code not in auth_codes:
        raise HTTPException(400, "Invalid code")
    
    auth_code = auth_codes.pop(code)  # 一次性使用
    if auth_code["client_id"] != client_id:
        raise HTTPException(400, "Code mismatch")
    if auth_code["expires_at"] < time.time():
        raise HTTPException(400, "Code expired")
    
    # 生成令牌
    access_token = secrets.token_urlsafe(32)
    refresh_token = secrets.token_urlsafe(32)
    
    access_tokens[access_token] = {
        "user_id": auth_code["user_id"],
        "scope": auth_code["scope"],
        "expires_at": time.time() + 3600
    }
    
    refresh_tokens[refresh_token] = {
        "user_id": auth_code["user_id"],
        "client_id": client_id,
        "scope": auth_code["scope"]
    }
    
    return {
        "access_token": access_token,
        "token_type": "Bearer",
        "expires_in": 3600,
        "refresh_token": refresh_token,
        "scope": auth_code["scope"]
    }

import httpx

class OAuthClient:
    def __init__(self, client_id: str, client_secret: str, redirect_uri: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
        self.auth_server = "https://auth.example.com"
    
    def get_authorization_url(self, state: str) -> str:
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": "read write",
            "state": state
        }
        return f"{self.auth_server}/authorize?" + "&".join(f"{k}={v}" for k, v in params.items())
    
    async def exchange_code(self, code: str) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.auth_server}/token",
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "redirect_uri": self.redirect_uri
                }
            )
            return response.json()
    
    async def refresh_access_token(self, refresh_token: str) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.auth_server}/token",
                data={
                    "grant_type": "refresh_token",
                    "refresh_token": refresh_token,
                    "client_id": self.client_id,
                    "client_secret": self.client_secret
                }
            )
            return response.json()

import base64
import hashlib
import secrets

def generate_pkce():
    """生成 PKCE 参数"""
    # 生成 code_verifier（43-128 字符）
    code_verifier = secrets.token_urlsafe(64)[:128]
    
    # 生成 code_challenge
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).rstrip(b'=').decode()
    
    return code_verifier, code_challenge

# 客户端使用
code_verifier, code_challenge = generate_pkce()

# 授权请求（带 code_challenge）
auth_url = f"""
{auth_server}/authorize?
    response_type=code&
    client_id={client_id}&
    redirect_uri={redirect_uri}&
    code_challenge={code_challenge}&
    code_challenge_method=S256&
    state={state}
"""

# 令牌交换（带 code_verifier）
async def exchange_code_with_pkce(code: str, code_verifier: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{auth_server}/token",
            data={
                "grant_type": "authorization_code",
                "code": code,
                "client_id": client_id,
                "redirect_uri": redirect_uri,
                "code_verifier": code_verifier
                # 注意：不需要 client_secret
            }
        )
        return response.json()

@app.post("/token")
async def token_with_pkce(
    grant_type: str,
    code: str = None,
    client_id: str = None,
    code_verifier: str = None,
    redirect_uri: str = None
):
    # ... 验证授权码 ...
    
    auth_code = auth_codes[code]
    
    # 验证 PKCE
    if "code_challenge" in auth_code:
        expected_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).rstrip(b'=').decode()
        
        if expected_challenge != auth_code["code_challenge"]:
            raise HTTPException(400, "Invalid code_verifier")
    
    # ... 生成令牌 ...

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"

def create_jwt_token(user_id: str, scope: str) -> str:
    payload = {
        "sub": user_id,
        "scope": scope,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=1),
        "iss": "auth.example.com"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_jwt_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

# 资源服务器验证
from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    payload = verify_jwt_token(credentials.credentials)
    return {"user_id": payload["sub"], "scope": payload["scope"]}

@app.get("/api/resource")
async def protected_resource(user: dict = Depends(get_current_user)):
    if "read" not in user["scope"]:
        raise HTTPException(403, "Insufficient scope")
    return {"data": "protected content"}