diff --git a/mozilla_django_oidc/_jose.py b/mozilla_django_oidc/_jose.py new file mode 100644 index 00000000..85ce21a4 --- /dev/null +++ b/mozilla_django_oidc/_jose.py @@ -0,0 +1,79 @@ +""" +Helpers for dealing with Javascript Object Signing and Encryption (JOSE) in an OpenID +Connect context. +""" +import json +from typing import Dict, Any, TypeAlias, Union + +from django.core.exceptions import SuspiciousOperation +from django.utils.encoding import smart_bytes + +from josepy.jwk import JWK +from josepy.jws import JWS + +# values could be narrowed down to relevant JSON types. +Payload: TypeAlias = Dict[str, Any] + + +def verify_jws_and_decode( + token: bytes, + key, + signing_algorithm: str = "", + decode_json: bool = False, # for backwards compatibility reasons +) -> Union[Payload, bytes]: + """ + Cryptographically verify the passed in token and return the decoded payload. + + Verification is done with utilities from the josepy library. + + :arg token: the raw binary content of the JWT/token. + :arg key: the key to verify the signature with. This may be a key obtained from + the OIDC_OP_JWKS_ENDPOINT or a shared key as a string (XXX CONFIRM THIS!) + :arg signing_algorithm: If provided, the token must employ this exact signing + algorithm. Values must be valid names from algorithms in :mod:`josepy.jwa`. + :arg decode_json: If true, the payload will be json-decoded and return a Python + object rather than a bytestring. + :return: the extracted payload object, deserialized from JSON. + :raises SuspiciousOperation: if the token verification fails. + """ + jws = JWS.from_compact(token) + + # validate the signing algorithm + if (alg := jws.signature.combined.alg) is None: + msg = "No alg value found in header" + raise SuspiciousOperation(msg) + + if signing_algorithm and signing_algorithm != alg.name: + msg = ( + "The provider algorithm {!r} does not match the client's " + "expected signing algorithm.".format(alg) + ) + raise SuspiciousOperation(msg) + + # one of the most common implementation weaknesses -> attacker can supply 'none' + # algorithm + # XXX: check if this is okay, technically users can now set + # settings.OIDC_RP_SIGN_ALGO = "none" and things should work? + if alg.name == "none": + raise SuspiciousOperation("'none' for alg value is not allowed") + + # process the key parameter which was/may have been loaded from keys endpoint + if isinstance(key, str): + # Use smart_bytes here since the key string comes from settings. + jwk = JWK.load(smart_bytes(key)) + else: + # The key is a json returned from the IDP JWKS endpoint. + jwk = JWK.from_json(key) + # address some missing upstream Self type declarations + assert isinstance(jwk, JWK) + + if not jws.verify(jwk): + msg = "JWS token verification failed." + raise SuspiciousOperation(msg) + + # return the decoded JSON or the raw bytestring payload + payload = jws.payload + if not decode_json: + return payload + else: + return json.loads(payload.decode("utf-8")) diff --git a/mozilla_django_oidc/auth.py b/mozilla_django_oidc/auth.py index f8243fe3..869a3346 100644 --- a/mozilla_django_oidc/auth.py +++ b/mozilla_django_oidc/auth.py @@ -9,15 +9,15 @@ 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 requests.auth import HTTPBasicAuth from requests.exceptions import HTTPError from mozilla_django_oidc.utils import absolutify, import_from_settings +from ._jose import verify_jws_and_decode LOGGER = logging.getLogger(__name__) @@ -127,33 +127,12 @@ 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) - - try: - alg = jws.signature.combined.alg.name - except KeyError: - msg = "No alg value found in header" - raise SuspiciousOperation(msg) - - if alg != self.OIDC_RP_SIGN_ALGO: - msg = ( - "The provider algorithm {!r} does not match the client's " - "OIDC_RP_SIGN_ALGO.".format(alg) - ) - raise SuspiciousOperation(msg) - - if isinstance(key, str): - # Use smart_bytes here since the key string comes from settings. - jwk = JWK.load(smart_bytes(key)) - else: - # The key is a json returned from the IDP JWKS endpoint. - jwk = JWK.from_json(key) - - if not jws.verify(jwk): - msg = "JWS token verification failed." - raise SuspiciousOperation(msg) - - return jws.payload + return verify_jws_and_decode( + token=payload, + key=key, + signing_algorithm=self.OIDC_RP_SIGN_ALGO, + decode_json=False, + ) def retrieve_matching_jwk(self, token): """Get the signing key by exploring the JWKS endpoint of the OP."""