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

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

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

API 具有以下端点：

根终端节点 [ / ]：将浏览器重定向到我的 AWS Cognito 用户池的登录页面。
重定向端点 [ /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 的内置支持）