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

Question

我正在尝试学习和使用 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 应用程序中调用。此外，还将有一个 Web 表单与端点进行交互。

在此阶段，我可以使用 FastAPI jinja2 模板来实现和托管 Web 表单。如果我选择此选项，大概我可以让/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 是否存在并执行令牌验证？

快速API

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()))

Answer 1

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

您可以在文档中了解有关 FastAPI 中的安全架构（以及如何使用 OAuth2 的内置支持）的更多信息