Verify the remote pubkey and peer_id

- Add `from_bytes` in RSAPublicKey and Secp256k1PublicKey
- Add `pubkey_from_protobuf` to parse pubkey from protobuf
- Verify key and peer_id in `InsecureSession.run_handshake`

libp2p/crypto/rsa.py

@@ -11,6 +11,11 @@ class RSAPublicKey(PublicKey):
     def to_bytes(self) -> bytes:
         return self.impl.export_key("DER")
 
+    @classmethod
+    def from_bytes(cls, key_bytes: bytes) -> "RSAPublicKey":
+        rsakey = RSA.import_key(key_bytes)
+        return cls(rsakey)
+
     def get_type(self) -> KeyType:
         return KeyType.RSA
 
@@ -30,6 +35,11 @@ class RSAPrivateKey(PrivateKey):
     def to_bytes(self) -> bytes:
         return self.impl.export_key("DER")
 
+    @classmethod
+    def from_bytes(cls, key_bytes: bytes) -> "RSAPrivateKey":
+        rsakey = RSA.import_key(key_bytes)
+        return cls(rsakey)
+
     def get_type(self) -> KeyType:
         return KeyType.RSA
 

libp2p/crypto/secp256k1.py

@@ -10,6 +10,11 @@ class Secp256k1PublicKey(PublicKey):
     def to_bytes(self) -> bytes:
         return self.impl.format()
 
+    @classmethod
+    def from_bytes(cls, key_bytes: bytes) -> "Secp256k1PublicKey":
+        secp256k1_pubkey = coincurve.PublicKey(key_bytes)
+        return cls(secp256k1_pubkey)
+
     def get_type(self) -> KeyType:
         return KeyType.Secp256k1
 

libp2p/crypto/utils.py

@@ -0,0 +1,16 @@
+from .keys import PublicKey
+from .pb import crypto_pb2 as protobuf
+from .rsa import RSAPublicKey
+from .secp256k1 import Secp256k1PublicKey
+
+
+def pubkey_from_protobuf(pubkey_pb: protobuf.PublicKey) -> PublicKey:
+    if pubkey_pb.key_type == protobuf.RSA:
+        return RSAPublicKey.from_bytes(pubkey_pb.data)
+    # TODO: Test against secp256k1 keys
+    elif pubkey_pb.key_type == protobuf.Secp256k1:
+        return Secp256k1PublicKey.from_bytes(pubkey_pb.data)
+    else:
+        raise ValueError(
+            f"unsupported key_type={pubkey_pb.key_type}, data={pubkey_pb.data!r}"
+        )