【FastAPI】基于 PasswordBearer Token 的 OAuth2 认证

什么是 OAuth2 认证

规范：共同准守的规范，需要学习这种规范、流程

OAuth2是一个规范，它定义了几种处理身份认证和授权的方法。

基本的规范结构

不标准的代码，只是讲解规范

需要校验 token 的接口，把 OAuth2PasswordBearer 实例化对象作为依赖就好。

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
"""
1、OAuth2PasswordBearer 只是接收 URL 作为参数的一个类，\
    客户端会向该 URL 发送用于请求 token 的参数（username、password ...）
2、OAuth2PasswordBearer 不会创建相应的 URL 路径操作，\
    只是指明客户端用来请求 Token 的 URL 地址
3、把 OAuth2PasswordBearer 的实例作为 依赖，\
    FastAPI 会检查请求的 Authorization头信息，如果没有找到会返回 401状态码（UNAUTHORIZED）
"""

# 实例化 OAuth2PasswordBearer 指明请求 Token 地址
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token') # http://127.0.0.1:8000/token

# 请求 token 时具体的处理
@app.post('/token')  # 对应上面的 OAuth2PasswordBearer 实例化参数
async def get_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """ 这里使用了 OAuth2 规范的表单类 """
    """ OAuth2 规范的 Token 返回结构 """
    return {'access_token': user.username, 'token_type': 'bearer'}

# 使用 oauth2_scheme 作为依赖，会自动校验 请求的 Authorization头信息
async def get_user_info(user_token: str = Depends(oauth2_scheme)):
    return user_token

# 使用 get_user_info 作为依赖，直接请求会返回 401状态码（UNAUTHORIZED） 
@app.post('/user/me')
async def read_user_me(user: str = Depends(get_user_info)):
    return user

实例

项目中不会这样使用（便于理解），项目更多的是使用 Json Web Token 认证模式

import uvicorn
from pydantic import BaseModel
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

fake_user_db = {  # 模拟用户数据库
    'Bob': {'username': 'Bob',
            'password': '123456'},
    'Mary': {'username': 'Mary',
             'password': '987654'}
}

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

# 实例化 OAuth2PasswordBearer 类，指明请求 token 接口地址
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

@app.post('/token')  # 获取 token 的接口具体方法
async def get_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_user_db.get(form_data.username)  # 数据库获取用户
    if not user_dict:
        # 检查 用户 是否存在
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail='Incorrect username or password')
    user = User(**user_dict)
    if form_data.password != user.password:
        # 检查 密码 是否正确
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail='Incorrect username or password')
    """
    access_token 是返回的 token （这里是为了方便，所以使用 username 作用 token）
    token_type 是对应 实例化 OAuth2Password'Bearer'
    """
    return {'access_token': user.username, 'token_type': 'bearer'}

# 依赖 oauth2_scheme 自动校验 Authorization 头信息 
async def get_user_info(user_token: str = Depends(oauth2_scheme)):
    return User(**fake_user_db[user_token])

# 获取 token 的用户详细信息
@app.post('/user/me')
async def read_user_me(user: User = Depends(get_user_info)):
    return user

if __name__ == '__main__':
    uvicorn.run(app)