coincurve.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119
  1. from eth_utils import (
  2. big_endian_to_int,
  3. )
  4. from eth_keys.datatypes import (
  5. BaseSignature,
  6. NonRecoverableSignature,
  7. PrivateKey,
  8. PublicKey,
  9. Signature,
  10. )
  11. from eth_keys.exceptions import (
  12. BadSignature,
  13. )
  14. from eth_keys.utils import (
  15. der,
  16. )
  17. from eth_keys.utils.numeric import (
  18. coerce_low_s,
  19. )
  20. from eth_keys.validation import (
  21. validate_uncompressed_public_key_bytes,
  22. )
  23. from .base import (
  24. BaseECCBackend,
  25. )
  26. def is_coincurve_available() -> bool:
  27. try:
  28. import coincurve # noqa: F401
  29. except ImportError:
  30. return False
  31. else:
  32. return True
  33. class CoinCurveECCBackend(BaseECCBackend):
  34. def __init__(self) -> None:
  35. try:
  36. import coincurve
  37. except ImportError:
  38. raise ImportError(
  39. "The CoinCurveECCBackend requires the coincurve "
  40. "library which is not available for import."
  41. )
  42. self.keys = coincurve.keys
  43. self.ecdsa = coincurve.ecdsa
  44. super().__init__()
  45. def ecdsa_sign(self, msg_hash: bytes, private_key: PrivateKey) -> Signature:
  46. private_key_bytes = private_key.to_bytes()
  47. signature_bytes = self.keys.PrivateKey(private_key_bytes).sign_recoverable(
  48. msg_hash,
  49. hasher=None,
  50. )
  51. signature = Signature(signature_bytes, backend=self)
  52. return signature
  53. def ecdsa_sign_non_recoverable(
  54. self, msg_hash: bytes, private_key: PrivateKey
  55. ) -> NonRecoverableSignature:
  56. private_key_bytes = private_key.to_bytes()
  57. der_encoded_signature = self.keys.PrivateKey(private_key_bytes).sign(
  58. msg_hash,
  59. hasher=None,
  60. )
  61. rs = der.two_int_sequence_decoder(der_encoded_signature)
  62. signature = NonRecoverableSignature(rs=rs, backend=self)
  63. return signature
  64. def ecdsa_verify(
  65. self, msg_hash: bytes, signature: BaseSignature, public_key: PublicKey
  66. ) -> bool:
  67. # coincurve rejects signatures with a high s,
  68. # so convert to the equivalent low s form
  69. low_s = coerce_low_s(signature.s)
  70. der_encoded_signature = der.two_int_sequence_encoder(signature.r, low_s)
  71. coincurve_public_key = self.keys.PublicKey(b"\x04" + public_key.to_bytes())
  72. return coincurve_public_key.verify(
  73. der_encoded_signature,
  74. msg_hash,
  75. hasher=None,
  76. )
  77. def ecdsa_recover(self, msg_hash: bytes, signature: Signature) -> PublicKey:
  78. signature_bytes = signature.to_bytes()
  79. try:
  80. public_key_bytes = self.keys.PublicKey.from_signature_and_message(
  81. signature_bytes,
  82. msg_hash,
  83. hasher=None,
  84. ).format(compressed=False)[1:]
  85. except Exception as err:
  86. raise BadSignature(str(err))
  87. public_key = PublicKey(public_key_bytes, backend=self)
  88. return public_key
  89. def private_key_to_public_key(self, private_key: PrivateKey) -> PublicKey:
  90. public_key_bytes = self.keys.PrivateKey(
  91. private_key.to_bytes()
  92. ).public_key.format(compressed=False,)[1:]
  93. return PublicKey(public_key_bytes, backend=self)
  94. def decompress_public_key_bytes(self, compressed_public_key_bytes: bytes) -> bytes:
  95. public_key = self.keys.PublicKey(compressed_public_key_bytes)
  96. return public_key.format(compressed=False)[1:]
  97. def compress_public_key_bytes(self, uncompressed_public_key_bytes: bytes) -> bytes:
  98. validate_uncompressed_public_key_bytes(uncompressed_public_key_bytes)
  99. point = (
  100. big_endian_to_int(uncompressed_public_key_bytes[:32]),
  101. big_endian_to_int(uncompressed_public_key_bytes[32:]),
  102. )
  103. public_key = self.keys.PublicKey.from_point(*point)
  104. return public_key.format(compressed=True)