python-3.x - 是否有 FastAPI 库可用于将端点标记为 protected 并验证 HTTP Only Cookie 中的身份验证 JWT token ？

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 终端节点来解决代码质询。我可以在控制台日志输出中看到身份、访问和刷新 token 已成功检索。

FastAPI 另外还有一些 protected 端点，这些端点将从 Web 应用程序中调用。此外，还将有一个 Web 表单与端点交互。

在此阶段，我可以使用 FastAPI jinja2 模板实现和托管 Web 表单。如果我选择此选项，大概我可以让 /aws_cognito_redirect 端点在仅 HTTP session cookie 中返回 token 。这样，每个后续的客户端请求都会自动包含 cookie，而不会在浏览器本地存储中暴露任何 token 。我知道我必须使用此选项来处理 XSRF/CSRF。

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

在这两者中，考虑到我的要求，我认为我倾向于使用 jinja2 模板在 FastAPI 上托管 Web 应用程序，并在成功登录时返回仅 HTTP session cookie。

如果我选择此实现路线，是否有 FastAPI 功能或 Python 库允许使用 auth required 来装饰/标记端点，以检查 session cookie 是否存在并执行 token 验证？

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

Answer 1

FastAPI 高度依赖依赖注入(inject)，它也可以用于身份验证。您需要做的就是编写一个简单的依赖项来检查 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

并在您的 View 中将其用作依赖项:

@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 注册的 View 中检索此值，只需在 View 参数中定义此依赖项即可。 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 依赖项，但每个还需要访问当前用户的 View 都可以另外具有 get_auth_user 定义了依赖项，因此身份验证将始终发生(并且始终仅发生一次)，并且仅在需要时才从数据库中获取用户。

您可以在 documentation 中了解有关 FastAPI 中的安全架构(以及如何使用 OAuth2 的内置支持)的更多信息。