mirror of
https://github.com/varun-r-mallya/py-libp2p.git
synced 2025-12-31 20:36:24 +00:00
feat: Add Windows compatibility using coincurve
This commit is contained in:
@ -1,11 +1,4 @@
|
||||
from fastecdsa import (
|
||||
keys,
|
||||
point,
|
||||
)
|
||||
from fastecdsa import curve as curve_types
|
||||
from fastecdsa.encoding.sec1 import (
|
||||
SEC1Encoder,
|
||||
)
|
||||
import sys
|
||||
|
||||
from libp2p.crypto.keys import (
|
||||
KeyPair,
|
||||
@ -14,67 +7,126 @@ from libp2p.crypto.keys import (
|
||||
PublicKey,
|
||||
)
|
||||
|
||||
if sys.platform != "win32":
|
||||
from fastecdsa import (
|
||||
keys,
|
||||
point,
|
||||
)
|
||||
from fastecdsa import curve as curve_types
|
||||
from fastecdsa.encoding.sec1 import (
|
||||
SEC1Encoder,
|
||||
)
|
||||
else:
|
||||
from coincurve import PrivateKey as CPrivateKey
|
||||
from coincurve import PublicKey as CPublicKey
|
||||
|
||||
def infer_local_type(curve: str) -> curve_types.Curve:
|
||||
|
||||
def infer_local_type(curve: str) -> object:
|
||||
"""
|
||||
Convert a ``str`` representation of some elliptic curve to a
|
||||
Convert a str representation of some elliptic curve to a
|
||||
representation understood by the backend of this module.
|
||||
"""
|
||||
if curve == "P-256":
|
||||
if curve != "P-256":
|
||||
raise NotImplementedError("Only P-256 curve is supported")
|
||||
|
||||
if sys.platform != "win32":
|
||||
return curve_types.P256
|
||||
else:
|
||||
raise NotImplementedError()
|
||||
return "P-256" # coincurve only supports P-256
|
||||
|
||||
|
||||
class ECCPublicKey(PublicKey):
|
||||
def __init__(self, impl: point.Point, curve: curve_types.Curve) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
if sys.platform != "win32":
|
||||
|
||||
def to_bytes(self) -> bytes:
|
||||
return SEC1Encoder.encode_public_key(self.impl, compressed=False)
|
||||
class ECCPublicKey(PublicKey):
|
||||
def __init__(self, impl: point.Point, curve: curve_types.Curve) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
|
||||
@classmethod
|
||||
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 to_bytes(self) -> bytes:
|
||||
return SEC1Encoder.encode_public_key(self.impl, compressed=False)
|
||||
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
@classmethod
|
||||
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 verify(self, data: bytes, signature: bytes) -> bool:
|
||||
raise NotImplementedError()
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
|
||||
def verify(self, data: bytes, signature: bytes) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
class ECCPrivateKey(PrivateKey):
|
||||
def __init__(self, impl: int, curve: curve_types.Curve) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
class ECCPrivateKey(PrivateKey):
|
||||
def __init__(self, impl: int, curve: curve_types.Curve) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
|
||||
@classmethod
|
||||
def new(cls, curve: str) -> "ECCPrivateKey":
|
||||
curve_type = infer_local_type(curve)
|
||||
private_key_impl = keys.gen_private_key(curve_type)
|
||||
return cls(private_key_impl, curve_type)
|
||||
@classmethod
|
||||
def new(cls, curve: str) -> "ECCPrivateKey":
|
||||
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 keys.export_key(self.impl, self.curve)
|
||||
def to_bytes(self) -> bytes:
|
||||
return keys.export_key(self.impl, self.curve)
|
||||
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
|
||||
def sign(self, data: bytes) -> bytes:
|
||||
raise NotImplementedError()
|
||||
def sign(self, data: bytes) -> bytes:
|
||||
raise NotImplementedError()
|
||||
|
||||
def get_public_key(self) -> PublicKey:
|
||||
public_key_impl = keys.get_public_key(self.impl, self.curve)
|
||||
return ECCPublicKey(public_key_impl, self.curve)
|
||||
def get_public_key(self) -> PublicKey:
|
||||
public_key_impl = keys.get_public_key(self.impl, self.curve)
|
||||
return ECCPublicKey(public_key_impl, self.curve)
|
||||
|
||||
else:
|
||||
|
||||
class ECCPublicKey(PublicKey):
|
||||
def __init__(self, impl: CPublicKey, curve: str) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
|
||||
def to_bytes(self) -> bytes:
|
||||
return self.impl.format(compressed=False)
|
||||
|
||||
@classmethod
|
||||
def from_bytes(cls, data: bytes, curve: str) -> "ECCPublicKey":
|
||||
curve_type = infer_local_type(curve)
|
||||
return cls(CPublicKey(data), curve_type) # type: ignore[arg-type]
|
||||
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
|
||||
def verify(self, data: bytes, signature: bytes) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
class ECCPrivateKey(PrivateKey):
|
||||
def __init__(self, impl: CPrivateKey, curve: str) -> None:
|
||||
self.impl = impl
|
||||
self.curve = curve
|
||||
|
||||
@classmethod
|
||||
def new(cls, curve: str) -> "ECCPrivateKey":
|
||||
curve_type = infer_local_type(curve)
|
||||
return cls(CPrivateKey(), curve_type) # type: ignore[arg-type]
|
||||
|
||||
def to_bytes(self) -> bytes:
|
||||
return self.impl.secret
|
||||
|
||||
def get_type(self) -> KeyType:
|
||||
return KeyType.ECC_P256
|
||||
|
||||
def sign(self, data: bytes) -> bytes:
|
||||
raise NotImplementedError()
|
||||
|
||||
def get_public_key(self) -> PublicKey:
|
||||
return ECCPublicKey(self.impl.public_key, self.curve)
|
||||
|
||||
|
||||
def create_new_key_pair(curve: str) -> KeyPair:
|
||||
"""
|
||||
Return a new ECC keypair with the requested ``curve`` type, e.g.
|
||||
Return a new ECC keypair with the requested curve type, e.g.
|
||||
"P-256".
|
||||
"""
|
||||
private_key = ECCPrivateKey.new(curve)
|
||||
|
||||
@ -1,11 +1,26 @@
|
||||
import sys
|
||||
from typing import (
|
||||
Callable,
|
||||
cast,
|
||||
)
|
||||
|
||||
from fastecdsa.encoding import (
|
||||
util,
|
||||
)
|
||||
if sys.platform != "win32":
|
||||
from fastecdsa.encoding import (
|
||||
util,
|
||||
)
|
||||
|
||||
int_bytelen = util.int_bytelen
|
||||
else:
|
||||
from math import (
|
||||
ceil,
|
||||
log2,
|
||||
)
|
||||
|
||||
def int_bytelen(n: int) -> int:
|
||||
if n == 0:
|
||||
return 1
|
||||
return ceil(log2(abs(n) + 1) / 8)
|
||||
|
||||
|
||||
from libp2p.crypto.ecc import (
|
||||
ECCPrivateKey,
|
||||
@ -18,8 +33,6 @@ from libp2p.crypto.keys import (
|
||||
|
||||
SharedKeyGenerator = Callable[[bytes], bytes]
|
||||
|
||||
int_bytelen = util.int_bytelen
|
||||
|
||||
|
||||
def create_ephemeral_key_pair(curve_type: str) -> tuple[PublicKey, SharedKeyGenerator]:
|
||||
"""Facilitates ECDH key exchange."""
|
||||
@ -32,9 +45,15 @@ def create_ephemeral_key_pair(curve_type: str) -> tuple[PublicKey, SharedKeyGene
|
||||
private_key = cast(ECCPrivateKey, key_pair.private_key)
|
||||
|
||||
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")
|
||||
|
||||
if sys.platform != "win32":
|
||||
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")
|
||||
else:
|
||||
# Windows implementation using coincurve
|
||||
shared_key = private_key.impl.ecdh(remote_point.impl.public_key)
|
||||
return shared_key
|
||||
|
||||
return key_pair.public_key, _key_exchange
|
||||
|
||||
Reference in New Issue
Block a user