#!/usr/bin/env python3 # Copyright (c) 2017-2024 The Zcash developers # Distributed under the MIT software license, see the accompanying # file COPYING or https://www.opensource.org/licenses/mit-license.php . from decimal import Decimal from test_framework.test_framework import BitcoinTestFramework from test_framework.authproxy import JSONRPCException from test_framework.util import ( assert_equal, assert_greater_than, assert_raises_message, start_nodes, initialize_chain_clean, connect_nodes_bi, wait_and_assert_operationid_status, ) from test_framework.zip317 import conventional_fee from functools import reduce import logging import sys logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO, stream=sys.stdout) class ZkeyImportExportTest (BitcoinTestFramework): def setup_chain(self): print("Initializing test directory "+self.options.tmpdir) initialize_chain_clean(self.options.tmpdir, 5) def setup_network(self, split=False): self.nodes = start_nodes(5, self.options.tmpdir, extra_args=[[ '-allowdeprecated=getnewaddress', '-allowdeprecated=z_getnewaddress', '-allowdeprecated=z_getbalance', '-allowdeprecated=z_gettotalbalance', ]] * 5) connect_nodes_bi(self.nodes,0,1) connect_nodes_bi(self.nodes,1,2) connect_nodes_bi(self.nodes,0,2) connect_nodes_bi(self.nodes,0,3) connect_nodes_bi(self.nodes,0,4) self.is_network_split=False self.sync_all() def run_test(self): [alice, bob, charlie, daira, miner] = self.nodes fee = conventional_fee(2) # the sender loses 'amount' plus fee; to_addr receives exactly 'amount' def z_send(from_node, from_addr, to_addr, amount): recipients = [{"address": to_addr, "amount": amount}] opid = from_node.z_sendmany(from_addr, recipients, 1, fee) wait_and_assert_operationid_status(from_node, opid) self.sync_all() miner.generate(1) self.sync_all() def verify_utxos(node, amounts, zaddr): amounts.sort(reverse=True) txs = node.z_listreceivedbyaddress(zaddr) txs.sort(key=lambda x: x["amount"], reverse=True) print("Sorted txs", txs) print("amounts", amounts) try: assert_equal(amounts, [tx["amount"] for tx in txs]) for tx in txs: # make sure Sapling outputs exist and have valid values assert_equal("outindex" in tx, True) assert_greater_than(tx["outindex"], -1) except AssertionError: logging.error( 'Expected amounts: %r; txs: %r', amounts, txs) raise def get_private_balance(node): balance = node.z_gettotalbalance() return balance['private'] # Seed Alice with some funds alice.generate(10) self.sync_all() miner.generate(100) self.sync_all() # Shield Alice's coinbase funds to her zaddr alice_zaddr = alice.z_getnewaddress() res = alice.z_shieldcoinbase("*", alice_zaddr) wait_and_assert_operationid_status(alice, res['opid']) self.sync_all() miner.generate(1) self.sync_all() # Now get a pristine z-address for receiving transfers: bob_zaddr = bob.z_getnewaddress() verify_utxos(bob, [], bob_zaddr) assert_raises_message(JSONRPCException, "From address does not belong to this node", charlie.z_listreceivedbyaddress, bob_zaddr) # the amounts of each txn embodied which generates a single UTXO: amounts = list(map(Decimal, ['2.3', '3.7', '0.1', '0.5', '1.0', '0.19'])) # Internal test consistency assertion: assert_greater_than( Decimal(get_private_balance(alice)), reduce(Decimal.__add__, amounts)) logging.info("Sending pre-export txns...") for amount in amounts[0:2]: z_send(alice, alice_zaddr, bob_zaddr, amount) logging.info("Exporting privkey from bob...") bob_privkey = bob.z_exportkey(bob_zaddr) logging.info("Sending post-export txns...") for amount in amounts[2:4]: z_send(alice, alice_zaddr, bob_zaddr, amount) verify_utxos(bob, amounts[:4], bob_zaddr) assert_raises_message(JSONRPCException, "From address does not belong to this node", charlie.z_listreceivedbyaddress, bob_zaddr) logging.info("Importing bob_privkey into charlie...") # z_importkey rescan defaults to "whenkeyisnew", so should rescan here ipk_zaddr = charlie.z_importkey(bob_privkey) # z_importkey should have rescanned for new key, so this should pass: verify_utxos(charlie, amounts[:4], ipk_zaddr["address"]) # address is Sapling assert_equal(ipk_zaddr["address_type"], "sapling") # Verify idempotent behavior: ipk_zaddr2 = charlie.z_importkey(bob_privkey) assert_equal(ipk_zaddr["address"], ipk_zaddr2["address"]) # amounts should be unchanged verify_utxos(charlie, amounts[:4], ipk_zaddr2["address"]) logging.info("Sending post-import txns...") for amount in amounts[4:]: z_send(alice, alice_zaddr, bob_zaddr, amount) verify_utxos(bob, amounts, bob_zaddr) verify_utxos(charlie, amounts, ipk_zaddr["address"]) verify_utxos(charlie, amounts, ipk_zaddr2["address"]) # keep track of bob's expected balance bob_balance = sum(amounts) # Try to reproduce zombie balance reported in #1936 # At generated zaddr, receive ZEC, and send ZEC back out. bob -> alice for amount in amounts[:2]: print("Sending amount from bob to alice: ", amount) z_send(bob, bob_zaddr, alice_zaddr, amount) bob_balance -= amount + fee assert_equal(bob.z_getbalance(bob_zaddr), bob_balance) # z_import onto new node "daira" (rescans by default because the key is new) d_ipk_zaddr = daira.z_importkey(bob_privkey) # Check if amount bob spent is deducted for charlie and daira assert_equal(charlie.z_getbalance(ipk_zaddr["address"]), bob_balance) assert_equal(daira.z_getbalance(d_ipk_zaddr["address"]), bob_balance) if __name__ == '__main__': ZkeyImportExportTest().main()