Convert python to use cryptography

Author: martinthomson

python/http_ece/__init__.py

@@ -9,7 +9,7 @@
 from cryptography.hazmat.primitives.ciphers import (
     Cipher, algorithms, modes
 )
-from pyelliptic import ecc
+from cryptography.hazmat.primitives.asymmetric import ec
 
 keys = {}
 labels = {}
@@ -74,20 +74,23 @@ def derive_dh(mode, version, private_key, dh, label):
         def length_prefix(key):
             return struct.pack("!H", len(key)) + key
 
+        encoded = private_key.public_key().public_numbers().encode_point()
         if mode == "encrypt":
-            sender_pub_key = private_key.get_pubkey()
+            sender_pub_key = encoded
             receiver_pub_key = dh
         else:
             sender_pub_key = dh
-            receiver_pub_key = private_key.get_pubkey()
+            receiver_pub_key = encoded
 
         if version == "aes128gcm":
             context = b"WebPush: info\x00" + receiver_pub_key + sender_pub_key
         else:
             context = (label + b"\0" + length_prefix(receiver_pub_key) +
                        length_prefix(sender_pub_key))
 
-        return private_key.get_ecdh_key(dh), context
+        numbers = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), dh)
+        pubkey = numbers.public_key(default_backend())
+        return private_key.exchange(ec.ECDH(), pubkey), context
 
     if version not in versions:
         raise ECEException(u"Invalid version")
@@ -251,7 +254,8 @@ def unpad(data, last):
         keymap = keys
     if keylabels is None:
         keylabels = labels
-    if keyid is not None and keyid in keymap and isinstance(keymap[keyid], ecc.ECC):
+    if keyid is not None and keyid in keymap and \
+       isinstance(keymap[keyid], ec.EllipticCurvePrivateKey):
         private_key = keymap[keyid]
 
     if version == "aes128gcm":
@@ -318,7 +322,7 @@ def encrypt(content, salt=None, key=None,
     :type key: object
     :param keyid: Internal key identifier for private key info
     :type keyid: str
-    :param dh: Remote Diffie Hellman sequence
+    :param dh: Remote Diffie Hellman sequence (omit for aes128gcm)
     :type dh: str
     :param rs: Record size
     :type rs: int
@@ -382,7 +386,8 @@ def compose_aes128gcm(salt, content, rs, keyid):
     if salt is None:
         salt = os.urandom(16)
         version = "aes128gcm"
-    if keyid is not None and keyid in keymap and isinstance(keymap[keyid], ecc.ECC):
+    if keyid is not None and keyid in keymap and \
+       isinstance(keymap[keyid], ec.EllipticCurvePrivateKey):
         private_key = keymap[keyid]
 
     (key_, nonce_) = derive_key(mode="encrypt", version=version,
@@ -412,7 +417,7 @@ 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.get_pubkey()
+            kid = private_key.public_key().public_numbers().encode_point()
         else:
             kid = (keyid or '').encode('utf-8')
         return compose_aes128gcm(salt, result, rs, keyid=kid)

python/http_ece/tests/test_ece.py

@@ -1,13 +1,13 @@
 import base64
 import json
 import os
-import pyelliptic
 import struct
 import unittest
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.asymmetric import ec
 
 from nose.tools import eq_, assert_raises
 
-
 import http_ece as ece
 from http_ece import ECEException
 
@@ -43,11 +43,14 @@ 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())
+
 
 class TestEce(unittest.TestCase):
 
     def setUp(self):
-        self.keymap = {'valid': pyelliptic.ECC(curve="prime256v1")}
+        self.keymap = {'valid': make_key()}
         self.keylabels = {'valid': 'P-256'}
         self.m_key = os.urandom(16)
         self.m_salt = os.urandom(16)
@@ -396,34 +399,47 @@ def use_key_id(self, version):
                              decrypt_params, version=version)
 
     def use_dh(self, version):
+        def pubbytes(k):
+            return k.public_key().public_numbers().encode_point()
+
+        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):
+                b = struct.pack("!L", (d >> i) & 0xffffffff) + b
+            return b
+
+        def logec(s, k):
+            logbuf(s + " private", privbytes(k))
+            logbuf(s + " public", pubbytes(k))
+
         def is_uncompressed(k):
-            b1 = k.get_pubkey()[0:1]
+            b1 = pubbytes(k)[0:1]
             assert struct.unpack("B", b1)[0] == 4, "is an uncompressed point"
 
         # the static key is used by the receiver
-        static_key = pyelliptic.ECC(curve="prime256v1")
+        static_key = make_key()
         is_uncompressed(static_key)
 
-        logbuf("Receiver private", static_key.get_privkey())
-        logbuf("Receiver public", static_key.get_pubkey())
+
+        logec("receiver", static_key)
 
         # the ephemeral key is used by the sender
-        ephemeral_key = pyelliptic.ECC(curve="prime256v1")
+        ephemeral_key = make_key()
         is_uncompressed(ephemeral_key)
 
-        logbuf("Sender private", ephemeral_key.get_privkey())
-        logbuf("Sender public", ephemeral_key.get_pubkey())
+        logec("sender", ephemeral_key)
 
         auth_secret = os.urandom(16)
 
         if version != "aes128gcm":
-            decrypt_dh = ephemeral_key.get_pubkey()
+            decrypt_dh = pubbytes(ephemeral_key)
         else:
             decrypt_dh = None
 
         encrypt_params = {
             "private_key": ephemeral_key,
-            "dh": static_key.get_pubkey(),
+            "dh": pubbytes(static_key),
             "auth_secret": auth_secret,
         }
         decrypt_params = {
@@ -484,11 +500,14 @@ def _run(self, mode):
 
             if 'keys' in data:
                 key = None
-                private_key = pyelliptic.ECC(
-                    curve='prime256v1',
-                    pubkey=b64d(data['keys'][local]['public']),
-                    privkey=b64d(data['keys'][local]['private']),
-                )
+                decode_pub = ec.EllipticCurvePublicNumbers.from_encoded_point
+                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):
+                    d = (d << 32) + struct.unpack('!L', dbin[i:i + 4])[0]
+                privnum = ec.EllipticCurvePrivateNumbers(d, pubnum)
+                private_key = privnum.private_key(default_backend())
             else:
                 key = b64d(p['key'])
                 private_key = None

python/setup.py

@@ -10,7 +10,7 @@
 
 setup(
     name='http_ece',
-    version='0.7.3',
+    version='0.8.0',
     author='Martin Thomson',
     author_email='[email protected]',
     scripts=[],
@@ -27,8 +27,7 @@
     ],
     keywords='crypto http',
     install_requires=[
-        'pyelliptic==1.5.7',
-        'cryptography',
+        'cryptography~=1.1.0',
     ],
     tests_require=[
         'nose',