Sem-ver: bugfix Typing fix ups - this commit will be merged into an earlier one

Signed-off-by: David Black <dblack@atlassian.com>

Author: dbaxa

atlassian_jwt_auth/auth.py

@@ -26,7 +26,7 @@ def create(
         cls,
         issuer: str,
         key_identifier: Union[KeyIdentifier, str],
-        private_key_pem: str,
+        private_key_pem: Union[str, bytes],
         audience: Union[str, Iterable[str]],
         **kwargs: Any,
     ) -> "BaseJWTAuth":
@@ -39,4 +39,4 @@ def create(
     def _get_header_value(self) -> bytes:
         return b"Bearer " + self._signer.generate_jwt(
             self._audience, additional_claims=self._additional_claims
-        ).encode("utf-8")
+        )

atlassian_jwt_auth/frameworks/common/asap.py

@@ -37,7 +37,7 @@ def _process_asap_token(
         if verifier is None:
             verifier = backend.get_verifier(settings=settings)
         asap_claims = verifier.verify_jwt(
-            token.decode("utf-8") if isinstance(token, bytes) else token,
+            token,
             settings.ASAP_VALID_AUDIENCE,
             leeway=settings.ASAP_VALID_LEEWAY,
         )
@@ -94,5 +94,9 @@ def _verify_issuers(
 ) -> None:
     """Verify that the issuer in the claims is valid and is expected."""
     claim_iss = asap_claims.get("iss")
-    if issuers is not None and claim_iss is not None and claim_iss not in issuers:
+    if issuers is None:
+        return None
+
+    if claim_iss is None or claim_iss not in issuers:
         raise InvalidIssuerError
+    return None

atlassian_jwt_auth/frameworks/common/decorators.py

@@ -72,23 +72,22 @@ def restrict_asap_wrapper(request, *args, **kwargs) -> Any:
             asap_claims = getattr(request, "asap_claims", None)
             error_response = None
 
-            if required and not asap_claims:
-                if backend is not None:
+            if not asap_claims:
+                if required:
                     return backend.get_401_response("Unauthorized", request=request)
-
+                else:
+                    # Claims are not required and asap claims are not present.
+                    return func(request, *args, **kwargs)
             try:
-                if asap_claims is not None:
-                    _verify_issuers(asap_claims, settings.ASAP_VALID_ISSUERS)
+                _verify_issuers(asap_claims, settings.ASAP_VALID_ISSUERS)
             except InvalidIssuerError:
-                if backend is not None:
-                    error_response = backend.get_403_response(
-                        "Forbidden: Invalid token issuer", request=request
-                    )
+                error_response = backend.get_403_response(
+                    "Forbidden: Invalid token issuer", request=request
+                )
             except InvalidTokenError:
-                if backend is not None:
-                    error_response = backend.get_401_response(
-                        "Unauthorized: Invalid token", request=request
-                    )
+                error_response = backend.get_401_response(
+                    "Unauthorized: Invalid token", request=request
+                )
 
             if error_response and required:
                 return error_response

atlassian_jwt_auth/frameworks/django/decorators.py

@@ -56,7 +56,10 @@ def restrict_asap(
                                                 must match the issuer for a
                                                 token to be considered valid.
     """
-    issuers = issuers if issuers is not None else []
     return _restrict_asap(
-        func, DjangoBackend(), issuers, required, subject_should_match_issuer=None
+        func,
+        DjangoBackend(),
+        issuers,
+        required,
+        subject_should_match_issuer=None,
     )

atlassian_jwt_auth/frameworks/django/tests/test_django.py

@@ -68,7 +68,7 @@ def check_response(
         token=None,
         authorization=None,
         retriever_key=None,
-    ):
+    ) -> None:
         if authorization is None:
             if token is None:
                 if private_key is None:
@@ -211,7 +211,7 @@ def test_request_non_decorated_subject_is_rejected(self):
     def test_request_using_settings_only_is_allowed(self):
         self.check_response("unneeded", "two")
 
-    def test_request_subject_does_not_need_to_match_issuer_from_settings(self):
+    def test_request_subject_does_not_need_to_match_issuer_from_settings(self) -> None:
         self.test_settings["ASAP_SUBJECT_SHOULD_MATCH_ISSUER"] = False
         self.check_response("needed", "one", 200, subject="different_than_is")
 

atlassian_jwt_auth/frameworks/flask/decorators.py

@@ -27,6 +27,8 @@ def with_asap(
                                                 must match the issuer for a
                                                 token to be considered valid.
     """
+    if required is None:
+        required = True
     return _with_asap(
-        func, FlaskBackend(), issuers, required or False, subject_should_match_issuer
+        func, FlaskBackend(), issuers, required, subject_should_match_issuer
     )

atlassian_jwt_auth/signer.py

@@ -72,11 +72,11 @@ def _generate_claims(
     def _now(self) -> datetime.datetime:
         return datetime.datetime.now(datetime.timezone.utc)
 
-    def generate_jwt(self, audience: Union[str, Iterable[str]], **kwargs: Any) -> str:
+    def generate_jwt(self, audience: Union[str, Iterable[str]], **kwargs: Any) -> bytes:
         """returns a new signed jwt for use."""
         key_identifier, private_key_pem = self.private_key_retriever.load(self.issuer)
         private_key = self._obtain_private_key(key_identifier, private_key_pem)
-        token = jwt.encode(
+        token: str = jwt.encode(
             self._generate_claims(audience, **kwargs),
             key=private_key,
             algorithm=self.algorithm,
@@ -86,9 +86,7 @@ def generate_jwt(self, audience: Union[str, Iterable[str]], **kwargs: Any) -> st
                 else key_identifier
             },
         )
-        if isinstance(token, bytes):
-            return token.decode("utf-8")
-        return token
+        return token.encode("utf-8")
 
 
 class TokenReusingJWTAuthSigner(JWTAuthSigner):
@@ -102,7 +100,7 @@ def __init__(
 
     def get_cached_token(
         self, audience: Union[str, Iterable[str]], **kwargs: Any
-    ) -> Optional[str]:
+    ) -> Optional[bytes]:
         """returns the cached token. If there is no matching cached token
         then None is returned.
         """
@@ -139,7 +137,7 @@ def can_reuse_token(self, existing_token, claims) -> bool:
                 return False
         return True
 
-    def generate_jwt(self, audience: Union[str, Iterable[str]], **kwargs: Any) -> str:
+    def generate_jwt(self, audience: Union[str, Iterable[str]], **kwargs: Any) -> bytes:
         existing_token = self.get_cached_token(audience, **kwargs)
         claims = self._generate_claims(audience, **kwargs)
         if existing_token and self.can_reuse_token(existing_token, claims):

atlassian_jwt_auth/tests/utils.py

@@ -45,7 +45,7 @@ def create_token(
     issuer: str,
     audience: Union[str, Iterable[str]],
     key_id: Union[KeyIdentifier, str],
-    private_key: str,
+    private_key: Union[str, bytes],
     subject: Optional[str] = None,
 ):
     """ " returns a token based upon the supplied parameters."""

atlassian_jwt_auth/verifier.py

@@ -53,7 +53,11 @@ def __init__(
         self._check_jti_uniqueness = kwargs.get("check_jti_uniqueness", False)
 
     def verify_jwt(
-        self, a_jwt: str, audience: str, leeway: int = 0, **requests_kwargs: Any
+        self,
+        a_jwt: Union[str, bytes],
+        audience: str,
+        leeway: int = 0,
+        **requests_kwargs: Any,
     ) -> Dict[Any, Any]:
         """Verify if the token is correct
 
@@ -85,7 +89,7 @@ def _load_public_key(self, public_key: str, algorithm: Optional[str]) -> Any:
 
     def _decode_jwt(
         self,
-        a_jwt: str,
+        a_jwt: Union[str, bytes],
         key_identifier: KeyIdentifier,
         jwt_key: Union[AllowedPublicKeys, PyJWK, str, bytes],
         audience: Optional[Union[str, Iterable[str]]] = None,