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