是否有 FastAPI 库可用于将端点标记为受保护并验证仅 HTTP Cookie 中的 Auth JWT 令牌？

编程问答 2022-05-03

问题描述

我正在尝试学习和使用 AWS Cognito 用户池，并与 Python FastAPI 集成和实现 API。到目前为止，我正在使用授权代码流和我的 Cognito 用户池重定向到 FastAPI 上的端点来解决代码挑战。源代码附加在此查询的末尾。

API 具有以下端点：

  1. 根终端节点 [ / ]：将浏览器重定向到我的 AWS Cognito 用户池的登录页面。
  2. 重定向端点 [ /aws_cognito_redirect ]：在成功登录用户池后激活。从 Cognito 用户池接收代码质询。在下面显示的代码中，aws_cognito_redirect 端点通过将代码质询、redirect_uri、client_id 等发送到 AWS Cognito 用户池 oauth2/token 端点来解决代码质询。我可以在控制台日志输出中看到身份、访问和刷新令牌已成功检索。

FastAPI 还将有一些受保护的端点，这些端点将从 Web 应用程序调用。还将有一个网络表单与端点进行交互。

在这个阶段，我可以使用 FastAPI jinja2 模板实现和托管网络表单。如果我选择了这个选项，大概我可以让 /aws_cognito_redirect 端点在仅 HTTP 会话 cookie 中返回令牌。这样，每个后续客户端请求将自动包含 cookie，而不会在浏览器本地存储中公开令牌。我知道我必须使用此选项处理 XSRF/CSRF。

或者，我可以使用 Angular/React 来实现前端。据推测，推荐的做法似乎是我必须将授权流程重新配置为使用 PKCE 的身份验证代码？在这种情况下，Angular/React Web 客户端将直接与 AWS Cognito 通信，以检索将转发到 FastAPI 端点的令牌。这些令牌将存储在浏览器的本地存储中，然后在每个后续请求的授权标头中发送。我知道这种方法会受到 XSS 攻击。

在这两者中，鉴于我的要求，我认为我倾向于使用 jinja2 模板在 FastAPI 上托管 web 应用程序，并在成功登录时返回仅 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("Tokens\n" + 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 将仅解析（并执行）一次此依赖项。

你甚至可以做一些更复杂的事情，比如创建 2 个嵌套的依赖项，一个简单地检查身份验证，第二个从数据库中检索用户帐户：

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 依赖项，因此会发生身份验证始终（并且始终只有一次）并且仅在需要时才从数据库中获取用户。

您可以在 documentation

中了解有关 FastAPI 中安全架构的更多信息（以及如何使用对 OAuth2 的内置支持）
amazon-cognitofastapiopenidopenidopenidopenid-connectpython-3.x