FastAPI 中的 OAuth2PasswordBearer 授权方法详解

FastAPI 中的 OAuth2PasswordBearer 授权方法详解

官方文档：https://fastapi.tiangolo.com/zh/tutorial/security/first-steps/#fastapi-oauth2passwordbearer

FastAPI管理后台框架：https://gitee.com/ktianc/kinit

FastAPI 中的 OAuth2PasswordBearer 授权方法主要是为 FastAPI 应用提供密码流授权方案。

OAuth2 其他授权模式

OAuth2 定义了多种授权方式，用于实现不同场景下的身份验证和授权需求。以下是 OAuth2 中常见的认证方式：

1. 授权码（Authorization Code）: 这是最常见的 OAuth2 授权方式。它适用于 Web 应用和服务器端应用。在这种模式下，用户在客户端应用中点击登录，被重定向到授权服务器，在授权服务器登录后，将授权码传递回客户端应用，然后客户端应用使用授权码从授权服务器获取访问令牌。
2. 密码（Resource Owner Password Credentials）: 这种方式允许用户将自己的用户名和密码直接提供给客户端应用，然后客户端应用使用这些凭据从授权服务器获取访问令牌。这种方式通常在受信任的客户端应用和用户信任的授权服务器之间使用。
3. 隐式（Implicit）: 这种方式适用于纯前端应用，如单页面应用（SPA）。在这种模式下，令牌直接在授权服务器进行重定向时通过 URL 参数传递给前端应用，因此不需要授权码交换步骤。
4. 客户端凭证（Client Credentials）: 这种方式适用于客户端应用之间的认证，而不涉及用户。客户端应用使用自己的凭证（客户端ID和客户端秘钥）从授权服务器获取访问令牌。
5. 刷新令牌（Refresh Token）: 刷新令牌是一种特殊类型的令牌，用于获取新的访问令牌。当访问令牌过期时，客户端可以使用刷新令牌从授权服务器获取新的访问令牌，而无需用户干预。

不同的授权方式适用于不同的应用场景和安全需求。选择合适的授权方式取决于你的应用程序类型、用户交互方式以及安全要求。

OAuth2PasswordBearer 授权方法参数详解

OAuth2PasswordBearer 的定义方法：

class OAuth2PasswordBearer(OAuth2):
    def __init__(
        self,
        tokenUrl: str,
        scheme_name: Optional[str] = None,
        scopes: Optional[Dict[str, str]] = None,
        description: Optional[str] = None,
        auto_error: bool = True,
    ):
        if not scopes:
            scopes = {}
        flows = OAuthFlowsModel(
            password=cast(Any, {"tokenUrl": tokenUrl, "scopes": scopes})
        )
        super().__init__(
            flows=flows,
            scheme_name=scheme_name,
            description=description,
            auto_error=auto_error,
        )

    async def __call__(self, request: Request) -> Optional[str]:
        authorization = request.headers.get("Authorization")
        scheme, param = get_authorization_scheme_param(authorization)
        if not authorization or scheme.lower() != "bearer":
            if self.auto_error:
                raise HTTPException(
                    status_code=HTTP_401_UNAUTHORIZED,
                    detail="Not authenticated",
                    headers={"WWW-Authenticate": "Bearer"},
                )
            else:
                return None
        return param
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

这段代码是 FastAPI 中的 OAuth2PasswordBearer 类的定义代码。该类是用于在 FastAPI 中实现 OAuth2 密码授权方式的身份验证工具。下面详细解释每个参数与方法的含义：

以下每个参数的设置都与 Swagger UI API 文档息息相关。

1. tokenUrl (str)：必需参数。指定密码认证方法的 URL。
2. scheme_name (Optional[str])：可选参数。指定身份验证方案的名称。如果你在应用中使用多种身份验证方式，可以用来区分不同的身份验证方式。
3. scopes (Optional[Dict[str, str]])：可选参数。指定访问令牌的授权范围。OAuth2 密码授权流程可能需要特定的授权作用域，这个参数可以用来限制访问令牌的作用域。
4. description (Optional[str])：可选参数。为身份验证方案提供描述性信息。
5. auto_error (bool)：可选参数，默认为 True。当验证失败时，如果设置为 True，FastAPI 将自动返回一个 401 未授权的响应。如果设置为 False，你需要自己处理身份验证失败的情况。

在 __init__ 方法中，首先检查是否提供了 scopes，如果没有，则默认设置为空字典。然后创建一个包含密码授权流程信息的 OAuthFlowsModel 对象。这个对象指定了密码授权流程的信息，其中包括了 tokenUrl 和 scopes。

接下来，调用父类 OAuth2 的构造函数，将前面创建的 flows、scheme_name、description 和 auto_error 参数传递进去，以创建一个 OAuth2PasswordBearer 实例。

__call__ 方法是用于执行身份验证逻辑的方法。它接受一个 FastAPI 的 Request 对象作为参数，并从请求头中获取授权信息。如果没有提供授权信息或授权方式不是 “bearer”，根据 auto_error 参数的设置，可能会引发一个 401 未授权的异常或者返回 None。

tokenUrl

必填参数，参数类型为字符串，指定后可以在 Swagger UI API 文档中使用该参数的 URL 接口进行认证用户身份。

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

app = FastAPI()

# 指定认证 URL 为 /token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")


@app.post("/token")
async def login(data: OAuth2PasswordRequestForm = Depends()):
    """
    密码认证身份方法
    :param data: 用户通过 form 表单提供的用户名和密码
    :return:
    """
    print(data.username)
    print(data.password)
    return {"access_token": data.username, "token_type": "bearer"}


@app.get("/items")
async def read_items(token: str = Depends(oauth2_scheme)):
    """
    需要通过认证才能访问的接口方法，否则无法访问
    :param token:
    :return:
    """
    return {"token": token}


if __name__ == '__main__':
    # 启动项目
    uvicorn.run(app='main:app', host='127.0.0.1', port=9000)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

在以上代码中，指定了认证 URL 为 /token：

# 指定认证 URL 为 /token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
1
2

启动项目后，打开 Swagger UI API 文档：http://127.0.0.1:9000/docs

以上小锁为开着的状态时说明接口还没有通过认证授权

还没认证授权通过时，请求需要认证授权的接口会报错 401：

这时候就需要认证授权了：

点击 Authorize 按钮后可以看到弹出一个需要认证授权的窗口，并且里面提示认证授权的 URL 地址与我们配置的 tokenUrl 参数一致。

填写好用户名密码（这里可以随便填写，因为我们在上面代码中并没有设定真正的用户名密码）后，就可以点击 Authorize 按钮进行认证了。

以下为通过认证：

通过认证授权后，小锁为关闭的状态：

程序打印：

请求需要认证授权的接口，返回200 请求成功：

一个小提示

当我们给 /token 认证授权接口加上用户名密码验证时，如果在 Swagger UI API 文档中提供的用户名或密码错误是错误的，一定要在返回响应时将 状态码改为 错误的状态码，否则如果状态码为 200，即使没有返回 token，同样会表示认证成功。

auto_error

可选参数，参数类型为 布尔值，默认为 True。当验证失败时，如果设置为 True，FastAPI 将自动返回一个 401 未授权的响应。如果设置为 False，你需要自己处理身份验证失败的情况。

在以上示例中，我们并没有设置 auto_error 参数，所以参数默认为 True

我们在没有通过认证授权的情况下去请求需要认证授权的接口，会直接返回我们401的报错响应。

但是如果我们将 auto_error 设置为 False 后，在没有通过认证授权的情况下再去访问需要认证授权的接口，会返回我们 200 访问成功的响应。

# 指定认证 URL 为 /token
# 指定 auto_error 为 False，不需要自动报错 401
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)
1
2
3

截图中可以看到小锁开着的状态，仍然可以请求成功，所以要谨慎使用 auto_error 参数。

description

可选参数，参数类型为字符串类型，为身份验证方案提供描述性信息。

添加后会 Swagger UI API 文档中认证时显示：

# 指定认证 URL 为 /token
# 指定 auto_error 为 True，错误时自动报错 401
# description：添加认证授权描述信息
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=True, description="这是测试认证")
1
2
3
4

查看 Swagger UI API 文档：

scheme_name

可选参数，参数类型为字符串。指定身份验证方案的名称。如果你在应用中使用多种身份验证方式，可以用来区分不同的身份验证方式。

使用双认证方案的错误示例

如果我们在项目中需要使用多认证方案时，仅仅只是添加了多个 OAuth2PasswordBearer 认证授权实例并没有指定 scheme_name 参数时，这样第一个 OAuth2PasswordBearer 实例是不会生效的，只是 第二个 OAuth2PasswordBearer 会生效。

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.responses import ORJSONResponse
import uvicorn

app = FastAPI()

# 指定认证 URL 为 /token/password
# 指定 auto_error 为 True，错误时自动报错 401
# description：添加认证授权描述信息
password_oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/token/password",
    auto_error=True,
    description="密码认证方式"
)

# 指定认证 URL 为 /token/sms/code
# 指定 auto_error 为 True，错误时自动报错 401
# description：添加认证授权描述信息
sms_code_oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/token/sms/code",
    auto_error=True,
    description="短信验证码认证方式"
)


@app.post("/token/password")
async def password_login(data: OAuth2PasswordRequestForm = Depends()):
    """
    密码认证身份方法
    :param data: 用户通过 form 表单提供的用户名和密码
    :return:
    """
    if data.username != 'kinit' or data.password != 'kinit':
        return ORJSONResponse(status_code=401, content='用户名或密码错误')
    return {"access_token": f"{data.username}-password", "token_type": "bearer"}


@app.post("/token/sms/code")
async def sms_code_login(data: OAuth2PasswordRequestForm = Depends()):
    """
    短信验证码认证身份方法
    :param data: 用户通过 form 表单提供的手机号和短信验证码
    :return:
    """
    if data.username != '15020230821' or data.password != '8888':
        return ORJSONResponse(status_code=401, content='手机号或短信验证码错误')
    return {"access_token": f"{data.username}-sms", "token_type": "bearer"}


@app.get("/items")
async def read_items(token: str = Depends(password_oauth2_scheme)):
    """
    需要通过密码认证才能访问的接口方法，否则无法访问
    :param token:
    :return:
    """
    if token.split("-")[1] != 'password':
        return ORJSONResponse(status_code=401, content='认证信息错误')
    return {"token": token, 'message': '访问成功'}


@app.get("/users")
async def read_users(token: str = Depends(sms_code_oauth2_scheme)):
    """
    需要通过短信验证码认证才能访问的接口方法，否则无法访问
    :param token:
    :return:
    """
    if token.split("-")[1] != 'sms':
        return ORJSONResponse(status_code=401, content='认证信息错误')
    return {"token": token, 'message': '访问成功'}


if __name__ == '__main__':
    # 启动项目
    uvicorn.run(app='main:app', host='127.0.0.1', port=9000)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

在以上代码中我们指定了两个 OAuth2PasswordBearer 认证授权实例分别为：

1. password_oauth2_scheme
2. sms_code_oauth2_scheme

我们这时候打开查看 Swagger UI API 文档中，只会看到 sms_code_oauth2_scheme 认证授权实例：

因为第二个会把第一个给覆盖掉，并且在分别指定的需要授权访问接口也是一样的。

使用双认证方案的正确示例

这时候我们就需要在两个 OAuth2PasswordBearer 认证授权实例分别加上 scheme_name 属性即可：

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.responses import ORJSONResponse
import uvicorn

app = FastAPI()

# 指定认证 URL 为 /token/password
# 指定 auto_error 为 True，错误时自动报错 401
# description：添加认证授权描述信息
password_oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/token/password",
    scheme_name="密码认证授权方式",
    auto_error=True,
    description="密码认证方式"
)

# 指定认证 URL 为 /token/sms/code
# 指定 auto_error 为 True，错误时自动报错 401
# description：添加认证授权描述信息
sms_code_oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/token/sms/code",
    scheme_name="短信验证码认证授权方式",
    auto_error=True,
    description="短信验证码认证方式"
)


@app.post("/token/password")
async def password_login(data: OAuth2PasswordRequestForm = Depends()):
    """
    密码认证身份方法
    :param data: 用户通过 form 表单提供的用户名和密码
    :return:
    """
    if data.username != 'kinit' or data.password != 'kinit':
        return ORJSONResponse(status_code=401, content='用户名或密码错误')
    return {"access_token": f"{data.username}-password", "token_type": "bearer"}


@app.post("/token/sms/code")
async def sms_code_login(data: OAuth2PasswordRequestForm = Depends()):
    """
    短信验证码认证身份方法
    :param data: 用户通过 form 表单提供的手机号和短信验证码
    :return:
    """
    if data.username != '15020230821' or data.password != '8888':
        return ORJSONResponse(status_code=401, content='手机号或短信验证码错误')
    return {"access_token": f"{data.username}-sms", "token_type": "bearer"}


@app.get("/items")
async def read_items(token: str = Depends(password_oauth2_scheme)):
    """
    需要通过密码认证才能访问的接口方法，否则无法访问
    :param token:
    :return:
    """
    if token.split("-")[1] != 'password':
        return ORJSONResponse(status_code=401, content='认证信息错误')
    return {"token": token, 'message': '访问成功'}


@app.get("/users")
async def read_users(token: str = Depends(sms_code_oauth2_scheme)):
    """
    需要通过短信验证码认证才能访问的接口方法，否则无法访问
    :param token:
    :return:
    """
    if token.split("-")[1] != 'sms':
        return ORJSONResponse(status_code=401, content='认证信息错误')
    return {"token": token, 'message': '访问成功'}


if __name__ == '__main__':
    # 启动项目
    uvicorn.run(app='main:app', host='127.0.0.1', port=9000)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

再查看 Swagger UI API 文档，就会看到两个认证授权方式：

1. 密码认证授权方式：

2. 短信验证码认证授权方式：

单个认证：

scopes

可选参数，参数类型为字典，字典的建和值都为字符串。

指定后可以在 Swagger UI API 文档中进行认证用户身份时，选择指定的 scope。

import json
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm, SecurityScopes
from fastapi.responses import ORJSONResponse
import uvicorn
from starlette import status

app = FastAPI()


# 指定认证 URL 为 /token
# scopes：提供两种可选的作用域，用来访问对应作用域的接口
oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/token",
    scopes={"items": "Read items.", "users": "Read users."}
)


@app.post("/token")
async def password_login(data: OAuth2PasswordRequestForm = Depends()):
    """
    密码认证身份方法
    :param data: 用户通过 form 表单提供的用户名和密码和作用域
    :return:
    """
    if data.username != 'kinit' or data.password != 'kinit':
        return ORJSONResponse(status_code=401, content='用户名或密码错误')

    access_token = json.dumps({"sub": data.username, "scopes": data.scopes})
    return {"access_token": access_token, "token_type": "bearer"}


def validation_scope(security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]) -> str:
    """
    解析 token 获取到 token 中的 scopes
    判断接口中需要的作用域是否在 token 存在
    如果存在则返回用户名
    如果不存在则报错

    SecurityScopes的属性scopes，是一个包含所有它需要的scopes以及所有依赖项(把它作为子依赖项)的列表。
    SecurityScopes的属性scope_str，是包含所有scopes的一个字符串(以空格分隔)。
    :param security_scopes: 请求接口必须存在的 scopes
    :param token: token
    :return: 用户名
    """
    # 解析 token
    payload = json.loads(token)
    username: str = payload.get("sub", '')
    if username is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="认证信息错误")
    # token 中的 scopes
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无权限访问的 scope")
    return username


@app.get("/items")
async def read_items(username: Annotated[str, Security(validation_scope, scopes=["items"])]):
    """
    需要存在 items scope 的 token 才能访问该接口方法，否则无法访问

    Security实际上是Depends的子类，只不过多了一个参数，可以接收scopes的列表信息。
    通过使用Security而不是Depends，FastAPI将会知道它会声明并内部使用scopes信息，并且在交互式文档中显示这些信息。
    :param username: 用户名
    :return:
    """
    print(username)
    return {'message': '访问成功', 'username': username}


@app.get("/users")
async def read_users(username: Annotated[str, Security(validation_scope, scopes=["users"])]):
    """
    需要存在 users scope 的 token 才能访问该接口方法，否则无法访问
    :param username:
    :return:
    """
    print(username)
    return {'message': '访问成功', 'username': username}


if __name__ == '__main__':
    # 启动项目
    uvicorn.run(app='main:app', host='127.0.0.1', port=9000)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

打开并查看 Swagger UI API 文档，就会在认证时看到两个我们在 OAuth2PasswordBearer 中配置的 scopes：

选中对应的 scopes 登录后，就会将选中的 scopes 通过 form 表单传入 OAuth2PasswordRequestForm，就可以在登录视图中获取到了。