FastAPI（58）- 使用 OAuth2PasswordBearer 的简单栗子

小菠萝测试笔记

发布于 2021-10-09 10:56:37

2.6K0

发布于 2021-10-09 10:56:37

文章被收录于专栏：自动化、性能测试

背景

假设在某个域中拥有后端 API（127.0.0.1:8080）
并且在另一个域或同一域的不同路径（或移动应用程序）中有一个前端（127.0.0.1:8081）
并且希望有一种方法让前端使用用户名和密码与后端进行身份验证
可以使用 OAuth2 通过 FastAPI 来构建它，通过 FastAPI 提供的工具来处理安全性

OAuth2 的授权模式

授权码授权模式 Authorization Code Grant
隐式授权模式 Implicit Grant
密码授权模式 Resource Owner Password Credentials Grant
客户端凭证授权模式 Client Credentials Grant

这里讲 FastAPI 的是第三种

密码授权模式的简易流程图

用户在客户端输入用户名、密码
客户端携带用户名、密码去请求授权服务器，访问获取 token 的接口
授权服务器验证用户名、密码（身份验证）
验证通过后，返回这个用户的 token 到客户端
客户端存储 token，在后续发送请求携带该 token，就能通过身份验证了

FastAPI 中使用 OAuth2 的简单栗子

import uvicorn
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

if __name__ == '__main__':
    uvicorn.run(app="49_bearer:app", reload=True, host="127.0.0.1", port=8080)

代码解析

OAuth2 旨在使后端或 API 可以独立于对用户进行身份验证的服务器
但在这种情况下，同一个 FastAPI 应用程序将同时处理 API 和身份验证
前端请求 /items 的之前要先进行身份验证，也就是用户名和密码，这个验证的路径就是 tokenUrl，是相对路径，POST请求
oauth2_scheme 中接收一个 str 类型的 token，就是当验证通过后，要返回给客户端的一个令牌（常说的 token）
方便下次请求携带这个 token 就可以通过身份认证，这个 token 有过期时间，过期后需要重新验证

OAuth2PasswordBearer

使用 OAuth2、密码授权模式、Bearer Token（不记名 token），就是通过 OAuth2PasswordBearer 来完成
OAuth2PasswordBearer 是接收 URL 作为参数的一个类
客户端会向该 URL 发送 username 和 password 参数（通过表单的格式发送），然后得到一个 token 值
OAuth2PasswordBearer 并不会创建相应的 URL 路径操作，只是指明了客户端用来获取 token 的目标 URL

tokenUrl 是相对路径

如果 API 位于 https://example.com/，那么它将引用 https://example.com/token
如果API 位于 https://example.com/api/v1/，那么它将引用 https://example.com/api/v1/token

oauth2_scheme

该变量是 OAuth2PasswordBearer 的一个实例，但它也是一个可调用对象，所以它可以用于依赖项

async def read_items(token: str = Depends(oauth2_scheme)):

OAuth2PasswordBearer 会做什么

客户端发送请求的时候，FastAPI 会检查请求的 Authorization 头信息，如果没有找到 Authorization 头信息
或者头信息的内容不是 Bearer token，它会返回 401 状态码( UNAUTHORIZED )

传递 token 的请求结果

目前因为没有对 token 做验证，所以 token 传什么值都可以验证通过

看看 OAuth2PasswordBearer 的源码

查看 Swagger API 文档

多了个 Authorize 按钮，点击它

可以看到一个包含用户名、密码还有其他可选字段的授权表单

上述代码的问题

还没有获取 token 的路径操作

完善 OAuth2

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠萝测试笔记
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/10/6 12:05 下午
# file: 49_bearer.py
"""
from typing import Optional

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

# 模拟数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


# 模拟 hash 加密算法
def fake_hash_password(password: str) -> str:
    return "fakehashed" + password


# 返回给客户端的 User Model，不需要包含密码
class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


# 继承 User，用于密码验证，所以要包含密码
class UserInDB(User):
    hashed_password: str


# OAuth2 获取 token 的请求路径
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 1、获取客户端传过来的用户名、密码
    username = form_data.username
    password = form_data.password
    # 2、模拟从数据库中根据用户名查找对应的用户
    user_dict = fake_users_db.get(username)
    if not user_dict:
        # 3、若没有找到用户则返回错误码
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码不正确")

    # 4、找到用户
    user = UserInDB(**user_dict)
    # 5、将传进来的密码模拟 hash 加密
    hashed_password = fake_hash_password(password)
    # 6、如果 hash 后的密码和数据库中存储的密码不相等，则返回错误码
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码不正确")

    # 7、用户名、密码验证通过后，返回一个 JSON
    return {"access_token": user.username, "token_type": "bearer"}


# 模拟从数据库中根据用户名查找用户
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


# 模拟验证 token，验证通过则返回对应的用户信息
def fake_decode_token(token):
    user = get_user(fake_users_db, token)
    return user


# 根据当前用户的 token 获取用户，token 已失效则返回错误码
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(user: User = Depends(get_current_user)):
    if user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid User")
    return user


# 获取当前用户信息
@app.get("/user/me")
async def read_user(user: User = Depends(get_current_active_user)):
    return user


# 正常的请求
@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}


if __name__ == '__main__':
    uvicorn.run(app="49_bearer:app", reload=True, host="127.0.0.1", port=8080)

/token 路径操作函数的响应

# 7、用户名、密码验证通过后，返回一个 JSON
return {"access_token": user.username, "token_type": "bearer"}

获取 token 的接口的响应必须是一个 JSON 对象（返回一个 dict 即可）
它应该有一个 token_type，当使用 Bearer toklen 时，令牌类型应该是 bearer
它应该有一个 access_token，一个包含访问 token 的字符串
对于上面简单的例子，返回的 token 是用户名，这是不安全，只是作为栗子好理解一点

返回 401 的HTTPException

# 根据当前用户的 token 获取用户，token 已失效则返回错误码
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