Merge pull request #50 from web-push-libs/feat/49

feat: update public key calls

Author: jrconlin

python/http_ece/__init__.py

@@ -9,6 +9,9 @@
 from cryptography.hazmat.primitives.ciphers import (
     Cipher, algorithms, modes
 )
+from cryptography.hazmat.primitives.serialization import (
+    Encoding, PublicFormat
+)
 from cryptography.hazmat.primitives.asymmetric import ec
 
 MAX_RECORD_SIZE = pow(2, 31) - 1
@@ -24,11 +27,13 @@
     "aesgcm128": {"pad": 1},
 }
 
+
 class ECEException(Exception):
     """Exception for ECE encryption functions"""
     def __init__(self, message):
         self.message = message
 
+
 def derive_key(mode, version, salt, key,
                private_key, dh, auth_secret,
                keyid, keylabel="P-256"):
@@ -64,15 +69,20 @@ def build_info(base, info_context):
     def derive_dh(mode, version, private_key, dh, keylabel):
         def length_prefix(key):
             return struct.pack("!H", len(key)) + key
-
         if isinstance(dh, ec.EllipticCurvePublicKey):
             pubkey = dh
-            dh = dh.public_numbers().encode_point()
+            dh = dh.public_bytes(
+                Encoding.X962,
+                PublicFormat.UncompressedPoint)
         else:
-            numbers = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), dh)
-            pubkey = numbers.public_key(default_backend())
-
-        encoded = private_key.public_key().public_numbers().encode_point()
+            pubkey = ec.EllipticCurvePublicKey.from_encoded_point(
+                ec.SECP256R1(),
+                dh
+            )
+
+        encoded = private_key.public_key().public_bytes(
+            Encoding.X962,
+            PublicFormat.UncompressedPoint)
         if mode == "encrypt":
             sender_pub_key = encoded
             receiver_pub_key = dh
@@ -243,7 +253,7 @@ def unpad(data, last):
     if version == "aes128gcm":
         try:
             content_header = parse_content_header(content)
-        except:
+        except Exception:
             raise ECEException("Could not parse the content header")
         salt = content_header['salt']
         rs = content_header['rs']
@@ -386,7 +396,9 @@ def compose_aes128gcm(salt, content, rs, keyid):
         counter += 1
     if version == "aes128gcm":
         if keyid is None and private_key is not None:
-            kid = private_key.public_key().public_numbers().encode_point()
+            kid = private_key.public_key().public_bytes(
+                Encoding.X962,
+                PublicFormat.UncompressedPoint)
         else:
             kid = (keyid or '').encode('utf-8')
         return compose_aes128gcm(salt, result, rs, keyid=kid)

python/http_ece/tests/test_ece.py

@@ -5,6 +5,9 @@
 import unittest
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.serialization import (
+    Encoding, PublicFormat
+)
 
 from nose.tools import eq_, assert_raises
 
@@ -43,6 +46,7 @@ def b64d(arg):
         return None
     return base64.urlsafe_b64decode(str(arg) + '===='[:len(arg) % 4:])
 
+
 def make_key():
     return ec.generate_private_key(ec.SECP256R1(), default_backend())
 
@@ -131,7 +135,8 @@ class TestEceChecking(unittest.TestCase):
     def setUp(self):
         self.m_key = os.urandom(16)
         self.m_input = os.urandom(5)
-        # This header is specific to the padding tests, but can be used elsewhere
+        # This header is specific to the padding tests, but can be used
+        # elsewhere
         self.m_header = b'\xaa\xd2\x05}3S\xb7\xff7\xbd\xe4*\xe1\xd5\x0f\xda'
         self.m_header += struct.pack('!L', 32) + b'\0'
 
@@ -189,47 +194,54 @@ def test_encrypt_long_keyid(self):
                 self.m_input,
                 version='aes128gcm',
                 key=self.m_key,
-                keyid=b64e(os.urandom(192)), # 256 bytes
+                keyid=b64e(os.urandom(192)),  # 256 bytes
             )
         eq_(ex.exception.message, "keyid is too long")
 
     def test_overlong_padding(self):
         with assert_raises(ECEException) as ex:
             ece.decrypt(
-                self.m_header + b'\xbb\xc7\xb9ev\x0b\xf0f+\x93\xf4\xe5\xd6\x94\xb7e\xf0\xcd\x15\x9b(\x01\xa5',
+                self.m_header + b'\xbb\xc7\xb9ev\x0b\xf0f+\x93\xf4'
+                                b'\xe5\xd6\x94\xb7e\xf0\xcd\x15\x9b(\x01\xa5',
                 version='aes128gcm',
                 key=b'd\xc7\x0ed\xa7%U\x14Q\xf2\x08\xdf\xba\xa0\xb9r',
-                keyid=b64e(os.urandom(192)), # 256 bytes
+                keyid=b64e(os.urandom(192)),  # 256 bytes
             )
         eq_(ex.exception.message, "all zero record plaintext")
 
     def test_bad_early_delimiter(self):
         with assert_raises(ECEException) as ex:
             ece.decrypt(
-                self.m_header + b'\xb9\xc7\xb9ev\x0b\xf0\x9eB\xb1\x08C8u\xa3\x06\xc9x\x06\n\xfc|}\xe9R\x85\x91\x8bX\x02`\xf3' + b'E8z(\xe5%f/H\xc1\xc32\x04\xb1\x95\xb5N\x9ep\xd4\x0e<\xf3\xef\x0cg\x1b\xe0\x14I~\xdc',
+                self.m_header + b'\xb9\xc7\xb9ev\x0b\xf0\x9eB\xb1\x08C8u'
+                                b'\xa3\x06\xc9x\x06\n\xfc|}\xe9R\x85\x91'
+                                b'\x8bX\x02`\xf3' +
+                b'E8z(\xe5%f/H\xc1\xc32\x04\xb1\x95\xb5N\x9ep\xd4\x0e<\xf3'
+                b'\xef\x0cg\x1b\xe0\x14I~\xdc',
                 version='aes128gcm',
                 key=b'd\xc7\x0ed\xa7%U\x14Q\xf2\x08\xdf\xba\xa0\xb9r',
-                keyid=b64e(os.urandom(192)), # 256 bytes
+                keyid=b64e(os.urandom(192)),  # 256 bytes
             )
         eq_(ex.exception.message, "record delimiter != 1")
 
     def test_bad_final_delimiter(self):
         with assert_raises(ECEException) as ex:
             ece.decrypt(
-                self.m_header + b'\xba\xc7\xb9ev\x0b\xf0\x9eB\xb1\x08Ji\xe4P\x1b\x8dI\xdb\xc6y#MG\xc2W\x16',
+                self.m_header + b'\xba\xc7\xb9ev\x0b\xf0\x9eB\xb1\x08Ji'
+                                b'\xe4P\x1b\x8dI\xdb\xc6y#MG\xc2W\x16',
                 version='aes128gcm',
                 key=b'd\xc7\x0ed\xa7%U\x14Q\xf2\x08\xdf\xba\xa0\xb9r',
-                keyid=b64e(os.urandom(192)), # 256 bytes
+                keyid=b64e(os.urandom(192)),  # 256 bytes
             )
         eq_(ex.exception.message, "last record delimiter != 2")
 
     def test_damage(self):
         with assert_raises(ECEException) as ex:
             ece.decrypt(
-                self.m_header + b'\xbb\xc6\xb1\x1dF:~\x0f\x07+\xbe\xaaD\xe0\xd6.K\xe5\xf9]%\xe3\x86q\xe0}',
+                self.m_header + b'\xbb\xc6\xb1\x1dF:~\x0f\x07+\xbe\xaaD'
+                                b'\xe0\xd6.K\xe5\xf9]%\xe3\x86q\xe0}',
                 version='aes128gcm',
                 key=b'd\xc7\x0ed\xa7%U\x14Q\xf2\x08\xdf\xba\xa0\xb9r',
-                keyid=b64e(os.urandom(192)), # 256 bytes
+                keyid=b64e(os.urandom(192)),  # 256 bytes
             )
         eq_(ex.exception.message, "Decryption error: InvalidTag()")
 
@@ -246,14 +258,14 @@ def tearDown(self):
 
     def _rsoverhead(self, version):
         if version == 'aesgcm128':
-            return 1;
+            return 1
         if version == 'aesgcm':
-            return 2;
-        return 18;
+            return 2
+        return 18
 
     def _generate_input(self, minLen=0):
         length = struct.unpack('!B', os.urandom(1))[0] + minLen
-        return os.urandom(length);
+        return os.urandom(length)
 
     def encrypt_decrypt(self, input, encrypt_params, decrypt_params=None,
                         version=None):
@@ -263,7 +275,7 @@ def encrypt_decrypt(self, input, encrypt_params, decrypt_params=None,
         :type length: bytearray
         :param encrypt_params: Dictionary of encryption parameters
         :type encrypt_params: dict
-        :param decrypt_params: Optional dictionary of decryption paramseters
+        :param decrypt_params: Optional dictionary of decryption parameters
         :type decrypt_params: dict
         :param version: Content-Type of the body, formulating encryption
         :type enumerate("aes128gcm", "aesgcm", "aesgcm128"):
@@ -299,7 +311,8 @@ def encrypt_decrypt(self, input, encrypt_params, decrypt_params=None,
                                 dh=decrypt_params.get("dh"),
                                 private_key=decrypt_params.get("private_key"),
                                 auth_secret=decrypt_params.get("auth_secret"),
-                                rs=decrypt_params.get("rs", decrypt_rs_default),
+                                rs=decrypt_params.get("rs",
+                                                      decrypt_rs_default),
                                 version=version)
         logbuf("Decrypted", decrypted)
         eq_(input, decrypted)
@@ -346,12 +359,17 @@ def detect_truncation(self, version):
 
     def use_dh(self, version):
         def pubbytes(k):
-            return k.public_key().public_numbers().encode_point()
+            return k.public_key().public_bytes(
+                Encoding.X962,
+                PublicFormat.UncompressedPoint
+            )
 
         def privbytes(k):
             d = k.private_numbers().private_value
             b = b''
-            for i in range(0, k.private_numbers().public_numbers.curve.key_size, 32):
+            for i in range(0,
+                           k.private_numbers().public_numbers.curve.key_size,
+                           32):
                 b = struct.pack("!L", (d >> i) & 0xffffffff) + b
             return b
 
@@ -424,13 +442,11 @@ def _run(self, mode):
         if mode == 'encrypt':
             func = ece.encrypt
             local = 'sender'
-            remote = 'receiver'
             inp = 'input'
             outp = 'encrypted'
         else:
             func = ece.decrypt
             local = 'receiver'
-            remote = 'sender'
             inp = 'encrypted'
             outp = 'input'
 
@@ -445,7 +461,8 @@ def _run(self, mode):
             if 'keys' in data:
                 key = None
                 decode_pub = ec.EllipticCurvePublicNumbers.from_encoded_point
-                pubnum = decode_pub(ec.SECP256R1(), b64d(data['keys'][local]['public']))
+                pubnum = decode_pub(ec.SECP256R1(),
+                                    b64d(data['keys'][local]['public']))
                 d = 0
                 dbin = b64d(data['keys'][local]['private'])
                 for i in range(0, len(dbin), 4):