1 #!/usr/bin/env python3
2 # Copyright (c) 2015-2021 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 """Test the setban rpc call."""
6 7 from contextlib import ExitStack
8 from test_framework.test_framework import LimenkaTestFramework
9 from test_framework.util import (
10 p2p_port,
11 assert_equal,
12 )
13 14 class SetBanTests(LimenkaTestFramework):
15 def set_test_params(self):
16 self.num_nodes = 2
17 self.setup_clean_chain = True
18 self.extra_args = [[],[]]
19 20 def is_banned(self, node, addr):
21 return any(e['address'] == addr for e in node.listbanned())
22 23 def run_test(self):
24 # Node 0 connects to Node 1, check that the noban permission is not granted
25 self.connect_nodes(0, 1)
26 peerinfo = self.nodes[1].getpeerinfo()[0]
27 assert not "noban" in peerinfo["permissions"]
28 29 # Node 0 get banned by Node 1
30 self.nodes[1].setban("127.0.0.1", "add")
31 32 # Node 0 should not be able to reconnect
33 context = ExitStack()
34 context.enter_context(self.nodes[1].assert_debug_log(expected_msgs=['dropped (banned)\n'], timeout=50))
35 # When disconnected right after connecting, a v2 node will attempt to reconnect with v1.
36 # Wait for that to happen so that it cannot mess with later tests.
37 if self.options.v2transport:
38 context.enter_context(self.nodes[0].assert_debug_log(expected_msgs=['trying v1 connection'], timeout=50))
39 with context:
40 self.restart_node(1, [])
41 self.nodes[0].addnode("127.0.0.1:" + str(p2p_port(1)), "onetry")
42 43 # However, node 0 should be able to reconnect if it has noban permission
44 self.restart_node(1, ['-whitelist=127.0.0.1'])
45 self.connect_nodes(0, 1)
46 peerinfo = self.nodes[1].getpeerinfo()[0]
47 assert "noban" in peerinfo["permissions"]
48 49 # If we remove the ban, Node 0 should be able to reconnect even without noban permission
50 self.nodes[1].setban("127.0.0.1", "remove")
51 self.restart_node(1, [])
52 self.connect_nodes(0, 1)
53 peerinfo = self.nodes[1].getpeerinfo()[0]
54 assert not "noban" in peerinfo["permissions"]
55 56 self.log.info("Test that a non-IP address can be banned/unbanned")
57 node = self.nodes[1]
58 tor_addr = "pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion"
59 ip_addr = "1.2.3.4"
60 assert not self.is_banned(node, tor_addr)
61 assert not self.is_banned(node, ip_addr)
62 63 node.setban(tor_addr, "add")
64 assert self.is_banned(node, tor_addr)
65 assert not self.is_banned(node, ip_addr)
66 67 node.setban(tor_addr, "remove")
68 assert not self.is_banned(self.nodes[1], tor_addr)
69 assert not self.is_banned(node, ip_addr)
70 71 self.log.info("Test -bantime")
72 self.restart_node(1, ["-bantime=1234"])
73 self.nodes[1].setban("127.0.0.1", "add")
74 banned = self.nodes[1].listbanned()[0]
75 assert_equal(banned['ban_duration'], 1234)
76 77 if __name__ == '__main__':
78 SetBanTests(__file__).main()
79