我正在尝试学习和使用AWS Cognito用户池，并与使用Python FastAPI实现的API集成。到目前为止，我正在使用授权代码流和我的Cognito用户池重定向到FastAPI上的端点来解决代码挑战。源代码附加在该查询的末尾。
API具有以下端点：
- 根端点[/]：将浏览器重定向到我的AWS Cognito用户池的登录页面
-
重定向端点[/aws_cognito_Redirect]：成功登录用户池后激活。接收来自cognito用户池的代码质询。在下面显示的代码中，
aws_cognito_redirect端点通过向AWS Cognito用户池oauth2/token端点发送代码质询、redirect_uri、client_id等来解决代码质询。我可以在控制台日志输出中看到，已成功检索到标识、访问和刷新令牌
FastAPI还将具有一些受保护的端点，这些端点将从web应用程序调用。此外，还将有一个与端点交互的web表单。
在这个阶段，我可以使用FastAPI jinja2模板来实现和托管网络表单。如果我使用这个选项，大概我可以让/aws_cognito_redirect端点在仅HTTP会话cookie中返回令牌。这样，每个后续的客户端请求都会自动包含cookie，而浏览器本地存储中没有暴露任何令牌。我知道我必须用这个选项来处理XSRF/CSRF。
或者，我可以使用Angular/React实现前端。据推测，推荐的做法似乎是，我必须将授权流重新配置为使用PKCE的Auth Code？在这种情况下，Angular/React web客户端将直接与AWS Cognito通信，以检索将转发到FastAPI端点的令牌。这些令牌将存储在浏览器的本地存储中，然后在每个后续请求的授权标头中发送。我知道这种方法会受到XSS攻击。
在这两者中，考虑到我的要求，我认为我倾向于使用jinja2模板在FastAPI上托管网络应用程序，并在成功登录时返回仅HTTP会话cookie。
如果我选择了这个实现路线，是否有FastAPI功能或Python库允许用auth required装饰/标记端点，以检查会话cookie的存在并执行令牌验证？
FastAPI
import base64
from functools import lru_cache
import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.responses import RedirectResponse
from . import config
app = FastAPI()
@lru_cache()
def get_settings():
"""Create config settings instance encapsulating app config."""
return config.Settings()
def encode_auth_header(client_id: str, client_secret: str):
"""Encode client id and secret as base64 client_id:client_secret."""
secret = base64.b64encode(
bytes(client_id, "utf-8") + b":" + bytes(client_secret, "utf-8")
)
return "Basic " + secret.decode()
@app.get("/")
def read_root(settings: config.Settings = Depends(get_settings)):
login_url = (
"https://"
+ settings.domain
+ ".auth."
+ settings.region
+ ".amazoncognito.com/login?client_id="
+ settings.client_id
+ "&response_type=code&scope=email+openid&redirect_uri="
+ settings.redirect_uri
)
print("Redirecting to " + login_url)
return RedirectResponse(login_url)
@app.get("/aws_cognito_redirect")
async def read_code_challenge(
request: Request, settings: config.Settings = Depends(get_settings)
):
"""Retrieve tokens from oauth2/token endpoint"""
code = request.query_params["code"]
print("/aws_cognito_redirect received code := ", code)
auth_secret = encode_auth_header(settings.client_id, settings.client_secret)
headers = {"Authorization": auth_secret}
print("Authorization:" + str(headers["Authorization"]))
payload = {
"client_id": settings.client_id,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": settings.redirect_uri,
}
token_url = (
"https://"
+ settings.domain
+ ".auth."
+ settings.region
+ ".amazoncognito.com/oauth2/token"
)
async with httpx.AsyncClient() as client:
tokens = await client.post(
token_url,
data=payload,
headers=headers,
)
print("Tokensn" + str(tokens.json()))
FastAPI高度依赖依赖依赖项注入，它也可以用于身份验证。你所需要做的就是编写一个简单的依赖项来检查cookie：
async def verify_access(secret_token: Optional[str] = Cookie(None)):
if secret_token is None or secret_token not in valid_tokens:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
)
return secret_token
并将其作为依赖项在您的视图中使用：
@app.get("/")
def read_root(settings: config.Settings = Depends(get_settings), auth_token = Depends(verify_access)):
...
如果你想保护一组端点，你可以定义额外的路由器，它将始终包括verify_access作为依赖项：
app = FastAPI()
auth_required_router = APIRouter()
app.include_router(
auth_required_router, dependencies=[Depends(verify_access)],
)
@auth_required_router.get("/")
def read_root(settings: config.Settings = Depends(get_settings)):
...
请注意，身份验证依赖项返回的值是任意的，因此您可以在那里返回任何在您的用例中有意义的内容(例如经过身份验证的用户帐户(。如果要在auth_required_router注册的视图中检索此值，只需在视图参数中定义此依赖项即可。FastAPI将只解析(并执行(此依赖项一次。
你甚至可以做一些更复杂的事情，比如创建两个嵌套的依赖项，一个简单地检查身份验证，第二个从数据库中检索用户帐户：
async def authenticate(...):
... # Verifies the auth data without fetching the user
async def get_auth_user(auth = Depends(authenticate):
... # Gets the user from the database, based on the auth data
现在，您的auth_required_router只能具有authenticate依赖项，但每个也需要访问当前用户的视图都可以额外定义get_auth_user依赖项，因此身份验证将始终(并且始终仅发生一次(，并且只有在需要时才会从数据库中获取用户。
您可以在文档中了解更多关于FastAPI中的安全体系结构(以及如何使用对OAuth2的内置支持(