1  #!/usr/bin/env python3
  2  # Copyright (c) 2015-2022 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 (
  9      assert_equal,
 10      str_to_b64str,
 11  )
 12  
 13  import http.client
 14  import urllib.parse
 15  import subprocess
 16  from random import SystemRandom
 17  import string
 18  import configparser
 19  import sys
 20  
 21  
 22  def call_with_auth(node, user, password):
 23      url = urllib.parse.urlparse(node.url)
 24      headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(user, password))}
 25  
 26      conn = http.client.HTTPConnection(url.hostname, url.port)
 27      conn.connect()
 28      conn.request('POST', '/', '{"method": "getbestblockhash"}', headers)
 29      resp = conn.getresponse()
 30      conn.close()
 31      return resp
 32  
 33  
 34  class HTTPBasicsTest(BitcoinTestFramework):
 35      def set_test_params(self):
 36          self.num_nodes = 2
 37          self.supports_cli = False
 38  
 39      def conf_setup(self):
 40          #Append rpcauth to bitcoin.conf before initialization
 41          self.rtpassword = "cA773lm788buwYe4g4WT+05pKyNruVKjQ25x3n0DQcM="
 42          rpcauth = "rpcauth=rt:93648e835a54c573682c2eb19f882535$7681e9c5b74bdd85e78166031d2058e1069b3ed7ed967c93fc63abba06f31144"
 43  
 44          self.rpcuser = "rpcuser💻"
 45          self.rpcpassword = "rpcpassword🔑"
 46  
 47          config = configparser.ConfigParser()
 48          config.read_file(open(self.options.configfile))
 49          gen_rpcauth = config['environment']['RPCAUTH']
 50  
 51          # Generate RPCAUTH with specified password
 52          self.rt2password = "8/F3uMDw4KSEbw96U3CA1C4X05dkHDN2BPFjTgZW4KI="
 53          p = subprocess.Popen([sys.executable, gen_rpcauth, 'rt2', self.rt2password], stdout=subprocess.PIPE, text=True)
 54          lines = p.stdout.read().splitlines()
 55          rpcauth2 = lines[1]
 56  
 57          # Generate RPCAUTH without specifying password
 58          self.user = ''.join(SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(10))
 59          p = subprocess.Popen([sys.executable, gen_rpcauth, self.user], stdout=subprocess.PIPE, text=True)
 60          lines = p.stdout.read().splitlines()
 61          rpcauth3 = lines[1]
 62          self.password = lines[3]
 63  
 64          with open(self.nodes[0].datadir_path / "bitcoin.conf", "a", encoding="utf8") as f:
 65              f.write(rpcauth + "\n")
 66              f.write(rpcauth2 + "\n")
 67              f.write(rpcauth3 + "\n")
 68          with open(self.nodes[1].datadir_path / "bitcoin.conf", "a", encoding="utf8") as f:
 69              f.write("rpcuser={}\n".format(self.rpcuser))
 70              f.write("rpcpassword={}\n".format(self.rpcpassword))
 71          self.restart_node(0)
 72          self.restart_node(1)
 73  
 74      def test_auth(self, node, user, password):
 75          self.log.info('Correct...')
 76          assert_equal(200, call_with_auth(node, user, password).status)
 77  
 78          self.log.info('Wrong...')
 79          assert_equal(401, call_with_auth(node, user, password + 'wrong').status)
 80  
 81          self.log.info('Wrong...')
 82          assert_equal(401, call_with_auth(node, user + 'wrong', password).status)
 83  
 84          self.log.info('Wrong...')
 85          assert_equal(401, call_with_auth(node, user + 'wrong', password + 'wrong').status)
 86  
 87      def run_test(self):
 88          self.conf_setup()
 89          self.log.info('Check correctness of the rpcauth config option')
 90          url = urllib.parse.urlparse(self.nodes[0].url)
 91  
 92          self.test_auth(self.nodes[0], url.username, url.password)
 93          self.test_auth(self.nodes[0], 'rt', self.rtpassword)
 94          self.test_auth(self.nodes[0], 'rt2', self.rt2password)
 95          self.test_auth(self.nodes[0], self.user, self.password)
 96  
 97          self.log.info('Check correctness of the rpcuser/rpcpassword config options')
 98          url = urllib.parse.urlparse(self.nodes[1].url)
 99  
100          self.test_auth(self.nodes[1], self.rpcuser, self.rpcpassword)
101  
102          init_error = 'Error: Unable to start HTTP server. See debug log for details.'
103  
104          self.log.info('Check -rpcauth are validated')
105          # Empty -rpcauth= are ignored
106          self.restart_node(0, extra_args=['-rpcauth='])
107          self.stop_node(0)
108          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error, extra_args=['-rpcauth=foo'])
109          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error, extra_args=['-rpcauth=foo:bar'])
110          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error, extra_args=['-rpcauth=foo:bar:baz'])
111          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error, extra_args=['-rpcauth=foo$bar:baz'])
112          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error, extra_args=['-rpcauth=foo$bar$baz'])
113  
114          self.log.info('Check that failure to write cookie file will abort the node gracefully')
115          (self.nodes[0].chain_path / ".cookie.tmp").mkdir()
116          self.nodes[0].assert_start_raises_init_error(expected_msg=init_error)
117  
118  
119  if __name__ == '__main__':
120      HTTPBasicsTest().main()