WIP: Ruff format

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

Author: dbaxa

atlassian_jwt_auth/__init__.py

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

atlassian_jwt_auth/algorithms.py

@@ -1,13 +1,13 @@
 def get_permitted_algorithm_names():
-    """ returns permitted algorithm names. """
+    """returns permitted algorithm names."""
     return [
-        'RS256',
-        'RS384',
-        'RS512',
-        'ES256',
-        'ES384',
-        'ES512',
-        'PS256',
-        'PS384',
-        'PS512'
+        "RS256",
+        "RS384",
+        "RS512",
+        "ES256",
+        "ES384",
+        "ES512",
+        "PS256",
+        "PS384",
+        "PS512",
     ]

atlassian_jwt_auth/auth.py

@@ -10,20 +10,29 @@
 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', {})
+        self._additional_claims = kwargs.get("additional_claims", {})
 
     @classmethod
-    def create(cls, issuer: str, key_identifier: Union[KeyIdentifier, str], private_key_pem: str, audience: str,
-               **kwargs: Any) -> "BaseJWTAuth":
+    def create(
+        cls,
+        issuer: str,
+        key_identifier: Union[KeyIdentifier, str],
+        private_key_pem: str,
+        audience: str,
+        **kwargs: Any,
+    ) -> "BaseJWTAuth":
         """Instantiate a JWTAuth while creating the signer inline"""
-        signer = atlassian_jwt_auth.create_signer(issuer, key_identifier,
-                                                  private_key_pem, **kwargs)
+        signer = atlassian_jwt_auth.create_signer(
+            issuer, key_identifier, private_key_pem, **kwargs
+        )
         return cls(signer, audience)
 
     def _get_header_value(self) -> bytes:
-        return b'Bearer ' + self._signer.generate_jwt(
-            self._audience, additional_claims=self._additional_claims).encode("utf-8")
+        return b"Bearer " + self._signer.generate_jwt(
+            self._audience, additional_claims=self._additional_claims
+        ).encode("utf-8")

atlassian_jwt_auth/contrib/aiohttp/__init__.py

@@ -1,4 +1,5 @@
 """Provide asyncio support"""
+
 import sys
 
 if sys.version_info >= (3, 5):
@@ -10,6 +11,7 @@
         from .verifier import JWTAuthVerifier  # noqa
     except ImportError as e:
         import warnings
+
         warnings.warn(str(e))
 
 

atlassian_jwt_auth/contrib/aiohttp/auth.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Union
+from typing import Any, Union
 
 from aiohttp import BasicAuth
 
@@ -11,15 +11,20 @@ class JWTAuth(BaseJWTAuth, BasicAuth):
 
     It should be aiohttp.BasicAuth subclass, so redefine its `__new__` method.
     """
+
     def __new__(cls, *args, **kwargs):
-        return super().__new__(cls, '')
+        return super().__new__(cls, "")
 
     def encode(self) -> str:
         return self._get_header_value().decode(self.encoding)
 
 
 def create_jwt_auth(
-        issuer: str, key_identifier: Union[KeyIdentifier, str], private_key_pem: str, audience: str, **kwargs: Any) -> BaseJWTAuth:
+    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)
+    return JWTAuth.create(issuer, key_identifier, private_key_pem, audience, **kwargs)

atlassian_jwt_auth/contrib/aiohttp/key.py

@@ -7,16 +7,17 @@
 
 from atlassian_jwt_auth.exceptions import PublicKeyRetrieverException
 from atlassian_jwt_auth.key import PEM_FILE_TYPE
-from atlassian_jwt_auth.key import \
-    HTTPSPublicKeyRetriever as _HTTPSPublicKeyRetriever
+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
@@ -25,34 +26,41 @@ def __init__(self, base_url: str, *,
     def _get_session(self) -> aiohttp.ClientSession:  # type: ignore[override]
         if HTTPSPublicKeyRetriever._class_session is None:
             HTTPSPublicKeyRetriever._class_session = aiohttp.ClientSession(
-                loop=self.loop)
+                loop=self.loop
+            )
         return HTTPSPublicKeyRetriever._class_session
 
     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).
+        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).
         """
         proxy = None
-        if 'proxies' in requests_kwargs:
+        if "proxies" in requests_kwargs:
             scheme = urllib.parse.urlparse(url).scheme
-            proxy = requests_kwargs['proxies'].get(scheme, None)
-            del requests_kwargs['proxies']
-            requests_kwargs['proxy'] = proxy
+            proxy = requests_kwargs["proxies"].get(scheme, None)
+            del requests_kwargs["proxies"]
+            requests_kwargs["proxy"] = proxy
         return requests_kwargs
 
     async def _retrieve(
-            self, url: str, requests_kwargs: Dict[Any, Any]) -> Awaitable[str]:
-        requests_kwargs = self._convert_proxies_to_proxy_arg(
-            url, requests_kwargs)
+        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': # type: ignore[misc]
-                                                         PEM_FILE_TYPE},
-                                           **requests_kwargs)
+            resp = await self._session.get(
+                url,
+                headers={
+                    "accept":  # type: ignore[misc]
+                    PEM_FILE_TYPE
+                },
+                **requests_kwargs,
+            )
             resp.raise_for_status()
-            self._check_content_type(url, resp.headers['content-type'])
+            self._check_content_type(url, resp.headers["content-type"])
             return await resp.text()
         except aiohttp.ClientError as e:
-            status_code = getattr(e, 'code', None)
+            status_code = getattr(e, "code", None)
             raise PublicKeyRetrieverException(e, status_code=status_code)

atlassian_jwt_auth/contrib/aiohttp/verifier.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Any, Dict, Coroutine, Union
+from typing import Any, Dict
 
 import jwt
 
@@ -8,8 +8,13 @@
 
 
 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]:
+    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
 
         Returns:
@@ -24,8 +29,8 @@ async def verify_jwt(self, a_jwt: str, audience: str,  # type: ignore[override]
         if asyncio.iscoroutine(public_key):
             public_key = await public_key
 
-        alg = jwt.get_unverified_header(a_jwt).get('alg', None)
+        alg = jwt.get_unverified_header(a_jwt).get("alg", None)
         public_key_obj = self._load_public_key(public_key, alg)
         return self._decode_jwt(
-            a_jwt, key_identifier, public_key_obj,
-            audience=audience, leeway=leeway)
+            a_jwt, key_identifier, public_key_obj, audience=audience, leeway=leeway
+        )

atlassian_jwt_auth/contrib/django/__init__.py

@@ -3,5 +3,6 @@
 warnings.warn(
     "The atlassian_jwt_auth.contrib.django package is deprecated in 4.0.0 "
     "in favour of atlassian_jwt_auth.frameworks.django.",
-    DeprecationWarning, stacklevel=2
+    DeprecationWarning,
+    stacklevel=2,
 )

atlassian_jwt_auth/contrib/django/decorators.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Iterable, Optional, Sequence
+from typing import Iterable, Optional
 
 from django.http.response import HttpResponse
 

atlassian_jwt_auth/contrib/django/middleware.py

@@ -3,8 +3,7 @@
 from django.conf import settings
 from django.utils.deprecation import MiddlewareMixin
 
-from atlassian_jwt_auth.frameworks.django.middleware import \
-    OldStyleASAPMiddleware
+from atlassian_jwt_auth.frameworks.django.middleware import OldStyleASAPMiddleware
 
 
 class ProxiedAsapMiddleware(OldStyleASAPMiddleware, MiddlewareMixin):
@@ -19,17 +18,17 @@ def __init__(self, get_response: Optional[Any] = None) -> None:
 
         # Rely on this header to tell us if a request has been forwarded
         # from an ASAP-enabled service; will overwrite X-Forwarded-For
-        self.xfwd = getattr(settings, 'ASAP_PROXIED_FORWARDED_FOR_HEADER',
-                            'HTTP_X_ASAP_FORWARDED_FOR')
+        self.xfwd = getattr(
+            settings, "ASAP_PROXIED_FORWARDED_FOR_HEADER", "HTTP_X_ASAP_FORWARDED_FOR"
+        )
 
         # This header won't always be set, i.e. some users will be anonymous
-        self.xauth = getattr(settings, 'ASAP_PROXIED_AUTHORIZATION_HEADER',
-                             'HTTP_X_ASAP_AUTHORIZATION')
+        self.xauth = getattr(
+            settings, "ASAP_PROXIED_AUTHORIZATION_HEADER", "HTTP_X_ASAP_AUTHORIZATION"
+        )
 
     def process_request(self, request) -> Optional[str]:
-        error_response = super(ProxiedAsapMiddleware, self).process_request(
-            request
-        )
+        error_response = super(ProxiedAsapMiddleware, self).process_request(request)
 
         if error_response:
             return error_response
@@ -39,28 +38,29 @@ def process_request(self, request) -> Optional[str]:
             return None
 
         request.asap_forwarded = True
-        request.META['HTTP_X_FORWARDED_FOR'] = forwarded_for
+        request.META["HTTP_X_FORWARDED_FOR"] = forwarded_for
 
-        asap_auth = request.META.pop('HTTP_AUTHORIZATION', None)
+        asap_auth = request.META.pop("HTTP_AUTHORIZATION", None)
         orig_auth = request.META.pop(self.xauth, None)
 
         # Swap original client header in to allow regular auth middleware
         if orig_auth is not None:
-            request.META['HTTP_AUTHORIZATION'] = orig_auth
+            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:
-        if not hasattr(request, 'asap_forwarded'):
+    def process_view(
+        self, request: Any, view_func: Callable, view_args: Any, view_kwargs: Any
+    ) -> None:
+        if not hasattr(request, "asap_forwarded"):
             return None
 
         # swap headers back into place
         asap_auth = request.META.pop(self.xauth, None)
-        orig_auth = request.META.pop('HTTP_AUTHORIZATION', None)
+        orig_auth = request.META.pop("HTTP_AUTHORIZATION", None)
 
         if asap_auth is not None:
-            request.META['HTTP_AUTHORIZATION'] = asap_auth
+            request.META["HTTP_AUTHORIZATION"] = asap_auth
         if orig_auth is not None:
             request.META[self.xauth] = orig_auth