Sem-ver: bugfix Apply ruff formatting and use ruff for linting

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

Author: dbaxa

.github/workflows/build.yml

@@ -19,11 +19,11 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip wheel setuptools
-        pip install -q pycodestyle==2.9.1 flake8==5.0.4
+        pip install -q ruff
     - name: Lint
       run: |
-        pycodestyle .
-        flake8 .
+        ruff check .
+        ruff format --check .
     - name: Test
       run: |
         pip install wheel

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

@@ -9,16 +9,17 @@ class BaseJWTAuth(object):
     def __init__(self, signer, audience, *args, **kwargs):
         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, key_identifier, private_key_pem, audience,
-               **kwargs):
+    def create(cls, issuer, key_identifier, private_key_pem, audience, **kwargs):
         """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):
-        return b'Bearer ' + self._signer.generate_jwt(
-            self._audience, additional_claims=self._additional_claims)
+        return b"Bearer " + self._signer.generate_jwt(
+            self._audience, additional_claims=self._additional_claims
+        )

atlassian_jwt_auth/contrib/aiohttp/__init__.py

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

atlassian_jwt_auth/contrib/aiohttp/auth.py

@@ -8,15 +8,14 @@ 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):
         return self._get_header_value().decode(self.encoding)
 
 
-def create_jwt_auth(
-        issuer, key_identifier, private_key_pem, audience, **kwargs):
+def create_jwt_auth(issuer, key_identifier, private_key_pem, audience, **kwargs):
     """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

@@ -6,12 +6,15 @@
 from atlassian_jwt_auth.exceptions import PublicKeyRetrieverException
 from atlassian_jwt_auth.key import (
     PEM_FILE_TYPE,
-    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, *, loop=None):
@@ -23,32 +26,32 @@ def __init__(self, base_url, *, loop=None):
     def _get_session(self):
         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, requests_kwargs):
-        """ 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).
+        """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, requests_kwargs):
-        requests_kwargs = self._convert_proxies_to_proxy_arg(
-            url, requests_kwargs)
+        requests_kwargs = self._convert_proxies_to_proxy_arg(url, requests_kwargs)
         try:
-            resp = await self._session.get(url, headers={'accept':
-                                                         PEM_FILE_TYPE},
-                                           **requests_kwargs)
+            resp = await self._session.get(
+                url, headers={"accept": 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

@@ -22,8 +22,8 @@ async def verify_jwt(self, a_jwt, audience, leeway=0, **requests_kwargs):
         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

@@ -1,8 +1,8 @@
 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.",
-    DeprecationWarning, stacklevel=2
+    DeprecationWarning,
+    stacklevel=2,
 )

atlassian_jwt_auth/contrib/django/decorators.py

@@ -16,30 +16,32 @@ def validate_asap(issuers=None, subjects=None, required=True):
     :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
 
 
@@ -48,7 +50,9 @@ def requires_asap(issuers=None, subject_should_match_issuer=None, func=None):
 
     :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,9 +1,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):
@@ -18,17 +16,17 @@ def __init__(self, get_response=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):
-        error_response = super(ProxiedAsapMiddleware, self).process_request(
-            request
-        )
+        error_response = super(ProxiedAsapMiddleware, self).process_request(request)
 
         if error_response:
             return error_response
@@ -38,26 +36,26 @@ def process_request(self, request):
             return
 
         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
 
     def process_view(self, request, view_func, view_args, view_kwargs):
-        if not hasattr(request, 'asap_forwarded'):
+        if not hasattr(request, "asap_forwarded"):
             return
 
         # 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