mirror of
https://github.com/cryptoadvance/specter-desktop.git
synced 2026-08-13 12:33:29 +02:00
swap to embit for xpub and address derivation (#588)
This commit is contained in:
Stepan Snigirev
GitHub
parent
1825df9628
commit
17c8f82bd0
9 changed files with 357 additions and 255 deletions
|
|
@ -14,3 +14,4 @@ requests==2.23.0
|
|||
pysocks==1.7.1
|
||||
six==1.12.0
|
||||
stem==1.8.0
|
||||
embit==0.1.1
|
||||
|
|
|
|||
|
|
@ -94,6 +94,9 @@ ecdsa==0.13.3 \
|
|||
--hash=sha256:163c80b064a763ea733870feb96f9dd9b92216cfcacd374837af18e4e8ec3d4d \
|
||||
--hash=sha256:9814e700890991abeceeb2242586024d4758c8fc18445b194a49bd62d85861db \
|
||||
# via bitbox02, hwi
|
||||
embit==0.1.1 \
|
||||
--hash=sha256:15ab4f35036113e35cfcddd786a4123cecb2059ca2417c12ee52eba68bd91615 \
|
||||
# via -r requirements.in
|
||||
flask-cors==3.0.8 \
|
||||
--hash=sha256:72170423eb4612f0847318afff8c247b38bd516b7737adfc10d1c2cdbb382d16 \
|
||||
--hash=sha256:f4d97201660e6bbcff2d89d082b5b6d31abee04b1b3003ee073a6fd25ad1d69a \
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import os
|
||||
import shutil
|
||||
from mnemonic import Mnemonic
|
||||
from embit import bip39, bip32, networks
|
||||
from ..device import Device
|
||||
from ..helpers import alias
|
||||
from ..util.descriptor import AddChecksum
|
||||
|
|
@ -41,8 +41,11 @@ class BitcoinCore(Device):
|
|||
testnet,
|
||||
keys_range=[0, 1000],
|
||||
):
|
||||
seed = Mnemonic.to_seed(mnemonic, passphrase)
|
||||
xprv = seed_to_hd_master_key(seed, testnet=testnet)
|
||||
seed = bip39.mnemonic_to_seed(mnemonic, passphrase)
|
||||
root = bip32.HDKey.from_seed(seed)
|
||||
network = networks.NETWORKS["test" if testnet else "main"]
|
||||
root.version = network["xprv"]
|
||||
xprv = root.to_base58()
|
||||
# Load the wallet if not loaded
|
||||
self._load_wallet(wallet_manager)
|
||||
rpc = wallet_manager.rpc.wallet(
|
||||
|
|
@ -54,7 +57,9 @@ class BitcoinCore(Device):
|
|||
[
|
||||
{
|
||||
"desc": AddChecksum(
|
||||
"sh(wpkh({}{}/0/*))".format(xprv, path.replace("m", ""))
|
||||
"sh(wpkh({}{}/0/*))".format(
|
||||
xprv, path.rstrip("/").replace("m", "")
|
||||
)
|
||||
),
|
||||
"range": keys_range,
|
||||
"timestamp": "now",
|
||||
|
|
@ -64,7 +69,9 @@ class BitcoinCore(Device):
|
|||
+ [
|
||||
{
|
||||
"desc": AddChecksum(
|
||||
"sh(wpkh({}{}/1/*))".format(xprv, path.replace("m", ""))
|
||||
"sh(wpkh({}{}/1/*))".format(
|
||||
xprv, path.rstrip("/").replace("m", "")
|
||||
)
|
||||
),
|
||||
"range": keys_range,
|
||||
"timestamp": "now",
|
||||
|
|
@ -75,41 +82,16 @@ class BitcoinCore(Device):
|
|||
)
|
||||
|
||||
xpubs_str = ""
|
||||
paths = ["m"] + paths
|
||||
xpubs = derive_xpubs_from_xprv(xprv, paths, wallet_manager.rpc)
|
||||
# it's not parent fingerprint, it's self fingerprint
|
||||
master_fpr = get_xpub_fingerprint(xpubs[0]).hex()
|
||||
xpubs = [root.derive(path).to_public().to_base58() for path in paths]
|
||||
# root fingerprint is fingerprint field of the first child
|
||||
master_fpr = root.child(0).fingerprint.hex()
|
||||
|
||||
slip132_paths = [
|
||||
"m/49'/0'/0'",
|
||||
"m/84'/0'/0'",
|
||||
"m/48'/0'/0'/1'",
|
||||
"m/48'/0'/0'/2'",
|
||||
"m/49'/1'/0'",
|
||||
"m/84'/1'/0'",
|
||||
"m/48'/1'/0'/1'",
|
||||
"m/48'/1'/0'/2'",
|
||||
]
|
||||
slip132_prefixes = [
|
||||
b"\x04\x9d\x7c\xb2",
|
||||
b"\x04\xb2\x47\x46",
|
||||
b"\x02\x95\xb4\x3f",
|
||||
b"\x02\xaa\x7e\xd3",
|
||||
b"\x04\x4a\x52\x62",
|
||||
b"\x04\x5f\x1c\xf6",
|
||||
b"\x02\x42\x89\xef",
|
||||
b"\x02\x57\x54\x83",
|
||||
]
|
||||
for i in range(1, len(paths)):
|
||||
for i in range(len(paths)):
|
||||
path = paths[i]
|
||||
xpub = xpubs[i]
|
||||
slip132_prefix = None
|
||||
for j, slip132_path in enumerate(slip132_paths):
|
||||
if path.replace("h", "'").startswith(slip132_path):
|
||||
slip132_prefix = slip132_prefixes[j]
|
||||
break
|
||||
if slip132_prefix:
|
||||
xpub = convert_xpub_prefix(xpub, slip132_prefix)
|
||||
# detect slip132 version for xpubs
|
||||
slip132_prefix = bip32.detect_version(path, default="xpub", network=network)
|
||||
xpub = convert_xpub_prefix(xpub, slip132_prefix)
|
||||
xpubs_str += "[{}{}]{}\n".format(master_fpr, path.replace("m", ""), xpub)
|
||||
|
||||
keys, failed = Key.parse_xpubs(xpubs_str)
|
||||
|
|
@ -192,127 +174,3 @@ class BitcoinCore(Device):
|
|||
break
|
||||
except:
|
||||
pass # We tried...
|
||||
|
||||
|
||||
# We need to copy it like this because HWI uses it as a dependency,
|
||||
# but requires v0.18 which doesn't have this function.
|
||||
|
||||
|
||||
def seed_to_hd_master_key(seed, testnet=False) -> str:
|
||||
"""Converts bip32 seed to xprv"""
|
||||
if len(seed) < 16 or len(seed) > 64:
|
||||
raise ValueError("Provided seed should be between 16 and 64 bytes")
|
||||
|
||||
# Compute HMAC-SHA512 of seed
|
||||
seed = hmac.new(b"Bitcoin seed", seed, digestmod="sha512").digest()
|
||||
|
||||
# Serialization format can be found at:
|
||||
# https://github.com/bitcoin/bips/blob/master/bip-0032.mediawiki#Serialization_format
|
||||
xprv = b"\x04\x88\xad\xe4" # Version for private mainnet
|
||||
if testnet:
|
||||
xprv = b"\x04\x35\x83\x94" # Version for private testnet
|
||||
xprv += b"\x00" * 9 # Depth, parent fingerprint, and child number
|
||||
xprv += seed[32:] # Chain code
|
||||
xprv += b"\x00" + seed[:32] # Master key
|
||||
|
||||
return encode_base58_checksum(xprv)
|
||||
|
||||
|
||||
def derive_xpubs_from_xprv(xprv, paths: list, rpc):
|
||||
"""
|
||||
Derives xpubs from root xprv and list of paths.
|
||||
Requires running BitcoinRPC instance to derive xpub from xprv
|
||||
"""
|
||||
derived_xprvs = []
|
||||
for path in paths:
|
||||
derivation = parse_path(path)
|
||||
if len(derivation) == 0:
|
||||
# tuple: (parent, derived)
|
||||
derived_xprvs.append((None, xprv))
|
||||
else:
|
||||
# we need parent for fingerprint
|
||||
parent = xprv
|
||||
for idx in derivation[:-1]:
|
||||
parent = get_child(parent, idx, rpc)
|
||||
child = get_child(parent, derivation[-1], rpc)
|
||||
derived_xprvs.append((parent, child))
|
||||
xpubs = []
|
||||
for parent, child in derived_xprvs:
|
||||
res = rpc.getdescriptorinfo(f"wpkh({child})")
|
||||
xpub = res["descriptor"].split("(")[1].split(")")[0]
|
||||
if parent is not None:
|
||||
res = rpc.getdescriptorinfo(f"wpkh({parent})")
|
||||
parent_xpub = res["descriptor"].split("(")[1].split(")")[0]
|
||||
fingerprint = get_xpub_fingerprint(parent_xpub)
|
||||
xpub = swap_fingerprint(xpub, fingerprint)
|
||||
xpubs.append(xpub)
|
||||
return xpubs
|
||||
|
||||
|
||||
def swap_fingerprint(xpub, fingerprint):
|
||||
"""Replaces fingerprint in xpub"""
|
||||
raw = decode_base58(xpub)
|
||||
swapped = raw[:5] + fingerprint + raw[9:]
|
||||
return encode_base58_checksum(swapped)
|
||||
|
||||
|
||||
# curve order
|
||||
N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
|
||||
|
||||
|
||||
def get_child(xprv, index, rpc=None):
|
||||
"""Derives a child from xprv but without fingerprint information"""
|
||||
if index > 0xFFFFFFFF or index < 0:
|
||||
raise ValueError("Index should be between 0 and 2^32")
|
||||
|
||||
stream = BytesIO(decode_base58(xprv))
|
||||
version = stream.read(4)
|
||||
depth = stream.read(1)[0]
|
||||
fingerprint = stream.read(4)
|
||||
child_number = int.from_bytes(stream.read(4), "big")
|
||||
chain_code = stream.read(32)
|
||||
stream.read(1)
|
||||
secret = stream.read(32)
|
||||
key = b"\x00" + secret
|
||||
|
||||
if index < 0x80000000:
|
||||
if rpc is None:
|
||||
raise ValueError("Can't do non-hardened without rpc")
|
||||
# non hardened - we need pubkey
|
||||
else:
|
||||
# convert to public
|
||||
res = rpc.getdescriptorinfo(f"wpkh({xprv})")
|
||||
xpub = res["descriptor"].split("(")[1].split(")")[0]
|
||||
# decode pubkey
|
||||
key = decode_base58(xpub)[-33:]
|
||||
|
||||
data = key + index.to_bytes(4, "big")
|
||||
raw = hmac.new(chain_code, data, digestmod="sha512").digest()
|
||||
tweak = raw[:32]
|
||||
chain_code = raw[32:]
|
||||
|
||||
new_secret = (int.from_bytes(secret, "big") + int.from_bytes(tweak, "big")) % N
|
||||
res = version + bytes([depth + 1]) + fingerprint + index.to_bytes(4, "big")
|
||||
res += chain_code + b"\x00" + new_secret.to_bytes(32, "big")
|
||||
return encode_base58_checksum(res)
|
||||
|
||||
|
||||
def parse_path(path: str) -> list:
|
||||
"""
|
||||
Converts derivation path of the form
|
||||
m/44h/1'/0'/0/32 to int array
|
||||
"""
|
||||
arr = path.split("/")
|
||||
if arr[0] == "m":
|
||||
arr = arr[1:]
|
||||
if len(arr) == 0:
|
||||
return []
|
||||
if arr[-1] == "":
|
||||
# trailing slash
|
||||
arr = arr[:-1]
|
||||
for i, e in enumerate(arr):
|
||||
if e[-1] == "h" or e[-1] == "'":
|
||||
arr[i] = int(e[:-1]) + 0x80000000
|
||||
else:
|
||||
arr[i] = int(e)
|
||||
return arr
|
||||
|
|
|
|||
|
|
@ -13,7 +13,6 @@ from collections import OrderedDict
|
|||
from mnemonic import Mnemonic
|
||||
from hwilib.serializations import PSBT, CTransaction
|
||||
from .persistence import read_json_file, write_json_file
|
||||
from .util.descriptor import AddChecksum
|
||||
from .util.bcur import bcur_decode
|
||||
import threading
|
||||
from io import BytesIO
|
||||
|
|
@ -171,49 +170,6 @@ def get_devices_with_keys_by_type(app, cosigners, wallet_type):
|
|||
return devices
|
||||
|
||||
|
||||
def sort_descriptor(rpc, descriptor, index=None, change=False):
|
||||
descriptor = descriptor.replace("sortedmulti", "multi")
|
||||
if index is not None:
|
||||
descriptor = descriptor.replace("*", f"{index}")
|
||||
# remove checksum
|
||||
descriptor = descriptor.split("#")[0]
|
||||
# get address (should be already imported to the wallet)
|
||||
address = rpc.deriveaddresses(AddChecksum(descriptor), change=change)[0]
|
||||
|
||||
# get pubkeys involved
|
||||
address_info = rpc.getaddressinfo(address)
|
||||
if "pubkeys" in address_info:
|
||||
pubkeys = address_info["pubkeys"]
|
||||
elif "embedded" in address_info and "pubkeys" in address_info["embedded"]:
|
||||
pubkeys = address_info["embedded"]["pubkeys"]
|
||||
else:
|
||||
raise Exception(
|
||||
"Could not find 'pubkeys' in address info:\n%s"
|
||||
% json.dumps(address_info, indent=2)
|
||||
)
|
||||
|
||||
# get xpubs from the descriptor
|
||||
arr = descriptor.split("(multi(")[1].split(")")[0].split(",")
|
||||
|
||||
# getting [wsh] or [sh, wsh]
|
||||
prefix = descriptor.split("(multi(")[0].split("(")
|
||||
sigs_required = arr[0]
|
||||
keys = arr[1:]
|
||||
|
||||
# sort them according to sortedmulti
|
||||
z = sorted(zip(pubkeys, keys), key=lambda x: x[0])
|
||||
keys = [zz[1] for zz in z]
|
||||
inner = f"{sigs_required}," + ",".join(keys)
|
||||
desc = f"multi({inner})"
|
||||
|
||||
# Write from the inside out
|
||||
prefix.reverse()
|
||||
for p in prefix:
|
||||
desc = f"{p}({desc})"
|
||||
|
||||
return AddChecksum(desc)
|
||||
|
||||
|
||||
def clean_psbt(b64psbt):
|
||||
psbt = PSBT()
|
||||
psbt.deserialize(b64psbt)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import re
|
||||
from embit import bip32, ec, networks, script
|
||||
|
||||
# From: https://github.com/bitcoin/bitcoin/blob/master/src/script/descriptor.cpp
|
||||
|
||||
|
|
@ -53,6 +54,21 @@ def AddChecksum(desc):
|
|||
return desc + "#" + DescriptorChecksum(desc)
|
||||
|
||||
|
||||
def derive_pubkey(key, path_suffix=None, idx=None):
|
||||
# if SEC pubkey - just return it
|
||||
if key[:2] in ["02", "03", "04"]:
|
||||
return ec.PublicKey.parse(bytes.fromhex(key))
|
||||
# otherwise - xpub or xprv
|
||||
hd = bip32.HDKey.from_base58(key)
|
||||
if hd.is_private:
|
||||
hd = hd.to_public()
|
||||
# if we have path suffix
|
||||
path = "m" + (path_suffix or "")
|
||||
if idx is not None:
|
||||
path = path.replace("*", str(idx))
|
||||
return hd.derive(path).key
|
||||
|
||||
|
||||
class Descriptor:
|
||||
def __init__(
|
||||
self,
|
||||
|
|
@ -68,6 +84,7 @@ class Descriptor:
|
|||
wsh=None,
|
||||
multisig_M=None,
|
||||
multisig_N=None,
|
||||
sort_keys=True,
|
||||
):
|
||||
self.origin_fingerprint = origin_fingerprint
|
||||
self.origin_path = origin_path
|
||||
|
|
@ -82,10 +99,21 @@ class Descriptor:
|
|||
self.multisig_M = multisig_M
|
||||
self.multisig_N = multisig_N
|
||||
self.m_path = None
|
||||
self.sort_keys = sort_keys
|
||||
|
||||
if origin_path and not isinstance(origin_path, list):
|
||||
self.m_path_base = "m" + origin_path
|
||||
self.m_path = "m" + origin_path + (path_suffix or "")
|
||||
elif isinstance(origin_path, list):
|
||||
self.m_path_base = []
|
||||
self.m_path = []
|
||||
for i in range(0, len(origin_path)):
|
||||
if origin_path[i]:
|
||||
self.m_path_base.append("m" + origin_path[i])
|
||||
self.m_path.append("m" + origin_path[i] + (path_suffix[i] or ""))
|
||||
else:
|
||||
self.m_path_base.append(None)
|
||||
self.m_path.append(None)
|
||||
|
||||
@classmethod
|
||||
def parse(cls, desc, testnet=False):
|
||||
|
|
@ -101,6 +129,7 @@ class Descriptor:
|
|||
path_suffix = None
|
||||
multisig_M = None
|
||||
multisig_N = None
|
||||
sort_keys = True
|
||||
|
||||
# Check the checksum
|
||||
check_split = desc.split("#")
|
||||
|
|
@ -133,8 +162,13 @@ class Descriptor:
|
|||
return None
|
||||
# get the list of keys only
|
||||
keys = desc.split(",", 1)[1].split(")", 1)[0].split(",")
|
||||
sort_keys = "sortedmulti" in desc
|
||||
if "sortedmulti" in desc:
|
||||
keys.sort(key=lambda x: x if "]" not in x else x.split("]")[1])
|
||||
# sorting makes sense only if individual pubkeys are provided
|
||||
base_keys = [x if "]" not in x else x.split("]")[1] for x in keys]
|
||||
bare_pubkeys = [k for k in base_keys if k[:2] in ["02", "03", "04"]]
|
||||
if len(bare_pubkeys) == len(keys):
|
||||
keys.sort(key=lambda x: x if "]" not in x else x.split("]")[1])
|
||||
multisig_M = desc.split(",")[0].split("(")[-1]
|
||||
multisig_N = len(keys)
|
||||
else:
|
||||
|
|
@ -180,6 +214,7 @@ class Descriptor:
|
|||
sh,
|
||||
sh_wsh,
|
||||
wsh,
|
||||
sort_keys,
|
||||
)
|
||||
)
|
||||
if len(descriptors) == 1:
|
||||
|
|
@ -199,29 +234,170 @@ class Descriptor:
|
|||
wsh,
|
||||
multisig_M,
|
||||
multisig_N,
|
||||
sort_keys,
|
||||
)
|
||||
|
||||
@property
|
||||
def is_multisig(self):
|
||||
return bool(self.multisig_N)
|
||||
|
||||
def derive(self, idx, keep_xpubs=False):
|
||||
"""
|
||||
Derives a descriptor with index idx up to the pubkeys.
|
||||
If keep_xpubs is False all xpubs will be replaces by pubkeys
|
||||
so [fgp/path]xpub/suffix changes to [fgp/path/suffix]pubkey
|
||||
Otherwise xpubs will be sorted according to pubkeys
|
||||
but remain in the descriptor
|
||||
"""
|
||||
if self.is_multisig:
|
||||
keys = []
|
||||
for i, key in enumerate(self.base_key):
|
||||
keys.append(
|
||||
(
|
||||
derive_pubkey(key, self.path_suffix[i], idx),
|
||||
key,
|
||||
self.origin_fingerprint[i],
|
||||
self.origin_path[i],
|
||||
self.path_suffix[i],
|
||||
)
|
||||
)
|
||||
if self.sort_keys:
|
||||
keys = sorted(keys, key=lambda k: k[0])
|
||||
origin_fingerprint = [k[2] for k in keys]
|
||||
if keep_xpubs:
|
||||
base_key = [k[1] for k in keys]
|
||||
origin_path = [k[3] for k in keys]
|
||||
path_suffix = [(k[4] or "") for k in keys]
|
||||
path_suffix = [(p.replace("*", str(idx)) or None) for p in path_suffix]
|
||||
else:
|
||||
base_key = [k[0].sec().hex() for k in keys]
|
||||
origin_path = [(k[3] or "") + (k[4] or "") for k in keys]
|
||||
origin_path = [(p.replace("*", str(idx)) or None) for p in origin_path]
|
||||
path_suffix = [None for k in keys]
|
||||
else:
|
||||
origin_fingerprint = self.origin_fingerprint
|
||||
if keep_xpubs:
|
||||
base_key = self.base_key
|
||||
origin_path = self.origin_path
|
||||
path_suffix = self.path_suffix
|
||||
path_suffix = (
|
||||
path_suffix.replace("*", str(idx)) if path_suffix else None
|
||||
)
|
||||
else:
|
||||
base_key = (
|
||||
derive_pubkey(self.base_key, self.path_suffix, idx).sec().hex()
|
||||
)
|
||||
origin_path = (self.origin_path or "") + (self.path_suffix or "")
|
||||
origin_path = origin_path.replace("*", str(idx)) or None
|
||||
path_suffix = None
|
||||
return Descriptor(
|
||||
origin_fingerprint,
|
||||
origin_path,
|
||||
base_key,
|
||||
path_suffix,
|
||||
self.testnet,
|
||||
self.sh_wpkh,
|
||||
self.wpkh,
|
||||
self.sh,
|
||||
self.sh_wsh,
|
||||
self.wsh,
|
||||
self.multisig_M,
|
||||
self.multisig_N,
|
||||
self.sort_keys,
|
||||
)
|
||||
|
||||
def scriptpubkey(self, idx=None):
|
||||
if idx is None and "*" in self.serialize():
|
||||
raise RuntimeError("Index is required")
|
||||
if self.is_multisig:
|
||||
keys = []
|
||||
for i, key in enumerate(self.base_key):
|
||||
keys.append(derive_pubkey(key, self.path_suffix[i], idx))
|
||||
if self.sort_keys:
|
||||
keys = sorted(keys)
|
||||
sc = script.multisig(int(self.multisig_M), keys)
|
||||
if self.sh:
|
||||
return script.p2sh(sc)
|
||||
elif self.sh_wsh:
|
||||
return script.p2sh(script.p2wsh(sc))
|
||||
elif self.wsh:
|
||||
return script.p2wsh(sc)
|
||||
else:
|
||||
key = derive_pubkey(self.base_key, self.path_suffix, idx)
|
||||
if self.wpkh:
|
||||
return script.p2wpkh(key)
|
||||
elif self.sh_wpkh:
|
||||
return script.p2sh(script.p2wpkh(key))
|
||||
else:
|
||||
return script.p2pkh(key)
|
||||
|
||||
def address(self, idx=None, network=None):
|
||||
if network is None:
|
||||
net = networks.NETWORKS["test" if self.testnet else "main"]
|
||||
else:
|
||||
net = networks.NETWORKS[network]
|
||||
return self.scriptpubkey(idx).address(net)
|
||||
|
||||
def serialize(self):
|
||||
descriptor_open = "pkh("
|
||||
descriptor_close = ")"
|
||||
origin = ""
|
||||
path_suffix = ""
|
||||
|
||||
if self.wpkh:
|
||||
descriptor_open = "wpkh("
|
||||
elif self.sh_wpkh:
|
||||
descriptor_open = "sh(wpkh("
|
||||
descriptor_close = "))"
|
||||
elif self.sh or self.sh_wsh or self.wsh:
|
||||
# serialize multisig descriptor is not supported yet.
|
||||
return None
|
||||
elif self.sh:
|
||||
descriptor_open = "sh("
|
||||
descriptor_close = ")"
|
||||
elif self.sh_wsh:
|
||||
descriptor_open = "sh(wsh("
|
||||
descriptor_close = "))"
|
||||
elif self.wsh:
|
||||
descriptor_open = "wsh("
|
||||
descriptor_close = ")"
|
||||
|
||||
if self.origin_fingerprint and self.origin_path:
|
||||
origin = "[" + self.origin_fingerprint + self.origin_path + "]"
|
||||
if self.is_multisig:
|
||||
multi = "sortedmulti" if self.sort_keys else "multi"
|
||||
base_open = f"{multi}({self.multisig_M},"
|
||||
base_close = ")"
|
||||
origins = []
|
||||
for i in range(self.multisig_N):
|
||||
path_suffix = ""
|
||||
origin = ""
|
||||
if self.origin_fingerprint[i] and self.origin_path[i]:
|
||||
origin = (
|
||||
"[" + self.origin_fingerprint[i] + self.origin_path[i] + "]"
|
||||
)
|
||||
|
||||
if self.path_suffix:
|
||||
path_suffix = self.path_suffix
|
||||
if self.path_suffix[i]:
|
||||
path_suffix = self.path_suffix[i]
|
||||
|
||||
return AddChecksum(
|
||||
descriptor_open + origin + self.base_key + path_suffix + descriptor_close
|
||||
)
|
||||
origins.append(origin + self.base_key[i] + path_suffix)
|
||||
|
||||
base = base_open + ",".join(origins) + base_close
|
||||
else:
|
||||
origin = ""
|
||||
path_suffix = ""
|
||||
if self.origin_fingerprint and self.origin_path:
|
||||
origin = "[" + self.origin_fingerprint + self.origin_path + "]"
|
||||
|
||||
if self.path_suffix:
|
||||
path_suffix = self.path_suffix
|
||||
|
||||
base = origin + self.base_key + path_suffix
|
||||
|
||||
return AddChecksum(descriptor_open + base + descriptor_close)
|
||||
|
||||
|
||||
def sort_descriptor(descriptor, index=None):
|
||||
"""
|
||||
Sorts descriptor to maintain compatibility with Core 19
|
||||
as it doesn't support sortedmulti.
|
||||
Returns a derived multi() descriptor with sorted xpubs inside.
|
||||
"""
|
||||
desc = Descriptor.parse(descriptor)
|
||||
desc.sort_keys = True
|
||||
sorted_desc = desc.derive(index, keep_xpubs=True)
|
||||
sorted_desc.sort_keys = False
|
||||
return sorted_desc.serialize()
|
||||
|
|
|
|||
75
src/cryptoadvance/specter/util/tx.py
Normal file
75
src/cryptoadvance/specter/util/tx.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
from embit.transaction import Transaction
|
||||
from embit.networks import NETWORKS
|
||||
from hashlib import sha256
|
||||
import math
|
||||
|
||||
TYPES_MAP = {
|
||||
"p2wpkh": "witness_v0_keyhash",
|
||||
}
|
||||
|
||||
|
||||
def decoderawinput(vin):
|
||||
result = {
|
||||
"txid": vin.txid.hex(),
|
||||
"vout": vin.vout,
|
||||
"scriptSig": {
|
||||
# TODO: asm
|
||||
# "asm": "0014c08fd0c4658b89678b9e0726838c2c2c2f41c3df",
|
||||
"hex": vin.script_sig.data.hex()
|
||||
},
|
||||
"sequence": vin.sequence,
|
||||
}
|
||||
if vin.is_segwit:
|
||||
result["txinwitness"] = [item.hex() for item in vin.witness.items]
|
||||
return result
|
||||
|
||||
|
||||
def decoderawoutput(vout, chain):
|
||||
result = {
|
||||
"value": vout.value * 1e-8,
|
||||
"scriptPubKey": {
|
||||
# TODO: asm
|
||||
# "asm": "0 f81b3e69f5cafc2f1e69ed5625d07876e3558e69",
|
||||
"hex": vout.script_pubkey.data.hex(),
|
||||
# TODO: reqSigs, type, address only if you can
|
||||
# "reqSigs": 1,
|
||||
# "type": "witness_v0_keyhash",
|
||||
},
|
||||
}
|
||||
try:
|
||||
result["addresses"] = [vout.script_pubkey.address(NETWORKS[chain])]
|
||||
except:
|
||||
pass
|
||||
return result
|
||||
|
||||
|
||||
def decoderawtransaction(hextx, chain="main"):
|
||||
raw = bytes.fromhex(hextx)
|
||||
tx = Transaction.parse(raw)
|
||||
txhash = sha256(sha256(raw).digest()).digest()[::-1].hex()
|
||||
txsize = len(raw)
|
||||
if tx.is_segwit:
|
||||
# tx size - flag - marker - witness
|
||||
non_witness_size = (
|
||||
txsize - 2 - sum([len(inp.witness.serialize()) for inp in tx.vin])
|
||||
)
|
||||
witness_size = txsize - non_witness_size
|
||||
weight = non_witness_size * 4 + witness_size
|
||||
vsize = math.ceil(weight / 4)
|
||||
else:
|
||||
vsize = txsize
|
||||
weight = txsize * 4
|
||||
result = {
|
||||
"txid": tx.txid().hex(),
|
||||
"hash": txhash,
|
||||
"version": tx.version,
|
||||
"size": txsize,
|
||||
"vsize": vsize,
|
||||
"weight": weight,
|
||||
"locktime": tx.locktime,
|
||||
"vin": [decoderawinput(vin) for vin in tx.vin],
|
||||
"vout": [
|
||||
dict(decoderawoutput(vout, chain), n=i) for i, vout in enumerate(tx.vout)
|
||||
],
|
||||
}
|
||||
return result
|
||||
|
|
@ -1,13 +1,13 @@
|
|||
import copy, hashlib, json, logging, os
|
||||
import time
|
||||
from hwilib.descriptor import AddChecksum
|
||||
from .device import Device
|
||||
from .key import Key
|
||||
from .util.merkleblock import is_valid_merkle_proof
|
||||
from .helpers import der_to_bytes, sort_descriptor, parse_utxo
|
||||
from .helpers import der_to_bytes, parse_utxo
|
||||
from .util.base58 import decode_base58
|
||||
from .util.descriptor import Descriptor
|
||||
from .util.descriptor import Descriptor, sort_descriptor, AddChecksum
|
||||
from .util.xpub import get_xpub_fingerprint
|
||||
from .util.tx import decoderawtransaction
|
||||
from .persistence import write_json_file
|
||||
from hwilib.serializations import PSBT, CTransaction
|
||||
from io import BytesIO
|
||||
|
|
@ -426,8 +426,8 @@ class Wallet:
|
|||
if tx["confirmations"] == 0 and (
|
||||
tx["category"] == "send" and tx["bip125-replaceable"] == "yes"
|
||||
):
|
||||
raw_tx = self.rpc.decoderawtransaction(
|
||||
self.rpc.gettransaction(tx["txid"])["hex"]
|
||||
raw_tx = decoderawtransaction(
|
||||
self.rpc.gettransaction(tx["txid"])["hex"], self.manager.chain
|
||||
)
|
||||
tx["vsize"] = raw_tx["vsize"]
|
||||
|
||||
|
|
@ -665,16 +665,7 @@ class Wallet:
|
|||
if pool < index + self.GAP_LIMIT:
|
||||
self.keypoolrefill(pool, index + self.GAP_LIMIT, change=change)
|
||||
desc = self.change_descriptor if change else self.recv_descriptor
|
||||
if self.is_multisig:
|
||||
try:
|
||||
# first try with sortedmulti
|
||||
addr = self.rpc.deriveaddresses(desc, [index, index + 1])[0]
|
||||
except Exception:
|
||||
# if sortedmulti is not supported
|
||||
desc = sort_descriptor(self.rpc, desc, index=index, change=change)
|
||||
addr = self.rpc.deriveaddresses(desc)[0]
|
||||
return addr
|
||||
return self.rpc.deriveaddresses(desc, [index, index + 1])[0]
|
||||
return Descriptor.parse(desc).address(index, self.manager.chain)
|
||||
|
||||
def get_descriptor(self, index=None, change=False, address=None):
|
||||
"""
|
||||
|
|
@ -685,24 +676,12 @@ class Wallet:
|
|||
return self.rpc.getaddressinfo(address).get("desc", "")
|
||||
if index is None:
|
||||
index = self.change_index if change else self.address_index
|
||||
desc = self.rpc.getaddressinfo(self.get_address(index, change)).get("desc", "")
|
||||
result = {"descriptor": desc, "xpubs_descriptor": None}
|
||||
if self.is_multisig:
|
||||
if address is not None:
|
||||
d = self.rpc.getaddressinfo(address)["desc"]
|
||||
path = d.split("[")[1].split("]")[0].split("/")
|
||||
change = bool(int(path[-2]))
|
||||
index = int(path[-1])
|
||||
if index is None:
|
||||
index = self.change_index if change else self.address_index
|
||||
desc = self.change_descriptor if change else self.recv_descriptor
|
||||
try:
|
||||
result["xpubs_descriptor"] = sort_descriptor(
|
||||
self.rpc, desc, index=index, change=change
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
return result
|
||||
desc = self.change_descriptor if change else self.recv_descriptor
|
||||
derived_desc = Descriptor.parse(desc).derive(index).serialize()
|
||||
derived_desc_xpubs = (
|
||||
Descriptor.parse(desc).derive(index, keep_xpubs=True).serialize()
|
||||
)
|
||||
return {"descriptor": derived_desc, "xpubs_descriptor": derived_desc_xpubs}
|
||||
|
||||
def get_electrum_watchonly(self):
|
||||
if len(self.keys) == 1:
|
||||
|
|
@ -757,7 +736,7 @@ class Wallet:
|
|||
available.update(balance)
|
||||
for tx in locked_utxo:
|
||||
tx_data = self.rpc.gettransaction(tx["txid"])
|
||||
raw_tx = self.rpc.decoderawtransaction(tx_data["hex"])
|
||||
raw_tx = decoderawtransaction(tx_data["hex"], self.manager.chain)
|
||||
delta = raw_tx["vout"][tx["vout"]]["value"]
|
||||
if "confirmations" not in tx_data or tx_data["confirmations"] == 0:
|
||||
available["untrusted_pending"] -= delta
|
||||
|
|
@ -813,9 +792,7 @@ class Wallet:
|
|||
arg.pop("range")
|
||||
batch = []
|
||||
for i in range(start, end):
|
||||
sorted_desc = sort_descriptor(
|
||||
self.rpc, desc, index=i, change=change
|
||||
)
|
||||
sorted_desc = sort_descriptor(desc, index=i)
|
||||
# create fresh object
|
||||
obj = {}
|
||||
obj.update(arg)
|
||||
|
|
|
|||
23
tests/test_decodetx.py
Normal file
23
tests/test_decodetx.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
from cryptoadvance.specter.util.tx import *
|
||||
|
||||
|
||||
def test_decoderawtransaction():
|
||||
hextx = "02000000000101902666609a245e45e426ead256ad47dca8a2b4dd65d1a634ad35b4a1c0603c0e0000000017160014c08fd0c4658b89678b9e0726838c2c2c2f41c3dffeffffff02b44d5a0300000000160014f81b3e69f5cafc2f1e69ed5625d07876e3558e6900e1f50500000000160014255fe80139657184c1de8e04f71c74c667dd7c3f02473044022038a29d1958d295a9739798c1a5f138404711d824f3bf103221d31bcbc11ff0010220635b4996de17290980c78e3de2547717de119e8312a52eb54fce73c27728c5e00121020356c34dd0931251a6e306704a912dbfeedb0f550a30a89689a1c8434cefa91000000000"
|
||||
res = decoderawtransaction(hextx, "regtest")
|
||||
assert res["size"] == 245
|
||||
assert res["vsize"] == 164
|
||||
assert res["weight"] == 653
|
||||
|
||||
# coinbase tx
|
||||
hextx = "010000000001010000000000000000000000000000000000000000000000000000000000000000ffffffff5f03a2030a1c2f5669614254432f4d696e656420627920736762756c6f686b79632f2cfabe6d6ded74726703908120aa4fa4f10557f50c66aa7419c7285edab024872b7bdd96bf1000000000000000101772a30fd92f0420e9dce2e133ebaa2affffffff0488b49929000000001976a914536ffa992491508dca0354e52f32a3a7a679a53a88ac00000000000000002b6a2952534b424c4f434b3a6f819a142bdad27916d8e8daf419907fa2cd7b085019d035823f3125002b83500000000000000000266a24b9e11b6d563b4d7c8486af0c15dac4c2764a39be25365cfb8990db6d170d794a562080660000000000000000266a24aa21a9ed256d453640cd18d3fe21bdc7f127e9c075e630b28be937a0c8e402807bc010360120000000000000000000000000000000000000000000000000000000000000000000000000"
|
||||
res = decoderawtransaction(hextx, "regtest")
|
||||
assert res["size"] == 362
|
||||
assert res["vsize"] == 335
|
||||
assert res["weight"] == 1340
|
||||
|
||||
# legacy tx
|
||||
hextx = "020000000191f381c648c70f2388cce607f5955fe6b9f0b50a49c9bfa618413f931e55cf16000000006a4730440220543b92a31ed7cd00781cdce8cac4ef37fbfdce30a9dfc1f8e00a77f2dd35a2ec02201eb21ec97126f0dad8f0f066e0ae1cf44de8a3027caa99b819511ec57ba632c70121020f9c0041942551b00abcf1ba8d00f6ac93e67ddb378eecd0fb240a9ef3ddc9c0ffffffff0182480a010000000017a9143524696d526f50ab583c829bcca02553af9c64fa8700000000"
|
||||
res = decoderawtransaction(hextx, "regtest")
|
||||
assert res["size"] == 189
|
||||
assert res["vsize"] == 189
|
||||
assert res["weight"] == 756
|
||||
33
tests/test_descriptor.py
Normal file
33
tests/test_descriptor.py
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
from cryptoadvance.specter.util.descriptor import *
|
||||
|
||||
|
||||
def test_parse():
|
||||
descs = [
|
||||
"wpkh([5d5c5649/84h/1h/0h]tpubDCB5nE2GEEuX9xyFigt33xT1RidfkSsH2VSqDx93D1TrvghcZgoDBTjWnWwKTtA6DfvW7fKDAzPJoSduEbt1QkUW2YGaC2CgYxvmF9RyRZS/0/*)#ypamvruf",
|
||||
"sh(wpkh([5d5c5649/84h/1h/0h]tpubDCB5nE2GEEuX9xyFigt33xT1RidfkSsH2VSqDx93D1TrvghcZgoDBTjWnWwKTtA6DfvW7fKDAzPJoSduEbt1QkUW2YGaC2CgYxvmF9RyRZS/0/*))",
|
||||
"wsh(sortedmulti(2,[5d5c5649/48h/1h/0h/2h]tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/*,[0b9fb36b/48h/1h/0h/2h]tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/*))#2mdfgjf6",
|
||||
"sh(wsh(sortedmulti(2,[5d5c5649/48h/1h/0h/2h]tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/*,[0b9fb36b/48h/1h/0h/2h]tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/*)))",
|
||||
"sh(wsh(sortedmulti(2,tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/*,[0b9fb36b/48h/1h/0h/2h]tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/*,tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV,tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/10,03d568305d7ce6185f2512472bcad032a672626cf15dc2c6b5f68fdd2f3e5898ef)))",
|
||||
]
|
||||
|
||||
for desc in descs:
|
||||
d = Descriptor.parse(desc, True)
|
||||
|
||||
|
||||
def test_derive():
|
||||
desc = "sh(wsh(sortedmulti(2,tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/*,[0b9fb36b/48h/1h/0h/2h]tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/*,tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV,tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/10,03d568305d7ce6185f2512472bcad032a672626cf15dc2c6b5f68fdd2f3e5898ef)))"
|
||||
d = Descriptor.parse(desc, True)
|
||||
assert (
|
||||
d.derive(1).serialize()
|
||||
== "sh(wsh(sortedmulti(2,[0b9fb36b/48'/1'/0'/2'/10]0249f0282636a8f3fac54a37387686705ddf717ab255cc18d4cc60fff284b8585c,[0b9fb36b/48'/1'/0'/2']02c0ca2aa23a2c83039437973d7eb44d15978900733569583103d03c705aa8383a,[0b9fb36b/48'/1'/0'/2'/0/1]036e5e49573aa861e10c3a01342bc7badcaf8acb88a02aaf4bdae46187260ca262,[0b9fb36b/48'/1'/0'/2']03d568305d7ce6185f2512472bcad032a672626cf15dc2c6b5f68fdd2f3e5898ef,03d7b53b60cbf4c0a9075d65911ce37f759d513f9ac3eca45752256f8aa2d82a9e)))#6n9weurx"
|
||||
)
|
||||
assert (
|
||||
d.derive(1, keep_xpubs=True).serialize()
|
||||
== "sh(wsh(sortedmulti(2,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/10,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/1,[0b9fb36b/48'/1'/0'/2']03d568305d7ce6185f2512472bcad032a672626cf15dc2c6b5f68fdd2f3e5898ef,tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/1)))#7r3s692f"
|
||||
)
|
||||
assert (
|
||||
sort_descriptor(desc, 11)
|
||||
== "sh(wsh(multi(2,tpubDEizCJr6sdiKWC6Be8b5EB7akzS7omSX8CHfAYNYewweRDzjmX2kgDAnig9RcVxqtcxdKuYQSKhkjHecYjyej22b7WThS8r1RBmY3Rfczb9/0/11,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/10,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV,[0b9fb36b/48'/1'/0'/2']tpubDEhAkijE6ovaiWkJpnGnLhW3VJSSbbQeAtRWQro8EaWgNarWFv2TumZ1sj4iBPReCufziRnnb9QSYSEE8tgZQbbaXJTdLtGtQgQTGXEJdfV/0/11,[0b9fb36b/48'/1'/0'/2']03d568305d7ce6185f2512472bcad032a672626cf15dc2c6b5f68fdd2f3e5898ef)))#sys0qqe8"
|
||||
)
|
||||
assert d.address(10) == "2N1TgzrzxjdgkWSuJLoNUtoLhZBJQSakRRk"
|
||||
assert d.address(10, "main") == "39uUw84w8BBQJfGkffkcGrMSLq6Ee4A2f7"
|
||||
Loading…
Add table
Add a link
Reference in a new issue