Remove more usages of six (mpdavis#253)

Author: asherf

jose/backends/base.py

@@ -1,6 +1,4 @@
-import six
-
-from ..utils import base64url_encode
+from ..utils import base64url_encode, ensure_binary
 
 
 class Key:
@@ -79,7 +77,7 @@ def unwrap_key(self, wrapped_key):
 
 class DIRKey(Key):
     def __init__(self, key_data, algorithm):
-        self._key = six.ensure_binary(key_data)
+        self._key = ensure_binary(key_data)
         self._alg = algorithm
 
     def to_dict(self):

jose/backends/cryptography_backend.py

@@ -1,7 +1,6 @@
 import math
 import warnings
 
-import six
 from cryptography.exceptions import InvalidSignature, InvalidTag
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.bindings.openssl.binding import Binding
@@ -17,7 +16,7 @@
 
 from ..constants import ALGORITHMS
 from ..exceptions import JWEError, JWKError
-from ..utils import base64_to_long, base64url_decode, base64url_encode, long_to_base64
+from ..utils import base64_to_long, base64url_decode, base64url_encode, ensure_binary, long_to_base64
 from .base import Key
 
 _binding = None
@@ -486,7 +485,7 @@ def to_dict(self):
         return data
 
     def encrypt(self, plain_text, aad=None):
-        plain_text = six.ensure_binary(plain_text)
+        plain_text = ensure_binary(plain_text)
         try:
             iv = get_random_bytes(algorithms.AES.block_size//8)
             mode = self._mode(iv)
@@ -509,9 +508,9 @@ def encrypt(self, plain_text, aad=None):
             raise JWEError(e)
 
     def decrypt(self, cipher_text, iv=None, aad=None, tag=None):
-        cipher_text = six.ensure_binary(cipher_text)
+        cipher_text = ensure_binary(cipher_text)
         try:
-            iv = six.ensure_binary(iv)
+            iv = ensure_binary(iv)
             mode = self._mode(iv)
             if mode.name == "GCM":
                 if tag is None:
@@ -537,12 +536,12 @@ def decrypt(self, cipher_text, iv=None, aad=None, tag=None):
             raise JWEError(e)
 
     def wrap_key(self, key_data):
-        key_data = six.ensure_binary(key_data)
+        key_data = ensure_binary(key_data)
         cipher_text = aes_key_wrap(self._key, key_data, default_backend())
         return cipher_text  # IV, cipher text, auth tag
 
     def unwrap_key(self, wrapped_key):
-        wrapped_key = six.ensure_binary(wrapped_key)
+        wrapped_key = ensure_binary(wrapped_key)
         try:
             plain_text = aes_key_unwrap(self._key, wrapped_key, default_backend())
         except InvalidUnwrap as cause:
@@ -611,15 +610,15 @@ def to_dict(self):
         }
 
     def sign(self, msg):
-        msg = six.ensure_binary(msg)
+        msg = ensure_binary(msg)
         h = hmac.HMAC(self.prepared_key, self._hash_alg, backend=default_backend())
         h.update(msg)
         signature = h.finalize()
         return signature
 
     def verify(self, msg, sig):
-        msg = six.ensure_binary(msg)
-        sig = six.ensure_binary(sig)
+        msg = ensure_binary(msg)
+        sig = ensure_binary(sig)
         h = hmac.HMAC(self.prepared_key, self._hash_alg, backend=default_backend())
         h.update(msg)
         try:

jose/jwe.py

@@ -3,8 +3,6 @@
 import zlib
 from struct import pack
 
-import six
-
 try:
     from collections.abc import Mapping  # Python 3
 except ImportError:
@@ -14,7 +12,7 @@
 from .backends import get_random_bytes
 from .constants import ALGORITHMS, ZIPS
 from .exceptions import JWEError, JWEParseError
-from .utils import base64url_decode, base64url_encode
+from .utils import base64url_decode, base64url_encode, ensure_binary
 
 
 def encrypt(plaintext, key, encryption=ALGORITHMS.A256GCM,
@@ -49,7 +47,7 @@ def encrypt(plaintext, key, encryption=ALGORITHMS.A256GCM,
         'eyJhbGciOiJkaXIiLCJlbmMiOiJBMTI4R0NNIn0..McILMB3dYsNJSuhcDzQshA.OfX9H_mcUpHDeRM4IA.CcnTWqaqxNsjT4eCaUABSg'
 
     """
-    plaintext = six.ensure_binary(plaintext)  # Make sure it's bytes
+    plaintext = ensure_binary(plaintext)  # Make sure it's bytes
     if algorithm not in ALGORITHMS.SUPPORTED:
         raise JWEError('Algorithm %s not supported.' % algorithm)
     if encryption not in ALGORITHMS.SUPPORTED:
@@ -277,7 +275,7 @@ def _jwe_compact_deserialize(jwe_bytes):
     # Vector, the JWE Ciphertext, the JWE Authentication Tag, and the
     # JWE AAD, following the restriction that no line breaks,
     # whitespace, or other additional characters have been used.
-    jwe_bytes = six.ensure_binary(jwe_bytes)
+    jwe_bytes = ensure_binary(jwe_bytes)
     try:
         header_segment, encrypted_key_segment, iv_segment, \
             cipher_text_segment, auth_tag_segment = jwe_bytes.split(b'.', 4)
@@ -305,9 +303,9 @@ def _jwe_compact_deserialize(jwe_bytes):
     # values that together comprise the JOSE Header.
 
     try:
-        header = json.loads(six.ensure_str(header_data))
+        header = json.loads(header_data)
     except ValueError as e:
-        raise JWEParseError('Invalid header string: %s' % e)
+        raise JWEParseError(f'Invalid header string: {e}')
 
     if not isinstance(header, Mapping):
         raise JWEParseError('Invalid header string: must be a json object')
@@ -598,7 +596,7 @@ def _jwe_compact_serialize(encoded_header, encrypted_cek, iv, cipher_text, auth_
     Returns:
         (str): JWE compact serialized string
     """
-    cipher_text = six.ensure_binary(cipher_text)  # Maker sure it's bytes
+    cipher_text = ensure_binary(cipher_text)
     encoded_encrypted_cek = base64url_encode(encrypted_cek)
     encoded_iv = base64url_encode(iv)
     encoded_cipher_text = base64url_encode(cipher_text)

jose/utils.py

@@ -95,3 +95,13 @@ def timedelta_total_seconds(delta):
         delta (timedelta): A timedelta to convert to seconds.
     """
     return delta.days * 24 * 60 * 60 + delta.seconds
+
+
+def ensure_binary(s):
+    """Coerce **s** to bytes."""
+
+    if isinstance(s, bytes):
+        return s
+    if isinstance(s, str):
+        return s.encode('utf-8', 'strict')
+    raise TypeError(f"not expecting type '{type(s)}'")

requirements.txt

@@ -1,5 +1,4 @@
 pycryptodome
-six
 rsa
 ecdsa != 0.15
 pyasn1

setup.py

@@ -26,11 +26,8 @@ def get_packages(package):
     'pycrypto': ['pycrypto >=2.6.0, <2.7.0'] + pyasn1,
     'pycryptodome': ['pycryptodome >=3.3.1, <4.0.0'] + pyasn1,
 }
-legacy_backend_requires = ['ecdsa != 0.15', 'rsa'] + pyasn1
-install_requires = ['six <2.0']
-
 # TODO: work this into the extras selection instead.
-install_requires += legacy_backend_requires
+install_requires = ['ecdsa != 0.15', 'rsa'] + pyasn1
 
 
 setup(
@@ -70,7 +67,6 @@ def get_packages(package):
         'setuptools>=39.2.0',
     ],
     tests_require=[
-        'six',
         'ecdsa != 0.15',
         'pytest',
         'pytest-cov',

tests/algorithms/test_AES.py

@@ -1,7 +1,6 @@
 from binascii import hexlify, unhexlify
 
 import pytest
-import six
 
 from jose.constants import ALGORITHMS
 

tests/test_jwe.py

@@ -1,7 +1,6 @@
 import json
 
 import pytest
-import six
 
 import jose.backends
 from jose import jwe
@@ -429,7 +428,7 @@ def test_alg_enc_headers(self):
         enc = ALGORITHMS.A256CBC_HS512
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert header["enc"] == enc
         assert header["alg"] == alg
 
@@ -439,15 +438,15 @@ def test_cty_header_present_when_provided(self):
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg,
                                 cty="expected")
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert header["cty"] == "expected"
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
     def test_cty_header_not_present_when_not_provided(self):
         enc = ALGORITHMS.A256CBC_HS512
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert "cty" not in header
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
@@ -456,15 +455,15 @@ def test_zip_header_present_when_provided(self):
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt(b"Text", PUBLIC_KEY_PEM, enc, alg,
                                 zip=ZIPS.DEF)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert header["zip"] == ZIPS.DEF
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
     def test_zip_header_not_present_when_not_provided(self):
         enc = ALGORITHMS.A256CBC_HS512
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt(b"Text", PUBLIC_KEY_PEM, enc, alg)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert "zip" not in header
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
@@ -473,7 +472,7 @@ def test_zip_header_not_present_when_none(self):
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg,
                                 zip=ZIPS.NONE)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert "zip" not in header
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
@@ -482,13 +481,13 @@ def test_kid_header_present_when_provided(self):
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg,
                                 kid="expected")
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert header["kid"] == "expected"
 
     @pytest.mark.skipif(AESKey is None, reason="No AES backend")
     def test_kid_header_not_present_when_not_provided(self):
         enc = ALGORITHMS.A256CBC_HS512
         alg = ALGORITHMS.RSA_OAEP_256
         encrypted = jwe.encrypt("Text", PUBLIC_KEY_PEM, enc, alg)
-        header = json.loads(six.ensure_str(base64url_decode(encrypted.split(b".")[0])))
+        header = json.loads(base64url_decode(encrypted.split(b".")[0]))
         assert "kid" not in header