Sem-ver: feature Provide type information to consumers of the library

Signed-off-by: David Black <[email protected]>

Sem-ver: bugfix Have mypy install stubs as part of ci
Signed-off-by: David Black <[email protected]>

Sem-ver: bugfix Typing fix ups - this commit will be merged into an earlier one
Signed-off-by: David Black <[email protected]>

Remove the unused UnitTestProtocol code

Signed-off-by: David Black <[email protected]>

Author: dbaxa

.github/workflows/build.yml

@@ -18,6 +18,8 @@ jobs:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
+        pip install -r requirements.txt
+        pip install -r test-requirements.txt
         python -m pip install --upgrade pip wheel setuptools
         pip install -q ruff mypy
     - name: Lint

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/py.typed

@@ -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/signer.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."""
@@ -65,18 +65,6 @@ def get_new_private_key_in_pem_format(self) -> bytes:
         raise NotImplementedError("not implemented.")
 
 
-class UnitTestProtocol(Protocol):
-    def assertEqual(self, a, b): ...
-
-    def assertIsNotNone(self, a): ...
-
-    def assertTrue(self, a): ...
-
-    def assertIn(self, a, b): ...
-
-    def assertNotEqual(self, a, b): ...
-
-
 class KeyMixInProtocol(Protocol):
     @property
     def algorithm(self) -> str: ...