from typing import Any, Optional, Sequence from .auth_access_token_types import AccessCheck from .auth_bearer_token import parse_bearer_token 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 #: 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 )