WIP

dbaxa · dbaxa · commit bb349b47bf8e · 2025-09-24T16:34:54.000+10:00
Signed-off-by: David Black &lt;dblack@atlassian.com&gt;
diff --git a/atlassian_jwt_auth/auth.py b/atlassian_jwt_auth/auth.py
@@ -26,4 +26,4 @@ def create(cls, issuer: str, key_identifier: Union[KeyIdentifier, str], private_
 
     def _get_header_value(self) -> bytes:
         return b'Bearer ' + self._signer.generate_jwt(
-            self._audience, additional_claims=self._additional_claims)
+            self._audience, additional_claims=self._additional_claims).encode("utf-8")
diff --git a/atlassian_jwt_auth/contrib/aiohttp/auth.py b/atlassian_jwt_auth/contrib/aiohttp/auth.py
@@ -19,7 +19,7 @@ def encode(self) -> str:
 
 
 def create_jwt_auth(
-        issuer: str, key_identifier: Union[KeyIdentifier, str], private_key_pem: str, audience: str, **kwargs: Any) -> JWTAuth:
+        issuer: str, key_identifier: Union[KeyIdentifier, str], private_key_pem: str, audience: str, **kwargs: Any) -> BaseJWTAuth:
     """Instantiate a JWTAuth while creating the signer inline"""
     return JWTAuth.create(
         issuer, key_identifier, private_key_pem, audience, **kwargs)
diff --git a/atlassian_jwt_auth/contrib/aiohttp/key.py b/atlassian_jwt_auth/contrib/aiohttp/key.py
@@ -1,7 +1,7 @@
 import asyncio
 import urllib.parse
 from asyncio import AbstractEventLoop
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Awaitable
 
 import aiohttp
 
@@ -22,7 +22,7 @@ def __init__(self, base_url: str, *,
         self.loop = loop
         super().__init__(base_url)
 
-    def _get_session(self) -> aiohttp.ClientSession:
+    def _get_session(self) -> aiohttp.ClientSession:  # type: ignore[override]
         if HTTPSPublicKeyRetriever._class_session is None:
             HTTPSPublicKeyRetriever._class_session = aiohttp.ClientSession(
                 loop=self.loop)
@@ -43,11 +43,11 @@ def _convert_proxies_to_proxy_arg(
         return requests_kwargs
 
     async def _retrieve(
-            self, url: str, requests_kwargs: Dict[Any, Any]) -> str:
+            self, url: str, requests_kwargs: Dict[Any, Any]) -> Awaitable[str]:
         requests_kwargs = self._convert_proxies_to_proxy_arg(
             url, requests_kwargs)
         try:
-            resp = await self._session.get(url, headers={'accept':
+            resp = await self._session.get(url, headers={'accept': # type: ignore[misc]
                                                          PEM_FILE_TYPE},
                                            **requests_kwargs)
             resp.raise_for_status()
diff --git a/atlassian_jwt_auth/contrib/aiohttp/verifier.py b/atlassian_jwt_auth/contrib/aiohttp/verifier.py
@@ -1,14 +1,14 @@
 import asyncio
-from typing import Any, Dict
+from typing import Any, Dict, Coroutine, Union
 
 import jwt
 
 from atlassian_jwt_auth import key
 from atlassian_jwt_auth.verifier import JWTAuthVerifier as _JWTAuthVerifier
 
 
-class JWTAuthVerifier(_JWTAuthVerifier):
-    async def verify_jwt(self, a_jwt: str, audience: str,
+class JWTAuthVerifier(_JWTAuthVerifier):  # type: ignore[override]
+    async def verify_jwt(self, a_jwt: str, audience: str,  # type: ignore[override]
                          leeway: int = 0, **requests_kwargs: Any) -> Dict[Any, Any]:
         """Verify if the token is correct
 
diff --git a/atlassian_jwt_auth/contrib/django/middleware.py b/atlassian_jwt_auth/contrib/django/middleware.py
@@ -49,6 +49,7 @@ def process_request(self, request) -> Optional[str]:
             request.META['HTTP_AUTHORIZATION'] = orig_auth
         if asap_auth is not None:
             request.META[self.xauth] = asap_auth
+        return None
 
     def process_view(self, request: Any, view_func: Callable,
                      view_args: Any, view_kwargs: Any) -> None:
diff --git a/atlassian_jwt_auth/contrib/requests.py b/atlassian_jwt_auth/contrib/requests.py
@@ -12,8 +12,8 @@
 class JWTAuth(AuthBase, BaseJWTAuth):
     """Adds a JWT bearer token to the request per the ASAP specification"""
 
-    def __call__(self, r: requests.Request):
-        r.headers['Authorization'] = self._get_header_value()
+    def __call__(self, r:  requests.models.PreparedRequest) -> requests.models.PreparedRequest:
+        r.headers['Authorization'] = self._get_header_value() # type: ignore[assignment]
         return r
 
 
diff --git a/atlassian_jwt_auth/contrib/tests/aiohttp/test_auth.py b/atlassian_jwt_auth/contrib/tests/aiohttp/test_auth.py
@@ -13,7 +13,7 @@ class BaseAuthTest(test_requests.BaseRequestsTest):
     def _get_auth_header(self, auth) -> bytes:
         return auth.encode().encode('latin1')
 
-    def create_jwt_auth(self, *args: Any, **kwargs: Dict):
+    def create_jwt_auth(self, *args: Any, **kwargs: Any):
         return create_jwt_auth(*args, **kwargs)
 
 
diff --git a/atlassian_jwt_auth/contrib/tests/aiohttp/test_public_key_provider.py b/atlassian_jwt_auth/contrib/tests/aiohttp/test_public_key_provider.py
@@ -23,8 +23,8 @@ class DummyHTTPSPublicKeyRetriever(HTTPSPublicKeyRetriever):
     def set_headers(self, headers) -> None:
         self._session.get.return_value.headers.update(headers)
 
-    def set_text(self, text: str) -> None:
-        self._session.get.return_value.text.return_value = text
+    def set_text(self, text: str | bytes) -> None:
+        self._session.get.return_value.text.return_value = text  # type: ignore[attr-defined]
 
     def _get_session(self) -> Mock:
         session = Mock(spec=aiohttp.ClientSession)
@@ -41,16 +41,16 @@ class BaseHTTPSPublicKeyRetrieverTestMixin(object):
     """Tests for aiohttp.HTTPSPublicKeyRetriever class for RS256 algorithm"""
 
     def setUp(self) -> None:
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        self._private_key_pem = self.get_new_private_key_in_pem_format() # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem)
         self.base_url = 'https://example.com'
 
     async def test_retrieve(self) -> None:
         """Check if retrieve method returns public key"""
         retriever = DummyHTTPSPublicKeyRetriever(self.base_url)
-        retriever.set_text(self._public_key_pem)
-        self.assertEqual(
+        retriever.set_text(self._public_key_pem) # type: ignore[misc]
+        self.assertEqual( # type: ignore[attr-defined]
             await retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -61,7 +61,7 @@ async def test_retrieve_with_charset_in_content_type_h(self) -> None:
         retriever.set_text(self._public_key_pem)
         retriever.set_headers(headers)
 
-        self.assertEqual(
+        self.assertEqual( # type: ignore[attr-defined]
             await retriever.retrieve('example/eg'),
             self._public_key_pem)
 
@@ -71,10 +71,10 @@ async def test_retrieve_fails_with_different_content_type(self) -> None:
         """
         headers = {'content-type': 'different/not-supported'}
         retriever = DummyHTTPSPublicKeyRetriever(self.base_url)
-        retriever.set_text(self._public_key_pem)
+        retriever.set_text(self._public_key_pem) # type: ignore[arg-type]
         retriever.set_headers(headers)
 
-        with self.assertRaises(ValueError):
+        with self.assertRaises(ValueError): # type: ignore[attr-defined]
             await retriever.retrieve('example/eg')
 
     async def test_retrieve_session_uses_env_proxy(self) -> None:
@@ -87,9 +87,9 @@ async def test_retrieve_session_uses_env_proxy(self) -> None:
             proxy_location)
         with mock.patch.dict(os.environ, proxy_dict, clear=True):
             retriever = DummyHTTPSPublicKeyRetriever(self.base_url)
-            self.assertEqual(retriever._proxies, expected_proxies)
+            self.assertEqual(retriever._proxies, expected_proxies) # type: ignore[attr-defined]
             await retriever.retrieve(key_id)
-            retriever._session.get.assert_called_once_with(
+            retriever._session.get.assert_called_once_with( #type: ignore[attr-defined]
                 f'{self.base_url}/{key_id}', headers={'accept': PEM_FILE_TYPE},
                 proxy=expected_proxies[self.base_url.split(':')[0]]
             )
diff --git a/atlassian_jwt_auth/contrib/tests/test_requests.py b/atlassian_jwt_auth/contrib/tests/test_requests.py
@@ -1,6 +1,6 @@
 import unittest
 from datetime import timedelta
-from typing import Any
+from typing import Any, Type
 
 import jwt
 from requests import Request
@@ -14,7 +14,7 @@
 class BaseRequestsTest(object):
 
     """ tests for the contrib.requests.JWTAuth class """
-    auth_cls = JWTAuth
+    auth_cls: Type[BaseJWTAuth] = JWTAuth
 
     def setUp(self) -> None:
         self._private_key_pem = self.get_new_private_key_in_pem_format()
@@ -33,7 +33,7 @@ def assert_authorization_header_is_valid(self, auth) -> Any:
         return jwt.decode(bearer, self._public_key_pem.decode(),
                           audience='audience', algorithms=algorithms)
 
-    def _get_auth_header(self, auth) -> str:
+    def _get_auth_header(self, auth) -> bytes:
         request = auth(Request())
         auth_header = request.headers['Authorization']
         return auth_header
diff --git a/atlassian_jwt_auth/contrib/tests/utils.py b/atlassian_jwt_auth/contrib/tests/utils.py
@@ -1,22 +1,23 @@
-from typing import Any, Dict
+from typing import Any, Dict, Type
 
 import requests
 
 import atlassian_jwt_auth
 from atlassian_jwt_auth import JWTAuthVerifier
+from atlassian_jwt_auth.key import BasePublicKeyRetriever
 
 
-def get_static_retriever_class(keys: Dict[str, Any]):
+def get_static_retriever_class(keys: Dict[str, Any]) -> Type[BasePublicKeyRetriever]:
 
-    class StaticPublicKeyRetriever(object):
+    class StaticPublicKeyRetriever(BasePublicKeyRetriever):
         """ Retrieves a key from a static dict of public keys
         (for use in tests only) """
 
         def __init__(self, *args: Any, **
-                     kwargs: Any) -> requests.PreparedRequest:
+                     kwargs: Any) -> None:
             self.keys: Dict[str, Any] = keys
 
-        def retrieve(self, key_identifier, **requests_kwargs) -> str:
+        def retrieve(self, key_identifier, **requests_kwargs) -> Any:
             return self.keys[key_identifier.key_id]
 
     return StaticPublicKeyRetriever
diff --git a/atlassian_jwt_auth/frameworks/common/asap.py b/atlassian_jwt_auth/frameworks/common/asap.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Iterable, Optional, Union
 
 from jwt.exceptions import InvalidIssuerError, InvalidTokenError
 
@@ -28,7 +28,7 @@ def _process_asap_token(request: Any, backend: Backend, settings: SettingsDict,
         if verifier is None:
             verifier = backend.get_verifier(settings=settings)
         asap_claims = verifier.verify_jwt(
-            token,
+            token.decode('utf-8') if isinstance(token, bytes) else token,
             settings.ASAP_VALID_AUDIENCE,
             leeway=settings.ASAP_VALID_LEEWAY,
         )
@@ -49,7 +49,7 @@ def _process_asap_token(request: Any, backend: Backend, settings: SettingsDict,
             # will return 403 for a missing file to avoid leaking
             # information.
             raise
-            logger.warning('Could not retrieve the matching public key')
+        logger.warning('Could not retrieve the matching public key')
         error_response = backend.get_401_response(
             'Unauthorized: Key not found', request=request
         )
@@ -79,11 +79,12 @@ def _process_asap_token(request: Any, backend: Backend, settings: SettingsDict,
 
     if error_response is not None and settings.ASAP_REQUIRED:
         return error_response
+    return None
 
 
 def _verify_issuers(
         asap_claims: Dict[Any, Any], issuers: Optional[Iterable[str]] = None) -> None:
     """Verify that the issuer in the claims is valid and is expected."""
     claim_iss = asap_claims.get('iss')
-    if issuers and claim_iss not in issuers:
+    if issuers is not None and claim_iss is not None and claim_iss not in issuers:
         raise InvalidIssuerError
diff --git a/atlassian_jwt_auth/frameworks/common/decorators.py b/atlassian_jwt_auth/frameworks/common/decorators.py
@@ -9,8 +9,8 @@
 
 
 def _with_asap(
-    func: Callable = None,
-    backend: Backend = None,
+    func: Optional[Callable] = None,
+    backend: Optional[Backend] = None,
     issuers: Optional[Iterable[str]] = None,
     required: bool = True,
     subject_should_match_issuer: Optional[bool] = None,
@@ -48,7 +48,7 @@ def with_asap_wrapper(*args: Any, **kwargs: Any) -> Optional[str]:
 
 
 def _restrict_asap(
-    func: Callable = None,
+    func: Optional[Callable] = None,
     backend: Optional[Backend] = None,
     issuers: Optional[Iterable[str]] = None,
     required: bool = True,
@@ -61,6 +61,8 @@ def _restrict_asap(
     def restrict_asap_decorator(func: Callable) -> Optional[Any]:
         @wraps(func)
         def restrict_asap_wrapper(request, *args, **kwargs) -> Any:
+            if backend is None:
+                raise ValueError("Backend cannot be None")
             settings = _update_settings_from_kwargs(
                 backend.settings,
                 issuers=issuers,
@@ -71,19 +73,23 @@ def restrict_asap_wrapper(request, *args, **kwargs) -> Any:
             error_response = None
 
             if required and not asap_claims:
-                return backend.get_401_response(
-                    "Unauthorized", request=request)
+                if backend is not None:
+                    return backend.get_401_response(
+                        "Unauthorized", request=request)
 
             try:
-                _verify_issuers(asap_claims, settings.ASAP_VALID_ISSUERS)
+                if asap_claims is not None:
+                    _verify_issuers(asap_claims, settings.ASAP_VALID_ISSUERS)
             except InvalidIssuerError:
-                error_response = backend.get_403_response(
-                    "Forbidden: Invalid token issuer", request=request
-                )
+                if backend is not None:
+                    error_response = backend.get_403_response(
+                        "Forbidden: Invalid token issuer", request=request
+                    )
             except InvalidTokenError:
-                error_response = backend.get_401_response(
-                    "Unauthorized: Invalid token", request=request
-                )
+                if backend is not None:
+                    error_response = backend.get_401_response(
+                        "Unauthorized: Invalid token", request=request
+                    )
 
             if error_response and required:
                 return error_response
diff --git a/atlassian_jwt_auth/frameworks/common/utils.py b/atlassian_jwt_auth/frameworks/common/utils.py
@@ -19,7 +19,7 @@ def _hash_key(self) -> frozenset[Any]:
             keys_and_values.append("%s %s" % (key, hash(value)))
         return frozenset(keys_and_values)
 
-    def __hash__(self) -> int:
+    def __hash__(self) -> int: # type: ignore[override]
         return hash(self._hash_key())
 
     def __eq__(self, other) -> bool:
diff --git a/atlassian_jwt_auth/frameworks/django/decorators.py b/atlassian_jwt_auth/frameworks/django/decorators.py
@@ -5,7 +5,7 @@
 from .backend import DjangoBackend
 
 
-def with_asap(func: Optional[Callable] = None, issuers: Optional[Iterable[str]] = None, required: Optional[bool] = None,
+def with_asap(func: Optional[Callable] = None, issuers: Optional[Iterable[str]] = None, required: bool=True,
               subject_should_match_issuer: Optional[bool] = None) -> Callable:
     """Decorator to allow endpoint-specific ASAP authentication.
 
@@ -29,7 +29,7 @@ def with_asap(func: Optional[Callable] = None, issuers: Optional[Iterable[str]]
     )
 
 
-def restrict_asap(func: Optional[Callable] = None, backend: Optional[Backend] = None, issuers: Iterable[str] = None,
+def restrict_asap(func: Optional[Callable] = None, backend: Optional[Backend] = None, issuers: Optional[Iterable[str]] = None,
                   required: bool = True, subject_should_match_issuer: Optional[bool] = None) -> Callable:
     """Decorator to allow endpoint-specific ASAP authorization policies.
 
@@ -48,6 +48,7 @@ def restrict_asap(func: Optional[Callable] = None, backend: Optional[Backend] =
                                                 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
diff --git a/atlassian_jwt_auth/frameworks/django/middleware.py b/atlassian_jwt_auth/frameworks/django/middleware.py
@@ -38,3 +38,4 @@ def process_request(self, request: HttpRequest) -> Optional[str]:
         )
         if error_response is not None:
             return error_response
+        return None
diff --git a/atlassian_jwt_auth/frameworks/django/tests/test_django.py b/atlassian_jwt_auth/frameworks/django/tests/test_django.py
@@ -1,5 +1,5 @@
 import os
-from typing import Optional
+from typing import Optional, Dict, Any
 
 import django
 from django.test.testcases import SimpleTestCase
@@ -24,16 +24,17 @@ def setUpClass(cls) -> None:
             'atlassian_jwt_auth.frameworks.django.tests.settings')
 
         django.setup()
-        super(DjangoAsapMixin, cls).setUpClass()
+
+        super(DjangoAsapMixin, cls).setUpClass() # type: ignore[misc]
 
     @classmethod
     def tearDownClass(cls) -> None:
-        super(DjangoAsapMixin, cls).tearDownClass()
+        super(DjangoAsapMixin, cls).tearDownClass() # type: ignore[misc]
         del os.environ['DJANGO_SETTINGS_MODULE']
 
     def setUp(self) -> None:
-        super(DjangoAsapMixin, self).setUp()
-        self._private_key_pem = self.get_new_private_key_in_pem_format()
+        super(DjangoAsapMixin, self).setUp() # type: ignore[misc]
+        self._private_key_pem = self.get_new_private_key_in_pem_format() # type: ignore[attr-defined]
         self._public_key_pem = utils.get_public_key_pem_for_private_key_pem(
             self._private_key_pem
         )
@@ -42,7 +43,7 @@ def setUp(self) -> None:
             'client-app/key01': self._public_key_pem
         })
 
-        self.test_settings = {
+        self.test_settings: Dict[Any, Any] = {
             'ASAP_KEY_RETRIEVER_CLASS': self.retriever
         }
 
diff --git a/atlassian_jwt_auth/frameworks/flask/backend.py b/atlassian_jwt_auth/frameworks/flask/backend.py
diff --git a/atlassian_jwt_auth/frameworks/flask/decorators.py b/atlassian_jwt_auth/frameworks/flask/decorators.py
diff --git a/atlassian_jwt_auth/key.py b/atlassian_jwt_auth/key.py
diff --git a/atlassian_jwt_auth/signer.py b/atlassian_jwt_auth/signer.py
diff --git a/atlassian_jwt_auth/verifier.py b/atlassian_jwt_auth/verifier.py
diff --git a/pyproject.toml b/pyproject.toml
