Sprint review: JWTPlugin gets the middleware's checks; object dispatch leaves service.py

py2http/auth.py

@@ -0,0 +1,174 @@
+"""Framework-independent credential checks, shared by the aiohttp middleware
+(``py2http.middleware``) and the Bottle plugins (``py2http.bottle_plugins``), so that
+both refuse the same tokens and keys for the same reasons.
+
+>>> bearer_token('Bearer abc'), bearer_token('bearer abc'), bearer_token('Basic abc')
+('abc', 'abc', '')
+>>> key_matches('k-123', 'k-123'), key_matches('k-12', 'k-123'), key_matches(None, 'k')
+(True, False, False)
+"""
+
+import hmac
+import logging
+from collections.abc import Iterable, Mapping
+from typing import Optional
+
+logger = logging.getLogger("py2http")
+
+DFLT_JWT_ALGORITHMS = ("HS256",)
+# ``jwt.decode`` arguments the verification setup owns, so ``decode_kwargs`` can't
+_RESERVED_DECODE_KWARGS = frozenset({"jwt", "key", "algorithms", "options"})
+_BEARER_PREFIX = "bearer "
+
+
+class CredentialsRejected(Exception):
+    """A request's credentials were refused. The message is safe to show the client:
+    it never contains the token, nor anything about the server's key."""
+
+
+def _algorithm_family(algorithm: str) -> str:
+    """The key family an algorithm verifies with: a shared secret for ``HS*``, a
+    public key otherwise.
+
+    >>> _algorithm_family('HS512'), _algorithm_family('RS256'), _algorithm_family('ES256')
+    ('hmac', 'public-key', 'public-key')
+    """
+    return "hmac" if algorithm.upper().startswith("HS") else "public-key"
+
+
+def _names(name: str, value: Iterable[str]) -> list:
+    if isinstance(value, (str, bytes)):
+        raise TypeError(f"{name} must be a list of names, e.g. [{value!r}]")
+    return list(value)
+
+
+def jwt_algorithms(
+    secret,
+    *,
+    verify: bool = True,
+    algorithms: Iterable[str] = DFLT_JWT_ALGORITHMS,
+    allow_mixed_algorithm_families: bool = False,
+) -> list:
+    """Validate a JWT verification setup, returning the algorithms as a list.
+
+    Raises at construction time, rather than accepting forgeries at request time, when:
+
+    - ``algorithms`` is empty, a bare string, or names an unknown or unsigned (``none``)
+      algorithm (PyJWT names are case-sensitive);
+    - ``verify`` is True and ``secret`` is empty (an empty HMAC secret lets anyone sign
+      tokens), or an HMAC algorithm is asked for with a non-``str``/``bytes`` secret;
+    - ``verify`` is True and HMAC and public-key algorithms are mixed with one
+      ``secret`` (the classic algorithm-confusion setup), unless
+      ``allow_mixed_algorithm_families`` is True.
+
+    >>> jwt_algorithms('s3cret')
+    ['HS256']
+    >>> jwt_algorithms('', algorithms=['HS256'])
+    Traceback (most recent call last):
+      ...
+    ValueError: secret must be a non-empty key when verify=True
+    """
+    import jwt
+
+    algorithms = _names("algorithms", algorithms)
+    if not algorithms:
+        raise ValueError("algorithms must name at least one JWT algorithm")
+    known = set(jwt.algorithms.get_default_algorithms()) - {"none"}
+    if unknown := [a for a in algorithms if a not in known]:
+        raise ValueError(
+            f"Unknown or unsigned JWT algorithm(s) {unknown}; names are "
+            f"case-sensitive, choose from {sorted(known)}"
+        )
+    if verify and not secret:
+        raise ValueError("secret must be a non-empty key when verify=True")
+    families = {_algorithm_family(a) for a in algorithms}
+    if verify and "hmac" in families and not isinstance(secret, (str, bytes)):
+        raise TypeError(f"an HMAC secret must be str or bytes, got {type(secret)}")
+    if verify and len(families) > 1 and not allow_mixed_algorithm_families:
+        raise ValueError(
+            f"algorithms={algorithms} mixes HMAC and public-key algorithms with one "
+            "secret, which allows algorithm-confusion forgeries; use one family, or "
+            "pass allow_mixed_algorithm_families=True if you really mean it"
+        )
+    return algorithms
+
+
+def jwt_decode_kwargs(decode_kwargs: Optional[Mapping] = None) -> dict:
+    """Validate extra ``jwt.decode`` arguments (``audience``, ``issuer``,
+    ``leeway``...), refusing those the verification setup owns.
+
+    >>> jwt_decode_kwargs({'audience': 'my-api'})
+    {'audience': 'my-api'}
+    >>> jwt_decode_kwargs({'options': {'verify_signature': False}})
+    Traceback (most recent call last):
+      ...
+    ValueError: decode_kwargs may not set ['options']; use verify= and algorithms=
+    """
+    decode_kwargs = dict(decode_kwargs or {})
+    if reserved := sorted(_RESERVED_DECODE_KWARGS & set(decode_kwargs)):
+        raise ValueError(
+            f"decode_kwargs may not set {reserved}; use verify= and algorithms="
+        )
+    return decode_kwargs
+
+
+def bearer_token(authorization_header) -> str:
+    """The token of an ``Authorization: Bearer <token>`` header, or ``''``."""
+    header = authorization_header or ""
+    if header[: len(_BEARER_PREFIX)].lower() != _BEARER_PREFIX:
+        return ""
+    return header[len(_BEARER_PREFIX) :].strip()
+
+
+def decode_jwt(
+    token: str,
+    secret,
+    *,
+    algorithms: list,
+    verify: bool = True,
+    decode_kwargs: Optional[Mapping] = None,
+) -> dict:
+    """The claims of ``token``, or ``CredentialsRejected`` with a client-safe reason.
+
+    With ``verify=False`` neither the signature nor the claims (``exp``...) are checked.
+    ``decode_kwargs`` (``audience``, ``issuer``, ``leeway``...) go to ``jwt.decode``.
+    """
+    import jwt
+
+    try:
+        return jwt.decode(
+            token,
+            secret,
+            options={"verify_signature": verify},
+            algorithms=algorithms,
+            **(decode_kwargs or {}),
+        )
+    except jwt.InvalidTokenError as error:  # its text describes the token
+        raise CredentialsRejected(f"Invalid authentication token: {error}") from error
+    except (jwt.PyJWTError, UnicodeError) as error:
+        # a key unfit for the token's alg (InvalidKeyError...) describes the server's
+        # key, so it is logged rather than sent; a non-UTF-8 header is just refused
+        if not isinstance(error, UnicodeError):
+            logger.warning("JWT rejected for a server-side reason: %r", error)
+        raise CredentialsRejected(
+            "Invalid authentication token: not acceptable"
+        ) from error
+
+
+def required_key(key, *, name: str = "api_key"):
+    """``key``, refused when empty: an empty key would let a request with no
+    ``Authorization`` header through."""
+    if not key or not isinstance(key, (str, bytes)):
+        raise ValueError(f"{name} must be a non-empty str or bytes")
+    return key
+
+
+def key_matches(given, expected) -> bool:
+    """Whether the ``given`` credential equals ``expected``, in constant time."""
+    if not given or not expected:
+        return False
+
+    def _bytes(x):  # aiohttp keeps non-UTF-8 header bytes as surrogate escapes
+        return x.encode(errors="surrogateescape") if isinstance(x, str) else x
+
+    return hmac.compare_digest(_bytes(given), _bytes(expected))

py2http/bottle_plugins.py

@@ -1,16 +1,42 @@
 """Plugins for adding middleware functionality to Bottle apps."""
 
-from bottle import request, response, abort
-from functools import wraps
 import json
-import jwt
+from functools import wraps
 from typing import Iterable
 from warnings import warn
+
+from bottle import request, response
+
+from py2http.auth import (
+    DFLT_JWT_ALGORITHMS,
+    CredentialsRejected,
+    bearer_token,
+    decode_jwt,
+    jwt_algorithms,
+    jwt_decode_kwargs,
+    key_matches,
+    required_key,
+)
 from py2http.constants import JSON_CONTENT_TYPE
 
 OPTIONS = "OPTIONS"
 
 
+def _authorization_header() -> str:
+    """The request's ``Authorization`` header; ``''`` if it is not valid UTF-8 (Bottle
+    raises when decoding it), which every check then refuses."""
+    try:
+        return request.headers.get("Authorization", "")
+    except UnicodeError:
+        return ""
+
+
+def _unauthorized(message):
+    response.status = 401
+    response.content_type = JSON_CONTENT_TYPE
+    return json.dumps({"error": message})
+
+
 class JWTPlugin:
     """A plugin for validating JWTs and extracting their payloads.
 
@@ -25,70 +51,82 @@ def __init__(
         mapper: dict = None,
         ignore_methods: Iterable[str] = None,
         algorithms: Iterable[str] = None,
+        *,
+        allow_mixed_algorithm_families: bool = False,
+        decode_kwargs: dict = None,
     ):
         """Creates a new JWTPlugin instance.
 
-        :param secret: (Optional) The JWT public key (RS256) or synchronous secret (HS256) used to validate tokens.
+        :param secret: The JWT public key (RS256) or shared secret (HS256) used to
+            validate tokens. Required (non-empty) when ``verify`` is True.
         :param verify: (Optional) If True, will verify JWT signatures against the provided secret,
         and reject unverified requests.
         :param mapper: (Optional) A dict that specifies how to map JWT claim value names in the output.
         :param ignore_methods: (Optional) A list of method names for the plugin to ignore.
+        :param algorithms: (Optional) The algorithms a token may be signed with
+            (default ``("HS256",)``); validated as ``py2http.auth.jwt_algorithms``
+            describes, as is ``allow_mixed_algorithm_families``.
+        :param decode_kwargs: (Optional) Extra ``jwt.decode`` arguments, such as
+            ``audience``, ``issuer`` or ``leeway``.
         """
         self._secret = secret
         self._verify = verify
         self._mapper = mapper if mapper else {}
         self._ignore_methods = ignore_methods if ignore_methods else []
-        self._algorithms = algorithms if algorithms else ["HS256"]
+        self._algorithms = jwt_algorithms(
+            secret,
+            verify=verify,
+            algorithms=algorithms or DFLT_JWT_ALGORITHMS,
+            allow_mixed_algorithm_families=allow_mixed_algorithm_families,
+        )
+        self._decode_kwargs = jwt_decode_kwargs(decode_kwargs)
+
+    def _claims(self):
+        """The (renamed) claims of the request's token; raises ``CredentialsRejected``."""
+        decoded = decode_jwt(
+            bearer_token(_authorization_header()),
+            self._secret,
+            algorithms=self._algorithms,
+            verify=self._verify,
+            decode_kwargs=self._decode_kwargs,
+        )
+        for k, v in self._mapper.items():
+            if k in decoded:
+                decoded[v] = decoded.pop(k)
+        return decoded
 
     def __call__(self, handler):
-        if self._ignore_methods and handler.method_name in self._ignore_methods:
+        if getattr(handler, "method_name", None) in self._ignore_methods:
             return handler
 
         @wraps(handler)
         def wrapped_handler(*args, **kwargs):
             if request.method == OPTIONS:
-                return handler(*args, *kwargs)
-            auth_header = request.headers.get("Authorization", "")
-            token = auth_header[7:]
-            try:
-                decoded = jwt.decode(
-                    token,
-                    self._secret,
-                    options={"verify_signature": self._verify},
-                    algorithms=self._algorithms,
-                )
-                for k, v in self._mapper.items():
-                    if k in decoded:
-                        decoded[v] = decoded.pop(k)
-                request.token = decoded
                 return handler(*args, **kwargs)
-            except jwt.DecodeError as error:
+            try:
+                request.token = self._claims()
+            except CredentialsRejected as rejection:
                 if self._verify:
-                    response.status = 401
-                    response.content_type = JSON_CONTENT_TYPE
-                    return json.dumps(
-                        {
-                            "error": f'Invalid authentication token "{token}", {str(error)}'
-                        }
-                    )
-                warn(f"Invalid JWT: {token}")
-                return handler(*args, **kwargs)
+                    return _unauthorized(str(rejection))
+                warn(f"Invalid JWT: {rejection.__cause__}")
+            return handler(*args, **kwargs)
 
         return wrapped_handler
 
 
 class ApiKeyAuthPlugin:
+    """A plugin letting through only requests whose ``Authorization`` header is
+    exactly ``api_key`` (compared in constant time; an empty key is refused)."""
+
     def __init__(self, api_key: str):
-        self._api_key = api_key
+        self._api_key = required_key(api_key)
 
     def __call__(self, handler):
+        @wraps(handler)
         def wrapped_handler(*args, **kwargs):
-            auth_header = request.headers.get("Authorization", "")
-            if auth_header == self._api_key:
+            if key_matches(_authorization_header(), self._api_key):
                 return handler(*args, **kwargs)
-            response.status = 401
-            response.content_type = JSON_CONTENT_TYPE
-            return json.dumps({"error": "invalid API key"})
+            return _unauthorized("invalid API key")
 
         return wrapped_handler
 
@@ -100,6 +138,7 @@ def __init__(self, origins: str = "*"):
         self._origins = origins
 
     def __call__(self, handler):
+        @wraps(handler)
         def wrapped_handler(*args, **kwargs):
             response.headers["Access-Control-Allow-Origin"] = self._origins
             response.headers["Access-Control-Allow-Methods"] = (