aboutsummaryrefslogtreecommitdiff
path: root/qa/rpc-tests
diff options
context:
space:
mode:
Diffstat (limited to 'qa/rpc-tests')
-rwxr-xr-xqa/rpc-tests/rpcbind_test.py219
-rw-r--r--qa/rpc-tests/test_framework/util.py10
-rwxr-xr-xqa/rpc-tests/wallet-dump.py120
3 files changed, 221 insertions, 128 deletions
diff --git a/qa/rpc-tests/rpcbind_test.py b/qa/rpc-tests/rpcbind_test.py
index 572273566b..bf1cc87126 100755
--- a/qa/rpc-tests/rpcbind_test.py
+++ b/qa/rpc-tests/rpcbind_test.py
@@ -5,143 +5,108 @@
# Test for -rpcbind, as well as -rpcallowip and -rpcconnect
-# TODO extend this test from the test framework (like all other tests)
-
import tempfile
import traceback
+from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import *
from test_framework.netutil import *
-def run_bind_test(tmpdir, allow_ips, connect_to, addresses, expected):
- '''
- Start a node with requested rpcallowip and rpcbind parameters,
- then try to connect, and check if the set of bound addresses
- matches the expected set.
- '''
- expected = [(addr_to_hex(addr), port) for (addr, port) in expected]
- base_args = ['-disablewallet', '-nolisten']
- if allow_ips:
- base_args += ['-rpcallowip=' + x for x in allow_ips]
- binds = ['-rpcbind='+addr for addr in addresses]
- nodes = start_nodes(self.num_nodes, tmpdir, [base_args + binds], connect_to)
- try:
- pid = bitcoind_processes[0].pid
- assert_equal(set(get_bind_addrs(pid)), set(expected))
- finally:
- stop_nodes(nodes)
- wait_bitcoinds()
-
-def run_allowip_test(tmpdir, allow_ips, rpchost, rpcport):
- '''
- Start a node with rpcwallow IP, and request getinfo
- at a non-localhost IP.
- '''
- base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips]
- nodes = start_nodes(self.num_nodes, tmpdir, [base_args])
- try:
- # connect to node through non-loopback interface
- url = "http://rt:rt@%s:%d" % (rpchost, rpcport,)
- node = get_rpc_proxy(url, 1)
- node.getinfo()
- finally:
- node = None # make sure connection will be garbage collected and closed
- stop_nodes(nodes)
- wait_bitcoinds()
-
-
-def run_test(tmpdir):
- assert(sys.platform.startswith('linux')) # due to OS-specific network stats queries, this test works only on Linux
- # find the first non-loopback interface for testing
- non_loopback_ip = None
- for name,ip in all_interfaces():
- if ip != '127.0.0.1':
- non_loopback_ip = ip
- break
- if non_loopback_ip is None:
- assert(not 'This test requires at least one non-loopback IPv4 interface')
- print("Using interface %s for testing" % non_loopback_ip)
+class RPCBindTest(BitcoinTestFramework):
- defaultport = rpc_port(0)
+ def __init__(self):
+ super().__init__()
+ self.setup_clean_chain = True
+ self.num_nodes = 1
- # check default without rpcallowip (IPv4 and IPv6 localhost)
- run_bind_test(tmpdir, None, '127.0.0.1', [],
- [('127.0.0.1', defaultport), ('::1', defaultport)])
- # check default with rpcallowip (IPv6 any)
- run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', [],
- [('::0', defaultport)])
- # check only IPv4 localhost (explicit)
- run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
- [('127.0.0.1', defaultport)])
- # check only IPv4 localhost (explicit) with alternative port
- run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
- [('127.0.0.1', 32171)])
- # check only IPv4 localhost (explicit) with multiple alternative ports on same host
- run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
- [('127.0.0.1', 32171), ('127.0.0.1', 32172)])
- # check only IPv6 localhost (explicit)
- run_bind_test(tmpdir, ['[::1]'], '[::1]', ['[::1]'],
- [('::1', defaultport)])
- # check both IPv4 and IPv6 localhost (explicit)
- run_bind_test(tmpdir, ['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
- [('127.0.0.1', defaultport), ('::1', defaultport)])
- # check only non-loopback interface
- run_bind_test(tmpdir, [non_loopback_ip], non_loopback_ip, [non_loopback_ip],
- [(non_loopback_ip, defaultport)])
-
- # Check that with invalid rpcallowip, we are denied
- run_allowip_test(tmpdir, [non_loopback_ip], non_loopback_ip, defaultport)
- try:
- run_allowip_test(tmpdir, ['1.1.1.1'], non_loopback_ip, defaultport)
- assert(not 'Connection not denied by rpcallowip as expected')
- except ValueError:
+ def setup_network(self):
pass
-def main():
- import optparse
-
- parser = optparse.OptionParser(usage="%prog [options]")
- parser.add_option("--nocleanup", dest="nocleanup", default=False, action="store_true",
- help="Leave bitcoinds and test.* datadir on exit or error")
- parser.add_option("--srcdir", dest="srcdir", default="../../src",
- help="Source directory containing bitcoind/bitcoin-cli (default: %default%)")
- parser.add_option("--tmpdir", dest="tmpdir", default=tempfile.mkdtemp(prefix="test"),
- help="Root directory for datadirs")
- (options, args) = parser.parse_args()
-
- os.environ['PATH'] = options.srcdir+":"+os.environ['PATH']
-
- check_json_precision()
-
- success = False
- nodes = []
- try:
- print("Initializing test directory "+options.tmpdir)
- if not os.path.isdir(options.tmpdir):
- os.makedirs(options.tmpdir)
- initialize_chain(options.tmpdir)
-
- run_test(options.tmpdir)
-
- success = True
-
- except AssertionError as e:
- print("Assertion failed: "+e.message)
- except Exception as e:
- print("Unexpected exception caught during testing: "+str(e))
- traceback.print_tb(sys.exc_info()[2])
-
- if not options.nocleanup:
- print("Cleaning up")
- wait_bitcoinds()
- shutil.rmtree(options.tmpdir)
+ def setup_nodes(self):
+ pass
- if success:
- print("Tests successful")
- sys.exit(0)
- else:
- print("Failed")
- sys.exit(1)
+ def run_bind_test(self, allow_ips, connect_to, addresses, expected):
+ '''
+ Start a node with requested rpcallowip and rpcbind parameters,
+ then try to connect, and check if the set of bound addresses
+ matches the expected set.
+ '''
+ expected = [(addr_to_hex(addr), port) for (addr, port) in expected]
+ base_args = ['-disablewallet', '-nolisten']
+ if allow_ips:
+ base_args += ['-rpcallowip=' + x for x in allow_ips]
+ binds = ['-rpcbind='+addr for addr in addresses]
+ self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args + binds], connect_to)
+ try:
+ pid = bitcoind_processes[0].pid
+ assert_equal(set(get_bind_addrs(pid)), set(expected))
+ finally:
+ stop_nodes(self.nodes)
+ wait_bitcoinds()
+
+ def run_allowip_test(self, allow_ips, rpchost, rpcport):
+ '''
+ Start a node with rpcwallow IP, and request getinfo
+ at a non-localhost IP.
+ '''
+ base_args = ['-disablewallet', '-nolisten'] + ['-rpcallowip='+x for x in allow_ips]
+ self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, [base_args])
+ try:
+ # connect to node through non-loopback interface
+ node = get_rpc_proxy(rpc_url(0, "%s:%d" % (rpchost, rpcport)), 0)
+ node.getinfo()
+ finally:
+ node = None # make sure connection will be garbage collected and closed
+ stop_nodes(self.nodes)
+ wait_bitcoinds()
+
+ def run_test(self):
+ # due to OS-specific network stats queries, this test works only on Linux
+ assert(sys.platform.startswith('linux'))
+ # find the first non-loopback interface for testing
+ non_loopback_ip = None
+ for name,ip in all_interfaces():
+ if ip != '127.0.0.1':
+ non_loopback_ip = ip
+ break
+ if non_loopback_ip is None:
+ assert(not 'This test requires at least one non-loopback IPv4 interface')
+ print("Using interface %s for testing" % non_loopback_ip)
+
+ defaultport = rpc_port(0)
+
+ # check default without rpcallowip (IPv4 and IPv6 localhost)
+ self.run_bind_test(None, '127.0.0.1', [],
+ [('127.0.0.1', defaultport), ('::1', defaultport)])
+ # check default with rpcallowip (IPv6 any)
+ self.run_bind_test(['127.0.0.1'], '127.0.0.1', [],
+ [('::0', defaultport)])
+ # check only IPv4 localhost (explicit)
+ self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1'],
+ [('127.0.0.1', defaultport)])
+ # check only IPv4 localhost (explicit) with alternative port
+ self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171'],
+ [('127.0.0.1', 32171)])
+ # check only IPv4 localhost (explicit) with multiple alternative ports on same host
+ self.run_bind_test(['127.0.0.1'], '127.0.0.1:32171', ['127.0.0.1:32171', '127.0.0.1:32172'],
+ [('127.0.0.1', 32171), ('127.0.0.1', 32172)])
+ # check only IPv6 localhost (explicit)
+ self.run_bind_test(['[::1]'], '[::1]', ['[::1]'],
+ [('::1', defaultport)])
+ # check both IPv4 and IPv6 localhost (explicit)
+ self.run_bind_test(['127.0.0.1'], '127.0.0.1', ['127.0.0.1', '[::1]'],
+ [('127.0.0.1', defaultport), ('::1', defaultport)])
+ # check only non-loopback interface
+ self.run_bind_test([non_loopback_ip], non_loopback_ip, [non_loopback_ip],
+ [(non_loopback_ip, defaultport)])
+
+ # Check that with invalid rpcallowip, we are denied
+ self.run_allowip_test([non_loopback_ip], non_loopback_ip, defaultport)
+ try:
+ self.run_allowip_test(['1.1.1.1'], non_loopback_ip, defaultport)
+ assert(not 'Connection not denied by rpcallowip as expected')
+ except JSONRPCException:
+ pass
if __name__ == '__main__':
- main()
+ RPCBindTest ().main ()
diff --git a/qa/rpc-tests/test_framework/util.py b/qa/rpc-tests/test_framework/util.py
index 32fe79efc3..8aa34265c5 100644
--- a/qa/rpc-tests/test_framework/util.py
+++ b/qa/rpc-tests/test_framework/util.py
@@ -171,7 +171,15 @@ def rpc_auth_pair(n):
def rpc_url(i, rpchost=None):
rpc_u, rpc_p = rpc_auth_pair(i)
- return "http://%s:%s@%s:%d" % (rpc_u, rpc_p, rpchost or '127.0.0.1', rpc_port(i))
+ host = '127.0.0.1'
+ port = rpc_port(i)
+ if rpchost:
+ parts = rpchost.split(':')
+ if len(parts) == 2:
+ host, port = parts
+ else:
+ host = rpchost
+ return "http://%s:%s@%s:%d" % (rpc_u, rpc_p, host, int(port))
def wait_for_bitcoind_start(process, url, i):
'''
diff --git a/qa/rpc-tests/wallet-dump.py b/qa/rpc-tests/wallet-dump.py
new file mode 100755
index 0000000000..dd675f57fc
--- /dev/null
+++ b/qa/rpc-tests/wallet-dump.py
@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+# Copyright (c) 2016 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import *
+import os
+import shutil
+
+
+class WalletDumpTest(BitcoinTestFramework):
+
+ def __init__(self):
+ super().__init__()
+ self.setup_clean_chain = False
+ self.num_nodes = 1
+
+ def setup_network(self, split=False):
+ extra_args = [["-keypool=100"]]
+ self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, extra_args)
+
+ def run_test (self):
+ tmpdir = self.options.tmpdir
+
+ #generate 20 addresses to compare against the dump
+ test_addr_count = 20
+ addrs = []
+ for i in range(0,test_addr_count):
+ addr = self.nodes[0].getnewaddress()
+ vaddr= self.nodes[0].validateaddress(addr) #required to get hd keypath
+ addrs.append(vaddr)
+
+ # dump unencrypted wallet
+ self.nodes[0].dumpwallet(tmpdir + "/node0/wallet.unencrypted.dump")
+
+ #open file
+ inputfile = open(tmpdir + "/node0/wallet.unencrypted.dump")
+ found_addr = 0
+ found_addr_chg = 0
+ found_addr_rsv = 0
+ hdmasteraddr = ""
+ for line in inputfile:
+ #only read non comment lines
+ if line[0] != "#" and len(line) > 10:
+ #split out some data
+ keyLabel, comment = line.split("#")
+ key = keyLabel.split(" ")[0]
+ keytype = keyLabel.split(" ")[2]
+ if len(comment) > 1:
+ addrKeypath = comment.split(" addr=")[1]
+ addr = addrKeypath.split(" ")[0]
+ keypath = ""
+ if keytype != "hdmaster=1":
+ keypath = addrKeypath.rstrip().split("hdkeypath=")[1]
+ else:
+ #keep hd master for later comp.
+ hdmasteraddr = addr
+
+ #count key types
+ for addrObj in addrs:
+ if (addrObj['address'] == addr and addrObj['hdkeypath'] == keypath and keytype == "label="):
+ found_addr+=1
+ break
+ elif (keytype == "change=1"):
+ found_addr_chg+=1
+ break
+ elif (keytype == "reserve=1"):
+ found_addr_rsv+=1
+ break
+ assert(found_addr == test_addr_count) #all keys must be in the dump
+ assert(found_addr_chg == 50) #50 blocks where mined
+ assert(found_addr_rsv == 100) #100 reserve keys (keypool)
+
+ #encrypt wallet, restart, unlock and dump
+ self.nodes[0].encryptwallet('test')
+ bitcoind_processes[0].wait()
+ self.nodes[0] = start_node(0, self.options.tmpdir)
+ self.nodes[0].walletpassphrase('test', 10)
+ self.nodes[0].dumpwallet(tmpdir + "/node0/wallet.encrypted.dump")
+
+ #open dump done with an encrypted wallet
+ inputfile = open(tmpdir + "/node0/wallet.encrypted.dump")
+ found_addr = 0
+ found_addr_chg = 0
+ found_addr_rsv = 0
+ for line in inputfile:
+ if line[0] != "#" and len(line) > 10:
+ keyLabel, comment = line.split("#")
+ key = keyLabel.split(" ")[0]
+ keytype = keyLabel.split(" ")[2]
+ if len(comment) > 1:
+ addrKeypath = comment.split(" addr=")[1]
+ addr = addrKeypath.split(" ")[0]
+ keypath = ""
+ if keytype != "hdmaster=1":
+ keypath = addrKeypath.rstrip().split("hdkeypath=")[1]
+ else:
+ #ensure we have generated a new hd master key
+ assert(hdmasteraddr != addr)
+ if keytype == "inactivehdmaster=1":
+ #ensure the old master is still available
+ assert(hdmasteraddr == addr)
+ for addrObj in addrs:
+ if (addrObj['address'] == addr and addrObj['hdkeypath'] == keypath and keytype == "label="):
+ found_addr+=1
+ break
+ elif (keytype == "change=1"):
+ found_addr_chg+=1
+ break
+ elif (keytype == "reserve=1"):
+ found_addr_rsv+=1
+ break
+
+ assert(found_addr == test_addr_count)
+ assert(found_addr_chg == 150) #old reserve keys are marked as change now
+ assert(found_addr_rsv == 100) #keypool size
+
+if __name__ == '__main__':
+ WalletDumpTest().main ()
generated by cgit v1.2.3 (git 2.26.2) at 2024-11-09 01:36:20 +0000