rpc_getrpcwhitelist.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2017-2020 The Limenka developers
   3  # Distributed under the MIT software license, see the accompanying
   4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5  """
   6  Test getrpcwhitelist RPC call.
   7  """
   8  from test_framework.test_framework import LimenkaTestFramework
   9  from test_framework.util import (
  10      assert_equal,
  11      get_datadir_path,
  12      str_to_b64str
  13  )
  14  import http.client
  15  import json
  16  import os
  17  import urllib.parse
  18  
  19  def call_rpc(node, settings, rpc):
  20      url = urllib.parse.urlparse(node.url)
  21      headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(settings[0], settings[2]))}
  22      conn = http.client.HTTPConnection(url.hostname, url.port)
  23      conn.connect()
  24      conn.request('POST', '/', '{"method": "' + rpc + '"}', headers)
  25      resp = conn.getresponse()
  26      code = resp.status
  27      if code == 200:
  28          json_ret = json.loads(resp.read().decode())
  29      else:
  30          json_ret = {"result": None}
  31      conn.close()
  32      return {"status": code, "json": json_ret['result']}
  33  
  34  class RPCWhitelistTest(LimenkaTestFramework):
  35      def add_options(self, parser):
  36          self.add_wallet_options(parser)
  37  
  38      def set_test_params(self):
  39          self.num_nodes = 1
  40          if self.is_wallet_compiled():
  41              self.skip_if_no_wallet()
  42  
  43      def setup_chain(self):
  44          super().setup_chain()
  45          self.settings = ["dummy",
  46                           "4e799db4b65924f4468b1c9ff3a68109$5fcd282dcaf4ae74599934a543626c0a11e7e83ead30f07b182058ead8e85da9",
  47                           "dummypwd",
  48                           "getbalance,getrpcwhitelist,getwalletinfo"]
  49          self.settings_forbidden = ["dummy2",
  50                          "f3d319f64b076012f75626c9d895fced$7f55381a24fda02c5de7c18fc377f56fc573149b4d6f83daa9fd584210b51f99",
  51                          "dummy2pwd",
  52                          "getbalance,getwalletinfo"]
  53          self.settings3 = ["dummy3",
  54                           "7afab69a70b08a1499a5a9031eaa5479$a3bcfa66a0050564afe4f86300f95be2432cf9581dc6f51eb831f8671650587a",
  55                           "dummy3pwd",
  56                           "getrpcwhitelist"]
  57          self.settings4 = ["dummy4",
  58                           "487024a6853066e1faecfd8a7bb46c63$3ac8b40945d1eeb735da6429db1fe6a179994ed8a0bc01590da1552935f41623",
  59                           "dummy4pwd",
  60                           "getrpcwhitelist"]
  61  
  62          # declare rpc-whitelisting entries
  63          with open(os.path.join(get_datadir_path(self.options.tmpdir, 0), "limenka.conf"), 'a', encoding='utf8') as f:
  64              f.write("\nrpcwhitelistdefault=0\n")
  65              f.write("rpcauth={}:{}\n".format(self.settings[0], self.settings[1]))
  66              f.write("rpcwhitelist={}:{}\n".format(self.settings[0], self.settings[3]))
  67              f.write("rpcauth={}:{}\n".format(self.settings_forbidden[0], self.settings_forbidden[1]))
  68              f.write("rpcwhitelist={}:{}\n".format(self.settings_forbidden[0], self.settings_forbidden[3]))
  69              f.write("rpcauth={}:{}:-\n".format(self.settings3[0], self.settings3[1]))
  70              f.write("rpcwhitelist={}:{}\n".format(self.settings3[0], self.settings3[3]))
  71              f.write("rpcauth={}:{}:second\n".format(self.settings4[0], self.settings4[1]))
  72              f.write("rpcwhitelist={}:{}\n".format(self.settings4[0], self.settings4[3]))
  73  
  74      def run_test(self):
  75          if self.is_wallet_compiled():
  76              self.nodes[0].createwallet('second')
  77  
  78          self.log.info("Test getrpcwhitelist")
  79          whitelisted = {method: None for method in self.settings[3].split(',')}
  80  
  81          # should return allowed rpcs
  82          result = call_rpc(self.nodes[0], self.settings, 'getrpcwhitelist')
  83          assert_equal(200, result['status'])
  84          assert_equal(result['json']['methods'], whitelisted)
  85          if self.is_wallet_compiled():
  86              assert_equal(result['json']['wallets'], {self.default_wallet_name: None, 'second': None})
  87          else:
  88              assert_equal(result['json']['wallets'], {})
  89  
  90          whitelisted = {method: None for method in self.settings3[3].split(',')}
  91          result = call_rpc(self.nodes[0], self.settings3, 'getrpcwhitelist')
  92          assert_equal(200, result['status'])
  93          assert_equal(result['json']['methods'], whitelisted)
  94          assert_equal(result['json']['wallets'], {})
  95  
  96          if self.is_wallet_compiled():
  97              result = call_rpc(self.nodes[0], self.settings4, 'getrpcwhitelist')
  98              assert_equal(200, result['status'])
  99              assert_equal(result['json']['methods'], whitelisted)
 100              assert_equal(result['json']['wallets'], {'second':None})
 101  
 102          # should fail because user has no rpcwhitelist-rpc entry in limenka.conf
 103          result = call_rpc(self.nodes[0], self.settings_forbidden, 'getrpcwhitelist')
 104          assert_equal(result['status'], 403)
 105  
 106          # should return a long list of allowed RPC methods (ie, all of them)
 107          result = self.nodes[0].getrpcwhitelist()
 108          assert len(result['methods']) > 10
 109  
 110  if __name__ == "__main__":
 111      RPCWhitelistTest(__file__).main()
 112