5.4.2. Build helpers

In the auth module, there is a slightly special part: helper functions. In this section, we will set up those functions. First, create the following files: get_authorization_token.py, get_info_from_claims.py, and get_public_keys.py.

Get Authorization Token Helper

Open get_authorization_token.py and add the following code.

Import the necessary modules:

# Import errors
from core.error import AppError, ClientError, is_standard_error

def get_authorization_token(ctx) -> str | None:
    """
    Lấy token từ trong headers. Mặc định sẽ lấy Bearer token.

    Args:
        ctx: internal context

    Returns:
        str: bearer token
    """
    headers = ctx.params.get("headers")

    # Lấy Authorization header (ưu tiên chuẩn, fallback lowercase)
    auth_header = headers.get("Authorization") or headers.get("authorization") or None

    try:
        if not auth_header or not auth_header.startswith("Bearer "):
            err = ClientError("Missing or invalid Authorization header")
            err.add_error_detail({"source": "get_authorization_token"})
            err.as_http_error("BadRequest")
            raise err

        parts = auth_header.split(" ", 1)
        token = parts[1] if len(parts) > 1 else None

        if token is None:
            err = ClientError("Bearer token not found")
            err.add_error_detail({"source": "get_authorization_token"})
            err.as_http_error("BadRequest")
            raise err

        return token
    except Exception as e:
        if not is_standard_error(e):
            msg = str(e)
            e = AppError("Cannot get token from authorization header")
            e.add_error_detail({"source": "get_authorization_token", "desc": msg})
            e.as_http_error("InternalServerError")

        if ctx.options.get("can_catch_error"):
            raise e

        return None

This function is designed to retrieve the Bearer Token value, which will be used in subsequent stages of the process.

5.4.2.1

Get Information from Claims Helper

Open get_info_from_claims.py and add the following code.

Import the necessary modules:

from typing import Any, Dict

def get_info_from_claims(claims: Dict[str, Any]) -> Dict[str, Any]:
    """
    Tổng hợp và lấy thông tin người dùng trong cognito claims.

    Args:
        claims (Dict[str, Any]): cognito claims.
    Return:
        dict: chứa username, team, role.
    """
    return {
        "username": claims.get("username"),
        "team": claims.get("cognito:groups", [None])[0],
        "role": claims.get("custom:role"),
    }

5.4.2.2

Get Public Keys Helper

Open get_public_keys.py and add the following code.

Import the necessary modules:

import os
import requests

from core.error.AppError import AppError

# Import constants
from utils.configs.main import Configs

def get_public_keys():
    """
    Lấy các public keys trong Cognito User Pool.
    Trả về list keys hoặc AppError nếu thất bại.
    """
    region = Configs.Aws_Region
    user_pool_id = Configs.Cognito_User_Pool_Id

    jwks_url = f"https://cognito-idp.{region}.amazonaws.com/{user_pool_id}/.well-known/jwks.json"

    try:
        response = requests.get(jwks_url, timeout=10)
        response.raise_for_status()
        data = response.json()

        return data.get("keys", [])
    except requests.RequestException as e:
        err = AppError("Cannot get public keys from Cognito User Pool")
        err.add_error_detail({"source": "get_public_keys", "desc": str(e)})
        err.as_http_error("InternalServerError")

        return err

5.4.2.3

Finally, create __init__.py in the helpers folder and add the following code.

from .get_authorization_token import get_authorization_token
from .get_public_keys import get_public_keys
from .get_info_from_claims import get_info_from_claims

__all__ = ["get_authorization_token", "get_info_from_claims", "get_public_keys"]

5.4.2.4

Now, we have built all the necessary helper functions in this module.