LBRYFoundation/LBRY-Vault
1
0
Fork 0
mirror of https://github.com/LBRYFoundation/LBRY-Vault.git synced 2025-08-29 16:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #4381 from SomberNight/coincurve4

Browse source 
crypto refactoring take3
This commit is contained in:
ThomasV 2018-05-25 18:49:36 +02:00 • committed by GitHub
commit 770f438249
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
33 changed files with 1117 additions and 539 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.travis.yml
View file

@ -3,6 +3,12 @@ language: python
python:
- 3.5
- 3.6
addons:
apt:
sources:
- sourceline: 'ppa:tah83/secp256k1'
packages:
- libsecp256k1-0
install:
- pip install -r contrib/requirements/requirements-travis.txt
cache:
@ -27,7 +33,7 @@ jobs:
- sudo apt-key add Release.key
- sudo apt-add-repository https://dl.winehq.org/wine-builds/ubuntu/
- sudo apt-get update -qq
- sudo apt-get install -qq winehq-stable dirmngr gnupg2 p7zip-full
- sudo apt-get install -qq winehq-stable dirmngr gnupg2 p7zip-full mingw-w64
before_script: ls -lah /tmp/electrum-build
script: ./contrib/build-wine/build.sh
after_success: true

3
contrib/build-wine/README.md
View file

@ -14,13 +14,14 @@ Usage:
- gpg
- 7Zip
- Wine (>= v2)
- mingw-w64
For example:
```
$ sudo apt-get install wine-development dirmngr gnupg2 p7zip-full
$ sudo apt-get install wine-development dirmngr gnupg2 p7zip-full mingw-w64
$ wine --version
wine-2.0 (Debian 2.0-3+b2)
```

33
contrib/build-wine/build-secp256k1.sh Executable file
View file

@ -0,0 +1,33 @@
#!/bin/bash
# heavily based on https://github.com/ofek/coincurve/blob/417e726f553460f88d7edfa5dc67bfda397c4e4a/.travis/build_windows_wheels.sh
set -e
build_dll() {
#sudo apt-get install -y mingw-w64
./autogen.sh
echo "LDFLAGS = -no-undefined" >> Makefile.am
./configure --host=$1 --enable-module-recovery --enable-experimental --enable-module-ecdh --disable-jni
make
}
cd /tmp/electrum-build
if [ ! -d secp256k1 ]; then
git clone https://github.com/bitcoin-core/secp256k1.git
cd secp256k1;
else
cd secp256k1
git pull
fi
git reset --hard 452d8e4d2a2f9f1b5be6b02e18f1ba102e5ca0b4
git clean -f -x -q
build_dll i686-w64-mingw32 # 64-bit would be: x86_64-w64-mingw32
mv .libs/libsecp256k1-0.dll libsecp256k1.dll
find -exec touch -d '2000-11-11T11:11:11+00:00' {} +
echo "building libsecp256k1 finished"

2
contrib/build-wine/build.sh
View file

@ -17,6 +17,8 @@ mkdir -p /tmp/electrum-build
mkdir -p /tmp/electrum-build/pip-cache
export PIP_CACHE_DIR="/tmp/electrum-build/pip-cache"
$here/build-secp256k1.sh || exit 1
$here/prepare-wine.sh || exit 1
echo "Resetting modification time in C:\Python..."

2
contrib/build-wine/deterministic.spec
View file

@ -28,6 +28,8 @@ binaries = [(PYHOME+"/libusb-1.0.dll", ".")]
# Workaround for "Retro Look":
binaries += [b for b in collect_dynamic_libs('PyQt5') if 'qwindowsvista' in b[0]]
binaries += [('C:/tmp/libsecp256k1.dll', '.')]
datas = [
(home+'lib/currencies.json', 'electrum'),
(home+'lib/servers.json', 'electrum'),

3
contrib/build-wine/prepare-wine.sh
View file

@ -139,4 +139,7 @@ cp libusb/MS32/dll/libusb-1.0.dll $WINEPREFIX/drive_c/python$PYTHON_VERSION/
# add dlls needed for pyinstaller:
cp $WINEPREFIX/drive_c/python$PYTHON_VERSION/Lib/site-packages/PyQt5/Qt/bin/* $WINEPREFIX/drive_c/python$PYTHON_VERSION/
mkdir -p $WINEPREFIX/drive_c/tmp
cp secp256k1/libsecp256k1.dll $WINEPREFIX/drive_c/tmp/
echo "Wine is configured."

7
gui/qt/main_window.py
View file

@ -39,7 +39,7 @@ import PyQt5.QtCore as QtCore
from .exception_window import Exception_Hook
from PyQt5.QtWidgets import *
from electrum import keystore, simple_config
from electrum import keystore, simple_config, ecc
from electrum.bitcoin import COIN, is_address, TYPE_ADDRESS
from electrum import constants
from electrum.plugins import run_hook
@ -2177,7 +2177,7 @@ class ElectrumWindow(QMainWindow, MessageBoxMixin, PrintError):
try:
# This can throw on invalid base64
sig = base64.b64decode(str(signature.toPlainText()))
verified = bitcoin.verify_message(address, sig, message)
verified = ecc.verify_message_with_address(address, sig, message)
except Exception as e:
verified = False
if verified:
@ -2243,7 +2243,8 @@ class ElectrumWindow(QMainWindow, MessageBoxMixin, PrintError):
message = message_e.toPlainText()
message = message.encode('utf-8')
try:
encrypted = bitcoin.encrypt_message(message, pubkey_e.text())
public_key = ecc.ECPubkey(bfh(pubkey_e.text()))
encrypted = public_key.encrypt_message(message)
encrypted_e.setText(encrypted.decode('ascii'))
except BaseException as e:
traceback.print_exc(file=sys.stdout)

442
lib/bitcoin.py
View file

@ -24,19 +24,14 @@
# SOFTWARE.
import hashlib
import base64
import hmac
import os
import json
import ecdsa
import pyaes
from .util import bfh, bh2u, to_string, BitcoinException
from .util import bfh, bh2u, BitcoinException, print_error, assert_bytes, to_bytes, inv_dict
from . import version
from .util import print_error, InvalidPassword, assert_bytes, to_bytes, inv_dict
from . import segwit_addr
from . import constants
from . import ecc
from .crypto import Hash, sha256, hash_160
################################## transactions
@ -49,94 +44,6 @@ TYPE_ADDRESS = 0
TYPE_PUBKEY = 1
TYPE_SCRIPT = 2
# AES encryption
try:
from Cryptodome.Cipher import AES
except:
AES = None
class InvalidPadding(Exception):
pass
def append_PKCS7_padding(data):
assert_bytes(data)
padlen = 16 - (len(data) % 16)
return data + bytes([padlen]) * padlen
def strip_PKCS7_padding(data):
assert_bytes(data)
if len(data) % 16 != 0 or len(data) == 0:
raise InvalidPadding("invalid length")
padlen = data[-1]
if padlen > 16:
raise InvalidPadding("invalid padding byte (large)")
for i in data[-padlen:]:
if i != padlen:
raise InvalidPadding("invalid padding byte (inconsistent)")
return data[0:-padlen]
def aes_encrypt_with_iv(key, iv, data):
assert_bytes(key, iv, data)
data = append_PKCS7_padding(data)
if AES:
e = AES.new(key, AES.MODE_CBC, iv).encrypt(data)
else:
aes_cbc = pyaes.AESModeOfOperationCBC(key, iv=iv)
aes = pyaes.Encrypter(aes_cbc, padding=pyaes.PADDING_NONE)
e = aes.feed(data) + aes.feed() # empty aes.feed() flushes buffer
return e
def aes_decrypt_with_iv(key, iv, data):
assert_bytes(key, iv, data)
if AES:
cipher = AES.new(key, AES.MODE_CBC, iv)
data = cipher.decrypt(data)
else:
aes_cbc = pyaes.AESModeOfOperationCBC(key, iv=iv)
aes = pyaes.Decrypter(aes_cbc, padding=pyaes.PADDING_NONE)
data = aes.feed(data) + aes.feed() # empty aes.feed() flushes buffer
try:
return strip_PKCS7_padding(data)
except InvalidPadding:
raise InvalidPassword()
def EncodeAES(secret, s):
assert_bytes(s)
iv = bytes(os.urandom(16))
ct = aes_encrypt_with_iv(secret, iv, s)
e = iv + ct
return base64.b64encode(e)
def DecodeAES(secret, e):
e = bytes(base64.b64decode(e))
iv, e = e[:16], e[16:]
s = aes_decrypt_with_iv(secret, iv, e)
return s
def pw_encode(s, password):
if password:
secret = Hash(password)
return EncodeAES(secret, to_bytes(s, "utf8")).decode('utf8')
else:
return s
def pw_decode(s, password):
if password is not None:
secret = Hash(password)
try:
d = to_string(DecodeAES(secret, s), "utf8")
except Exception:
raise InvalidPassword()
return d
else:
return s
def rev_hex(s):
return bh2u(bfh(s)[::-1])
@ -233,17 +140,6 @@ def add_number_to_script(i: int) -> bytes:
return bfh(push_script(script_num_to_hex(i)))
def sha256(x):
x = to_bytes(x, 'utf8')
return bytes(hashlib.sha256(x).digest())
def Hash(x):
x = to_bytes(x, 'utf8')
out = bytes(sha256(sha256(x)))
return out
hash_encode = lambda x: bh2u(x[::-1])
hash_decode = lambda x: bfh(x)[::-1]
hmac_sha_512 = lambda x, y: hmac.new(x, y, hashlib.sha512).digest()
@ -287,40 +183,10 @@ def seed_type(x):
is_seed = lambda x: bool(seed_type(x))
# pywallet openssl private key implementation
def i2o_ECPublicKey(pubkey, compressed=False):
# public keys are 65 bytes long (520 bits)
# 0x04 + 32-byte X-coordinate + 32-byte Y-coordinate
# 0x00 = point at infinity, 0x02 and 0x03 = compressed, 0x04 = uncompressed
# compressed keys: <sign> <x> where <sign> is 0x02 if y is even and 0x03 if y is odd
if compressed:
if pubkey.point.y() & 1:
key = '03' + '%064x' % pubkey.point.x()
else:
key = '02' + '%064x' % pubkey.point.x()
else:
key = '04' + \
'%064x' % pubkey.point.x() + \
'%064x' % pubkey.point.y()
return bfh(key)
# end pywallet openssl private key implementation
############ functions from pywallet #####################
def hash_160(public_key):
try:
md = hashlib.new('ripemd160')
md.update(sha256(public_key))
return md.digest()
except BaseException:
from . import ripemd
md = ripemd.new(sha256(public_key))
return md.digest()
def hash160_to_b58_address(h160, addrtype):
def hash160_to_b58_address(h160: bytes, addrtype):
s = bytes([addrtype])
s += h160
return base_encode(s+Hash(s)[0:4], base=58)
@ -342,7 +208,7 @@ def hash160_to_p2sh(h160, *, net=None):
net = constants.net
return hash160_to_b58_address(h160, net.ADDRTYPE_P2SH)
def public_key_to_p2pkh(public_key):
def public_key_to_p2pkh(public_key: bytes) -> str:
return hash160_to_p2pkh(hash_160(public_key))
def hash_to_segwit_addr(h, witver, *, net=None):
@ -437,7 +303,7 @@ __b43chars = b'0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ$*+-./:'
assert len(__b43chars) == 43
def base_encode(v, base):
def base_encode(v: bytes, base: int) -> str:
""" encode v, which is a string of bytes, to base58."""
assert_bytes(v)
if base not in (58, 43):
@ -535,7 +401,10 @@ SCRIPT_TYPES = {
}
def serialize_privkey(secret, compressed, txin_type, internal_use=False):
def serialize_privkey(secret: bytes, compressed: bool, txin_type: str,
internal_use: bool=False) -> str:
# we only export secrets inside curve range
secret = ecc.ECPrivkey.normalize_secret_bytes(secret)
if internal_use:
prefix = bytes([(SCRIPT_TYPES[txin_type] + constants.net.WIF_PREFIX) & 255])
else:
@ -549,7 +418,7 @@ def serialize_privkey(secret, compressed, txin_type, internal_use=False):
return '{}:{}'.format(txin_type, base58_wif)
def deserialize_privkey(key):
def deserialize_privkey(key: str) -> (str, bytes, bool):
if is_minikey(key):
return 'p2pkh', minikey_to_private_key(key), True
@ -581,34 +450,19 @@ def deserialize_privkey(key):
if len(vch) not in [33, 34]:
raise BitcoinException('invalid vch len for WIF key: {}'.format(len(vch)))
compressed = len(vch) == 34
return txin_type, vch[1:33], compressed
def regenerate_key(pk):
assert len(pk) == 32
return EC_KEY(pk)
def GetPubKey(pubkey, compressed=False):
return i2o_ECPublicKey(pubkey, compressed)
def GetSecret(pkey):
return bfh('%064x' % pkey.secret)
secret_bytes = vch[1:33]
# we accept secrets outside curve range; cast into range here:
secret_bytes = ecc.ECPrivkey.normalize_secret_bytes(secret_bytes)
return txin_type, secret_bytes, compressed
def is_compressed(sec):
return deserialize_privkey(sec)[2]
def public_key_from_private_key(pk, compressed):
pkey = regenerate_key(pk)
public_key = GetPubKey(pkey.pubkey, compressed)
return bh2u(public_key)
def address_from_private_key(sec):
txin_type, privkey, compressed = deserialize_privkey(sec)
public_key = public_key_from_private_key(privkey, compressed)
public_key = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)
return pubkey_to_address(txin_type, public_key)
def is_segwit_address(addr):
@ -654,242 +508,12 @@ def is_minikey(text):
def minikey_to_private_key(text):
return sha256(text)
from ecdsa.ecdsa import curve_secp256k1, generator_secp256k1
from ecdsa.curves import SECP256k1
from ecdsa.ellipticcurve import Point
from ecdsa.util import string_to_number, number_to_string
def msg_magic(message):
length = bfh(var_int(len(message)))
return b"\x18Bitcoin Signed Message:\n" + length + message
def verify_message(address, sig, message):
assert_bytes(sig, message)
try:
h = Hash(msg_magic(message))
public_key, compressed = pubkey_from_signature(sig, h)
# check public key using the address
pubkey = point_to_ser(public_key.pubkey.point, compressed)
for txin_type in ['p2pkh','p2wpkh','p2wpkh-p2sh']:
addr = pubkey_to_address(txin_type, bh2u(pubkey))
if address == addr:
break
else:
raise Exception("Bad signature")
# check message
public_key.verify_digest(sig[1:], h, sigdecode = ecdsa.util.sigdecode_string)
return True
except Exception as e:
print_error("Verification error: {0}".format(e))
return False
def encrypt_message(message, pubkey, magic=b'BIE1'):
return EC_KEY.encrypt_message(message, bfh(pubkey), magic)
def chunks(l, n):
return [l[i:i+n] for i in range(0, len(l), n)]
def ECC_YfromX(x,curved=curve_secp256k1, odd=True):
_p = curved.p()
_a = curved.a()
_b = curved.b()
for offset in range(128):
Mx = x + offset
My2 = pow(Mx, 3, _p) + _a * pow(Mx, 2, _p) + _b % _p
My = pow(My2, (_p+1)//4, _p )
if curved.contains_point(Mx,My):
if odd == bool(My&1):
return [My,offset]
return [_p-My,offset]
raise Exception('ECC_YfromX: No Y found')
def negative_point(P):
return Point( P.curve(), P.x(), -P.y(), P.order() )
def point_to_ser(P, comp=True ):
if comp:
return bfh( ('%02x'%(2+(P.y()&1)))+('%064x'%P.x()) )
return bfh( '04'+('%064x'%P.x())+('%064x'%P.y()) )
def ser_to_point(Aser):
curve = curve_secp256k1
generator = generator_secp256k1
_r = generator.order()
assert Aser[0] in [0x02, 0x03, 0x04]
if Aser[0] == 0x04:
return Point( curve, string_to_number(Aser[1:33]), string_to_number(Aser[33:]), _r )
Mx = string_to_number(Aser[1:])
return Point( curve, Mx, ECC_YfromX(Mx, curve, Aser[0] == 0x03)[0], _r )
class MyVerifyingKey(ecdsa.VerifyingKey):
@classmethod
def from_signature(klass, sig, recid, h, curve):
""" See http://www.secg.org/download/aid-780/sec1-v2.pdf, chapter 4.1.6 """
from ecdsa import util, numbertheory
from . import msqr
curveFp = curve.curve
G = curve.generator
order = G.order()
# extract r,s from signature
r, s = util.sigdecode_string(sig, order)
# 1.1
x = r + (recid//2) * order
# 1.3
alpha = ( x * x * x + curveFp.a() * x + curveFp.b() ) % curveFp.p()
beta = msqr.modular_sqrt(alpha, curveFp.p())
y = beta if (beta - recid) % 2 == 0 else curveFp.p() - beta
# 1.4 the constructor checks that nR is at infinity
R = Point(curveFp, x, y, order)
# 1.5 compute e from message:
e = string_to_number(h)
minus_e = -e % order
# 1.6 compute Q = r^-1 (sR - eG)
inv_r = numbertheory.inverse_mod(r,order)
Q = inv_r * ( s * R + minus_e * G )
return klass.from_public_point( Q, curve )
def pubkey_from_signature(sig, h):
if len(sig) != 65:
raise Exception("Wrong encoding")
nV = sig[0]
if nV < 27 or nV >= 35:
raise Exception("Bad encoding")
if nV >= 31:
compressed = True
nV -= 4
else:
compressed = False
recid = nV - 27
return MyVerifyingKey.from_signature(sig[1:], recid, h, curve = SECP256k1), compressed
class MySigningKey(ecdsa.SigningKey):
"""Enforce low S values in signatures"""
def sign_number(self, number, entropy=None, k=None):
curve = SECP256k1
G = curve.generator
order = G.order()
r, s = ecdsa.SigningKey.sign_number(self, number, entropy, k)
if s > order//2:
s = order - s
return r, s
class EC_KEY(object):
def __init__( self, k ):
secret = string_to_number(k)
self.pubkey = ecdsa.ecdsa.Public_key( generator_secp256k1, generator_secp256k1 * secret )
self.privkey = ecdsa.ecdsa.Private_key( self.pubkey, secret )
self.secret = secret
def get_public_key(self, compressed=True):
return bh2u(point_to_ser(self.pubkey.point, compressed))
def sign(self, msg_hash):
private_key = MySigningKey.from_secret_exponent(self.secret, curve = SECP256k1)
public_key = private_key.get_verifying_key()
signature = private_key.sign_digest_deterministic(msg_hash, hashfunc=hashlib.sha256, sigencode = ecdsa.util.sigencode_string)
assert public_key.verify_digest(signature, msg_hash, sigdecode = ecdsa.util.sigdecode_string)
return signature
def sign_message(self, message, is_compressed):
message = to_bytes(message, 'utf8')
signature = self.sign(Hash(msg_magic(message)))
for i in range(4):
sig = bytes([27 + i + (4 if is_compressed else 0)]) + signature
try:
self.verify_message(sig, message)
return sig
except Exception as e:
continue
else:
raise Exception("error: cannot sign message")
def verify_message(self, sig, message):
assert_bytes(message)
h = Hash(msg_magic(message))
public_key, compressed = pubkey_from_signature(sig, h)
# check public key
if point_to_ser(public_key.pubkey.point, compressed) != point_to_ser(self.pubkey.point, compressed):
raise Exception("Bad signature")
# check message
public_key.verify_digest(sig[1:], h, sigdecode = ecdsa.util.sigdecode_string)
# ECIES encryption/decryption methods; AES-128-CBC with PKCS7 is used as the cipher; hmac-sha256 is used as the mac
@classmethod
def encrypt_message(self, message, pubkey, magic=b'BIE1'):
assert_bytes(message)
pk = ser_to_point(pubkey)
if not ecdsa.ecdsa.point_is_valid(generator_secp256k1, pk.x(), pk.y()):
raise Exception('invalid pubkey')
ephemeral_exponent = number_to_string(ecdsa.util.randrange(pow(2,256)), generator_secp256k1.order())
ephemeral = EC_KEY(ephemeral_exponent)
ecdh_key = point_to_ser(pk * ephemeral.privkey.secret_multiplier)
key = hashlib.sha512(ecdh_key).digest()
iv, key_e, key_m = key[0:16], key[16:32], key[32:]
ciphertext = aes_encrypt_with_iv(key_e, iv, message)
ephemeral_pubkey = bfh(ephemeral.get_public_key(compressed=True))
encrypted = magic + ephemeral_pubkey + ciphertext
mac = hmac.new(key_m, encrypted, hashlib.sha256).digest()
return base64.b64encode(encrypted + mac)
def decrypt_message(self, encrypted, magic=b'BIE1'):
encrypted = base64.b64decode(encrypted)
if len(encrypted) < 85:
raise Exception('invalid ciphertext: length')
magic_found = encrypted[:4]
ephemeral_pubkey = encrypted[4:37]
ciphertext = encrypted[37:-32]
mac = encrypted[-32:]
if magic_found != magic:
raise Exception('invalid ciphertext: invalid magic bytes')
try:
ephemeral_pubkey = ser_to_point(ephemeral_pubkey)
except AssertionError as e:
raise Exception('invalid ciphertext: invalid ephemeral pubkey')
if not ecdsa.ecdsa.point_is_valid(generator_secp256k1, ephemeral_pubkey.x(), ephemeral_pubkey.y()):
raise Exception('invalid ciphertext: invalid ephemeral pubkey')
ecdh_key = point_to_ser(ephemeral_pubkey * self.privkey.secret_multiplier)
key = hashlib.sha512(ecdh_key).digest()
iv, key_e, key_m = key[0:16], key[16:32], key[32:]
if mac != hmac.new(key_m, encrypted[:-32], hashlib.sha256).digest():
raise InvalidPassword()
return aes_decrypt_with_iv(key_e, iv, ciphertext)
###################################### BIP32 ##############################
random_seed = lambda n: "%032x"%ecdsa.util.randrange( pow(2,n) )
BIP32_PRIME = 0x80000000
def get_pubkeys_from_secret(secret):
# public key
private_key = ecdsa.SigningKey.from_string( secret, curve = SECP256k1 )
public_key = private_key.get_verifying_key()
K = public_key.to_string()
K_compressed = GetPubKey(public_key.pubkey,True)
return K, K_compressed
# Child private key derivation function (from master private key)
# k = master private key (32 bytes)
# c = master chain code (extra entropy for key derivation) (32 bytes)
@ -904,12 +528,13 @@ def CKD_priv(k, c, n):
def _CKD_priv(k, c, s, is_prime):
order = generator_secp256k1.order()
keypair = EC_KEY(k)
cK = GetPubKey(keypair.pubkey,True)
keypair = ecc.ECPrivkey(k)
cK = keypair.get_public_key_bytes(compressed=True)
data = bytes([0]) + k + s if is_prime else cK + s
I = hmac.new(c, data, hashlib.sha512).digest()
k_n = number_to_string( (string_to_number(I[0:32]) + string_to_number(k)) % order , order )
k_n = ecc.number_to_string(
(ecc.string_to_number(I[0:32]) + ecc.string_to_number(k)) % ecc.CURVE_ORDER,
ecc.CURVE_ORDER)
c_n = I[32:]
return k_n, c_n
@ -920,18 +545,15 @@ def _CKD_priv(k, c, s, is_prime):
# This function allows us to find the nth public key, as long as n is
# non-negative. If n is negative, we need the master private key to find it.
def CKD_pub(cK, c, n):
if n & BIP32_PRIME: raise
if n & BIP32_PRIME: raise Exception()
return _CKD_pub(cK, c, bfh(rev_hex(int_to_hex(n,4))))
# helper function, callable with arbitrary string
def _CKD_pub(cK, c, s):
order = generator_secp256k1.order()
I = hmac.new(c, cK + s, hashlib.sha512).digest()
curve = SECP256k1
pubkey_point = string_to_number(I[0:32])*curve.generator + ser_to_point(cK)
public_key = ecdsa.VerifyingKey.from_public_point( pubkey_point, curve = SECP256k1 )
pubkey = ecc.ECPrivkey(I[0:32]) + ecc.ECPubkey(cK)
cK_n = pubkey.get_public_key_bytes(compressed=True)
c_n = I[32:]
cK_n = GetPubKey(public_key.pubkey,True)
return cK_n, c_n
@ -949,7 +571,7 @@ def xpub_header(xtype, *, net=None):
def serialize_xprv(xtype, c, k, depth=0, fingerprint=b'\x00'*4,
child_number=b'\x00'*4, *, net=None):
if not (0 < string_to_number(k) < SECP256k1.order):
if not ecc.is_secret_within_curve_range(k):
raise BitcoinException('Impossible xprv (not within curve order)')
xprv = xprv_header(xtype, net=net) \
+ bytes([depth]) + fingerprint + child_number + c + bytes([0]) + k
@ -982,7 +604,7 @@ def deserialize_xkey(xkey, prv, *, net=None):
xtype = list(headers.keys())[list(headers.values()).index(header)]
n = 33 if prv else 32
K_or_k = xkey[13+n:]
if prv and not (0 < string_to_number(K_or_k) < SECP256k1.order):
if prv and not ecc.is_secret_within_curve_range(K_or_k):
raise BitcoinException('Impossible xprv (not within curve order)')
return xtype, depth, fingerprint, child_number, c, K_or_k
@ -1015,7 +637,7 @@ def is_xprv(text):
def xpub_from_xprv(xprv):
xtype, depth, fingerprint, child_number, c, k = deserialize_xprv(xprv)
K, cK = get_pubkeys_from_secret(k)
cK = ecc.ECPrivkey(k).get_public_key_bytes(compressed=True)
return serialize_xpub(xtype, c, cK, depth, fingerprint, child_number)
@ -1023,14 +645,16 @@ def bip32_root(seed, xtype):
I = hmac.new(b"Bitcoin seed", seed, hashlib.sha512).digest()
master_k = I[0:32]
master_c = I[32:]
K, cK = get_pubkeys_from_secret(master_k)
# create xprv first, as that will check if master_k is within curve order
xprv = serialize_xprv(xtype, master_c, master_k)
cK = ecc.ECPrivkey(master_k).get_public_key_bytes(compressed=True)
xpub = serialize_xpub(xtype, master_c, cK)
return xprv, xpub
def xpub_from_pubkey(xtype, cK):
assert cK[0] in [0x02, 0x03]
if cK[0] not in (0x02, 0x03):
raise ValueError('Unexpected first byte: {}'.format(cK[0]))
return serialize_xpub(xtype, b'\x00'*32, cK)
@ -1064,10 +688,10 @@ def bip32_private_derivation(xprv, branch, sequence):
parent_k = k
k, c = CKD_priv(k, c, i)
depth += 1
_, parent_cK = get_pubkeys_from_secret(parent_k)
parent_cK = ecc.ECPrivkey(parent_k).get_public_key_bytes(compressed=True)
fingerprint = hash_160(parent_cK)[0:4]
child_number = bfh("%08X"%i)
K, cK = get_pubkeys_from_secret(k)
cK = ecc.ECPrivkey(k).get_public_key_bytes(compressed=True)
xpub = serialize_xpub(xtype, c, cK, depth, fingerprint, child_number)
xprv = serialize_xprv(xtype, c, k, depth, fingerprint, child_number)
return xprv, xpub

14
lib/commands.py
View file

@ -33,7 +33,7 @@ import base64
from functools import wraps
from decimal import Decimal
from .import util
from .import util, ecc
from .util import bfh, bh2u, format_satoshis, json_decode, print_error, json_encode
from .import bitcoin
from .bitcoin import is_address, hash_160, COIN, TYPE_ADDRESS
@ -219,7 +219,7 @@ class Commands:
sec = txin.get('privkey')
if sec:
txin_type, privkey, compressed = bitcoin.deserialize_privkey(sec)
pubkey = bitcoin.public_key_from_private_key(privkey, compressed)
pubkey = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)
keypairs[pubkey] = privkey, compressed
txin['type'] = txin_type
txin['x_pubkeys'] = [pubkey]
@ -237,8 +237,8 @@ class Commands:
tx = Transaction(tx)
if privkey:
txin_type, privkey2, compressed = bitcoin.deserialize_privkey(privkey)
pubkey = bitcoin.public_key_from_private_key(privkey2, compressed)
h160 = bitcoin.hash_160(bfh(pubkey))
pubkey_bytes = ecc.ECPrivkey(privkey2).get_public_key_bytes(compressed=compressed)
h160 = bitcoin.hash_160(pubkey_bytes)
x_pubkey = 'fd' + bh2u(b'\x00' + h160)
tx.sign({x_pubkey:(privkey2, compressed)})
else:
@ -405,7 +405,7 @@ class Commands:
"""Verify a signature."""
sig = base64.b64decode(signature)
message = util.to_bytes(message)
return bitcoin.verify_message(address, sig, message)
return ecc.verify_message_with_address(address, sig, message)
def _mktx(self, outputs, fee, change_addr, domain, nocheck, unsigned, rbf, password, locktime=None):
self.nocheck = nocheck
@ -527,7 +527,9 @@ class Commands:
@command('')
def encrypt(self, pubkey, message):
"""Encrypt a message with a public key. Use quotes if the message contains whitespaces."""
return bitcoin.encrypt_message(message, pubkey)
public_key = ecc.ECPubkey(bfh(pubkey))
encrypted = public_key.encrypt_message(message)
return encrypted
@command('wp')
def decrypt(self, pubkey, encrypted, password=None):

142
lib/crypto.py Normal file
View file

@ -0,0 +1,142 @@
# -*- coding: utf-8 -*-
#
# Electrum - lightweight Bitcoin client
# Copyright (C) 2018 The Electrum developers
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import base64
import os
import hashlib
import pyaes
from .util import assert_bytes, InvalidPassword, to_bytes, to_string
try:
from Cryptodome.Cipher import AES
except:
AES = None
class InvalidPadding(Exception):
pass
def append_PKCS7_padding(data):
assert_bytes(data)
padlen = 16 - (len(data) % 16)
return data + bytes([padlen]) * padlen
def strip_PKCS7_padding(data):
assert_bytes(data)
if len(data) % 16 != 0 or len(data) == 0:
raise InvalidPadding("invalid length")
padlen = data[-1]
if padlen > 16:
raise InvalidPadding("invalid padding byte (large)")
for i in data[-padlen:]:
if i != padlen:
raise InvalidPadding("invalid padding byte (inconsistent)")
return data[0:-padlen]
def aes_encrypt_with_iv(key, iv, data):
assert_bytes(key, iv, data)
data = append_PKCS7_padding(data)
if AES:
e = AES.new(key, AES.MODE_CBC, iv).encrypt(data)
else:
aes_cbc = pyaes.AESModeOfOperationCBC(key, iv=iv)
aes = pyaes.Encrypter(aes_cbc, padding=pyaes.PADDING_NONE)
e = aes.feed(data) + aes.feed() # empty aes.feed() flushes buffer
return e
def aes_decrypt_with_iv(key, iv, data):
assert_bytes(key, iv, data)
if AES:
cipher = AES.new(key, AES.MODE_CBC, iv)
data = cipher.decrypt(data)
else:
aes_cbc = pyaes.AESModeOfOperationCBC(key, iv=iv)
aes = pyaes.Decrypter(aes_cbc, padding=pyaes.PADDING_NONE)
data = aes.feed(data) + aes.feed() # empty aes.feed() flushes buffer
try:
return strip_PKCS7_padding(data)
except InvalidPadding:
raise InvalidPassword()
def EncodeAES(secret, s):
assert_bytes(s)
iv = bytes(os.urandom(16))
ct = aes_encrypt_with_iv(secret, iv, s)
e = iv + ct
return base64.b64encode(e)
def DecodeAES(secret, e):
e = bytes(base64.b64decode(e))
iv, e = e[:16], e[16:]
s = aes_decrypt_with_iv(secret, iv, e)
return s
def pw_encode(s, password):
if password:
secret = Hash(password)
return EncodeAES(secret, to_bytes(s, "utf8")).decode('utf8')
else:
return s
def pw_decode(s, password):
if password is not None:
secret = Hash(password)
try:
d = to_string(DecodeAES(secret, s), "utf8")
except Exception:
raise InvalidPassword()
return d
else:
return s
def sha256(x: bytes) -> bytes:
x = to_bytes(x, 'utf8')
return bytes(hashlib.sha256(x).digest())
def Hash(x: bytes) -> bytes:
x = to_bytes(x, 'utf8')
out = bytes(sha256(sha256(x)))
return out
def hash_160(x: bytes) -> bytes:
try:
md = hashlib.new('ripemd160')
md.update(sha256(x))
return md.digest()
except BaseException:
from . import ripemd
md = ripemd.new(sha256(x))
return md.digest()

407
lib/ecc.py Normal file
View file

@ -0,0 +1,407 @@
# -*- coding: utf-8 -*-
#
# Electrum - lightweight Bitcoin client
# Copyright (C) 2018 The Electrum developers
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import base64
import hmac
import hashlib
from typing import Union
import ecdsa
from ecdsa.ecdsa import curve_secp256k1, generator_secp256k1
from ecdsa.curves import SECP256k1
from ecdsa.ellipticcurve import Point
from ecdsa.util import string_to_number, number_to_string
from .util import bfh, bh2u, assert_bytes, print_error, to_bytes, InvalidPassword, profiler
from .crypto import (Hash, aes_encrypt_with_iv, aes_decrypt_with_iv)
from .ecc_fast import do_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1
do_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1()
CURVE_ORDER = SECP256k1.order
def generator():
return ECPubkey.from_point(generator_secp256k1)
def sig_string_from_der_sig(der_sig):
r, s = ecdsa.util.sigdecode_der(der_sig, CURVE_ORDER)
return ecdsa.util.sigencode_string(r, s, CURVE_ORDER)
def der_sig_from_sig_string(sig_string):
r, s = ecdsa.util.sigdecode_string(sig_string, CURVE_ORDER)
return ecdsa.util.sigencode_der_canonize(r, s, CURVE_ORDER)
def der_sig_from_r_and_s(r, s):
return ecdsa.util.sigencode_der_canonize(r, s, CURVE_ORDER)
def get_r_and_s_from_sig_string(sig_string):
r, s = ecdsa.util.sigdecode_string(sig_string, CURVE_ORDER)
return r, s
def sig_string_from_r_and_s(r, s):
return ecdsa.util.sigencode_string_canonize(r, s, CURVE_ORDER)
def point_to_ser(P, compressed=True) -> bytes:
if isinstance(P, tuple):
assert len(P) == 2, 'unexpected point: %s' % P
x, y = P
else:
x, y = P.x(), P.y()
if compressed:
return bfh(('%02x' % (2+(y&1))) + ('%064x' % x))
return bfh('04'+('%064x' % x)+('%064x' % y))
def get_y_coord_from_x(x, odd=True):
curve = curve_secp256k1
_p = curve.p()
_a = curve.a()
_b = curve.b()
for offset in range(128):
Mx = x + offset
My2 = pow(Mx, 3, _p) + _a * pow(Mx, 2, _p) + _b % _p
My = pow(My2, (_p + 1) // 4, _p)
if curve.contains_point(Mx, My):
if odd == bool(My & 1):
return My
return _p - My
raise Exception('ECC_YfromX: No Y found')
def ser_to_point(ser: bytes) -> (int, int):
if ser[0] not in (0x02, 0x03, 0x04):
raise ValueError('Unexpected first byte: {}'.format(ser[0]))
if ser[0] == 0x04:
return string_to_number(ser[1:33]), string_to_number(ser[33:])
x = string_to_number(ser[1:])
return x, get_y_coord_from_x(x, ser[0] == 0x03)
def _ser_to_python_ecdsa_point(ser: bytes) -> ecdsa.ellipticcurve.Point:
x, y = ser_to_point(ser)
return Point(curve_secp256k1, x, y, CURVE_ORDER)
class InvalidECPointException(Exception):
"""e.g. not on curve, or infinity"""
class _MyVerifyingKey(ecdsa.VerifyingKey):
@classmethod
def from_signature(klass, sig, recid, h, curve): # TODO use libsecp??
""" See http://www.secg.org/download/aid-780/sec1-v2.pdf, chapter 4.1.6 """
from ecdsa import util, numbertheory
from . import msqr
curveFp = curve.curve
G = curve.generator
order = G.order()
# extract r,s from signature
r, s = util.sigdecode_string(sig, order)
# 1.1
x = r + (recid//2) * order
# 1.3
alpha = ( x * x * x + curveFp.a() * x + curveFp.b() ) % curveFp.p()
beta = msqr.modular_sqrt(alpha, curveFp.p())
y = beta if (beta - recid) % 2 == 0 else curveFp.p() - beta
# 1.4 the constructor checks that nR is at infinity
try:
R = Point(curveFp, x, y, order)
except:
raise InvalidECPointException()
# 1.5 compute e from message:
e = string_to_number(h)
minus_e = -e % order
# 1.6 compute Q = r^-1 (sR - eG)
inv_r = numbertheory.inverse_mod(r,order)
Q = inv_r * ( s * R + minus_e * G )
return klass.from_public_point( Q, curve )
class _MySigningKey(ecdsa.SigningKey):
"""Enforce low S values in signatures"""
def sign_number(self, number, entropy=None, k=None):
r, s = ecdsa.SigningKey.sign_number(self, number, entropy, k)
if s > CURVE_ORDER//2:
s = CURVE_ORDER - s
return r, s
class ECPubkey(object):
def __init__(self, b: bytes):
assert_bytes(b)
point = _ser_to_python_ecdsa_point(b)
self._pubkey = ecdsa.ecdsa.Public_key(generator_secp256k1, point)
@classmethod
def from_sig_string(cls, sig_string: bytes, recid: int, msg_hash: bytes):
assert_bytes(sig_string)
if len(sig_string) != 64:
raise Exception('Wrong encoding')
if recid < 0 or recid > 3:
raise ValueError('recid is {}, but should be 0 <= recid <= 3'.format(recid))
ecdsa_verifying_key = _MyVerifyingKey.from_signature(sig_string, recid, msg_hash, curve=SECP256k1)
ecdsa_point = ecdsa_verifying_key.pubkey.point
return ECPubkey(point_to_ser(ecdsa_point))
@classmethod
def from_signature65(cls, sig: bytes, msg_hash: bytes):
if len(sig) != 65:
raise Exception("Wrong encoding")
nV = sig[0]
if nV < 27 or nV >= 35:
raise Exception("Bad encoding")
if nV >= 31:
compressed = True
nV -= 4
else:
compressed = False
recid = nV - 27
return cls.from_sig_string(sig[1:], recid, msg_hash), compressed
@classmethod
def from_point(cls, point):
_bytes = point_to_ser(point, compressed=False) # faster than compressed
return ECPubkey(_bytes)
def get_public_key_bytes(self, compressed=True):
return point_to_ser(self.point(), compressed)
def get_public_key_hex(self, compressed=True):
return bh2u(self.get_public_key_bytes(compressed))
def point(self) -> (int, int):
return self._pubkey.point.x(), self._pubkey.point.y()
def __mul__(self, other: int):
if not isinstance(other, int):
raise TypeError('multiplication not defined for ECPubkey and {}'.format(type(other)))
ecdsa_point = self._pubkey.point * other
return self.from_point(ecdsa_point)
def __rmul__(self, other: int):
return self * other
def __add__(self, other):
if not isinstance(other, ECPubkey):
raise TypeError('addition not defined for ECPubkey and {}'.format(type(other)))
ecdsa_point = self._pubkey.point + other._pubkey.point
return self.from_point(ecdsa_point)
def __eq__(self, other):
return self.get_public_key_bytes() == other.get_public_key_bytes()
def __ne__(self, other):
return not (self == other)
def verify_message_for_address(self, sig65: bytes, message: bytes) -> None:
assert_bytes(message)
h = Hash(msg_magic(message))
public_key, compressed = self.from_signature65(sig65, h)
# check public key
if public_key != self:
raise Exception("Bad signature")
# check message
self.verify_message_hash(sig65[1:], h)
def verify_message_hash(self, sig_string: bytes, msg_hash: bytes) -> None:
assert_bytes(sig_string)
if len(sig_string) != 64:
raise Exception('Wrong encoding')
ecdsa_point = self._pubkey.point
verifying_key = _MyVerifyingKey.from_public_point(ecdsa_point, curve=SECP256k1)
verifying_key.verify_digest(sig_string, msg_hash, sigdecode=ecdsa.util.sigdecode_string)
def encrypt_message(self, message: bytes, magic: bytes = b'BIE1'):
"""
ECIES encryption/decryption methods; AES-128-CBC with PKCS7 is used as the cipher; hmac-sha256 is used as the mac
"""
assert_bytes(message)
randint = ecdsa.util.randrange(CURVE_ORDER)
ephemeral_exponent = number_to_string(randint, CURVE_ORDER)
ephemeral = ECPrivkey(ephemeral_exponent)
ecdh_key = (self * ephemeral.secret_scalar).get_public_key_bytes(compressed=True)
key = hashlib.sha512(ecdh_key).digest()
iv, key_e, key_m = key[0:16], key[16:32], key[32:]
ciphertext = aes_encrypt_with_iv(key_e, iv, message)
ephemeral_pubkey = ephemeral.get_public_key_bytes(compressed=True)
encrypted = magic + ephemeral_pubkey + ciphertext
mac = hmac.new(key_m, encrypted, hashlib.sha256).digest()
return base64.b64encode(encrypted + mac)
@classmethod
def order(cls):
return CURVE_ORDER
def msg_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_message_with_address(address: str, sig65: bytes, message: bytes):
from .bitcoin import pubkey_to_address
assert_bytes(sig65, message)
try:
h = Hash(msg_magic(message))
public_key, compressed = ECPubkey.from_signature65(sig65, h)
# check public key using the address
pubkey_hex = public_key.get_public_key_hex(compressed)
for txin_type in ['p2pkh','p2wpkh','p2wpkh-p2sh']:
addr = pubkey_to_address(txin_type, pubkey_hex)
if address == addr:
break
else:
raise Exception("Bad signature")
# check message
public_key.verify_message_hash(sig65[1:], h)
return True
except Exception as e:
print_error("Verification error: {0}".format(e))
return False
def is_secret_within_curve_range(secret: Union[int, bytes]) -> bool:
if isinstance(secret, bytes):
secret = string_to_number(secret)
return 0 < secret < CURVE_ORDER
class ECPrivkey(ECPubkey):
def __init__(self, privkey_bytes: bytes):
assert_bytes(privkey_bytes)
if len(privkey_bytes) != 32:
raise Exception('unexpected size for secret. should be 32 bytes, not {}'.format(len(privkey_bytes)))
secret = string_to_number(privkey_bytes)
if not is_secret_within_curve_range(secret):
raise Exception('Invalid secret scalar (not within curve order)')
self.secret_scalar = secret
point = generator_secp256k1 * secret
super().__init__(point_to_ser(point))
self._privkey = ecdsa.ecdsa.Private_key(self._pubkey, secret)
@classmethod
def from_secret_scalar(cls, secret_scalar: int):
secret_bytes = number_to_string(secret_scalar, CURVE_ORDER)
return ECPrivkey(secret_bytes)
@classmethod
def from_arbitrary_size_secret(cls, privkey_bytes: bytes):
"""This method is only for legacy reasons. Do not introduce new code that uses it.
Unlike the default constructor, this method does not require len(privkey_bytes) == 32,
and the secret does not need to be within the curve order either.
"""
return ECPrivkey(cls.normalize_secret_bytes(privkey_bytes))
@classmethod
def normalize_secret_bytes(cls, privkey_bytes: bytes) -> bytes:
scalar = string_to_number(privkey_bytes) % CURVE_ORDER
if scalar == 0:
raise Exception('invalid EC private key scalar: zero')
privkey_32bytes = number_to_string(scalar, CURVE_ORDER)
return privkey_32bytes
def sign_transaction(self, hashed_preimage):
private_key = _MySigningKey.from_secret_exponent(self.secret_scalar, curve=SECP256k1)
sig = private_key.sign_digest_deterministic(hashed_preimage, hashfunc=hashlib.sha256,
sigencode=ecdsa.util.sigencode_der)
public_key = private_key.get_verifying_key()
if not public_key.verify_digest(sig, hashed_preimage, sigdecode=ecdsa.util.sigdecode_der):
raise Exception('Sanity check verifying our own signature failed.')
return sig
def sign_message(self, message, is_compressed):
def sign_with_python_ecdsa(msg_hash):
private_key = _MySigningKey.from_secret_exponent(self.secret_scalar, curve=SECP256k1)
public_key = private_key.get_verifying_key()
signature = private_key.sign_digest_deterministic(msg_hash, hashfunc=hashlib.sha256, sigencode=ecdsa.util.sigencode_string)
if not public_key.verify_digest(signature, msg_hash, sigdecode=ecdsa.util.sigdecode_string):
raise Exception('Sanity check verifying our own signature failed.')
return signature
def bruteforce_recid(sig_string):
for recid in range(4):
sig65 = construct_sig65(sig_string, recid, is_compressed)
try:
self.verify_message_for_address(sig65, message)
return sig65, recid
except Exception as e:
continue
else:
raise Exception("error: cannot sign message. no recid fits..")
message = to_bytes(message, 'utf8')
msg_hash = Hash(msg_magic(message))
sig_string = sign_with_python_ecdsa(msg_hash)
sig65, recid = bruteforce_recid(sig_string)
try:
self.verify_message_for_address(sig65, message)
return sig65
except Exception as e:
raise Exception("error: cannot sign message. self-verify sanity check failed")
def decrypt_message(self, encrypted, magic=b'BIE1'):
encrypted = base64.b64decode(encrypted)
if len(encrypted) < 85:
raise Exception('invalid ciphertext: length')
magic_found = encrypted[:4]
ephemeral_pubkey_bytes = encrypted[4:37]
ciphertext = encrypted[37:-32]
mac = encrypted[-32:]
if magic_found != magic:
raise Exception('invalid ciphertext: invalid magic bytes')
try:
ecdsa_point = _ser_to_python_ecdsa_point(ephemeral_pubkey_bytes)
except AssertionError as e:
raise Exception('invalid ciphertext: invalid ephemeral pubkey') from e
if not ecdsa.ecdsa.point_is_valid(generator_secp256k1, ecdsa_point.x(), ecdsa_point.y()):
raise Exception('invalid ciphertext: invalid ephemeral pubkey')
ephemeral_pubkey = ECPubkey(point_to_ser(ecdsa_point))
ecdh_key = (ephemeral_pubkey * self.secret_scalar).get_public_key_bytes(compressed=True)
key = hashlib.sha512(ecdh_key).digest()
iv, key_e, key_m = key[0:16], key[16:32], key[32:]
if mac != hmac.new(key_m, encrypted[:-32], hashlib.sha256).digest():
raise InvalidPassword()
return aes_decrypt_with_iv(key_e, iv, ciphertext)
def construct_sig65(sig_string, recid, is_compressed):
comp = 4 if is_compressed else 0
return bytes([27 + recid + comp]) + sig_string

216
lib/ecc_fast.py Normal file
View file

@ -0,0 +1,216 @@
# taken (with minor modifications) from pycoin
# https://github.com/richardkiss/pycoin/blob/01b1787ed902df23f99a55deb00d8cd076a906fe/pycoin/ecdsa/native/secp256k1.py
import os
import sys
import traceback
import ctypes
from ctypes.util import find_library
from ctypes import (
byref, c_byte, c_int, c_uint, c_char_p, c_size_t, c_void_p, create_string_buffer, CFUNCTYPE, POINTER
)
import ecdsa
from .util import print_stderr, print_error
SECP256K1_FLAGS_TYPE_MASK = ((1 << 8) - 1)
SECP256K1_FLAGS_TYPE_CONTEXT = (1 << 0)
SECP256K1_FLAGS_TYPE_COMPRESSION = (1 << 1)
# /** The higher bits contain the actual data. Do not use directly. */
SECP256K1_FLAGS_BIT_CONTEXT_VERIFY = (1 << 8)
SECP256K1_FLAGS_BIT_CONTEXT_SIGN = (1 << 9)
SECP256K1_FLAGS_BIT_COMPRESSION = (1 << 8)
# /** Flags to pass to secp256k1_context_create. */
SECP256K1_CONTEXT_VERIFY = (SECP256K1_FLAGS_TYPE_CONTEXT | SECP256K1_FLAGS_BIT_CONTEXT_VERIFY)
SECP256K1_CONTEXT_SIGN = (SECP256K1_FLAGS_TYPE_CONTEXT | SECP256K1_FLAGS_BIT_CONTEXT_SIGN)
SECP256K1_CONTEXT_NONE = (SECP256K1_FLAGS_TYPE_CONTEXT)
SECP256K1_EC_COMPRESSED = (SECP256K1_FLAGS_TYPE_COMPRESSION | SECP256K1_FLAGS_BIT_COMPRESSION)
SECP256K1_EC_UNCOMPRESSED = (SECP256K1_FLAGS_TYPE_COMPRESSION)
def load_library():
if sys.platform == 'darwin':
library_path = 'libsecp256k1.dylib'
elif sys.platform in ('windows', 'win32'):
library_path = 'libsecp256k1.dll'
else:
library_path = 'libsecp256k1.so.0'
secp256k1 = ctypes.cdll.LoadLibrary(library_path)
if not secp256k1:
print_stderr('[ecc] warning: libsecp256k1 library failed to load')
return None
try:
secp256k1.secp256k1_context_create.argtypes = [c_uint]
secp256k1.secp256k1_context_create.restype = c_void_p
secp256k1.secp256k1_context_randomize.argtypes = [c_void_p, c_char_p]
secp256k1.secp256k1_context_randomize.restype = c_int
secp256k1.secp256k1_ec_pubkey_create.argtypes = [c_void_p, c_void_p, c_char_p]
secp256k1.secp256k1_ec_pubkey_create.restype = c_int
secp256k1.secp256k1_ecdsa_sign.argtypes = [c_void_p, c_char_p, c_char_p, c_char_p, c_void_p, c_void_p]
secp256k1.secp256k1_ecdsa_sign.restype = c_int
secp256k1.secp256k1_ecdsa_verify.argtypes = [c_void_p, c_char_p, c_char_p, c_char_p]
secp256k1.secp256k1_ecdsa_verify.restype = c_int
secp256k1.secp256k1_ec_pubkey_parse.argtypes = [c_void_p, c_char_p, c_char_p, c_size_t]
secp256k1.secp256k1_ec_pubkey_parse.restype = c_int
secp256k1.secp256k1_ec_pubkey_serialize.argtypes = [c_void_p, c_char_p, c_void_p, c_char_p, c_uint]
secp256k1.secp256k1_ec_pubkey_serialize.restype = c_int
secp256k1.secp256k1_ecdsa_signature_parse_compact.argtypes = [c_void_p, c_char_p, c_char_p]
secp256k1.secp256k1_ecdsa_signature_parse_compact.restype = c_int
secp256k1.secp256k1_ecdsa_signature_serialize_compact.argtypes = [c_void_p, c_char_p, c_char_p]
secp256k1.secp256k1_ecdsa_signature_serialize_compact.restype = c_int
secp256k1.secp256k1_ec_pubkey_tweak_mul.argtypes = [c_void_p, c_char_p, c_char_p]
secp256k1.secp256k1_ec_pubkey_tweak_mul.restype = c_int
secp256k1.ctx = secp256k1.secp256k1_context_create(SECP256K1_CONTEXT_SIGN | SECP256K1_CONTEXT_VERIFY)
r = secp256k1.secp256k1_context_randomize(secp256k1.ctx, os.urandom(32))
if r:
return secp256k1
else:
print_stderr('[ecc] warning: secp256k1_context_randomize failed')
return None
except (OSError, AttributeError):
#traceback.print_exc(file=sys.stderr)
print_stderr('[ecc] warning: libsecp256k1 library was found and loaded but there was an error when using it')
return None
class _patched_functions:
prepared_to_patch = False
monkey_patching_active = False
def _prepare_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1():
if not _libsecp256k1:
return
# save original functions so that we can undo patching (needed for tests)
_patched_functions.orig_sign = staticmethod(ecdsa.ecdsa.Private_key.sign)
_patched_functions.orig_verify = staticmethod(ecdsa.ecdsa.Public_key.verifies)
_patched_functions.orig_mul = staticmethod(ecdsa.ellipticcurve.Point.__mul__)
curve_secp256k1 = ecdsa.ecdsa.curve_secp256k1
curve_order = ecdsa.curves.SECP256k1.order
point_at_infinity = ecdsa.ellipticcurve.INFINITY
def mul(self: ecdsa.ellipticcurve.Point, other: int):
if self.curve() != curve_secp256k1:
# this operation is not on the secp256k1 curve; use original implementation
return _patched_functions.orig_mul(self, other)
other %= curve_order
if self == point_at_infinity or other == 0:
return point_at_infinity
pubkey = create_string_buffer(64)
public_pair_bytes = b'\4' + self.x().to_bytes(32, byteorder="big") + self.y().to_bytes(32, byteorder="big")
r = _libsecp256k1.secp256k1_ec_pubkey_parse(
_libsecp256k1.ctx, pubkey, public_pair_bytes, len(public_pair_bytes))
if not r:
return False
r = _libsecp256k1.secp256k1_ec_pubkey_tweak_mul(_libsecp256k1.ctx, pubkey, other.to_bytes(32, byteorder="big"))
if not r:
return point_at_infinity
pubkey_serialized = create_string_buffer(65)
pubkey_size = c_size_t(65)
_libsecp256k1.secp256k1_ec_pubkey_serialize(
_libsecp256k1.ctx, pubkey_serialized, byref(pubkey_size), pubkey, SECP256K1_EC_UNCOMPRESSED)
x = int.from_bytes(pubkey_serialized[1:33], byteorder="big")
y = int.from_bytes(pubkey_serialized[33:], byteorder="big")
return ecdsa.ellipticcurve.Point(curve_secp256k1, x, y, curve_order)
def sign(self: ecdsa.ecdsa.Private_key, hash: int, random_k: int):
# note: random_k is ignored
if self.public_key.curve != curve_secp256k1:
# this operation is not on the secp256k1 curve; use original implementation
return _patched_functions.orig_sign(self, hash, random_k)
secret_exponent = self.secret_multiplier
nonce_function = None
sig = create_string_buffer(64)
sig_hash_bytes = hash.to_bytes(32, byteorder="big")
_libsecp256k1.secp256k1_ecdsa_sign(
_libsecp256k1.ctx, sig, sig_hash_bytes, secret_exponent.to_bytes(32, byteorder="big"), nonce_function, None)
compact_signature = create_string_buffer(64)
_libsecp256k1.secp256k1_ecdsa_signature_serialize_compact(_libsecp256k1.ctx, compact_signature, sig)
r = int.from_bytes(compact_signature[:32], byteorder="big")
s = int.from_bytes(compact_signature[32:], byteorder="big")
return ecdsa.ecdsa.Signature(r, s)
def verify(self: ecdsa.ecdsa.Public_key, hash: int, signature: ecdsa.ecdsa.Signature):
if self.curve != curve_secp256k1:
# this operation is not on the secp256k1 curve; use original implementation
return _patched_functions.orig_verify(self, hash, signature)
sig = create_string_buffer(64)
input64 = signature.r.to_bytes(32, byteorder="big") + signature.s.to_bytes(32, byteorder="big")
r = _libsecp256k1.secp256k1_ecdsa_signature_parse_compact(_libsecp256k1.ctx, sig, input64)
if not r:
return False
r = _libsecp256k1.secp256k1_ecdsa_signature_normalize(_libsecp256k1.ctx, sig, sig)
public_pair_bytes = b'\4' + self.point.x().to_bytes(32, byteorder="big") + self.point.y().to_bytes(32, byteorder="big")
pubkey = create_string_buffer(64)
r = _libsecp256k1.secp256k1_ec_pubkey_parse(
_libsecp256k1.ctx, pubkey, public_pair_bytes, len(public_pair_bytes))
if not r:
return False
return 1 == _libsecp256k1.secp256k1_ecdsa_verify(_libsecp256k1.ctx, sig, hash.to_bytes(32, byteorder="big"), pubkey)
# save new functions so that we can (re-)do patching
_patched_functions.fast_sign = sign
_patched_functions.fast_verify = verify
_patched_functions.fast_mul = mul
_patched_functions.prepared_to_patch = True
def do_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1():
if not _libsecp256k1:
print_stderr('[ecc] warning: libsecp256k1 library not available, falling back to python-ecdsa')
return
if not _patched_functions.prepared_to_patch:
raise Exception("can't patch python-ecdsa without preparations")
ecdsa.ecdsa.Private_key.sign = _patched_functions.fast_sign
ecdsa.ecdsa.Public_key.verifies = _patched_functions.fast_verify
ecdsa.ellipticcurve.Point.__mul__ = _patched_functions.fast_mul
# ecdsa.ellipticcurve.Point.__add__ = ... # TODO??
_patched_functions.monkey_patching_active = True
def undo_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1():
if not _libsecp256k1:
return
if not _patched_functions.prepared_to_patch:
raise Exception("can't patch python-ecdsa without preparations")
ecdsa.ecdsa.Private_key.sign = _patched_functions.orig_sign
ecdsa.ecdsa.Public_key.verifies = _patched_functions.orig_verify
ecdsa.ellipticcurve.Point.__mul__ = _patched_functions.orig_mul
_patched_functions.monkey_patching_active = False
def is_using_fast_ecc():
return _patched_functions.monkey_patching_active
try:
_libsecp256k1 = load_library()
except:
_libsecp256k1 = None
traceback.print_exc(file=sys.stderr)
_prepare_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1()

33
lib/keystore.py
View file

@ -26,8 +26,10 @@
from unicodedata import normalize
from . import bitcoin
from . import bitcoin, ecc
from .bitcoin import *
from .ecc import string_to_number, number_to_string
from .crypto import pw_decode, pw_encode
from . import constants
from .util import (PrintError, InvalidPassword, hfu, WalletFileException,
BitcoinException)
@ -90,12 +92,12 @@ class Software_KeyStore(KeyStore):
def sign_message(self, sequence, message, password):
privkey, compressed = self.get_private_key(sequence, password)
key = regenerate_key(privkey)
key = ecc.ECPrivkey(privkey)
return key.sign_message(message, compressed)
def decrypt_message(self, sequence, message, password):
privkey, compressed = self.get_private_key(sequence, password)
ec = regenerate_key(privkey)
ec = ecc.ECPrivkey(privkey)
decrypted = ec.decrypt_message(message)
return decrypted
@ -141,7 +143,7 @@ class Imported_KeyStore(Software_KeyStore):
def import_privkey(self, sec, password):
txin_type, privkey, compressed = deserialize_privkey(sec)
pubkey = public_key_from_private_key(privkey, compressed)
pubkey = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)
# re-serialize the key so the internal storage format is consistent
serialized_privkey = serialize_privkey(
privkey, compressed, txin_type, internal_use=True)
@ -159,7 +161,7 @@ class Imported_KeyStore(Software_KeyStore):
sec = pw_decode(self.keypairs[pubkey], password)
txin_type, privkey, compressed = deserialize_privkey(sec)
# this checks the password
if pubkey != public_key_from_private_key(privkey, compressed):
if pubkey != ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed):
raise InvalidPassword()
return privkey, compressed
@ -381,9 +383,8 @@ class Old_KeyStore(Deterministic_KeyStore):
@classmethod
def mpk_from_seed(klass, seed):
secexp = klass.stretch_key(seed)
master_private_key = ecdsa.SigningKey.from_secret_exponent(secexp, curve = SECP256k1)
master_public_key = master_private_key.get_verifying_key().to_string()
return bh2u(master_public_key)
privkey = ecc.ECPrivkey.from_secret_scalar(secexp)
return privkey.get_public_key_hex(compressed=False)[2:]
@classmethod
def stretch_key(self, seed):
@ -399,18 +400,16 @@ class Old_KeyStore(Deterministic_KeyStore):
@classmethod
def get_pubkey_from_mpk(self, mpk, for_change, n):
z = self.get_sequence(mpk, for_change, n)
master_public_key = ecdsa.VerifyingKey.from_string(bfh(mpk), curve = SECP256k1)
pubkey_point = master_public_key.pubkey.point + z*SECP256k1.generator
public_key2 = ecdsa.VerifyingKey.from_public_point(pubkey_point, curve = SECP256k1)
return '04' + bh2u(public_key2.to_string())
master_public_key = ecc.ECPubkey(bfh('04'+mpk))
public_key = master_public_key + z*ecc.generator()
return public_key.get_public_key_hex(compressed=False)
def derive_pubkey(self, for_change, n):
return self.get_pubkey_from_mpk(self.mpk, for_change, n)
def get_private_key_from_stretched_exponent(self, for_change, n, secexp):
order = generator_secp256k1.order()
secexp = (secexp + self.get_sequence(self.mpk, for_change, n)) % order
pk = number_to_string(secexp, generator_secp256k1.order())
secexp = (secexp + self.get_sequence(self.mpk, for_change, n)) % ecc.CURVE_ORDER
pk = number_to_string(secexp, ecc.CURVE_ORDER)
return pk
def get_private_key(self, sequence, password):
@ -423,8 +422,8 @@ class Old_KeyStore(Deterministic_KeyStore):
def check_seed(self, seed):
secexp = self.stretch_key(seed)
master_private_key = ecdsa.SigningKey.from_secret_exponent( secexp, curve = SECP256k1 )
master_public_key = master_private_key.get_verifying_key().to_string()
master_private_key = ecc.ECPrivkey.from_secret_scalar(secexp)
master_public_key = master_private_key.get_public_key_bytes(compressed=False)[1:]
if master_public_key != bfh(self.mpk):
print_error('invalid password (mpk)', self.mpk, bh2u(master_public_key))
raise InvalidPassword()

10
lib/paymentrequest.py
View file

@ -38,6 +38,7 @@ except ImportError:
sys.exit("Error: could not find paymentrequest_pb2.py. Create it with 'protoc --proto_path=lib/ --python_out=lib/ lib/paymentrequest.proto'")
from . import bitcoin
from . import ecc
from . import util
from .util import print_error, bh2u, bfh
from .util import export_meta, import_meta
@ -206,9 +207,9 @@ class PaymentRequest:
if pr.pki_type == "dnssec+btc":
self.requestor = alias
address = info.get('address')
pr.signature = ''
pr.signature = b''
message = pr.SerializeToString()
if bitcoin.verify_message(address, sig, message):
if ecc.verify_message_with_address(address, sig, message):
self.error = 'Verified with DNSSEC'
return True
else:
@ -321,10 +322,9 @@ def sign_request_with_alias(pr, alias, alias_privkey):
pr.pki_type = 'dnssec+btc'
pr.pki_data = str(alias)
message = pr.SerializeToString()
ec_key = bitcoin.regenerate_key(alias_privkey)
address = bitcoin.address_from_private_key(alias_privkey)
ec_key = ecc.ECPrivkey(alias_privkey)
compressed = bitcoin.is_compressed(alias_privkey)
pr.signature = ec_key.sign_message(message, compressed, address)
pr.signature = ec_key.sign_message(message, compressed)
def verify_cert_chain(chain):

23
lib/storage.py
View file

@ -33,10 +33,11 @@ import pbkdf2, hmac, hashlib
import base64
import zlib
from .util import PrintError, profiler, InvalidPassword, WalletFileException
from .util import PrintError, profiler, InvalidPassword, WalletFileException, bfh
from .plugins import run_hook, plugin_loaders
from .keystore import bip44_derivation
from . import bitcoin
from . import ecc
# seed_version is now used for the version of the wallet file
@ -162,9 +163,10 @@ class WalletStorage(PrintError):
def file_exists(self):
return self.path and os.path.exists(self.path)
def get_key(self, password):
secret = pbkdf2.PBKDF2(password, '', iterations = 1024, macmodule = hmac, digestmodule = hashlib.sha512).read(64)
ec_key = bitcoin.EC_KEY(secret)
@staticmethod
def get_eckey_from_password(password):
secret = pbkdf2.PBKDF2(password, '', iterations=1024, macmodule=hmac, digestmodule=hashlib.sha512).read(64)
ec_key = ecc.ECPrivkey.from_arbitrary_size_secret(secret)
return ec_key
def _get_encryption_magic(self):
@ -177,13 +179,13 @@ class WalletStorage(PrintError):
raise WalletFileException('no encryption magic for version: %s' % v)
def decrypt(self, password):
ec_key = self.get_key(password)
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))
else:
s = None
self.pubkey = ec_key.get_public_key()
self.pubkey = ec_key.get_public_key_hex()
s = s.decode('utf8')
self.load_data(s)
@ -191,7 +193,7 @@ class WalletStorage(PrintError):
"""Raises an InvalidPassword exception on invalid password"""
if not self.is_encrypted():
return
if self.pubkey and self.pubkey != self.get_key(password).get_public_key():
if self.pubkey and self.pubkey != self.get_eckey_from_password(password).get_public_key_hex():
raise InvalidPassword()
def set_keystore_encryption(self, enable):
@ -202,8 +204,8 @@ class WalletStorage(PrintError):
if enc_version is None:
enc_version = self._encryption_version
if password and enc_version != STO_EV_PLAINTEXT:
ec_key = self.get_key(password)
self.pubkey = ec_key.get_public_key()
ec_key = self.get_eckey_from_password(password)
self.pubkey = ec_key.get_public_key_hex()
self._encryption_version = enc_version
else:
self.pubkey = None
@ -253,7 +255,8 @@ class WalletStorage(PrintError):
s = bytes(s, 'utf8')
c = zlib.compress(s)
enc_magic = self._get_encryption_magic()
s = bitcoin.encrypt_message(c, self.pubkey, enc_magic)
public_key = ecc.ECPubkey(bfh(self.pubkey))
s = public_key.encrypt_message(c, enc_magic)
s = s.decode('utf8')
temp_path = "%s.tmp.%s" % (self.path, os.getpid())

17
lib/tests/__init__.py
View file

@ -1,9 +1,24 @@
import unittest
import threading
from lib import constants
class TestCaseForTestnet(unittest.TestCase):
# some unit tests are modifying globals; sorry.
class SequentialTestCase(unittest.TestCase):
test_lock = threading.Lock()
def setUp(self):
super().setUp()
self.test_lock.acquire()
def tearDown(self):
super().tearDown()
self.test_lock.release()
class TestCaseForTestnet(SequentialTestCase):
@classmethod
def setUpClass(cls):

133
lib/tests/test_bitcoin.py
View file

@ -1,21 +1,26 @@
import base64
import unittest
import sys
from ecdsa.util import number_to_string
from lib import bitcoin
from lib.bitcoin import (
generator_secp256k1, point_to_ser, public_key_to_p2pkh, EC_KEY,
bip32_root, bip32_public_derivation, bip32_private_derivation, pw_encode,
pw_decode, Hash, public_key_from_private_key, address_from_private_key,
public_key_to_p2pkh,
bip32_root, bip32_public_derivation, bip32_private_derivation,
Hash, address_from_private_key,
is_address, is_private_key, xpub_from_xprv, is_new_seed, is_old_seed,
var_int, op_push, address_to_script, regenerate_key,
verify_message, deserialize_privkey, serialize_privkey, is_segwit_address,
var_int, op_push, address_to_script,
deserialize_privkey, serialize_privkey, is_segwit_address,
is_b58_address, address_to_scripthash, is_minikey, is_compressed, is_xpub,
xpub_type, is_xprv, is_bip32_derivation, seed_type, EncodeBase58Check,
script_num_to_hex, push_script, add_number_to_script)
from lib import ecc, crypto, ecc_fast
from lib.ecc import number_to_string, string_to_number
from lib.transaction import opcodes
from lib.util import bfh, bh2u
from lib import constants
from lib.storage import WalletStorage
from . import SequentialTestCase
from . import TestCaseForTestnet
@ -26,27 +31,54 @@ except ImportError:
sys.exit("Error: python-ecdsa does not seem to be installed. Try 'sudo pip install ecdsa'")
class Test_bitcoin(unittest.TestCase):
def needs_test_with_all_ecc_implementations(func):
"""Function decorator to run a unit test twice:
once when libsecp256k1 is not available, once when it is.
NOTE: this is inherently sequential;
tests running in parallel would break things
"""
def run_test(*args, **kwargs):
ecc_fast.undo_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1()
try:
# first test without libsecp
func(*args, **kwargs)
finally:
# if libsecp is not available, we are done
if not ecc_fast._libsecp256k1:
return
ecc_fast.do_monkey_patching_of_python_ecdsa_internals_with_libsecp256k1()
# if libsecp is available, test again now
func(*args, **kwargs)
return run_test
class Test_bitcoin(SequentialTestCase):
def test_libsecp256k1_is_available(self):
# we want the unit testing framework to test with libsecp256k1 available.
self.assertTrue(bool(ecc_fast._libsecp256k1))
@needs_test_with_all_ecc_implementations
def test_crypto(self):
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):
G = generator_secp256k1
G = ecc.generator()
_r = G.order()
pvk = ecdsa.util.randrange( pow(2,256) ) %_r
pvk = ecdsa.util.randrange(_r)
Pub = pvk*G
pubkey_c = point_to_ser(Pub,True)
pubkey_c = Pub.get_public_key_bytes(True)
#pubkey_u = point_to_ser(Pub,False)
addr_c = public_key_to_p2pkh(pubkey_c)
#print "Private key ", '%064x'%pvk
eck = EC_KEY(number_to_string(pvk,_r))
eck = ecc.ECPrivkey(number_to_string(pvk,_r))
#print "Compressed public key ", pubkey_c.encode('hex')
enc = EC_KEY.encrypt_message(message, pubkey_c)
enc = ecc.ECPubkey(pubkey_c).encrypt_message(message)
dec = eck.decrypt_message(enc)
self.assertEqual(message, dec)
@ -57,15 +89,16 @@ class Test_bitcoin(unittest.TestCase):
signature = eck.sign_message(message, True)
#print signature
EC_KEY.verify_message(eck, signature, message)
eck.verify_message_for_address(signature, message)
@needs_test_with_all_ecc_implementations
def test_msg_signing(self):
msg1 = b'Chancellor on brink of second bailout for banks'
msg2 = b'Electrum'
def sign_message_with_wif_privkey(wif_privkey, msg):
txin_type, privkey, compressed = deserialize_privkey(wif_privkey)
key = regenerate_key(privkey)
key = ecc.ECPrivkey(privkey)
return key.sign_message(msg, compressed)
sig1 = sign_message_with_wif_privkey(
@ -81,30 +114,61 @@ class Test_bitcoin(unittest.TestCase):
self.assertEqual(sig1_b64, b'H/9jMOnj4MFbH3d7t4yCQ9i7DgZU/VZ278w3+ySv2F4yIsdqjsc5ng3kmN8OZAThgyfCZOQxZCWza9V5XzlVY0Y=')
self.assertEqual(sig2_b64, b'G84dmJ8TKIDKMT9qBRhpX2sNmR0y5t+POcYnFFJCs66lJmAs3T8A6Sbpx7KA6yTQ9djQMabwQXRrDomOkIKGn18=')
self.assertTrue(verify_message(addr1, sig1, msg1))
self.assertTrue(verify_message(addr2, sig2, msg2))
self.assertTrue(ecc.verify_message_with_address(addr1, sig1, msg1))
self.assertTrue(ecc.verify_message_with_address(addr2, sig2, msg2))
self.assertFalse(verify_message(addr1, b'wrong', msg1))
self.assertFalse(verify_message(addr1, sig2, msg1))
self.assertFalse(ecc.verify_message_with_address(addr1, b'wrong', msg1))
self.assertFalse(ecc.verify_message_with_address(addr1, sig2, msg1))
@needs_test_with_all_ecc_implementations
def test_decrypt_message(self):
key = WalletStorage.get_eckey_from_password('pw123')
self.assertEqual(b'me<(s_s)>age', key.decrypt_message(b'QklFMQMDFtgT3zWSQsa+Uie8H/WvfUjlu9UN9OJtTt3KlgKeSTi6SQfuhcg1uIz9hp3WIUOFGTLr4RNQBdjPNqzXwhkcPi2Xsbiw6UCNJncVPJ6QBg=='))
self.assertEqual(b'me<(s_s)>age', key.decrypt_message(b'QklFMQKXOXbylOQTSMGfo4MFRwivAxeEEkewWQrpdYTzjPhqjHcGBJwdIhB7DyRfRQihuXx1y0ZLLv7XxLzrILzkl/H4YUtZB4uWjuOAcmxQH4i/Og=='))
self.assertEqual(b'hey_there' * 100, key.decrypt_message(b'QklFMQLOOsabsXtGQH8edAa6VOUa5wX8/DXmxX9NyHoAx1a5bWgllayGRVPeI2bf0ZdWK0tfal0ap0ZIVKbd2eOJybqQkILqT6E1/Syzq0Zicyb/AA1eZNkcX5y4gzloxinw00ubCA8M7gcUjJpOqbnksATcJ5y2YYXcHMGGfGurWu6uJ/UyrNobRidWppRMW5yR9/6utyNvT6OHIolCMEf7qLcmtneoXEiz51hkRdZS7weNf9mGqSbz9a2NL3sdh1A0feHIjAZgcCKcAvksNUSauf0/FnIjzTyPRpjRDMeDC8Ci3sGiuO3cvpWJwhZfbjcS26KmBv2CHWXfRRNFYOInHZNIXWNAoBB47Il5bGSMd+uXiGr+SQ9tNvcu+BiJNmFbxYqg+oQ8dGAl1DtvY2wJVY8k7vO9BIWSpyIxfGw7EDifhc5vnOmGe016p6a01C3eVGxgl23UYMrP7+fpjOcPmTSF4rk5U5ljEN3MSYqlf1QEv0OqlI9q1TwTK02VBCjMTYxDHsnt04OjNBkNO8v5uJ4NR+UUDBEp433z53I59uawZ+dbk4v4ZExcl8EGmKm3Gzbal/iJ/F7KQuX2b/ySEhLOFVYFWxK73X1nBvCSK2mC2/8fCw8oI5pmvzJwQhcCKTdEIrz3MMvAHqtPScDUOjzhXxInQOCb3+UBj1PPIdqkYLvZss1TEaBwYZjLkVnK2MBj7BaqT6Rp6+5A/fippUKHsnB6eYMEPR2YgDmCHL+4twxHJG6UWdP3ybaKiiAPy2OHNP6PTZ0HrqHOSJzBSDD+Z8YpaRg29QX3UEWlqnSKaan0VYAsV1VeaN0XFX46/TWO0L5tjhYVXJJYGqo6tIQJymxATLFRF6AZaD1Mwd27IAL04WkmoQoXfO6OFfwdp/shudY/1gBkDBvGPICBPtnqkvhGF+ZF3IRkuPwiFWeXmwBxKHsRx/3+aJu32Ml9+za41zVk2viaxcGqwTc5KMexQFLAUwqhv+aIik7U+5qk/gEVSuRoVkihoweFzKolNF+BknH2oB4rZdPixag5Zje3DvgjsSFlOl69W/67t/Gs8htfSAaHlsB8vWRQr9+v/lxTbrAw+O0E+sYGoObQ4qQMyQshNZEHbpPg63eWiHtJJnrVBvOeIbIHzoLDnMDsWVWZSMzAQ1vhX1H5QLgSEbRlKSliVY03kDkh/Nk/KOn+B2q37Ialq4JcRoIYFGJ8AoYEAD0tRuTqFddIclE75HzwaNG7NyKW1plsa72ciOPwsPJsdd5F0qdSQ3OSKtooTn7uf6dXOc4lDkfrVYRlZ0PX'))
@needs_test_with_all_ecc_implementations
def test_encrypt_message(self):
key = WalletStorage.get_eckey_from_password('secret_password77')
msgs = [
bytes([0] * 555),
b'cannot think of anything funny'
]
for plaintext in msgs:
ciphertext1 = key.encrypt_message(plaintext)
ciphertext2 = key.encrypt_message(plaintext)
self.assertEqual(plaintext, key.decrypt_message(ciphertext1))
self.assertEqual(plaintext, key.decrypt_message(ciphertext2))
self.assertNotEqual(ciphertext1, ciphertext2)
@needs_test_with_all_ecc_implementations
def test_sign_transaction(self):
eckey1 = ecc.ECPrivkey(bfh('7e1255fddb52db1729fc3ceb21a46f95b8d9fe94cc83425e936a6c5223bb679d'))
sig1 = eckey1.sign_transaction(bfh('5a548b12369a53faaa7e51b5081829474ebdd9c924b3a8230b69aa0be254cd94'))
self.assertEqual(bfh('3045022100902a288b98392254cd23c0e9a49ac6d7920f171b8249a48e484b998f1874a2010220723d844826828f092cf400cb210c4fa0b8cd1b9d1a7f21590e78e022ff6476b9'), sig1)
eckey2 = ecc.ECPrivkey(bfh('c7ce8c1462c311eec24dff9e2532ac6241e50ae57e7d1833af21942136972f23'))
sig2 = eckey2.sign_transaction(bfh('642a2e66332f507c92bda910158dfe46fc10afbf72218764899d3af99a043fac'))
self.assertEqual(bfh('30440220618513f4cfc87dde798ce5febae7634c23e7b9254a1eabf486be820f6a7c2c4702204fef459393a2b931f949e63ced06888f35e286e446dc46feb24b5b5f81c6ed52'), sig2)
def test_aes_homomorphic(self):
"""Make sure AES is homomorphic."""
payload = u'\u66f4\u7a33\u5b9a\u7684\u4ea4\u6613\u5e73\u53f0'
password = u'secret'
enc = pw_encode(payload, password)
dec = pw_decode(enc, password)
enc = crypto.pw_encode(payload, password)
dec = crypto.pw_decode(enc, password)
self.assertEqual(dec, payload)
def test_aes_encode_without_password(self):
"""When not passed a password, pw_encode is noop on the payload."""
payload = u'\u66f4\u7a33\u5b9a\u7684\u4ea4\u6613\u5e73\u53f0'
enc = pw_encode(payload, None)
enc = crypto.pw_encode(payload, None)
self.assertEqual(payload, enc)
def test_aes_deencode_without_password(self):
"""When not passed a password, pw_decode is noop on the payload."""
payload = u'\u66f4\u7a33\u5b9a\u7684\u4ea4\u6613\u5e73\u53f0'
enc = pw_decode(payload, None)
enc = crypto.pw_decode(payload, None)
self.assertEqual(payload, enc)
def test_aes_decode_with_invalid_password(self):
@ -112,8 +176,8 @@ class Test_bitcoin(unittest.TestCase):
payload = u"blah"
password = u"uber secret"
wrong_password = u"not the password"
enc = pw_encode(payload, password)
self.assertRaises(Exception, pw_decode, enc, wrong_password)
enc = crypto.pw_encode(payload, password)
self.assertRaises(Exception, crypto.pw_decode, enc, wrong_password)
def test_hash(self):
"""Make sure the Hash function does sha256 twice"""
@ -238,7 +302,7 @@ class Test_bitcoin_testnet(TestCaseForTestnet):
self.assertEqual(address_to_script('2NE4ZdmxFmUgwu5wtfoN2gVniyMgRDYq1kk'), 'a914e4567743d378957cd2ee7072da74b1203c1a7a0b87')
class Test_xprv_xpub(unittest.TestCase):
class Test_xprv_xpub(SequentialTestCase):
xprv_xpub = (
# Taken from test vectors in https://en.bitcoin.it/wiki/BIP_0032_TestVectors
@ -269,6 +333,7 @@ class Test_xprv_xpub(unittest.TestCase):
return xpub, xprv
@needs_test_with_all_ecc_implementations
def test_bip32(self):
# see https://en.bitcoin.it/wiki/BIP_0032_TestVectors
xpub, xprv = self._do_test_bip32("000102030405060708090a0b0c0d0e0f", "m/0'/1/2'/2/1000000000")
@ -279,12 +344,14 @@ class Test_xprv_xpub(unittest.TestCase):
self.assertEqual("xpub6FnCn6nSzZAw5Tw7cgR9bi15UV96gLZhjDstkXXxvCLsUXBGXPdSnLFbdpq8p9HmGsApME5hQTZ3emM2rnY5agb9rXpVGyy3bdW6EEgAtqt", xpub)
self.assertEqual("xprvA2nrNbFZABcdryreWet9Ea4LvTJcGsqrMzxHx98MMrotbir7yrKCEXw7nadnHM8Dq38EGfSh6dqA9QWTyefMLEcBYJUuekgW4BYPJcr9E7j", xprv)
@needs_test_with_all_ecc_implementations
def test_xpub_from_xprv(self):
"""We can derive the xpub key from a xprv."""
for xprv_details in self.xprv_xpub:
result = xpub_from_xprv(xprv_details['xprv'])
self.assertEqual(result, xprv_details['xpub'])
@needs_test_with_all_ecc_implementations
def test_is_xpub(self):
for xprv_details in self.xprv_xpub:
xpub = xprv_details['xpub']
@ -292,11 +359,13 @@ class Test_xprv_xpub(unittest.TestCase):
self.assertFalse(is_xpub('xpub1nval1d'))
self.assertFalse(is_xpub('xpub661MyMwAqRbcFWohJWt7PHsFEJfZAvw9ZxwQoDa4SoMgsDDM1T7WK3u9E4edkC4ugRnZ8E4xDZRpk8Rnts3Nbt97dPwT52WRONGBADWRONG'))
@needs_test_with_all_ecc_implementations
def test_xpub_type(self):
for xprv_details in self.xprv_xpub:
xpub = xprv_details['xpub']
self.assertEqual(xprv_details['xtype'], xpub_type(xpub))
@needs_test_with_all_ecc_implementations
def test_is_xprv(self):
for xprv_details in self.xprv_xpub:
xprv = xprv_details['xprv']
@ -388,7 +457,7 @@ class Test_xprv_xpub_testnet(TestCaseForTestnet):
self.assertTrue(xkey_b58.startswith(xpub_headers_b58[xtype]))
class Test_keyImport(unittest.TestCase):
class Test_keyImport(SequentialTestCase):
priv_pub_addr = (
{'priv': 'KzMFjMC2MPadjvX5Cd7b8AKKjjpBSoRKUTpoAtN6B3J9ezWYyXS6',
@ -475,19 +544,22 @@ class Test_keyImport(unittest.TestCase):
'scripthash': '60ad5a8b922f758cd7884403e90ee7e6f093f8d21a0ff24c9a865e695ccefdf1'},
)
@needs_test_with_all_ecc_implementations
def test_public_key_from_private_key(self):
for priv_details in self.priv_pub_addr:
txin_type, privkey, compressed = deserialize_privkey(priv_details['priv'])
result = public_key_from_private_key(privkey, compressed)
result = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)
self.assertEqual(priv_details['pub'], result)
self.assertEqual(priv_details['txin_type'], txin_type)
self.assertEqual(priv_details['compressed'], compressed)
@needs_test_with_all_ecc_implementations
def test_address_from_private_key(self):
for priv_details in self.priv_pub_addr:
addr2 = address_from_private_key(priv_details['priv'])
self.assertEqual(priv_details['address'], addr2)
@needs_test_with_all_ecc_implementations
def test_is_valid_address(self):
for priv_details in self.priv_pub_addr:
addr = priv_details['address']
@ -503,6 +575,7 @@ class Test_keyImport(unittest.TestCase):
self.assertFalse(is_address("not an address"))
@needs_test_with_all_ecc_implementations
def test_is_private_key(self):
for priv_details in self.priv_pub_addr:
self.assertTrue(is_private_key(priv_details['priv']))
@ -511,30 +584,34 @@ class Test_keyImport(unittest.TestCase):
self.assertFalse(is_private_key(priv_details['address']))
self.assertFalse(is_private_key("not a privkey"))
@needs_test_with_all_ecc_implementations
def test_serialize_privkey(self):
for priv_details in self.priv_pub_addr:
txin_type, privkey, compressed = deserialize_privkey(priv_details['priv'])
priv2 = serialize_privkey(privkey, compressed, txin_type)
self.assertEqual(priv_details['exported_privkey'], priv2)
@needs_test_with_all_ecc_implementations
def test_address_to_scripthash(self):
for priv_details in self.priv_pub_addr:
sh = address_to_scripthash(priv_details['address'])
self.assertEqual(priv_details['scripthash'], sh)
@needs_test_with_all_ecc_implementations
def test_is_minikey(self):
for priv_details in self.priv_pub_addr:
minikey = priv_details['minikey']
priv = priv_details['priv']
self.assertEqual(minikey, is_minikey(priv))
@needs_test_with_all_ecc_implementations
def test_is_compressed(self):
for priv_details in self.priv_pub_addr:
self.assertEqual(priv_details['compressed'],
is_compressed(priv_details['priv']))
class Test_seeds(unittest.TestCase):
class Test_seeds(SequentialTestCase):
""" Test old and new seeds. """
mnemonics = {

7
lib/tests/test_dnssec.py
View file

@ -1,11 +1,14 @@
import unittest
import dns
from lib import dnssec
from . import SequentialTestCase
from .test_bitcoin import needs_test_with_all_ecc_implementations
class TestDnsSec(unittest.TestCase):
class TestDnsSec(SequentialTestCase):
@needs_test_with_all_ecc_implementations
def test_python_validate_rrsig_ecdsa(self):
rrset = dns.rrset.from_text("getmonero.org.", 3599, 1, 48,
"257 3 13 mdsswUyr3DPW132mOi8V9xESWE8jTo0d xCjjnopKl+GqJxpVXckHAeF+KkxLbxIL fDLUT0rAK9iUzy1L53eKGQ==",

4
lib/tests/test_interface.py
View file

@ -2,8 +2,10 @@ import unittest
from lib import interface
from . import SequentialTestCase
class TestInterface(unittest.TestCase):
class TestInterface(SequentialTestCase):
def test_match_host_name(self):
self.assertTrue(interface._match_hostname('asd.fgh.com', 'asd.fgh.com'))

8
lib/tests/test_mnemonic.py
View file

@ -4,8 +4,10 @@ from lib import mnemonic
from lib import old_mnemonic
from lib.util import bh2u
from . import SequentialTestCase
class Test_NewMnemonic(unittest.TestCase):
class Test_NewMnemonic(SequentialTestCase):
def test_to_seed(self):
seed = mnemonic.Mnemonic.mnemonic_to_seed(mnemonic='foobar', passphrase='none')
@ -22,7 +24,7 @@ class Test_NewMnemonic(unittest.TestCase):
self.assertEqual(m.mnemonic_encode(i), seed)
class Test_OldMnemonic(unittest.TestCase):
class Test_OldMnemonic(SequentialTestCase):
def test(self):
seed = '8edad31a95e7d59f8837667510d75a4d'
@ -31,7 +33,7 @@ class Test_OldMnemonic(unittest.TestCase):
self.assertEqual(result, words.split())
self.assertEqual(old_mnemonic.mn_decode(result), seed)
class Test_BIP39Checksum(unittest.TestCase):
class Test_BIP39Checksum(SequentialTestCase):
def test(self):
mnemonic = u'gravity machine north sort system female filter attitude volume fold club stay feature office ecology stable narrow fog'

6
lib/tests/test_simple_config.py
View file

@ -8,8 +8,10 @@ import shutil
from io import StringIO
from lib.simple_config import (SimpleConfig, read_user_config)
from . import SequentialTestCase
class Test_SimpleConfig(unittest.TestCase):
class Test_SimpleConfig(SequentialTestCase):
def setUp(self):
super(Test_SimpleConfig, self).setUp()
@ -109,7 +111,7 @@ class Test_SimpleConfig(unittest.TestCase):
self.assertEqual({"something": "a"}, result)
class TestUserConfig(unittest.TestCase):
class TestUserConfig(SequentialTestCase):
def setUp(self):
super(TestUserConfig, self).setUp()

2
lib/tests/test_storage_upgrade.py
View file

@ -6,6 +6,8 @@ from lib.wallet import Wallet
from lib.tests.test_wallet import WalletTestCase
from . import SequentialTestCase
# TODO add other wallet types: 2fa, xpub-only
# TODO hw wallet with client version 2.6.x (single-, and multiacc)

10
lib/tests/test_transaction.py
View file

@ -5,12 +5,16 @@ from lib.bitcoin import TYPE_ADDRESS
from lib.keystore import xpubkey_to_address
from lib.util import bh2u, bfh
from . import SequentialTestCase
from .test_bitcoin import needs_test_with_all_ecc_implementations
unsigned_blob = '01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65bf38633b424eb4031000000005701ff4c53ff0488b21e03ef2afea18000000089689bff23e1e7fb2f161daa37270a97a3d8c2e537584b2d304ecb47b86d21fc021b010d3bd425f8cf2e04824bfdf1f1f5ff1d51fadd9a41f9e3fb8dd3403b1bfe00000000ffffffff0140420f00000000001976a914230ac37834073a42146f11ef8414ae929feaafc388ac00000000'
signed_blob = '01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65bf38633b424eb4031000000006c493046022100a82bbc57a0136751e5433f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7de89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d985012102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae35cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a42146f11ef8414ae929feaafc388ac00000000'
v2_blob = "0200000001191601a44a81e061502b7bfbc6eaa1cef6d1e6af5308ef96c9342f71dbf4b9b5000000006b483045022100a6d44d0a651790a477e75334adfb8aae94d6612d01187b2c02526e340a7fd6c8022028bdf7a64a54906b13b145cd5dab21a26bd4b85d6044e9b97bceab5be44c2a9201210253e8e0254b0c95776786e40984c1aa32a7d03efa6bdacdea5f421b774917d346feffffff026b20fa04000000001976a914024db2e87dd7cfd0e5f266c5f212e21a31d805a588aca0860100000000001976a91421919b94ae5cefcdf0271191459157cdb41c4cbf88aca6240700"
signed_segwit_blob = "01000000000101b66d722484f2db63e827ebf41d02684fed0c6550e85015a6c9d41ef216a8a6f00000000000fdffffff0280c3c90100000000160014b65ce60857f7e7892b983851c2a8e3526d09e4ab64bac30400000000160014c478ebbc0ab2097706a98e10db7cf101839931c4024730440220789c7d47f876638c58d98733c30ae9821c8fa82b470285dcdf6db5994210bf9f02204163418bbc44af701212ad42d884cc613f3d3d831d2d0cc886f767cca6e0235e012103083a6dc250816d771faa60737bfe78b23ad619f6b458e0a1f1688e3a0605e79c00000000"
class TestBCDataStream(unittest.TestCase):
class TestBCDataStream(SequentialTestCase):
def test_compact_size(self):
s = transaction.BCDataStream()
@ -51,8 +55,9 @@ class TestBCDataStream(unittest.TestCase):
self.assertEqual(s.read_bytes(4), b'r')
self.assertEqual(s.read_bytes(1), b'')
class TestTransaction(unittest.TestCase):
class TestTransaction(SequentialTestCase):
@needs_test_with_all_ecc_implementations
def test_tx_unsigned(self):
expected = {
'inputs': [{
@ -97,6 +102,7 @@ class TestTransaction(unittest.TestCase):
blob = str(tx)
self.assertEqual(transaction.deserialize(blob), expected)
@needs_test_with_all_ecc_implementations
def test_tx_signed(self):
expected = {
'inputs': [{

5
lib/tests/test_util.py
View file

@ -1,7 +1,10 @@
import unittest
from lib.util import format_satoshis, parse_URI
class TestUtil(unittest.TestCase):
from . import SequentialTestCase
class TestUtil(SequentialTestCase):
def test_format_satoshis(self):
result = format_satoshis(1234)

4
lib/tests/test_wallet.py
View file

@ -8,6 +8,8 @@ import json
from io import StringIO
from lib.storage import WalletStorage, FINAL_SEED_VERSION
from . import SequentialTestCase
class FakeSynchronizer(object):
@ -18,7 +20,7 @@ class FakeSynchronizer(object):
self.store.append(address)
class WalletTestCase(unittest.TestCase):
class WalletTestCase(SequentialTestCase):
def setUp(self):
super(WalletTestCase, self).setUp()

19
lib/tests/test_wallet_vertical.py
View file

@ -13,6 +13,8 @@ from lib.wallet import TX_HEIGHT_UNCONFIRMED, TX_HEIGHT_UNCONF_PARENT
from plugins.trustedcoin import trustedcoin
from . import TestCaseForTestnet
from . import SequentialTestCase
from .test_bitcoin import needs_test_with_all_ecc_implementations
class WalletIntegrityHelper:
@ -57,8 +59,9 @@ class WalletIntegrityHelper:
# TODO passphrase/seed_extension
class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
class TestWalletKeystoreAddressIntegrityForMainnet(SequentialTestCase):
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_seed_standard(self, mock_write):
seed_words = 'cycle rocket west magnet parrot shuffle foot correct salt library feed song'
@ -78,6 +81,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '1NNkttn1YvVGdqBW4PR6zvc3Zx3H5owKRf')
self.assertEqual(w.get_change_addresses()[0], '1KSezYMhAJMWqFbVFB2JshYg69UpmEXR4D')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_seed_segwit(self, mock_write):
seed_words = 'bitter grass shiver impose acquire brush forget axis eager alone wine silver'
@ -97,6 +101,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], 'bc1q3g5tmkmlvxryhh843v4dz026avatc0zzr6h3af')
self.assertEqual(w.get_change_addresses()[0], 'bc1qdy94n2q5qcp0kg7v9yzwe6wvfkhnvyzje7nx2p')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_seed_old(self, mock_write):
seed_words = 'powerful random nobody notice nothing important anyway look away hidden message over'
@ -115,6 +120,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '1FJEEB8ihPMbzs2SkLmr37dHyRFzakqUmo')
self.assertEqual(w.get_change_addresses()[0], '1KRW8pH6HFHZh889VDq6fEKvmrsmApwNfe')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_seed_2fa(self, mock_write):
seed_words = 'kiss live scene rude gate step hip quarter bunker oxygen motor glove'
@ -148,6 +154,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '35L8XmCDoEBKeaWRjvmZvoZvhp8BXMMMPV')
self.assertEqual(w.get_change_addresses()[0], '3PeZEcumRqHSPNN43hd4yskGEBdzXgY8Cy')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_bip39_seed_bip44_standard(self, mock_write):
seed_words = 'treat dwarf wealth gasp brass outside high rent blood crowd make initial'
@ -166,6 +173,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '16j7Dqk3Z9DdTdBtHcCVLaNQy9MTgywUUo')
self.assertEqual(w.get_change_addresses()[0], '1GG5bVeWgAp5XW7JLCphse14QaC4qiHyWn')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_bip39_seed_bip49_p2sh_segwit(self, mock_write):
seed_words = 'treat dwarf wealth gasp brass outside high rent blood crowd make initial'
@ -184,6 +192,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '35ohQTdNykjkF1Mn9nAVEFjupyAtsPAK1W')
self.assertEqual(w.get_change_addresses()[0], '3KaBTcviBLEJajTEMstsA2GWjYoPzPK7Y7')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_bip39_seed_bip84_native_segwit(self, mock_write):
# test case from bip84
@ -203,6 +212,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], 'bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu')
self.assertEqual(w.get_change_addresses()[0], 'bc1q8c6fshw2dlwun7ekn9qwf37cu2rn755upcp6el')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_multisig_seed_standard(self, mock_write):
seed_words = 'blast uniform dragon fiscal ensure vast young utility dinosaur abandon rookie sure'
@ -225,6 +235,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '32ji3QkAgXNz6oFoRfakyD3ys1XXiERQYN')
self.assertEqual(w.get_change_addresses()[0], '36XWwEHrrVCLnhjK5MrVVGmUHghr9oWTN1')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_electrum_multisig_seed_segwit(self, mock_write):
seed_words = 'snow nest raise royal more walk demise rotate smooth spirit canyon gun'
@ -247,6 +258,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], 'bc1qvzezdcv6vs5h45ugkavp896e0nde5c5lg5h0fwe2xyfhnpkxq6gq7pnwlc')
self.assertEqual(w.get_change_addresses()[0], 'bc1qxqf840dqswcmu7a8v82fj6ej0msx08flvuy6kngr7axstjcaq6us9hrehd')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_bip39_multisig_seed_bip45_standard(self, mock_write):
seed_words = 'treat dwarf wealth gasp brass outside high rent blood crowd make initial'
@ -269,6 +281,7 @@ class TestWalletKeystoreAddressIntegrityForMainnet(unittest.TestCase):
self.assertEqual(w.get_receiving_addresses()[0], '3JPTQ2nitVxXBJ1yhMeDwH6q417UifE3bN')
self.assertEqual(w.get_change_addresses()[0], '3FGyDuxgUDn2pSZe5xAJH1yUwSdhzDMyEE')
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_bip39_multisig_seed_p2sh_segwit(self, mock_write):
# bip39 seed: pulse mixture jazz invite dune enrich minor weapon mosquito flight fly vapor
@ -332,6 +345,7 @@ class TestWalletSending(TestCaseForTestnet):
ks = keystore.from_seed(seed_words, '', False)
return WalletIntegrityHelper.create_standard_wallet(ks, gap_limit=2)
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_sending_between_p2wpkh_and_compressed_p2pkh(self, mock_write):
wallet1 = self.create_standard_wallet_from_seed('bitter grass shiver impose acquire brush forget axis eager alone wine silver')
@ -382,6 +396,7 @@ class TestWalletSending(TestCaseForTestnet):
self.assertEqual((0, funding_output_value - 250000 - 5000 + 100000, 0), wallet1.get_balance())
self.assertEqual((0, 250000 - 5000 - 100000, 0), wallet2.get_balance())
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_sending_between_p2sh_2of3_and_uncompressed_p2pkh(self, mock_write):
wallet1a = WalletIntegrityHelper.create_multisig_wallet(
@ -451,6 +466,7 @@ class TestWalletSending(TestCaseForTestnet):
self.assertEqual((0, funding_output_value - 370000 - 5000 + 100000, 0), wallet1a.get_balance())
self.assertEqual((0, 370000 - 5000 - 100000, 0), wallet2.get_balance())
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_sending_between_p2wsh_2of3_and_p2wsh_p2sh_2of2(self, mock_write):
wallet1a = WalletIntegrityHelper.create_multisig_wallet(
@ -538,6 +554,7 @@ class TestWalletSending(TestCaseForTestnet):
self.assertEqual((0, funding_output_value - 165000 - 5000 + 100000, 0), wallet1a.get_balance())
self.assertEqual((0, 165000 - 5000 - 100000, 0), wallet2a.get_balance())
@needs_test_with_all_ecc_implementations
@mock.patch.object(storage.WalletStorage, '_write')
def test_sending_between_p2sh_1of2_and_p2wpkh_p2sh(self, mock_write):
wallet1a = WalletIntegrityHelper.create_multisig_wallet(

34
lib/transaction.py
View file

@ -31,6 +31,7 @@ from typing import Sequence, Union
from .util import print_error, profiler
from . import ecc
from . import bitcoin
from .bitcoin import *
import struct
@ -653,18 +654,18 @@ class Transaction:
if sig in sigs1:
continue
pre_hash = Hash(bfh(self.serialize_preimage(i)))
# der to string
order = ecdsa.ecdsa.generator_secp256k1.order()
r, s = ecdsa.util.sigdecode_der(bfh(sig[:-2]), order)
sig_string = ecdsa.util.sigencode_string(r, s, order)
compressed = True
sig_string = ecc.sig_string_from_der_sig(bfh(sig[:-2]))
for recid in range(4):
public_key = MyVerifyingKey.from_signature(sig_string, recid, pre_hash, curve = SECP256k1)
pubkey = bh2u(point_to_ser(public_key.pubkey.point, compressed))
if pubkey in pubkeys:
public_key.verify_digest(sig_string, pre_hash, sigdecode = ecdsa.util.sigdecode_string)
j = pubkeys.index(pubkey)
print_error("adding sig", i, j, pubkey, sig)
try:
public_key = ecc.ECPubkey.from_sig_string(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:
public_key.verify_message_hash(sig_string, pre_hash)
j = pubkeys.index(pubkey_hex)
print_error("adding sig", i, j, pubkey_hex, sig)
self.add_signature_to_txin(self._inputs[i], j, sig)
#self._inputs[i]['x_pubkeys'][j] = pubkey
break
@ -1067,7 +1068,7 @@ class Transaction:
if x_pubkey in keypairs.keys():
print_error("adding signature for", x_pubkey)
sec, compressed = keypairs.get(x_pubkey)
pubkey = public_key_from_private_key(sec, compressed)
pubkey = ecc.ECPrivkey(sec).get_public_key_hex(compressed=compressed)
# add signature
sig = self.sign_txin(i, sec)
self.add_signature_to_txin(txin, j, sig)
@ -1079,13 +1080,8 @@ class Transaction:
def sign_txin(self, txin_index, privkey_bytes):
pre_hash = Hash(bfh(self.serialize_preimage(txin_index)))
pkey = regenerate_key(privkey_bytes)
secexp = pkey.secret
private_key = bitcoin.MySigningKey.from_secret_exponent(secexp, curve=SECP256k1)
public_key = private_key.get_verifying_key()
sig = private_key.sign_digest_deterministic(pre_hash, hashfunc=hashlib.sha256, sigencode=ecdsa.util.sigencode_der)
if not public_key.verify_digest(sig, pre_hash, sigdecode=ecdsa.util.sigdecode_der):
raise Exception('Sanity check verifying our own signature failed.')
privkey = ecc.ECPrivkey(privkey_bytes)
sig = privkey.sign_transaction(pre_hash)
sig = bh2u(sig) + '01'
return sig

2
lib/wallet.py
View file

@ -114,7 +114,7 @@ def append_utxos_to_inputs(inputs, network, pubkey, txin_type, imax):
def sweep_preparations(privkeys, network, imax=100):
def find_utxos_for_privkey(txin_type, privkey, compressed):
pubkey = bitcoin.public_key_from_private_key(privkey, compressed)
pubkey = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)
append_utxos_to_inputs(inputs, network, pubkey, txin_type, imax)
keypairs[pubkey] = privkey, compressed
inputs = []

9
plugins/cosigner_pool/qt.py
View file

@ -30,7 +30,7 @@ from PyQt5.QtGui import *
from PyQt5.QtCore import *
from PyQt5.QtWidgets import QPushButton
from electrum import bitcoin, util, keystore
from electrum import bitcoin, util, keystore, ecc
from electrum import transaction
from electrum.plugins import BasePlugin, hook
from electrum.i18n import _
@ -174,7 +174,8 @@ class Plugin(BasePlugin):
if not self.cosigner_can_sign(tx, xpub):
continue
raw_tx_bytes = bfh(str(tx))
message = bitcoin.encrypt_message(raw_tx_bytes, bh2u(K)).decode('ascii')
public_key = ecc.ECPubkey(K)
message = public_key.encrypt_message(raw_tx_bytes).decode('ascii')
try:
server.put(_hash, message)
except Exception as e:
@ -214,8 +215,8 @@ class Plugin(BasePlugin):
if not xprv:
return
try:
k = bh2u(bitcoin.deserialize_xprv(xprv)[-1])
EC = bitcoin.EC_KEY(bfh(k))
k = bitcoin.deserialize_xprv(xprv)[-1]
EC = ecc.ECPrivkey(k)
message = bh2u(EC.decrypt_message(message))
except Exception as e:
traceback.print_exc(file=sys.stdout)

34
plugins/digitalbitbox/digitalbitbox.py
View file

@ -5,8 +5,11 @@
try:
import electrum
from electrum.bitcoin import TYPE_ADDRESS, push_script, var_int, msg_magic, Hash, verify_message, pubkey_from_signature, point_to_ser, public_key_to_p2pkh, EncodeAES, DecodeAES, MyVerifyingKey, is_address
from electrum.bitcoin import serialize_xpub, deserialize_xpub
from electrum.crypto import Hash, EncodeAES, DecodeAES
from electrum.bitcoin import (TYPE_ADDRESS, push_script, var_int, public_key_to_p2pkh, is_address,
serialize_xpub, deserialize_xpub)
from electrum import ecc
from electrum.ecc import msg_magic
from electrum.wallet import Standard_Wallet
from electrum import constants
from electrum.transaction import Transaction
@ -27,9 +30,6 @@ try:
import base64
import os
import sys
from ecdsa.ecdsa import generator_secp256k1
from ecdsa.util import sigencode_der
from ecdsa.curves import SECP256k1
DIGIBOX = True
except ImportError as e:
DIGIBOX = False
@ -476,19 +476,21 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
if 'recid' in reply['sign'][0]:
# firmware > v2.1.1
sig = bytes([27 + int(reply['sign'][0]['recid'], 16) + 4]) + binascii.unhexlify(reply['sign'][0]['sig'])
pk, compressed = pubkey_from_signature(sig, msg_hash)
pk = point_to_ser(pk.pubkey.point, compressed)
addr = public_key_to_p2pkh(pk)
if verify_message(addr, sig, message) is False:
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 = ecc.ECPubkey.from_signature65(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:
raise Exception(_("Could not sign message"))
elif 'pubkey' in reply['sign'][0]:
# firmware <= v2.1.1
for i in range(4):
sig = bytes([27 + i + 4]) + binascii.unhexlify(reply['sign'][0]['sig'])
for recid in range(4):
sig_string = binascii.unhexlify(reply['sign'][0]['sig'])
sig = ecc.construct_sig65(sig_string, recid, True)
try:
addr = public_key_to_p2pkh(binascii.unhexlify(reply['sign'][0]['pubkey']))
if verify_message(addr, sig, message):
if ecc.verify_message_with_address(addr, sig, message):
break
except Exception:
continue
@ -634,8 +636,8 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
recid = int(signed['recid'], 16)
s = binascii.unhexlify(signed['sig'])
h = inputhasharray[i]
pk = MyVerifyingKey.from_signature(s, recid, h, curve = SECP256k1)
pk = to_hexstr(point_to_ser(pk.pubkey.point, True))
pk = ecc.ECPubkey.from_sig_string(s, recid, h)
pk = pk.get_public_key_hex(compressed=True)
elif 'pubkey' in signed:
# firmware <= v2.1.1
pk = signed['pubkey']
@ -643,7 +645,7 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
continue
sig_r = int(signed['sig'][:64], 16)
sig_s = int(signed['sig'][64:], 16)
sig = sigencode_der(sig_r, sig_s, generator_secp256k1.order())
sig = ecc.der_sig_from_r_and_s(sig_r, sig_s)
sig = to_hexstr(sig) + '01'
Transaction.add_signature_to_txin(txin, ii, sig)
tx._inputs[i] = txin

4
plugins/trustedcoin/trustedcoin.py
View file

@ -31,7 +31,7 @@ from urllib.parse import urljoin
from urllib.parse import quote
import electrum
from electrum import bitcoin
from electrum import bitcoin, ecc
from electrum import constants
from electrum import keystore
from electrum.bitcoin import *
@ -593,7 +593,7 @@ class TrustedCoinPlugin(BasePlugin):
def f(xprv):
_, _, _, _, c, k = deserialize_xprv(xprv)
pk = bip32_private_key([0, 0], k, c)
key = regenerate_key(pk)
key = ecc.ECPrivkey(pk)
sig = key.sign_message(message, True)
return base64.b64encode(sig).decode()

11
setup.py
View file

@ -40,13 +40,18 @@ if platform.system() in ['Linux', 'FreeBSD', 'DragonFly']:
(os.path.join(usr_share, icons_dirname), ['icons/electrum.png'])
]
extras_require = {
'hardware': requirements_hw,
'fast': ['pycryptodomex'],
}
extras_require['full'] = extras_require['hardware'] + extras_require['fast']
setup(
name="Electrum",
version=version.ELECTRUM_VERSION,
install_requires=requirements,
extras_require={
'full': requirements_hw + ['pycryptodomex'],
},
extras_require=extras_require,
packages=[
'electrum',
'electrum_gui',

2
tox.ini
View file

@ -8,3 +8,5 @@ deps=
commands=
coverage run --source=lib -m py.test -v
coverage report
extras=
fast