1 使用Oauth2Password 进行令牌捕获 设想你在某个域名下拥有一个 后端 API。 并且你在另一个域名或同一域名的不同路径下拥有一个 前端 (或者是一个移动应用程序)。 你希望前端能够使用 用户名 和 密码 向后端进行身份验证。 我们可以使用 OAuth2 来通过 FastAPI 构建这一功能。
1.1 快速使用 1 2 3 4 5 6 7 8 9 10 11 12 from typing import Annotatedfrom fastapi import Depends, FastAPIfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) @app.get("/items/" ) async def read_items (token: Annotated[str , Depends(oauth2_scheme )] ): return {"token" : token}
信息
当你运行 pip install "fastapi[standard]" 命令时, python-multipart 包会自动随 FastAPI 一起安装。
>
但是,如果你使用 pip install fastapi 命令,则默认不会包含 python-multipart 包。
要手动安装它,请确保创建一个 虚拟环境 ,激活它,然后使用以下命令进行安装:
1 $ pip install python-multipart
这是因为 OAuth2 使用“表单数据”来发送 username 和 password 。
授权按钮!
你现在拥有了一个崭新的“授权”按钮。并且你的 路径操作 在右上角有一个小锁图标,你可以点击它。
点击它,会出现一个小的授权表单,用于输入 username 和 password (以及其他可选字段)。
注意
在表单中输入任何内容都没用,因为还没实现后续逻辑。但我们会讲到那一步。
1.2 password 流程 password “流程”是 OAuth2 中定义的一种用于处理安全性和身份验证的方式(“流程”)之一。OAuth2 的设计初衷是让后端或 API 可以独立于对用户进行身份验证的服务器。
但在本例中,同一个 FastAPI 应用程序将同时处理 API 和身份验证。所以,让我们从这个简化的角度回顾一下:
用户在前端输入 username 和 password ,然后按下 Enter 。
前端(在用户的浏览器中运行)将该 username 和 password 发送到我们 API 中的特定 URL(通过 tokenUrl="token" 声明)。
API 检查该 username 和 password ,并返回一个“令牌”(我们还没有实现这些)。
“令牌”只是一个包含某些内容的字符串,我们可以稍后使用它来验证该用户。
通常,令牌会被设置为在一段时间后过期。因此,用户稍后必须重新登录。 如果令牌被盗,风险也较小。它不像永久密钥那样(在大多数情况下)永远有效。
前端会将该令牌临时存储在某处。
用户点击前端进入前端 Web 应用程序的另一个部分。
前端需要从 API 获取更多数据。
但该特定端点需要身份验证。因此,为了与我们的 API 进行身份验证,它会发送一个 Authorization 请求头,其值为 Bearer 加上令牌。
如果令牌内容为 foobar ,则 Authorization 请求头的内容将是: Bearer foobar 。
1.3 FastAPI 的 OAuth2PasswordBearer FastAPI 在不同的抽象层级上提供了多种工具来实现这些安全功能。 在本例中,我们将使用 OAuth2 ,配合 Password 流程和 Bearer 令牌。我们通过 OAuth2PasswordBearer 类来实现这一点。
token_url参数本质为令牌验证的url处理方法的路径
>
但对于我们的用例来说,它是最好的选择。
除非你是 OAuth2 专家,清楚地知道为什么有其他选项更适合你的需求,否则它可能是大多数用例的最佳选择。
在这种情况下, FastAPI 也为你提供了构建其他方案的工具。
当我们创建 OAuth2PasswordBearer 类的实例时,我们会传入 tokenUrl 参数。该参数包含客户端(运行在用户浏览器中的前端)用于发送 username 和 password 以获取令牌的 URL。
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Annotatedfrom fastapi import Depends, FastAPIfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) @app.get("/items/" ) async def read_items (token: Annotated[str , Depends(oauth2_scheme )] ): return {"token" : token}
提示
这里 tokenUrl="token" 指的是一个相对 URL token ,我们还没创建它 。作为相对 URL,它等同于 ./token 。
>
因为我们使用的是相对 URL,如果你的 API 位于 https://example.com/ ,那么它将指向 https://example.com/token 。但如果你的 API 位于 https://example.com/api/v1/ ,那么它将指向 https://example.com/api/v1/token 。
使用相对 URL 很重要,这可以确保你的应用程序即使在 代理后 (Behind a Proxy) 这样的高级用例中也能正常工作。
此参数不会自动创建该端点 / 路径操作 ,但它声明了 /token URL 是客户端应该用来获取令牌的地址。该信息在 OpenAPI 以及随后的交互式 API 文档系统中会被使用。
我们很快也会创建实际的路径操作。
oauth2_scheme 变量是 OAuth2PasswordBearer 的一个实例,但它也是一个“可调用对象 ”因为重载了__call__方法。
1 oauth2_scheme(some, parameters)
现在你可以通过 Depends 将该 oauth2_scheme 用作依赖项。
1 2 3 4 5 6 7 8 9 10 11 12 from typing import Annotatedfrom fastapi import Depends, FastAPIfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) @app.get("/items/" ) async def read_items (token: Annotated[str , Depends(oauth2_scheme )] ): return {"token" : token}
此依赖项将提供一个 str 类型的值,并将其赋值给 路径操作函数 的 token 参数。
FastAPI 将知道它可以利用此依赖项在 OpenAPI 模式(以及自动 API 文档)中定义一个“安全方案”。
技术细节
FastAPI 知道可以使用 OAuth2PasswordBearer 类(在依赖项中声明)在 OpenAPI 中定义安全方案,因为它继承自 fastapi.security.oauth2.OAuth2 ,而后者又继承自 fastapi.security.base.SecurityBase 。
>
所有与 OpenAPI 集成(以及自动 API 文档)的安全工具都继承自 SecurityBase ,这就是 FastAPI 知道如何在 OpenAPI 中集成它们的方式。
它的作用 它会去请求中查找那个 Authorization 请求头,检查值是否为 Bearer 加上某个令牌,并以 str 的形式返回该令牌。
如果它没发现 Authorization 请求头,或者值中没有 Bearer 令牌,它会直接返回一个 401 状态码错误( UNAUTHORIZED )。
你甚至不需要检查令牌是否存在就返回错误。你可以确信,一旦你的函数被执行,该令牌中一定包含一个 str 。
你现在就可以在交互式文档中进行尝试了。
我们还没验证令牌的有效性,但这是一个良好的开端。
2. 通过得到的令牌解码出用户信息 2.1 解码令牌并得到用户信息,通过用户模型返回前端 首先,创建 Pydantic 用户模型。与使用 Pydantic 声明请求体相同,并且可在任何位置使用:
创建pydantic模型 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 from typing import Annotatedfrom fastapi import Depends, FastAPIfrom fastapi.security import OAuth2PasswordBearerfrom pydantic import BaseModelapp = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) class User (BaseModel ): username: str email: str | None = None full_name: str | None = None disabled: bool | None = None def fake_decode_token (token ): return User( username=token + "fakedecoded" , email="john@example.com" , full_name="John Doe" ) async def get_current_user (token: Annotated[str , Depends(oauth2_scheme )] ): user = fake_decode_token(token) return user @app.get("/users/me" ) async def read_users_me (current_user: Annotated[User, Depends(get_current_user )] ): return current_user
2.2 整体流程 1. 从令牌中解码得到用户,通过子依赖项 创建 get_current_user 依赖项。还记得依赖项支持子依赖项吗?get_current_user 使用 oauth2_scheme 作为依赖项。与之前直接在路径操作中的做法相同,新的 get_current_user 依赖项从子依赖项 oauth2_scheme 中接收 str 类型的 token :get_current_user 使用创建的(伪)工具函数,该函数接收 str 类型的令牌,并返回 Pydantic 的 User 模型:
2. 将解码得到的用户信息注入到用户模型中 在 路径操作 的 Depends 中使用 get_current_user 。注意,此处把 current_user 的类型声明为 Pydantic 的 User 模型。
这有助于在函数内部使用代码补全和类型检查。
Fast自动通过Annotated分别请求方法参数类型
还记得请求体也是使用 Pydantic 模型声明的吧。放心,因为使用了 Depends , FastAPI 不会搞混。
>
依赖系统的这种设计方式可以支持不同的依赖项返回同一个 User 模型。而不是局限于只能有一个返回该类型数据的依赖项。
2.1.2 返回的用户模型没有字段限制 接下来,直接在 路径操作函数 中获取当前用户,并用 Depends 在 依赖注入 系统中处理安全机制。**开发者可以使用任何模型或数据满足安全需求(本例中是 Pydantic 的 User 模型)。而且,不局限于只能使用特定的数据模型、类或类型。
不想在模型中使用 username ,而是使用 id 和 email ?当然可以。这些工具也支持。 只想使用字符串?或字典?甚至是数据库类模型的实例?工作方式都一样。 实际上,就算登录应用的不是用户,而是只拥有访问令牌的机器人、程序或其它系统?工作方式也一样。 尽管使用应用所需的任何模型、类、数据库。 FastAPI 通过依赖注入系统都能帮您搞定。
3. 添加/token路由处理方法来实现token验证 本章添加上一章示例中欠缺的部分,实现完整的安全流。
3.1.1 获取 username 和 password 使用 FastAPI 安全工具获取 username 和 password 。
前端表单中必需的两个字段
OAuth2 规范要求使用“密码流”时,客户端或用户必须以表单数据形式发送 username 和 password 字段。 并且,这两个字段必须命名为 username 和 password ,不能使用 user-name 或 email 等其它名称。 该规范要求必须以表单数据形式发送 username 和 password ,因此,不能使用 JSON 对象。
不过也不用担心,前端仍可以显示终端用户所需的名称。数据库模型也可以使用所需的名称。 但对于登录 路径操作 ,则要使用兼容规范的 username 和 password ,(例如,实现与 API 文档集成)。
通过scope字段来配合后端做用户权限管理 OAuth2 还支持客户端发送 scope 表单字段。虽然表单字段的名称是 scope (单数),但实际上,它是以空格分隔的,由多个 scope 组成的长字符串。 常用于声明指定安全权限,例如:
常见用例为, users:read 或 users:write
脸书和 Instagram 使用 instagram_basic
谷歌使用 https://www.googleapis.com/auth/drive
信息
OAuth2 中, 作用域 只是声明指定权限的字符串。
>
是否使用冒号 : 等符号,或是不是 URL 并不重要。这些细节只是针对特定的实现方式权限管理系统的实现方式。对 OAuth2对象 来说,都只是字符串而已。
首先,导入 OAuth2PasswordRequestForm ,然后,在 /token 路径操作 中,用 Depends 把该类作为依赖项。
完整代码 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 from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom pydantic import BaseModelfake_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() def fake_hash_password (password: str ): return "fakehashed" + password oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) class User (BaseModel ): username: str email: str | None = None full_name: str | None = None disabled: bool | None = None class UserInDB (User ): hashed_password: str def get_user (db, username: str ): if username in db: user_dict = db[username] return UserInDB(**user_dict) def fake_decode_token (token ): user = get_user(fake_users_db, token) return user async def get_current_user (token: Annotated[str , Depends(oauth2_scheme )] ): user = fake_decode_token(token) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated" , headers={"WWW-Authenticate" : "Bearer" }, ) return user async def get_current_active_user ( current_user: Annotated[User, Depends(get_current_user )], ): if current_user.disabled: raise HTTPException(status_code=400 , detail="Inactive user" ) return current_user @app.post("/token" ) async def login (form_data: Annotated[OAuth2PasswordRequestForm, Depends( )] ): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) user = UserInDB(**user_dict) hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) return {"access_token" : user.username, "token_type" : "bearer" } @app.get("/users/me" ) async def read_users_me ( current_user: Annotated[User, Depends(get_current_active_user )], ): return current_user
OAuth2PasswordRequestForm 是用以下几项内容声明表单请求体的类依赖项:
username
password
可选的 scope 字段,由多个空格分隔的字符串组成的长字符串
可选的 grant_type
提示
实际上,OAuth2 规范 要求 grant_type 字段使用固定值 password ,但 OAuth2PasswordRequestForm 没有作强制约束。
>
如需强制使用固定值 password ,则不要用 OAuth2PasswordRequestForm ,而是用 OAuth2PasswordRequestFormStrict 。
可选的 client_id (本例未使用)
可选的 client_secret (本例未使用)
信息
OAuth2PasswordRequestForm 并不像 OAuth2PasswordBearer 那样是 FastAPI 的特殊类。
>
FastAPI 把 OAuth2PasswordBearer 识别为安全方案。因此,可以通过这种方式把它添加至 OpenAPI。
但 OAuth2PasswordRequestForm 只是可以自行编写的类依赖项,也可以直接声明 Form 参数。
但由于这种用例很常见,FastAPI 为了简便,就直接提供了对它的支持。
3.1.3 主要流程 1. 使用表单数据并和数据库中的用户信息进行比对
提示
OAuth2PasswordRequestForm 类依赖项的实例没有以空格分隔的长字符串属性 scope ,但它支持 scopes 属性,由已发送的 scope 字符串列表组成。
>
本例没有使用 scopes ,但开发者也可以根据需要使用该属性。
现在,即可使用表单字段 username ,从(伪)数据库中获取用户数据。如果不存在指定用户,则返回错误消息,提示 用户名或密码错误 。本例使用 HTTPException 异常显示此错误:
1 2 3 4 5 6 7 8 9 10 11 12 13 @app.post("/token" ) async def login (form_data: Annotated[OAuth2PasswordRequestForm, Depends( )] ): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) user = UserInDB(**user_dict) hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) return {"access_token" : user.username, "token_type" : "bearer" }
至此,我们已经从数据库中获取了用户数据,但尚未校验密码。接下来,首先将数据放入 Pydantic 的 UserInDB 模型。
注意
永远不要保存明文密码,本例暂时先使用(伪)哈希密码系统。如果密码不匹配,则返回与上面相同的错误。
##### 密码哈希
为什么使用密码哈希 哈希是指,将指定内容(本例中为密码)转换为形似乱码的字节序列(其实就是字符串)。
每次传入完全相同的内容(比如,完全相同的密码)时,得到的都是完全相同的乱码。
但这个乱码无法转换回传入的密码。
原因很简单,假如数据库被盗,窃贼无法获取用户的明文密码,得到的只是哈希值。
这样一来,窃贼就无法在其它应用中使用窃取的密码,要知道,很多用户在所有系统中都使用相同的密码,风险超大。
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 from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom pydantic import BaseModelfake_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() def fake_hash_password (password: str ): return "fakehashed" + password oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token" ) @app.post("/token" ) async def login (form_data: Annotated[OAuth2PasswordRequestForm, Depends( )] ): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) user = UserInDB(**user_dict) hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400 , detail="Incorrect username or password" ) return {"access_token" : user.username, "token_type" : "bearer" } @app.get("/users/me" ) async def read_users_me ( current_user: Annotated[User, Depends(get_current_active_user )], ): return current_user
2. 返回 Token token 端点的响应必须是 JSON 对象。 响应返回的内容应该包含 token_type 。本例中用的是 Bearer Token,因此, Token 类型应为 bearer 。 返回内容还应包含 access_token 字段,它是包含权限 Token 的字符串。
本例只是简单的演示,返回的 Token 就是 username ,但这种方式极不安全。
提示
下一章介绍使用哈希密码和 JWT Token 的真正安全机制。
>
但现在,仅关注所需的特定细节。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from typing import Annotated from fastapi import Depends , FastAPI , HTTPException , statusfrom fastapi.security import OAuth2PasswordBearer , OAuth2PasswordRequestForm from pydantic import BaseModel @app.post ("/token" ) async def login (form_data : Annotated [OAuth2PasswordRequestForm , Depends ()]): user_dict = fake_users_db.get (form_data.username ) if not user_dict : raise HTTPException (status_code=400 , detail="Incorrect username or password" ) user = UserInDB (**user_dict) hashed_password = fake_hash_password (form_data.password ) if not hashed_password == user.hashed_password : raise HTTPException (status_code=400 , detail="Incorrect username or password" ) return {"access_token" : user.username , "token_type" : "bearer" }
提示
按规范的要求,应像本示例一样,返回带有 access_token 和 token_type 的 JSON 对象。
>
这是开发者必须在代码中自行完成的工作,并且要确保使用这些 JSON 的键。
这几乎是唯一需要开发者牢记在心,并按规范要求正确执行的事。
FastAPI 则负责处理其它的工作。
3. 更新依赖项 接下来,更新依赖项。
使之仅在当前用户为激活状态时,才能获取 current_user 。
为此,要再创建一个依赖项 get_current_active_user ,此依赖项以 get_current_user 依赖项为基础。
如果用户不存在,或状态为未激活,这两个依赖项都会返回 HTTP 错误。
因此,在端点中,只有当用户存在、通过身份验证、且状态为激活时,才能获得该用户:
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 from typing import Annotated from fastapi import Depends , FastAPI , HTTPException , statusfrom 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 () def fake_hash_password (password : str): return "fakehashed" + password oauth2_scheme = OAuth2PasswordBearer (tokenUrl="token" ) class User (BaseModel ): username : str email : str | None = None full_name : str | None = None disabled : bool | None = None class UserInDB (User ): hashed_password : str def get_user (db, username : str): if username in db : user_dict = db[username] return UserInDB (**user_dict) def fake_decode_token (token): # This doesn't provide any security at all # Check the next version user = get_user(fake_users_db, token) return user async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): user = fake_decode_token(token) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)], ): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user @app.post("/token") async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]): user_dict = fake_users_db.get(form_data.username) if not user_dict: raise HTTPException(status_code=400, detail="Incorrect username or password") user = UserInDB(**user_dict) hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400, detail="Incorrect username or password") return {"access_token": user.username, "token_type": "bearer"} @app.get("/users/me") async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)], ): return current_user
信息
此处返回值为 Bearer 的响应头 WWW-Authenticate 也是规范的一部分。
>
任何 401“UNAUTHORIZED”HTTP(错误)状态码都应返回 WWW-Authenticate 响应头。
本例中,因为使用的是 Bearer Token,该响应头的值应为 Bearer 。
实际上,忽略这个附加响应头,也不会有什么问题。
之所以在此提供这个附加响应头,是为了符合规范的要求。
说不定什么时候,就有工具用得上它,而且,开发者或用户也可能用得上。
这就是遵循标准的好处…
实际效果 点击“Authorize”按钮。 使用以下凭证: 用户名: johndoe 密码: secret
通过身份验证后,显示下图所示的内容:
获取当前用户数据 使用 /users/me 路径的 GET 操作。 可以提取如下当前用户数据:
1 2 3 4 5 6 7 { "username" : "johndoe" , "email" : "johndoe@example.com" , "full_name" : "John Doe" , "disabled" : false , "hashed_password" : "fakehashedsecret" }
点击小锁图标,注销后,再执行同样的操作,则会得到 HTTP 401 错误:
1 2 3 { "detail" : "Not authenticated" }
未激活用户 测试未激活用户,输入以下信息,进行身份验证: 用户名: alice 密码: secret2 然后,执行 /users/me 路径的 GET 操作。
显示下列 未激活用户 错误信息:
1 2 3 4 { "detail" : "Inactive user" }