Repository: jpadilla/pyjwt
Pull request #664 (2021-06-16 08:51:34)

DOC: Clarify RSA encoding and decoding depend on the cryptography package
Followup to #208

I faced a similar issue because I read the installation part a little too fast, but I believe this should be clarified over the example many other developers will refer to.

Thanks for this project ❤️ 


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 2c33e32..505e374 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/psf/black
-    rev: 21.6b0
+    rev: 21.7b0
     hooks:
       - id: black
         args: ["--target-version=py36"]
@@ -18,7 +18,7 @@ repos:
         language_version: python3.8
 
   - repo: https://github.com/PyCQA/isort
-    rev: 5.8.0
+    rev: 5.9.3
     hooks:
       - id: isort
 
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index b138569..b049e9e 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -13,6 +13,9 @@ Changed
 Fixed
 ~~~~~
 
+- Prefer `headers["alg"]` to `algorithm` in `jwt.encode()`. `#673 <https://github.com/jpadilla/pyjwt/pull/673>`__
+- Fix aud validation to support {'aud': null} case. `#670 <https://github.com/jpadilla/pyjwt/pull/670>`__
+
 Added
 ~~~~~
 
diff --git a/docs/api.rst b/docs/api.rst
index d23d4bd..f1720b9 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -13,8 +13,9 @@ API Reference
         * for **asymmetric algorithms**: PEM-formatted private key, a multiline string
         * for **symmetric algorithms**: plain string, sufficiently long for security
 
-    :param str algorithm: algorithm to sign the token with, e.g. ``"ES256"``
-    :param dict headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``
+    :param str algorithm: algorithm to sign the token with, e.g. ``"ES256"``.
+        If ``headers`` includes ``alg``, it will be preferred to this parameter.
+    :param dict headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``.
     :param json.JSONEncoder json_encoder: custom JSON encoder for ``payload`` and ``headers``
     :rtype: str
     :returns: a JSON Web Token
diff --git a/docs/installation.rst b/docs/installation.rst
index 6da2ccd..925da97 100644
--- a/docs/installation.rst
+++ b/docs/installation.rst
@@ -7,6 +7,9 @@ You can install ``PyJWT`` with ``pip``:
 
     $ pip install pyjwt
 
+
+.. _installation_cryptography
+
 Cryptographic Dependencies (Optional)
 -------------------------------------
 
diff --git a/docs/usage.rst b/docs/usage.rst
index c5372af..2eacb96 100644
--- a/docs/usage.rst
+++ b/docs/usage.rst
@@ -17,6 +17,8 @@ Encoding & Decoding Tokens with HS256
 Encoding & Decoding Tokens with RS256 (RSA)
 -------------------------------------------
 
+RSA encoding and decoding require the ``cryptography`` module. See :ref:`installation_cryptography`.
+
 .. code-block:: pycon
 
     >>> import jwt
diff --git a/jwt/api_jws.py b/jwt/api_jws.py
index 6d13699..6cefb2c 100644
--- a/jwt/api_jws.py
+++ b/jwt/api_jws.py
@@ -77,7 +77,7 @@ class PyJWS:
         self,
         payload: bytes,
         key: str,
-        algorithm: str = "HS256",
+        algorithm: Optional[str] = "HS256",
         headers: Optional[Dict] = None,
         json_encoder: Optional[Type[json.JSONEncoder]] = None,
     ) -> str:
@@ -86,6 +86,10 @@ class PyJWS:
         if algorithm is None:
             algorithm = "none"
 
+        # Prefer headers["alg"] if present to algorithm parameter.
+        if headers and "alg" in headers and headers["alg"]:
+            algorithm = headers["alg"]
+
         if algorithm not in self._valid_algs:
             pass
 
diff --git a/jwt/api_jwt.py b/jwt/api_jwt.py
index 70a5e53..48a9316 100644
--- a/jwt/api_jwt.py
+++ b/jwt/api_jwt.py
@@ -38,7 +38,7 @@ class PyJWT:
         self,
         payload: Dict[str, Any],
         key: str,
-        algorithm: str = "HS256",
+        algorithm: Optional[str] = "HS256",
         headers: Optional[Dict] = None,
         json_encoder: Optional[Type[json.JSONEncoder]] = None,
     ) -> str:
@@ -177,19 +177,18 @@ class PyJWT:
             raise ExpiredSignatureError("Signature has expired")
 
     def _validate_aud(self, payload, audience):
-        if audience is None and "aud" not in payload:
-            return
+        if audience is None:
+            if "aud" not in payload or not payload["aud"]:
+                return
+            # Application did not specify an audience, but
+            # the token has the 'aud' claim
+            raise InvalidAudienceError("Invalid audience")
 
-        if audience is not None and "aud" not in payload:
+        if "aud" not in payload or not payload["aud"]:
             # Application specified an audience, but it could not be
             # verified since the token does not contain a claim.
             raise MissingRequiredClaimError("aud")
 
-        if audience is None and "aud" in payload:
-            # Application did not specify an audience, but
-            # the token has the 'aud' claim
-            raise InvalidAudienceError("Invalid audience")
-
         audience_claims = payload["aud"]
 
         if isinstance(audience_claims, str):


Test patch:
diff --git a/tests/test_api_jws.py b/tests/test_api_jws.py
index b731edc..fbeda02 100644
--- a/tests/test_api_jws.py
+++ b/tests/test_api_jws.py
@@ -166,6 +166,32 @@ class TestJWS:
         exception = context.value
         assert str(exception) == "Algorithm not supported"
 
+    def test_encode_with_headers_alg_none(self, jws, payload):
+        msg = jws.encode(payload, key=None, headers={"alg": "none"})
+        with pytest.raises(DecodeError) as context:
+            jws.decode(msg, algorithms=["none"])
+        assert str(context.value) == "Signature verification failed"
+
+    @crypto_required
+    def test_encode_with_headers_alg_es256(self, jws, payload):
+        with open(key_path("testkey_ec.priv"), "rb") as ec_priv_file:
+            priv_key = load_pem_private_key(ec_priv_file.read(), password=None)
+        with open(key_path("testkey_ec.pub"), "rb") as ec_pub_file:
+            pub_key = load_pem_public_key(ec_pub_file.read())
+
+        msg = jws.encode(payload, priv_key, headers={"alg": "ES256"})
+        assert b"hello world" == jws.decode(msg, pub_key, algorithms=["ES256"])
+
+    @crypto_required
+    def test_encode_with_alg_hs256_and_headers_alg_es256(self, jws, payload):
+        with open(key_path("testkey_ec.priv"), "rb") as ec_priv_file:
+            priv_key = load_pem_private_key(ec_priv_file.read(), password=None)
+        with open(key_path("testkey_ec.pub"), "rb") as ec_pub_file:
+            pub_key = load_pem_public_key(ec_pub_file.read())
+
+        msg = jws.encode(payload, priv_key, algorithm="HS256", headers={"alg": "ES256"})
+        assert b"hello world" == jws.decode(msg, pub_key, algorithms=["ES256"])
+
     def test_decode_algorithm_param_should_be_case_sensitive(self, jws):
         example_jws = (
             "eyJhbGciOiJoczI1NiIsInR5cCI6IkpXVCJ9"  # alg = hs256
diff --git a/tests/test_api_jwt.py b/tests/test_api_jwt.py
index 56de90c..1faa05f 100644
--- a/tests/test_api_jwt.py
+++ b/tests/test_api_jwt.py
@@ -202,6 +202,16 @@ class TestJWT:
         with pytest.raises(DecodeError):
             jwt.decode(example_jwt, "secret", algorithms=["HS256"])
 
+    def test_decode_raises_exception_if_aud_is_none(self, jwt):
+        # >>> jwt.encode({'aud': None}, 'secret')
+        example_jwt = (
+            "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9."
+            "eyJhdWQiOm51bGx9."
+            "-Peqc-pTugGvrc5C8Bnl0-X1V_5fv-aVb_7y7nGBVvQ"
+        )
+        decoded = jwt.decode(example_jwt, "secret", algorithms=["HS256"])
+        assert decoded["aud"] is None
+
     def test_encode_datetime(self, jwt):
         secret = "secret"
         current_datetime = datetime.utcnow()
@@ -413,6 +423,15 @@ class TestJWT:
 
         assert exc.value.claim == "aud"
 
+    def test_raise_exception_token_with_aud_none_and_without_audience(self, jwt):
+        payload = {"some": "payload", "aud": None}
+        token = jwt.encode(payload, "secret")
+
+        with pytest.raises(MissingRequiredClaimError) as exc:
+            jwt.decode(token, "secret", audience="urn:me", algorithms=["HS256"])
+
+        assert exc.value.claim == "aud"
+
     def test_check_issuer_when_valid(self, jwt):
         issuer = "urn:foo"
         payload = {"some": "payload", "iss": "urn:foo"}