Repository: jpadilla/pyjwt
Pull request #562 (2020-12-19 17:15:23)

Remove unused argument "payload" from PyJWS._verify_signature()


Patch:
diff --git a/jwt/__init__.py b/jwt/__init__.py
index 9d95053..55a5ae3 100644
--- a/jwt/__init__.py
+++ b/jwt/__init__.py
@@ -1,12 +1,10 @@
-from .api_jws import PyJWS
-from .api_jwt import (
-    PyJWT,
-    decode,
-    encode,
+from .api_jws import (
+    PyJWS,
     get_unverified_header,
     register_algorithm,
     unregister_algorithm,
 )
+from .api_jwt import PyJWT, decode, encode
 from .exceptions import (
     DecodeError,
     ExpiredSignatureError,
diff --git a/jwt/api_jws.py b/jwt/api_jws.py
index 17c49db..152c237 100644
--- a/jwt/api_jws.py
+++ b/jwt/api_jws.py
@@ -1,7 +1,7 @@
 import binascii
 import json
 from collections.abc import Mapping
-from typing import Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type
 
 from .algorithms import (
     Algorithm,
@@ -77,7 +77,7 @@ class PyJWS:
 
     def encode(
         self,
-        payload: Union[Dict, bytes],
+        payload: bytes,
         key: str,
         algorithm: str = "HS256",
         headers: Optional[Dict] = None,
@@ -127,15 +127,14 @@ class PyJWS:
 
         return encoded_string.decode("utf-8")
 
-    def decode(
+    def decode_complete(
         self,
         jwt: str,
         key: str = "",
         algorithms: List[str] = None,
         options: Dict = None,
-        complete: bool = False,
         **kwargs,
-    ):
+    ) -> Dict[str, Any]:
         if options is None:
             options = {}
         merged_options = {**self.options, **options}
@@ -150,17 +149,25 @@ class PyJWS:
 
         if verify_signature:
             self._verify_signature(
-                payload, signing_input, header, signature, key, algorithms
+                signing_input, header, signature, key, algorithms
             )
 
-        if complete:
-            return {
-                "payload": payload,
-                "header": header,
-                "signature": signature,
-            }
+        return {
+            "payload": payload,
+            "header": header,
+            "signature": signature,
+        }
 
-        return payload
+    def decode(
+        self,
+        jwt: str,
+        key: str = "",
+        algorithms: List[str] = None,
+        options: Dict = None,
+        **kwargs,
+    ) -> str:
+        decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs)
+        return decoded["payload"]
 
     def get_unverified_header(self, jwt):
         """Returns back the JWT header parameters as a dict()
@@ -213,7 +220,6 @@ class PyJWS:
 
     def _verify_signature(
         self,
-        payload,
         signing_input,
         header,
         signature,
@@ -249,6 +255,7 @@ class PyJWS:
 
 _jws_global_obj = PyJWS()
 encode = _jws_global_obj.encode
+decode_complete = _jws_global_obj.decode_complete
 decode = _jws_global_obj.decode
 register_algorithm = _jws_global_obj.register_algorithm
 unregister_algorithm = _jws_global_obj.unregister_algorithm
diff --git a/jwt/api_jwt.py b/jwt/api_jwt.py
index 06c89f4..da273c5 100644
--- a/jwt/api_jwt.py
+++ b/jwt/api_jwt.py
@@ -4,7 +4,7 @@ from collections.abc import Iterable, Mapping
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional, Type, Union
 
-from .api_jws import PyJWS
+from . import api_jws
 from .exceptions import (
     DecodeError,
     ExpiredSignatureError,
@@ -16,8 +16,11 @@ from .exceptions import (
 )
 
 
-class PyJWT(PyJWS):
-    header_type = "JWT"
+class PyJWT:
+    def __init__(self, options=None):
+        if options is None:
+            options = {}
+        self.options = {**self._get_default_options(), **options}
 
     @staticmethod
     def _get_default_options() -> Dict[str, Union[bool, List[str]]]:
@@ -33,7 +36,7 @@ class PyJWT(PyJWS):
 
     def encode(
         self,
-        payload: Union[Dict, bytes],
+        payload: Dict[str, Any],
         key: str,
         algorithm: str = "HS256",
         headers: Optional[Dict] = None,
@@ -59,20 +62,18 @@ class PyJWT(PyJWS):
             payload, separators=(",", ":"), cls=json_encoder
         ).encode("utf-8")
 
-        return super().encode(
+        return api_jws.encode(
             json_payload, key, algorithm, headers, json_encoder
         )
 
-    def decode(
+    def decode_complete(
         self,
         jwt: str,
         key: str = "",
         algorithms: List[str] = None,
         options: Dict = None,
-        complete: bool = False,
         **kwargs,
     ) -> Dict[str, Any]:
-
         if options is None:
             options = {"verify_signature": True}
         else:
@@ -83,20 +84,16 @@ class PyJWT(PyJWS):
                 'It is required that you pass in a value for the "algorithms" argument when calling decode().'
             )
 
-        decoded = super().decode(
+        decoded = api_jws.decode_complete(
             jwt,
             key=key,
             algorithms=algorithms,
             options=options,
-            complete=complete,
             **kwargs,
         )
 
         try:
-            if complete:
-                payload = json.loads(decoded["payload"])
-            else:
-                payload = json.loads(decoded)
+            payload = json.loads(decoded["payload"])
         except ValueError as e:
             raise DecodeError("Invalid payload string: %s" % e)
         if not isinstance(payload, dict):
@@ -106,11 +103,19 @@ class PyJWT(PyJWS):
             merged_options = {**self.options, **options}
             self._validate_claims(payload, merged_options, **kwargs)
 
-        if complete:
-            decoded["payload"] = payload
-            return decoded
+        decoded["payload"] = payload
+        return decoded
 
-        return payload
+    def decode(
+        self,
+        jwt: str,
+        key: str = "",
+        algorithms: List[str] = None,
+        options: Dict = None,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs)
+        return decoded["payload"]
 
     def _validate_claims(
         self, payload, options, audience=None, issuer=None, leeway=0, **kwargs
@@ -215,7 +220,5 @@ class PyJWT(PyJWS):
 
 _jwt_global_obj = PyJWT()
 encode = _jwt_global_obj.encode
+decode_complete = _jwt_global_obj.decode_complete
 decode = _jwt_global_obj.decode
-register_algorithm = _jwt_global_obj.register_algorithm
-unregister_algorithm = _jwt_global_obj.unregister_algorithm
-get_unverified_header = _jwt_global_obj.get_unverified_header
diff --git a/jwt/jwks_client.py b/jwt/jwks_client.py
index 707724e..ecd10cb 100644
--- a/jwt/jwks_client.py
+++ b/jwt/jwks_client.py
@@ -2,7 +2,7 @@ import json
 import urllib.request
 
 from .api_jwk import PyJWKSet
-from .api_jwt import decode as decode_token
+from .api_jwt import decode_complete as decode_token
 from .exceptions import PyJWKClientError
 
 
@@ -50,8 +50,6 @@ class PyJWKClient:
         return signing_key
 
     def get_signing_key_from_jwt(self, token):
-        unverified = decode_token(
-            token, complete=True, options={"verify_signature": False}
-        )
+        unverified = decode_token(token, options={"verify_signature": False})
         header = unverified["header"]
         return self.get_signing_key(header.get("kid"))
diff --git a/jwt/utils.py b/jwt/utils.py
index 495c8c7..f9579d3 100644
--- a/jwt/utils.py
+++ b/jwt/utils.py
@@ -32,7 +32,7 @@ def base64url_decode(input):
     return base64.urlsafe_b64decode(input)
 
 
-def base64url_encode(input):
+def base64url_encode(input: bytes) -> bytes:
     return base64.urlsafe_b64encode(input).replace(b"=", b"")
 
 


Test patch:
diff --git a/tests/test_api_jws.py b/tests/test_api_jws.py
index a0456dc..2e8dd90 100644
--- a/tests/test_api_jws.py
+++ b/tests/test_api_jws.py
@@ -215,6 +215,27 @@ class TestJWS:
 
         assert decoded_payload == payload
 
+    def test_decodes_complete_valid_jws(self, jws, payload):
+        example_secret = "secret"
+        example_jws = (
+            b"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9."
+            b"aGVsbG8gd29ybGQ."
+            b"gEW0pdU4kxPthjtehYdhxB9mMOGajt1xCKlGGXDJ8PM"
+        )
+
+        decoded = jws.decode_complete(
+            example_jws, example_secret, algorithms=["HS256"]
+        )
+
+        assert decoded == {
+            "header": {"alg": "HS256", "typ": "JWT"},
+            "payload": payload,
+            "signature": (
+                b"\x80E\xb4\xa5\xd58\x93\x13\xed\x86;^\x85\x87a\xc4"
+                b"\x1ff0\xe1\x9a\x8e\xddq\x08\xa9F\x19p\xc9\xf0\xf3"
+            ),
+        }
+
     # 'Control' Elliptic Curve jws created by another library.
     # Used to test for regressions that could affect both
     # encoding / decoding operations equally (causing tests
diff --git a/tests/test_api_jwt.py b/tests/test_api_jwt.py
index 35ba6ba..dda896f 100644
--- a/tests/test_api_jwt.py
+++ b/tests/test_api_jwt.py
@@ -47,6 +47,27 @@ class TestJWT:
 
         assert decoded_payload == example_payload
 
+    def test_decodes_complete_valid_jwt(self, jwt):
+        example_payload = {"hello": "world"}
+        example_secret = "secret"
+        example_jwt = (
+            b"eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9"
+            b".eyJoZWxsbyI6ICJ3b3JsZCJ9"
+            b".tvagLDLoaiJKxOKqpBXSEGy7SYSifZhjntgm9ctpyj8"
+        )
+        decoded = jwt.decode_complete(
+            example_jwt, example_secret, algorithms=["HS256"]
+        )
+
+        assert decoded == {
+            "header": {"alg": "HS256", "typ": "JWT"},
+            "payload": example_payload,
+            "signature": (
+                b'\xb6\xf6\xa0,2\xe8j"J\xc4\xe2\xaa\xa4\x15\xd2'
+                b"\x10l\xbbI\x84\xa2}\x98c\x9e\xd8&\xf5\xcbi\xca?"
+            ),
+        }
+
     def test_load_verify_valid_jwt(self, jwt):
         example_payload = {"hello": "world"}
         example_secret = "secret"
@@ -313,13 +334,12 @@ class TestJWT:
         secret = "secret"
         jwt_message = jwt.encode(payload, secret)
 
-        decoded_payload, signing, header, signature = jwt._load(jwt_message)
-
         # With 3 seconds leeway, should be ok
         for leeway in (3, timedelta(seconds=3)):
-            jwt.decode(
+            decoded = jwt.decode(
                 jwt_message, secret, leeway=leeway, algorithms=["HS256"]
             )
+            assert decoded == payload
 
         # With 1 seconds, should fail
         for leeway in (1, timedelta(seconds=1)):