mirror of
https://github.com/bitcoin/bitcoin.git
synced 2025-02-12 11:19:08 -05:00
Merge f0e5e4cdbe
into 85f96b01b7
This commit is contained in:
commit
98dc09f9fa
1 changed files with 62 additions and 13 deletions
|
@ -13,6 +13,7 @@ from test_framework.util import (
|
||||||
import http.client
|
import http.client
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
|
|
||||||
|
|
||||||
def rpccall(node, user, method):
|
def rpccall(node, user, method):
|
||||||
url = urllib.parse.urlparse(node.url)
|
url = urllib.parse.urlparse(node.url)
|
||||||
headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(user[0], user[3]))}
|
headers = {"Authorization": "Basic " + str_to_b64str('{}:{}'.format(user[0], user[3]))}
|
||||||
|
@ -24,7 +25,12 @@ def rpccall(node, user, method):
|
||||||
return resp
|
return resp
|
||||||
|
|
||||||
|
|
||||||
|
def get_permissions(whitelist):
|
||||||
|
return [perm for perm in whitelist.replace(" ", "").split(",") if perm]
|
||||||
|
|
||||||
|
|
||||||
class RPCWhitelistTest(BitcoinTestFramework):
|
class RPCWhitelistTest(BitcoinTestFramework):
|
||||||
|
|
||||||
def set_test_params(self):
|
def set_test_params(self):
|
||||||
self.num_nodes = 1
|
self.num_nodes = 1
|
||||||
|
|
||||||
|
@ -48,7 +54,9 @@ class RPCWhitelistTest(BitcoinTestFramework):
|
||||||
["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount, getbestblockhash", "s7R4nG3R7H1nGZ"],
|
["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount, getbestblockhash", "s7R4nG3R7H1nGZ"],
|
||||||
["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount", "s7R4nG3R7H1nGZ"],
|
["strangedude4", "990c895760a70df83949e8278665e19a$8f0906f20431ff24cb9e7f5b5041e4943bdf2a5c02a19ef4960dcf45e72cde1c", ":getblockcount", "s7R4nG3R7H1nGZ"],
|
||||||
# Testing the same permission twice
|
# Testing the same permission twice
|
||||||
["strangedude5", "d12c6e962d47a454f962eb41225e6ec8$2dd39635b155536d3c1a2e95d05feff87d5ba55f2d5ff975e6e997a836b717c9", ":getblockcount,getblockcount", "s7R4nG3R7H1nGZ"]
|
["strangedude5", "d12c6e962d47a454f962eb41225e6ec8$2dd39635b155536d3c1a2e95d05feff87d5ba55f2d5ff975e6e997a836b717c9", ":getblockcount,getblockcount", "s7R4nG3R7H1nGZ"],
|
||||||
|
# Test non-whitelisted user
|
||||||
|
["strangedude6", "ab02e4fb22ef4ab004cca217a49ee8d2$90dd09b08edd12d552d9d8a5ada838dcef2ac587789fa7e9c47f5990e80cdf93", None, "password123"]
|
||||||
]
|
]
|
||||||
# These commands shouldn't be allowed for any user to test failures
|
# These commands shouldn't be allowed for any user to test failures
|
||||||
self.never_allowed = ["getnetworkinfo"]
|
self.never_allowed = ["getnetworkinfo"]
|
||||||
|
@ -60,21 +68,11 @@ class RPCWhitelistTest(BitcoinTestFramework):
|
||||||
# Special cases
|
# Special cases
|
||||||
for strangedude in self.strange_users:
|
for strangedude in self.strange_users:
|
||||||
f.write("rpcauth=" + strangedude[0] + ":" + strangedude[1] + "\n")
|
f.write("rpcauth=" + strangedude[0] + ":" + strangedude[1] + "\n")
|
||||||
f.write("rpcwhitelist=" + strangedude[0] + strangedude[2] + "\n")
|
if strangedude[2] is not None:
|
||||||
|
f.write("rpcwhitelist=" + strangedude[0] + strangedude[2] + "\n")
|
||||||
self.restart_node(0)
|
self.restart_node(0)
|
||||||
|
|
||||||
for user in self.users:
|
for user in self.users:
|
||||||
permissions = user[2].replace(" ", "").split(",")
|
|
||||||
# Pop all empty items
|
|
||||||
i = 0
|
|
||||||
while i < len(permissions):
|
|
||||||
if permissions[i] == '':
|
|
||||||
permissions.pop(i)
|
|
||||||
|
|
||||||
i += 1
|
|
||||||
for permission in permissions:
|
|
||||||
self.log.info("[" + user[0] + "]: Testing a permitted permission (" + permission + ")")
|
|
||||||
assert_equal(200, rpccall(self.nodes[0], user, permission).status)
|
|
||||||
for permission in self.never_allowed:
|
for permission in self.never_allowed:
|
||||||
self.log.info("[" + user[0] + "]: Testing a non permitted permission (" + permission + ")")
|
self.log.info("[" + user[0] + "]: Testing a non permitted permission (" + permission + ")")
|
||||||
assert_equal(403, rpccall(self.nodes[0], user, permission).status)
|
assert_equal(403, rpccall(self.nodes[0], user, permission).status)
|
||||||
|
@ -92,5 +90,56 @@ class RPCWhitelistTest(BitcoinTestFramework):
|
||||||
self.log.info("Strange test 5")
|
self.log.info("Strange test 5")
|
||||||
assert_equal(200, rpccall(self.nodes[0], self.strange_users[4], "getblockcount").status)
|
assert_equal(200, rpccall(self.nodes[0], self.strange_users[4], "getblockcount").status)
|
||||||
|
|
||||||
|
self.test_users_permissions()
|
||||||
|
self.test_rpcwhitelistdefault_0_no_permissions()
|
||||||
|
|
||||||
|
# Replace file configurations
|
||||||
|
self.nodes[0].replace_in_config([("rpcwhitelistdefault=0", "rpcwhitelistdefault=1")])
|
||||||
|
with open(self.nodes[0].datadir_path / "bitcoin.conf", 'a', encoding='utf8') as f:
|
||||||
|
f.write("rpcwhitelist=__cookie__:getblockcount,getblockchaininfo,getmempoolinfo,stop\n")
|
||||||
|
self.restart_node(0)
|
||||||
|
|
||||||
|
# Test rpcwhitelistdefault=1
|
||||||
|
self.test_users_permissions()
|
||||||
|
self.test_rpcwhitelistdefault_1_no_permissions()
|
||||||
|
|
||||||
|
def test_users_permissions(self):
|
||||||
|
"""
|
||||||
|
* Permissions:
|
||||||
|
(user1): getbestblockhash,getblockcount
|
||||||
|
(user2): getblockcount
|
||||||
|
Expected result: * users can only access whitelisted methods
|
||||||
|
"""
|
||||||
|
for user in self.users:
|
||||||
|
permissions = get_permissions(user[2])
|
||||||
|
for permission in permissions:
|
||||||
|
self.log.info("[" + user[0] + "]: Testing whitelisted user permission (" + permission + ")")
|
||||||
|
assert_equal(200, rpccall(self.nodes[0], user, permission).status)
|
||||||
|
self.log.info("[" + user[0] + "]: Testing non-permitted permission: getblockchaininfo")
|
||||||
|
assert_equal(403, rpccall(self.nodes[0], user, "getblockchaininfo").status)
|
||||||
|
|
||||||
|
def test_rpcwhitelistdefault_0_no_permissions(self):
|
||||||
|
"""
|
||||||
|
* rpcwhitelistdefault=0
|
||||||
|
* No Permissions defined
|
||||||
|
Expected result: * strangedude6 (not whitelisted) can access any method
|
||||||
|
"""
|
||||||
|
unrestricted_user = self.strange_users[6]
|
||||||
|
for permission in ["getbestblockhash", "getblockchaininfo"]:
|
||||||
|
self.log.info("[" + unrestricted_user[0] + "]: Testing unrestricted user permission (" + permission + ")")
|
||||||
|
assert_equal(200, rpccall(self.nodes[0], unrestricted_user, permission).status)
|
||||||
|
|
||||||
|
def test_rpcwhitelistdefault_1_no_permissions(self):
|
||||||
|
"""
|
||||||
|
* rpcwhitelistdefault=1
|
||||||
|
* No Permissions defined
|
||||||
|
Expected result: * strangedude6 (not whitelisted) can not access any method
|
||||||
|
"""
|
||||||
|
|
||||||
|
for permission in ["getbestblockhash", "getblockchaininfo"]:
|
||||||
|
self.log.info("[" + self.strange_users[6][0] + "]: Testing rpcwhitelistdefault=1 no specified permission (" + permission + ")")
|
||||||
|
assert_equal(403, rpccall(self.nodes[0], self.strange_users[6], permission).status)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
RPCWhitelistTest(__file__).main()
|
RPCWhitelistTest(__file__).main()
|
||||||
|
|
Loading…
Add table
Reference in a new issue