Fix JWT authentication security issues

docs/changelog.rst

@@ -15,6 +15,7 @@ Changelog
    applied. Uninstalling from the same panel removes the PAS plugin
    and the per-user JWT signing secrets.
 
+- #90 Fix JWT authentication security issues
 - #88 Add Bearer token (JWT) authentication via PAS plugin
 - #89 Fix inherited schema fields missing for multi-schema Dexterity types
 - #87 Inject additional analyses fields

src/senaite/jsonapi/pas/plugin.py

@@ -89,36 +89,47 @@ def authenticateCredentials(self, credentials):  # noqa camelCase
             return None
 
         # Peek the payload (no signature check) to learn the userid the
-        # token claims to belong to
+        # token claims to belong to. The unverified userid is only used
+        # below as a key into the keystorage; it is never trusted on its
+        # own (see peek_userid docstring).
         userid = peek_userid(token)
         if not userid:
             return None
 
-        # Verify signature and expiration with that user's secret
+        # Look up the user *before* touching the keystorage. This avoids
+        # an anonymous DoS where an attacker sprays tokens with random
+        # userids and forces the plugin to create a new signing-secret
+        # entry on the portal's OOBTree annotation for each one.
+        user = api.get_user(userid)
+        if not user:
+            return None
+
+        # Verify signature and expiration with that user's existing
+        # secret. decode_token returns None if no secret is stored for
+        # the user yet (i.e. the user has never logged in via /login).
         payload = decode_token(token, userid)
         if not payload:
             return None
 
-        # Defensive: ensure the verified payload still names the same user
+        # Defensive: ensure the verified payload still names the same
+        # user as the (now signature-verified) claim.
         if api.to_utf8(payload.get("userid"), default=None) != userid:
             return None
 
-        # Make sure the user still exists
-        user = api.get_user(userid)
-        if not user:
-            return None
-
         return userid, userid
 
 
 def get_jwt_token(request):
     """Extracts the JWT token from the request, in this order of preference:
     Authorization: Bearer header, "token" cookie, X-JWT-Auth-Token header.
     """
-    # Read from Authorization header
-    auth = request._auth  # noqa
-    if auth and auth[:7].lower() == "bearer ":
-        return auth.split()[-1]
+    # Read from Authorization header. Zope/PAS may consume the
+    # Authorization header from the WSGI environ before our plugin
+    # runs (so getHeader("Authorization") returns None), but the raw
+    # value is cached on request._auth at request init.
+    auth = request._auth or ""  # noqa
+    if auth[:7].lower() == "bearer ":
+        return auth[7:].strip()
 
     # Read from cookie
     token = request.cookies.get(JWT_COOKIE_ID)
@@ -160,7 +171,7 @@ def signing_secret(userid):
 
 def rotate_secret(userid):
     """Discards the signing secret for the given userid, so a new one is
-    generated on the next call to ``signing_secret``. All previously
+    generated on the next call to `signing_secret`. All previously
     issued tokens for the user are invalidated.
     """
     userid = api.to_utf8(userid, default=None)
@@ -180,36 +191,68 @@ def timestamp(seconds=0, minutes=0, hours=0, days=0):
 
 
 def peek_userid(token):
-    """Returns the ``userid`` claim of the given token without verifying
+    """Returns the `userid` claim of the given token without verifying
     the signature, or None if the token cannot be decoded.
 
     Decoding the token without verifying its signature is safe in this
-    context: the returned ``userid`` is only used to look up *that
+    context: the returned `userid` is only used to look up *that
     user's* signing secret, which is then used to verify the signature
-    in ``decode_token``. A forged token that claims to belong to user
+    in `decode_token`. A forged token that claims to belong to user
     X but was signed with anything other than X's real secret fails
-    signature verification and is rejected. The unverified ``userid``
+    signature verification and is rejected. The unverified `userid`
     is never trusted on its own.
     """
     token = api.to_utf8(token, default=None)
     if not token:
         return None
     try:
-        payload = jwt.decode(token, verify=False)
+        # Disable signature verification at peek time, but still
+        # require HS256 as the algorithm so a forged "alg": "none"
+        # token cannot reach signature verification with no secret.
+        # The `verify=False` kwarg was removed in PyJWT 2.x; the
+        # `options` form is supported by both 1.x and 2.x.
+        payload = jwt.decode(
+            token, options={"verify_signature": False},
+            algorithms=["HS256"],
+        )
     except (ValueError, TypeError, jwt.InvalidTokenError):
         return None
     return api.to_utf8(payload.get("userid"), default=None)
 
 
+def get_signing_secret(userid):
+    """Returns the existing signing secret for `userid` or None.
+
+    Unlike `signing_secret`, this never creates a new secret entry,
+    so it is safe to call with an unverified userid claim (e.g. during
+    token verification).
+    """
+    userid = api.to_utf8(userid, default=None)
+    if not userid:
+        return None
+    data = get_keystorage().get(userid)
+    if not data:
+        return None
+    return data.get("key")
+
+
 def decode_token(token, userid):
-    """Decodes the given JWT, verifies the signature with ``userid``'s
+    """Decodes the given JWT, verifies the signature with `userid`'s
     secret and checks the expiration. Returns the payload or None.
+
+    Returns None if no signing secret exists for `userid` yet; the
+    keystorage entry is only created by `create_token` (called from
+    `/login`), never by the verification path. This prevents an
+    anonymous attacker from forcing the portal to create one
+    keystorage entry per token claim.
     """
     token = api.to_utf8(token, default=None)
     if not token:
         return None
+    secret = get_signing_secret(userid)
+    if not secret:
+        return None
     try:
-        secret = signing_secret(userid)
         return jwt.decode(token, secret, algorithms=["HS256"])
     except (ValueError, TypeError, jwt.InvalidTokenError):
         return None

src/senaite/jsonapi/tests/doctests/bearer.rst

@@ -67,7 +67,7 @@ once the user is authenticated):
     >>> isinstance(token, basestring) and len(token) > 0
     True
 
-A fresh browser carrying that token in the ``Authorization: Bearer``
+A fresh browser carrying that token in the `Authorization: Bearer`
 header is authenticated by the JWT PAS plugin:
 
     >>> is_authenticated(with_bearer(fresh_browser(), token))
@@ -77,9 +77,9 @@ header is authenticated by the JWT PAS plugin:
 The X-JWT-Auth-Token fallback header authenticates the request
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-Clients that cannot set the ``Authorization`` header (some embedded
+Clients that cannot set the `Authorization` header (some embedded
 HTTP libraries, browser extensions, etc.) can use the
-``X-JWT-Auth-Token`` fallback header instead:
+`X-JWT-Auth-Token` fallback header instead:
 
     >>> fallback = fresh_browser()
     >>> fallback.addHeader("X-JWT-Auth-Token", token.encode("ascii"))
@@ -110,10 +110,31 @@ verification:
     False
 
 
+A token for an unknown user does not create a keystorage entry
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Tokens that claim to belong to a user who does not exist must be
+rejected without touching the per-user keystorage. Otherwise an
+anonymous attacker can force unbounded ZODB writes by spraying
+tokens with random userids.
+
+    >>> from senaite.jsonapi.pas.plugin import get_keystorage
+    >>> before = set(get_keystorage().keys())
+
+    >>> attacker = pyjwt.encode(
+    ...     {"userid": "nobody-12345", "exp": timestamp(seconds=3600)},
+    ...     "attacker-secret", algorithm="HS256")
+    >>> is_authenticated(with_bearer(fresh_browser(), attacker))
+    False
+
+    >>> set(get_keystorage().keys()) == before
+    True
+
+
 An expired token does not authenticate
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-A token whose ``exp`` claim lies in the past is rejected by PyJWT:
+A token whose `exp` claim lies in the past is rejected by PyJWT:
 
     >>> expired_token = create_token(TEST_USER_ID, exp=timestamp(seconds=-60))
     >>> transaction.commit()
@@ -124,7 +145,7 @@ A token whose ``exp`` claim lies in the past is rejected by PyJWT:
 Rotating the user's secret revokes existing tokens
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-After ``rotate_secret``, the previously-issued token no longer
+After `rotate_secret`, the previously-issued token no longer
 verifies:
 
     >>> rotate_secret(TEST_USER_ID)

src/senaite/jsonapi/tests/doctests/login.rst

@@ -50,7 +50,7 @@ Issue a token via /login
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
 An already-authenticated user calls /login and obtains a JWT in the
-response body. The setup helper ``self.getBrowser()`` returns a browser
+response body. The setup helper `self.getBrowser()` returns a browser
 that has a Plone session cookie, so the request is authenticated by
 the cookie auth plugin before the route runs:
 
@@ -69,9 +69,9 @@ the cookie auth plugin before the route runs:
 
 The response also sets the JWT as an HttpOnly, Secure cookie. The
 test browser drives the API over plain HTTP, so the Secure cookie is
-not retained by ``browser.cookies``; the assertion below would only
+not retained by `browser.cookies`; the assertion below would only
 hold over HTTPS. Clients running on HTTP must therefore use the
-``Authorization: Bearer`` header (covered next).
+`Authorization: Bearer` header (covered next).
 
 The Set-Cookie header still carries the expected security attributes
 even when the browser drops the cookie itself: