tMerge pull request #4381 from SomberNight/coincurve4 - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
DIR commit 770f438249078b598964d44b72242fc64e4ece75
DIR parent 756cc323e795cc9fcadceaa6eca54cf3d77c823b
HTML Author: ThomasV <thomasv@electrum.org>
Date: Fri, 25 May 2018 18:49:36 +0200
Merge pull request #4381 from SomberNight/coincurve4
crypto refactoring take3
Diffstat:
M .travis.yml | 8 +++++++-
M contrib/build-wine/README.md | 3 ++-
A contrib/build-wine/build-secp256k1… | 33 +++++++++++++++++++++++++++++++
M contrib/build-wine/build.sh | 2 ++
M contrib/build-wine/deterministic.s… | 2 ++
M contrib/build-wine/prepare-wine.sh | 3 +++
M gui/qt/main_window.py | 7 ++++---
M lib/bitcoin.py | 442 +++----------------------------
M lib/commands.py | 14 ++++++++------
A lib/crypto.py | 142 +++++++++++++++++++++++++++++++
A lib/ecc.py | 407 +++++++++++++++++++++++++++++++
A lib/ecc_fast.py | 216 +++++++++++++++++++++++++++++++
M lib/keystore.py | 33 +++++++++++++++----------------
M lib/paymentrequest.py | 10 +++++-----
M lib/storage.py | 23 +++++++++++++----------
M lib/tests/__init__.py | 17 ++++++++++++++++-
M lib/tests/test_bitcoin.py | 135 ++++++++++++++++++++++++-------
M lib/tests/test_dnssec.py | 7 +++++--
M lib/tests/test_interface.py | 4 +++-
M lib/tests/test_mnemonic.py | 8 +++++---
M lib/tests/test_simple_config.py | 6 ++++--
M lib/tests/test_storage_upgrade.py | 2 ++
M lib/tests/test_transaction.py | 10 ++++++++--
M lib/tests/test_util.py | 5 ++++-
M lib/tests/test_wallet.py | 4 +++-
M lib/tests/test_wallet_vertical.py | 19 ++++++++++++++++++-
M lib/transaction.py | 34 ++++++++++++++-----------------
M lib/wallet.py | 2 +-
M plugins/cosigner_pool/qt.py | 9 +++++----
M plugins/digitalbitbox/digitalbitbo… | 34 ++++++++++++++++---------------
M plugins/trustedcoin/trustedcoin.py | 4 ++--
M setup.py | 11 ++++++++---
M tox.ini | 2 ++
33 files changed, 1118 insertions(+), 540 deletions(-)
---
DIR diff --git a/.travis.yml b/.travis.yml
t@@ -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:
t@@ -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
DIR diff --git a/contrib/build-wine/README.md b/contrib/build-wine/README.md
t@@ -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)
```
DIR diff --git a/contrib/build-wine/build-secp256k1.sh b/contrib/build-wine/build-secp256k1.sh
t@@ -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"
DIR diff --git a/contrib/build-wine/build.sh b/contrib/build-wine/build.sh
t@@ -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..."
DIR diff --git a/contrib/build-wine/deterministic.spec b/contrib/build-wine/deterministic.spec
t@@ -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'),
DIR diff --git a/contrib/build-wine/prepare-wine.sh b/contrib/build-wine/prepare-wine.sh
t@@ -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."
DIR diff --git a/gui/qt/main_window.py b/gui/qt/main_window.py
t@@ -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
t@@ -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:
t@@ -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)
DIR diff --git a/lib/bitcoin.py b/lib/bitcoin.py
t@@ -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
t@@ -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])
t@@ -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()
t@@ -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)
t@@ -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):
t@@ -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):
t@@ -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:
t@@ -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
t@@ -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):
t@@ -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)
t@@ -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
t@@ -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
t@@ -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
t@@ -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
t@@ -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)
t@@ -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)
t@@ -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
DIR diff --git a/lib/commands.py b/lib/commands.py
t@@ -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
t@@ -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]
t@@ -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:
t@@ -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
t@@ -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):
DIR diff --git a/lib/crypto.py b/lib/crypto.py
t@@ -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()
DIR diff --git a/lib/ecc.py b/lib/ecc.py
t@@ -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
DIR diff --git a/lib/ecc_fast.py b/lib/ecc_fast.py
t@@ -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()
DIR diff --git a/lib/keystore.py b/lib/keystore.py
t@@ -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)
t@@ -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
t@@ -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)
t@@ -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
t@@ -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):
t@@ -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):
t@@ -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()
DIR diff --git a/lib/paymentrequest.py b/lib/paymentrequest.py
t@@ -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
t@@ -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:
t@@ -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):
DIR diff --git a/lib/storage.py b/lib/storage.py
t@@ -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
t@@ -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):
t@@ -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)
t@@ -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):
t@@ -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
t@@ -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())
DIR diff --git a/lib/tests/__init__.py b/lib/tests/__init__.py
t@@ -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):
DIR diff --git a/lib/tests/test_bitcoin.py b/lib/tests/test_bitcoin.py
t@@ -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
t@@ -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)
t@@ -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(
t@@ -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.assertFalse(verify_message(addr1, b'wrong', msg1))
- self.assertFalse(verify_message(addr1, sig2, msg1))
+ self.assertTrue(ecc.verify_message_with_address(addr1, sig1, msg1))
+ self.assertTrue(ecc.verify_message_with_address(addr2, sig2, msg2))
+
+ self.assertFalse(ecc.verify_message_with_address(addr1, b'wrong', msg1))
+ self.assertFalse(ecc.verify_message_with_address(addr1, sig2, msg1))
+
+ @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=='))
parazyd.org:70 /git/electrum/commit/770f438249078b598964d44b72242fc64e4ece75.gph:1969: line too long