auth.access-token

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, Sequence

from .auth_access_token_types import AccessToken
from .auth_jwt_sign_jwt import sign_jwt
from .time_unix_to_iso import unix_to_iso  ← from time.unix-to-iso ^1.0.0 · built alongside by fune


def _is_whole(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def issue_access_token(
    subject: str,
    name: str,
    email: str,
    token_version: int,
    jti: str,
    now: int,
    ttl_seconds: int,
    secret: Sequence[int],
) -> AccessToken:
    """A signed access token for a user who has just logged in, with the claims
    this API relies on: sub, name, email, ver (token version), jti, iat and
    exp. Everything that varies (the time, the random jti) is passed in."""
    if not isinstance(subject, str) or len(subject) == 0:
        raise TypeError("subject must be a non-empty string")
    if not isinstance(name, str):
        raise TypeError("name must be a string")
    if not isinstance(email, str):
        raise TypeError("email must be a string")
    if not _is_whole(token_version) or token_version < 0:
        raise ValueError("tokenVersion must be a whole number, 0 or more")
    if not isinstance(jti, str) or len(jti) == 0:
        raise TypeError("jti must be a non-empty string")
    if not _is_whole(now):
        raise TypeError("now must be a whole number of Unix seconds")
    if not _is_whole(ttl_seconds) or ttl_seconds < 1:
        raise ValueError("ttlSeconds must be a whole number of at least 1")
    exp = now + ttl_seconds
    expires_at = unix_to_iso(exp)
    token = sign_jwt(
        {"sub": subject, "name": name, "email": email, "ver": token_version, "jti": jti, "iat": now, "exp": exp},
        secret,
    )
    return AccessToken(access_token=token, token_type="Bearer", expires_at=expires_at, expires_in=ttl_seconds)

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
    )

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 Optional

from .auth_access_token_read_access_token import access_denied  ← readAccessToken, another function of this group · built into the same file, even by a slim install
from .auth_access_token_types import AccessCheck


def confirm_access_token(check: AccessCheck, current_token_version: Optional[int], revoked: bool) -> AccessCheck:
    """The last word on a token read_access_token accepted, once the caller has
    looked up the user and the revoked list: a logged-out token, or one issued
    before the user's token version was bumped (a password change), no longer
    stands."""
    if current_token_version is not None and (
        isinstance(current_token_version, bool) or not isinstance(current_token_version, int)
    ):
        raise TypeError("currentTokenVersion must be a whole number or null")
    if not isinstance(revoked, bool):
        raise TypeError("revoked must be true or false")
    if not check.ok:
        return check
    if current_token_version is None:
        return access_denied("user_not_found", "the token's user no longer exists")
    if revoked:
        return access_denied("token_revoked", "the token has been revoked by logging out")
    if check.token_version != current_token_version:
        return access_denied(
            "token_revoked", "the token was issued before the user's tokens were revoked (password changed)"
        )
    return check