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