FastAPI实现OAuth2认证的完整指南-尧图网络科技

1. FastAPI与OAuth2认证基础

在构建现代Web应用时，认证系统是保护API安全的第一道防线。FastAPI作为高性能的Python框架，提供了对OAuth2协议的优雅支持。OAuth2是目前最流行的授权框架，被Google、GitHub等大型平台广泛采用。

OAuth2的核心思想是允许用户授权第三方应用访问其存储在服务提供者上的资源，而无需直接暴露用户名和密码。FastAPI通过内置的security模块简化了OAuth2的实现过程，特别是对密码授权模式(Password Flow)的支持。

重要提示：虽然我们示例中使用的是简化版的密码流，但在生产环境中应始终结合HTTPS使用OAuth2，并考虑添加CSRF保护等额外安全措施。

2. 项目环境搭建与基础配置

2.1 初始化FastAPI应用

首先确保已安装Python 3.7+和FastAPI：

pip install fastapi uvicorn

创建基础应用结构：

from fastapi import FastAPI app = FastAPI() @app.get("/") async def root(): return {"message": "Welcome to OAuth2 Demo"}

2.2 添加安全依赖

FastAPI的安全工具位于fastapi.security模块中：

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

这里OAuth2PasswordBearer是核心类，它：

定义令牌获取URL(tokenUrl)
自动处理Authorization请求头
集成到OpenAPI/Swagger UI中

2.3 用户模型设计

使用Pydantic定义用户模型：

from pydantic import BaseModel from typing import Optional class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None class UserInDB(User): hashed_password: str

3. 实现OAuth2密码流认证

3.1 模拟用户数据库

为演示目的，我们先使用内存数据库：

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

注意：生产环境应使用真实数据库，且密码必须使用bcrypt等安全哈希算法。

3.2 创建令牌端点

from fastapi import Depends, HTTPException, status @app.post("/token") async def login(form_data: OAuth2PasswordRequestForm = Depends()): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password" ) user = UserInDB(**user_dict) if not form_data.password + "fakehashed" == user.hashed_password: raise HTTPException( status_code=400, detail="Incorrect username or password" ) return {"access_token": user.username, "token_type": "bearer"}

3.3 用户认证依赖项

创建获取当前用户的依赖项：

async def get_current_user(token: str = Depends(oauth2_scheme)): user = fake_decode_token(token) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user

4. 保护API端点

4.1 创建受保护路由

@app.get("/users/me") async def read_users_me(current_user: User = Depends(get_current_active_user)): return current_user

4.2 测试认证流程

启动服务：
```
uvicorn main:app --reload
```
访问http://127.0.0.1:8000/docs打开交互文档
点击"Authorize"按钮，输入测试凭据
测试/users/me端点

5. 安全增强实践

5.1 密码哈希处理

使用passlib进行真正的密码哈希：

from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password)

5.2 JWT令牌实现

安装PyJWT：

pip install python-jose[cryptography]

实现JWT令牌：

from jose import JWTError, jwt from datetime import datetime, timedelta SECRET_KEY = "your-secret-key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 def create_access_token(data: dict, expires_delta: 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