test/functional/importmulti.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2014-2016 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 the importmulti RPC."""
   6 from test_framework.test_framework import BitcoinTestFramework
   7 from test_framework.util import *
   8
   9 class ImportMultiTest (BitcoinTestFramework):
  10     def __init__(self):
  11         super().__init__()
  12         self.num_nodes = 2
  13         self.setup_clean_chain = True
  14
  15     def setup_network(self):
  16         self.setup_nodes()
  17
  18     def run_test (self):
  19         self.log.info("Mining blocks...")
  20         self.nodes[0].generate(1)
  21         self.nodes[1].generate(1)
  22         timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
  23
  24         # keyword definition
  25         PRIV_KEY = 'privkey'
  26         PUB_KEY = 'pubkey'
  27         ADDRESS_KEY = 'address'
  28         SCRIPT_KEY = 'script'
  29
  30
  31         node0_address1 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  32         node0_address2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  33         node0_address3 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  34
  35         #Check only one address
  36         assert_equal(node0_address1['ismine'], True)
  37
  38         #Node 1 sync test
  39         assert_equal(self.nodes[1].getblockcount(),1)
  40
  41         #Address Test - before import
  42         address_info = self.nodes[1].validateaddress(node0_address1['address'])
  43         assert_equal(address_info['iswatchonly'], False)
  44         assert_equal(address_info['ismine'], False)
  45
  46
  47         # RPC importmulti -----------------------------------------------
  48
  49         # Bitcoin Address
  50         self.log.info("Should import an address")
  51         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  52         result = self.nodes[1].importmulti([{
  53             "scriptPubKey": {
  54                 "address": address['address']
  55             },
  56             "timestamp": "now",
  57         }])
  58         assert_equal(result[0]['success'], True)
  59         address_assert = self.nodes[1].validateaddress(address['address'])
  60         assert_equal(address_assert['iswatchonly'], True)
  61         assert_equal(address_assert['ismine'], False)
  62         assert_equal(address_assert['timestamp'], timestamp)
  63         watchonly_address = address['address']
  64         watchonly_timestamp = timestamp
  65
  66         self.log.info("Should not import an invalid address")
  67         result = self.nodes[1].importmulti([{
  68             "scriptPubKey": {
  69                 "address": "not valid address",
  70             },
  71             "timestamp": "now",
  72         }])
  73         assert_equal(result[0]['success'], False)
  74         assert_equal(result[0]['error']['code'], -5)
  75         assert_equal(result[0]['error']['message'], 'Invalid address')
  76
  77         # ScriptPubKey + internal
  78         self.log.info("Should import a scriptPubKey with internal flag")
  79         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  80         result = self.nodes[1].importmulti([{
  81             "scriptPubKey": address['scriptPubKey'],
  82             "timestamp": "now",
  83             "internal": True
  84         }])
  85         assert_equal(result[0]['success'], True)
  86         address_assert = self.nodes[1].validateaddress(address['address'])
  87         assert_equal(address_assert['iswatchonly'], True)
  88         assert_equal(address_assert['ismine'], False)
  89         assert_equal(address_assert['timestamp'], timestamp)
  90
  91         # ScriptPubKey + !internal
  92         self.log.info("Should not import a scriptPubKey without internal flag")
  93         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
  94         result = self.nodes[1].importmulti([{
  95             "scriptPubKey": address['scriptPubKey'],
  96             "timestamp": "now",
  97         }])
  98         assert_equal(result[0]['success'], False)
  99         assert_equal(result[0]['error']['code'], -8)
 100         assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
 101         address_assert = self.nodes[1].validateaddress(address['address'])
 102         assert_equal(address_assert['iswatchonly'], False)
 103         assert_equal(address_assert['ismine'], False)
 104         assert_equal('timestamp' in address_assert, False)
 105
 106
 107         # Address + Public key + !Internal
 108         self.log.info("Should import an address with public key")
 109         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 110         result = self.nodes[1].importmulti([{
 111             "scriptPubKey": {
 112                 "address": address['address']
 113             },
 114             "timestamp": "now",
 115             "pubkeys": [ address['pubkey'] ]
 116         }])
 117         assert_equal(result[0]['success'], True)
 118         address_assert = self.nodes[1].validateaddress(address['address'])
 119         assert_equal(address_assert['iswatchonly'], True)
 120         assert_equal(address_assert['ismine'], False)
 121         assert_equal(address_assert['timestamp'], timestamp)
 122
 123
 124         # ScriptPubKey + Public key + internal
 125         self.log.info("Should import a scriptPubKey with internal and with public key")
 126         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 127         request = [{
 128             "scriptPubKey": address['scriptPubKey'],
 129             "timestamp": "now",
 130             "pubkeys": [ address['pubkey'] ],
 131             "internal": True
 132         }]
 133         result = self.nodes[1].importmulti(request)
 134         assert_equal(result[0]['success'], True)
 135         address_assert = self.nodes[1].validateaddress(address['address'])
 136         assert_equal(address_assert['iswatchonly'], True)
 137         assert_equal(address_assert['ismine'], False)
 138         assert_equal(address_assert['timestamp'], timestamp)
 139
 140         # ScriptPubKey + Public key + !internal
 141         self.log.info("Should not import a scriptPubKey without internal and with public key")
 142         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 143         request = [{
 144             "scriptPubKey": address['scriptPubKey'],
 145             "timestamp": "now",
 146             "pubkeys": [ address['pubkey'] ]
 147         }]
 148         result = self.nodes[1].importmulti(request)
 149         assert_equal(result[0]['success'], False)
 150         assert_equal(result[0]['error']['code'], -8)
 151         assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
 152         address_assert = self.nodes[1].validateaddress(address['address'])
 153         assert_equal(address_assert['iswatchonly'], False)
 154         assert_equal(address_assert['ismine'], False)
 155         assert_equal('timestamp' in address_assert, False)
 156
 157         # Address + Private key + !watchonly
 158         self.log.info("Should import an address with private key")
 159         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 160         result = self.nodes[1].importmulti([{
 161             "scriptPubKey": {
 162                 "address": address['address']
 163             },
 164             "timestamp": "now",
 165             "keys": [ self.nodes[0].dumpprivkey(address['address']) ]
 166         }])
 167         assert_equal(result[0]['success'], True)
 168         address_assert = self.nodes[1].validateaddress(address['address'])
 169         assert_equal(address_assert['iswatchonly'], False)
 170         assert_equal(address_assert['ismine'], True)
 171         assert_equal(address_assert['timestamp'], timestamp)
 172
 173         # Address + Private key + watchonly
 174         self.log.info("Should not import an address with private key and with watchonly")
 175         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 176         result = self.nodes[1].importmulti([{
 177             "scriptPubKey": {
 178                 "address": address['address']
 179             },
 180             "timestamp": "now",
 181             "keys": [ self.nodes[0].dumpprivkey(address['address']) ],
 182             "watchonly": True
 183         }])
 184         assert_equal(result[0]['success'], False)
 185         assert_equal(result[0]['error']['code'], -8)
 186         assert_equal(result[0]['error']['message'], 'Incompatibility found between watchonly and keys')
 187         address_assert = self.nodes[1].validateaddress(address['address'])
 188         assert_equal(address_assert['iswatchonly'], False)
 189         assert_equal(address_assert['ismine'], False)
 190         assert_equal('timestamp' in address_assert, False)
 191
 192         # ScriptPubKey + Private key + internal
 193         self.log.info("Should import a scriptPubKey with internal and with private key")
 194         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 195         result = self.nodes[1].importmulti([{
 196             "scriptPubKey": address['scriptPubKey'],
 197             "timestamp": "now",
 198             "keys": [ self.nodes[0].dumpprivkey(address['address']) ],
 199             "internal": True
 200         }])
 201         assert_equal(result[0]['success'], True)
 202         address_assert = self.nodes[1].validateaddress(address['address'])
 203         assert_equal(address_assert['iswatchonly'], False)
 204         assert_equal(address_assert['ismine'], True)
 205         assert_equal(address_assert['timestamp'], timestamp)
 206
 207         # ScriptPubKey + Private key + !internal
 208         self.log.info("Should not import a scriptPubKey without internal and with private key")
 209         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 210         result = self.nodes[1].importmulti([{
 211             "scriptPubKey": address['scriptPubKey'],
 212             "timestamp": "now",
 213             "keys": [ self.nodes[0].dumpprivkey(address['address']) ]
 214         }])
 215         assert_equal(result[0]['success'], False)
 216         assert_equal(result[0]['error']['code'], -8)
 217         assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
 218         address_assert = self.nodes[1].validateaddress(address['address'])
 219         assert_equal(address_assert['iswatchonly'], False)
 220         assert_equal(address_assert['ismine'], False)
 221         assert_equal('timestamp' in address_assert, False)
 222
 223
 224         # P2SH address
 225         sig_address_1 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 226         sig_address_2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 227         sig_address_3 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 228         multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['address'], sig_address_2['address'], sig_address_3['pubkey']])
 229         self.nodes[1].generate(100)
 230         transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
 231         self.nodes[1].generate(1)
 232         timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
 233         transaction = self.nodes[1].gettransaction(transactionid)
 234
 235         self.log.info("Should import a p2sh")
 236         result = self.nodes[1].importmulti([{
 237             "scriptPubKey": {
 238                 "address": multi_sig_script['address']
 239             },
 240             "timestamp": "now",
 241         }])
 242         assert_equal(result[0]['success'], True)
 243         address_assert = self.nodes[1].validateaddress(multi_sig_script['address'])
 244         assert_equal(address_assert['isscript'], True)
 245         assert_equal(address_assert['iswatchonly'], True)
 246         assert_equal(address_assert['timestamp'], timestamp)
 247         p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
 248         assert_equal(p2shunspent['spendable'], False)
 249         assert_equal(p2shunspent['solvable'], False)
 250
 251
 252         # P2SH + Redeem script
 253         sig_address_1 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 254         sig_address_2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 255         sig_address_3 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 256         multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['address'], sig_address_2['address'], sig_address_3['pubkey']])
 257         self.nodes[1].generate(100)
 258         transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
 259         self.nodes[1].generate(1)
 260         timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
 261         transaction = self.nodes[1].gettransaction(transactionid)
 262
 263         self.log.info("Should import a p2sh with respective redeem script")
 264         result = self.nodes[1].importmulti([{
 265             "scriptPubKey": {
 266                 "address": multi_sig_script['address']
 267             },
 268             "timestamp": "now",
 269             "redeemscript": multi_sig_script['redeemScript']
 270         }])
 271         assert_equal(result[0]['success'], True)
 272         address_assert = self.nodes[1].validateaddress(multi_sig_script['address'])
 273         assert_equal(address_assert['timestamp'], timestamp)
 274
 275         p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
 276         assert_equal(p2shunspent['spendable'], False)
 277         assert_equal(p2shunspent['solvable'], True)
 278
 279
 280         # P2SH + Redeem script + Private Keys + !Watchonly
 281         sig_address_1 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 282         sig_address_2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 283         sig_address_3 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 284         multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['address'], sig_address_2['address'], sig_address_3['pubkey']])
 285         self.nodes[1].generate(100)
 286         transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
 287         self.nodes[1].generate(1)
 288         timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
 289         transaction = self.nodes[1].gettransaction(transactionid)
 290
 291         self.log.info("Should import a p2sh with respective redeem script and private keys")
 292         result = self.nodes[1].importmulti([{
 293             "scriptPubKey": {
 294                 "address": multi_sig_script['address']
 295             },
 296             "timestamp": "now",
 297             "redeemscript": multi_sig_script['redeemScript'],
 298             "keys": [ self.nodes[0].dumpprivkey(sig_address_1['address']), self.nodes[0].dumpprivkey(sig_address_2['address'])]
 299         }])
 300         assert_equal(result[0]['success'], True)
 301         address_assert = self.nodes[1].validateaddress(multi_sig_script['address'])
 302         assert_equal(address_assert['timestamp'], timestamp)
 303
 304         p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
 305         assert_equal(p2shunspent['spendable'], False)
 306         assert_equal(p2shunspent['solvable'], True)
 307
 308         # P2SH + Redeem script + Private Keys + Watchonly
 309         sig_address_1 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 310         sig_address_2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 311         sig_address_3 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 312         multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['address'], sig_address_2['address'], sig_address_3['pubkey']])
 313         self.nodes[1].generate(100)
 314         transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
 315         self.nodes[1].generate(1)
 316         timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
 317         transaction = self.nodes[1].gettransaction(transactionid)
 318
 319         self.log.info("Should import a p2sh with respective redeem script and private keys")
 320         result = self.nodes[1].importmulti([{
 321             "scriptPubKey": {
 322                 "address": multi_sig_script['address']
 323             },
 324             "timestamp": "now",
 325             "redeemscript": multi_sig_script['redeemScript'],
 326             "keys": [ self.nodes[0].dumpprivkey(sig_address_1['address']), self.nodes[0].dumpprivkey(sig_address_2['address'])],
 327             "watchonly": True
 328         }])
 329         assert_equal(result[0]['success'], False)
 330         assert_equal(result[0]['error']['code'], -8)
 331         assert_equal(result[0]['error']['message'], 'Incompatibility found between watchonly and keys')
 332
 333
 334         # Address + Public key + !Internal + Wrong pubkey
 335         self.log.info("Should not import an address with a wrong public key")
 336         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 337         address2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 338         result = self.nodes[1].importmulti([{
 339             "scriptPubKey": {
 340                 "address": address['address']
 341             },
 342             "timestamp": "now",
 343             "pubkeys": [ address2['pubkey'] ]
 344         }])
 345         assert_equal(result[0]['success'], False)
 346         assert_equal(result[0]['error']['code'], -5)
 347         assert_equal(result[0]['error']['message'], 'Consistency check failed')
 348         address_assert = self.nodes[1].validateaddress(address['address'])
 349         assert_equal(address_assert['iswatchonly'], False)
 350         assert_equal(address_assert['ismine'], False)
 351         assert_equal('timestamp' in address_assert, False)
 352
 353
 354         # ScriptPubKey + Public key + internal + Wrong pubkey
 355         self.log.info("Should not import a scriptPubKey with internal and with a wrong public key")
 356         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 357         address2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 358         request = [{
 359             "scriptPubKey": address['scriptPubKey'],
 360             "timestamp": "now",
 361             "pubkeys": [ address2['pubkey'] ],
 362             "internal": True
 363         }]
 364         result = self.nodes[1].importmulti(request)
 365         assert_equal(result[0]['success'], False)
 366         assert_equal(result[0]['error']['code'], -5)
 367         assert_equal(result[0]['error']['message'], 'Consistency check failed')
 368         address_assert = self.nodes[1].validateaddress(address['address'])
 369         assert_equal(address_assert['iswatchonly'], False)
 370         assert_equal(address_assert['ismine'], False)
 371         assert_equal('timestamp' in address_assert, False)
 372
 373
 374         # Address + Private key + !watchonly + Wrong private key
 375         self.log.info("Should not import an address with a wrong private key")
 376         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 377         address2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 378         result = self.nodes[1].importmulti([{
 379             "scriptPubKey": {
 380                 "address": address['address']
 381             },
 382             "timestamp": "now",
 383             "keys": [ self.nodes[0].dumpprivkey(address2['address']) ]
 384         }])
 385         assert_equal(result[0]['success'], False)
 386         assert_equal(result[0]['error']['code'], -5)
 387         assert_equal(result[0]['error']['message'], 'Consistency check failed')
 388         address_assert = self.nodes[1].validateaddress(address['address'])
 389         assert_equal(address_assert['iswatchonly'], False)
 390         assert_equal(address_assert['ismine'], False)
 391         assert_equal('timestamp' in address_assert, False)
 392
 393
 394         # ScriptPubKey + Private key + internal + Wrong private key
 395         self.log.info("Should not import a scriptPubKey with internal and with a wrong private key")
 396         address = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 397         address2 = self.nodes[0].validateaddress(self.nodes[0].getnewaddress())
 398         result = self.nodes[1].importmulti([{
 399             "scriptPubKey": address['scriptPubKey'],
 400             "timestamp": "now",
 401             "keys": [ self.nodes[0].dumpprivkey(address2['address']) ],
 402             "internal": True
 403         }])
 404         assert_equal(result[0]['success'], False)
 405         assert_equal(result[0]['error']['code'], -5)
 406         assert_equal(result[0]['error']['message'], 'Consistency check failed')
 407         address_assert = self.nodes[1].validateaddress(address['address'])
 408         assert_equal(address_assert['iswatchonly'], False)
 409         assert_equal(address_assert['ismine'], False)
 410         assert_equal('timestamp' in address_assert, False)
 411
 412
 413         # Importing existing watch only address with new timestamp should replace saved timestamp.
 414         assert_greater_than(timestamp, watchonly_timestamp)
 415         self.log.info("Should replace previously saved watch only timestamp.")
 416         result = self.nodes[1].importmulti([{
 417             "scriptPubKey": {
 418                 "address": watchonly_address,
 419             },
 420             "timestamp": "now",
 421         }])
 422         assert_equal(result[0]['success'], True)
 423         address_assert = self.nodes[1].validateaddress(watchonly_address)
 424         assert_equal(address_assert['iswatchonly'], True)
 425         assert_equal(address_assert['ismine'], False)
 426         assert_equal(address_assert['timestamp'], timestamp)
 427         watchonly_timestamp = timestamp
 428
 429
 430         # restart nodes to check for proper serialization/deserialization of watch only address
 431         self.stop_nodes()
 432         self.nodes = self.start_nodes(2, self.options.tmpdir)
 433         address_assert = self.nodes[1].validateaddress(watchonly_address)
 434         assert_equal(address_assert['iswatchonly'], True)
 435         assert_equal(address_assert['ismine'], False)
 436         assert_equal(address_assert['timestamp'], watchonly_timestamp)
 437
 438         # Bad or missing timestamps
 439         self.log.info("Should throw on invalid or missing timestamp values")
 440         assert_raises_message(JSONRPCException, 'Missing required timestamp field for key',
 441             self.nodes[1].importmulti, [{
 442                 "scriptPubKey": address['scriptPubKey'],
 443             }])
 444         assert_raises_message(JSONRPCException, 'Expected number or "now" timestamp value for key. got type string',
 445             self.nodes[1].importmulti, [{
 446                 "scriptPubKey": address['scriptPubKey'],
 447                 "timestamp": "",
 448             }])
 449
 450
 451 if __name__ == '__main__':
 452     ImportMultiTest ().main ()