Replace crypto lib ecdsa with coincurve

v4-client-py-v2/dydx_v4_client/indexer/rest/noble_client.py

@@ -1,15 +1,13 @@
 from typing import List, Optional
 
 import grpc
-from ecdsa.util import sigencode_string_canonize
 from v4_proto.cosmos.auth.v1beta1 import query_pb2_grpc as auth
 from v4_proto.cosmos.auth.v1beta1.auth_pb2 import BaseAccount
 from v4_proto.cosmos.auth.v1beta1.query_pb2 import QueryAccountRequest
 from v4_proto.cosmos.bank.v1beta1 import query_pb2 as bank_query
 from v4_proto.cosmos.bank.v1beta1 import query_pb2_grpc as bank_query_grpc
 from v4_proto.cosmos.base.abci.v1beta1.abci_pb2 import TxResponse
 from v4_proto.cosmos.base.v1beta1.coin_pb2 import Coin
-from v4_proto.cosmos.crypto.secp256k1.keys_pb2 import PubKey
 from v4_proto.cosmos.tx.signing.v1beta1.signing_pb2 import SignMode
 from v4_proto.cosmos.tx.v1beta1 import service_pb2_grpc
 from v4_proto.cosmos.tx.v1beta1.service_pb2 import (
@@ -29,7 +27,8 @@
 
 from dydx_v4_client.config import GAS_MULTIPLIER
 from dydx_v4_client.node.builder import as_any
-from dydx_v4_client.wallet import from_mnemonic
+from dydx_v4_client.key_pair import KeyPair
+from dydx_v4_client.wallet import Wallet
 
 
 class NobleClient:
@@ -72,8 +71,8 @@ async def connect(self, mnemonic: str):
         """
         if not mnemonic:
             raise ValueError("Mnemonic not provided")
-        private_key = from_mnemonic(mnemonic)
-        self.wallet = private_key
+        key_pair = KeyPair.from_mnemonic(mnemonic)
+        self.wallet = Wallet(key_pair, 0, 0)
         self.channel = grpc.secure_channel(
             self.rest_endpoint,
             grpc.ssl_channel_credentials(),
@@ -178,26 +177,19 @@ async def send(
 
         # Sign and broadcast the transaction
         signer_info = SignerInfo(
-            public_key=as_any(
-                PubKey(key=self.wallet.get_verifying_key().to_string("compressed"))
-            ),
+            public_key=as_any(self.wallet.public_key),
             mode_info=ModeInfo(single=ModeInfo.Single(mode=SignMode.SIGN_MODE_DIRECT)),
-            sequence=self.get_account(
-                self.wallet.get_verifying_key().to_string()
-            ).sequence,
+            sequence=self.get_account(self.wallet.address).sequence,
         )
         body = TxBody(messages=messages, memo=memo or self.default_client_memo)
         auth_info = AuthInfo(signer_infos=[signer_info], fee=fee)
-        signature = self.wallet.sign(
+        signature = self.wallet.key.sign(
             SignDoc(
                 body_bytes=body.SerializeToString(),
                 auth_info_bytes=auth_info.SerializeToString(),
-                account_number=self.get_account(
-                    self.wallet.get_verifying_key().to_string()
-                ).account_number,
+                account_number=self.get_account(self.wallet.address).account_number,
                 chain_id=self.chain_id,
-            ).SerializeToString(),
-            sigencode=sigencode_string_canonize,
+            ).SerializeToString()
         )
 
         tx = Tx(body=body, auth_info=auth_info, signatures=[signature])
@@ -233,13 +225,9 @@ async def simulate_transaction(
 
         # Get simulated response
         signer_info = SignerInfo(
-            public_key=as_any(
-                PubKey(key=self.wallet.get_verifying_key().to_string("compressed"))
-            ),
+            public_key=as_any(self.wallet.public_key),
             mode_info=ModeInfo(single=ModeInfo.Single(mode=SignMode.SIGN_MODE_DIRECT)),
-            sequence=self.get_account(
-                self.wallet.get_verifying_key().to_string()
-            ).sequence,
+            sequence=self.get_account(self.wallet.address).sequence,
         )
         body = TxBody(messages=messages, memo=memo or self.default_client_memo)
         auth_info = AuthInfo(signer_infos=[signer_info], fee=Fee(gas_limit=0))

v4-client-py-v2/dydx_v4_client/key_pair.py

@@ -0,0 +1,78 @@
+"""
+This module implements ECDSA (Elliptic Curve Digital Signature Algorithm) key pair wrapper class. Initaialy `ecdsa.SigningKey` was directly used. However due to security concrens and to avoid direct dependency on specific implementation, `KeyPair class was introduced. This class provides a wrapper around the `coincurve.PrivateKey` and mimics how `ecdsa` was used before.
+"""
+
+from dataclasses import dataclass
+from typing import Tuple
+
+from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coins
+from coincurve import PrivateKey
+from coincurve.utils import GROUP_ORDER_INT, int_to_bytes
+
+
+def bytes_from_mnemonic(mnemonic: str) -> bytes:
+    seed = Bip39SeedGenerator(mnemonic).Generate()
+    return (
+        Bip44.FromSeed(seed, Bip44Coins.COSMOS)
+        .DeriveDefaultPath()
+        .PrivateKey()
+        .Raw()
+        .ToBytes()
+    )
+
+
+@dataclass
+class KeyPair:
+    """
+    Wrapper class around `coincurve.PrivateKey` to mimic `ecdsa.SigningKey` behavior.
+    """
+
+    key: PrivateKey
+
+    @staticmethod
+    def from_mnemonic(mnemonic: str) -> "KeyPair":
+        """
+        Creates a private key from a mnemonic.
+        """
+        return KeyPair(PrivateKey(bytes_from_mnemonic(mnemonic)))
+
+    @staticmethod
+    def from_hex(hex_key: str) -> "KeyPair":
+        """
+        Creates a private key from a hex string.
+        """
+        return KeyPair(PrivateKey.from_hex(hex_key))
+
+    def sign(self, message: bytes) -> bytes:
+        """
+        Signs a message using the private key. Signature is encoded the same way as `ecdsa.util.sigencode_string_canonize`, to cointains 64-bytes.
+        """
+        signature = self.key.sign_recoverable(message)
+        return coinsign_canonize(signature)
+
+    @property
+    def public_key_bytes(self) -> bytes:
+        """
+        Returns the public key bytes of the key pair in compressed format.
+        """
+        return self.key.public_key.format(compressed=True)
+
+
+def coinsign_extract(signature: bytes) -> Tuple[int, int]:
+    assert len(signature) == 65
+
+    r = int.from_bytes(signature[:32], "big")
+    s = int.from_bytes(signature[32:64], "big")
+
+    return r, s
+
+
+def coinsign_canonize(signature: bytes) -> bytes:
+    r, s = coinsign_extract(signature)
+
+    if s > GROUP_ORDER_INT // 2:
+        s = GROUP_ORDER_INT - s
+
+    r_bytes = int_to_bytes(r)
+    s_bytes = int_to_bytes(s)
+    return r_bytes.rjust(32, b"\x00") + s_bytes.rjust(32, b"\x00")

v4-client-py-v2/dydx_v4_client/node/builder.py

@@ -2,7 +2,6 @@
 from typing import List
 
 import google
-from ecdsa.util import sigencode_string_canonize
 from google.protobuf.message import Message
 from v4_proto.cosmos.base.v1beta1.coin_pb2 import Coin
 from v4_proto.cosmos.tx.signing.v1beta1.signing_pb2 import SignMode
@@ -34,17 +33,15 @@ def get_signer_info(public_key, sequence):
     )
 
 
-def get_signature(private_key, body, auth_info, account_number, chain_id):
+def get_signature(key_pair, body, auth_info, account_number, chain_id):
     signdoc = SignDoc(
         body_bytes=body.SerializeToString(),
         auth_info_bytes=auth_info.SerializeToString(),
         account_number=account_number,
         chain_id=chain_id,
     )
 
-    return private_key.sign(
-        signdoc.SerializeToString(), sigencode=sigencode_string_canonize
-    )
+    return key_pair.sign(signdoc.SerializeToString())
 
 
 DEFAULT_FEE = Fee(

v4-client-py-v2/dydx_v4_client/wallet.py

@@ -1,56 +1,38 @@
 import hashlib
 from dataclasses import dataclass
-from functools import partial
 from typing import TYPE_CHECKING
 
 import bech32
-import ecdsa
 from Crypto.Hash import RIPEMD160
-from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coins
+
 from v4_proto.cosmos.crypto.secp256k1.keys_pb2 import PubKey
 
+from dydx_v4_client.key_pair import KeyPair
+
 if TYPE_CHECKING:
     from dydx_v4_client.node.client import NodeClient
 
 
-from_string = partial(
-    ecdsa.SigningKey.from_string, curve=ecdsa.SECP256k1, hashfunc=hashlib.sha256
-)
-
-
-def bytes_from_mnemonic(mnemonic: str) -> bytes:
-    seed = Bip39SeedGenerator(mnemonic).Generate()
-    return (
-        Bip44.FromSeed(seed, Bip44Coins.COSMOS)
-        .DeriveDefaultPath()
-        .PrivateKey()
-        .Raw()
-        .ToBytes()
-    )
-
-
-def from_mnemonic(mnemonic: str) -> ecdsa.SigningKey:
-    return from_string(bytes_from_mnemonic(mnemonic))
-
-
 @dataclass
 class Wallet:
-    key: ecdsa.SigningKey
+    key: KeyPair
     account_number: int
     sequence: int
 
     @staticmethod
     async def from_mnemonic(node: "NodeClient", mnemonic: str, address: str):
         account = await node.get_account(address)
-        return Wallet(from_mnemonic(mnemonic), account.account_number, account.sequence)
+        return Wallet(
+            KeyPair.from_mnemonic(mnemonic), account.account_number, account.sequence
+        )
 
     @property
     def public_key(self) -> PubKey:
-        return PubKey(key=self.key.get_verifying_key().to_string("compressed"))
+        return PubKey(key=self.key.public_key_bytes)
 
     @property
     def address(self) -> str:
-        public_key_bytes = self.public_key.key
+        public_key_bytes = self.key.public_key_bytes
         sha256_hash = hashlib.sha256(public_key_bytes).digest()
         ripemd160_hash = RIPEMD160.new(sha256_hash).digest()
         return bech32.bech32_encode("dydx", bech32.convertbits(ripemd160_hash, 8, 5))

v4-client-py-v2/examples/basic_adder.py

@@ -13,7 +13,8 @@
 from dydx_v4_client.network import TESTNET
 from dydx_v4_client.node.client import NodeClient
 from dydx_v4_client.node.message import order, order_id
-from dydx_v4_client.wallet import Wallet, from_string
+from dydx_v4_client.key_pair import KeyPair
+from dydx_v4_client.wallet import Wallet
 from tests.conftest import DYDX_TEST_PRIVATE_KEY
 
 logging.basicConfig(
@@ -37,7 +38,7 @@ def __init__(
         self, node_client: NodeClient, address: str, key: str, subaccount_number: int
     ):
         self.address = address
-        self.key = from_string(bytes.fromhex(key))
+        self.key = KeyPair.from_hex(key)
         self.subaccount_number = subaccount_number
         self.node_client = node_client
         self.testnet_indexer_socket = IndexerSocket(