impl/python/read_access_token.py

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Any, Optional, Sequence

from .auth_access_token_types import AccessCheck
from .auth_bearer_token import parse_bearer_token  ← from auth.bearer-token ^1.0.0 · built alongside by fune
from .auth_jwt_decode_jwt import check_jwt_secret
from .auth_jwt_verify_jwt import verify_jwt
from .time_unix_to_iso import unix_to_iso  ← from time.unix-to-iso ^1.0.0 · built alongside by fune

#: The latest second time.unix-to-iso can write.
MAX_EXP = 253402300799


def access_denied(error: str, message: str) -> AccessCheck:
    """A failed check: every field but the reason is None."""
    return AccessCheck(
        ok=False, subject=None, token_version=None, jti=None, expires_at=None, claims=None, error=error, message=message
    )


def _whole(value: Any) -> Optional[int]:
    # A JSON 1.0 is a float in Python and 1 in JavaScript; both are whole.
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    return None


def read_access_token(authorization: Optional[str], secret: Sequence[int], now: int, leeway_seconds: int) -> AccessCheck:
    """Who is making this request? The Bearer token from the Authorization
    header, verified, and required to carry the claims issue_access_token
    writes. A missing or bad token is an answer (401), never an exception."""
    check_jwt_secret(secret)
    if isinstance(now, bool) or not isinstance(now, int):
        raise TypeError("now must be a whole number of Unix seconds")
    if isinstance(leeway_seconds, bool) or not isinstance(leeway_seconds, int) or leeway_seconds < 0:
        raise ValueError("leewaySeconds must be a whole number, 0 or more")

    token = parse_bearer_token(authorization)
    if token is None:
        return access_denied("missing_token", "the request has no Bearer token")
    verified = verify_jwt(token, secret, now, leeway_seconds)
    if not verified.valid or verified.claims is None:
        return access_denied(verified.error or "malformed_token", verified.message or "the token is not a well-formed JWT")
    claims = verified.claims
    sub, jti = claims.get("sub"), claims.get("jti")
    ver, exp = _whole(claims.get("ver")), _whole(claims.get("exp"))
    if (
        not isinstance(sub, str) or len(sub) == 0
        or not isinstance(jti, str) or len(jti) == 0
        or ver is None or ver < 0
        or exp is None or exp < 0 or exp > MAX_EXP
    ):
        return access_denied("invalid_claims", "the token lacks the sub, jti, ver or exp this API issues")
    return AccessCheck(
        ok=True, subject=sub, token_version=ver, jti=jti, expires_at=unix_to_iso(exp), claims=claims, error=None, message=None
    )