auth.jwt

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

import json
from typing import Any, Mapping, Sequence

from .auth_jwt_decode_jwt import check_jwt_secret  ← decodeJwt, another function of this group · built into the same file, even by a slim install
from .crypto_hmac_sha256 import hmac_sha256  ← from crypto.hmac-sha256 ^1.0.0 · built alongside by fune
from .encoding_base64_base64_url_encode import base64_url_encode
from .encoding_utf8_utf8_encode import utf8_encode

#: base64url of {"alg":"HS256","typ":"JWT"}: the header is fixed.
HEADER = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
MAX_DEPTH = 32
MAX_SAFE = 9007199254740991


def _canonical(value: Any, depth: int) -> Any:
    """The value with every number made a whole int, checked the way the other
    languages check it; json.dumps then writes it with sorted keys."""
    if value is None or isinstance(value, (bool, str)):
        return value
    if isinstance(value, (int, float)):
        if isinstance(value, float) and not value.is_integer():
            raise ValueError("claims may only hold whole numbers from -(2^53 - 1) to 2^53 - 1")
        if abs(value) > MAX_SAFE:
            raise ValueError("claims may only hold whole numbers from -(2^53 - 1) to 2^53 - 1")
        return int(value)
    if isinstance(value, (list, tuple)):
        if depth > MAX_DEPTH:
            raise ValueError("claims are nested more than 32 deep")
        return [_canonical(item, depth + 1) for item in value]
    if isinstance(value, Mapping):
        if depth > MAX_DEPTH:
            raise ValueError("claims are nested more than 32 deep")
        out = {}
        for key, item in value.items():
            if not isinstance(key, str):
                raise TypeError("claims must hold only JSON values")
            out[key] = _canonical(item, depth + 1)
        return out
    raise TypeError("claims must hold only JSON values")


def sign_jwt(claims: Mapping[str, Any], secret: Sequence[int]) -> str:
    """An HS256 JWT for the claims: header {"alg":"HS256","typ":"JWT"}, the
    claims as canonical JSON (no spaces, keys sorted by code point, UTF-8
    rather than \\u escapes), and the HMAC-SHA256 of the two."""
    check_jwt_secret(secret)
    if not isinstance(claims, Mapping):
        raise TypeError("claims must be a JSON object")
    payload = json.dumps(_canonical(claims, 1), sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
    signing_input = HEADER + "." + base64_url_encode(utf8_encode(payload))
    return signing_input + "." + base64_url_encode(hmac_sha256(secret, utf8_encode(signing_input)))

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 Sequence

from .auth_jwt_decode_jwt import check_jwt_secret, split_jwt  ← decodeJwt, another function of this group · built into the same file, even by a slim install
from .auth_jwt_types import JwtVerification
from .crypto_constant_time_equal import constant_time_equal  ← from crypto.constant-time-equal ^1.0.0 · built alongside by fune
from .crypto_hmac_sha256 import hmac_sha256  ← from crypto.hmac-sha256 ^1.0.0 · built alongside by fune
from .encoding_utf8_utf8_encode import utf8_encode


def _fail(error: str, message: str) -> JwtVerification:
    return JwtVerification(valid=False, claims=None, error=error, message=message)


def verify_jwt(token: str, secret: Sequence[int], now: int, leeway_seconds: int) -> JwtVerification:
    """Check an HS256 token: shape, algorithm (so "none" and algorithm-confusion
    tokens are refused before any key is used), signature in constant time,
    then exp, nbf and iat against now with leeway. A bad token is an answer,
    never an exception; only a bad secret, now or leeway raises."""
    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")

    parts = split_jwt(token)
    if parts is None:
        return _fail("malformed_token", "the token is not a well-formed JWT")
    header, claims, signing_input, signature = parts
    if header.get("alg") != "HS256":
        return _fail("unsupported_algorithm", "only HS256 tokens are accepted")
    if "crit" in header:
        return _fail("unsupported_algorithm", "the token requires header extensions (crit) this verifier does not support")

    expected = hmac_sha256(secret, utf8_encode(signing_input))
    if not constant_time_equal(expected, signature):
        return _fail("invalid_signature", "the token's signature does not match")

    for name in ("exp", "nbf", "iat"):
        if name in claims and (isinstance(claims[name], bool) or not isinstance(claims[name], (int, float))):
            return _fail("invalid_claims", "the token's %s claim is not a number" % name)
    # RFC 7519 4.1.4: the current time MUST be before exp.
    if "exp" in claims and now >= claims["exp"] + leeway_seconds:
        return _fail("token_expired", "the token has expired")
    # RFC 7519 4.1.5: the current time MUST be at or after nbf.
    if "nbf" in claims and now + leeway_seconds < claims["nbf"]:
        return _fail("token_not_yet_valid", "the token is not valid yet")
    if "iat" in claims and claims["iat"] > now + leeway_seconds:
        return _fail("token_issued_in_future", "the token was issued in the future")
    return JwtVerification(valid=True, claims=claims, error=None, message=None)

import json
import math
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple

from .encoding_base64_base64_url_decode import base64_url_decode

#: Containers nested deeper than this are refused, in every language, rather
#: than left to each parser's own limit.
MAX_DEPTH = 32
MAX_SAFE = 9007199254740991


def _sextet(ch: str) -> int:
    if "A" <= ch <= "Z":
        return ord(ch) - 65
    if "a" <= ch <= "z":
        return ord(ch) - 71
    if "0" <= ch <= "9":
        return ord(ch) + 4
    if ch == "-":
        return 62
    if ch == "_":
        return 63
    return -1


def _is_segment(text: str, allow_empty: bool) -> bool:
    """One canonical, unpadded base64url segment? Checked before decoding so a
    hostile token is an answer (None), never an exception."""
    if len(text) == 0:
        return allow_empty
    if len(text) % 4 == 1:
        return False
    last = 0
    for ch in text:
        last = _sextet(ch)
        if last < 0:
            return False
    if len(text) % 4 == 2:
        return last & 15 == 0
    if len(text) % 4 == 3:
        return last & 3 == 0
    return True


def _reject_constant(name: str) -> Any:
    # json.loads accepts NaN and Infinity, which are not JSON.
    raise ValueError("not JSON: %s" % name)


def _has_surrogate(text: str) -> bool:
    return any(0xD800 <= ord(ch) <= 0xDFFF for ch in text)


def _is_portable(value: Any, depth: int) -> bool:
    """Refuse what the three languages' parsers would answer differently:
    integers beyond 2^53, non-finite numbers, lone surrogates, deep nesting."""
    if isinstance(value, bool) or value is None:
        return True
    if isinstance(value, int):
        return abs(value) <= MAX_SAFE
    if isinstance(value, float):
        return math.isfinite(value) and (not value.is_integer() or abs(value) <= MAX_SAFE)
    if isinstance(value, str):
        return not _has_surrogate(value)
    if depth > MAX_DEPTH:
        return False
    if isinstance(value, list):
        return all(_is_portable(item, depth + 1) for item in value)
    return all(not _has_surrogate(k) and _is_portable(v, depth + 1) for k, v in value.items())


def _parse_object(segment: str) -> Optional[Dict[str, Any]]:
    try:
        text = bytes(base64_url_decode(segment)).decode("utf-8")
        parsed = json.loads(text, parse_constant=_reject_constant)
    except (ValueError, RecursionError):
        return None
    if not isinstance(parsed, dict):
        return None
    return parsed if _is_portable(parsed, 1) else None


def split_jwt(token: str) -> Optional[Tuple[Dict[str, Any], Dict[str, Any], str, List[int]]]:
    """(header, claims, signing input, signature bytes) of a compact JWT, or
    None when it is not three canonical base64url segments whose first two
    are JSON objects. Nothing is verified here; verify_jwt builds on it."""
    if not isinstance(token, str):
        return None
    parts = token.split(".")
    if len(parts) != 3:
        return None
    if not _is_segment(parts[0], False) or not _is_segment(parts[1], False) or not _is_segment(parts[2], True):
        return None
    header = _parse_object(parts[0])
    claims = _parse_object(parts[1])
    if header is None or claims is None:
        return None
    return header, claims, parts[0] + "." + parts[1], base64_url_decode(parts[2])


def check_jwt_secret(secret: Sequence[int]) -> None:
    """RFC 7518 section 3.2: an HS256 key MUST be at least 256 bits."""
    if isinstance(secret, (str, dict)) or not hasattr(secret, "__len__"):
        raise TypeError("secret must be a list of integers from 0 to 255")
    for b in secret:
        if type(b) is not int or b < 0 or b > 255:
            raise ValueError("secret must be a list of integers from 0 to 255")
    if len(secret) < 32:
        raise ValueError("secret must be at least 32 bytes (RFC 7518 section 3.2), received %d" % len(secret))


def decode_jwt(token: str) -> Optional[Mapping[str, Any]]:
    """The claims of a JWT WITHOUT checking its signature, or None when it is
    malformed. For reading exp or name from a token you hold; never for
    deciding whether to trust one."""
    parts = split_jwt(token)
    return None if parts is None else parts[1]