1 #!/usr/bin/env python3
2 # Copyright (c) 2015-present 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 """Test the setban rpc call."""
6 7 from contextlib import ExitStack
8 from test_framework.test_framework import BitcoinTestFramework
9 from test_framework.util import (
10 p2p_port,
11 assert_equal,
12 )
13 14 class SetBanTests(BitcoinTestFramework):
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 "noban" not in peerinfo["permissions"]
28 29 # Node 0 gets banned by Node 1
30 self.nodes[1].setban("127.0.0.1", "add")
31 self.wait_until(lambda: not self.nodes[0].is_connected_to(self.nodes[1]))
32 33 # Node 0 should not be able to reconnect
34 self.restart_node(1, [])
35 context = ExitStack()
36 context.enter_context(self.nodes[1].assert_debug_log(expected_msgs=["dropped (banned)\n"]))
37 # When disconnected right after connecting, a v2 node will attempt to reconnect with v1.
38 # Wait for all disconnects on node0, so that it cannot mess with later tests.
39 context.enter_context(self.nodes[0].assert_debug_log(
40 expected_msgs=[
41 "retrying with v1 transport protocol for peer=2",
42 "Cleared nodestate for peer=2",
43 "Cleared nodestate for peer=3",
44 ] if self.options.v2transport else [
45 "Cleared nodestate for peer=2", # Just one v1 disconnect to wait for
46 ],
47 timeout=8))
48 with context:
49 self.nodes[0].addnode("127.0.0.1:" + str(p2p_port(1)), "onetry")
50 assert not self.nodes[0].is_connected_to(self.nodes[1])
51 52 # However, node 0 should be able to reconnect if it has noban permission
53 self.restart_node(1, ['-whitelist=127.0.0.1'])
54 self.connect_nodes(0, 1)
55 peerinfo = self.nodes[1].getpeerinfo()[0]
56 assert "noban" in peerinfo["permissions"]
57 58 # If we remove the ban, Node 0 should be able to reconnect even without noban permission
59 self.nodes[1].setban("127.0.0.1", "remove")
60 self.restart_node(1, [])
61 self.connect_nodes(0, 1)
62 peerinfo = self.nodes[1].getpeerinfo()[0]
63 assert "noban" not in peerinfo["permissions"]
64 65 self.log.info("Test that a non-IP address can be banned/unbanned")
66 node = self.nodes[1]
67 tor_addr = "pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion"
68 ip_addr = "1.2.3.4"
69 assert not self.is_banned(node, tor_addr)
70 assert not self.is_banned(node, ip_addr)
71 72 node.setban(tor_addr, "add")
73 assert self.is_banned(node, tor_addr)
74 assert not self.is_banned(node, ip_addr)
75 76 node.setban(tor_addr, "remove")
77 assert not self.is_banned(self.nodes[1], tor_addr)
78 assert not self.is_banned(node, ip_addr)
79 80 self.log.info("Test -bantime")
81 self.restart_node(1, ["-bantime=1234"])
82 self.nodes[1].setban("127.0.0.1", "add")
83 banned = self.nodes[1].listbanned()[0]
84 assert_equal(banned['ban_duration'], 1234)
85 86 if __name__ == '__main__':
87 SetBanTests(__file__).main()
88