WIP: add typing

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

Author: dbaxa

atlassian_jwt_auth/__init__.py

@@ -1,15 +1,6 @@
 from atlassian_jwt_auth.algorithms import get_permitted_algorithm_names  # noqa
-
+from atlassian_jwt_auth.key import (HTTPSPublicKeyRetriever,  # noqa
+                                    KeyIdentifier)
 from atlassian_jwt_auth.signer import (  # noqa
-    create_signer,
-    create_signer_from_file_private_key_repository,
-)
-
-from atlassian_jwt_auth.key import (  # noqa
-    KeyIdentifier,
-    HTTPSPublicKeyRetriever,
-)
-
-from atlassian_jwt_auth.verifier import (  # noqa
-    JWTAuthVerifier,
-)
+    create_signer, create_signer_from_file_private_key_repository)
+from atlassian_jwt_auth.verifier import JWTAuthVerifier  # noqa

atlassian_jwt_auth/auth.py

@@ -1,6 +1,6 @@
 from __future__ import absolute_import
 
-from typing import Any
+from typing import Any, Union
 
 import atlassian_jwt_auth
 from atlassian_jwt_auth import KeyIdentifier
@@ -10,7 +10,8 @@
 class BaseJWTAuth(object):
     """Adds a JWT bearer token to the request per the ASAP specification"""
 
-    def __init__(self, signer: JWTAuthSigner, audience: str, *args: Any, **kwargs: Any) -> None:
+    def __init__(self, signer: JWTAuthSigner, audience: str,
+                 *args: Any, **kwargs: Any) -> None:
         self._audience = audience
         self._signer = signer
         self._additional_claims = kwargs.get('additional_claims', {})

atlassian_jwt_auth/contrib/aiohttp/__init__.py

@@ -4,6 +4,7 @@
 if sys.version_info >= (3, 5):
     try:
         import aiohttp  # noqa
+
         from .auth import JWTAuth  # noqa
         from .key import HTTPSPublicKeyRetriever  # noqa
         from .verifier import JWTAuthVerifier  # noqa

atlassian_jwt_auth/contrib/aiohttp/auth.py

@@ -1,4 +1,4 @@
-from typing import Union, Any, Dict
+from typing import Any, Dict, Union
 
 from aiohttp import BasicAuth
 
@@ -14,7 +14,7 @@ class JWTAuth(BaseJWTAuth, BasicAuth):
     def __new__(cls, *args, **kwargs):
         return super().__new__(cls, '')
 
-    def encode(self) -> str :
+    def encode(self) -> str:
         return self._get_header_value().decode(self.encoding)
 
 

atlassian_jwt_auth/contrib/aiohttp/key.py

@@ -1,22 +1,22 @@
 import asyncio
 import urllib.parse
 from asyncio import AbstractEventLoop
-from typing import Dict, Any, Optional
+from typing import Any, Dict, Optional
 
 import aiohttp
 
 from atlassian_jwt_auth.exceptions import PublicKeyRetrieverException
-from atlassian_jwt_auth.key import (
-    PEM_FILE_TYPE,
+from atlassian_jwt_auth.key import PEM_FILE_TYPE
+from atlassian_jwt_auth.key import \
     HTTPSPublicKeyRetriever as _HTTPSPublicKeyRetriever
-)
 
 
 class HTTPSPublicKeyRetriever(_HTTPSPublicKeyRetriever):
     """A class for retrieving JWT public keys with aiohttp"""
     _class_session = None
 
-    def __init__(self, base_url:str, *, loop: Optional[AbstractEventLoop]=None) -> None:
+    def __init__(self, base_url: str, *,
+                 loop: Optional[AbstractEventLoop] = None) -> None:
         if loop is None:
             loop = asyncio.get_event_loop()
         self.loop = loop
@@ -28,7 +28,8 @@ def _get_session(self) -> aiohttp.ClientSession:
                 loop=self.loop)
         return HTTPSPublicKeyRetriever._class_session
 
-    def _convert_proxies_to_proxy_arg(self, url: str, requests_kwargs: Dict[Any, Any]) -> Dict[str, Any]:
+    def _convert_proxies_to_proxy_arg(
+            self, url: str, requests_kwargs: Dict[Any, Any]) -> Dict[str, Any]:
         """ returns a modified requests_kwargs dict that contains proxy
             information in a form that aiohttp accepts
             (it wants proxy information instead of a dict of proxies).
@@ -41,7 +42,8 @@ def _convert_proxies_to_proxy_arg(self, url: str, requests_kwargs: Dict[Any, Any
             requests_kwargs['proxy'] = proxy
         return requests_kwargs
 
-    async def _retrieve(self, url:str , requests_kwargs: Dict[Any, Any]) -> str:
+    async def _retrieve(
+            self, url: str, requests_kwargs: Dict[Any, Any]) -> str:
         requests_kwargs = self._convert_proxies_to_proxy_arg(
             url, requests_kwargs)
         try:

atlassian_jwt_auth/contrib/aiohttp/verifier.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Dict, Any
+from typing import Any, Dict
 
 import jwt
 
@@ -8,7 +8,8 @@
 
 
 class JWTAuthVerifier(_JWTAuthVerifier):
-    async def verify_jwt(self, a_jwt: str, audience: str, leeway: int=0, **requests_kwargs: Any) -> Dict[Any, Any]:
+    async def verify_jwt(self, a_jwt: str, audience: str,
+                         leeway: int = 0, **requests_kwargs: Any) -> Dict[Any, Any]:
         """Verify if the token is correct
 
         Returns:

atlassian_jwt_auth/contrib/django/__init__.py

@@ -1,6 +1,5 @@
 import warnings
 
-
 warnings.warn(
     "The atlassian_jwt_auth.contrib.django package is deprecated in 4.0.0 "
     "in favour of atlassian_jwt_auth.frameworks.django.",

atlassian_jwt_auth/contrib/django/decorators.py

@@ -1,13 +1,17 @@
 from collections.abc import Callable
 from functools import wraps
+from typing import Iterable, Optional, Sequence
 
 from django.http.response import HttpResponse
 
 from atlassian_jwt_auth.frameworks.django.decorators import with_asap
-from typing import Optional, Sequence, Iterable
 
 
-def validate_asap(issuers: Optional[Iterable[str]]=None, subjects: Optional[Iterable[str]]=None, required: bool=True) -> Callable[]:
+def validate_asap(
+    issuers: Optional[Iterable[str]] = None,
+    subjects: Optional[Iterable[str]] = None,
+    required: bool = True,
+) -> Callable:
     """Decorator to allow endpoint-specific ASAP authorization, assuming ASAP
     authentication has already occurred.
 
@@ -18,39 +22,47 @@ def validate_asap(issuers: Optional[Iterable[str]]=None, subjects: Optional[Iter
     :param boolean required: Whether or not to require ASAP on this endpoint.
         Note that requirements will be still be verified if claims are present.
     """
+
     def validate_asap_decorator(func):
         @wraps(func)
         def validate_asap_wrapper(request, *args, **kwargs):
-            asap_claims = getattr(request, 'asap_claims', None)
+            asap_claims = getattr(request, "asap_claims", None)
             if required and not asap_claims:
-                message = 'Unauthorized: Invalid or missing token'
+                message = "Unauthorized: Invalid or missing token"
                 response = HttpResponse(message, status=401)
-                response['WWW-Authenticate'] = 'Bearer'
+                response["WWW-Authenticate"] = "Bearer"
                 return response
 
             if asap_claims:
-                iss = asap_claims['iss']
+                iss = asap_claims["iss"]
                 if issuers and iss not in issuers:
-                    message = 'Forbidden: Invalid token issuer'
+                    message = "Forbidden: Invalid token issuer"
                     return HttpResponse(message, status=403)
 
-                sub = asap_claims.get('sub')
+                sub = asap_claims.get("sub")
                 if subjects and sub not in subjects:
-                    message = 'Forbidden: Invalid token subject'
+                    message = "Forbidden: Invalid token subject"
                     return HttpResponse(message, status=403)
 
             return func(request, *args, **kwargs)
 
         return validate_asap_wrapper
+
     return validate_asap_decorator
 
 
-def requires_asap(issuers: Optional[Iterable[str]]=None, subject_should_match_issuer: Optional[bool]=None, func: Optional[Callable]=None) ->:
+def requires_asap(
+    issuers: Optional[Iterable[str]] = None,
+    subject_should_match_issuer: Optional[bool] = None,
+    func: Optional[Callable] = None,
+) -> Callable:
     """Decorator for Django endpoints to require ASAP
 
     :param list issuers: *required The 'iss' claims that this endpoint is from.
     """
-    return with_asap(func=func,
-                     required=True,
-                     issuers=issuers,
-                     subject_should_match_issuer=subject_should_match_issuer)
+    return with_asap(
+        func=func,
+        required=True,
+        issuers=issuers,
+        subject_should_match_issuer=subject_should_match_issuer,
+    )

atlassian_jwt_auth/contrib/django/middleware.py

@@ -1,11 +1,10 @@
-from typing import Optional, Any, Callable
+from typing import Any, Callable, Optional
 
 from django.conf import settings
 from django.utils.deprecation import MiddlewareMixin
 
-from atlassian_jwt_auth.frameworks.django.middleware import (
+from atlassian_jwt_auth.frameworks.django.middleware import \
     OldStyleASAPMiddleware
-)
 
 
 class ProxiedAsapMiddleware(OldStyleASAPMiddleware, MiddlewareMixin):
@@ -14,7 +13,7 @@ class ProxiedAsapMiddleware(OldStyleASAPMiddleware, MiddlewareMixin):
 
     This must come before any authentication middleware."""
 
-    def __init__(self, get_response: Optional[Any]=None) -> None:
+    def __init__(self, get_response: Optional[Any] = None) -> None:
         super(ProxiedAsapMiddleware, self).__init__()
         self.get_response = get_response
 
@@ -51,7 +50,8 @@ def process_request(self, request) -> Optional[str]:
         if asap_auth is not None:
             request.META[self.xauth] = asap_auth
 
-    def process_view(self, request: Any, view_func: Callable, view_args: Any, view_kwargs: Any) -> None:
+    def process_view(self, request: Any, view_func: Callable,
+                     view_args: Any, view_kwargs: Any) -> None:
         if not hasattr(request, 'asap_forwarded'):
             return None
 

atlassian_jwt_auth/contrib/flask_app/__init__.py

@@ -2,7 +2,6 @@
 
 from .decorators import requires_asap  # noqa
 
-
 warnings.warn(
     "The atlassian_jwt_auth.contrib.flask_app package is deprecated in 4.0.0 "
     "in favour of atlassian_jwt_auth.frameworks.flask.",