Replace josepy with jwcrypto

mozilla_django_oidc/auth.py

@@ -9,11 +9,11 @@
 from django.contrib.auth.backends import ModelBackend
 from django.core.exceptions import ImproperlyConfigured, SuspiciousOperation
 from django.urls import reverse
-from django.utils.encoding import force_bytes, smart_bytes, smart_str
+from django.utils.encoding import force_bytes, smart_str
 from django.utils.module_loading import import_string
-from josepy.b64 import b64decode
-from josepy.jwk import JWK
-from josepy.jws import JWS, Header
+from jwcrypto.common import base64url_decode
+from jwcrypto.jwk import JWK
+from jwcrypto.jws import JWS, InvalidJWSSignature
 from requests.auth import HTTPBasicAuth
 from requests.exceptions import HTTPError
 
@@ -127,10 +127,11 @@ def update_user(self, user, claims):
 
     def _verify_jws(self, payload, key):
         """Verify the given JWS payload with the given key and return the payload"""
-        jws = JWS.from_compact(payload)
+        jws = JWS()
+        jws.deserialize(smart_str(payload))
 
         try:
-            alg = jws.signature.combined.alg.name
+            alg = jws.jose_header["alg"]
         except KeyError:
             msg = "No alg value found in header"
             raise SuspiciousOperation(msg)
@@ -143,13 +144,17 @@ def _verify_jws(self, payload, key):
             raise SuspiciousOperation(msg)
 
         if isinstance(key, str):
-            # Use smart_bytes here since the key string comes from settings.
-            jwk = JWK.load(smart_bytes(key))
+            try:
+                jwk = JWK.from_pem(force_bytes(key))
+            except ValueError:
+                jwk = JWK.from_password(key)
         else:
             # The key is a json returned from the IDP JWKS endpoint.
-            jwk = JWK.from_json(key)
+            jwk = JWK(**key)
 
-        if not jws.verify(jwk):
+        try:
+            jws.verify(jwk)
+        except InvalidJWSSignature:
             msg = "JWS token verification failed."
             raise SuspiciousOperation(msg)
 
@@ -167,17 +172,16 @@ def retrieve_matching_jwk(self, token):
         jwks = response_jwks.json()
 
         # Compute the current header from the given token to find a match
-        jws = JWS.from_compact(token)
-        json_header = jws.signature.protected
-        header = Header.json_loads(json_header)
+        jws = JWS()
+        jws.deserialize(smart_str(token))
 
         key = None
         for jwk in jwks["keys"]:
             if import_from_settings("OIDC_VERIFY_KID", True) and jwk[
                 "kid"
-            ] != smart_str(header.kid):
+            ] != smart_str(jws.jose_header["kid"]):
                 continue
-            if "alg" in jwk and jwk["alg"] != smart_str(header.alg):
+            if "alg" in jwk and jwk["alg"] != smart_str(jws.jose_header["alg"]):
                 continue
             key = jwk
         if key is None:
@@ -188,11 +192,11 @@ def get_payload_data(self, token, key):
         """Helper method to get the payload of the JWT token."""
         if self.get_settings("OIDC_ALLOW_UNSECURED_JWT", False):
             header, payload_data, signature = token.split(b".")
-            header = json.loads(smart_str(b64decode(header)))
+            header = json.loads(base64url_decode(smart_str(header)))
 
             # If config allows unsecured JWTs check the header and return the decoded payload
             if "alg" in header and header["alg"] == "none":
-                return b64decode(payload_data)
+                return base64url_decode(smart_str(payload_data))
 
         # By default fallback to verify JWT signatures
         return self._verify_jws(token, key)

mozilla_django_oidc/utils.py

@@ -4,8 +4,7 @@
 from hashlib import sha256
 from urllib.request import parse_http_list, parse_keqv_list
 
-# Make it obvious that these aren't the usual base64 functions
-import josepy.b64
+from jwcrypto.common import base64url_encode
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 
@@ -55,36 +54,6 @@ def is_authenticated(user):
     return user.is_authenticated
 
 
-def base64_url_encode(bytes_like_obj):
-    """Return a URL-Safe, base64 encoded version of bytes_like_obj
-
-    Implements base64urlencode as described in
-    https://datatracker.ietf.org/doc/html/rfc7636#appendix-A
-    """
-
-    s = josepy.b64.b64encode(bytes_like_obj).decode("ascii")  # base64 encode
-    # the josepy base64 encoder (strips '='s padding) automatically
-    s = s.replace("+", "-")  # 62nd char of encoding
-    s = s.replace("/", "_")  # 63rd char of encoding
-
-    return s
-
-
-def base64_url_decode(string_like_obj):
-    """Return the bytes encoded in a URL-Safe, base64 encoded string.
-    Implements inverse of base64urlencode as described in
-    https://datatracker.ietf.org/doc/html/rfc7636#appendix-A
-    This function is not used by the OpenID client; it's just for testing PKCE related functions.
-    """
-    s = string_like_obj
-
-    s = s.replace("_", "/")  # 63rd char of encoding
-    s = s.replace("-", "+")  # 62nd char of encoding
-    b = josepy.b64.b64decode(s)  # josepy base64 encoder (decodes without '='s padding)
-
-    return b
-
-
 def generate_code_challenge(code_verifier, method):
     """Return a code_challege, which proves knowledge of the code_verifier.
     The code challenge is generated according to method which must be one
@@ -99,7 +68,7 @@ def generate_code_challenge(code_verifier, method):
         return code_verifier
 
     elif method == "S256":
-        return base64_url_encode(sha256(code_verifier.encode("ascii")).digest())
+        return base64url_encode(sha256(code_verifier.encode("ascii")).digest())
 
     else:
         raise ValueError("code challenge method must be 'plain' or 'S256'.")

setup.py

@@ -34,6 +34,8 @@
 install_requirements = [
     "Django >= 3.2",
     "josepy",
+    "pyjwt",
+    "jwcrypto",
     "requests",
     "cryptography",
 ]