test/functional/multi_rpc.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2015-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 multiple RPC users."""
   6
   7 from test_framework.test_framework import BitcoinTestFramework
   8 from test_framework.util import str_to_b64str, assert_equal
   9
  10 import os
  11 import http.client
  12 import urllib.parse
  13
  14 class HTTPBasicsTest (BitcoinTestFramework):
  15
  16     def __init__(self):
  17         super().__init__()
  18         self.setup_clean_chain = False
  19         self.num_nodes = 1
  20
  21     def setup_chain(self):
  22         super().setup_chain()
  23         #Append rpcauth to bitcoin.conf before initialization
  24         rpcauth = "rpcauth=rt:93648e835a54c573682c2eb19f882535$7681e9c5b74bdd85e78166031d2058e1069b3ed7ed967c93fc63abba06f31144"
  25         rpcauth2 = "rpcauth=rt2:f8607b1a88861fac29dfccf9b52ff9f$ff36a0c23c8c62b4846112e50fa888416e94c17bfd4c42f88fd8f55ec6a3137e"
  26         with open(os.path.join(self.options.tmpdir+"/node0", "bitcoin.conf"), 'a', encoding='utf8') as f:
  27             f.write(rpcauth+"\n")
  28             f.write(rpcauth2+"\n")
  29
  30     def setup_network(self):
  31         self.nodes = self.setup_nodes()
  32
  33     def run_test(self):
  34
  35         ##################################################
  36         # Check correctness of the rpcauth config option #
  37         ##################################################
  38         url = urllib.parse.urlparse(self.nodes[0].url)
  39
  40         #Old authpair
  41         authpair = url.username + ':' + url.password
  42
  43         #New authpair generated via share/rpcuser tool
  44         password = "cA773lm788buwYe4g4WT+05pKyNruVKjQ25x3n0DQcM="
  45
  46         #Second authpair with different username
  47         password2 = "8/F3uMDw4KSEbw96U3CA1C4X05dkHDN2BPFjTgZW4KI="
  48         authpairnew = "rt:"+password
  49
  50         headers = {"Authorization": "Basic " + str_to_b64str(authpair)}
  51
  52         conn = http.client.HTTPConnection(url.hostname, url.port)
  53         conn.connect()
  54         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  55         resp = conn.getresponse()
  56         assert_equal(resp.status==401, False)
  57         conn.close()
  58
  59         #Use new authpair to confirm both work
  60         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  61
  62         conn = http.client.HTTPConnection(url.hostname, url.port)
  63         conn.connect()
  64         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  65         resp = conn.getresponse()
  66         assert_equal(resp.status==401, False)
  67         conn.close()
  68
  69         #Wrong login name with rt's password
  70         authpairnew = "rtwrong:"+password
  71         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  72
  73         conn = http.client.HTTPConnection(url.hostname, url.port)
  74         conn.connect()
  75         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  76         resp = conn.getresponse()
  77         assert_equal(resp.status==401, True)
  78         conn.close()
  79
  80         #Wrong password for rt
  81         authpairnew = "rt:"+password+"wrong"
  82         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  83
  84         conn = http.client.HTTPConnection(url.hostname, url.port)
  85         conn.connect()
  86         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  87         resp = conn.getresponse()
  88         assert_equal(resp.status==401, True)
  89         conn.close()
  90
  91         #Correct for rt2
  92         authpairnew = "rt2:"+password2
  93         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  94
  95         conn = http.client.HTTPConnection(url.hostname, url.port)
  96         conn.connect()
  97         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  98         resp = conn.getresponse()
  99         assert_equal(resp.status==401, False)
 100         conn.close()
 101
 102         #Wrong password for rt2
 103         authpairnew = "rt2:"+password2+"wrong"
 104         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
 105
 106         conn = http.client.HTTPConnection(url.hostname, url.port)
 107         conn.connect()
 108         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 109         resp = conn.getresponse()
 110         assert_equal(resp.status==401, True)
 111         conn.close()
 112
 113
 114 if __name__ == '__main__':
 115     HTTPBasicsTest ().main ()