wallet_labels.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2016-present The Bitcoin Core developers 3 # Distributed under the MIT software license, see the accompanying 4 # file COPYING or http://www.opensource.org/licenses/mit-license.php. 5 """Test label RPCs. 6 7 RPCs tested are: 8 - getaddressesbylabel 9 - listaddressgroupings 10 - setlabel 11 """ 12 from collections import defaultdict 13 14 from test_framework.blocktools import COINBASE_MATURITY 15 from test_framework.descriptors import descsum_create 16 from test_framework.test_framework import BitcoinTestFramework 17 from test_framework.util import assert_equal, assert_raises_rpc_error 18 from test_framework.wallet_util import test_address 19 20 21 class WalletLabelsTest(BitcoinTestFramework): 22 def set_test_params(self): 23 self.setup_clean_chain = True 24 self.num_nodes = 2 25 26 def skip_test_if_missing_module(self): 27 self.skip_if_no_wallet() 28 29 def invalid_label_name_test(self): 30 node = self.nodes[0] 31 address = node.getnewaddress() 32 pubkey = node.getaddressinfo(address)['pubkey'] 33 rpc_calls = [ 34 [node.getnewaddress], 35 [node.setlabel, address], 36 [node.getaddressesbylabel], 37 [node.getreceivedbylabel], 38 [node.listsinceblock, node.getblockhash(0), 1, False, True, False], 39 ] 40 response = node.importdescriptors([{ 41 'desc': f'pkh({pubkey})', 42 'label': '*', 43 'timestamp': 'now', 44 }]) 45 46 assert_equal(response[0]['success'], False) 47 assert_equal(response[0]['error']['code'], -11) 48 assert_equal(response[0]['error']['message'], "Invalid label name") 49 50 for rpc_call in rpc_calls: 51 assert_raises_rpc_error(-11, "Invalid label name", *rpc_call, label="*") 52 53 def test_label_named_parameter_handling(self): 54 """Test that getnewaddress with labels containing '=' characters is handled correctly in -named mode""" 55 self.log.info("Test getnewaddress label parameter handling") 56 node = self.nodes[0] 57 58 # Test getnewaddress with explicit named parameter containing '=' 59 label_with_equals = "wallet=wallet" 60 result = node.cli("-named", "getnewaddress", f"label={label_with_equals}").send_cli() 61 address = result.strip() 62 addr_info = node.getaddressinfo(address) 63 assert_equal(addr_info.get('labels', []), [label_with_equals]) 64 65 self.log.info("Test bitcoin-cli -named passes parameter containing '=' by position if it does not specify a known parameter name and is in a string position") 66 equals_label = "my=label" 67 result = node.cli("-named", "getnewaddress", equals_label).send_cli() 68 address = result.strip() 69 addr_info = node.getaddressinfo(address) 70 assert_equal(addr_info.get('labels', []), [equals_label]) 71 72 self.log.info("getnewaddress label parameter handling test completed successfully") 73 74 def run_test(self): 75 # Check that there's no UTXO on the node 76 node = self.nodes[0] 77 assert_equal(len(node.listunspent()), 0) 78 79 self.log.info("Checking listlabels' invalid parameters") 80 assert_raises_rpc_error(-8, "Invalid 'purpose' argument, must be a known purpose string, typically 'send', or 'receive'.", node.listlabels, "notavalidpurpose") 81 assert_raises_rpc_error(-8, "Invalid 'purpose' argument, must be a known purpose string, typically 'send', or 'receive'.", node.listlabels, "unknown") 82 83 # Note each time we call generate, all generated coins go into 84 # the same address, so we call twice to get two addresses w/50 each 85 self.generatetoaddress(node, nblocks=1, address=node.getnewaddress(label='coinbase')) 86 self.generatetoaddress(node, nblocks=COINBASE_MATURITY + 1, address=node.getnewaddress(label='coinbase')) 87 assert_equal(node.getbalance(), 100) 88 89 # there should be 2 address groups 90 # each with 1 address with a balance of 50 Bitcoins 91 address_groups = node.listaddressgroupings() 92 assert_equal(len(address_groups), 2) 93 # the addresses aren't linked now, but will be after we send to the 94 # common address 95 linked_addresses = set() 96 for address_group in address_groups: 97 assert_equal(len(address_group), 1) 98 assert_equal(len(address_group[0]), 3) 99 assert_equal(address_group[0][1], 50) 100 assert_equal(address_group[0][2], 'coinbase') 101 linked_addresses.add(address_group[0][0]) 102 103 # send 50 from each address to a third address not in this wallet 104 common_address = "msf4WtN1YQKXvNtvdFYt9JBnUD2FB41kjr" 105 node.sendmany( 106 amounts={common_address: 100}, 107 subtractfeefrom=[common_address], 108 minconf=1, 109 ) 110 # there should be 1 address group, with the previously 111 # unlinked addresses now linked (they both have 0 balance) 112 address_groups = node.listaddressgroupings() 113 assert_equal(len(address_groups), 1) 114 assert_equal(len(address_groups[0]), 2) 115 assert_equal(set([a[0] for a in address_groups[0]]), linked_addresses) 116 assert_equal([a[1] for a in address_groups[0]], [0, 0]) 117 118 self.generate(node, 1) 119 120 # we want to reset so that the "" label has what's expected. 121 # otherwise we're off by exactly the fee amount as that's mined 122 # and matures in the next 100 blocks 123 amount_to_send = 1.0 124 125 # Create labels and make sure subsequent label API calls 126 # recognize the label/address associations. 127 labels = [Label(name) for name in ("a", "b", "c", "d", "e")] 128 for label in labels: 129 address = node.getnewaddress(label.name) 130 label.add_receive_address(address) 131 label.verify(node) 132 133 # Check listlabels when passing 'purpose' 134 node2_addr = self.nodes[1].getnewaddress() 135 node.setlabel(node2_addr, "node2_addr") 136 assert_equal(node.listlabels(purpose="send"), ["node2_addr"]) 137 assert_equal(node.listlabels(purpose="receive"), sorted(['coinbase'] + [label.name for label in labels])) 138 139 # Check all labels are returned by listlabels. 140 assert_equal(node.listlabels(), sorted(['coinbase'] + [label.name for label in labels] + ["node2_addr"])) 141 142 # Send a transaction to each label. 143 for label in labels: 144 node.sendtoaddress(label.addresses[0], amount_to_send) 145 label.verify(node) 146 147 # Check the amounts received. 148 self.generate(node, 1) 149 for label in labels: 150 assert_equal( 151 node.getreceivedbyaddress(label.addresses[0]), amount_to_send) 152 assert_equal(node.getreceivedbylabel(label.name), amount_to_send) 153 154 for i, label in enumerate(labels): 155 to_label = labels[(i + 1) % len(labels)] 156 node.sendtoaddress(to_label.addresses[0], amount_to_send) 157 self.generate(node, 1) 158 for label in labels: 159 address = node.getnewaddress(label.name) 160 label.add_receive_address(address) 161 label.verify(node) 162 assert_equal(node.getreceivedbylabel(label.name), 2) 163 label.verify(node) 164 self.generate(node, COINBASE_MATURITY + 1) 165 166 # Check that setlabel can assign a label to a new unused address. 167 for label in labels: 168 address = node.getnewaddress() 169 node.setlabel(address, label.name) 170 label.add_address(address) 171 label.verify(node) 172 assert_raises_rpc_error(-11, "No addresses with label", node.getaddressesbylabel, "") 173 174 # Check that setlabel can change the label of an address from a 175 # different label. 176 change_label(node, labels[0].addresses[0], labels[0], labels[1]) 177 178 # Check that setlabel can set the label of an address already 179 # in the label. This is a no-op. 180 change_label(node, labels[2].addresses[0], labels[2], labels[2]) 181 182 self.invalid_label_name_test() 183 self.test_label_named_parameter_handling() 184 185 # This is a descriptor wallet test because of segwit v1+ addresses 186 self.log.info('Check watchonly labels') 187 node.createwallet(wallet_name='watch_only', disable_private_keys=True) 188 wallet_watch_only = node.get_wallet_rpc('watch_only') 189 BECH32_VALID = { 190 '✔️_VER15_PROG40': 'bcrt10qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqxkg7fn', 191 '✔️_VER16_PROG03': 'bcrt1sqqqqq8uhdgr', 192 '✔️_VER16_PROB02': 'bcrt1sqqqq4wstyw', 193 } 194 BECH32_INVALID = { 195 '❌_VER15_PROG41': 'bcrt1sqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqajlxj8', 196 '❌_VER16_PROB01': 'bcrt1sqq5r4036', 197 } 198 for l in BECH32_VALID: 199 ad = BECH32_VALID[l] 200 import_res = wallet_watch_only.importdescriptors([{"desc": descsum_create(f"addr({ad})"), "timestamp": "now", "label": l}]) 201 assert_equal(import_res[0]["success"], True) 202 self.generatetoaddress(node, 1, ad) 203 assert_equal(wallet_watch_only.getaddressesbylabel(label=l), {ad: {'purpose': 'receive'}}) 204 assert_equal(wallet_watch_only.getreceivedbylabel(label=l), 0) 205 for l in BECH32_INVALID: 206 ad = BECH32_INVALID[l] 207 import_res = wallet_watch_only.importdescriptors([{"desc": descsum_create(f"addr({ad})"), "timestamp": "now", "label": l}]) 208 assert_equal(import_res[0]["success"], False) 209 assert_equal(import_res[0]["error"]["code"], -5) 210 assert_equal(import_res[0]["error"]["message"], "Address is not valid") 211 212 213 class Label: 214 def __init__(self, name): 215 # Label name 216 self.name = name 217 # Current receiving address associated with this label. 218 self.receive_address = None 219 # List of all addresses assigned with this label 220 self.addresses = [] 221 # Map of address to address purpose 222 self.purpose = defaultdict(lambda: "receive") 223 224 def add_address(self, address): 225 assert_equal(address not in self.addresses, True) 226 self.addresses.append(address) 227 228 def add_receive_address(self, address): 229 self.add_address(address) 230 231 def verify(self, node): 232 if self.receive_address is not None: 233 assert self.receive_address in self.addresses 234 for address in self.addresses: 235 test_address(node, address, labels=[self.name]) 236 assert self.name in node.listlabels() 237 assert_equal( 238 node.getaddressesbylabel(self.name), 239 {address: {"purpose": self.purpose[address]} for address in self.addresses}) 240 241 def change_label(node, address, old_label, new_label): 242 assert_equal(address in old_label.addresses, True) 243 node.setlabel(address, new_label.name) 244 245 old_label.addresses.remove(address) 246 new_label.add_address(address) 247 248 old_label.verify(node) 249 new_label.verify(node) 250 251 if __name__ == '__main__': 252 WalletLabelsTest(__file__).main()