1  #!/usr/bin/env python3
 2  # Copyright (c) 2017-2019 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  """
 6  A test for RPC users with restricted permissions
 7  """
 8  from test_framework.test_framework import BitcoinTestFramework
 9  from test_framework.util import (
10      assert_equal,
11      str_to_b64str,
12  )
13  import http.client
14  import urllib.parse
15  
16  def rpccall(node, user, method):
17      url = urllib.parse.urlparse(node.url)
18      headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(user[0], user[3]))}
19      conn = http.client.HTTPConnection(url.hostname, url.port)
20      conn.connect()
21      conn.request('POST', '/', '{"method": "' + method + '"}', headers)
22      resp = conn.getresponse()
23      conn.close()
24      return resp
25  
26  
27  class RPCWhitelistTest(BitcoinTestFramework):
28      def set_test_params(self):
29          self.num_nodes = 1
30  
31      def run_test(self):
32          # 0 => Username
33          # 1 => Password (Hashed)
34          # 2 => Permissions
35          # 3 => Password Plaintext
36          self.users = [
37              ["user1", "50358aa884c841648e0700b073c32b2e$b73e95fff0748cc0b517859d2ca47d9bac1aa78231f3e48fa9222b612bd2083e", "getbestblockhash,getblockcount,", "12345"],
38              ["user2", "8650ba41296f62092377a38547f361de$4620db7ba063ef4e2f7249853e9f3c5c3592a9619a759e3e6f1c63f2e22f1d21", "getblockcount", "54321"]
39          ]
40          # For exceptions
41          self.strange_users = [
42              # Test empty
43              ["strangedude", "62d67dffec03836edd698314f1b2be62$c2fb4be29bb0e3646298661123cf2d8629640979cabc268ef05ea613ab54068d", ":", "s7R4nG3R7H1nGZ"],
44              ["strangedude2", "575c012c7fe4b1e83b9d809412da3ef7$09f448d0acfc19924dd62ecb96004d3c2d4b91f471030dfe43c6ea64a8f658c1", "", "s7R4nG3R7H1nGZ"],
45              # Test trailing comma
46              ["strangedude3", "23189c561b5975a56f4cf94030495d61$3a2f6aac26351e2257428550a553c4c1979594e36675bbd3db692442387728c0", ":getblockcount,", "s7R4nG3R7H1nGZ"],
47              # Test overwrite
48              ["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount, getbestblockhash", "s7R4nG3R7H1nGZ"],
49              ["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount", "s7R4nG3R7H1nGZ"],
50              # Testing the same permission twice
51              ["strangedude5", "d12c6e962d47a454f962eb41225e6ec8$2dd39635b155536d3c1a2e95d05feff87d5ba55f2d5ff975e6e997a836b717c9", ":getblockcount,getblockcount", "s7R4nG3R7H1nGZ"]
52          ]
53          # These commands shouldn't be allowed for any user to test failures
54          self.never_allowed = ["getnetworkinfo"]
55          with open(self.nodes[0].datadir_path / "bitcoin.conf", "a", encoding="utf8") as f:
56              f.write("\nrpcwhitelistdefault=0\n")
57              for user in self.users:
58                  f.write("rpcauth=" + user[0] + ":" + user[1] + "\n")
59                  f.write("rpcwhitelist=" + user[0] + ":" + user[2] + "\n")
60              # Special cases
61              for strangedude in self.strange_users:
62                  f.write("rpcauth=" + strangedude[0] + ":" + strangedude[1] + "\n")
63                  f.write("rpcwhitelist=" + strangedude[0] + strangedude[2] + "\n")
64          self.restart_node(0)
65  
66          for user in self.users:
67              permissions = user[2].replace(" ", "").split(",")
68              # Pop all empty items
69              i = 0
70              while i < len(permissions):
71                  if permissions[i] == '':
72                      permissions.pop(i)
73  
74                  i += 1
75              for permission in permissions:
76                  self.log.info("[" + user[0] + "]: Testing a permitted permission (" + permission + ")")
77                  assert_equal(200, rpccall(self.nodes[0], user, permission).status)
78              for permission in self.never_allowed:
79                  self.log.info("[" + user[0] + "]: Testing a non permitted permission (" + permission + ")")
80                  assert_equal(403, rpccall(self.nodes[0], user, permission).status)
81          # Now test the strange users
82          for permission in self.never_allowed:
83              self.log.info("Strange test 1")
84              assert_equal(403, rpccall(self.nodes[0], self.strange_users[0], permission).status)
85          for permission in self.never_allowed:
86              self.log.info("Strange test 2")
87              assert_equal(403, rpccall(self.nodes[0], self.strange_users[1], permission).status)
88          self.log.info("Strange test 3")
89          assert_equal(200, rpccall(self.nodes[0], self.strange_users[2], "getblockcount").status)
90          self.log.info("Strange test 4")
91          assert_equal(403, rpccall(self.nodes[0], self.strange_users[3], "getbestblockhash").status)
92          self.log.info("Strange test 5")
93          assert_equal(200, rpccall(self.nodes[0], self.strange_users[4], "getblockcount").status)
94  
95  if __name__ == "__main__":
96      RPCWhitelistTest().main()