Get Decred account key from seed.

This commit is contained in:
tecnovert 2024-04-22 20:59:22 +02:00
parent 89ca350ff2
commit f1822e1443
14 changed files with 330 additions and 32 deletions

View file

@ -502,7 +502,9 @@ class BasicSwap(BaseApp):
self.coin_clients[Coins.LTC_MWEB] = self.coin_clients[coin] self.coin_clients[Coins.LTC_MWEB] = self.coin_clients[coin]
if self.coin_clients[coin]['connection_type'] == 'rpc': if self.coin_clients[coin]['connection_type'] == 'rpc':
if coin == Coins.XMR: if coin == Coins.DCR:
self.coin_clients[coin]['walletrpcport'] = chain_client_settings['walletrpcport']
elif coin == Coins.XMR:
self.coin_clients[coin]['rpctimeout'] = chain_client_settings.get('rpctimeout', 60) self.coin_clients[coin]['rpctimeout'] = chain_client_settings.get('rpctimeout', 60)
self.coin_clients[coin]['walletrpctimeout'] = chain_client_settings.get('walletrpctimeout', 120) self.coin_clients[coin]['walletrpctimeout'] = chain_client_settings.get('walletrpctimeout', 120)
self.coin_clients[coin]['walletrpctimeoutlong'] = chain_client_settings.get('walletrpctimeoutlong', 600) self.coin_clients[coin]['walletrpctimeoutlong'] = chain_client_settings.get('walletrpctimeoutlong', 600)

View file

@ -38,6 +38,7 @@ from basicswap.util.address import (
pubkeyToAddress, pubkeyToAddress,
) )
from basicswap.util.crypto import ( from basicswap.util.crypto import (
hash160,
sha256, sha256,
) )
from coincurve.keys import ( from coincurve.keys import (
@ -70,8 +71,7 @@ from basicswap.contrib.test_framework.script import (
OP_DROP, OP_DROP,
OP_HASH160, OP_EQUAL, OP_HASH160, OP_EQUAL,
SIGHASH_ALL, SIGHASH_ALL,
SegwitV0SignatureHash, SegwitV0SignatureHash)
hash160)
from basicswap.basicswap_util import ( from basicswap.basicswap_util import (
TxLockTypes) TxLockTypes)

View file

@ -5,6 +5,8 @@
# Distributed under the MIT software license, see the accompanying # Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php. # file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import logging
from basicswap.chainparams import Coins from basicswap.chainparams import Coins
from basicswap.interface.btc import Secp256k1Interface from basicswap.interface.btc import Secp256k1Interface
from basicswap.util.address import ( from basicswap.util.address import (
@ -13,8 +15,10 @@ from basicswap.util.address import (
) )
from basicswap.util.crypto import ( from basicswap.util.crypto import (
blake256, blake256,
hash160,
ripemd160, ripemd160,
) )
from basicswap.util.extkey import ExtKeyPair
from basicswap.interface.dcr.rpc import make_rpc_func from basicswap.interface.dcr.rpc import make_rpc_func
@ -45,7 +49,15 @@ class DCRInterface(Secp256k1Interface):
self._rpc_host = coin_settings.get('rpchost', '127.0.0.1') self._rpc_host = coin_settings.get('rpchost', '127.0.0.1')
self._rpcport = coin_settings['rpcport'] self._rpcport = coin_settings['rpcport']
self._rpcauth = coin_settings['rpcauth'] self._rpcauth = coin_settings['rpcauth']
self._sc = swap_client
self._log = self._sc.log if self._sc and self._sc.log else logging
self.rpc = make_rpc_func(self._rpcport, self._rpcauth, host=self._rpc_host) self.rpc = make_rpc_func(self._rpcport, self._rpcauth, host=self._rpc_host)
if 'walletrpcport' in coin_settings:
self.rpc_wallet = make_rpc_func(coin_settings['walletrpcport'], self._rpcauth, host=self._rpc_host)
else:
self.rpc_wallet = None
self._use_segwit = coin_settings['use_segwit']
def pkh(self, pubkey: bytes) -> bytes: def pkh(self, pubkey: bytes) -> bytes:
return ripemd160(blake256(pubkey)) return ripemd160(blake256(pubkey))
@ -69,6 +81,41 @@ class DCRInterface(Secp256k1Interface):
def testDaemonRPC(self, with_wallet=True) -> None: def testDaemonRPC(self, with_wallet=True) -> None:
if with_wallet: if with_wallet:
self.rpc_wallet('getwalletinfo') self.rpc_wallet('getinfo')
else: else:
self.rpc('getblockchaininfo') self.rpc('getblockchaininfo')
def checkWallets(self) -> int:
# Only one wallet possible?
return 1
def initialiseWallet(self, key: bytes) -> None:
# Load with --create
pass
def getDaemonVersion(self):
return self.rpc('getnetworkinfo')['version']
def getBlockchainInfo(self):
return self.rpc('getblockchaininfo')
def using_segwit(self) -> bool:
return self._use_segwit
def getWalletInfo(self):
rv = self.rpc_wallet('getinfo')
return rv
def getSeedHash(self, seed: bytes) -> bytes:
# m / purpose' / coin_type' / account' / change / address_index
# m/44'/coin_type'/0'/0/0
ek = ExtKeyPair(self.coin_type())
ek.set_seed(seed)
coin_type = self.chainparams_network()['bip44']
ek_purpose = ek.derive(44 | (1 << 31))
ek_coin = ek_purpose.derive(coin_type | (1 << 31))
ek_account = ek_coin.derive(0 | (1 << 31))
return hash160(ek_account.encode_p())

View file

@ -16,7 +16,6 @@ def callrpc(rpc_port, auth, method, params=[], host='127.0.0.1'):
x.__handler = None x.__handler = None
v = x.json_request(method, params) v = x.json_request(method, params)
x.close() x.close()
print('[rm] v', v)
r = json.loads(v.decode('utf-8')) r = json.loads(v.decode('utf-8'))
except Exception as ex: except Exception as ex:
traceback.print_exc() traceback.print_exc()

View file

@ -76,11 +76,11 @@ class NAVInterface(BTCInterface):
# p2sh-p2wsh # p2sh-p2wsh
return True return True
def seedToMnemonic(self, key): def seedToMnemonic(self, key: bytes) -> None:
return Mnemonic('english').to_mnemonic(key) return Mnemonic('english').to_mnemonic(key)
def initialiseWallet(self, key): def initialiseWallet(self, key):
# load with -importmnemonic= parameter # Load with -importmnemonic= parameter
pass pass
def getWalletSeedID(self): def getWalletSeedID(self):

View file

@ -93,7 +93,7 @@ class PARTInterface(BTCInterface):
index_info = self.rpc('getinsightinfo' if int(str(version)[:2]) > 19 else 'getindexinfo') index_info = self.rpc('getinsightinfo' if int(str(version)[:2]) > 19 else 'getindexinfo')
return index_info['spentindex'] return index_info['spentindex']
def initialiseWallet(self, key): def initialiseWallet(self, key: bytes) -> None:
raise ValueError('TODO') raise ValueError('TODO')
def withdrawCoin(self, value, addr_to, subfee): def withdrawCoin(self, value, addr_to, subfee):

View file

@ -156,7 +156,7 @@ class XMRInterface(CoinInterface):
pass pass
self.rpc_wallet('open_wallet', params) self.rpc_wallet('open_wallet', params)
def initialiseWallet(self, key_view, key_spend, restore_height=None): def initialiseWallet(self, key_view: bytes, key_spend: bytes, restore_height=None) -> None:
with self._mx_wallet: with self._mx_wallet:
try: try:
self.openWallet(self._wallet_filename) self.openWallet(self._wallet_filename)

View file

@ -90,7 +90,7 @@ def getKeyID(key_data: bytes) -> bytes:
return ripemd160(sha256(key_data)) return ripemd160(sha256(key_data))
def bech32Decode(hrp, addr): def bech32Decode(hrp: str, addr: str) -> bytes:
hrpgot, data = bech32_decode(addr) hrpgot, data = bech32_decode(addr)
if hrpgot != hrp: if hrpgot != hrp:
return None return None
@ -100,14 +100,14 @@ def bech32Decode(hrp, addr):
return bytes(decoded) return bytes(decoded)
def bech32Encode(hrp, data): def bech32Encode(hrp: str, data: bytes) -> str:
ret = bech32_encode(hrp, convertbits(data, 8, 5)) ret = bech32_encode(hrp, convertbits(data, 8, 5))
if bech32Decode(hrp, ret) is None: if bech32Decode(hrp, ret) is None:
return None return None
return ret return ret
def decodeAddress(address: str): def decodeAddress(address: str) -> bytes:
addr_data = b58decode(address) addr_data = b58decode(address)
if addr_data is None: if addr_data is None:
return None return None

View file

@ -4,9 +4,10 @@
# Distributed under the MIT software license, see the accompanying # Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php. # file LICENSE or http://www.opensource.org/licenses/mit-license.php.
from Crypto.Hash import RIPEMD160, SHA256 # pycryptodome
from basicswap.contrib.blake256.blake256 import blake_hash from basicswap.contrib.blake256.blake256 import blake_hash
from Crypto.Hash import HMAC, RIPEMD160, SHA256, SHA512 # pycryptodome
def sha256(data: bytes) -> bytes: def sha256(data: bytes) -> bytes:
h = SHA256.new() h = SHA256.new()
@ -14,6 +15,12 @@ def sha256(data: bytes) -> bytes:
return h.digest() return h.digest()
def sha512(data: bytes) -> bytes:
h = SHA512.new()
h.update(data)
return h.digest()
def ripemd160(data: bytes) -> bytes: def ripemd160(data: bytes) -> bytes:
h = RIPEMD160.new() h = RIPEMD160.new()
h.update(data) h.update(data)
@ -24,5 +31,11 @@ def blake256(data: bytes) -> bytes:
return blake_hash(data) return blake_hash(data)
def hash160(s: bytes) -> bytes: def hash160(data: bytes) -> bytes:
return ripemd160(sha256(s)) return ripemd160(sha256(data))
def hmac_sha512(secret: bytes, data: bytes) -> bytes:
h = HMAC.new(secret, digestmod=SHA512)
h.update(data)
return h.digest()

116
basicswap/util/extkey.py Normal file
View file

@ -0,0 +1,116 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2024 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
from .crypto import blake256, hash160, hmac_sha512, ripemd160
from coincurve.keys import (
PrivateKey,
PublicKey)
def BIP32Hash(chaincode: bytes, child_no: int, key_data_type: int, keydata: bytes):
return hmac_sha512(chaincode, key_data_type.to_bytes(1) + keydata + child_no.to_bytes(4, 'big'))
def hash160_dcr(data: bytes) -> bytes:
return ripemd160(blake256(data))
class ExtKeyPair():
__slots__ = ('_depth', '_fingerprint', '_child_no', '_chaincode', '_key', '_pubkey', 'hash_func')
def __init__(self, coin_type=1):
if coin_type == 4:
self.hash_func = hash160_dcr
else:
self.hash_func = hash160
def set_seed(self, seed: bytes) -> None:
hashout: bytes = hmac_sha512(b'Bitcoin seed', seed)
self._key = hashout[:32]
self._pubkey = None
self._chaincode = hashout[32:]
self._depth = 0
self._child_no = 0
self._fingerprint = b'\0' * 4
def has_key(self) -> bool:
return False if self._key is None else True
def neuter(self) -> None:
if self._key is None:
raise ValueError('Already neutered')
self._pubkey = PublicKey.from_secret(self._key).format()
self._key = None
def derive(self, child_no: int):
out = ExtKeyPair()
out._depth = self._depth + 1
out._child_no = child_no
if (child_no >> 31) == 0:
if self._key:
K = PublicKey.from_secret(self._key)
k_encoded = K.format()
else:
K = PublicKey(self._pubkey)
k_encoded = self._pubkey
out._fingerprint = self.hash_func(k_encoded)[:4]
new_hash = BIP32Hash(self._chaincode, child_no, k_encoded[0], k_encoded[1:])
out._chaincode = new_hash[32:]
if self._key:
k = PrivateKey(self._key)
k.add(new_hash[:32], update=True)
out._key = k.secret
out._pubkey = None
else:
K.add(new_hash[:32], update=True)
out._key = None
out._pubkey = K.format()
else:
k = PrivateKey(self._key)
out._fingerprint = self.hash_func(self._pubkey if self._pubkey else PublicKey.from_secret(self._key).format())[:4]
new_hash = BIP32Hash(self._chaincode, child_no, 0, self._key)
out._chaincode = new_hash[32:]
k.add(new_hash[:32], update=True)
out._key = k.secret
out._pubkey = None
out.hash_func = self.hash_func
return out
def encode_v(self) -> bytes:
return self._depth.to_bytes(1) + \
self._fingerprint + \
self._child_no.to_bytes(4, 'big') + \
self._chaincode + \
b'\x00' + \
self._key
def encode_p(self) -> bytes:
pubkey = PublicKey.from_secret(self._key).format() if self._pubkey is None else self._pubkey
return self._depth.to_bytes(1) + \
self._fingerprint + \
self._child_no.to_bytes(4, 'big') + \
self._chaincode + \
pubkey
def decode(self, data: bytes) -> None:
if len(data) != 74:
raise ValueError('Unexpected extkey length')
self._depth = data[0]
self._fingerprint = data[1:5]
self._child_no = int.from_bytes(data[5:9], 'big')
self._chaincode = data[9:41]
if data[41] == 0:
self._key = data[42:]
self._pubkey = None
else:
self._key = None
self._pubkey = data[41:]

View file

@ -91,6 +91,8 @@ known_coins = {
disabled_coins = [ disabled_coins = [
'navcoin', 'navcoin',
'namecoin', # Needs update
'decred', # In-progress
] ]
expected_key_ids = { expected_key_ids = {

View file

@ -81,6 +81,7 @@ def startDaemon(node_dir, bin_dir, daemon_bin, opts=[], extra_config={}):
args.append('-datadir=' + datadir_path) args.append('-datadir=' + datadir_path)
args += opts args += opts
logging.info('Starting node ' + daemon_bin + ' ' + (('-datadir=' + node_dir) if add_datadir else '')) logging.info('Starting node ' + daemon_bin + ' ' + (('-datadir=' + node_dir) if add_datadir else ''))
logging.info('[rm] {}'.format(' '.join(args)))
opened_files = [] opened_files = []
if extra_config.get('stdout_to_file', False): if extra_config.get('stdout_to_file', False):

View file

@ -7,6 +7,8 @@
import logging import logging
import os import os
import subprocess
import select
import unittest import unittest
import basicswap.config as cfg import basicswap.config as cfg
@ -14,6 +16,7 @@ import basicswap.config as cfg
from basicswap.basicswap import ( from basicswap.basicswap import (
Coins, Coins,
) )
from basicswap.util.crypto import hash160
from basicswap.interface.dcr.rpc import ( from basicswap.interface.dcr.rpc import (
callrpc, callrpc,
) )
@ -38,6 +41,7 @@ DCR_CLI = os.getenv('DCR_CLI', 'dcrctl' + cfg.bin_suffix)
DCR_BASE_PORT = 44932 DCR_BASE_PORT = 44932
DCR_BASE_RPC_PORT = 45932 DCR_BASE_RPC_PORT = 45932
DCR_BASE_WALLET_RPC_PORT = 45952
def make_rpc_func(node_id, base_rpc_port): def make_rpc_func(node_id, base_rpc_port):
@ -50,28 +54,43 @@ def make_rpc_func(node_id, base_rpc_port):
return rpc_func return rpc_func
def prepareDCDDataDir(datadir, node_id, conf_file, dir_prefix, base_p2p_port, base_rpc_port, num_nodes=3): def prepareDCDDataDir(datadir, node_id, conf_file, dir_prefix, num_nodes=3):
node_dir = os.path.join(datadir, dir_prefix + str(node_id)) node_dir = os.path.join(datadir, dir_prefix + str(node_id))
if not os.path.exists(node_dir): if not os.path.exists(node_dir):
os.makedirs(node_dir) os.makedirs(node_dir)
cfg_file_path = os.path.join(node_dir, conf_file) cfg_file_path = os.path.join(node_dir, conf_file)
if os.path.exists(cfg_file_path): if os.path.exists(cfg_file_path):
return return
config = [
'simnet=1\n',
'debuglevel=debug\n',
f'listen=127.0.0.1:{DCR_BASE_PORT + node_id}\n',
f'rpclisten=127.0.0.1:{DCR_BASE_RPC_PORT + node_id}\n',
f'rpcuser=test{node_id}\n',
f'rpcpass=test_pass{node_id}\n',
'notls=1\n',]
for i in range(0, num_nodes):
if node_id == i:
continue
config.append('addpeer=127.0.0.1:{}\n'.format(DCR_BASE_PORT + i))
with open(cfg_file_path, 'w+') as fp: with open(cfg_file_path, 'w+') as fp:
config = [ for line in config:
'regnet=1\n', # or simnet? fp.write(line)
'debuglevel=debug\n',
f'listen=127.0.0.1:{base_p2p_port + node_id}\n',
f'rpclisten=127.0.0.1:{base_rpc_port + node_id}\n',
f'rpcuser=test{node_id}\n',
f'rpcpass=test_pass{node_id}\n',
'notls=1\n',]
for i in range(0, num_nodes): config = [
if node_id == i: 'simnet=1\n',
continue 'debuglevel=debug\n',
config.append('addpeer=127.0.0.1:{}\n'.format(base_p2p_port + i)) f'rpclisten=127.0.0.1:{DCR_BASE_WALLET_RPC_PORT + node_id}\n',
f'rpcconnect=127.0.0.1:{DCR_BASE_RPC_PORT + node_id}\n',
f'username=test{node_id}\n',
f'password=test_pass{node_id}\n',
'noservertls=1\n',
'noclienttls=1\n',]
wallet_cfg_file_path = os.path.join(node_dir, 'dcrwallet.conf')
with open(wallet_cfg_file_path, 'w+') as fp:
for line in config: for line in config:
fp.write(line) fp.write(line)
@ -83,6 +102,12 @@ class Test(BaseTest):
start_ltc_nodes = False start_ltc_nodes = False
start_xmr_nodes = False start_xmr_nodes = False
hex_seeds = [
'e8574b2a94404ee62d8acc0258cab4c0defcfab8a5dfc2f4954c1f9d7e09d72a',
'10689fc6378e5f318b663560012673441dcdd8d796134e6021a4248cc6342cc6',
'efc96ffe4fee469407826841d9700ef0a0735b0aa5ec5e7a4aa9bc1afd9a9a30', # Won't match main seed, as it's set randomly
]
@classmethod @classmethod
def prepareExtraCoins(cls): def prepareExtraCoins(cls):
pass pass
@ -103,7 +128,7 @@ class Test(BaseTest):
def prepareExtraDataDir(cls, i): def prepareExtraDataDir(cls, i):
extra_opts = [] extra_opts = []
if not cls.restore_instance: if not cls.restore_instance:
data_dir = prepareDCDDataDir(cfg.TEST_DATADIRS, i, 'dcrd.conf', 'dcr_', base_p2p_port=DCR_BASE_PORT, base_rpc_port=DCR_BASE_RPC_PORT) data_dir = prepareDCDDataDir(cfg.TEST_DATADIRS, i, 'dcrd.conf', 'dcr_')
appdata = os.path.join(cfg.TEST_DATADIRS, 'dcr_' + str(i)) appdata = os.path.join(cfg.TEST_DATADIRS, 'dcr_' + str(i))
datadir = os.path.join(appdata, 'data') datadir = os.path.join(appdata, 'data')
@ -113,12 +138,52 @@ class Test(BaseTest):
waitForRPC(make_rpc_func(i, base_rpc_port=DCR_BASE_RPC_PORT), test_delay_event, rpc_command='getnetworkinfo', max_tries=12) waitForRPC(make_rpc_func(i, base_rpc_port=DCR_BASE_RPC_PORT), test_delay_event, rpc_command='getnetworkinfo', max_tries=12)
logging.info('Creating wallet')
extra_opts.append('--pass=test_pass')
args = [os.path.join(DCR_BINDIR, DCR_WALLET), '--create'] + extra_opts
(pipe_r, pipe_w) = os.pipe() # subprocess.PIPE is buffered, blocks when read
p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=pipe_w, stderr=pipe_w)
try:
while p.poll() is None:
while len(select.select([pipe_r], [], [], 0)[0]) == 1:
buf = os.read(pipe_r, 1024).decode('utf-8')
response = None
if 'Use the existing configured private passphrase' in buf:
response = b'y\n'
elif 'Do you want to add an additional layer of encryption' in buf:
response = b'n\n'
elif 'Do you have an existing wallet seed' in buf:
response = b'y\n'
elif 'Enter existing wallet seed' in buf:
response = (cls.hex_seeds[i] + '\n').encode('utf-8')
else:
raise ValueError(f'Unexpected output: {buf}')
if response is not None:
p.stdin.write(response)
p.stdin.flush()
test_delay_event.wait(0.1)
except Exception as e:
logging.error(f'{DCR_WALLET} --create failed: {e}')
finally:
if p.poll() is None:
p.terminate()
os.close(pipe_r)
os.close(pipe_w)
p.stdin.close()
cls.dcr_daemons.append(startDaemon(appdata, DCR_BINDIR, DCR_WALLET, opts=extra_opts, extra_config={'add_datadir': False, 'stdout_to_file': True, 'stdout_filename': 'dcrwallet_stdout.log'}))
logging.info('Started %s %d', DCR_WALLET, cls.dcr_daemons[-1].handle.pid)
waitForRPC(make_rpc_func(i, base_rpc_port=DCR_BASE_WALLET_RPC_PORT), test_delay_event, rpc_command='getinfo', max_tries=12)
@classmethod @classmethod
def addCoinSettings(cls, settings, datadir, node_id): def addCoinSettings(cls, settings, datadir, node_id):
settings['chainclients']['decred'] = { settings['chainclients']['decred'] = {
'connection_type': 'rpc', 'connection_type': 'rpc',
'manage_daemon': False, 'manage_daemon': False,
'rpcport': DCR_BASE_RPC_PORT + node_id, 'rpcport': DCR_BASE_RPC_PORT + node_id,
'walletrpcport': DCR_BASE_WALLET_RPC_PORT + node_id,
'rpcuser': 'test' + str(node_id), 'rpcuser': 'test' + str(node_id),
'rpcpassword': 'test_pass' + str(node_id), 'rpcpassword': 'test_pass' + str(node_id),
'datadir': os.path.join(datadir, 'dcr_' + str(node_id)), 'datadir': os.path.join(datadir, 'dcr_' + str(node_id)),
@ -128,7 +193,7 @@ class Test(BaseTest):
'blocks_confirmed': 1, 'blocks_confirmed': 1,
} }
def test_001_decred(self): def test_0001_decred_address(self):
logging.info('---------- Test {}'.format(self.test_coin_from.name)) logging.info('---------- Test {}'.format(self.test_coin_from.name))
coin_settings = {'rpcport': 0, 'rpcauth': 'none'} coin_settings = {'rpcport': 0, 'rpcauth': 'none'}
@ -146,6 +211,26 @@ class Test(BaseTest):
data = ci.decode_address(address) data = ci.decode_address(address)
assert (data[2:] == pkh) assert (data[2:] == pkh)
def test_001_segwit(self):
logging.info('---------- Test {} segwit'.format(self.test_coin_from.name))
swap_clients = self.swap_clients
ci = swap_clients[0].ci(self.test_coin_from)
assert (ci.using_segwit() is True)
for i, sc in enumerate(swap_clients):
loop_ci = sc.ci(self.test_coin_from)
root_key = sc.getWalletKey(Coins.DCR, 1)
masterpubkey = loop_ci.rpc_wallet('getmasterpubkey')
masterpubkey_data = loop_ci.decode_address(masterpubkey)[4:]
seed_hash = loop_ci.getSeedHash(root_key)
if i == 0:
assert (masterpubkey == 'spubVV1z2AFYjVZvzM45FSaWMPRqyUoUwyW78wfANdjdNG6JGCXrr8AbRvUgYb3Lm1iun9CgHew1KswdePryNLKEnBSQ82AjNpYdQgzXPUme9c6')
if i < 2:
assert (seed_hash == hash160(masterpubkey_data))
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -23,8 +23,10 @@ from coincurve.keys import (
PrivateKey) PrivateKey)
from basicswap.util import i2b, h2b from basicswap.util import i2b, h2b
from basicswap.util.integer import encode_varint, decode_varint from basicswap.util.address import decodeAddress
from basicswap.util.crypto import ripemd160, hash160, blake256 from basicswap.util.crypto import ripemd160, hash160, blake256
from basicswap.util.extkey import ExtKeyPair
from basicswap.util.integer import encode_varint, decode_varint
from basicswap.util.network import is_private_ip_address from basicswap.util.network import is_private_ip_address
from basicswap.util.rfc2440 import rfc2440_hash_password from basicswap.util.rfc2440 import rfc2440_hash_password
from basicswap.util_xmr import encode_address as xmr_encode_address from basicswap.util_xmr import encode_address as xmr_encode_address
@ -312,7 +314,7 @@ class Test(unittest.TestCase):
assert ('10.00000000' == format_amount(amount_to_recreate, scale_to)) assert ('10.00000000' == format_amount(amount_to_recreate, scale_to))
coin_settings = {'rpcport': 0, 'rpcauth': 'none', 'walletrpcport': 0, 'walletrpcauth': 'none'} coin_settings = {'rpcport': 0, 'rpcauth': 'none', 'walletrpcport': 0, 'walletrpcauth': 'none'}
coin_settings.update(self.REQUIRED_SETTINGS) coin_settings.update(REQUIRED_SETTINGS)
ci_xmr = XMRInterface(coin_settings, 'regtest') ci_xmr = XMRInterface(coin_settings, 'regtest')
ci_btc = BTCInterface(coin_settings, 'regtest') ci_btc = BTCInterface(coin_settings, 'regtest')
@ -480,6 +482,37 @@ class Test(unittest.TestCase):
for expect_hash, data in test_vectors: for expect_hash, data in test_vectors:
assert (blake256(data).hex() == expect_hash) assert (blake256(data).hex() == expect_hash)
def test_extkey(self):
test_key = 'XPARHAr37YxmFP8wyjkaHAQWmp84GiyLikL7EL8j9BCx4LkB8Q1Bw5Kr8sA1GA3Ym53zNLcaxxFHr6u81JVTeCaD61c6fKS1YRAuti8Zu5SzJCjh'
test_key_c0 = 'XPARHAt1XMcNYAwP5wEnQXknBAkGSzaetdZt2eoJZehdB4WXfV1xbSjpgHe44AivmumcSejW5KaYx6L5M6MyR1WyXrsWTwaiUEfHq2RrqCfXj3ZW'
test_key_c0_p = 'PPARTKPL4rp5WLnrYP6jZfuRjx6jrmvbsz5QdHofPfFqJdm918mQwdPLq6Dd9TkdbQeKUqjbHWkyzWe7Pftd7itzm7ETEoUMq4cbG4fY9FKH1YSU'
test_key_c0h = 'XPARHAt1XMcNgWbv48LwoQbjs1bC8kCXKomzvJLRT5xmbQ2GKf9e8Vfr1MMcfiWJC34RyDp5HvAfjeiNyLDfkFm1UrRCrPkVC9GGaAWa3nXMWew8'
ek_data = decodeAddress(test_key)[4:]
ek = ExtKeyPair()
ek.decode(ek_data)
assert (ek.encode_v() == ek_data)
m_0 = ek.derive(0)
ek_c0_data = decodeAddress(test_key_c0)[4:]
assert (m_0.encode_v() == ek_c0_data)
child_no: int = 0 | (1 << 31)
m_0h = ek.derive(child_no)
ek_c0h_data = decodeAddress(test_key_c0h)[4:]
assert (m_0h.encode_v() == ek_c0h_data)
ek.neuter()
assert (ek.has_key() is False)
m_0 = ek.derive(0)
ek_c0_p_data = decodeAddress(test_key_c0_p)[4:]
assert (m_0.encode_p() == ek_c0_p_data)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()