Lint with ruff

.github/workflows/test.yml

@@ -16,8 +16,14 @@ jobs:
           do_not_skip: '["pull_request"]'
           cancel_others: 'true'
           concurrent_skipping: same_content
-  test:
+  ruff:
+    runs-on: ubuntu-latest
     needs: pre_job
+    steps:
+      - uses: actions/checkout@v4
+      - uses: chartboost/ruff-action@v1
+  test:
+    needs: ruff
     if: ${{ needs.pre_job.outputs.should_skip != 'true' }}
     runs-on: ubuntu-latest
     strategy:
@@ -43,6 +49,6 @@ jobs:
         run: |
           poetry run pytest -vvv -ra --cov=cryptojwt --cov-report=xml --isort --black
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v4
         with:
           token: ${{ secrets.CODECOV_TOKEN }}

.pre-commit-config.yaml

@@ -0,0 +1,15 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.6.0
+    hooks:
+      - id: check-merge-conflict
+      - id: debug-statements
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-json
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.4.9
+    hooks:
+      - id: ruff
+      - id: ruff-format

doc/conf.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+#
+# ruff: noqa
 # -*- coding: utf-8 -*-
 #
 # CryptoJWT documentation build configuration file, created by

pyproject.toml

@@ -42,7 +42,7 @@ python = "^3.9"
 cryptography = ">=3.4.6"
 requests = "^2.25.1"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.dev.dependencies]
 alabaster = "^0.7.12"
 black = "^24.4.2"
 isort = "^5.13.2"
@@ -54,7 +54,27 @@ responses = "^0.13.0"
 sphinx = "^3.5.2"
 sphinx-autobuild = "^2021.3.14"
 coverage = "^7"
+ruff = "^0.4.6"
+pytest-ruff = "^0.3.2"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.ruff.lint]
+select = [
+    # pycodestyle
+    "E",
+    # Pyflakes
+    "F",
+    # pyupgrade
+    "UP",
+    # flake8-bugbear
+    "B",
+    # flake8-simplify
+    "SIM",
+    # isort
+    "I",
+]
+ignore = ["E501", "I001", "SIM102"]
+exclude = ["examples/*"]

src/cryptojwt/__init__.py

@@ -16,15 +16,23 @@
 from .utils import b64encode_item
 from .utils import split_token
 
-try:
-    from builtins import hex
-    from builtins import str
-    from builtins import zip
-except ImportError:
-    pass
-
 __version__ = version("cryptojwt")
 
+__all__ = [
+    "JWE",
+    "JWE",
+    "JWK",
+    "JWS",
+    "JWT",
+    "KeyBundle",
+    "KeyJar",
+    "BadSyntax",
+    "as_unicode",
+    "b64d",
+    "b64encode_item",
+    "split_token",
+]
+
 logger = logging.getLogger(__name__)
 
 JWT_TYPES = ("JWT", "application/jws", "JWS", "JWE")

src/cryptojwt/exception.py

@@ -20,7 +20,7 @@ def __init__(self, value, msg):
         self.msg = msg
 
     def __str__(self):
-        return "%s: %r" % (self.msg, self.value)
+        return f"{self.msg}: {self.value!r}"
 
 
 class BadSignature(Invalid):

src/cryptojwt/jwe/__init__.py

@@ -38,7 +38,7 @@
 }
 
 
-class Encrypter(object):
+class Encrypter:
     """Abstract base class for encryption algorithms."""
 
     def __init__(self, with_digest=False):

src/cryptojwt/jwe/aes.py

@@ -30,7 +30,7 @@ def __init__(self, key_len=32, key=None, msg_padding="PKCS7"):
             self.padder = PKCS7(128).padder()
             self.unpadder = PKCS7(128).unpadder()
         else:
-            raise Unsupported("Message padding: {}".format(msg_padding))
+            raise Unsupported(f"Message padding: {msg_padding}")
 
         self.iv = None
 

src/cryptojwt/jwe/jwe.py

@@ -1,14 +1,13 @@
+import contextlib
 import logging
 
 from ..jwk.asym import AsymmetricKey
 from ..jwk.ec import ECKey
 from ..jwk.hmac import SYMKey
 from ..jwk.jwk import key_from_jwk_dict
-from ..jwk.rsa import RSAKey
 from ..jwx import JWx
 from .exception import DecryptionFailed
 from .exception import NoSuitableDecryptionKey
-from .exception import NoSuitableECDHKey
 from .exception import NoSuitableEncryptionKey
 from .exception import NotSupportedAlgorithm
 from .exception import WrongEncryptionAlgorithm
@@ -133,7 +132,7 @@ def encrypt(self, keys=None, cek="", iv="", **kwargs):
             except TypeError as err:
                 raise err
             else:
-                logger.debug("Encrypted message using key with kid={}".format(key.kid))
+                logger.debug(f"Encrypted message using key with kid={key.kid}")
                 return token
 
         # logger.error("Could not find any suitable encryption key")
@@ -159,10 +158,8 @@ def decrypt(self, token=None, keys=None, alg=None, cek=None):
         else:
             keys = self.pick_keys(self._get_keys(), use="enc", alg=_alg)
 
-        try:
+        with contextlib.suppress(KeyError):
             keys.append(key_from_jwk_dict(_jwe.headers["jwk"]))
-        except KeyError:
-            pass
 
         if not keys and not cek:
             raise NoSuitableDecryptionKey(_alg)
@@ -194,18 +191,15 @@ def decrypt(self, token=None, keys=None, alg=None, cek=None):
                 return msg
 
         for key in keys:
-            if isinstance(key, AsymmetricKey):
-                _key = key.private_key()
-            else:
-                _key = key.key
+            _key = key.private_key() if isinstance(key, AsymmetricKey) else key.key
 
             try:
                 msg = decrypter.decrypt(_jwe, _key)
                 self["cek"] = decrypter.cek if "cek" in decrypter else None
             except (KeyError, DecryptionFailed):
                 pass
             else:
-                logger.debug("Decrypted message using key with kid=%s" % key.kid)
+                logger.debug(f"Decrypted message using key with kid={key.kid}")
                 return msg
 
         raise DecryptionFailed("No available key that could decrypt the message")

src/cryptojwt/jwe/jwe_ec.py

@@ -1,3 +1,4 @@
+import contextlib
 import struct
 
 from cryptography.hazmat.primitives.asymmetric import ec
@@ -110,8 +111,8 @@ def enc_setup(self, msg, key=None, auth_data=b"", **kwargs):
         if self.alg == "ECDH-ES":
             try:
                 dk_len = KEY_LEN[self.enc]
-            except KeyError:
-                raise ValueError("Unknown key length for algorithm %s" % self.enc)
+            except KeyError as exc:
+                raise ValueError(f"Unknown key length for algorithm {self.enc}") from exc
 
             cek = ecdh_derive_key(_epk, key.pub_key, apu, apv, str(self.enc).encode(), dk_len)
         elif self.alg in ["ECDH-ES+A128KW", "ECDH-ES+A192KW", "ECDH-ES+A256KW"]:
@@ -121,7 +122,7 @@ def enc_setup(self, msg, key=None, auth_data=b"", **kwargs):
             cek = self._generate_key(self.enc, cek=cek)
             encrypted_key = aes_key_wrap(kek, cek)
         else:
-            raise Exception("Unsupported algorithm %s" % self.alg)
+            raise Exception(f"Unsupported algorithm {self.alg}")
 
         return cek, encrypted_key, iv, params, epk
 
@@ -152,8 +153,8 @@ def dec_setup(self, token, key=None, **kwargs):
         if self.headers["alg"] == "ECDH-ES":
             try:
                 dk_len = KEY_LEN[self.headers["enc"]]
-            except KeyError:
-                raise Exception("Unknown key length for algorithm")
+            except KeyError as exc:
+                raise Exception("Unknown key length for algorithm") from exc
 
             self.cek = ecdh_derive_key(
                 key,
@@ -173,7 +174,7 @@ def dec_setup(self, token, key=None, **kwargs):
             kek = ecdh_derive_key(key, epubkey.pub_key, apu, apv, str(_post).encode(), klen)
             self.cek = aes_key_unwrap(kek, token.encrypted_key())
         else:
-            raise Exception("Unsupported algorithm %s" % self.headers["alg"])
+            raise Exception("Unsupported algorithm {}".format(self.headers["alg"]))
 
         return self.cek
 
@@ -190,10 +191,8 @@ def encrypt(self, key=None, iv="", cek="", **kwargs):
         _msg = as_bytes(self.msg)
 
         _args = self._dict
-        try:
+        with contextlib.suppress(KeyError):
             _args["kid"] = kwargs["kid"]
-        except KeyError:
-            pass
 
         if "params" in kwargs:
             if "apu" in kwargs["params"]:
@@ -204,23 +203,20 @@ def encrypt(self, key=None, iv="", cek="", **kwargs):
                 _args["epk"] = kwargs["params"]["epk"]
 
         jwe = JWEnc(**_args)
-        ctxt, tag, cek = super(JWE_EC, self).enc_setup(
+        ctxt, tag, cek = super().enc_setup(
             self["enc"], _msg, auth_data=jwe.b64_encode_header(), key=cek, iv=iv
         )
         if "encrypted_key" in kwargs:
             return jwe.pack(parts=[kwargs["encrypted_key"], iv, ctxt, tag])
         return jwe.pack(parts=[iv, ctxt, tag])
 
     def decrypt(self, token=None, **kwargs):
-        if isinstance(token, JWEnc):
-            jwe = token
-        else:
-            jwe = JWEnc().unpack(token)
+        jwe = token if isinstance(token, JWEnc) else JWEnc().unpack(token)
 
         if not self.cek:
             raise Exception("Content Encryption Key is Not Yet Set")
 
-        msg = super(JWE_EC, self)._decrypt(
+        msg = super()._decrypt(
             self.headers["enc"],
             self.cek,
             self.ctxt,

src/cryptojwt/jwe/jwe_hmac.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import zlib
 
@@ -34,10 +35,8 @@ def encrypt(self, key, iv="", cek="", **kwargs):
         _msg = as_bytes(self.msg)
 
         _args = self._dict
-        try:
+        with contextlib.suppress(KeyError):
             _args["kid"] = kwargs["kid"]
-        except KeyError:
-            pass
 
         jwe = JWEnc(**_args)
 
@@ -68,10 +67,7 @@ def decrypt(self, token, key=None, cek=None):
         if not key and not cek:
             raise MissingKey("On of key or cek must be specified")
 
-        if isinstance(token, JWEnc):
-            jwe = token
-        else:
-            jwe = JWEnc().unpack(token)
+        jwe = token if isinstance(token, JWEnc) else JWEnc().unpack(token)
 
         if len(jwe) != 5:
             raise WrongNumberOfParts(len(jwe))

src/cryptojwt/jwe/jwe_rsa.py

@@ -49,7 +49,7 @@ def encrypt(self, key, iv="", cek="", **kwargs):
             if self["zip"] == "DEF":
                 _msg = zlib.compress(_msg)
             else:
-                raise ParameterError("Zip has unknown value: %s" % self["zip"])
+                raise ParameterError("Zip has unknown value: {}".format(self["zip"]))
 
         kwarg_cek = cek or None
 
@@ -58,7 +58,7 @@ def encrypt(self, key, iv="", cek="", **kwargs):
         cek = self._generate_key(_enc, cek)
         self["cek"] = cek
 
-        logger.debug("cek: %s, iv: %s" % ([c for c in cek], [c for c in iv]))
+        logger.debug(f"cek: {[c for c in cek]}, iv: {[c for c in iv]}")
 
         _encrypt = RSAEncrypter(self.with_digest).encrypt
 
@@ -92,10 +92,7 @@ def decrypt(self, token, key, cek=None):
         :param cek: Ephemeral cipher key
         :return: The decrypted message
         """
-        if not isinstance(token, JWEnc):
-            jwe = JWEnc().unpack(token)
-        else:
-            jwe = token
+        jwe = JWEnc().unpack(token) if not isinstance(token, JWEnc) else token
 
         self.jwt = jwe.encrypted_key()
         jek = jwe.encrypted_key()

src/cryptojwt/jwe/jwekey.py

@@ -29,8 +29,8 @@ def _generate_key(encalg, cek=""):
         except KeyError:
             try:
                 _key = get_random_bytes(KEY_LEN_BYTES[encalg])
-            except KeyError:
-                raise ValueError("Unsupported encryption algorithm %s" % encalg)
+            except KeyError as exc:
+                raise ValueError(f"Unsupported encryption algorithm {encalg}") from exc
 
         return _key
 
@@ -77,7 +77,7 @@ def _decrypt(enc, key, ctxt, iv, tag, auth_data=b""):
         elif enc in ["A128CBC-HS256", "A192CBC-HS384", "A256CBC-HS512"]:
             aes = AES_CBCEncrypter(key=key)
         else:
-            raise Exception("Unsupported encryption algorithm %s" % enc)
+            raise Exception(f"Unsupported encryption algorithm {enc}")
 
         try:
             return aes.decrypt(ctxt, iv=iv, auth_data=auth_data, tag=tag)

src/cryptojwt/jwe/jwenc.py

@@ -48,7 +48,7 @@ def is_jwe(self):
         if "alg" in self.headers and "enc" in self.headers:
             for typ in ["alg", "enc"]:
                 if self.headers[typ] not in SUPPORTED[typ]:
-                    logger.debug("Not supported %s algorithm: %s" % (typ, self.headers[typ]))
+                    logger.debug(f"Not supported {typ} algorithm: {self.headers[typ]}")
                     return False
         else:
             return False

src/cryptojwt/jwe/utils.py

@@ -7,8 +7,6 @@
 from cryptography.hazmat.primitives.hashes import SHA384
 from cryptography.hazmat.primitives.hashes import SHA512
 
-from ..utils import b64e
-
 LENMET = {32: (16, SHA256), 48: (24, SHA384), 64: (32, SHA512)}
 
 
@@ -20,8 +18,8 @@ def get_keys_seclen_dgst(key, iv):
     # Select the digest to use based on key length
     try:
         seclen, hash_method = LENMET[len(key)]
-    except KeyError:
-        raise Exception("Invalid CBC+HMAC key length: %s bytes" % len(key))
+    except KeyError as exc:
+        raise Exception(f"Invalid CBC+HMAC key length: {len(key)} bytes") from exc
 
     # Split the key
     ka = key[:seclen]