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]