26.FastAPI安全性
软件开发中,安全是永恒的话题,FastAPI作为一个优秀的Python Web开发框架,为用户提供了多种工具,帮助用户以标准的方式轻松快速地解决软件开发中的安全性。
FastAPI 的 fastapi.security 模块中为各种安全方案提供了一些工具,这些工具简化了这些安全机制的使用方法。
26.1 OAuth2PasswordBearer
FastAPI提供的OAuth2PasswordBearer是使用 OAuth2的密码授权模式的Bearer Token(不记名 token) 。创建OAuth2PasswordBearer 实例需要接收URL作为参数。
客户端会向该 URL 通过表单的格式发送 username 和 password 参数,然后得到一个 token 值;OAuth2PasswordBearer 并不会创建相应的 URL 路径操作,只是指明了客户端用来获取 token 的目标 URL。
代码示例:
# coding: utf-8from fastapi import FastAPIfrom fastapi import Dependsfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI()oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")@app.get(path='/')async def root(token: str = Depends(oauth2_scheme)): return "Hello world"
在上面的代码中, tokenUrl="token"指的token是相对 URL 。
此时访问,其返回结果:
curl http://127.0.0.1:8000 -iHTTP/1.1 401 Unauthorizeddate: Tue, 08 Feb 2022 09:28:05 GMTserver: uvicornwww-authenticate: Bearercontent-length: 30content-type: application/json{"detail":"Not authenticated"}
上面的结果表明:访问的内容以及被保护,必须经过授权后才可以访问。
26.2 OAuth2PasswordRequestForm
OAuth2PasswordRequestForm是一个用于获取用户和密码的请求表单类, OAuth2规定客户端必须将username和password字段作为表单数据发送,不可使用 JSON 。OAuth2PasswordRequestForm声明的请求表单:
- username
- password
- 可选scope字段,是一个由空格分隔的字符串组成的大字符串
- 可选的grant_type字段
- 可选的client_id字段
- 可选的 client_secret字段
当获取到表单数据后,需要进行密码校验,一般情况下,我们都会考虑使用哈希密码,PassLib 是一个用于处理哈希密码的非常好的 Python 包,它支持许多安全哈希算法以及配合算法使用的实用程序。
pip install passlib
具体passlib的使用方法可以查看其文档https://passlib.readthedocs.io/en/stable
下面的代码示例在上面代码的基础上增加用户登录及Token验证
# coding: utf-8from fastapi import FastAPIfrom fastapi import Dependsfrom fastapi import HTTPExceptionfrom fastapi.security import OAuth2PasswordBearerfrom fastapi.security import OAuth2PasswordRequestFormfrom passlib.hash import pbkdf2_sha256app = FastAPI()oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")# 用户列表usrs = [ { 'usr_id': 'u000010001', 'usr_acc': 'abc', 'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$fUI/40Zxj6.62GHgUX9PbDZ0SybccxZwn3Wl3qJ8U/M', # a1b2c3 'salt': '10181010' }, { 'usr_id': 'u000010002', 'usr_acc': 'xyz', 'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$Eu7mZV80f.tu4RDXSsst9gDjV4fXJ.9S7t1hgcGMMVk', # x1y2z3 'salt': '10101018' }]# Token校验async def verify_token(token: str): found = False for usr in usrs: if usr['usr_id'] == token: found = True break if not found: raise HTTPException(status_code=401)# 登录并返回Token@app.post(path='/token')async def login(form: OAuth2PasswordRequestForm = Depends()): foundUsr = None for usr in usrs: if usr['usr_acc'] == form.username: foundUsr = usr if foundUsr is None or pbkdf2_sha256.hash(form.password, salt=usr['salt'].encode()) != foundUsr['usr_pwd']: raise HTTPException(status_code=400, detail="Incorrect username or password") else: return {"access_token": foundUsr['usr_id'], "token_type": "bearer"}@app.get(path='/')async def root(token: str = Depends(oauth2_scheme)): await verify_token(token) return "Hello world"
启动应用并执行请求:
测试无效登录:
curl -d "username=xyz&password=x1y2z38" -X POST http://127.0.0.1:8000/token -iHTTP/1.1 400 Bad Requestdate: Tue, 08 Feb 2022 12:16:37 GMTserver: uvicorncontent-length: 43content-type: application/json{"detail":"Incorrect username or password"}
测试正常登录:
curl -d "username=xyz&password=x1y2z3" -X POST http://127.0.0.1:8000/token -iHTTP/1.1 200 OKdate: Tue, 08 Feb 2022 12:17:45 GMTserver: uvicorncontent-length: 51content-type: application/json{"access_token":"u000010002","token_type":"bearer"}
返回token,在Headers中使用token访问:
curl -H "Authorization:Bearer u000010002" http://127.0.0.1:8000 -iHTTP/1.1 200 OKdate: Tue, 08 Feb 2022 12:19:16 GMTserver: uvicorncontent-length: 13content-type: application/json"Hello world"
修改token后请求:
curl -H "Authorization:Bearer u000010007" http://127.0.0.1:8000 -iHTTP/1.1 401 Unauthorizeddate: Tue, 08 Feb 2022 12:19:56 GMTserver: uvicorncontent-length: 25content-type: application/json{"detail":"Unauthorized"}
上面的代码如果去掉 await verify_token(token) 行,则:
curl -H "Authorization:Bearer u000010007" http://127.0.0.1:8000 -i
会得到返回结果,原因是默认情况下,OAuth2PasswordBearer只负责请求头中是否具有Authorization:Bearer,如果有就会执行相应的请求,所以,为了验证Token的正确性,需要每个方法都执行相应的验证代码。
本例只作为例子,在实际开发中不会直接拿用户ID作为Token,为了提高系统的安全性,需要使用 JWT。下面我们就介绍 JWT。
26.3 JWT ( JSON Web Tokens )
JWT是一个将 JSON 对象编码为密集且没有空格的长字符串的标准。 具体学习和了解 JWT,请参考 https://jwt.io。
需要提到的主要是 JWT中的sub,JWT 的规范中有一个 sub 键,值为该令牌的主题。使用它并不是必须的,但这是我们放置用户标识的地方,所以一般情况下,我们在sub中存放用户ID, 为了避免 ID 冲突,当为创建 JWT 令牌时,可以在 sub 键的值前加上前缀,例如 username:、userid:等。
在 Python 中生成和校验 JWT 令牌 ,可以使用PyJWT,也可以使用 python-jose 。我们在本例中使用 python-jose 来编写代码。
pip install python-jose
使用:
from jose import jwt
使用 JWT,需要在系统中添加一个SECRET_KEY变量,用于生成令牌,如:
SECRET_KEY='5e9eb66688de11eca78070c94ec87656a649b0cc88de11eca8b470c94ec87656'
以下代码在上面代码的基础上使用 JWT 令牌。
# coding: utf-8from fastapi import FastAPIfrom fastapi import Dependsfrom fastapi import HTTPExceptionfrom fastapi.security import OAuth2PasswordBearerfrom fastapi.security import OAuth2PasswordRequestFormfrom passlib.hash import pbkdf2_sha256from datetime import datetimefrom datetime import timedeltafrom jose import jwtimport jsonapp = FastAPI()oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")SECRET_KEY='5e9eb66688de11eca78070c94ec87656a649b0cc88de11eca8b470c94ec87656'ALGORITHM = "HS256"ACCESS_TOKEN_EXPIRE_DAYS = 1# 用户列表usrs = [ { 'usr_id': 'u000010001', 'usr_acc': 'abc', 'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$fUI/40Zxj6.62GHgUX9PbDZ0SybccxZwn3Wl3qJ8U/M', # a1b2c3 'salt': '10181010' }, { 'usr_id': 'u000010002', 'usr_acc': 'xyz', 'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$Eu7mZV80f.tu4RDXSsst9gDjV4fXJ.9S7t1hgcGMMVk', # x1y2z3 'salt': '10101018' }]# 生成Tokendef build_access_token(usr: dict): for_encode = {'sub': json.dumps(usr)} expire = datetime.utcnow() + timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS) for_encode.update({"exp": expire}) jwt_code = jwt.encode(for_encode, SECRET_KEY, algorithm=ALGORITHM) return jwt_code# Token校验def verify_token(token: str): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) print(payload) except Exception as ex: print(str(ex)) raise HTTPException(status_code=401)# 登录并返回Token@app.post(path='/token')async def login(form: OAuth2PasswordRequestForm = Depends()): foundUsr = None for usr in usrs: if usr['usr_acc'] == form.username: foundUsr = usr if foundUsr is None or pbkdf2_sha256.hash(form.password, salt=usr['salt'].encode()) != foundUsr['usr_pwd']: raise HTTPException(status_code=400, detail="Incorrect username or password") else: return {"access_token": build_access_token(foundUsr), "token_type": "bearer"}@app.get(path='/')async def root(token: str = Depends(oauth2_scheme)): verify_token(token) return "Hello world"
与前面的代码差别之处:
1.生成Token的函数:build_access_token
2.校验Token的函数:verify_token
3.登录函数:login
请求测试:
登录:
curl -d "username=abc&password=a1b2c3" -X POST http://127.0.0.1:8000/token{"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAxXCIsIFwidXNyX2FjY1wiOiBcImFiY1wiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRmVUkvNDBaeGo2LjYyR0hnVVg5UGJEWjBTeWJjY3had24zV2wzcUo4VS9NXCIsIFwic2FsdFwiOiBcIjEwMTgxMDEwXCJ9IiwiZXhwIjoxNjQ0NDEzMDI5fQ.oer3oMlf3eN-d6RknIY5Yqjq7SRK4IxXE83ldyrVEw0","token_type":"bearer"}
令牌访问:
curl -H "Authorization:Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAxXCIsIFwidXNyX2FjY1wiOiBcImFiY1wiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRmVUkvNDBaeGo2LjYyR0hnVVg5UGJEWjBTeWJjY3had24zV2wzcUo4VS9NXCIsIFwic2FsdFwiOiBcIjEwMTgxMDEwXCJ9IiwiZXhwIjoxNjQ0NDEzMDI5fQ.oer3oMlf3eN-d6RknIY5Yqjq7SRK4IxXE83ldyrVEw0" http://127.0.0.1:8000"Hello world"
错误的令牌访问:
curl -H "Authorization:Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAxXCIsIFwidXNyX2FjY1wiOiBcImFiY1wiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRmVUkvNDBaeGo2LjYyR0hnVVg5UGJEWjBTeWJjY3had24zV2wzcUo4VS9NXCIsIFwic2FsdFwiOiBcIjEwMTgxMDEwXCJ9IiwiZXhwIjoxNjQ0NDEzMDI5fQ.oer3oMlf3eN-d6RknIY5Yqjq7SRK4IxXE83l" http://127.0.0.1:8000 -iHTTP/1.1 401 Unauthorizeddate: Tue, 08 Feb 2022 13:26:57 GMTserver: uvicorncontent-length: 25content-type: application/json{"detail":"Unauthorized"}
26.4 获取当前用户
在大部分应用程序中,当用户访问某个接口API的时候,都需要明确访问者的身份,所以在应用程序中需要随时获取当前用户,由于在 JWT 令牌的 sub 字段中已经保存了用户信息,所以获取当前用户只需要对令牌解码即可。
在上面的代码的基础上,增加两个函数,代码如下:
#获取当前用户def find_current_usr(token: str = Depends(oauth2_scheme)): current_usr = None try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) current_usr = json.loads(payload['sub']) except Exception as ex: print(str(ex)) raise HTTPException(status_code=401) if current_usr is None: raise HTTPException(status_code=401) return current_usr
# 获取登录者的信息@app.get(path='/myinfo')async def myinfo(current_usr: dict = Depends(find_current_usr)): return current_usr
请求测试:
curl -H "Authorization:Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAxXCIsIFwidXNyX2FjY1wiOiBcImFiY1wiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRmVUkvNDBaeGo2LjYyR0hnVVg5UGJEWjBTeWJjY3had24zV2wzcUo4VS9NXCIsIFwic2FsdFwiOiBcIjEwMTgxMDEwXCJ9IiwiZXhwIjoxNjQ0NDEzMDI5fQ.oer3oMlf3eN-d6RknIY5Yqjq7SRK4IxXE83ldyrVEw0" http://127.0.0.1:8000/myinfo{"usr_id":"u000010001","usr_acc":"abc","usr_pwd":"$pbkdf2-sha256$29000$MTAxMDEwMTg$fUI/40Zxj6.62GHgUX9PbDZ0SybccxZwn3Wl3qJ8U/M","salt":"10181010"}
以上,我们完成了一个简单的安全性示例,FastAPI提供的安全性框架帮助我们节约了很多代码,但在实际开发中,我们常常使用微服务的方式来开发,对于鉴权最好设计独立的微服务进行处理。后面我们会展示一个采用FastAPI开发的鉴权微服务,以便在此基础上进行业务系统的开发。
版权声明:内容来源于互联网和用户投稿 如有侵权请联系删除