Use a different ECC backend with a compatible serializer

This library has the ``SEC1`` encoder which is compatible
with the serialization of ECC keys/points used in the Go libp2p impl

libp2p/crypto/ecc.py

@@ -1,22 +1,34 @@
-from typing import cast
-
-from Crypto.PublicKey import ECC
-from Crypto.PublicKey.ECC import EccKey
+from fastecdsa import curve as curve_types
+from fastecdsa import keys, point
+from fastecdsa.encoding.sec1 import SEC1Encoder
 
 from libp2p.crypto.keys import KeyPair, KeyType, PrivateKey, PublicKey
 
 
+def infer_local_type(curve: str) -> curve_types.Curve:
+    """
+    converts a ``str`` representation of some elliptic curve to
+    a representation understood by the backend of this module.
+    """
+    if curve == "P-256":
+        return curve_types.P256
+    else:
+        raise NotImplementedError()
+
+
 class ECCPublicKey(PublicKey):
-    def __init__(self, impl: EccKey) -> None:
+    def __init__(self, impl: point.Point, curve: curve_types.Curve) -> None:
         self.impl = impl
+        self.curve = curve
 
     def to_bytes(self) -> bytes:
-        return cast(bytes, self.impl.export_key(format="DER"))
+        return SEC1Encoder.encode_public_key(self.impl, compressed=False)
 
     @classmethod
-    def from_bytes(cls, data: bytes) -> "ECCPublicKey":
-        public_key_impl = ECC.import_key(data)
-        return cls(public_key_impl)
+    def from_bytes(cls, data: bytes, curve: str) -> "ECCPublicKey":
+        curve_type = infer_local_type(curve)
+        public_key_impl = SEC1Encoder.decode_public_key(data, curve_type)
+        return cls(public_key_impl, curve_type)
 
     def get_type(self) -> KeyType:
         return KeyType.ECC_P256
@@ -26,16 +38,18 @@ class ECCPublicKey(PublicKey):
 
 
 class ECCPrivateKey(PrivateKey):
-    def __init__(self, impl: EccKey) -> None:
+    def __init__(self, impl: int, curve: curve_types.Curve) -> None:
         self.impl = impl
+        self.curve = curve
 
     @classmethod
     def new(cls, curve: str) -> "ECCPrivateKey":
-        private_key_impl = ECC.generate(curve=curve)
-        return cls(private_key_impl)
+        curve_type = infer_local_type(curve)
+        private_key_impl = keys.gen_private_key(curve_type)
+        return cls(private_key_impl, curve_type)
 
     def to_bytes(self) -> bytes:
-        return cast(bytes, self.impl.export_key(format="DER"))
+        return keys.export_key(self.impl, self.curve)
 
     def get_type(self) -> KeyType:
         return KeyType.ECC_P256
@@ -44,7 +58,8 @@ class ECCPrivateKey(PrivateKey):
         raise NotImplementedError
 
     def get_public_key(self) -> PublicKey:
-        return ECCPublicKey(self.impl.public_key())
+        public_key_impl = keys.get_public_key(self.impl, self.curve)
+        return ECCPublicKey(public_key_impl, self.curve)
 
 
 def create_new_key_pair(curve: str) -> KeyPair:

libp2p/crypto/key_exchange.py

@@ -1,9 +1,8 @@
 from typing import Callable, Tuple, cast
 
-from Crypto.Math.Numbers import Integer
-import Crypto.PublicKey.ECC as ECC
+from fastecdsa.encoding.util import int_bytelen
 
-from libp2p.crypto.ecc import ECCPrivateKey, create_new_key_pair
+from libp2p.crypto.ecc import ECCPrivateKey, ECCPublicKey, create_new_key_pair
 from libp2p.crypto.keys import PublicKey
 
 SharedKeyGenerator = Callable[[bytes], bytes]
@@ -19,11 +18,12 @@ def create_ephemeral_key_pair(curve_type: str) -> Tuple[PublicKey, SharedKeyGene
     key_pair = create_new_key_pair(curve_type)
 
     def _key_exchange(serialized_remote_public_key: bytes) -> bytes:
-        remote_public_key = ECC.import_key(serialized_remote_public_key)
-        curve_point = remote_public_key.pointQ
         private_key = cast(ECCPrivateKey, key_pair.private_key)
-        secret_point = curve_point * private_key.impl.d
-        byte_size = secret_point.size_in_bytes()
-        return cast(Integer, secret_point.x).to_bytes(byte_size)
+
+        remote_point = ECCPublicKey.from_bytes(serialized_remote_public_key, curve_type)
+        secret_point = remote_point.impl * private_key.impl
+        secret_x_coordinate = secret_point.x
+        byte_size = int_bytelen(secret_x_coordinate)
+        return secret_x_coordinate.to_bytes(byte_size, byteorder="big")
 
     return key_pair.public_key, _key_exchange

libp2p/security/secio/transport.py

@@ -283,7 +283,7 @@ async def _establish_session_parameters(
 
     remote_ephemeral_public_key_bytes = remote_exchange.ephemeral_public_key
     remote_ephemeral_public_key = ECCPublicKey.from_bytes(
-        remote_ephemeral_public_key_bytes
+        remote_ephemeral_public_key_bytes, curve_param
     )
     remote_encryption_parameters.ephemeral_public_key = remote_ephemeral_public_key
     remote_selection = (

setup.py

@@ -41,6 +41,7 @@ setuptools.setup(
         "lru-dict>=1.1.6",
         "protobuf==3.9.0",
         "coincurve>=10.0.0,<11.0.0",
+        "fastecdsa==1.7.4",
     ],
     extras_require=extras_require,
     packages=setuptools.find_packages(exclude=["tests", "tests.*"]),