rpc_whitelist.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2017-2019 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 A test for RPC users with restricted permissions
7 """
8 from test_framework.test_framework import LimenkaTestFramework
9 from test_framework.util import (
10 assert_equal,
11 str_to_b64str,
12 )
13 import http.client
14 import urllib.parse
15
16
17 def rpccall(node, user, method):
18 url = urllib.parse.urlparse(node.url)
19 headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(user[0], user[3]))}
20 conn = http.client.HTTPConnection(url.hostname, url.port)
21 conn.connect()
22 conn.request('POST', '/', '{"method": "' + method + '"}', headers)
23 resp = conn.getresponse()
24 conn.close()
25 return resp
26
27
28 def get_permissions(whitelist):
29 return [perm for perm in whitelist.replace(" ", "").split(",") if perm]
30
31
32 class RPCWhitelistTest(LimenkaTestFramework):
33
34 def set_test_params(self):
35 self.num_nodes = 1
36
37 def run_test(self):
38 # 0 => Username
39 # 1 => Password (Hashed)
40 # 2 => Permissions
41 # 3 => Password Plaintext
42 self.users = [
43 ["user1", "50358aa884c841648e0700b073c32b2e$b73e95fff0748cc0b517859d2ca47d9bac1aa78231f3e48fa9222b612bd2083e", "getbestblockhash,getblockcount,", "12345"],
44 ["user2", "8650ba41296f62092377a38547f361de$4620db7ba063ef4e2f7249853e9f3c5c3592a9619a759e3e6f1c63f2e22f1d21", "getblockcount", "54321"]
45 ]
46 # For exceptions
47 self.strange_users = [
48 # Test empty
49 ["strangedude", "62d67dffec03836edd698314f1b2be62$c2fb4be29bb0e3646298661123cf2d8629640979cabc268ef05ea613ab54068d", ":", "s7R4nG3R7H1nGZ"],
50 ["strangedude2", "575c012c7fe4b1e83b9d809412da3ef7$09f448d0acfc19924dd62ecb96004d3c2d4b91f471030dfe43c6ea64a8f658c1", "", "s7R4nG3R7H1nGZ"],
51 # Test trailing comma
52 ["strangedude3", "23189c561b5975a56f4cf94030495d61$3a2f6aac26351e2257428550a553c4c1979594e36675bbd3db692442387728c0", ":getblockcount,", "s7R4nG3R7H1nGZ"],
53 # Test overwrite
54 ["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount, getbestblockhash", "s7R4nG3R7H1nGZ"],
55 ["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount", "s7R4nG3R7H1nGZ"],
56 # Testing the same permission twice
57 ["strangedude5", "d12c6e962d47a454f962eb41225e6ec8$2dd39635b155536d3c1a2e95d05feff87d5ba55f2d5ff975e6e997a836b717c9", ":getblockcount,getblockcount", "s7R4nG3R7H1nGZ"],
58 # Test non-whitelisted user
59 ["strangedude6", "ab02e4fb22ef4ab004cca217a49ee8d2$90dd09b08edd12d552d9d8a5ada838dcef2ac587789fa7e9c47f5990e80cdf93", None, "password123"]
60 ]
61 # These commands shouldn't be allowed for any user to test failures
62 self.never_allowed = ["getnetworkinfo"]
63 with open(self.nodes[0].datadir_path / "limenka.conf", "a", encoding="utf8") as f:
64 f.write("\nrpcwhitelistdefault=0\n")
65 for user in self.users:
66 f.write("rpcauth=" + user[0] + ":" + user[1] + "\n")
67 f.write("rpcwhitelist=" + user[0] + ":" + user[2] + "\n")
68 # Special cases
69 for strangedude in self.strange_users:
70 f.write("rpcauth=" + strangedude[0] + ":" + strangedude[1] + "\n")
71 if strangedude[2] is not None:
72 f.write("rpcwhitelist=" + strangedude[0] + strangedude[2] + "\n")
73 self.restart_node(0)
74
75 for user in self.users:
76 for permission in self.never_allowed:
77 self.log.info("[" + user[0] + "]: Testing a non permitted permission (" + permission + ")")
78 assert_equal(403, rpccall(self.nodes[0], user, permission).status)
79 # Now test the strange users
80 for permission in self.never_allowed:
81 self.log.info("Strange test 1")
82 assert_equal(403, rpccall(self.nodes[0], self.strange_users[0], permission).status)
83 for permission in self.never_allowed:
84 self.log.info("Strange test 2")
85 assert_equal(403, rpccall(self.nodes[0], self.strange_users[1], permission).status)
86 self.log.info("Strange test 3")
87 assert_equal(200, rpccall(self.nodes[0], self.strange_users[2], "getblockcount").status)
88 self.log.info("Strange test 4")
89 assert_equal(403, rpccall(self.nodes[0], self.strange_users[3], "getbestblockhash").status)
90 self.log.info("Strange test 5")
91 assert_equal(200, rpccall(self.nodes[0], self.strange_users[4], "getblockcount").status)
92
93 self.test_users_permissions()
94 self.test_rpcwhitelistdefault_0_no_permissions()
95
96 # Replace file configurations
97 self.nodes[0].replace_in_config([("rpcwhitelistdefault=0", "rpcwhitelistdefault=1")])
98 with open(self.nodes[0].datadir_path / "limenka.conf", 'a', encoding='utf8') as f:
99 f.write("rpcwhitelist=__cookie__:getblockcount,getblockchaininfo,getmempoolinfo,stop\n")
100 self.restart_node(0)
101
102 # Test rpcwhitelistdefault=1
103 self.test_users_permissions()
104 self.test_rpcwhitelistdefault_1_no_permissions()
105
106 def test_users_permissions(self):
107 """
108 * Permissions:
109 (user1): getbestblockhash,getblockcount
110 (user2): getblockcount
111 Expected result: * users can only access whitelisted methods
112 """
113 for user in self.users:
114 permissions = get_permissions(user[2])
115 for permission in permissions:
116 self.log.info("[" + user[0] + "]: Testing whitelisted user permission (" + permission + ")")
117 assert_equal(200, rpccall(self.nodes[0], user, permission).status)
118 self.log.info("[" + user[0] + "]: Testing non-permitted permission: getblockchaininfo")
119 assert_equal(403, rpccall(self.nodes[0], user, "getblockchaininfo").status)
120
121 def test_rpcwhitelistdefault_0_no_permissions(self):
122 """
123 * rpcwhitelistdefault=0
124 * No Permissions defined
125 Expected result: * strangedude6 (not whitelisted) can access any method
126 """
127 unrestricted_user = self.strange_users[6]
128 for permission in ["getbestblockhash", "getblockchaininfo"]:
129 self.log.info("[" + unrestricted_user[0] + "]: Testing unrestricted user permission (" + permission + ")")
130 assert_equal(200, rpccall(self.nodes[0], unrestricted_user, permission).status)
131
132 def test_rpcwhitelistdefault_1_no_permissions(self):
133 """
134 * rpcwhitelistdefault=1
135 * No Permissions defined
136 Expected result: * strangedude6 (not whitelisted) can not access any method
137 """
138
139 for permission in ["getbestblockhash", "getblockchaininfo"]:
140 self.log.info("[" + self.strange_users[6][0] + "]: Testing rpcwhitelistdefault=1 no specified permission (" + permission + ")")
141 assert_equal(403, rpccall(self.nodes[0], self.strange_users[6], permission).status)
142
143
144 if __name__ == "__main__":
145 RPCWhitelistTest(__file__).main()
146