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     def set_test_params(self):
  16         self.num_nodes = 2
  17
  18     def setup_chain(self):
  19         super().setup_chain()
  20         #Append rpcauth to bitcoin.conf before initialization
  21         rpcauth = "rpcauth=rt:93648e835a54c573682c2eb19f882535$7681e9c5b74bdd85e78166031d2058e1069b3ed7ed967c93fc63abba06f31144"
  22         rpcauth2 = "rpcauth=rt2:f8607b1a88861fac29dfccf9b52ff9f$ff36a0c23c8c62b4846112e50fa888416e94c17bfd4c42f88fd8f55ec6a3137e"
  23         rpcuser = "rpcuser=rpcuser💻"
  24         rpcpassword = "rpcpassword=rpcpassword🔑"
  25         with open(os.path.join(self.options.tmpdir+"/node0", "bitcoin.conf"), 'a', encoding='utf8') as f:
  26             f.write(rpcauth+"\n")
  27             f.write(rpcauth2+"\n")
  28         with open(os.path.join(self.options.tmpdir+"/node1", "bitcoin.conf"), 'a', encoding='utf8') as f:
  29             f.write(rpcuser+"\n")
  30             f.write(rpcpassword+"\n")
  31
  32     def run_test(self):
  33
  34         ##################################################
  35         # Check correctness of the rpcauth config option #
  36         ##################################################
  37         url = urllib.parse.urlparse(self.nodes[0].url)
  38
  39         #Old authpair
  40         authpair = url.username + ':' + url.password
  41
  42         #New authpair generated via share/rpcuser tool
  43         password = "cA773lm788buwYe4g4WT+05pKyNruVKjQ25x3n0DQcM="
  44
  45         #Second authpair with different username
  46         password2 = "8/F3uMDw4KSEbw96U3CA1C4X05dkHDN2BPFjTgZW4KI="
  47         authpairnew = "rt:"+password
  48
  49         headers = {"Authorization": "Basic " + str_to_b64str(authpair)}
  50
  51         conn = http.client.HTTPConnection(url.hostname, url.port)
  52         conn.connect()
  53         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  54         resp = conn.getresponse()
  55         assert_equal(resp.status, 200)
  56         conn.close()
  57
  58         #Use new authpair to confirm both work
  59         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  60
  61         conn = http.client.HTTPConnection(url.hostname, url.port)
  62         conn.connect()
  63         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  64         resp = conn.getresponse()
  65         assert_equal(resp.status, 200)
  66         conn.close()
  67
  68         #Wrong login name with rt's password
  69         authpairnew = "rtwrong:"+password
  70         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  71
  72         conn = http.client.HTTPConnection(url.hostname, url.port)
  73         conn.connect()
  74         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  75         resp = conn.getresponse()
  76         assert_equal(resp.status, 401)
  77         conn.close()
  78
  79         #Wrong password for rt
  80         authpairnew = "rt:"+password+"wrong"
  81         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  82
  83         conn = http.client.HTTPConnection(url.hostname, url.port)
  84         conn.connect()
  85         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  86         resp = conn.getresponse()
  87         assert_equal(resp.status, 401)
  88         conn.close()
  89
  90         #Correct for rt2
  91         authpairnew = "rt2:"+password2
  92         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
  93
  94         conn = http.client.HTTPConnection(url.hostname, url.port)
  95         conn.connect()
  96         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
  97         resp = conn.getresponse()
  98         assert_equal(resp.status, 200)
  99         conn.close()
 100
 101         #Wrong password for rt2
 102         authpairnew = "rt2:"+password2+"wrong"
 103         headers = {"Authorization": "Basic " + str_to_b64str(authpairnew)}
 104
 105         conn = http.client.HTTPConnection(url.hostname, url.port)
 106         conn.connect()
 107         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 108         resp = conn.getresponse()
 109         assert_equal(resp.status, 401)
 110         conn.close()
 111
 112         ###############################################################
 113         # Check correctness of the rpcuser/rpcpassword config options #
 114         ###############################################################
 115         url = urllib.parse.urlparse(self.nodes[1].url)
 116
 117         # rpcuser and rpcpassword authpair
 118         rpcuserauthpair = "rpcuser💻:rpcpassword🔑"
 119
 120         headers = {"Authorization": "Basic " + str_to_b64str(rpcuserauthpair)}
 121
 122         conn = http.client.HTTPConnection(url.hostname, url.port)
 123         conn.connect()
 124         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 125         resp = conn.getresponse()
 126         assert_equal(resp.status, 200)
 127         conn.close()
 128
 129         #Wrong login name with rpcuser's password
 130         rpcuserauthpair = "rpcuserwrong:rpcpassword"
 131         headers = {"Authorization": "Basic " + str_to_b64str(rpcuserauthpair)}
 132
 133         conn = http.client.HTTPConnection(url.hostname, url.port)
 134         conn.connect()
 135         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 136         resp = conn.getresponse()
 137         assert_equal(resp.status, 401)
 138         conn.close()
 139
 140         #Wrong password for rpcuser
 141         rpcuserauthpair = "rpcuser:rpcpasswordwrong"
 142         headers = {"Authorization": "Basic " + str_to_b64str(rpcuserauthpair)}
 143
 144         conn = http.client.HTTPConnection(url.hostname, url.port)
 145         conn.connect()
 146         conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 147         resp = conn.getresponse()
 148         assert_equal(resp.status, 401)
 149         conn.close()
 150
 151
 152 if __name__ == '__main__':
 153     HTTPBasicsTest ().main ()