add multisig support; cc_adjust_hww_keystore now creates new dict - no more inplace adjustments; tests
This commit is contained in:
parent
15e703facb
commit
e69c6bee87
46
ckcc/cli.py
46
ckcc/cli.py
@ -28,9 +28,9 @@ from ckcc.constants import AF_CLASSIC, AF_P2SH, AF_P2WPKH, AF_P2WSH, AF_P2WPKH_P
|
||||
from ckcc.constants import STXN_FINALIZE, STXN_VISUALIZE, STXN_SIGNED
|
||||
from ckcc.client import ColdcardDevice, COINKITE_VID, CKCC_PID
|
||||
from ckcc.sigheader import FW_HEADER_SIZE, FW_HEADER_OFFSET, FW_HEADER_MAGIC
|
||||
from ckcc.utils import dfu_parse, calc_local_pincode, xfp2str, B2A, filepath_append_cc
|
||||
from ckcc.electrum import cc_adjust_hww_keystore
|
||||
|
||||
from ckcc.utils import dfu_parse, calc_local_pincode, xfp2str, B2A
|
||||
from ckcc.electrum import cc_adjust_hww_keystore, filepath_append_cc, is_multisig_wallet, \
|
||||
collect_multisig_hww_keystores_from_wallet, multisig_find_target
|
||||
|
||||
global force_serial
|
||||
force_serial = None
|
||||
@ -1007,13 +1007,21 @@ def get_storage_locker():
|
||||
click.echo(ls)
|
||||
|
||||
|
||||
keystore_keys = ["derivation", "hw_type", "label", "root_fingerprint", "soft_device_id", "xpub"]
|
||||
|
||||
@main.command('coldcardify')
|
||||
@click.argument('file', type=click.Path(exists=True), required=True)
|
||||
@click.option('--outfile', '-o', type=click.Path(),
|
||||
help="output file path where adjusted wallet file is written. "
|
||||
"If this is not specified output is written to <original_file>_cc")
|
||||
@click.option('--dry-run', '-n', default=False, is_flag=True, help="do not write files instead pretty print to console")
|
||||
def electrum_coldcardify(file, outfile, dry_run):
|
||||
@click.option('--key', '-k', type=click.Choice(keystore_keys),
|
||||
help="Multisig wallet keystore dict key based on which to match correct keystore "
|
||||
"(for example hw_type or root_fingerprint). Option required for multisig wallets")
|
||||
@click.option('--val', '-v', type=str, help="Multisig wallet value to match for specified key "
|
||||
"(for example ledger[hw_type] or fffffff0[root_fingerprint])"
|
||||
"Option required for multisig wallets")
|
||||
def electrum_coldcardify(file, outfile, dry_run, key, val):
|
||||
"""
|
||||
Coldcardify electrum wallet file.
|
||||
|
||||
@ -1046,17 +1054,35 @@ def electrum_coldcardify(file, outfile, dry_run):
|
||||
try:
|
||||
# open file only for reading and close it immediately after it is loaded into memory
|
||||
with open(file, "r") as f:
|
||||
contents = json.loads(f.read())
|
||||
wallet = json.loads(f.read())
|
||||
except Exception as e:
|
||||
click.echo("Failed to load wallet file: {}".format(e))
|
||||
sys.exit(1)
|
||||
|
||||
# for now only standard wallet
|
||||
assert contents["wallet_type"] == "standard", "Not a standard wallet"
|
||||
# below line adjust keystore dict in contents dict (in place)
|
||||
cc_adjust_hww_keystore(contents["keystore"], dev)
|
||||
wallet_type = wallet["wallet_type"]
|
||||
try:
|
||||
if wallet_type == "standard":
|
||||
new_keystore = cc_adjust_hww_keystore(wallet["keystore"], dev)
|
||||
wallet["keystore"] = new_keystore
|
||||
elif is_multisig_wallet(wallet):
|
||||
if key is None and val is None:
|
||||
click.echo("--key and --val have to be specified for multisig wallets")
|
||||
sys.exit(1)
|
||||
k, keystore = multisig_find_target(
|
||||
keystores=collect_multisig_hww_keystores_from_wallet(wallet),
|
||||
key=key,
|
||||
value=val,
|
||||
)
|
||||
new_keystore = cc_adjust_hww_keystore(keystore, dev)
|
||||
wallet[k] = new_keystore
|
||||
else:
|
||||
click.echo("Unsupported wallet type: {}".format(wallet_type))
|
||||
sys.exit(1)
|
||||
except RuntimeError as e:
|
||||
click.echo("Failed to adjust keystore: {}".format(e))
|
||||
sys.exit(1)
|
||||
|
||||
content_str = json.dumps(contents, indent=4)
|
||||
content_str = json.dumps(wallet, indent=4)
|
||||
if dry_run:
|
||||
click.echo(content_str)
|
||||
else:
|
||||
|
||||
@ -1,40 +1,98 @@
|
||||
# (c) Copyright 2021 by Coinkite Inc. This file is covered by license found in COPYING-CC.
|
||||
#
|
||||
import re
|
||||
import os
|
||||
import copy
|
||||
|
||||
from ckcc.utils import xfp2str
|
||||
from ckcc.client import ColdcardDevice
|
||||
from ckcc.protocol import CCProtocolPacker
|
||||
|
||||
|
||||
def cc_adjust_hww_keystore(keystore: dict, dev: ColdcardDevice = None):
|
||||
"""Modify electrum keystore dictionary in place"""
|
||||
assert keystore["type"] == "hardware", "Not a hardware wallet type"
|
||||
MULTISIG_WALLET_TYPE_PATTERN = r"^\d+of\d+$"
|
||||
MULTISIG_WALLET_KEY_PATTERN = r"^x\d+/$"
|
||||
|
||||
|
||||
def is_hww_keystore(keystore: dict) -> bool:
|
||||
return keystore["type"] == "hardware"
|
||||
|
||||
|
||||
def is_multisig_wallet(wallet: dict) -> bool:
|
||||
if re.match(MULTISIG_WALLET_TYPE_PATTERN, wallet["wallet_type"]):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def collect_multisig_hww_keystores_from_wallet(wallet: dict) -> dict:
|
||||
"""Find all hardware keystore objects in multisig wallet dict"""
|
||||
if not is_multisig_wallet(wallet):
|
||||
raise RuntimeError("Not an electrum multisig wallet")
|
||||
return {
|
||||
key: value
|
||||
for key, value in wallet.items()
|
||||
if re.match(MULTISIG_WALLET_KEY_PATTERN, key)
|
||||
if is_hww_keystore(value)
|
||||
}
|
||||
|
||||
|
||||
def multisig_find_target(keystores: dict, key: str, value: str) -> tuple:
|
||||
"""Find target keystore in list of keystores by key equals value"""
|
||||
result = [
|
||||
(k, keystore)
|
||||
for k, keystore in keystores.items()
|
||||
if keystore.get(key, None) == value
|
||||
]
|
||||
if len(result) != 1:
|
||||
# if this is true, we have found more than one keystore and therefore
|
||||
# key value pair is ambiguous
|
||||
raise RuntimeError(
|
||||
"Found {} keystores. Provided key/value is ambiguous".format(len(result))
|
||||
)
|
||||
return result[0]
|
||||
|
||||
|
||||
def filepath_append_cc(f_path):
|
||||
"""Append '_cc' suffix to file path. Do consider one file extension"""
|
||||
dirname = os.path.dirname(f_path)
|
||||
filename, file_ext = os.path.splitext(os.path.basename(f_path))
|
||||
result = os.path.join(dirname, "{}_cc".format(filename) + file_ext)
|
||||
return result
|
||||
|
||||
|
||||
def cc_adjust_hww_keystore(keystore: dict, dev: ColdcardDevice = None) -> dict:
|
||||
"""Create new updated version of keystore"""
|
||||
new_keystore = copy.deepcopy(keystore)
|
||||
|
||||
if not is_hww_keystore(keystore):
|
||||
raise RuntimeError("Not a hardware wallet type")
|
||||
|
||||
# 1-3 can be done without coldcard connected
|
||||
#
|
||||
# 1. change hw type to coldcard
|
||||
keystore["hw_type"] = "coldcard"
|
||||
new_keystore["hw_type"] = "coldcard"
|
||||
# 2. soft device id should be nullified
|
||||
keystore["soft_device_id"] = None
|
||||
new_keystore["soft_device_id"] = None
|
||||
# 3. remove cfg key if exists (ledger specific)
|
||||
if "cfg" in keystore:
|
||||
del keystore["cfg"]
|
||||
if "cfg" in new_keystore:
|
||||
del new_keystore["cfg"]
|
||||
# 4. label ? we can do something about it - at least remove the label that is no longer in use
|
||||
keystore["label"] = "Coldcard {}".format(keystore["root_fingerprint"])
|
||||
new_keystore["label"] = "Coldcard {}".format(new_keystore["root_fingerprint"])
|
||||
|
||||
# for next steps we need coldcard connected (unnecessary)
|
||||
if dev:
|
||||
# 4. label Coldcard + fingerprint
|
||||
xfp = dev.master_fingerprint
|
||||
xfp = xfp2str(xfp).lower() # if any letters - lower them
|
||||
if xfp != keystore["root_fingerprint"]:
|
||||
if xfp != new_keystore["root_fingerprint"]:
|
||||
raise RuntimeError(
|
||||
"Fingerprint missmatch! Is this a correct coldcard/wallet file?"
|
||||
" Make sure that your bip39 passphrase is in effect (if used)."
|
||||
" device fingerprint {}; wallet fingerprint {}".format(xfp, keystore["root_fingerprint"])
|
||||
" device fingerprint {}; wallet fingerprint {}".format(xfp, new_keystore["root_fingerprint"])
|
||||
)
|
||||
|
||||
label = "Coldcard {}".format(xfp)
|
||||
keystore["label"] = label
|
||||
new_keystore["label"] = label
|
||||
# 5. ckcc xpub (master xpub)
|
||||
master_ext_pubkey = dev.send_recv(CCProtocolPacker.get_xpub("m"), timeout=None)
|
||||
keystore["ckcc_xpub"] = master_ext_pubkey
|
||||
new_keystore["ckcc_xpub"] = master_ext_pubkey
|
||||
return new_keystore
|
||||
|
||||
@ -118,12 +118,4 @@ def calc_local_pincode(psbt_sha, next_local_code):
|
||||
|
||||
return '%06d' % (num % 1000000)
|
||||
|
||||
|
||||
def filepath_append_cc(f_path):
|
||||
"""Append '_cc' suffix to file path. Do consider one file extension"""
|
||||
dirname = os.path.dirname(f_path)
|
||||
filename, file_ext = os.path.splitext(os.path.basename(f_path))
|
||||
result = os.path.join(dirname, "{}_cc".format(filename) + file_ext)
|
||||
return result
|
||||
|
||||
# EOF
|
||||
|
||||
@ -1,9 +1,13 @@
|
||||
import os
|
||||
import json
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from ckcc.cli import electrum_coldcardify
|
||||
from ckcc.utils import filepath_append_cc
|
||||
from ckcc.electrum import (
|
||||
filepath_append_cc, multisig_find_target, collect_multisig_hww_keystores_from_wallet,
|
||||
cc_adjust_hww_keystore
|
||||
)
|
||||
|
||||
|
||||
# expecting cwd to be ckcc-protocol
|
||||
@ -11,6 +15,10 @@ test_data_path = os.path.join("tests", "test_data")
|
||||
encrypted_path = os.path.join(test_data_path, "encrypted")
|
||||
ledger_path = os.path.join(test_data_path, "ledger")
|
||||
trezor_path = os.path.join(test_data_path, "trezor.json")
|
||||
bip32_path = os.path.join(test_data_path, "bip32_wallet")
|
||||
a2fa_path = os.path.join(test_data_path, "2fa_wallet")
|
||||
import_path = os.path.join(test_data_path, "import_wallet")
|
||||
multi3of5_path = os.path.join(test_data_path, "multi3of5")
|
||||
|
||||
|
||||
def assert_keystore(keystore):
|
||||
@ -22,6 +30,41 @@ def assert_keystore(keystore):
|
||||
assert "ckcc_xpub" not in keystore
|
||||
|
||||
|
||||
def test_multisig_find_target():
|
||||
for pth in [ledger_path, trezor_path, bip32_path, a2fa_path]:
|
||||
with open(pth, "r") as f:
|
||||
wallet = json.loads(f.read())
|
||||
with pytest.raises(RuntimeError):
|
||||
multisig_find_target(
|
||||
keystores=collect_multisig_hww_keystores_from_wallet(wallet),
|
||||
key="hw_type", # key and value does not matter here as exc is raised before they can be used
|
||||
value="ledger"
|
||||
)
|
||||
|
||||
with open(multi3of5_path, "r") as f:
|
||||
wallet = json.loads(f.read())
|
||||
keystores = collect_multisig_hww_keystores_from_wallet(wallet)
|
||||
assert len(keystores) == 2 # only two are hardware
|
||||
for hw_type in ["trezor", "ledger"]:
|
||||
key, keystore = multisig_find_target(
|
||||
keystores=keystores,
|
||||
key="hw_type",
|
||||
value=hw_type
|
||||
)
|
||||
assert wallet[key] == keystore
|
||||
new_keystore = cc_adjust_hww_keystore(keystore)
|
||||
assert_keystore(new_keystore)
|
||||
|
||||
|
||||
def test_filepath_append_cc():
|
||||
assert filepath_append_cc("ledger_wallet") == "ledger_wallet_cc"
|
||||
assert filepath_append_cc("ledger_wallet.json") == "ledger_wallet_cc.json"
|
||||
assert filepath_append_cc("ledger wallet.json") == "ledger wallet_cc.json"
|
||||
assert filepath_append_cc("/ledger_wallet") == "/ledger_wallet_cc"
|
||||
assert filepath_append_cc("/tmp/.../ledger_wallet") == "/tmp/.../ledger_wallet_cc"
|
||||
assert filepath_append_cc("/user/local/h.ledger.wallet") == "/user/local/h.ledger_cc.wallet"
|
||||
|
||||
|
||||
def test_encrypted():
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(electrum_coldcardify, [encrypted_path, "--dry-run"])
|
||||
@ -72,3 +115,40 @@ def test_no_options():
|
||||
res = json.loads(f.read())
|
||||
assert_keystore(res["keystore"])
|
||||
os.remove(new_pth)
|
||||
|
||||
|
||||
def test_not_hww_wallet():
|
||||
runner = CliRunner()
|
||||
for pth in [bip32_path]:
|
||||
result = runner.invoke(electrum_coldcardify, [pth])
|
||||
assert result.exit_code == 1
|
||||
assert result.output == "Failed to adjust keystore: Not a hardware wallet type\n"
|
||||
|
||||
|
||||
def test_not_standard_wallet():
|
||||
runner = CliRunner()
|
||||
for name, pth in [("2fa", a2fa_path), ("imported", import_path)]:
|
||||
result = runner.invoke(electrum_coldcardify, [pth])
|
||||
assert result.exit_code == 1
|
||||
assert result.output == "Unsupported wallet type: {}\n".format(name)
|
||||
|
||||
|
||||
def test_multisig():
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(electrum_coldcardify, [multi3of5_path, "-k", "hw_type", "-v", "ledger"])
|
||||
assert result.exit_code == 0
|
||||
new_pth = filepath_append_cc(multi3of5_path)
|
||||
assert "New wallet file created: {}\n".format(new_pth) == result.output
|
||||
with open(new_pth, "r") as f:
|
||||
res = json.loads(f.read())
|
||||
# ledger is x2/ entry
|
||||
assert_keystore(res["x2/"])
|
||||
# make sure other entries are unchanged
|
||||
assert res["x3/"]["hw_type"] == "trezor"
|
||||
os.remove(new_pth)
|
||||
|
||||
result = runner.invoke(electrum_coldcardify, [multi3of5_path, "-k", "root_fingerprint", "-v", "7633218e"])
|
||||
assert result.exit_code == 1
|
||||
assert "Found 2 keystores" in result.output
|
||||
|
||||
|
||||
|
||||
@ -1,10 +0,0 @@
|
||||
from ckcc.utils import filepath_append_cc
|
||||
|
||||
|
||||
def test_filepath_append_cc():
|
||||
assert filepath_append_cc("ledger_wallet") == "ledger_wallet_cc"
|
||||
assert filepath_append_cc("ledger_wallet.json") == "ledger_wallet_cc.json"
|
||||
assert filepath_append_cc("ledger wallet.json") == "ledger wallet_cc.json"
|
||||
assert filepath_append_cc("/ledger_wallet") == "/ledger_wallet_cc"
|
||||
assert filepath_append_cc("/tmp/.../ledger_wallet") == "/tmp/.../ledger_wallet_cc"
|
||||
assert filepath_append_cc("/user/local/h.ledger.wallet") == "/user/local/h.ledger_cc.wallet"
|
||||
Loading…
Reference in New Issue
Block a user