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

Is There A FastAPI Library That Can be Used to Mark An Endpoint As Protected And Verify Auth JWT Tokens In HTTP Only Cookie?

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

API 具有以下端点:

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

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

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

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

在这两者中,根据我的要求,我认为我倾向于使用 jinja2 模板在 FastAPI 上托管 Web 应用程序,并 returning 仅 HTTP session cookie登录成功。

如果我选择这种实施方式,是否有 FastAPI 功能或 Python 库允许端点 decorated/marked 和 auth required 来检查是否存在 session 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()))

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)):
    ...

请注意,您的身份验证依赖项 return 的值是任意的,因此您可以 return 在您的用例中使用任何有意义的值(例如经过身份验证的用户帐户)。如果您想在 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 的内置支持)