From 5228ee370bad72efd98d27e2181fb1ee1c27bae2 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 17:51:42 +0200 Subject: [PATCH 01/19] tests: remove mistakenly committed file --- tests/test_framework/revault_network.py | 811 ------------------------ 1 file changed, 811 deletions(-) delete mode 100644 tests/test_framework/revault_network.py diff --git a/tests/test_framework/revault_network.py b/tests/test_framework/revault_network.py deleted file mode 100644 index 0a840f85..00000000 --- a/tests/test_framework/revault_network.py +++ /dev/null @@ -1,811 +0,0 @@ -import bip32 -import logging -import os -import random - -from ephemeral_port_reserve import reserve -from nacl.public import PrivateKey as Curve25519Private -from test_framework import serializations -from test_framework.bitcoind import BitcoindRpcProxy -from test_framework.coordinatord import Coordinatord -from test_framework.cosignerd import Cosignerd -from test_framework.miradord import Miradord -from test_framework.revaultd import ManagerRevaultd, StakeholderRevaultd, StkManRevaultd -from test_framework.utils import ( - get_descriptors, - get_participants, - finalize_input, - wait_for, - TIMEOUT, - WT_PLUGINS_DIR, -) - - -class RevaultNetwork: - # FIXME: we use a single bitcoind for all the wallets because it's much - # more efficient. Eventually, we may have to test with separate ones. - def __init__( - self, - root_dir, - bitcoind, - executor, - postgres_user, - postgres_pass, - postgres_host="localhost", - ): - self.root_dir = root_dir - self.bitcoind = bitcoind - self.daemons = [] - - self.executor = executor - - self.postgres_user = postgres_user - self.postgres_pass = postgres_pass - self.postgres_host = postgres_host - self.coordinator_port = reserve() - - self.stk_wallets = [] - self.stkman_wallets = [] - self.man_wallets = [] - - self.csv = None - self.emergency_address = None - - self.bitcoind_proxy = None - - def deploy( - self, - n_stakeholders, - n_managers, - n_stkmanagers=0, - csv=None, - managers_threshold=None, - with_cosigs=True, - with_watchtowers=True, - with_cpfp=True, - bitcoind_rpc_mocks=[], - ): - """ - Deploy a revault setup with {n_stakeholders} stakeholders, {n_managers} - managers. - """ - # They didn't provide it, defaults to n_managers - # PS: No I can't just managers_threshold=n_managers in the method's signature :( - if managers_threshold == None: - managers_threshold = n_managers + n_stkmanagers - - assert n_stakeholders + n_stkmanagers >= 2, "Not enough stakeholders" - assert n_managers + n_stkmanagers >= 1, "Not enough managers" - assert managers_threshold <= n_managers + n_stkmanagers, "Invalid threshold" - - # Connection info to bitcoind. Change the port depending on whether we are proxying - # the daemons' requests. - bitcoind_cookie = os.path.join(self.bitcoind.bitcoin_dir, "regtest", ".cookie") - if len(bitcoind_rpc_mocks) > 0: - self.bitcoind_proxy = BitcoindRpcProxy( - self.bitcoind.rpcport, bitcoind_cookie, bitcoind_rpc_mocks - ) - bitcoind_rpcport = self.bitcoind_proxy.rpcport - else: - bitcoind_rpcport = self.bitcoind.rpcport - - ( - stkonly_keychains, - stkonly_cosig_keychains, - manonly_keychains, - stkman_stk_keychains, - stkman_cosig_keychains, - stkman_man_keychains, - ) = get_participants(n_stakeholders, n_managers, n_stkmanagers, with_cosigs) - stks_keychains = stkonly_keychains + stkman_stk_keychains - cosigs_keychains = stkonly_cosig_keychains + stkman_cosig_keychains - mans_keychains = manonly_keychains + stkman_man_keychains - - if csv is None: - # Not more than 6 months - csv = random.randint(1, 26784) - self.csv = csv - - man_cpfp_seeds = [os.urandom(32) for _ in range(len(manonly_keychains))] - man_cpfp_privs = [ - bip32.BIP32.from_seed(seed, network="test") for seed in man_cpfp_seeds - ] - stkman_cpfp_seeds = [os.urandom(32) for _ in range(len(stkman_man_keychains))] - stkman_cpfp_privs = [ - bip32.BIP32.from_seed(seed, network="test") for seed in stkman_cpfp_seeds - ] - cpfp_xpubs = [c.get_xpub() for c in man_cpfp_privs + stkman_cpfp_privs] - stks_xpubs = [stk.get_xpub() for stk in stks_keychains] - cosigs_keys = [cosig.get_static_key().hex() for cosig in cosigs_keychains] - mans_xpubs = [man.get_xpub() for man in mans_keychains] - (self.deposit_desc, self.unvault_desc, self.cpfp_desc) = get_descriptors( - stks_xpubs, cosigs_keys, mans_xpubs, managers_threshold, cpfp_xpubs, csv - ) - # Generate a dummy 2of2 to be used as our Emergency address - desc = "wsh(multi(2,cRE7qAArQYnFQK7S1gXFTArFT4UWvh8J2v2EUajRWXbWFvRzxoeF,\ - cTzcgRCmHNqUqZuZgvCPLUDXXrQSoVQpZiXQZWQzsLEytcTr6iXi))" - checksum = self.bitcoind.rpc.getdescriptorinfo(desc)["checksum"] - desc = f"{desc}#{checksum}" - self.emergency_address = self.bitcoind.rpc.deriveaddresses(desc)[0] - desc_import = self.bitcoind.rpc.importdescriptors( - [ - { - "desc": desc, - "timestamp": "now", - "label": "revault-emergency", - } - ] - ) - if not desc_import[0]["success"]: - raise Exception(desc_import) - - # FIXME: this is getting dirty.. We should re-centralize information - # about each participant in specified data structures - stkonly_cosigners_ports = [] - stkman_cosigners_ports = [] - - # The Noise keys are interdependant, so generate everything in advance - # to avoid roundtrips - coordinator_noisepriv = os.urandom(32) - coordinator_noisepub = bytes( - Curve25519Private(coordinator_noisepriv).public_key - ) - - (stkonly_noiseprivs, stkonly_noisepubs) = ([], []) - (stkonly_wt_noiseprivs, stkonly_wt_noisepubs) = ([], []) - (stkonly_cosig_noiseprivs, stkonly_cosig_noisepubs) = ([], []) - for i in range(len(stkonly_keychains)): - stkonly_noiseprivs.append(os.urandom(32)) - stkonly_noisepubs.append( - bytes(Curve25519Private(stkonly_noiseprivs[i]).public_key) - ) - if with_cosigs: - stkonly_cosig_noiseprivs.append(os.urandom(32)) - stkonly_cosig_noisepubs.append( - bytes(Curve25519Private(stkonly_cosig_noiseprivs[i]).public_key) - ) - # Unused yet - stkonly_wt_noiseprivs.append(os.urandom(32)) - stkonly_wt_noisepubs.append( - bytes(Curve25519Private(stkonly_wt_noiseprivs[i]).public_key) - ) - - (stkman_noiseprivs, stkman_noisepubs) = ([], []) - (stkman_wt_noiseprivs, stkman_wt_noisepubs) = ([], []) - (stkman_cosig_noiseprivs, stkman_cosig_noisepubs) = ([], []) - for i in range(len(stkman_stk_keychains)): - stkman_noiseprivs.append(os.urandom(32)) - stkman_noisepubs.append( - bytes(Curve25519Private(stkman_noiseprivs[i]).public_key) - ) - if with_cosigs: - stkman_cosig_noiseprivs.append(os.urandom(32)) - stkman_cosig_noisepubs.append( - bytes(Curve25519Private(stkman_cosig_noiseprivs[i]).public_key) - ) - # Unused yet - stkman_wt_noiseprivs.append(os.urandom(32)) - stkman_wt_noisepubs.append( - bytes(Curve25519Private(stkman_wt_noiseprivs[i]).public_key) - ) - - (man_noiseprivs, man_noisepubs) = ([], []) - for i in range(len(manonly_keychains)): - man_noiseprivs.append(os.urandom(32)) - man_noisepubs.append(bytes(Curve25519Private(man_noiseprivs[i]).public_key)) - - logging.debug( - f"Using Noise pubkeys:\n- Stakeholders: {stkonly_noisepubs + stkman_noisepubs}" - f" (of which {len(stkman_noisepubs)} are also managers)" - f"\n- Managers: {man_noisepubs}\n- Watchtowers:" - f"{stkonly_wt_noisepubs + stkman_wt_noisepubs}\n" - ) - - # Spin up the "Sync Server" - coord_datadir = os.path.join(self.root_dir, "coordinatord") - os.makedirs(coord_datadir, exist_ok=True) - coordinatord = Coordinatord( - coord_datadir, - coordinator_noisepriv, - man_noisepubs + stkman_noisepubs, - stkonly_noisepubs + stkman_noisepubs, - stkonly_wt_noisepubs + stkman_wt_noisepubs, - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - self.postgres_user, - self.postgres_pass, - self.postgres_host, - ) - coordinatord.start() - self.daemons.append(coordinatord) - - cosigners_info = [] - for (i, noisepub) in enumerate(stkonly_cosig_noisepubs): - stkonly_cosigners_ports.append(reserve()) - cosigners_info.append( - { - "host": f"127.0.0.1:{stkonly_cosigners_ports[i]}", - "noise_key": noisepub, - } - ) - for (i, noisepub) in enumerate(stkman_cosig_noisepubs): - stkman_cosigners_ports.append(reserve()) - cosigners_info.append( - { - "host": f"127.0.0.1:{stkman_cosigners_ports[i]}", - "noise_key": noisepub, - } - ) - - # Start daemons in parallel, as it takes a few seconds for each - start_jobs = [] - # By default the watchtower should not revault anything - default_wt_plugin = { - "path": os.path.join(WT_PLUGINS_DIR, "revault_nothing.py"), - "conf": {}, - } - - # Spin up the stakeholders wallets and their cosigning servers - for i, stk in enumerate(stkonly_keychains): - if with_watchtowers: - datadir = os.path.join(self.root_dir, f"miradord-{i}") - os.makedirs(datadir) - wt_listen_port = reserve() - miradord = Miradord( - datadir, - str(self.deposit_desc), - str(self.unvault_desc), - str(self.cpfp_desc), - self.emergency_address, - wt_listen_port, - stkonly_wt_noiseprivs[i], - stkonly_noisepubs[i].hex(), - coordinator_noisepub.hex(), - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - plugins=[default_wt_plugin], - ) - start_jobs.append(self.executor.submit(miradord.start)) - self.daemons.append(miradord) - - datadir = os.path.join(self.root_dir, f"revaultd-stk-{i}") - os.makedirs(datadir, exist_ok=True) - stk_config = { - "keychain": stk, - "watchtowers": [ - { - "host": f"127.0.0.1:{wt_listen_port}", - "noise_key": stkonly_wt_noisepubs[i].hex(), - } - ] - if with_watchtowers - else [], - "emergency_address": self.emergency_address, - } - - revaultd = StakeholderRevaultd( - datadir, - str(self.deposit_desc), - str(self.unvault_desc), - str(self.cpfp_desc), - stkonly_noiseprivs[i], - coordinator_noisepub.hex(), - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - stk_config, - wt_process=miradord if with_watchtowers else None, - ) - start_jobs.append(self.executor.submit(revaultd.start)) - self.stk_wallets.append(revaultd) - - if with_cosigs: - datadir = os.path.join(self.root_dir, f"cosignerd-stk-{i}") - os.makedirs(datadir, exist_ok=True) - - cosignerd = Cosignerd( - datadir, - stkonly_cosig_noiseprivs[i], - stkonly_cosig_keychains[i].get_bitcoin_priv(), - stkonly_cosigners_ports[i], - man_noisepubs + stkman_noisepubs, - ) - start_jobs.append(self.executor.submit(cosignerd.start)) - self.daemons.append(cosignerd) - - # Spin up the stakeholder-managers wallets and their cosigning servers - for i, stkman in enumerate(stkman_stk_keychains): - if with_watchtowers: - datadir = os.path.join(self.root_dir, f"miradord-stkman-{i}") - os.makedirs(datadir) - wt_listen_port = reserve() - miradord = Miradord( - datadir, - str(self.deposit_desc), - str(self.unvault_desc), - str(self.cpfp_desc), - self.emergency_address, - wt_listen_port, - stkman_wt_noiseprivs[i], - stkman_noisepubs[i].hex(), - coordinator_noisepub.hex(), - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - plugins=[default_wt_plugin], - ) - start_jobs.append(self.executor.submit(miradord.start)) - self.daemons.append(miradord) - - datadir = os.path.join(self.root_dir, f"revaultd-stkman-{i}") - os.makedirs(datadir, exist_ok=True) - stk_config = { - "keychain": stkman, - "watchtowers": [ - { - "host": f"127.0.0.1:{wt_listen_port}", - "noise_key": stkman_wt_noisepubs[i].hex(), - } - ] - if with_watchtowers - else [], - "emergency_address": self.emergency_address, - } - man_config = { - "keychain": stkman_man_keychains[i], - "cosigners": cosigners_info, - } - - revaultd = StkManRevaultd( - datadir, - str(self.deposit_desc), - str(self.unvault_desc), - str(self.cpfp_desc), - stkman_noiseprivs[i], - coordinator_noisepub.hex(), - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - stk_config, - man_config, - wt_process=miradord if with_watchtowers else None, - cpfp_seed=stkman_cpfp_seeds[i] if with_cpfp else None, - ) - start_jobs.append(self.executor.submit(revaultd.start)) - self.stkman_wallets.append(revaultd) - - if with_cosigs: - datadir = os.path.join(self.root_dir, f"cosignerd-stkman-{i}") - os.makedirs(datadir, exist_ok=True) - - cosignerd = Cosignerd( - datadir, - stkman_cosig_noiseprivs[i], - stkman_cosig_keychains[i].get_bitcoin_priv(), - stkman_cosigners_ports[i], - man_noisepubs + stkman_noisepubs, - ) - start_jobs.append(self.executor.submit(cosignerd.start)) - self.daemons.append(cosignerd) - - # Spin up the managers (only) wallets - for i, man in enumerate(manonly_keychains): - datadir = os.path.join(self.root_dir, f"revaultd-man-{i}") - os.makedirs(datadir, exist_ok=True) - - man_config = {"keychain": man, "cosigners": cosigners_info} - daemon = ManagerRevaultd( - datadir, - str(self.deposit_desc), - str(self.unvault_desc), - str(self.cpfp_desc), - man_noiseprivs[i], - coordinator_noisepub.hex(), - self.coordinator_port, - bitcoind_rpcport, - bitcoind_cookie, - man_config, - cpfp_seed=man_cpfp_seeds[i] if with_cpfp else None, - ) - start_jobs.append(self.executor.submit(daemon.start)) - self.man_wallets.append(daemon) - - for j in start_jobs: - j.result(TIMEOUT) - - self.daemons += self.stk_wallets + self.stkman_wallets + self.man_wallets - - def mans(self): - return self.stkman_wallets + self.man_wallets - - def stks(self): - return self.stkman_wallets + self.stk_wallets - - def participants(self): - return self.stkman_wallets + self.stk_wallets + self.man_wallets - - def man(self, n): - """Get the {n}th manager (including the stakeholder-managers first)""" - mans = self.stkman_wallets + self.man_wallets - return mans[n] - - def stk(self, n): - """Get the {n}th stakeholder (including the stakeholder-managers first)""" - stks = self.stkman_wallets + self.stk_wallets - return stks[n] - - def signed_unvault_psbt(self, deposit, derivation_index): - """Get the fully-signed Unvault transaction for this deposit. - - This will raise if we don't have all the signatures. - """ - psbt_str = self.stks()[0].rpc.listpresignedtransactions([deposit])[ - "presigned_transactions" - ][0]["unvault"] - psbt = serializations.PSBT() - psbt.deserialize(psbt_str) - - finalize_input(self.deposit_desc, psbt.inputs[0], derivation_index) - psbt.tx.wit.vtxinwit.append(psbt.inputs[0].final_script_witness) - return psbt.tx.serialize_with_witness().hex() - - def signed_cancel_psbt(self, deposit, derivation_index): - """Get the fully-signed Cancel transaction for this deposit. - - This picks the lowest feerate version. - This will raise if we don't have all the signatures. - """ - psbt_str = self.stks()[0].rpc.listpresignedtransactions([deposit])[ - "presigned_transactions" - ][0]["cancel"][0] - psbt = serializations.PSBT() - psbt.deserialize(psbt_str) - - finalize_input(self.unvault_desc, psbt.inputs[0], derivation_index) - psbt.tx.wit.vtxinwit.append(psbt.inputs[0].final_script_witness) - return psbt.tx.serialize_with_witness().hex() - - def get_vault(self, address): - """Get a vault entry by outpoint or by address""" - for v in self.man(0).rpc.listvaults()["vaults"]: - if v["address"] == address: - return v - - def fund(self, amount=None): - """Deposit coins into the architectures, by paying to the deposit - descriptor and getting the tx 6 blocks confirmations.""" - assert ( - len(self.man_wallets + self.stkman_wallets) > 0 - ), "You must have deploy()ed first" - - man = self.man(0) - - if amount is None: - amount = 49.9999 - - addr = man.rpc.getdepositaddress()["address"] - txid = self.bitcoind.rpc.sendtoaddress(addr, amount) - man.wait_for_log(f"Got a new unconfirmed deposit at {txid}") - self.bitcoind.generate_block(6, wait_for_mempool=txid) - man.wait_for_log(f"Vault at {txid}.* is now confirmed") - - vaults = man.rpc.listvaults(["funded"])["vaults"] - for v in vaults: - if v["txid"] == txid: - for w in self.man_wallets + self.stk_wallets: - w.wait_for_deposits([f"{txid}:{v['vout']}"]) - return v - - raise Exception(f"Vault created by '{txid}' got in logs but not in listvaults?") - - def fundmany(self, amounts=[]): - """Deposit coins into the architectures in a single transaction""" - assert ( - len(self.man_wallets + self.stkman_wallets) > 0 - ), "You must have deploy()ed first" - assert len(amounts) > 0, "You must provide at least an amount!" - - man = self.man(0) - - curr_index = 0 - vaults = man.rpc.listvaults()["vaults"] - for v in vaults: - if v["derivation_index"] > curr_index: - curr_index = v["derivation_index"] - - indexes = list(range(curr_index + 1, curr_index + 1 + len(amounts))) - amounts_sendmany = {} - for i, amount in enumerate(amounts): - amounts_sendmany[man.rpc.getdepositaddress(indexes[i])["address"]] = amount - - txid = self.bitcoind.rpc.sendmany("", amounts_sendmany) - man.wait_for_logs( - [f"Got a new unconfirmed deposit at {txid}" for _ in range(len(amounts))], - timeout=TIMEOUT * max(1, len(amounts) / 10), - ) - self.bitcoind.generate_block(6, wait_for_mempool=txid) - man.wait_for_logs( - [f"Vault at {txid}.* is now confirmed" for _ in range(len(amounts))], - timeout=TIMEOUT * max(1, len(amounts) / 10), - ) - - # Return the vaults we created - all_vaults = man.rpc.listvaults(["funded"])["vaults"] - created_vaults = [] - for v in all_vaults: - if v["txid"] == txid: - created_vaults.append(v) - assert len(created_vaults) == len(amounts) - - return created_vaults - - def secure_vault(self, vault): - """Make all stakeholders share signatures for all revocation txs""" - deposit = f"{vault['txid']}:{vault['vout']}" - for stk in self.stks(): - stk.wait_for_deposits([deposit]) - psbts = stk.rpc.getrevocationtxs(deposit) - cancel_psbts = [ - stk.stk_keychain.sign_revocation_psbt(c, vault["derivation_index"]) - for c in psbts["cancel_txs"] - ] - emer_psbt = stk.stk_keychain.sign_revocation_psbt( - psbts["emergency_tx"], vault["derivation_index"] - ) - unemer_psbt = stk.stk_keychain.sign_revocation_psbt( - psbts["emergency_unvault_tx"], vault["derivation_index"] - ) - stk.rpc.revocationtxs(deposit, cancel_psbts, emer_psbt, unemer_psbt) - for w in self.participants(): - w.wait_for_secured_vaults([deposit]) - - def secure_vaults(self, vaults): - """Secure all these vaults, concurrently.""" - sec_jobs = [] - for v in vaults: - sec_jobs.append(self.executor.submit(self.secure_vault, v)) - for j in sec_jobs: - j.result(TIMEOUT) - - def activate_vault(self, vault): - """Make all stakeholders share signatures for the unvault tx""" - deposit = f"{vault['txid']}:{vault['vout']}" - for stk in self.stks(): - stk.wait_for_secured_vaults([deposit]) - unvault_psbt = stk.rpc.getunvaulttx(deposit)["unvault_tx"] - unvault_psbt = stk.stk_keychain.sign_unvault_psbt( - unvault_psbt, vault["derivation_index"] - ) - stk.rpc.unvaulttx(deposit, unvault_psbt) - for w in self.participants(): - w.wait_for_active_vaults([deposit]) - - def activate_fresh_vaults(self, vaults): - """Secure then activate all these vaults, concurrently.""" - # TODO: i'm sure we don't even need to wait for all sec jobs to be complete - # before starting the activate_vault futures, given a high enough TIMEOUT. - self.secure_vaults(vaults) - - act_jobs = [] - for v in vaults: - act_jobs.append(self.executor.submit(self.activate_vault, v)) - for j in act_jobs: - j.result(TIMEOUT) - - def broadcast_unvaults(self, vaults, destinations, feerate, priority=False): - """ - Broadcast the Unvault transactions for these {vaults}, advertizing a - Spend tx spending to these {destinations} (mapping of addresses to - amounts) - """ - man = self.man(0) - deposits = [] - deriv_indexes = [] - for v in vaults: - deposits.append(f"{v['txid']}:{v['vout']}") - deriv_indexes.append(v["derivation_index"]) - man.wait_for_active_vaults(deposits) - - spend_tx = man.rpc.getspendtx(deposits, destinations, feerate)["spend_tx"][ - "psbt" - ] - for man in self.mans(): - spend_tx = man.man_keychain.sign_spend_psbt(spend_tx, deriv_indexes) - man.rpc.updatespendtx(spend_tx) - - spend_psbt = serializations.PSBT() - spend_psbt.deserialize(spend_tx) - spend_psbt.tx.calc_sha256() - man.rpc.setspendtx(spend_psbt.tx.hash, priority) - return spend_psbt - - def unvault_vaults(self, vaults, destinations, feerate, priority=False): - """ - Unvault these {vaults}, advertizing a Spend tx spending to these {destinations} - (mapping of addresses to amounts) - """ - spend_psbt = self.broadcast_unvaults(vaults, destinations, feerate, priority) - deposits = [f"{v['txid']}:{v['vout']}" for v in vaults] - self.bitcoind.generate_block(1, wait_for_mempool=len(deposits)) - for w in self.participants(): - wait_for( - lambda: len(w.rpc.listvaults(["unvaulted"], deposits)["vaults"]) - == len(deposits) - ) - return spend_psbt - - def spend_vaults_unconfirmed(self, vaults, destinations, feerate, priority=False): - """ - Spend these {vaults} to these {destinations} (mapping of addresses to amounts), not - confirming the Spend transaction. - Make sure to call this only with revault deployment with a low (<500) CSV, or you'll encounter - an ugly timeout from bitcoinlib. - - :return: the list of spent deposits along with the Spend PSBT. - """ - assert len(vaults) > 0 - man = self.man(0) - deposits = [] - deriv_indexes = [] - for v in vaults: - deposits.append(f"{v['txid']}:{v['vout']}") - deriv_indexes.append(v["derivation_index"]) - - for man in self.mans(): - man.wait_for_active_vaults(deposits) - - spend_tx = man.rpc.getspendtx(deposits, destinations, feerate)["spend_tx"][ - "psbt" - ] - for man in self.mans(): - spend_tx = man.man_keychain.sign_spend_psbt(spend_tx, deriv_indexes) - man.rpc.updatespendtx(spend_tx) - - spend_psbt = serializations.PSBT() - spend_psbt.deserialize(spend_tx) - spend_psbt.tx.calc_sha256() - man.rpc.setspendtx(spend_psbt.tx.hash, priority) - - self.bitcoind.generate_block(1, wait_for_mempool=len(deposits)) - self.bitcoind.generate_block(self.csv) - man.wait_for_log( - f"Succesfully broadcasted Spend tx '{spend_psbt.tx.hash}'", - ) - for w in self.participants(): - wait_for( - lambda: len(w.rpc.listvaults(["spending"], deposits)["vaults"]) - == len(deposits) - ) - - return deposits, spend_psbt - - def spend_vaults(self, vaults, destinations, feerate, priority=False): - """ - Spend these {vaults} to these {destinations} (mapping of addresses to amounts). - Make sure to call this only with revault deployment with a low (<500) CSV, or you'll encounter - an ugly timeout from bitcoinlib. - - :return: the list of spent deposits along with the Spend PSBT. - """ - deposits, spend_psbt = self.spend_vaults_unconfirmed( - vaults, destinations, feerate, priority - ) - - self.bitcoind.generate_block(1, wait_for_mempool=[spend_psbt.tx.hash]) - for w in self.participants(): - wait_for( - lambda: len(w.rpc.listvaults(["spent"], deposits)["vaults"]) - == len(deposits) - ) - - return deposits, spend_psbt.tx.hash - - def _any_spend_data(self, vaults): - addr = self.bitcoind.rpc.getnewaddress() - total_spent = sum(v["amount"] for v in vaults) - feerate = 2 - fees = self.compute_spendtx_fees(feerate, len(vaults), 1) - return {addr: total_spent - fees}, feerate - - def unvault_vaults_anyhow(self, vaults, priority=False): - """ - Unvault these vaults with a random Spend transaction for a maximum amount and a - fixed feerate. - """ - destinations, feerate = self._any_spend_data(vaults) - return self.unvault_vaults(vaults, destinations, feerate, priority) - - def broadcast_unvaults_anyhow(self, vaults, priority=False): - """ - Broadcast the Unvault transactions for these vaults with a random Spend - transaction for a maximum amount and a fixed feerate. - """ - destinations, feerate = self._any_spend_data(vaults) - return self.broadcast_unvaults(vaults, destinations, feerate, priority) - - def spend_vaults_anyhow(self, vaults): - """Spend these vaults to a random address for a maximum amount for a fixed feerate""" - destinations, feerate = self._any_spend_data(vaults) - return self.spend_vaults(vaults, destinations, feerate) - - def spend_vaults_anyhow_unconfirmed(self, vaults, priority=False): - """ - Spend these vaults to a random address for a maximum amount for a fixed feerate, - not confirming the Spend transaction. - """ - destinations, feerate = self._any_spend_data(vaults) - return self.spend_vaults_unconfirmed(vaults, destinations, feerate, priority) - - def compute_spendtx_fees( - self, spendtx_feerate, n_vaults_spent, n_destinations, with_change=False - ): - """Get the fees necessary to include in a Spend transaction. - This assumes the destinations to be P2WPKH - """ - n_stk = len(self.stks()) - n_man = len(self.mans()) - - # witscript PUSH, keys , Unvault Script overhead, signatures - spend_witness_vb = ( - 1 + (n_man + n_stk * 2) * 34 + 15 + (n_man + n_stk) * 73 - ) // 4 - # Overhead, P2WPKH, P2WSH, inputs, witnesses - spend_witstrip_vb = ( - 11 - + 31 * n_destinations - + 43 * (1 + (1 if with_change else 0)) - + (32 + 4 + 4 + 1) * n_vaults_spent - ) - spendtx_vbytes = spend_witstrip_vb + spend_witness_vb * n_vaults_spent - - # witscript PUSH, keys , Deposit Script overhead, signatures - unvault_witness_vb = (1 + n_stk * (34 + 73) + 3) // 4 - # Overhead, P2WSH * 2, inputs + witness - unvaulttxs_vbytes = ( - 11 + 43 * 2 + (32 + 4 + 4 + 1) + unvault_witness_vb - ) * n_vaults_spent - - return ( - spendtx_vbytes * spendtx_feerate # Spend fees - + 2 * 32 * spendtx_vbytes # Spend CPFP - + unvaulttxs_vbytes * 24 # Unvault fees (6sat/WU feerate) - + 30_000 * n_vaults_spent # Unvault CPFP - ) - - def cancel_vault(self, vault): - deposit = f"{vault['txid']}:{vault['vout']}" - - for w in self.participants(): - wait_for( - lambda: len( - w.rpc.listvaults( - ["unvaulting", "unvaulted", "spending"], [deposit] - )["vaults"] - ) - == 1 - ) - - self.stk(0).rpc.revault(deposit) - self.bitcoind.generate_block(1, wait_for_mempool=1) - for w in self.participants(): - wait_for( - lambda: len(w.rpc.listvaults(["canceled"], [deposit])["vaults"]) == 1 - ) - - def stop_wallets(self): - jobs = [self.executor.submit(w.stop) for w in self.participants()] - for j in jobs: - j.result(TIMEOUT) - - def start_wallets(self): - jobs = [self.executor.submit(w.start) for w in self.participants()] - for j in jobs: - j.result(TIMEOUT) - - def cleanup(self): - for n in self.daemons: - n.cleanup() - if self.bitcoind_proxy is not None: - self.bitcoind_proxy.stop() From a067daf16b99a8bdaad1034f2e539c60108a1b4a Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 17:59:00 +0200 Subject: [PATCH 02/19] tests: use a ranged descriptor --- tests/fixtures.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/fixtures.py b/tests/fixtures.py index 74e61942..6a19b44b 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -117,7 +117,7 @@ def minisafed(bitcoind, directory): os.makedirs(datadir, exist_ok=True) bitcoind_cookie = os.path.join(bitcoind.bitcoin_dir, "regtest", ".cookie") - main_desc = "wsh(or_d(pk(02869ef67283b4bc9af9d8366efb31f718018bfd5970a69b3d16f22f51228f73dc),and_v(v:pkh(03bb4dc7ed08cc633893f457553ad941ff82195342467d350dbb63773dd17f113b),older(157680))))" + main_desc = "wsh(or_d(pk(tpubD9vQiBdDxYzU1V5D5UUmMTXF9FZC13PuQDs4aiv6rF7UCKQFvtVKZguYakX12C2bt8736ksioxu9Y9Nmp18gj4jDeNJEEqrBPEZXAxe5YcQ/*),and_v(v:pkh(tpubD9vQiBdDxYzU4cVFtApWj4devZrvcfWaPXX1zHdDc7GPfUsDKqGnbhraccfm7BAXgRgUbVQUV2v2o4NitjGEk7hpbuP85kvBrD4ahFDtNBJ/*),older(157680))))" minisafed = Minisafed( datadir, From 6ff43fcd9543acb79d6f4c343e102f5029f5480f Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:24:32 +0200 Subject: [PATCH 03/19] daemon: log the name of the current thread, too. --- src/bin/daemon.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/bin/daemon.rs b/src/bin/daemon.rs index ea467ea7..fe654182 100644 --- a/src/bin/daemon.rs +++ b/src/bin/daemon.rs @@ -2,7 +2,7 @@ use std::{ env, io::{self, Write}, path::PathBuf, - process, time, + process, thread, time, }; use minisafe::{config::Config, DaemonHandle}; @@ -25,7 +25,7 @@ fn setup_logger(log_level: log::LevelFilter) -> Result<(), fern::InitError> { let dispatcher = fern::Dispatch::new() .format(|out, message, record| { out.finish(format_args!( - "[{}][{}][{}] {}", + "[{}][{}][{}][thread {}] {}", time::SystemTime::now() .duration_since(time::UNIX_EPOCH) .unwrap_or_else(|e| { @@ -35,6 +35,7 @@ fn setup_logger(log_level: log::LevelFilter) -> Result<(), fern::InitError> { .as_secs(), record.target(), record.level(), + thread::current().name().unwrap_or("unnamed"), message )) }) From 6b2e90181484d5dd0ed4ee23b78ab705d98b5c36 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:25:14 +0200 Subject: [PATCH 04/19] bit poller: fix a off logged percentage --- src/bitcoin/poller/looper.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/bitcoin/poller/looper.rs b/src/bitcoin/poller/looper.rs index 7df006fa..0a212932 100644 --- a/src/bitcoin/poller/looper.rs +++ b/src/bitcoin/poller/looper.rs @@ -63,7 +63,7 @@ pub fn looper( let sync_progress = bit.sync_progress(); log::info!( "Block chain synchronization progress: {:.2}%", - sync_progress + sync_progress * 100.0 ); synced = sync_progress == 1.0; if !synced { From 86c1d32662e7fda9887da5f799e42cce73228448 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:25:28 +0200 Subject: [PATCH 05/19] bitcoin: name the Bitcoin poller thread --- src/bitcoin/poller/mod.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/bitcoin/poller/mod.rs b/src/bitcoin/poller/mod.rs index fa42471e..7ee7b5f0 100644 --- a/src/bitcoin/poller/mod.rs +++ b/src/bitcoin/poller/mod.rs @@ -23,10 +23,13 @@ impl Poller { poll_interval: time::Duration, ) -> Poller { let shutdown = sync::Arc::from(atomic::AtomicBool::from(false)); - let handle = thread::spawn({ - let shutdown = shutdown.clone(); - move || looper(bit, db, shutdown, poll_interval) - }); + let handle = thread::Builder::new() + .name("Bitcoin poller".to_string()) + .spawn({ + let shutdown = shutdown.clone(); + move || looper(bit, db, shutdown, poll_interval) + }) + .expect("Must not fail"); Poller { shutdown, handle } } From dd1b353a36feb04b1af5553e9c71763957996af0 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:26:05 +0200 Subject: [PATCH 06/19] commands: derive serde::Serialize for results --- src/commands/mod.rs | 18 +++++++++++++----- src/lib.rs | 6 +++--- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 8069e7bb..767e19ee 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -9,6 +9,7 @@ use miniscript::{ descriptor::{self, DescriptorTrait}, TranslatePk2, }; +use serde::{Deserialize, Serialize}; impl DaemonControl { /// Get information about the current state of the daemon @@ -26,29 +27,31 @@ impl DaemonControl { /// Get a new deposit address. This will always generate a new deposit address, regardless of /// whether it was actually used. - pub fn get_new_address(&self) -> bitcoin::Address { + pub fn get_new_address(&self) -> GetAddressResult { let mut db_conn = self.db.connection(); let index = db_conn.derivation_index(); // TODO: handle should we wrap around instead of failing? db_conn.update_derivation_index(index.increment().expect("TODO: handle wraparound")); - self.config + let address = self + .config .main_descriptor // TODO: have a descriptor newtype along with a derived descriptor one. .derive(index.into()) .translate_pk2(|xpk| xpk.derive_public_key(&self.secp)) .expect("All pubkeys were derived, no wildcard.") .address(self.config.bitcoind_config.network) - .expect("It's a wsh() descriptor") + .expect("It's a wsh() descriptor"); + GetAddressResult { address } } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize)] pub struct GetInfoDescriptors { pub main: descriptor::Descriptor, } /// Information about the daemon -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize)] pub struct GetInfoResult { pub version: String, pub network: bitcoin::Network, @@ -56,3 +59,8 @@ pub struct GetInfoResult { pub sync: f64, pub descriptors: GetInfoDescriptors, } + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GetAddressResult { + pub address: bitcoin::Address, +} diff --git a/src/lib.rs b/src/lib.rs index 8910971b..9207173a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -483,8 +483,8 @@ mod tests { // TODO: avoid scope creep. We should move the bitcoind-specific checks to the // bitcoind module, test the startup with a mocked bitcoind interface, and not test // commands here but in the commands module. - let addr = handle.control.get_new_address(); - let addr2 = handle.control.get_new_address(); + let addr = handle.control.get_new_address().address; + let addr2 = handle.control.get_new_address().address; assert_eq!( addr, bitcoin::Address::from_str( @@ -511,7 +511,7 @@ mod tests { let daemon_thread = thread::spawn(move || { let handle = DaemonHandle::start(config).unwrap(); // TODO: avoid scope creep. See above comment. - assert_ne!(handle.control.get_new_address(), addr); + assert_ne!(handle.control.get_new_address().address, addr); handle.shutdown(); }); complete_sanity_check(&server); From 0d55d6c45599c2a4996f10c5066041369c06d9b1 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:43:19 +0200 Subject: [PATCH 07/19] jsonrpc: a simple JSONRPC2 server --- src/bin/cli.rs | 10 +- src/bin/daemon.rs | 4 +- src/config.rs | 9 + src/jsonrpc/api.rs | 17 ++ src/jsonrpc/mod.rs | 161 ++++++++++++++++++ src/jsonrpc/server.rs | 380 ++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 44 ++++- 7 files changed, 615 insertions(+), 10 deletions(-) create mode 100644 src/jsonrpc/api.rs create mode 100644 src/jsonrpc/mod.rs create mode 100644 src/jsonrpc/server.rs diff --git a/src/bin/cli.rs b/src/bin/cli.rs index c80b154e..870bc271 100644 --- a/src/bin/cli.rs +++ b/src/bin/cli.rs @@ -14,7 +14,7 @@ use std::os::unix::net::UnixStream; // Exits with error fn show_usage() { eprintln!("Usage:"); - eprintln!(" revault-cli [--conf conf_path] [--raw] [ ...]"); + eprintln!(" minisafe-cli [--conf conf_path] [--raw] [ ...]"); process::exit(1); } @@ -73,7 +73,7 @@ fn rpc_request(method: String, params: Vec) -> Json { object.insert("jsonrpc".to_string(), Json::String("2.0".to_string())); object.insert( "id".to_string(), - Json::String(format!("revault-cli-{}", process::id())), + Json::String(format!("minisafe-cli-{}", process::id())), ); object.insert("method".to_string(), method); object.insert("params".to_string(), params); @@ -94,7 +94,7 @@ fn socket_file(conf_file: Option) -> PathBuf { [ data_dir, config.bitcoind_config.network.to_string().as_str(), - "revaultd_rpc", + "minisafed_rpc", ] .iter() .collect() @@ -127,7 +127,7 @@ fn main() { process::exit(1); }); socket - .write_all(request.to_string().as_bytes()) + .write_all(&[request.to_string().as_bytes(), b"\n"].concat()) .unwrap_or_else(|e| { eprintln!("Writing to {:?}: '{}'", &socket_file, e); process::exit(1); @@ -160,7 +160,7 @@ fn main() { println!("{:#}", serde_json::json!({ "error": e })); } else { log::warn!( - "revaultd response doesn't contain result or error: '{}'", + "minisafed response doesn't contain result or error: '{}'", response ); println!("{:#}", response); diff --git a/src/bin/daemon.rs b/src/bin/daemon.rs index fe654182..734cf40d 100644 --- a/src/bin/daemon.rs +++ b/src/bin/daemon.rs @@ -63,7 +63,9 @@ fn main() { // The panic hook will log::error panic!("Starting Minisafe daemon: {}", e); }); - daemon.shutdown(); + daemon + .rpc_server() + .expect("JSONRPC server must terminate cleanly"); // We are always logging to stdout, should it be then piped to the log file (if self) or // not. So just make sure that all messages were actually written. diff --git a/src/config.rs b/src/config.rs index 425d559f..ec41369c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -94,6 +94,15 @@ pub struct Config { pub bitcoind_config: BitcoindConfig, } +impl Config { + pub fn data_dir(&self) -> Option { + self.data_dir + .as_ref() + .map(Clone::clone) + .or_else(config_folder_path) + } +} + #[derive(PartialEq, Eq, Debug)] pub enum ConfigError { DatadirNotFound, diff --git a/src/jsonrpc/api.rs b/src/jsonrpc/api.rs new file mode 100644 index 00000000..f3a7f9a1 --- /dev/null +++ b/src/jsonrpc/api.rs @@ -0,0 +1,17 @@ +use crate::{ + jsonrpc::{Error, Request, Response}, + DaemonControl, +}; + +/// Handle an incoming JSONRPC2 request. +pub fn handle_request(control: &DaemonControl, req: Request) -> Result { + let result = match req.method.as_str() { + "getinfo" => serde_json::json!(&control.get_info()), + "getnewaddress" => serde_json::json!(&control.get_new_address()), + _ => { + return Err(Error::method_not_found()); + } + }; + + Ok(Response::success(req.id, result)) +} diff --git a/src/jsonrpc/mod.rs b/src/jsonrpc/mod.rs new file mode 100644 index 00000000..a771e666 --- /dev/null +++ b/src/jsonrpc/mod.rs @@ -0,0 +1,161 @@ +mod api; +pub mod server; + +use std::{error, fmt}; + +use serde::{self, Deserialize, Deserializer, Serialize, Serializer}; + +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(untagged)] +pub enum Params { + Array(Vec), + Map(serde_json::Map), +} + +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(untagged)] +pub enum ReqId { + Num(u64), + Str(String), +} + +/// A JSONRPC2 request. See https://www.jsonrpc.org/specification#request_object. +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Request { + /// Version. Must be "2.0". + pub jsonrpc: String, + /// Command name. + pub method: String, + /// Command parameters. + pub params: Option, + /// Request identifier. + pub id: ReqId, +} + +/// JSONRPC2 error codes. See https://www.jsonrpc.org/specification#error_object. +#[derive(Debug, PartialEq, Clone)] +pub enum ErrorCode { + /// The method does not exist / is not available. + MethodNotFound, + /// Invalid method parameter(s). + InvalidParams, + /// Reserved for implementation-defined server-errors. + ServerError(i64), +} + +impl Into for &ErrorCode { + fn into(self) -> i64 { + match self { + ErrorCode::MethodNotFound => -32601, + ErrorCode::InvalidParams => -32602, + ErrorCode::ServerError(code) => *code, + } + } +} + +impl From for ErrorCode { + fn from(code: i64) -> ErrorCode { + match code { + -32601 => ErrorCode::MethodNotFound, + -32602 => ErrorCode::InvalidParams, + code => ErrorCode::ServerError(code), + } + } +} + +impl<'a> Deserialize<'a> for ErrorCode { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'a>, + { + let code: i64 = Deserialize::deserialize(deserializer)?; + Ok(code.into()) + } +} + +impl Serialize for ErrorCode { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_i64(self.into()) + } +} + +/// JSONRPC2 error response. See https://www.jsonrpc.org/specification#error_object. +#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Error { + pub code: ErrorCode, + pub message: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, +} + +impl Error { + pub fn new(code: ErrorCode, message: impl Into) -> Error { + Error { + message: message.into(), + code, + data: None, + } + } + + pub fn method_not_found() -> Error { + Error::new(ErrorCode::MethodNotFound, "Method not found") + } + + pub fn invalid_params(message: impl Into) -> Error { + Error::new( + ErrorCode::InvalidParams, + format!("Invalid params: {}", message.into()), + ) + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + let code: i64 = (&self.code).into(); + write!(f, "{}: {}", code, self.message) + } +} + +impl error::Error for Error {} + +/// JSONRPC2 response. See https://www.jsonrpc.org/specification#response_object. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Response { + /// Version. Must be "2.0". + jsonrpc: String, + /// Required on success. Must not exist on error. + #[serde(skip_serializing_if = "Option::is_none")] + result: Option, + /// Required on error. Must not exist on success. + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + /// Request identifier. + id: ReqId, +} + +impl Response { + fn new(id: ReqId, result: Option, error: Option) -> Response { + Response { + jsonrpc: "2.0".to_string(), + result, + error, + id, + } + } + + pub fn success(id: ReqId, result: serde_json::Value) -> Response { + Response::new(id, Some(result), None) + } + + pub fn error(id: ReqId, error: Error) -> Response { + Response::new(id, None, Some(error)) + } +} diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs new file mode 100644 index 00000000..a0dad02d --- /dev/null +++ b/src/jsonrpc/server.rs @@ -0,0 +1,380 @@ +//! JSONRPC2 server +//! +//! This module implements the connections and streams handling logic for receiving +//! JSONRPC2 requests on a Unix Domain Socket. + +use crate::{ + jsonrpc::{api, Request, Response}, + DaemonControl, +}; + +use std::{ + io, + os::unix::net, + path, + sync::{self, atomic}, + thread, time, +}; + +// Maximum number of concurrent RPC connections we may accept. +const MAX_CONNECTIONS: u32 = 16; + +// Read a command from the stream. +// +// In order to both treat commands separately (respond as soon as we read one), and support +// multiple commands in a single read or in multiple parts, we are given the context as writable +// arguments: +// - `buf` is the buffer used to read from the socket. It will be extended as needed. It must be +// initialized. +// - `end`: The index of the end of the data read from the stream. Since `buf` needs to be +// initialized with dummy values, it can be very different from `buf.len()`. Used to not check +// for the separator character in the parts of the buffer with dummy values. +// - `cursor`: The index at which we checked for the separator character (`\n`). Used to not +// check twice for it on the same buffer chunk. +fn read_command( + stream: &mut dyn io::Read, + buf: &mut Vec, + end: &mut usize, + cursor: &mut usize, +) -> Result, io::Error> { + assert!(!buf.is_empty()); + + loop { + // First off, check if there are no existing commands in the buffer. + let pos = buf[*cursor..*end].iter().position(|byt| byt == &b'\n'); + log::trace!( + "pos: {:?}, buf[cur..end]: {:?}", + pos, + String::from_utf8_lossy(&buf[*cursor..*end]) + ); + if let Some(pos) = pos { + log::trace!( + "Parsing Request from: {:?}", + String::from_utf8_lossy(&buf[..*cursor + pos]) + ); + // TODO: don't return an io::Error here, instead try to parse a Request. Failing that, + // try to parse a serde_json::Value. Then return accordingly a JSONRPC "malformed + // request" or "invalid JSON" error. + let req: Request = serde_json::from_slice(&buf[..*cursor + pos])?; + *buf = buf[pos + 1..].to_vec(); // FIXME: can we avoid reallocating here? + *cursor = 0; + *end -= pos + 1; + + return Ok(Some(req)); + } + + // If nothing can be gathered from the buffer, continue reading. + let new_read = stream.read(&mut buf[*end..])?; + if new_read == 0 { + return Ok(None); + } + + // If we filled the buffer, increase its size and try again. + *end += new_read; + let buffer_filled = *end == buf.len(); + if buffer_filled { + buf.resize(buf.len() * 2, 0); + continue; + } + } +} + +// Handle all messages from this connection. +fn connection_handler( + control: sync::Arc>, + mut stream: net::UnixStream, +) -> Result<(), io::Error> { + let mut buf = vec![0; 2048]; + let mut end = 0; + let mut cursor = 0; + + loop { + let req = match read_command(&mut stream, &mut buf, &mut end, &mut cursor)? { + Some(req) => req, + None => { + // Connection closed. + return Ok(()); + } + }; + + // TODO: respond in case of invalid JSON or invalid JSONRPC request. + let req_id = req.id.clone(); + + log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req)); + let response = api::handle_request(&control.lock().unwrap(), req) + .unwrap_or_else(|e| Response::error(req_id, e)); + log::trace!("JSONRPC response: {:?}", serde_json::to_string(&response)); + if let Err(e) = serde_json::to_writer(&stream, &response) { + log::error!("Error writing response: '{}'", e); + return Ok(()); + } + } +} + +// FIXME: have a decent way to share the DaemonControl between connections. Maybe make it Clone? +/// The main event loop. Wait for connections, and treat requests sent through them. +pub fn rpcserver_loop( + listener: net::UnixListener, + daemon_control: sync::Arc>, +) -> Result<(), io::Error> { + // Keep it simple. We don't need great performances so just treat each connection in + // its thread, with a given maximum number of connections. + let connections_counter = sync::Arc::from(atomic::AtomicU32::new(0)); + + loop { + let (connection, _) = match listener.accept() { + Ok(c) => c, + Err(e) => { + log::error!("Accepting new connection: '{}'", e); + continue; + } + }; + log::trace!("New JSONRPC connection"); + + while connections_counter.load(atomic::Ordering::Relaxed) >= MAX_CONNECTIONS { + thread::sleep(time::Duration::from_millis(50)); + } + connections_counter.fetch_add(1, atomic::Ordering::Relaxed); + + let handler_id = connections_counter.load(atomic::Ordering::Relaxed); + thread::Builder::new() + .name(format!("minisafe-jsonrpc-{}", handler_id)) + .spawn({ + let control = daemon_control.clone(); + let counter = connections_counter.clone(); + + move || { + if let Err(e) = connection_handler(control, connection) { + log::error!("Error while handling connection {}: '{}'", handler_id, e); + } else { + log::trace!("Connection {} terminated without error.", handler_id); + } + counter.fetch_sub(1, atomic::Ordering::Relaxed); + } + })?; + } +} + +// Tries to bind to the socket, if we are told it's already in use try to connect +// to check there is actually someone listening and it's not a leftover from a +// crash. +fn bind(socket_path: &path::Path) -> Result { + match net::UnixListener::bind(socket_path) { + Ok(l) => Ok(l), + Err(e) => { + if e.kind() == io::ErrorKind::AddrInUse { + return match net::UnixStream::connect(socket_path) { + Ok(_) => Err(e), + Err(_) => { + // Ok, no one's here. Just delete the socket and bind. + log::debug!("Removing leftover rpc socket."); + std::fs::remove_file(socket_path)?; + net::UnixListener::bind(socket_path) + } + }; + } + + Err(e) + } + } +} + +/// Bind to the UDS at `socket_path` +pub fn rpcserver_setup(socket_path: &path::Path) -> Result { + log::debug!("Binding socket at {}", socket_path.display()); + // Create the socket with RW permissions only for the user + let old_umask = unsafe { libc::umask(0o177) }; + let listener = bind(&socket_path); + unsafe { + libc::umask(old_umask); + } + + listener +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::jsonrpc::{Params, ReqId}; + + use std::{env, fs, io::Write, process}; + + fn read_one_command(socket_path: &path::Path) -> thread::JoinHandle> { + let listener = rpcserver_setup(socket_path).unwrap(); + thread::spawn(move || { + let (mut conn, _) = listener.accept().unwrap(); + let mut buf = vec![0; 32]; + let mut end = 0; + let mut cursor = 0; + read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap() + }) + } + + fn read_all_commands(socket_path: &path::Path) -> thread::JoinHandle> { + let listener = rpcserver_setup(socket_path).unwrap(); + thread::spawn(move || { + let (mut conn, _) = listener.accept().unwrap(); + let mut buf = vec![0; 32]; + let mut end = 0; + let mut cursor = 0; + let mut reqs = Vec::new(); + + loop { + match read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap() { + Some(req) => { + reqs.push(req); + } + None => return reqs, + } + } + }) + } + + fn write_messages(socket_path: &path::Path, messages: &[&[u8]]) { + let mut client = net::UnixStream::connect(&socket_path).unwrap(); + for mess in messages { + client.write_all(&mess).unwrap(); + // Simulate throttling, this mimics real conditions and actually triggered a crash. + thread::sleep(time::Duration::from_millis(50)); + } + } + + #[test] + fn command_read_single() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // A simple dummy request + let t = read_all_commands(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": {"a": "b"}}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages(&socket_path, &[req, b"\n"]); + let read_req = t.join().unwrap(); + assert_eq!(parsed_req, read_req[0]); + + // Same, but with params as a list and a string id + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": "987-abc", "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages(&socket_path, &[req, b"\n"]); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_parts() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // A single request written in two parts + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages( + &socket_path, + &[&req[..req.len() / 2], &req[req.len() / 2..], b"\n"], + ); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + // A single request written in many parts + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + let tmp: Vec> = req.into_iter().map(|c| vec![*c]).collect(); + let mut to_send: Vec<&[u8]> = tmp.iter().map(|v| v.as_slice()).collect(); + to_send.push(b"\n"); + write_messages(&socket_path, &to_send); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_multiple() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // Multiple requests, in parts + let t = read_all_commands(&socket_path); + let reqs = [ + &br#"{"jsonrpc": "2.0", "id": 20478, "me"#[..], + br#"thod": "test", "params": ["a", 10]}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20499, "method": "t"#, + br#"e_edzA", "params": {"ttt": 980}}"#, + b"\n", + ]; + let parsed_reqs: Vec = vec![ + serde_json::from_slice(&[reqs[0], reqs[1]].concat()).unwrap(), + serde_json::from_slice(reqs[3]).unwrap(), + serde_json::from_slice(&[reqs[5], reqs[6]].concat()).unwrap(), + ]; + write_messages(&socket_path, &reqs); + let read_reqs = t.join().unwrap(); + assert_eq!(parsed_reqs, read_reqs); + + // The same requests, sent at once. + let t = read_all_commands(&socket_path); + let req_parts = [ + &br#"{"jsonrpc": "2.0", "id": 20478, "method": "test", "params": ["a", 10]}"#[..], + b"\n", + br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20499, "method": "te_edzA", "params": {"ttt": 980}}"#, + b"\n", + ] + .concat(); + write_messages(&socket_path, &[req_parts.as_slice()]); + let read_reqs = t.join().unwrap(); + assert_eq!(parsed_reqs, read_reqs); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_linebreak() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // Multiple requests, in parts + let t = read_one_command(&socket_path); + let mut params = serde_json::map::Map::new(); + params.insert( + "dummy param".to_string(), + "dummy value + with line + breaks" + .to_string() + .into(), + ); + let req = Request { + jsonrpc: "2.0".to_string(), + method: "dummy".to_string(), + params: Some(Params::Map(params)), + id: ReqId::Num(0), + }; + write_messages(&socket_path, &[&serde_json::to_vec(&req).unwrap(), b"\n"]); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 9207173a..526ffeb2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,15 +5,19 @@ pub mod config; mod daemonize; mod database; pub mod descriptors; +#[cfg(feature = "jsonrpc_server")] +mod jsonrpc; pub use miniscript; +#[cfg(feature = "jsonrpc_server")] +use crate::jsonrpc::server::{rpcserver_loop, rpcserver_setup}; use crate::{ bitcoin::{ d::{BitcoinD, BitcoindError}, poller, BitcoinInterface, }, - config::{config_folder_path, Config}, + config::Config, database::{ sqlite::{FreshDbOptions, SqliteDb, SqliteDbError}, DatabaseInterface, @@ -183,9 +187,8 @@ impl DaemonHandle { // First, check the data directory let mut data_dir = config - .data_dir - .clone() - .unwrap_or(config_folder_path().ok_or(StartupError::DefaultDataDirNotFound)?); + .data_dir() + .ok_or(StartupError::DefaultDataDirNotFound)?; data_dir.push(config.bitcoind_config.network.to_string()); let fresh_data_dir = !data_dir.as_path().exists(); if fresh_data_dir { @@ -259,6 +262,39 @@ impl DaemonHandle { }) } + /// Start the JSONRPC server and listen for incoming commands until we die. + /// Like DaemonHandle::shutdown(), this stops the Bitcoin poller at teardown. + #[cfg(feature = "jsonrpc_server")] + pub fn rpc_server(self) -> Result<(), io::Error> { + let DaemonHandle { + control, + bitcoin_poller: poller, + } = self; + + let rpc_socket: path::PathBuf = [ + control + .config + .data_dir() + .expect("Didn't fail at startup, must not now") + .as_path(), + path::Path::new(&control.config.bitcoind_config.network.to_string()), + path::Path::new("minisafed_rpc"), + ] + .iter() + .collect(); + let listener = rpcserver_setup(&rpc_socket)?; + log::info!("JSONRPC server started."); + + // FIXME: don't use a Mutex... + let control = sync::Arc::from(sync::Mutex::from(control)); + rpcserver_loop(listener, control)?; + log::info!("JSONRPC server stopped."); + + poller.stop(); + + Ok(()) + } + // NOTE: this moves out the data as it should not be reused after shutdown /// Shut down the Minisafe daemon. pub fn shutdown(self) { From 93098be7dcb82808f6644eee97005c6607eb0cda Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:45:09 +0200 Subject: [PATCH 08/19] tests: implement the connection to the daemon's RPC server And add some basic sanity check of the existing commands. --- tests/test_framework/minisafed.py | 4 + tests/test_framework/utils.py | 152 +++++++++++++++++++++++++++++- tests/test_rpc.py | 17 ++++ 3 files changed, 169 insertions(+), 4 deletions(-) create mode 100644 tests/test_rpc.py diff --git a/tests/test_framework/minisafed.py b/tests/test_framework/minisafed.py index ee92f77b..0b9135d0 100644 --- a/tests/test_framework/minisafed.py +++ b/tests/test_framework/minisafed.py @@ -1,6 +1,7 @@ import os from test_framework.utils import ( + UnixDomainSocketRpc, TailableProc, VERBOSE, LOG_LEVEL, @@ -22,6 +23,8 @@ class Minisafed(TailableProc): self.conf_file = os.path.join(datadir, "config.toml") self.cmd_line = [MINISAFED_PATH, "--conf", f"{self.conf_file}"] + socket_path = os.path.join(os.path.join(datadir, "regtest"), "minisafed_rpc") + self.rpc = UnixDomainSocketRpc(socket_path) with open(self.conf_file, "w") as f: f.write(f"data_dir = '{datadir}'\n") @@ -42,6 +45,7 @@ class Minisafed(TailableProc): [ "Database initialized and checked", "Connection to bitcoind established and checked.", + "JSONRPC server started.", ] ) diff --git a/tests/test_framework/utils.py b/tests/test_framework/utils.py index 6409a5e3..863b0be1 100644 --- a/tests/test_framework/utils.py +++ b/tests/test_framework/utils.py @@ -1,7 +1,9 @@ import itertools +import json import logging import os import re +import socket import subprocess import threading import time @@ -42,18 +44,160 @@ def wait_for(success, timeout=TIMEOUT, debug_fn=None): class RpcError(ValueError): - def __init__(self, method: str, payload: dict, error: str): + def __init__(self, method: str, params: dict, error: str): super(ValueError, self).__init__( - "RPC call failed: method: {}, payload: {}, error: {}".format( - method, payload, error + "RPC call failed: method: {}, params: {}, error: {}".format( + method, params, error ) ) self.method = method - self.payload = payload + self.params = params self.error = error +class UnixSocket(object): + """A wrapper for socket.socket that is specialized to unix sockets. + + Some OS implementations impose restrictions on the Unix sockets. + + - On linux OSs the socket path must be shorter than the in-kernel buffer + size (somewhere around 100 bytes), thus long paths may end up failing + the `socket.connect` call. + + This is a small wrapper that tries to work around these limitations. + + """ + + def __init__(self, path: str): + self.path = path + self.sock = None + self.connect() + + def connect(self) -> None: + try: + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect(self.path) + self.sock.settimeout(TIMEOUT) + except OSError as e: + self.close() + + if e.args[0] == "AF_UNIX path too long" and os.uname()[0] == "Linux": + # If this is a Linux system we may be able to work around this + # issue by opening our directory and using `/proc/self/fd/` to + # get a short alias for the socket file. + # + # This was heavily inspired by the Open vSwitch code see here: + # https://github.com/openvswitch/ovs/blob/master/python/ovs/socket_util.py + + dirname = os.path.dirname(self.path) + basename = os.path.basename(self.path) + + # Open an fd to our home directory, that we can then find + # through `/proc/self/fd` and access the contents. + dirfd = os.open(dirname, os.O_DIRECTORY | os.O_RDONLY) + short_path = "/proc/self/fd/%d/%s" % (dirfd, basename) + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect(short_path) + else: + # There is no good way to recover from this. + raise + + def close(self) -> None: + if self.sock is not None: + self.sock.close() + self.sock = None + + def sendall(self, b: bytes) -> None: + if self.sock is None: + raise socket.error("not connected") + + self.sock.sendall(b) + + def recv(self, length: int) -> bytes: + if self.sock is None: + raise socket.error("not connected") + + return self.sock.recv(length) + + def __del__(self) -> None: + self.close() + + +class UnixDomainSocketRpc(object): + def __init__(self, socket_path, logger=logging): + self.socket_path = socket_path + self.logger = logger + self.next_id = 0 + + def _readobj(self, sock): + """Read a JSON object""" + buff = b"" + while True: + n_to_read = max(2048, len(buff)) + chunk = sock.recv(n_to_read) + buff += chunk + if len(chunk) != n_to_read: + try: + return json.loads(buff) + except json.JSONDecodeError: + # There is more to read, continue + # FIXME: this is a workaround for large reads taken from revaultd. + # We should use the '\n' marker instead since minisafed uses that. + continue + + def __getattr__(self, name): + """Intercept any call that is not explicitly defined and call @call. + + We might still want to define the actual methods in the subclasses for + documentation purposes. + """ + name = name.replace("_", "-") + + def wrapper(*args, **kwargs): + if len(args) != 0 and len(kwargs) != 0: + raise RpcError( + name, {}, "Cannot mix positional and non-positional arguments" + ) + return self.call(name, params=kwargs) + + return wrapper + + def call(self, method, params={}): + self.logger.debug(f"Calling {method} with params {params}") + + # FIXME: we open a new socket for every readobj call... + sock = UnixSocket(self.socket_path) + msg = json.dumps( + { + "jsonrpc": "2.0", + "id": 0, + "method": method, + "params": params, + } + ) + sock.sendall(msg.encode() + b"\n") + this_id = self.next_id + resp = self._readobj(sock) + + self.logger.debug(f"Received response for {method} call: {resp}") + if "id" in resp and resp["id"] != this_id: + raise ValueError( + "Malformed response, id is not {}: {}.".format(this_id, resp) + ) + sock.close() + + if not isinstance(resp, dict): + raise ValueError( + f"Malformed response, response is not a dictionary: {resp}" + ) + elif "error" in resp: + raise RpcError(method, params, resp["error"]) + elif "result" not in resp: + raise ValueError('Malformed response, "result" missing.') + return resp["result"] + + class TailableProc(object): """A monitorable process that we can start, stop and tail. diff --git a/tests/test_rpc.py b/tests/test_rpc.py new file mode 100644 index 00000000..de67b85f --- /dev/null +++ b/tests/test_rpc.py @@ -0,0 +1,17 @@ +from fixtures import * + + +def test_getinfo(minisafed): + res = minisafed.rpc.getinfo() + assert res["version"] == "0.1" + assert res["network"] == "regtest" + assert res["blockheight"] == 101 + assert res["sync"] == 1.0 + assert "main" in res["descriptors"] + + +def test_getaddress(minisafed): + res = minisafed.rpc.getnewaddress() + assert "address" in res + # We'll get a new one at every call + assert res["address"] != minisafed.rpc.getnewaddress()["address"] From f365effacd12d0d3fce105dbc59e7ccf5f85488c Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Sun, 7 Aug 2022 12:41:45 +0200 Subject: [PATCH 09/19] Accept a custom Bitcoin interface when starting the daemon The Bitcoin interface was thought of as being generic, but a caller couldn't use one different from bitcoind. Make it so they can, and fix our trait and generics implementations. --- src/bin/daemon.rs | 2 +- src/bitcoin/mod.rs | 25 ++++++++--- src/bitcoin/poller/looper.rs | 2 +- src/bitcoin/poller/mod.rs | 2 +- src/commands/mod.rs | 2 +- src/config.rs | 1 + src/lib.rs | 80 ++++++++++++++++++++++++------------ 7 files changed, 78 insertions(+), 36 deletions(-) diff --git a/src/bin/daemon.rs b/src/bin/daemon.rs index 734cf40d..13e7b366 100644 --- a/src/bin/daemon.rs +++ b/src/bin/daemon.rs @@ -59,7 +59,7 @@ fn main() { process::exit(1); }); - let daemon = DaemonHandle::start(config).unwrap_or_else(|e| { + let daemon = DaemonHandle::start_default(config).unwrap_or_else(|e| { // The panic hook will log::error panic!("Starting Minisafe daemon: {}", e); }); diff --git a/src/bitcoin/mod.rs b/src/bitcoin/mod.rs index fee4f427..0bf4170d 100644 --- a/src/bitcoin/mod.rs +++ b/src/bitcoin/mod.rs @@ -28,20 +28,33 @@ pub trait BitcoinInterface: Send { fn is_in_chain(&self, tip: &BlockChainTip) -> bool; } -impl BitcoinInterface for sync::Arc> { +impl BitcoinInterface for d::BitcoinD { fn sync_progress(&self) -> f64 { - self.read().unwrap().sync_progress() + self.sync_progress() } fn chain_tip(&self) -> BlockChainTip { - self.read().unwrap().chain_tip() + self.chain_tip() } fn is_in_chain(&self, tip: &BlockChainTip) -> bool { - self.read() - .unwrap() - .get_block_hash(tip.height) + self.get_block_hash(tip.height) .map(|bh| bh == tip.hash) .unwrap_or(false) } } + +// FIXME: do we need to repeat the entire trait implemenation? Isn't there a nicer way? +impl BitcoinInterface for sync::Arc> { + fn sync_progress(&self) -> f64 { + self.lock().unwrap().sync_progress() + } + + fn chain_tip(&self) -> BlockChainTip { + self.lock().unwrap().chain_tip() + } + + fn is_in_chain(&self, tip: &BlockChainTip) -> bool { + self.lock().unwrap().is_in_chain(tip) + } +} diff --git a/src/bitcoin/poller/looper.rs b/src/bitcoin/poller/looper.rs index 0a212932..4bdb7350 100644 --- a/src/bitcoin/poller/looper.rs +++ b/src/bitcoin/poller/looper.rs @@ -39,7 +39,7 @@ fn update_tip(bit: &impl BitcoinInterface, db_conn: &mut Box>, db: impl DatabaseInterface, shutdown: sync::Arc, poll_interval: time::Duration, diff --git a/src/bitcoin/poller/mod.rs b/src/bitcoin/poller/mod.rs index 7ee7b5f0..23dd0b57 100644 --- a/src/bitcoin/poller/mod.rs +++ b/src/bitcoin/poller/mod.rs @@ -18,7 +18,7 @@ pub struct Poller { impl Poller { pub fn start( - bit: impl BitcoinInterface + 'static, + bit: sync::Arc>, db: impl DatabaseInterface + 'static, poll_interval: time::Duration, ) -> Poller { diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 767e19ee..733d2b56 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -2,7 +2,7 @@ //! //! External interface to the Minisafe daemon. -use crate::{DaemonControl, VERSION}; +use crate::{bitcoin::BitcoinInterface, DaemonControl, VERSION}; use miniscript::{ bitcoin, diff --git a/src/config.rs b/src/config.rs index ec41369c..42085b74 100644 --- a/src/config.rs +++ b/src/config.rs @@ -50,6 +50,7 @@ fn default_daemon() -> bool { false } +// TODO: separate Bitcoin config and bitcoind-specific config. /// Everything we need to know for talking to bitcoind serenely #[derive(Debug, Clone, Deserialize, Serialize)] pub struct BitcoindConfig { diff --git a/src/lib.rs b/src/lib.rs index 526ffeb2..a50dcdb9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -148,9 +148,35 @@ fn create_datadir(datadir_path: &path::Path) -> Result<(), StartupError> { }; } +// Connect to bitcoind. Setup the watchonly wallet, and do some sanity checks. +// If all went well, returns the interface to bitcoind. +fn setup_bitcoind( + config: &Config, + data_dir: &path::Path, + fresh_data_dir: bool, +) -> Result { + // Now set up the bitcoind interface + let wo_path: path::PathBuf = [data_dir, path::Path::new("minisafed_watchonly_wallet")] + .iter() + .collect(); + let bitcoind = BitcoinD::new( + &config.bitcoind_config, + wo_path.to_str().expect("Must be valid unicode").to_string(), + )?; + if fresh_data_dir { + bitcoind.create_watchonly_wallet(&config.main_descriptor)?; + log::info!("Created a new watchonly wallet on bitcoind."); + } + bitcoind.try_load_watchonly_wallet(); + bitcoind.sanity_check(&config.main_descriptor, config.bitcoind_config.network)?; + log::info!("Connection to bitcoind established and checked."); + + Ok(bitcoind.with_retry_limit(None)) +} + pub struct DaemonControl { config: Config, - bitcoin: Box, + bitcoin: sync::Arc>, db: Box, secp: secp256k1::Secp256k1, } @@ -158,7 +184,7 @@ pub struct DaemonControl { impl DaemonControl { pub fn new( config: Config, - bitcoin: Box, + bitcoin: sync::Arc>, db: Box, ) -> DaemonControl { let secp = secp256k1::Secp256k1::verification_only(); @@ -179,9 +205,15 @@ pub struct DaemonHandle { impl DaemonHandle { /// This starts the Minisafe daemon. Call `shutdown` to shut it down. /// + /// You may specify a custom Bitcoin interface through the `bitcoin` parameter. If `None`, the + /// default Bitcoin interface (`bitcoind` JSONRPC) will be used. + /// /// **Note**: we internally use threads, and set a panic hook. A downstream application must /// not overwrite this panic hook. - pub fn start(config: Config) -> Result { + pub fn start( + config: Config, + bitcoin: Option, + ) -> Result { #[cfg(not(test))] setup_panic_hook(); @@ -212,24 +244,15 @@ impl DaemonHandle { sqlite.sanity_check(config.bitcoind_config.network, &config.main_descriptor)?; log::info!("Database initialized and checked."); - // Now set up the bitcoind interface - let wo_path: path::PathBuf = [ - data_dir.as_path(), - path::Path::new("minisafed_watchonly_wallet"), - ] - .iter() - .collect(); - let bitcoind = BitcoinD::new( - &config.bitcoind_config, - wo_path.to_str().expect("Must be valid unicode").to_string(), - )?; - if fresh_data_dir { - bitcoind.create_watchonly_wallet(&config.main_descriptor)?; - log::info!("Created a new watchonly wallet on bitcoind."); - } - bitcoind.try_load_watchonly_wallet(); - bitcoind.sanity_check(&config.main_descriptor, config.bitcoind_config.network)?; - log::info!("Connection to bitcoind established and checked."); + // Now, set up the Bitcoin interface. + let bit = match bitcoin { + Some(bit) => sync::Arc::from(sync::Mutex::from(bit)), + None => sync::Arc::from(sync::Mutex::from(setup_bitcoind( + &config, + &data_dir, + fresh_data_dir, + )?)) as sync::Arc>, + }; // If we are on a UNIX system and they told us to daemonize, do it now. // NOTE: it's safe to daemonize now, as we don't carry any open DB connection @@ -246,15 +269,14 @@ impl DaemonHandle { } // Spawn the bitcoind poller with a retry limit high enough that we'd fail after that. - let bitcoind = sync::Arc::from(sync::RwLock::from(bitcoind.with_retry_limit(None))); let bitcoin_poller = poller::Poller::start( - bitcoind.clone(), + bit.clone(), sqlite.clone(), config.bitcoind_config.poll_interval_secs, ); // Finally, set up the API. - let control = DaemonControl::new(config, Box::from(bitcoind), Box::from(sqlite)); + let control = DaemonControl::new(config, bit, Box::from(sqlite)); Ok(Self { control, @@ -262,6 +284,12 @@ impl DaemonHandle { }) } + /// Start the Minisafe daemon with the default Bitcoin and database interfaces (`bitcoind` RPC + /// and SQLite). + pub fn start_default(config: Config) -> Result { + DaemonHandle::start(config, Option::::None) + } + /// Start the JSONRPC server and listen for incoming commands until we die. /// Like DaemonHandle::shutdown(), this stops the Bitcoin poller at teardown. #[cfg(feature = "jsonrpc_server")] @@ -515,7 +543,7 @@ mod tests { let daemon_thread = thread::spawn({ let config = config.clone(); move || { - let handle = DaemonHandle::start(config).unwrap(); + let handle = DaemonHandle::start_default(config).unwrap(); // TODO: avoid scope creep. We should move the bitcoind-specific checks to the // bitcoind module, test the startup with a mocked bitcoind interface, and not test // commands here but in the commands module. @@ -545,7 +573,7 @@ mod tests { // The datadir is created now, so if we restart it it won't create the wo wallet. let daemon_thread = thread::spawn(move || { - let handle = DaemonHandle::start(config).unwrap(); + let handle = DaemonHandle::start_default(config).unwrap(); // TODO: avoid scope creep. See above comment. assert_ne!(handle.control.get_new_address().address, addr); handle.shutdown(); From ea3595349d0051912ce187ca27de7a6df9b6e7e6 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Sun, 7 Aug 2022 13:01:46 +0200 Subject: [PATCH 10/19] Accept a custom database interface when starting the daemon --- src/bitcoin/poller/looper.rs | 2 +- src/bitcoin/poller/mod.rs | 2 +- src/commands/mod.rs | 2 +- src/database/mod.rs | 9 ++++++ src/lib.rs | 59 +++++++++++++++++++++++++----------- 5 files changed, 53 insertions(+), 21 deletions(-) diff --git a/src/bitcoin/poller/looper.rs b/src/bitcoin/poller/looper.rs index 4bdb7350..6e75d4f5 100644 --- a/src/bitcoin/poller/looper.rs +++ b/src/bitcoin/poller/looper.rs @@ -40,7 +40,7 @@ fn update_tip(bit: &impl BitcoinInterface, db_conn: &mut Box>, - db: impl DatabaseInterface, + db: sync::Arc>, shutdown: sync::Arc, poll_interval: time::Duration, ) { diff --git a/src/bitcoin/poller/mod.rs b/src/bitcoin/poller/mod.rs index 23dd0b57..68901d61 100644 --- a/src/bitcoin/poller/mod.rs +++ b/src/bitcoin/poller/mod.rs @@ -19,7 +19,7 @@ pub struct Poller { impl Poller { pub fn start( bit: sync::Arc>, - db: impl DatabaseInterface + 'static, + db: sync::Arc>, poll_interval: time::Duration, ) -> Poller { let shutdown = sync::Arc::from(atomic::AtomicBool::from(false)); diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 733d2b56..e56904e1 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -2,7 +2,7 @@ //! //! External interface to the Minisafe daemon. -use crate::{bitcoin::BitcoinInterface, DaemonControl, VERSION}; +use crate::{bitcoin::BitcoinInterface, database::DatabaseInterface, DaemonControl, VERSION}; use miniscript::{ bitcoin, diff --git a/src/database/mod.rs b/src/database/mod.rs index dac70224..6820661b 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -8,6 +8,8 @@ use crate::{ database::sqlite::{schema::DbTip, SqliteConn, SqliteDb}, }; +use std::sync; + use miniscript::bitcoin::util::bip32; pub trait DatabaseInterface: Send { @@ -20,6 +22,13 @@ impl DatabaseInterface for SqliteDb { } } +// FIXME: do we need to repeat the entire trait implemenation? Isn't there a nicer way? +impl DatabaseInterface for sync::Arc> { + fn connection(&self) -> Box { + self.lock().unwrap().connection() + } +} + pub trait DatabaseConnection { /// Get the tip of the best chain we've seen. fn chain_tip(&mut self) -> Option; diff --git a/src/lib.rs b/src/lib.rs index a50dcdb9..4e1f0a14 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -148,6 +148,31 @@ fn create_datadir(datadir_path: &path::Path) -> Result<(), StartupError> { }; } +// Connect to the SQLite database. Create it if starting fresh, and do some sanity checks. +// If all went well, returns the interface to the SQLite database. +fn setup_sqlite( + config: &Config, + data_dir: &path::Path, + fresh_data_dir: bool, +) -> Result { + let db_path: path::PathBuf = [data_dir, path::Path::new("minisafed.sqlite3")] + .iter() + .collect(); + let options = if fresh_data_dir { + Some(FreshDbOptions { + bitcoind_network: config.bitcoind_config.network, + main_descriptor: config.main_descriptor.clone(), + }) + } else { + None + }; + let sqlite = SqliteDb::new(db_path, options)?; + sqlite.sanity_check(config.bitcoind_config.network, &config.main_descriptor)?; + log::info!("Database initialized and checked."); + + Ok(sqlite) +} + // Connect to bitcoind. Setup the watchonly wallet, and do some sanity checks. // If all went well, returns the interface to bitcoind. fn setup_bitcoind( @@ -177,7 +202,8 @@ fn setup_bitcoind( pub struct DaemonControl { config: Config, bitcoin: sync::Arc>, - db: Box, + // FIXME: Should we require Sync on DatabaseInterface rather than using a Mutex? + db: sync::Arc>, secp: secp256k1::Secp256k1, } @@ -185,7 +211,7 @@ impl DaemonControl { pub fn new( config: Config, bitcoin: sync::Arc>, - db: Box, + db: sync::Arc>, ) -> DaemonControl { let secp = secp256k1::Secp256k1::verification_only(); DaemonControl { @@ -207,12 +233,15 @@ impl DaemonHandle { /// /// You may specify a custom Bitcoin interface through the `bitcoin` parameter. If `None`, the /// default Bitcoin interface (`bitcoind` JSONRPC) will be used. + /// You may specify a custom Database interface through the `db` parameter. If `None`, the + /// default Database interface (SQLite) will be used. /// /// **Note**: we internally use threads, and set a panic hook. A downstream application must /// not overwrite this panic hook. pub fn start( config: Config, bitcoin: Option, + db: Option, ) -> Result { #[cfg(not(test))] setup_panic_hook(); @@ -229,20 +258,14 @@ impl DaemonHandle { } // Then set up the database - let db_path: path::PathBuf = [data_dir.as_path(), path::Path::new("minisafed.sqlite3")] - .iter() - .collect(); - let options = if fresh_data_dir { - Some(FreshDbOptions { - bitcoind_network: config.bitcoind_config.network, - main_descriptor: config.main_descriptor.clone(), - }) - } else { - None + let db = match db { + Some(db) => sync::Arc::from(sync::Mutex::from(db)), + None => sync::Arc::from(sync::Mutex::from(setup_sqlite( + &config, + &data_dir, + fresh_data_dir, + )?)) as sync::Arc>, }; - let sqlite = SqliteDb::new(db_path, options)?; - sqlite.sanity_check(config.bitcoind_config.network, &config.main_descriptor)?; - log::info!("Database initialized and checked."); // Now, set up the Bitcoin interface. let bit = match bitcoin { @@ -271,12 +294,12 @@ impl DaemonHandle { // Spawn the bitcoind poller with a retry limit high enough that we'd fail after that. let bitcoin_poller = poller::Poller::start( bit.clone(), - sqlite.clone(), + db.clone(), config.bitcoind_config.poll_interval_secs, ); // Finally, set up the API. - let control = DaemonControl::new(config, bit, Box::from(sqlite)); + let control = DaemonControl::new(config, bit, db); Ok(Self { control, @@ -287,7 +310,7 @@ impl DaemonHandle { /// Start the Minisafe daemon with the default Bitcoin and database interfaces (`bitcoind` RPC /// and SQLite). pub fn start_default(config: Config) -> Result { - DaemonHandle::start(config, Option::::None) + DaemonHandle::start(config, Option::::None, Option::::None) } /// Start the JSONRPC server and listen for incoming commands until we die. From e510c0a30dd18445d595e403d5cddd8fe04ef1b6 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Tue, 9 Aug 2022 12:58:35 +0200 Subject: [PATCH 11/19] config: separate the Bitcoin and bitcoind-specifc settings --- src/bin/cli.rs | 2 +- src/commands/mod.rs | 4 +-- src/config.rs | 44 ++++++++++++++++++++----------- src/lib.rs | 33 +++++++++++++++-------- tests/test_framework/minisafed.py | 6 +++-- 5 files changed, 58 insertions(+), 31 deletions(-) diff --git a/src/bin/cli.rs b/src/bin/cli.rs index 870bc271..38c6f569 100644 --- a/src/bin/cli.rs +++ b/src/bin/cli.rs @@ -93,7 +93,7 @@ fn socket_file(conf_file: Option) -> PathBuf { [ data_dir, - config.bitcoind_config.network.to_string().as_str(), + config.bitcoin_config.network.to_string().as_str(), "minisafed_rpc", ] .iter() diff --git a/src/commands/mod.rs b/src/commands/mod.rs index e56904e1..444b6d05 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -16,7 +16,7 @@ impl DaemonControl { pub fn get_info(&self) -> GetInfoResult { GetInfoResult { version: VERSION.to_string(), - network: self.config.bitcoind_config.network, + network: self.config.bitcoin_config.network, blockheight: self.bitcoin.chain_tip().height, sync: self.bitcoin.sync_progress(), descriptors: GetInfoDescriptors { @@ -39,7 +39,7 @@ impl DaemonControl { .derive(index.into()) .translate_pk2(|xpk| xpk.derive_public_key(&self.secp)) .expect("All pubkeys were derived, no wildcard.") - .address(self.config.bitcoind_config.network) + .address(self.config.bitcoin_config.network) .expect("It's a wsh() descriptor"); GetAddressResult { address } } diff --git a/src/config.rs b/src/config.rs index 42085b74..168746fc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -54,13 +54,17 @@ fn default_daemon() -> bool { /// Everything we need to know for talking to bitcoind serenely #[derive(Debug, Clone, Deserialize, Serialize)] pub struct BitcoindConfig { - /// The network we are operating on, one of "bitcoin", "testnet", "regtest" - pub network: Network, /// Path to bitcoind's cookie file, to authenticate the RPC connection pub cookie_path: PathBuf, /// The IP:port bitcoind's RPC is listening on pub addr: SocketAddr, - /// The poll interval for bitcoind +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct BitcoinConfig { + /// The network we are operating on, one of "bitcoin", "testnet", "regtest", "signet" + pub network: Network, + /// The poll interval for the Bitcoin interface #[serde( deserialize_with = "deserialize_duration", serialize_with = "serialize_duration", @@ -91,8 +95,10 @@ pub struct Config { serialize_with = "serialize_to_string" )] pub main_descriptor: Descriptor, - /// Everything we need to know to talk to bitcoind - pub bitcoind_config: BitcoindConfig, + /// Settings for the Bitcoin interface + pub bitcoin_config: BitcoinConfig, + /// Settings specific to bitcoind as the Bitcoin interface + pub bitcoind_config: Option, } impl Config { @@ -195,7 +201,7 @@ impl Config { /// Make sure the settings are sane. pub fn check(&self) -> Result<(), ConfigError> { // Check the network of the xpubs in the descriptors - let expected_network = match self.bitcoind_config.network { + let expected_network = match self.bitcoin_config.network { Network::Bitcoin => Network::Bitcoin, _ => Network::Testnet, }; @@ -214,7 +220,7 @@ impl Config { if unexpected_net { return Err(ConfigError::Unexpected(format!( "Our bitcoin network is {} but one xpub is not for network {}", - self.bitcoind_config.network, expected_network + self.bitcoin_config.network, expected_network ))); } @@ -238,11 +244,13 @@ mod tests { log_level = "debug" main_descriptor = "wsh(andor(thresh(1,pk(xpub6BaZSKgpaVvibu2k78QsqeDWXp92xLHZxiu1WoqLB9hKhsBf3miBUDX7PJLgSPvkj66ThVHTqdnbXpeu8crXFmDUd4HeM4s4miQS2xsv3Qb/*)),and_v(v:multi(2,03b506a1dbe57b4bf48c95e0c7d417b87dd3b4349d290d2e7e9ba72c912652d80a,0295e7f5d12a2061f1fd2286cefec592dff656a19f55f4f01305d6aa56630880ce),older(4)),thresh(2,pkh(xpub6AHA9hZDN11k2ijHMeS5QqHx2KP9aMBRhTDqANMnwVtdyw2TDYRmF8PjpvwUFcL1Et8Hj59S3gTSMcUQ5gAqTz3Wd8EsMTmF3DChhqPQBnU/*),a:pkh(xpub6AaffFGfH6WXfm6pwWzmUMuECQnoLeB3agMKaLyEBZ5ZVfwtnS5VJKqXBt8o5ooCWVy2H87GsZshp7DeKE25eWLyd1Ccuh2ZubQUkgpiVux/*))))#532k8uvf" - [bitcoind_config] + [bitcoin_config] network = "bitcoin" + poll_interval_secs = 18 + + [bitcoind_config] cookie_path = "/home/user/.bitcoin/.cookie" addr = "127.0.0.1:8332" - poll_interval_secs = 18 "#.trim_start().replace(" ", ""); toml::from_str::(&toml_str).expect("Deserializing toml_str"); @@ -253,11 +261,13 @@ mod tests { log_level = 'TRACE' main_descriptor = 'wsh(andor(thresh(1,pk(xpub6BaZSKgpaVvibu2k78QsqeDWXp92xLHZxiu1WoqLB9hKhsBf3miBUDX7PJLgSPvkj66ThVHTqdnbXpeu8crXFmDUd4HeM4s4miQS2xsv3Qb/*)),and_v(v:multi(2,03b506a1dbe57b4bf48c95e0c7d417b87dd3b4349d290d2e7e9ba72c912652d80a,0295e7f5d12a2061f1fd2286cefec592dff656a19f55f4f01305d6aa56630880ce),older(4)),thresh(2,pkh(xpub6AHA9hZDN11k2ijHMeS5QqHx2KP9aMBRhTDqANMnwVtdyw2TDYRmF8PjpvwUFcL1Et8Hj59S3gTSMcUQ5gAqTz3Wd8EsMTmF3DChhqPQBnU/*),a:pkh(xpub6AaffFGfH6WXfm6pwWzmUMuECQnoLeB3agMKaLyEBZ5ZVfwtnS5VJKqXBt8o5ooCWVy2H87GsZshp7DeKE25eWLyd1Ccuh2ZubQUkgpiVux/*))))#532k8uvf' - [bitcoind_config] + [bitcoin_config] network = 'bitcoin' + poll_interval_secs = 18 + + [bitcoind_config] cookie_path = '/home/user/.bitcoin/.cookie' addr = '127.0.0.1:8332' - poll_interval_secs = 18 "#.trim_start().replace(" ", ""); let parsed = toml::from_str::(&toml_str).expect("Deserializing toml_str"); let serialized = toml::to_string_pretty(&parsed).expect("Serializing to toml"); @@ -273,16 +283,18 @@ mod tests { # The main descriptor semantics aren't checked, yet. main_descriptor = "wsh(andor(thresh(1,pk(xpub6BaZSKgpaVvibu2k78QsqeDWXp92xLHZxiu1WoqLB9hKhsBf3miBUDX7PJLgSPvkj66ThVHTqdnbXpeu8crXFmDUd4HeM4s4miQS2xsv3Qb/*)),and_v(v:multi(2,03b506a1dbe57b4bf48c95e0c7d417b87dd3b4349d290d2e7e9ba72c912652d80a,0295e7f5d12a2061f1fd2286cefec592dff656a19f55f4f01305d6aa56630880ce),older(4)),thresh(2,pkh(xpub6AHA9hZDN11k2ijHMeS5QqHx2KP9aMBRhTDqANMnwVtdyw2TDYRmF8PjpvwUFcL1Et8Hj59S3gTSMcUQ5gAqTz3Wd8EsMTmF3DChhqPQBnU/*),a:pkh(xpub6AaffFGfH6WXfm6pwWzmUMuECQnoLeB3agMKaLyEBZ5ZVfwtnS5VJKqXBt8o5ooCWVy2H87GsZshp7DeKE25eWLyd1Ccuh2ZubQUkgpiVux/*))))#532k88vf" - [bitcoind_config] + [bitcoin_config] network = "bitcoin" + poll_interval_secs = 18 + + [bitcoind_config] cookie_path = "/home/user/.bitcoin/.cookie" addr = "127.0.0.1:8332" - poll_interval_secs = 18 "#; let config_res: Result = toml::from_str(toml_str); config_res.expect_err("Deserializing an invalid toml_str"); - // Not enough parameters: missing the network + // Not enough parameters: missing the Bitcoin network let toml_str = r#" daemon = false log_level = "trace" @@ -291,10 +303,12 @@ mod tests { # The main descriptor semantics aren't checked, yet. main_descriptor = "wsh(andor(thresh(1,pk(xpub6BaZSKgpaVvibu2k78QsqeDWXp92xLHZxiu1WoqLB9hKhsBf3miBUDX7PJLgSPvkj66ThVHTqdnbXpeu8crXFmDUd4HeM4s4miQS2xsv3Qb/*)),and_v(v:multi(2,03b506a1dbe57b4bf48c95e0c7d417b87dd3b4349d290d2e7e9ba72c912652d80a,0295e7f5d12a2061f1fd2286cefec592dff656a19f55f4f01305d6aa56630880ce),older(4)),thresh(2,pkh(xpub6AHA9hZDN11k2ijHMeS5QqHx2KP9aMBRhTDqANMnwVtdyw2TDYRmF8PjpvwUFcL1Et8Hj59S3gTSMcUQ5gAqTz3Wd8EsMTmF3DChhqPQBnU/*),a:pkh(xpub6AaffFGfH6WXfm6pwWzmUMuECQnoLeB3agMKaLyEBZ5ZVfwtnS5VJKqXBt8o5ooCWVy2H87GsZshp7DeKE25eWLyd1Ccuh2ZubQUkgpiVux/*))))#532k8uvf" + [bitcoin_config] + poll_interval_secs = 18 + [bitcoind_config] cookie_path = "/home/user/.bitcoin/.cookie" addr = "127.0.0.1:8332" - poll_interval_secs = 18 "#; let config_res: Result = toml::from_str(toml_str); config_res.expect_err("Deserializing an invalid toml_str"); diff --git a/src/lib.rs b/src/lib.rs index 4e1f0a14..cac92090 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -80,6 +80,7 @@ pub enum StartupError { Io(io::Error), DefaultDataDirNotFound, DatadirCreation(path::PathBuf, io::Error), + MissingBitcoindConfig, Database(SqliteDbError), Bitcoind(BitcoindError), #[cfg(unix)] @@ -98,6 +99,10 @@ impl fmt::Display for StartupError { f, "Could not create data directory at '{}': '{}'", dir_path.display(), e ), + Self::MissingBitcoindConfig => write!( + f, + "Our Bitcoin interface is bitcoind but we have no 'bitcoind_config' entry in the configuration." + ), Self::Database(e) => write!(f, "Error initializing database: '{}'.", e), Self::Bitcoind(e) => write!(f, "Error setting up bitcoind interface: '{}'.", e), #[cfg(unix)] @@ -160,14 +165,14 @@ fn setup_sqlite( .collect(); let options = if fresh_data_dir { Some(FreshDbOptions { - bitcoind_network: config.bitcoind_config.network, + bitcoind_network: config.bitcoin_config.network, main_descriptor: config.main_descriptor.clone(), }) } else { None }; let sqlite = SqliteDb::new(db_path, options)?; - sqlite.sanity_check(config.bitcoind_config.network, &config.main_descriptor)?; + sqlite.sanity_check(config.bitcoin_config.network, &config.main_descriptor)?; log::info!("Database initialized and checked."); Ok(sqlite) @@ -185,7 +190,10 @@ fn setup_bitcoind( .iter() .collect(); let bitcoind = BitcoinD::new( - &config.bitcoind_config, + config + .bitcoind_config + .as_ref() + .ok_or(StartupError::MissingBitcoindConfig)?, wo_path.to_str().expect("Must be valid unicode").to_string(), )?; if fresh_data_dir { @@ -193,7 +201,7 @@ fn setup_bitcoind( log::info!("Created a new watchonly wallet on bitcoind."); } bitcoind.try_load_watchonly_wallet(); - bitcoind.sanity_check(&config.main_descriptor, config.bitcoind_config.network)?; + bitcoind.sanity_check(&config.main_descriptor, config.bitcoin_config.network)?; log::info!("Connection to bitcoind established and checked."); Ok(bitcoind.with_retry_limit(None)) @@ -250,7 +258,7 @@ impl DaemonHandle { let mut data_dir = config .data_dir() .ok_or(StartupError::DefaultDataDirNotFound)?; - data_dir.push(config.bitcoind_config.network.to_string()); + data_dir.push(config.bitcoin_config.network.to_string()); let fresh_data_dir = !data_dir.as_path().exists(); if fresh_data_dir { create_datadir(&data_dir)?; @@ -295,7 +303,7 @@ impl DaemonHandle { let bitcoin_poller = poller::Poller::start( bit.clone(), db.clone(), - config.bitcoind_config.poll_interval_secs, + config.bitcoin_config.poll_interval_secs, ); // Finally, set up the API. @@ -328,7 +336,7 @@ impl DaemonHandle { .data_dir() .expect("Didn't fail at startup, must not now") .as_path(), - path::Path::new(&control.config.bitcoind_config.network.to_string()), + path::Path::new(&control.config.bitcoin_config.network.to_string()), path::Path::new("minisafed_rpc"), ] .iter() @@ -356,7 +364,7 @@ impl DaemonHandle { #[cfg(all(test, unix))] mod tests { use super::*; - use crate::config::BitcoindConfig; + use crate::config::{BitcoinConfig, BitcoindConfig}; use miniscript::{bitcoin, Descriptor, DescriptorPublicKey}; use std::{ @@ -543,18 +551,21 @@ mod tests { net::SocketAddrV4::new(net::Ipv4Addr::new(127, 0, 0, 1), 0).into(); let server = net::TcpListener::bind(&addr).unwrap(); let addr = server.local_addr().unwrap(); - let bitcoind_config = BitcoindConfig { + let bitcoin_config = BitcoinConfig { network, + poll_interval_secs: time::Duration::from_secs(2), + }; + let bitcoind_config = BitcoindConfig { addr, cookie_path: cookie.clone(), - poll_interval_secs: time::Duration::from_secs(2), }; // Create a dummy config with this bitcoind let desc_str = "wsh(andor(pk(xpub68JJTXc1MWK8KLW4HGLXZBJknja7kDUJuFHnM424LbziEXsfkh1WQCiEjjHw4zLqSUm4rvhgyGkkuRowE9tCJSgt3TQB5J3SKAbZ2SdcKST/*),older(10000),pk(xpub68JJTXc1MWK8PEQozKsRatrUHXKFNkD1Cb1BuQU9Xr5moCv87anqGyXLyUd4KpnDyZgo3gz4aN1r3NiaoweFW8UutBsBbgKHzaD5HkTkifK/*)))#tk6wzexy"; let desc = Descriptor::::from_str(desc_str).unwrap(); let config = Config { - bitcoind_config, + bitcoin_config, + bitcoind_config: Some(bitcoind_config), data_dir: Some(data_dir.clone()), #[cfg(unix)] daemon: false, diff --git a/tests/test_framework/minisafed.py b/tests/test_framework/minisafed.py index 0b9135d0..7e365746 100644 --- a/tests/test_framework/minisafed.py +++ b/tests/test_framework/minisafed.py @@ -33,11 +33,13 @@ class Minisafed(TailableProc): f.write(f'main_descriptor = "{main_desc}"\n') - f.write("[bitcoind_config]\n") + f.write("[bitcoin_config]\n") f.write('network = "regtest"\n') + f.write("poll_interval_secs = 1\n") + + f.write("[bitcoind_config]\n") f.write(f"cookie_path = '{bitcoind_cookie_path}'\n") f.write(f"addr = '127.0.0.1:{bitcoind_rpc_port}'\n") - f.write("poll_interval_secs = 1\n") def start(self): TailableProc.start(self) From d03c46996798d9e5d68858c4d4631e6b3295f9b3 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Tue, 9 Aug 2022 15:18:47 +0200 Subject: [PATCH 12/19] Introduce a testutils module with a DummyMinisafe for unit tests So we can have unit tests without a dummy bitcoind thread, as we currently do for the startup test. This commit also implements sanity checks for the two existing commands using this mechanism. --- src/bitcoin/mod.rs | 2 +- src/bitcoin/poller/mod.rs | 5 ++ src/commands/mod.rs | 36 +++++++++++ src/lib.rs | 30 ++++----- src/testutils.rs | 124 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 179 insertions(+), 18 deletions(-) create mode 100644 src/testutils.rs diff --git a/src/bitcoin/mod.rs b/src/bitcoin/mod.rs index 0bf4170d..54406ec3 100644 --- a/src/bitcoin/mod.rs +++ b/src/bitcoin/mod.rs @@ -9,7 +9,7 @@ use std::sync; use miniscript::bitcoin; /// Information about the best block in the chain -#[derive(Debug, Clone, Eq, PartialEq)] +#[derive(Debug, Clone, Eq, PartialEq, Copy)] pub struct BlockChainTip { pub hash: bitcoin::BlockHash, pub height: i32, diff --git a/src/bitcoin/poller/mod.rs b/src/bitcoin/poller/mod.rs index 68901d61..97a1c22d 100644 --- a/src/bitcoin/poller/mod.rs +++ b/src/bitcoin/poller/mod.rs @@ -38,4 +38,9 @@ impl Poller { self.shutdown.store(true, atomic::Ordering::Relaxed); self.handle.join().expect("The poller loop must not fail"); } + + #[cfg(test)] + pub fn test_stop(&mut self) { + self.shutdown.store(true, atomic::Ordering::Relaxed); + } } diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 444b6d05..37a81aea 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -64,3 +64,39 @@ pub struct GetInfoResult { pub struct GetAddressResult { pub address: bitcoin::Address, } + +#[cfg(test)] +mod tests { + use super::*; + use crate::testutils::*; + use std::str::FromStr; + + #[test] + fn getinfo() { + let ms = DummyMinisafe::new(); + // We can query getinfo + ms.handle.control.get_info(); + ms.shutdown(); + } + + #[test] + fn getnewaddress() { + let ms = DummyMinisafe::new(); + + let control = &ms.handle.control; + // We can get an address + let addr = control.get_new_address().address; + assert_eq!( + addr, + bitcoin::Address::from_str( + "bc1qgudekhcrejgtlx3yhlvdul7t4q76e5lhm0vtcsndxs6aslh4r9jsqkqhwu" + ) + .unwrap() + ); + // We won't get the same twice. + let addr2 = control.get_new_address().address; + assert_ne!(addr, addr2); + + ms.shutdown(); + } +} diff --git a/src/lib.rs b/src/lib.rs index cac92090..fc425cb0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,6 +7,8 @@ mod database; pub mod descriptors; #[cfg(feature = "jsonrpc_server")] mod jsonrpc; +#[cfg(test)] +mod testutils; pub use miniscript; @@ -359,6 +361,12 @@ impl DaemonHandle { pub fn shutdown(self) { self.bitcoin_poller.stop(); } + + // We need a shutdown utility that does not move for implementing Drop for the DummyMinisafe + #[cfg(test)] + pub fn test_shutdown(&mut self) { + self.bitcoin_poller.test_stop(); + } } #[cfg(all(test, unix))] @@ -515,6 +523,10 @@ mod tests { stream.flush().unwrap(); } + // TODO: we could move the dummy bitcoind thread stuff to the bitcoind module to test the + // bitcoind interface, and use the DummyMinisafe from testutils to sanity check the startup. + // Note that startup as checked by this unit test is also tested in the functional test + // framework. #[test] fn daemon_startup() { let tmp_dir = env::temp_dir().join(format!( @@ -578,21 +590,7 @@ mod tests { let config = config.clone(); move || { let handle = DaemonHandle::start_default(config).unwrap(); - // TODO: avoid scope creep. We should move the bitcoind-specific checks to the - // bitcoind module, test the startup with a mocked bitcoind interface, and not test - // commands here but in the commands module. - let addr = handle.control.get_new_address().address; - let addr2 = handle.control.get_new_address().address; - assert_eq!( - addr, - bitcoin::Address::from_str( - "bc1qdu9dama0pwc6fd9lj4sqzq4f728y5q2ucqyj55mfzfvuxr268zks7yajm3" - ) - .unwrap() - ); - assert_ne!(addr, addr2); handle.shutdown(); - addr } }); complete_sanity_check(&server); @@ -603,13 +601,11 @@ mod tests { complete_wallet_check(&server, &wo_path); complete_desc_check(&server, desc_str); complete_sync_check(&server); - let addr = daemon_thread.join().unwrap(); + daemon_thread.join().unwrap(); // The datadir is created now, so if we restart it it won't create the wo wallet. let daemon_thread = thread::spawn(move || { let handle = DaemonHandle::start_default(config).unwrap(); - // TODO: avoid scope creep. See above comment. - assert_ne!(handle.control.get_new_address().address, addr); handle.shutdown(); }); complete_sanity_check(&server); diff --git a/src/testutils.rs b/src/testutils.rs new file mode 100644 index 00000000..0d8d5cea --- /dev/null +++ b/src/testutils.rs @@ -0,0 +1,124 @@ +use crate::{ + bitcoin::{BitcoinInterface, BlockChainTip}, + config::{BitcoinConfig, Config}, + database::{DatabaseConnection, DatabaseInterface}, + DaemonControl, DaemonHandle, +}; + +use std::{env, fs, path, process, str::FromStr, sync, thread, time}; + +use miniscript::{ + bitcoin::{self, util::bip32}, + descriptor, +}; + +pub struct DummyBitcoind {} + +impl BitcoinInterface for DummyBitcoind { + fn sync_progress(&self) -> f64 { + 1.0 + } + + fn chain_tip(&self) -> BlockChainTip { + let hash = bitcoin::BlockHash::from_str( + "000000007bc154e0fa7ea32218a72fe2c1bb9f86cf8c9ebf9a715ed27fdb229a", + ) + .unwrap(); + let height = 100; + BlockChainTip { hash, height } + } + + fn is_in_chain(&self, _: &BlockChainTip) -> bool { + // No reorg + true + } +} + +pub struct DummyDb { + curr_index: bip32::ChildNumber, + curr_tip: Option, +} + +impl DummyDb { + pub fn new() -> DummyDb { + DummyDb { + curr_index: 0.into(), + curr_tip: None, + } + } +} + +impl DatabaseInterface for sync::Arc> { + fn connection(&self) -> Box { + Box::new(DummyDbConn { db: self.clone() }) + } +} + +pub struct DummyDbConn { + db: sync::Arc>, +} + +impl DatabaseConnection for DummyDbConn { + fn chain_tip(&mut self) -> Option { + self.db.read().unwrap().curr_tip + } + + fn update_tip(&mut self, tip: &BlockChainTip) { + self.db.write().unwrap().curr_tip = Some(*tip); + } + + fn derivation_index(&mut self) -> bip32::ChildNumber { + self.db.read().unwrap().curr_index + } + + fn update_derivation_index(&mut self, index: bip32::ChildNumber) { + self.db.write().unwrap().curr_index = index; + } +} + +pub struct DummyMinisafe { + tmp_dir: path::PathBuf, + pub handle: DaemonHandle, +} + +impl DummyMinisafe { + pub fn new() -> DummyMinisafe { + let tmp_dir = env::temp_dir().join(format!( + "minisafed-unit-tests-{}-{:?}", + process::id(), + thread::current().id() + )); + fs::create_dir_all(&tmp_dir).unwrap(); + let data_dir: path::PathBuf = [tmp_dir.as_path(), path::Path::new("datadir")] + .iter() + .collect(); + + let network = bitcoin::Network::Bitcoin; + let bitcoin_config = BitcoinConfig { + network, + poll_interval_secs: time::Duration::from_secs(2), + }; + + let owner_key = descriptor::DescriptorPublicKey::from_str("xpub68JJTXc1MWK8KLW4HGLXZBJknja7kDUJuFHnM424LbziEXsfkh1WQCiEjjHw4zLqSUm4rvhgyGkkuRowE9tCJSgt3TQB5J3SKAbZ2SdcKST/*").unwrap(); + let heir_key = descriptor::DescriptorPublicKey::from_str("xpub68JJTXc1MWK8PEQozKsRatrUHXKFNkD1Cb1BuQU9Xr5moCv87anqGyXLyUd4KpnDyZgo3gz4aN1r3NiaoweFW8UutBsBbgKHzaD5HkTkifK/*").unwrap(); + let desc = crate::descriptors::inheritance_descriptor(owner_key, heir_key, 10_000).unwrap(); + let config = Config { + bitcoin_config, + bitcoind_config: None, + data_dir: Some(data_dir.clone()), + #[cfg(unix)] + daemon: false, + log_level: log::LevelFilter::Debug, + main_descriptor: desc, + }; + + let db = sync::Arc::from(sync::RwLock::from(DummyDb::new())); + let handle = DaemonHandle::start(config, Some(DummyBitcoind {}), Some(db)).unwrap(); + DummyMinisafe { tmp_dir, handle } + } + + pub fn shutdown(self) { + self.handle.shutdown(); + fs::remove_dir_all(&self.tmp_dir).unwrap(); + } +} From 726209cc0a07d12c71041573313d5b34def30bbd Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Wed, 10 Aug 2022 18:30:41 +0200 Subject: [PATCH 13/19] jsonrpc: add a 'stop' command This also implements a simple JSONRPC server sanity check And introduces a new tmp_dir() test utility, that i didn't have the courage to split from this commit --- src/database/sqlite/mod.rs | 15 +++------- src/jsonrpc/api.rs | 1 + src/jsonrpc/server.rs | 60 +++++++++++++++++++++++++++++++++----- src/lib.rs | 15 +++++----- src/testutils.rs | 41 +++++++++++++++++++++----- 5 files changed, 98 insertions(+), 34 deletions(-) diff --git a/src/database/sqlite/mod.rs b/src/database/sqlite/mod.rs index 1bc73aa2..2e81b54b 100644 --- a/src/database/sqlite/mod.rs +++ b/src/database/sqlite/mod.rs @@ -219,7 +219,8 @@ impl SqliteConn { #[cfg(test)] mod tests { use super::*; - use std::{env, fs, path, process, str::FromStr, thread}; + use crate::testutils::*; + use std::{fs, path, str::FromStr}; fn dummy_options() -> FreshDbOptions { let desc_str = "wsh(andor(pk(03b506a1dbe57b4bf48c95e0c7d417b87dd3b4349d290d2e7e9ba72c912652d80a),older(10000),pk(0295e7f5d12a2061f1fd2286cefec592dff656a19f55f4f01305d6aa56630880ce)))"; @@ -232,11 +233,7 @@ mod tests { #[test] fn db_startup_sanity_checks() { - let tmp_dir = env::temp_dir().join(format!( - "minisafed-unit-tests-{}-{:?}", - process::id(), - thread::current().id() - )); + let tmp_dir = tmp_dir(); fs::create_dir_all(&tmp_dir).unwrap(); let db_path: path::PathBuf = [tmp_dir.as_path(), path::Path::new("minisafed.sqlite3")] @@ -277,11 +274,7 @@ mod tests { #[test] fn db_tip_update() { - let tmp_dir = env::temp_dir().join(format!( - "minisafed-unit-tests-{}-{:?}", - process::id(), - thread::current().id() - )); + let tmp_dir = tmp_dir(); fs::create_dir_all(&tmp_dir).unwrap(); let db_path: path::PathBuf = [tmp_dir.as_path(), path::Path::new("minisafed.sqlite3")] diff --git a/src/jsonrpc/api.rs b/src/jsonrpc/api.rs index f3a7f9a1..82fe1a42 100644 --- a/src/jsonrpc/api.rs +++ b/src/jsonrpc/api.rs @@ -8,6 +8,7 @@ pub fn handle_request(control: &DaemonControl, req: Request) -> Result serde_json::json!(&control.get_info()), "getnewaddress" => serde_json::json!(&control.get_new_address()), + "stop" => serde_json::json!({}), _ => { return Err(Error::method_not_found()); } diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs index a0dad02d..b94d7150 100644 --- a/src/jsonrpc/server.rs +++ b/src/jsonrpc/server.rs @@ -83,12 +83,13 @@ fn read_command( fn connection_handler( control: sync::Arc>, mut stream: net::UnixStream, + shutdown: sync::Arc, ) -> Result<(), io::Error> { let mut buf = vec![0; 2048]; let mut end = 0; let mut cursor = 0; - loop { + while !shutdown.load(atomic::Ordering::Relaxed) { let req = match read_command(&mut stream, &mut buf, &mut end, &mut cursor)? { Some(req) => req, None => { @@ -97,8 +98,10 @@ fn connection_handler( } }; - // TODO: respond in case of invalid JSON or invalid JSONRPC request. let req_id = req.id.clone(); + if &req.method == "stop" { + shutdown.store(true, atomic::Ordering::Relaxed); + } log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req)); let response = api::handle_request(&control.lock().unwrap(), req) @@ -109,6 +112,8 @@ fn connection_handler( return Ok(()); } } + + Ok(()) } // FIXME: have a decent way to share the DaemonControl between connections. Maybe make it Clone? @@ -120,12 +125,14 @@ pub fn rpcserver_loop( // Keep it simple. We don't need great performances so just treat each connection in // its thread, with a given maximum number of connections. let connections_counter = sync::Arc::from(atomic::AtomicU32::new(0)); + let shutdown = sync::Arc::from(atomic::AtomicBool::new(false)); - loop { + listener.set_nonblocking(true)?; + while !shutdown.load(atomic::Ordering::Relaxed) { let (connection, _) = match listener.accept() { Ok(c) => c, - Err(e) => { - log::error!("Accepting new connection: '{}'", e); + Err(_) => { + thread::sleep(time::Duration::from_millis(100)); continue; } }; @@ -142,9 +149,10 @@ pub fn rpcserver_loop( .spawn({ let control = daemon_control.clone(); let counter = connections_counter.clone(); + let shutdown = shutdown.clone(); move || { - if let Err(e) = connection_handler(control, connection) { + if let Err(e) = connection_handler(control, connection, shutdown) { log::error!("Error while handling connection {}: '{}'", handler_id, e); } else { log::trace!("Connection {} terminated without error.", handler_id); @@ -153,6 +161,8 @@ pub fn rpcserver_loop( } })?; } + + Ok(()) } // Tries to bind to the socket, if we are told it's already in use try to connect @@ -195,7 +205,10 @@ pub fn rpcserver_setup(socket_path: &path::Path) -> Result usize { + unsafe { + let uid = COUNTER.load(sync::atomic::Ordering::Relaxed); + COUNTER.fetch_add(1, sync::atomic::Ordering::Relaxed); + uid + } +} + +pub fn tmp_dir() -> path::PathBuf { + env::temp_dir().join(format!( + "minisafed-unit-tests-{}-{:?}-{}-{}", + process::id(), + thread::current().id(), + time::SystemTime::now() + .duration_since(time::UNIX_EPOCH) + .unwrap() + .subsec_nanos(), + uid(), + )) +} + impl DummyMinisafe { pub fn new() -> DummyMinisafe { - let tmp_dir = env::temp_dir().join(format!( - "minisafed-unit-tests-{}-{:?}", - process::id(), - thread::current().id() - )); + let tmp_dir = tmp_dir(); fs::create_dir_all(&tmp_dir).unwrap(); let data_dir: path::PathBuf = [tmp_dir.as_path(), path::Path::new("datadir")] .iter() @@ -117,6 +135,13 @@ impl DummyMinisafe { DummyMinisafe { tmp_dir, handle } } + #[cfg(feature = "jsonrpc_server")] + pub fn rpc_server(self) -> Result<(), io::Error> { + self.handle.rpc_server()?; + fs::remove_dir_all(&self.tmp_dir)?; + Ok(()) + } + pub fn shutdown(self) { self.handle.shutdown(); fs::remove_dir_all(&self.tmp_dir).unwrap(); From eeac67dc0a7724ef4ca9003b7e734f546e62c0ae Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Wed, 10 Aug 2022 19:04:40 +0200 Subject: [PATCH 14/19] qa: shut down the daemon via the JSONRPC interface at teardown --- src/jsonrpc/server.rs | 1 + tests/test_framework/minisafed.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs index b94d7150..4100f990 100644 --- a/src/jsonrpc/server.rs +++ b/src/jsonrpc/server.rs @@ -101,6 +101,7 @@ fn connection_handler( let req_id = req.id.clone(); if &req.method == "stop" { shutdown.store(true, atomic::Ordering::Relaxed); + log::info!("Stopping the minisafe daemon."); } log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req)); diff --git a/tests/test_framework/minisafed.py b/tests/test_framework/minisafed.py index 7e365746..05e36658 100644 --- a/tests/test_framework/minisafed.py +++ b/tests/test_framework/minisafed.py @@ -1,3 +1,4 @@ +import logging import os from test_framework.utils import ( @@ -52,6 +53,14 @@ class Minisafed(TailableProc): ) def stop(self, timeout=5): + try: + self.rpc.stop() + self.wait_for_log( + "Stopping the minisafe daemon.", + ) + self.proc.wait(timeout) + except Exception as e: + logging.error(f"{self.prefix} : error when calling stop: '{e}'") return TailableProc.stop(self) def cleanup(self): From 14bd3890dcf6c5dc9fbe25992710c80c9ff107c7 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Wed, 10 Aug 2022 19:05:52 +0200 Subject: [PATCH 15/19] doc: add an API.md documenting the JSONRPC API. Co-Authored-By: Edouard Paris --- doc/API.md | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 doc/API.md diff --git a/doc/API.md b/doc/API.md new file mode 100644 index 00000000..3370b8c4 --- /dev/null +++ b/doc/API.md @@ -0,0 +1,66 @@ +# Minisafe API + +`minisafe` exposes a [JSON-RPC 2.0](https://www.jsonrpc.org/specification) +interface over a Unix Domain socket. + +Commands must be sent as valid JSONRPC 2.0 requests, ending with a `\n`. + +| Command | Description | +| ----------------------------------------------------------- | ---------------------------------------------------- | +| [`stop`](#stop) | Stops the minisafe daemon | +| [`getinfo`](#getinfo) | Get general information about the daemon | +| [`getnewaddress`](#getnewaddress) | Get a new receiving address | + +# Reference + +## General + +### `stop` + +Stops the minisafe daemon. + +#### Response + +Returns an empty response. + +| Field | Type | Description | +| ------------- | ------ | ----------- | + +### `getinfo` + +General information about the daemon + +#### Request + +This command does not take any parameter for now. + +| Field | Type | Description | +| ------------- | ----------------- | ----------------------------------------------------------- | + +#### Response + +| Field | Type | Description | +| -------------------- | ------- | -------------------------------------------------------------------------------------------- | +| `version` | string | Version following the [SimVer](http://www.simver.org/) format | +| `network` | string | Answer can be `mainnet`, `testnet`, `regtest` | +| `blockheight` | integer | Current block height | +| `sync` | float | The synchronization progress as percentage (`0 < sync < 1`) | +| `descriptors` | object | Object with the name of the descriptor as key and the descriptor string as value | + +### `getnewaddress` + +Get a new address for receiving coins. This will always generate a new address regardless of whether +it was used or not. + +#### Request + +This command does not take any parameter for now. + +| Field | Type | Description | +| ------------- | ----------------- | ----------------------------------------------------------- | + +#### Response + +| Field | Type | Description | +| ------------- | ------ | ------------------ | +| `address` | string | A Bitcoin address | From e6fcdd5b095232729cbeef4bd42bf6e57d77ffde Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Thu, 11 Aug 2022 13:03:51 +0200 Subject: [PATCH 16/19] jsonrpc: don't umask in unit tests It can interfere with the creation of dummy directories in other threads and cause spurious errors --- src/jsonrpc/server.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs index 4100f990..4d6a45c1 100644 --- a/src/jsonrpc/server.rs +++ b/src/jsonrpc/server.rs @@ -194,8 +194,10 @@ fn bind(socket_path: &path::Path) -> Result { pub fn rpcserver_setup(socket_path: &path::Path) -> Result { log::debug!("Binding socket at {}", socket_path.display()); // Create the socket with RW permissions only for the user + #[cfg(not(test))] let old_umask = unsafe { libc::umask(0o177) }; let listener = bind(&socket_path); + #[cfg(not(test))] unsafe { libc::umask(old_umask); } From 069bca546a23ce1b1714b9564973f13a6f090536 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Thu, 11 Aug 2022 13:12:26 +0200 Subject: [PATCH 17/19] jsonrpc: don't use a Mutex to share the DaemonControl between threads --- src/jsonrpc/server.rs | 8 ++++---- src/lib.rs | 3 +-- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs index 4d6a45c1..d2933e99 100644 --- a/src/jsonrpc/server.rs +++ b/src/jsonrpc/server.rs @@ -81,7 +81,7 @@ fn read_command( // Handle all messages from this connection. fn connection_handler( - control: sync::Arc>, + control: DaemonControl, mut stream: net::UnixStream, shutdown: sync::Arc, ) -> Result<(), io::Error> { @@ -105,8 +105,8 @@ fn connection_handler( } log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req)); - let response = api::handle_request(&control.lock().unwrap(), req) - .unwrap_or_else(|e| Response::error(req_id, e)); + let response = + api::handle_request(&control, req).unwrap_or_else(|e| Response::error(req_id, e)); log::trace!("JSONRPC response: {:?}", serde_json::to_string(&response)); if let Err(e) = serde_json::to_writer(&stream, &response) { log::error!("Error writing response: '{}'", e); @@ -121,7 +121,7 @@ fn connection_handler( /// The main event loop. Wait for connections, and treat requests sent through them. pub fn rpcserver_loop( listener: net::UnixListener, - daemon_control: sync::Arc>, + daemon_control: DaemonControl, ) -> Result<(), io::Error> { // Keep it simple. We don't need great performances so just treat each connection in // its thread, with a given maximum number of connections. diff --git a/src/lib.rs b/src/lib.rs index 5fcdac43..80f0f71c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -209,6 +209,7 @@ fn setup_bitcoind( Ok(bitcoind.with_retry_limit(None)) } +#[derive(Clone)] pub struct DaemonControl { config: Config, bitcoin: sync::Arc>, @@ -346,8 +347,6 @@ impl DaemonHandle { let listener = rpcserver_setup(&rpc_socket)?; log::info!("JSONRPC server started."); - // FIXME: don't use a Mutex... - let control = sync::Arc::from(sync::Mutex::from(control)); rpcserver_loop(listener, control)?; log::info!("JSONRPC server stopped."); From fe5de96312dc8d57f982c060d4b9b045c2194163 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Thu, 11 Aug 2022 15:07:32 +0200 Subject: [PATCH 18/19] Shorten the paths to the dummy datadirs in the unit tests It failed CI on MacOS because of overflowing SUN_PATH_LEN. --- src/jsonrpc/server.rs | 2 +- src/testutils.rs | 11 +++-------- 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs index d2933e99..ce18f94a 100644 --- a/src/jsonrpc/server.rs +++ b/src/jsonrpc/server.rs @@ -401,7 +401,7 @@ mod tests { let ms = DummyMinisafe::new(); let socket_path: path::PathBuf = [ ms.tmp_dir.as_path(), - path::Path::new("datadir"), + path::Path::new("d"), path::Path::new("bitcoin"), path::Path::new("minisafed_rpc"), ] diff --git a/src/testutils.rs b/src/testutils.rs index ff61a6d9..f38d73b3 100644 --- a/src/testutils.rs +++ b/src/testutils.rs @@ -92,13 +92,9 @@ fn uid() -> usize { pub fn tmp_dir() -> path::PathBuf { env::temp_dir().join(format!( - "minisafed-unit-tests-{}-{:?}-{}-{}", + "minisafed-{}-{:?}-{}", process::id(), thread::current().id(), - time::SystemTime::now() - .duration_since(time::UNIX_EPOCH) - .unwrap() - .subsec_nanos(), uid(), )) } @@ -107,9 +103,8 @@ impl DummyMinisafe { pub fn new() -> DummyMinisafe { let tmp_dir = tmp_dir(); fs::create_dir_all(&tmp_dir).unwrap(); - let data_dir: path::PathBuf = [tmp_dir.as_path(), path::Path::new("datadir")] - .iter() - .collect(); + // Use a shorthand for 'datadir', to avoid overflowing SUN_LEN on MacOS. + let data_dir: path::PathBuf = [tmp_dir.as_path(), path::Path::new("d")].iter().collect(); let network = bitcoin::Network::Bitcoin; let bitcoin_config = BitcoinConfig { From 798bf67e21977ad18ee680a14e23e534f64a3ea5 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Wed, 17 Aug 2022 19:39:39 +0200 Subject: [PATCH 19/19] commands: derive Deserialize for commands results --- src/commands/mod.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 37a81aea..f4c56fc8 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -45,13 +45,13 @@ impl DaemonControl { } } -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct GetInfoDescriptors { pub main: descriptor::Descriptor, } /// Information about the daemon -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct GetInfoResult { pub version: String, pub network: bitcoin::Network,