OAuth 2.0 授权流程详解与 FastAPI 实现

在现代网络应用中，OAuth 2.0 已成为授权和认证的标准协议。它允许用户将访问权限授予第三方应用，而无需暴露自己的用户名和密码。本文将详细介绍 OAuth 2.0 的常见授权流程，并展示如何在 FastAPI 中实现这些流程。
在这里插入图片描述

OAuth 2.0 简介

OAuth 2.0 是一个授权框架，它定义了多种授权流程，允许用户将权限授予第三方应用。这些流程包括：

授权码流程：适用于 Web 应用，通过授权码交换访问令牌。
密码凭据流程：适用于能够安全存储用户凭证的应用。
客户端凭据流程：适用于服务器到服务器的通信。
简化授权流程：适用于纯前端应用，如 JavaScript 应用。
设备授权流程：适用于输入受限的设备，如智能电视。

OAuth 2.0 授权流程详解

1. 授权码流程

适用场景：Web 应用。

流程：

用户访问客户端应用并请求授权。
客户端重定向用户到授权服务器。
用户登录并授权客户端应用。
授权服务器返回授权码给客户端。
客户端使用授权码请求访问令牌。

2. 密码凭据流程

适用场景：命令行工具、移动应用。

流程：

用户向客户端提供用户名和密码。
客户端使用这些凭证直接请求访问令牌。

3. 客户端凭据流程

适用场景：服务器到服务器的通信。

流程：

客户端直接向授权服务器进行身份验证。
授权服务器返回访问令牌给客户端。

4. 简化授权流程

适用场景：纯前端应用。

流程：

用户被重定向到授权服务器进行认证。
授权服务器直接返回访问令牌给客户端。

5. 设备授权流程

适用场景：输入受限的设备。

流程：

设备请求授权码。
用户在另一个设备上输入授权码进行授权。
设备使用授权码请求访问令牌。

FastAPI 中的 OAuth 2.0 实现

FastAPI 提供了与 OAuth 2.0 无缝集成的支持。以下是如何在 FastAPI 中实现 OAuth 2.0 授权流程的示例。

1. 环境准备

首先，安装必要的库：

pip install fastapi[all] python-jose[cryptography] passlib[bcrypt]

2. 代码实现

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional

app = FastAPI()

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    password: str

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    }
}

def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user:
        return False
    if user["disabled"]:
        return False
    if password != user["hashed_password"]:
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/items/")
async def get_items(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return {"username": user["username"], "items": ["item1", "item2"]}

@app.get("/token/fake")
async def fake_token():
    return create_access_token(data={"sub": "fake_user"})

3. 测试

使用 curl 测试 FastAPI 应用：

# 获取访问令牌
curl -X POST http://localhost:8000/token -d "username=johndoe&password=fakehashedsecret" -s

# 使用访问令牌访问受保护的端点
curl http://localhost:8000/items/ -H "Authorization: Bearer <ACCESS_TOKEN>"

替换 <ACCESS_TOKEN> 为你从 /token 端点获得的实际令牌。