davide/purple-electrumwallet
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

ecc: refactor/clean-up sign/verify APIs

Browse Source
This commit is contained in:
SomberNight
2024-04-11 15:13:41 +00:00
parent e72210b5d2
commit bd9d0ccc33
22 changed files with 165 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
electrum/channel_db.py
+3 -2
View File
@@ -46,6 +46,7 @@ from .lnutil import (LNPeerAddr, format_short_channel_id, ShortChannelID,
from .lnverifier import LNChannelVerifier, verify_sig_for_channel_update
from .lnmsg import decode_msg
from . import ecc
from .ecc import ECPubkey
from .crypto import sha256d
from .lnmsg import FailedToParseMsg
@@ -605,7 +606,7 @@ class ChannelDB(SqlDB):
pubkeys = [payload['node_id_1'], payload['node_id_2'], payload['bitcoin_key_1'], payload['bitcoin_key_2']]
sigs = [payload['node_signature_1'], payload['node_signature_2'], payload['bitcoin_signature_1'], payload['bitcoin_signature_2']]
for pubkey, sig in zip(pubkeys, sigs):
if not ecc.verify_signature(pubkey, sig, h):
if not ECPubkey(pubkey).ecdsa_verify(sig, h):
raise InvalidGossipMsg('signature failed')
@classmethod
@@ -613,7 +614,7 @@ class ChannelDB(SqlDB):
pubkey = payload['node_id']
signature = payload['signature']
h = sha256d(payload['raw'][66:])
if not ecc.verify_signature(pubkey, signature, h):
if not ECPubkey(pubkey).ecdsa_verify(signature, h):
raise InvalidGossipMsg('signature failed')
def add_node_announcements(self, msg_payloads):
electrum/commands.py
+1 -1
View File
@@ -713,7 +713,7 @@ class Commands:
"""Verify a signature."""
sig = base64.b64decode(signature)
message = util.to_bytes(message)
return ecc.verify_message_with_address(address, sig, message)
return ecc.verify_usermessage_with_address(address, sig, message)
@command('wp')
async def payto(self, destination, amount, fee=None, feerate=None, from_addr=None, from_coins=None, change_addr=None,
electrum/ecc.py
+74 -81
View File
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Electrum - lightweight Bitcoin client
# Copyright (C) 2018 The Electrum developers
# Copyright (C) 2018-2024 The Electrum developers
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
@@ -28,8 +28,7 @@ import hashlib
import functools
from typing import Union, Tuple, Optional
from ctypes import (
byref, c_byte, c_int, c_uint, c_char_p, c_size_t, c_void_p, create_string_buffer,
CFUNCTYPE, POINTER, cast
byref, c_char_p, c_size_t, create_string_buffer, cast,
)
from .util import bfh, assert_bytes, to_bytes, InvalidPassword, profiler, randrange
@@ -50,21 +49,22 @@ def string_to_number(b: bytes) -> int:
return int.from_bytes(b, byteorder='big', signed=False)
def sig_string_from_der_sig(der_sig: bytes) -> bytes:
r, s = get_r_and_s_from_der_sig(der_sig)
return sig_string_from_r_and_s(r, s)
def ecdsa_sig64_from_der_sig(der_sig: bytes) -> bytes:
r, s = get_r_and_s_from_ecdsa_der_sig(der_sig)
return ecdsa_sig64_from_r_and_s(r, s)
def der_sig_from_sig_string(sig_string: bytes) -> bytes:
r, s = get_r_and_s_from_sig_string(sig_string)
return der_sig_from_r_and_s(r, s)
def ecdsa_der_sig_from_ecdsa_sig64(sig64: bytes) -> bytes:
r, s = get_r_and_s_from_ecdsa_sig64(sig64)
return ecdsa_der_sig_from_r_and_s(r, s)
def der_sig_from_r_and_s(r: int, s: int) -> bytes:
sig_string = (int.to_bytes(r, length=32, byteorder="big") +
def ecdsa_der_sig_from_r_and_s(r: int, s: int) -> bytes:
sig64 = (
int.to_bytes(r, length=32, byteorder="big") +
int.to_bytes(s, length=32, byteorder="big"))
sig = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig_string)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig64)
if 1 != ret:
raise Exception("Bad signature")
ret = _libsecp256k1.secp256k1_ecdsa_signature_normalize(_libsecp256k1.ctx, sig, sig)
@@ -77,7 +77,7 @@ def der_sig_from_r_and_s(r: int, s: int) -> bytes:
return bytes(der_sig)[:der_sig_size]
def get_r_and_s_from_der_sig(der_sig: bytes) -> Tuple[int, int]:
def get_r_and_s_from_ecdsa_der_sig(der_sig: bytes) -> Tuple[int, int]:
assert isinstance(der_sig, bytes)
sig = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_der(_libsecp256k1.ctx, sig, der_sig, len(der_sig))
@@ -91,11 +91,11 @@ def get_r_and_s_from_der_sig(der_sig: bytes) -> Tuple[int, int]:
return r, s
def get_r_and_s_from_sig_string(sig_string: bytes) -> Tuple[int, int]:
if not (isinstance(sig_string, bytes) and len(sig_string) == 64):
raise Exception("sig_string must be bytes, and 64 bytes exactly")
def get_r_and_s_from_ecdsa_sig64(sig64: bytes) -> Tuple[int, int]:
if not (isinstance(sig64, bytes) and len(sig64) == 64):
raise Exception("sig64 must be bytes, and 64 bytes exactly")
sig = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig_string)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig64)
if 1 != ret:
raise Exception("Bad signature")
ret = _libsecp256k1.secp256k1_ecdsa_signature_normalize(_libsecp256k1.ctx, sig, sig)
@@ -106,11 +106,12 @@ def get_r_and_s_from_sig_string(sig_string: bytes) -> Tuple[int, int]:
return r, s
def sig_string_from_r_and_s(r: int, s: int) -> bytes:
sig_string = (int.to_bytes(r, length=32, byteorder="big") +
def ecdsa_sig64_from_r_and_s(r: int, s: int) -> bytes:
sig64 = (
int.to_bytes(r, length=32, byteorder="big") +
int.to_bytes(s, length=32, byteorder="big"))
sig = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig_string)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig64)
if 1 != ret:
raise Exception("Bad signature")
ret = _libsecp256k1.secp256k1_ecdsa_signature_normalize(_libsecp256k1.ctx, sig, sig)
@@ -155,28 +156,31 @@ class ECPubkey(object):
self._x, self._y = None, None
@classmethod
def from_sig_string(cls, sig_string: bytes, recid: int, msg_hash: bytes) -> 'ECPubkey':
assert_bytes(sig_string)
if len(sig_string) != 64:
raise Exception(f'wrong encoding used for signature? len={len(sig_string)} (should be 64)')
def from_ecdsa_sig64(cls, sig64: bytes, recid: int, msg32: bytes) -> 'ECPubkey':
assert_bytes(sig64)
if len(sig64) != 64:
raise Exception(f'wrong encoding used for signature? len={len(sig64)} (should be 64)')
if not (0 <= recid <= 3):
raise ValueError('recid is {}, but should be 0 <= recid <= 3'.format(recid))
assert isinstance(msg32, (bytes, bytearray)), type(msg32)
assert len(msg32) == 32, len(msg32)
sig65 = create_string_buffer(65)
ret = _libsecp256k1.secp256k1_ecdsa_recoverable_signature_parse_compact(
_libsecp256k1.ctx, sig65, sig_string, recid)
_libsecp256k1.ctx, sig65, sig64, recid)
if 1 != ret:
raise Exception('failed to parse signature')
pubkey = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_recover(_libsecp256k1.ctx, pubkey, sig65, msg_hash)
ret = _libsecp256k1.secp256k1_ecdsa_recover(_libsecp256k1.ctx, pubkey, sig65, msg32)
if 1 != ret:
raise InvalidECPointException('failed to recover public key')
return ECPubkey._from_libsecp256k1_pubkey_ptr(pubkey)
@classmethod
def from_signature65(cls, sig: bytes, msg_hash: bytes) -> Tuple['ECPubkey', bool, Optional[str]]:
if len(sig) != 65:
raise Exception(f'wrong encoding used for signature? len={len(sig)} (should be 65)')
nV = sig[0]
def from_ecdsa_sig65(cls, sig65: bytes, msg32: bytes) -> Tuple['ECPubkey', bool, Optional[str]]:
assert_bytes(sig65)
if len(sig65) != 65:
raise Exception(f'wrong encoding used for signature? len={len(sig65)} (should be 65)')
nV = sig65[0]
# as per BIP-0137:
# 27-30: p2pkh (uncompressed)
# 31-34: p2pkh (compressed)
@@ -199,7 +203,7 @@ class ECPubkey(object):
else:
compressed = False
recid = nV - 27
pubkey = cls.from_sig_string(sig[1:], recid, msg_hash)
pubkey = cls.from_ecdsa_sig64(sig65[1:], recid, msg32)
return pubkey, compressed, txin_type_guess
@classmethod
@@ -321,38 +325,36 @@ class ECPubkey(object):
p2 = ((other.x() or 0), (other.y() or 0))
return p1 < p2
def verify_message_for_address(self, sig65: bytes, message: bytes, algo=lambda x: sha256d(msg_magic(x))) -> bool:
assert_bytes(message)
h = algo(message)
def ecdsa_verify_recoverable(self, sig65: bytes, msg32: bytes) -> bool:
try:
public_key, compressed, txin_type_guess = self.from_signature65(sig65, h)
public_key, _compressed, _txin_type_guess = self.from_ecdsa_sig65(sig65, msg32)
except Exception:
return False
# check public key
if public_key != self:
return False
# check message
return self.verify_message_hash(sig65[1:], h)
return self.ecdsa_verify(sig65[1:], msg32)
def verify_message_hash(self, sig_string: bytes, msg_hash: bytes) -> bool:
assert_bytes(sig_string)
if len(sig_string) != 64:
def ecdsa_verify(self, sig64: bytes, msg32: bytes) -> bool:
assert_bytes(sig64)
if len(sig64) != 64:
return False
if not (isinstance(msg_hash, bytes) and len(msg_hash) == 32):
if not (isinstance(msg32, bytes) and len(msg32) == 32):
return False
sig = create_string_buffer(64)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig_string)
ret = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, sig64)
if 1 != ret:
return False
ret = _libsecp256k1.secp256k1_ecdsa_signature_normalize(_libsecp256k1.ctx, sig, sig)
pubkey = self._to_libsecp256k1_pubkey_ptr()
if 1 != _libsecp256k1.secp256k1_ecdsa_verify(_libsecp256k1.ctx, sig, msg_hash, pubkey):
if 1 != _libsecp256k1.secp256k1_ecdsa_verify(_libsecp256k1.ctx, sig, msg32, pubkey):
return False
return True
def verify_message_schnorr(self, sig64: bytes, msg32: bytes) -> bool:
def schnorr_verify(self, sig64: bytes, msg32: bytes) -> bool:
assert isinstance(sig64, bytes), type(sig64)
assert len(sig64) == 64, len(sig64)
assert isinstance(msg32, bytes), type(msg32)
@@ -363,7 +365,7 @@ class ECPubkey(object):
return False
return True
def encrypt_message(self, message: bytes, magic: bytes = b'BIE1') -> bytes:
def encrypt_message(self, message: bytes, *, magic: bytes = b'BIE1') -> bytes:
"""
ECIES encryption/decryption methods; AES-128-CBC with PKCS7 is used as the cipher; hmac-sha256 is used as the mac
"""
@@ -402,23 +404,19 @@ CURVE_ORDER = 0xFFFFFFFF_FFFFFFFF_FFFFFFFF_FFFFFFFE_BAAEDCE6_AF48A03B_BFD25E8C_D
POINT_AT_INFINITY = ECPubkey(None)
def msg_magic(message: bytes) -> bytes:
def usermessage_magic(message: bytes) -> bytes:
from .bitcoin import var_int
length = bfh(var_int(len(message)))
return b"\x18Bitcoin Signed Message:\n" + length + message
def verify_signature(pubkey: bytes, sig: bytes, h: bytes) -> bool:
return ECPubkey(pubkey).verify_message_hash(sig, h)
def verify_message_with_address(address: str, sig65: bytes, message: bytes, *, net=None) -> bool:
def verify_usermessage_with_address(address: str, sig65: bytes, message: bytes, *, net=None) -> bool:
from .bitcoin import pubkey_to_address
assert_bytes(sig65, message)
if net is None: net = constants.net
h = sha256d(msg_magic(message))
h = sha256d(usermessage_magic(message))
try:
public_key, compressed, txin_type_guess = ECPubkey.from_signature65(sig65, h)
public_key, compressed, txin_type_guess = ECPubkey.from_ecdsa_sig65(sig65, h)
except Exception as e:
return False
# check public key using the address
@@ -431,7 +429,7 @@ def verify_message_with_address(address: str, sig65: bytes, message: bytes, *, n
else:
return False
# check message
return public_key.verify_message_hash(sig65[1:], h)
return public_key.ecdsa_verify(sig65[1:], h)
def is_secret_within_curve_range(secret: Union[int, bytes]) -> bool:
@@ -487,18 +485,18 @@ class ECPrivkey(ECPubkey):
def get_secret_bytes(self) -> bytes:
return int.to_bytes(self.secret_scalar, length=32, byteorder='big', signed=False)
def sign(self, msg_hash: bytes, sigencode=None) -> bytes:
if not (isinstance(msg_hash, bytes) and len(msg_hash) == 32):
raise Exception("msg_hash to be signed must be bytes, and 32 bytes exactly")
def ecdsa_sign(self, msg32: bytes, *, sigencode=None) -> bytes:
if not (isinstance(msg32, bytes) and len(msg32) == 32):
raise Exception("msg32 to be signed must be bytes, and 32 bytes exactly")
if sigencode is None:
sigencode = sig_string_from_r_and_s
sigencode = ecdsa_sig64_from_r_and_s
privkey_bytes = self.secret_scalar.to_bytes(32, byteorder="big")
nonce_function = None
sig = create_string_buffer(64)
def sign_with_extra_entropy(extra_entropy):
ret = _libsecp256k1.secp256k1_ecdsa_sign(
_libsecp256k1.ctx, sig, msg_hash, privkey_bytes,
_libsecp256k1.ctx, sig, msg32, privkey_bytes,
nonce_function, extra_entropy)
if 1 != ret:
raise Exception('the nonce generation function failed, or the private key was invalid')
@@ -516,14 +514,14 @@ class ECPrivkey(ECPubkey):
extra_entropy = counter.to_bytes(32, byteorder="little")
r, s = sign_with_extra_entropy(extra_entropy=extra_entropy)
sig_string = sig_string_from_r_and_s(r, s)
if not self.verify_message_hash(sig_string, msg_hash):
sig64 = ecdsa_sig64_from_r_and_s(r, s)
if not self.ecdsa_verify(sig64, msg32):
raise Exception("sanity check failed: signature we just created does not verify!")
sig = sigencode(r, s)
return sig
def sign_schnorr(self, msg32: bytes, *, aux_rand32: bytes = None) -> bytes:
def schnorr_sign(self, msg32: bytes, *, aux_rand32: bytes = None) -> bytes:
assert isinstance(msg32, bytes), type(msg32)
assert len(msg32) == 32, len(msg32)
if aux_rand32 is None:
@@ -543,35 +541,30 @@ class ECPrivkey(ECPubkey):
sig64 = bytes(sig64)
if 1 != ret:
raise Exception('signing failure')
if not self.verify_message_schnorr(sig64, msg32):
if not self.schnorr_verify(sig64, msg32):
raise Exception("sanity check failed: signature we just created does not verify!")
return sig64
def sign_transaction(self, hashed_preimage: bytes) -> bytes:
return self.sign(hashed_preimage, sigencode=der_sig_from_r_and_s)
def sign_message(
self,
message: Union[bytes, str],
is_compressed: bool,
algo=lambda x: sha256d(msg_magic(x)),
) -> bytes:
def bruteforce_recid(sig_string):
def ecdsa_sign_recoverable(self, msg32: bytes, *, is_compressed: bool) -> bytes:
def bruteforce_recid(sig64: bytes):
for recid in range(4):
sig65 = construct_sig65(sig_string, recid, is_compressed)
if not self.verify_message_for_address(sig65, message, algo):
sig65 = construct_ecdsa_sig65(sig64, recid, is_compressed=is_compressed)
if not self.ecdsa_verify_recoverable(sig65, msg32):
continue
return sig65, recid
else:
raise Exception("error: cannot sign message. no recid fits..")
message = to_bytes(message, 'utf8')
msg_hash = algo(message)
sig_string = self.sign(msg_hash, sigencode=sig_string_from_r_and_s)
sig65, recid = bruteforce_recid(sig_string)
sig64 = self.ecdsa_sign(msg32, sigencode=ecdsa_sig64_from_r_and_s)
sig65, recid = bruteforce_recid(sig64)
return sig65
def decrypt_message(self, encrypted: Union[str, bytes], magic: bytes=b'BIE1') -> bytes:
def ecdsa_sign_usermessage(self, message: Union[bytes, str], *, is_compressed: bool) -> bytes:
message = to_bytes(message, 'utf8')
msg32 = sha256d(usermessage_magic(message))
return self.ecdsa_sign_recoverable(msg32, is_compressed=is_compressed)
def decrypt_message(self, encrypted: Union[str, bytes], *, magic: bytes=b'BIE1') -> bytes:
encrypted = base64.b64decode(encrypted) # type: bytes
if len(encrypted) < 85:
raise Exception('invalid ciphertext: length')
@@ -593,6 +586,6 @@ class ECPrivkey(ECPubkey):
return aes_decrypt_with_iv(key_e, iv, ciphertext)
def construct_sig65(sig_string: bytes, recid: int, is_compressed: bool) -> bytes:
def construct_ecdsa_sig65(sig64: bytes, recid: int, *, is_compressed: bool) -> bytes:
comp = 4 if is_compressed else 0
return bytes([27 + recid + comp]) + sig_string
return bytes([27 + recid + comp]) + sig64
electrum/gui/qml/qedaemon.py
+2 -2
View File
@@ -12,7 +12,7 @@ from electrum.util import WalletFileException, standardize_path, InvalidPassword
from electrum.plugin import run_hook
from electrum.lnchannel import ChannelState
from electrum.bitcoin import is_address
from electrum.ecc import verify_message_with_address
from electrum.ecc import verify_usermessage_with_address
from electrum.storage import StorageReadWriteError
from .auth import AuthMixin, auth_protect
@@ -372,7 +372,7 @@ class QEDaemon(AuthMixin, QObject):
try:
# This can throw on invalid base64
sig = base64.b64decode(str(signature.strip()))
verified = verify_message_with_address(address, sig, message)
verified = verify_usermessage_with_address(address, sig, message)
except Exception as e:
verified = False
return verified
electrum/gui/qt/main_window.py
+1 -1
View File
@@ -1982,7 +1982,7 @@ class ElectrumWindow(QMainWindow, MessageBoxMixin, Logger, QtEventListener):
try:
# This can throw on invalid base64
sig = base64.b64decode(str(signature.toPlainText()))
verified = ecc.verify_message_with_address(address, sig, message)
verified = ecc.verify_usermessage_with_address(address, sig, message)
except Exception as e:
verified = False
if verified:
electrum/gui/qt/update_checker.py
+1 -1
View File
@@ -123,7 +123,7 @@ class UpdateCheckThread(QThread, Logger):
continue
sig = base64.b64decode(sig)
msg = version_num.encode('utf-8')
if ecc.verify_message_with_address(address=address, sig65=sig, message=msg,
if ecc.verify_usermessage_with_address(address=address, sig65=sig, message=msg,
net=constants.BitcoinMainnet):
self.logger.info(f"valid sig for version announcement '{version_num}' from address '{address}'")
break
electrum/keystore.py
+1 -1
View File
@@ -223,7 +223,7 @@ class Software_KeyStore(KeyStore):
def sign_message(self, sequence, message, password, *, script_type=None) -> bytes:
privkey, compressed = self.get_private_key(sequence, password)
key = ecc.ECPrivkey(privkey)
return key.sign_message(message, compressed)
return key.ecdsa_sign_usermessage(message, is_compressed=compressed)
def decrypt_message(self, sequence, message, password) -> bytes:
privkey, compressed = self.get_private_key(sequence, password)
electrum/lnaddr.py
+4 -3
View File
@@ -252,8 +252,9 @@ def lnencode(addr: 'LnAddr', privkey) -> str:
# We actually sign the hrp, then data (padded to 8 bits with zeroes).
msg = hrp.encode("ascii") + data.tobytes()
msg32 = sha256(msg).digest()
privkey = ecc.ECPrivkey(privkey)
sig = privkey.sign_message(msg, is_compressed=False, algo=lambda x:sha256(x).digest())
sig = privkey.ecdsa_sign_recoverable(msg32, is_compressed=False)
recovery_flag = bytes([sig[0] - 27])
sig = bytes(sig[1:]) + recovery_flag
data += sig
@@ -550,13 +551,13 @@ def lndecode(invoice: str, *, verbose=False, net=None) -> LnAddr:
#
# A reader MUST use the `n` field to validate the signature instead of
# performing signature recovery if a valid `n` field is provided.
if not ecc.ECPubkey(addr.pubkey).verify_message_hash(sigdecoded[:64], hrp_hash):
if not ecc.ECPubkey(addr.pubkey).ecdsa_verify(sigdecoded[:64], hrp_hash):
raise LnDecodeException("bad signature")
pubkey_copy = addr.pubkey
class WrappedBytesKey:
serialize = lambda: pubkey_copy
addr.pubkey = WrappedBytesKey
else: # Recover pubkey from signature.
addr.pubkey = SerializableKey(ecc.ECPubkey.from_sig_string(sigdecoded[:64], sigdecoded[64], hrp_hash))
addr.pubkey = SerializableKey(ecc.ECPubkey.from_ecdsa_sig64(sigdecoded[:64], sigdecoded[64], hrp_hash))
return addr
electrum/lnchannel.py
+9 -8
View File
@@ -34,6 +34,7 @@ from aiorpcx import NetAddress
import attr
from . import ecc
from .ecc import ECPubkey
from . import constants, util
from .util import bfh, chunks, TxMinedInfo
from .invoices import PR_PAID
@@ -802,7 +803,7 @@ class Channel(AbstractChannel):
timestamp=now(),
)
sighash = sha256d(chan_upd[2 + 64:])
sig = ecc.ECPrivkey(self.lnworker.node_keypair.privkey).sign(sighash, ecc.sig_string_from_r_and_s)
sig = ecc.ECPrivkey(self.lnworker.node_keypair.privkey).ecdsa_sign(sighash, sigencode=ecc.ecdsa_sig64_from_r_and_s)
message_type, payload = decode_msg(chan_upd)
payload['signature'] = sig
chan_upd = encode_msg(message_type, **payload)
@@ -1099,7 +1100,7 @@ class Channel(AbstractChannel):
ctx_output_idx=ctx_output_idx,
htlc=htlc)
sig = bfh(htlc_tx.sign_txin(0, their_remote_htlc_privkey))
htlc_sig = ecc.sig_string_from_der_sig(sig[:-1])
htlc_sig = ecc.ecdsa_sig64_from_der_sig(sig[:-1])
htlcsigs.append((ctx_output_idx, htlc_sig))
htlcsigs.sort()
htlcsigs = [x[1] for x in htlcsigs]
@@ -1122,7 +1123,7 @@ class Channel(AbstractChannel):
pending_local_commitment = self.get_next_commitment(LOCAL)
preimage_hex = pending_local_commitment.serialize_preimage(0)
pre_hash = sha256d(bfh(preimage_hex))
if not ecc.verify_signature(self.config[REMOTE].multisig_key.pubkey, sig, pre_hash):
if not ECPubkey(self.config[REMOTE].multisig_key.pubkey).ecdsa_verify(sig, pre_hash):
raise LNProtocolWarning(
f'failed verifying signature for our updated commitment transaction. '
f'sig={sig.hex()}. '
@@ -1169,7 +1170,7 @@ class Channel(AbstractChannel):
preimage_hex = htlc_tx.serialize_preimage(0)
pre_hash = sha256d(bfh(preimage_hex))
remote_htlc_pubkey = derive_pubkey(self.config[REMOTE].htlc_basepoint.pubkey, pcp)
if not ecc.verify_signature(remote_htlc_pubkey, htlc_sig, pre_hash):
if not ECPubkey(remote_htlc_pubkey).ecdsa_verify(htlc_sig, pre_hash):
raise LNProtocolWarning(
f'failed verifying HTLC signatures: {htlc=}, {htlc_direction=}. '
f'htlc_tx={htlc_tx.serialize()}. '
@@ -1185,7 +1186,7 @@ class Channel(AbstractChannel):
data = self.config[LOCAL].current_htlc_signatures
htlc_sigs = list(chunks(data, 64))
htlc_sig = htlc_sigs[htlc_relative_idx]
remote_htlc_sig = ecc.der_sig_from_sig_string(htlc_sig) + Sighash.to_sigbytes(Sighash.ALL)
remote_htlc_sig = ecc.ecdsa_der_sig_from_ecdsa_sig64(htlc_sig) + Sighash.to_sigbytes(Sighash.ALL)
return remote_htlc_sig
def revoke_current_commitment(self):
@@ -1599,7 +1600,7 @@ class Channel(AbstractChannel):
outputs=outputs)
der_sig = bfh(closing_tx.sign_txin(0, self.config[LOCAL].multisig_key.privkey))
sig = ecc.sig_string_from_der_sig(der_sig[:-1])
sig = ecc.ecdsa_sig64_from_der_sig(der_sig[:-1])
return sig, closing_tx
def signature_fits(self, tx: PartialTransaction) -> bool:
@@ -1607,7 +1608,7 @@ class Channel(AbstractChannel):
preimage_hex = tx.serialize_preimage(0)
msg_hash = sha256d(bfh(preimage_hex))
assert remote_sig
res = ecc.verify_signature(self.config[REMOTE].multisig_key.pubkey, remote_sig, msg_hash)
res = ECPubkey(self.config[REMOTE].multisig_key.pubkey).ecdsa_verify(remote_sig, msg_hash)
return res
def force_close_tx(self) -> PartialTransaction:
@@ -1615,7 +1616,7 @@ class Channel(AbstractChannel):
assert self.signature_fits(tx)
tx.sign({self.config[LOCAL].multisig_key.pubkey.hex(): (self.config[LOCAL].multisig_key.privkey, True)})
remote_sig = self.config[LOCAL].current_commitment_signature
remote_sig = ecc.der_sig_from_sig_string(remote_sig) + Sighash.to_sigbytes(Sighash.ALL)
remote_sig = ecc.ecdsa_der_sig_from_ecdsa_sig64(remote_sig) + Sighash.to_sigbytes(Sighash.ALL)
tx.add_signature_to_txin(txin_idx=0,
signing_pubkey=self.config[REMOTE].multisig_key.pubkey.hex(),
sig=remote_sig.hex())
electrum/lnpeer.py
+10 -10
View File
@@ -19,7 +19,7 @@ from aiorpcx import ignore_after
from .crypto import sha256, sha256d
from . import bitcoin, util
from . import ecc
from .ecc import sig_string_from_r_and_s, der_sig_from_sig_string
from .ecc import ecdsa_sig64_from_r_and_s, ecdsa_der_sig_from_ecdsa_sig64, ECPubkey
from . import constants
from .util import (bfh, log_exceptions, ignore_exceptions, chunks, OldTaskGroup,
UnrelatedTransactionException, error_text_bytes_to_safe_str)
@@ -424,9 +424,9 @@ class Peer(Logger):
h = chan.get_channel_announcement_hash()
node_signature = payload["node_signature"]
bitcoin_signature = payload["bitcoin_signature"]
if not ecc.verify_signature(chan.config[REMOTE].multisig_key.pubkey, bitcoin_signature, h):
if not ECPubkey(chan.config[REMOTE].multisig_key.pubkey).ecdsa_verify(bitcoin_signature, h):
raise Exception("bitcoin_sig invalid in announcement_signatures")
if not ecc.verify_signature(self.pubkey, node_signature, h):
if not ECPubkey(self.pubkey).ecdsa_verify(node_signature, h):
raise Exception("node_sig invalid in announcement_signatures")
chan.config[REMOTE].announcement_node_sig = node_signature
chan.config[REMOTE].announcement_bitcoin_sig = bitcoin_signature
@@ -1453,7 +1453,7 @@ class Peer(Logger):
addrlen=len(addresses),
addresses=addresses)
h = sha256d(raw_msg[64+2:])
signature = ecc.ECPrivkey(self.privkey).sign(h, sig_string_from_r_and_s)
signature = ecc.ECPrivkey(self.privkey).ecdsa_sign(h, sigencode=ecdsa_sig64_from_r_and_s)
message_type, payload = decode_msg(raw_msg)
payload['signature'] = signature
raw_msg = encode_msg(message_type, **payload)
@@ -1516,8 +1516,8 @@ class Peer(Logger):
if not is_reply and chan.config[REMOTE].announcement_node_sig:
return
h = chan.get_channel_announcement_hash()
bitcoin_signature = ecc.ECPrivkey(chan.config[LOCAL].multisig_key.privkey).sign(h, sig_string_from_r_and_s)
node_signature = ecc.ECPrivkey(self.privkey).sign(h, sig_string_from_r_and_s)
bitcoin_signature = ecc.ECPrivkey(chan.config[LOCAL].multisig_key.privkey).ecdsa_sign(h, sigencode=ecdsa_sig64_from_r_and_s)
node_signature = ecc.ECPrivkey(self.privkey).ecdsa_sign(h, sigencode=ecdsa_sig64_from_r_and_s)
self.send_message(
"announcement_signatures",
channel_id=chan.channel_id,
@@ -2446,11 +2446,11 @@ class Peer(Logger):
closing_signed_tlvs=closing_signed_tlvs,
)
def verify_signature(tx, sig):
def verify_signature(tx: 'PartialTransaction', sig) -> bool:
their_pubkey = chan.config[REMOTE].multisig_key.pubkey
preimage_hex = tx.serialize_preimage(0)
pre_hash = sha256d(bfh(preimage_hex))
return ecc.verify_signature(their_pubkey, sig, pre_hash)
return ECPubkey(their_pubkey).ecdsa_verify(sig, pre_hash)
async def receive_closing_signed():
nonlocal our_sig, closing_tx
@@ -2571,11 +2571,11 @@ class Peer(Logger):
closing_tx.add_signature_to_txin(
txin_idx=0,
signing_pubkey=chan.config[LOCAL].multisig_key.pubkey.hex(),
sig=(der_sig_from_sig_string(our_sig) + Sighash.to_sigbytes(Sighash.ALL)).hex())
sig=(ecdsa_der_sig_from_ecdsa_sig64(our_sig) + Sighash.to_sigbytes(Sighash.ALL)).hex())
closing_tx.add_signature_to_txin(
txin_idx=0,
signing_pubkey=chan.config[REMOTE].multisig_key.pubkey.hex(),
sig=(der_sig_from_sig_string(their_sig) + Sighash.to_sigbytes(Sighash.ALL)).hex())
sig=(ecdsa_der_sig_from_ecdsa_sig64(their_sig) + Sighash.to_sigbytes(Sighash.ALL)).hex())
# save local transaction and set state
try:
self.lnworker.wallet.adb.add_transaction(closing_tx)
electrum/lnutil.py
+2 -2
View File
@@ -21,7 +21,7 @@ from .util import format_short_id as format_short_channel_id
from .crypto import sha256, pw_decode_with_version_and_mac
from .transaction import (Transaction, PartialTransaction, PartialTxInput, TxOutpoint,
PartialTxOutput, opcodes, TxOutput)
from .ecc import CURVE_ORDER, sig_string_from_der_sig, ECPubkey, string_to_number
from .ecc import CURVE_ORDER, ecdsa_sig64_from_der_sig, ECPubkey, string_to_number
from . import ecc, bitcoin, crypto, transaction
from . import descriptor
from .bitcoin import (push_script, redeem_script_to_address, address_to_script,
@@ -1084,7 +1084,7 @@ def make_commitment_output_to_remote_address(remote_payment_pubkey: bytes) -> st
def sign_and_get_sig_string(tx: PartialTransaction, local_config, remote_config):
tx.sign({local_config.multisig_key.pubkey.hex(): (local_config.multisig_key.privkey, True)})
sig = tx.inputs()[0].part_sigs[local_config.multisig_key.pubkey]
sig_64 = sig_string_from_der_sig(sig[:-1])
sig_64 = ecdsa_sig64_from_der_sig(sig[:-1])
return sig_64
def funding_output_script(local_config, remote_config) -> str:
electrum/lnverifier.py
+2 -1
View File
@@ -31,6 +31,7 @@ import aiorpcx
from . import bitcoin
from . import ecc
from .ecc import ECPubkey
from . import constants
from .util import bfh, NetworkJobOnDefaultServer
from .lnutil import funding_output_script_from_keys, ShortChannelID
@@ -183,6 +184,6 @@ def verify_sig_for_channel_update(chan_upd: dict, node_id: bytes) -> bool:
pre_hash = msg_bytes[2+64:]
h = sha256d(pre_hash)
sig = chan_upd['signature']
if not ecc.verify_signature(node_id, sig, h):
if not ECPubkey(node_id).ecdsa_verify(sig, h):
return False
return True
electrum/lnworker.py
+1 -1
View File
@@ -51,7 +51,7 @@ from .logging import Logger
from .lntransport import LNTransport, LNResponderTransport, LNTransportBase
from .lnpeer import Peer, LN_P2P_NETWORK_TIMEOUT
from .lnaddr import lnencode, LnAddr, lndecode
from .ecc import der_sig_from_sig_string
from .ecc import ecdsa_der_sig_from_ecdsa_sig64
from .lnchannel import Channel, AbstractChannel
from .lnchannel import ChannelState, PeerState, HTLCWithStatus
from .lnrater import LNRater
electrum/paymentrequest.py
+2 -2
View File
@@ -235,7 +235,7 @@ class PaymentRequest:
address = info.get('address')
pr.signature = b''
message = pr.SerializeToString()
if ecc.verify_message_with_address(address, sig, message):
if ecc.verify_usermessage_with_address(address, sig, message):
self._verified_success_msg = 'Verified with DNSSEC'
self._verified_success = True
return True
@@ -356,7 +356,7 @@ def sign_request_with_alias(pr, alias, alias_privkey):
message = pr.SerializeToString()
ec_key = ecc.ECPrivkey(alias_privkey)
compressed = bitcoin.is_compressed_privkey(alias_privkey)
pr.signature = ec_key.sign_message(message, compressed)
pr.signature = ec_key.ecdsa_sign_usermessage(message, is_compressed=compressed)
def verify_cert_chain(chain):
electrum/plugins/bitbox02/bitbox02.py
+1 -1
View File
@@ -537,7 +537,7 @@ class BitBox02Client(HardwareClientBase):
if len(sigs) != len(tx.inputs()):
raise Exception("Incorrect number of inputs signed.") # Should never occur
sighash = Sighash.to_sigbytes(Sighash.ALL).hex()
signatures = [ecc.der_sig_from_sig_string(x[1]).hex() + sighash for x in sigs]
signatures = [ecc.ecdsa_der_sig_from_ecdsa_sig64(x[1]).hex() + sighash for x in sigs]
tx.update_signatures(signatures)
def sign_message(self, keypath: str, message: bytes, script_type: str) -> bytes:
electrum/plugins/coldcard/coldcard.py
+1 -1
View File
@@ -45,7 +45,7 @@ try:
# verify a signature (65 bytes) over the session key, using the master bip32 node
# - customized to use specific EC library of Electrum.
pubkey = BIP32Node.from_xkey(expect_xpub).eckey
return pubkey.verify_message_hash(sig[1:65], self.session_key)
return pubkey.ecdsa_verify(sig[1:65], self.session_key)
except ImportError as e:
if not (isinstance(e, ModuleNotFoundError) and e.name == 'ckcc'):
electrum/plugins/digitalbitbox/digitalbitbox.py
+9 -9
View File
@@ -23,7 +23,7 @@ from electrum.bip32 import BIP32Node, convert_bip32_intpath_to_strpath, is_all_p
from electrum.bip32 import normalize_bip32_derivation
from electrum import descriptor
from electrum import ecc
from electrum.ecc import msg_magic
from electrum.ecc import usermessage_magic
from electrum.wallet import Standard_Wallet
from electrum import constants
from electrum.transaction import Transaction, PartialTransaction, PartialTxInput, Sighash
@@ -470,7 +470,7 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
message = message.encode('utf8')
inputPath = self.get_derivation_prefix() + "/%d/%d" % sequence
inputPath = normalize_bip32_derivation(inputPath, hardened_char="'")
msg_hash = sha256d(msg_magic(message))
msg_hash = sha256d(usermessage_magic(message))
inputHash = to_hexstr(msg_hash)
hasharray = []
hasharray.append({'hash': inputHash, 'keypath': inputPath})
@@ -500,19 +500,19 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
# firmware > v2.1.1
sig_string = binascii.unhexlify(reply['sign'][0]['sig'])
recid = int(reply['sign'][0]['recid'], 16)
sig = ecc.construct_sig65(sig_string, recid, True)
pubkey, compressed, txin_type_guess = ecc.ECPubkey.from_signature65(sig, msg_hash)
sig = ecc.construct_ecdsa_sig65(sig_string, recid, is_compressed=True)
pubkey, compressed, txin_type_guess = ecc.ECPubkey.from_ecdsa_sig65(sig, msg_hash)
addr = public_key_to_p2pkh(pubkey.get_public_key_bytes(compressed=compressed))
if ecc.verify_message_with_address(addr, sig, message) is False:
if ecc.verify_usermessage_with_address(addr, sig, message) is False:
raise Exception(_("Could not sign message"))
elif 'pubkey' in reply['sign'][0]:
# firmware <= v2.1.1
for recid in range(4):
sig_string = binascii.unhexlify(reply['sign'][0]['sig'])
sig = ecc.construct_sig65(sig_string, recid, True)
sig = ecc.construct_ecdsa_sig65(sig_string, recid, is_compressed=True)
try:
addr = public_key_to_p2pkh(binascii.unhexlify(reply['sign'][0]['pubkey']))
if ecc.verify_message_with_address(addr, sig, message):
if ecc.verify_usermessage_with_address(addr, sig, message):
break
except Exception:
continue
@@ -649,7 +649,7 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
recid = int(signed['recid'], 16)
s = binascii.unhexlify(signed['sig'])
h = inputhasharray[i]
pk = ecc.ECPubkey.from_sig_string(s, recid, h)
pk = ecc.ECPubkey.from_ecdsa_sig64(s, recid, h)
pk = pk.get_public_key_hex(compressed=True)
elif 'pubkey' in signed:
# firmware <= v2.1.1
@@ -658,7 +658,7 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
continue
sig_r = int(signed['sig'][:64], 16)
sig_s = int(signed['sig'][64:], 16)
sig = ecc.der_sig_from_r_and_s(sig_r, sig_s)
sig = ecc.ecdsa_der_sig_from_r_and_s(sig_r, sig_s)
sig = to_hexstr(sig) + Sighash.to_sigbytes(Sighash.ALL).hex()
tx.add_signature_to_txin(txin_idx=i, signing_pubkey=pubkey_bytes.hex(), sig=sig)
except UserCancelled:
electrum/plugins/trustedcoin/common_qt.py
+1 -1
View File
@@ -208,7 +208,7 @@ class TrustedcoinPluginQObject(PluginQObject):
def f(xprv):
rootnode = BIP32Node.from_xkey(xprv)
key = rootnode.subkey_at_private_derivation((0, 0)).eckey
sig = key.sign_message(message, True)
sig = key.ecdsa_sign_usermessage(message, is_compressed=True)
return base64.b64encode(sig).decode()
signatures = [f(x) for x in [xprv1, xprv2]]
electrum/storage.py
+2 -2
View File
@@ -190,7 +190,7 @@ class WalletStorage(Logger):
ec_key = self.get_eckey_from_password(password)
if self.raw:
enc_magic = self._get_encryption_magic()
s = zlib.decompress(ec_key.decrypt_message(self.raw, enc_magic))
s = zlib.decompress(ec_key.decrypt_message(self.raw, magic=enc_magic))
s = s.decode('utf8')
else:
s = ''
@@ -205,7 +205,7 @@ class WalletStorage(Logger):
c = zlib.compress(s, level=zlib.Z_BEST_SPEED)
enc_magic = self._get_encryption_magic()
public_key = ecc.ECPubkey(bfh(self.pubkey))
s = public_key.encrypt_message(c, enc_magic)
s = public_key.encrypt_message(c, magic=enc_magic)
s = s.decode('utf8')
return s
electrum/transaction.py
+4 -4
View File
@@ -2153,7 +2153,7 @@ class PartialTransaction(Transaction):
pre_hash = sha256d(bfh(self.serialize_preimage(txin_index,
bip143_shared_txdigest_fields=bip143_shared_txdigest_fields)))
privkey = ecc.ECPrivkey(privkey_bytes)
sig = privkey.sign_transaction(pre_hash)
sig = privkey.ecdsa_sign(pre_hash, sigencode=ecc.ecdsa_der_sig_from_r_and_s)
sig = sig.hex() + Sighash.to_sigbytes(sighash).hex()
return sig
@@ -2208,16 +2208,16 @@ class PartialTransaction(Transaction):
if bfh(sig) in list(txin.part_sigs.values()):
continue
pre_hash = sha256d(bfh(self.serialize_preimage(i)))
sig_string = ecc.sig_string_from_der_sig(bfh(sig[:-2]))
sig_string = ecc.ecdsa_sig64_from_der_sig(bfh(sig[:-2]))
for recid in range(4):
try:
public_key = ecc.ECPubkey.from_sig_string(sig_string, recid, pre_hash)
public_key = ecc.ECPubkey.from_ecdsa_sig64(sig_string, recid, pre_hash)
except ecc.InvalidECPointException:
# the point might not be on the curve for some recid values
continue
pubkey_hex = public_key.get_public_key_hex(compressed=True)
if pubkey_hex in pubkeys:
if not public_key.verify_message_hash(sig_string, pre_hash):
if not public_key.ecdsa_verify(sig_string, pre_hash):
continue
_logger.info(f"adding sig: txin_idx={i}, signing_pubkey={pubkey_hex}, sig={sig}")
self.add_signature_to_txin(txin_idx=i, signing_pubkey=pubkey_hex, sig=sig)
tests/test_bitcoin.py
+23 -20
View File
@@ -163,7 +163,7 @@ class Test_bitcoin(ElectrumTestCase):
for message in [b"Chancellor on brink of second bailout for banks", b'\xff'*512]:
self._do_test_crypto(message)
def _do_test_crypto(self, message):
def _do_test_crypto(self, message: bytes):
G = ecc.GENERATOR
_r = G.order()
pvk = randrange(_r)
@@ -186,9 +186,9 @@ class Test_bitcoin(ElectrumTestCase):
dec2 = eck.decrypt_message(enc)
self.assertEqual(message, dec2)
signature = eck.sign_message(message, True)
#print signature
self.assertTrue(eck.verify_message_for_address(signature, message))
msg32 = sha256d(ecc.usermessage_magic(message))
sig65 = eck.ecdsa_sign_recoverable(msg32, is_compressed=True)
self.assertTrue(eck.ecdsa_verify_recoverable(sig65, msg32))
def test_ecc_sanity(self):
G = ecc.GENERATOR
@@ -221,7 +221,7 @@ class Test_bitcoin(ElectrumTestCase):
def sign_message_with_wif_privkey(wif_privkey: str, msg: bytes) -> bytes:
txin_type, privkey, compressed = deserialize_privkey(wif_privkey)
key = ecc.ECPrivkey(privkey)
return key.sign_message(msg, compressed)
return key.ecdsa_sign_usermessage(msg, is_compressed=compressed)
def test_signmessage_legacy_address(self):
msg1 = b'Chancellor on brink of second bailout for banks'
@@ -240,11 +240,11 @@ class Test_bitcoin(ElectrumTestCase):
self.assertEqual(sig1_b64, b'Hzsu0U/THAsPz/MSuXGBKSULz2dTfmrg1NsAhFp+wH5aKfmX4Db7ExLGa7FGn0m6Mf43KsbEOWpvUUUBTM3Uusw=')
self.assertEqual(sig2_b64, b'HBQdYfv7kOrxmRewLJnG7sV6KlU71O04hUnE4tai97p7Pg+D+yKaWXsdGgHTrKw90caQMo/D6b//qX50ge9P9iI=')
self.assertTrue(ecc.verify_message_with_address(addr1, sig1, msg1))
self.assertTrue(ecc.verify_message_with_address(addr2, sig2, msg2))
self.assertTrue(ecc.verify_usermessage_with_address(addr1, sig1, msg1))
self.assertTrue(ecc.verify_usermessage_with_address(addr2, sig2, msg2))
self.assertFalse(ecc.verify_message_with_address(addr1, b'wrong', msg1))
self.assertFalse(ecc.verify_message_with_address(addr1, sig2, msg1))
self.assertFalse(ecc.verify_usermessage_with_address(addr1, b'wrong', msg1))
self.assertFalse(ecc.verify_usermessage_with_address(addr1, sig2, msg1))
def test_signmessage_segwit_witness_v0_address(self):
msg = b'Electrum'
@@ -252,14 +252,14 @@ class Test_bitcoin(ElectrumTestCase):
sig1 = self.sign_message_with_wif_privkey("p2wpkh-p2sh:L1cgMEnShp73r9iCukoPE3MogLeueNYRD9JVsfT1zVHyPBR3KqBY", msg)
addr1 = "3DYoBqQ5N6dADzyQjy9FT1Ls4amiYVaqTG"
self.assertEqual(base64.b64encode(sig1), b'HyFaND+87TtVbRhkTfT3mPNBCQcJ32XXtNZGW8sFldJsNpOPCegEmdcCf5Thy18hdMH88GLxZLkOby/EwVUuSeA=')
self.assertTrue(ecc.verify_message_with_address(addr1, sig1, msg))
self.assertFalse(ecc.verify_message_with_address(addr1, sig1, b'heyheyhey'))
self.assertTrue(ecc.verify_usermessage_with_address(addr1, sig1, msg))
self.assertFalse(ecc.verify_usermessage_with_address(addr1, sig1, b'heyheyhey'))
# p2wpkh
sig2 = self.sign_message_with_wif_privkey("p2wpkh:L1cgMEnShp73r9iCukoPE3MogLeueNYRD9JVsfT1zVHyPBR3KqBY", msg)
addr2 = "bc1qq2tmmcngng78nllq2pvrkchcdukemtj56uyue0"
self.assertEqual(base64.b64encode(sig2), b'HyFaND+87TtVbRhkTfT3mPNBCQcJ32XXtNZGW8sFldJsNpOPCegEmdcCf5Thy18hdMH88GLxZLkOby/EwVUuSeA=')
self.assertTrue(ecc.verify_message_with_address(addr2, sig2, msg))
self.assertFalse(ecc.verify_message_with_address(addr2, sig2, b'heyheyhey'))
self.assertTrue(ecc.verify_usermessage_with_address(addr2, sig2, msg))
self.assertFalse(ecc.verify_usermessage_with_address(addr2, sig2, b'heyheyhey'))
def test_signmessage_segwit_witness_v0_address_test_we_also_accept_sigs_from_trezor(self):
"""Trezor and some other projects use a slightly different scheme for message-signing
@@ -272,13 +272,13 @@ class Test_bitcoin(ElectrumTestCase):
addr2 = "bc1qannfxke2tfd4l7vhepehpvt05y83v3qsf6nfkk"
sig1 = bytes.fromhex("23744de4516fac5c140808015664516a32fead94de89775cec7e24dbc24fe133075ac09301c4cc8e197bea4b6481661d5b8e9bf19d8b7b8a382ecdb53c2ee0750d")
sig2 = bytes.fromhex("28b55d7600d9e9a7e2a49155ddf3cfdb8e796c207faab833010fa41fb7828889bc47cf62348a7aaa0923c0832a589fab541e8f12eb54fb711c90e2307f0f66b194")
self.assertTrue(ecc.verify_message_with_address(address=addr1, sig65=sig1, message=msg))
self.assertTrue(ecc.verify_message_with_address(address=addr2, sig65=sig2, message=msg))
self.assertTrue(ecc.verify_usermessage_with_address(address=addr1, sig65=sig1, message=msg))
self.assertTrue(ecc.verify_usermessage_with_address(address=addr2, sig65=sig2, message=msg))
# if there is type information in the header of the sig (first byte), enforce that:
sig1_wrongtype = bytes.fromhex("27744de4516fac5c140808015664516a32fead94de89775cec7e24dbc24fe133075ac09301c4cc8e197bea4b6481661d5b8e9bf19d8b7b8a382ecdb53c2ee0750d")
sig2_wrongtype = bytes.fromhex("24b55d7600d9e9a7e2a49155ddf3cfdb8e796c207faab833010fa41fb7828889bc47cf62348a7aaa0923c0832a589fab541e8f12eb54fb711c90e2307f0f66b194")
self.assertFalse(ecc.verify_message_with_address(address=addr1, sig65=sig1_wrongtype, message=msg))
self.assertFalse(ecc.verify_message_with_address(address=addr2, sig65=sig2_wrongtype, message=msg))
self.assertFalse(ecc.verify_usermessage_with_address(address=addr1, sig65=sig1_wrongtype, message=msg))
self.assertFalse(ecc.verify_usermessage_with_address(address=addr2, sig65=sig2_wrongtype, message=msg))
@needs_test_with_all_aes_implementations
def test_decrypt_message(self):
@@ -303,17 +303,20 @@ class Test_bitcoin(ElectrumTestCase):
def test_sign_transaction(self):
eckey1 = ecc.ECPrivkey(bfh('7e1255fddb52db1729fc3ceb21a46f95b8d9fe94cc83425e936a6c5223bb679d'))
sig1 = eckey1.sign_transaction(bfh('5a548b12369a53faaa7e51b5081829474ebdd9c924b3a8230b69aa0be254cd94'))
sig1 = eckey1.ecdsa_sign(bfh('5a548b12369a53faaa7e51b5081829474ebdd9c924b3a8230b69aa0be254cd94'),
sigencode=ecc.ecdsa_der_sig_from_r_and_s)
self.assertEqual('3044022066e7d6a954006cce78a223f5edece8aaedcf3607142e9677acef1cfcb91cfdde022065cb0b5401bf16959ce7b785ea7fd408be5e4cb7d8f1b1a32c78eac6f73678d9', sig1.hex())
eckey2 = ecc.ECPrivkey(bfh('c7ce8c1462c311eec24dff9e2532ac6241e50ae57e7d1833af21942136972f23'))
sig2 = eckey2.sign_transaction(bfh('642a2e66332f507c92bda910158dfe46fc10afbf72218764899d3af99a043fac'))
sig2 = eckey2.ecdsa_sign(bfh('642a2e66332f507c92bda910158dfe46fc10afbf72218764899d3af99a043fac'),
sigencode=ecc.ecdsa_der_sig_from_r_and_s)
self.assertEqual('30440220618513f4cfc87dde798ce5febae7634c23e7b9254a1eabf486be820f6a7c2c4702204fef459393a2b931f949e63ced06888f35e286e446dc46feb24b5b5f81c6ed52', sig2.hex())
@disable_ecdsa_r_value_grinding
def test_sign_transaction_without_ecdsa_r_value_grinding(self):
eckey1 = ecc.ECPrivkey(bfh('7e1255fddb52db1729fc3ceb21a46f95b8d9fe94cc83425e936a6c5223bb679d'))
sig1 = eckey1.sign_transaction(bfh('5a548b12369a53faaa7e51b5081829474ebdd9c924b3a8230b69aa0be254cd94'))
sig1 = eckey1.ecdsa_sign(bfh('5a548b12369a53faaa7e51b5081829474ebdd9c924b3a8230b69aa0be254cd94'),
sigencode=ecc.ecdsa_der_sig_from_r_and_s)
self.assertEqual('3045022100902a288b98392254cd23c0e9a49ac6d7920f171b8249a48e484b998f1874a2010220723d844826828f092cf400cb210c4fa0b8cd1b9d1a7f21590e78e022ff6476b9', sig1.hex())
@needs_test_with_all_aes_implementations
tests/test_ecc.py
+8 -8
View File
@@ -47,19 +47,19 @@ class TestSchnorr(ElectrumTestCase):
if seckey:
seckey = ECPrivkey(bytes.fromhex(seckey))
aux_rand = bytes.fromhex(aux_rand)
sig_created = seckey.sign_schnorr(msg32, aux_rand32=aux_rand)
sig_created = seckey.schnorr_sign(msg32, aux_rand32=aux_rand)
self.assertEqual(signature, sig_created)
is_sig_good = pubkey.verify_message_schnorr(signature, msg32)
is_sig_good = pubkey.schnorr_verify(signature, msg32)
expected_res = True if expected_res == "TRUE" else False
self.assertEqual(expected_res, is_sig_good)
def test_sign_schnorr_aux_rand(self):
seckey = ECPrivkey(bytes.fromhex("B7E151628AED2A6ABF7158809CF4F3C762E7160F38B4DA56A784D9045190CFEF"))
msg32 = sha256("hello there")
sig1 = seckey.sign_schnorr(msg32, aux_rand32=None)
sig2 = seckey.sign_schnorr(msg32, aux_rand32=b"\x00" * 32)
sig1 = seckey.schnorr_sign(msg32, aux_rand32=None)
sig2 = seckey.schnorr_sign(msg32, aux_rand32=b"\x00" * 32)
self.assertEqual(sig1, sig2)
sig3 = seckey.sign_schnorr(msg32, aux_rand32=bytes(range(32)))
sig3 = seckey.schnorr_sign(msg32, aux_rand32=bytes(range(32)))
self.assertNotEqual(sig1, sig3)
def test_y_parity_malleability(self):
@@ -79,6 +79,6 @@ class TestSchnorr(ElectrumTestCase):
self.assertNotEqual(pubkey1.get_public_key_bytes(True), pubkey2.get_public_key_bytes(True))
self.assertEqual(pubkey1.get_public_key_bytes(True)[1:], pubkey2.get_public_key_bytes(True)[1:])
msg32 = sha256("hello there")
sig = seckey.sign_schnorr(msg32, aux_rand32=None)
self.assertTrue(pubkey1.verify_message_schnorr(sig, msg32))
self.assertTrue(pubkey2.verify_message_schnorr(sig, msg32))
sig = seckey.schnorr_sign(msg32, aux_rand32=None)
self.assertTrue(pubkey1.schnorr_verify(sig, msg32))
self.assertTrue(pubkey2.schnorr_verify(sig, msg32))