Merge pull request #3564 from nanocurrency/network_limiting

Improve and simplify a number of network limiters

nano/boost/asio/ip/network_v6.hpp

@@ -0,0 +1,7 @@
+#pragma once
+
+#include <nano/boost/private/macro_warnings.hpp>
+
+DISABLE_ASIO_WARNINGS
+#include <boost/asio/ip/network_v6.hpp>
+REENABLE_WARNINGS

nano/core_test/network.cpp

@@ -1,3 +1,4 @@
+#include <nano/node/nodeconfig.hpp>
 #include <nano/node/transport/udp.hpp>
 #include <nano/test_common/network.hpp>
 #include <nano/test_common/system.hpp>
@@ -918,7 +919,10 @@ namespace transport
 {
 	TEST (network, peer_max_tcp_attempts_subnetwork)
 	{
-		nano::system system (1);
+		nano::node_flags node_flags;
+		node_flags.disable_max_peers_per_ip = true;
+		nano::system system;
+		system.add_node (node_flags);
 		auto node (system.nodes[0]);
 		for (auto i (0); i < node->network_params.network.max_peers_per_subnetwork; ++i)
 		{
@@ -927,9 +931,9 @@ namespace transport
 			ASSERT_FALSE (node->network.tcp_channels.reachout (endpoint));
 		}
 		ASSERT_EQ (0, node->network.size ());
-		ASSERT_EQ (0, node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_ip, nano::stat::dir::out));
+		ASSERT_EQ (0, node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_subnetwork, nano::stat::dir::out));
 		ASSERT_TRUE (node->network.tcp_channels.reachout (nano::endpoint (boost::asio::ip::make_address_v6 ("::ffff:127.0.0.1"), nano::get_available_port ())));
-		ASSERT_EQ (1, node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_ip, nano::stat::dir::out));
+		ASSERT_EQ (1, node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_subnetwork, nano::stat::dir::out));
 	}
 }
 }

nano/core_test/socket.cpp

@@ -1,27 +1,28 @@
+#include <nano/boost/asio/ip/address_v6.hpp>
+#include <nano/boost/asio/ip/network_v6.hpp>
 #include <nano/lib/threading.hpp>
 #include <nano/node/socket.hpp>
 #include <nano/test_common/system.hpp>
 #include <nano/test_common/testutil.hpp>
 
 #include <gtest/gtest.h>
 
+#include <map>
+#include <memory>
+#include <utility>
+#include <vector>
+
 using namespace std::chrono_literals;
 
 TEST (socket, max_connections)
 {
-	// this is here just so that ASSERT_TIMELY can be used
 	nano::system system;
 
-	auto node_flags = nano::inactive_node_flag_defaults ();
-	node_flags.read_only = false;
-	nano::inactive_node inactivenode (nano::unique_path (), node_flags);
-	auto node = inactivenode.node;
-
-	nano::thread_runner runner (node->io_ctx, 1);
+	auto node = system.add_node ();
 
-	auto server_port (nano::get_available_port ());
-	boost::asio::ip::tcp::endpoint listen_endpoint (boost::asio::ip::address_v6::any (), server_port);
-	boost::asio::ip::tcp::endpoint dst_endpoint (boost::asio::ip::address_v6::loopback (), server_port);
+	auto server_port = nano::get_available_port ();
+	boost::asio::ip::tcp::endpoint listen_endpoint{ boost::asio::ip::address_v6::any (), server_port };
+	boost::asio::ip::tcp::endpoint dst_endpoint{ boost::asio::ip::address_v6::loopback (), server_port };
 
 	// start a server socket that allows max 2 live connections
 	auto server_socket = std::make_shared<nano::server_socket> (*node, listen_endpoint, 2);
@@ -37,10 +38,10 @@ TEST (socket, max_connections)
 	});
 
 	// client side connection tracking
-	std::atomic<int> connection_attempts = 0;
+	std::atomic<size_t> connection_attempts = 0;
 	auto connect_handler = [&connection_attempts] (boost::system::error_code const & ec_a) {
 		ASSERT_EQ (ec_a.value (), 0);
-		connection_attempts++;
+		++connection_attempts;
 	};
 
 	// start 3 clients, 2 will persist but 1 will be dropped
@@ -102,8 +103,280 @@ TEST (socket, max_connections)
 	ASSERT_TIMELY (5s, server_sockets.size () == 5); // connections accepted by the server
 
 	node->stop ();
-	runner.stop_event_processing ();
-	runner.join ();
+}
+
+TEST (socket, max_connections_per_ip)
+{
+	nano::system system;
+
+	auto node = system.add_node ();
+	ASSERT_FALSE (node->flags.disable_max_peers_per_ip);
+
+	auto server_port = nano::get_available_port ();
+	boost::asio::ip::tcp::endpoint listen_endpoint{ boost::asio::ip::address_v6::any (), server_port };
+	boost::asio::ip::tcp::endpoint dst_endpoint{ boost::asio::ip::address_v6::loopback (), server_port };
+
+	const auto max_ip_connections = node->network_params.network.max_peers_per_ip;
+	ASSERT_TRUE (max_ip_connections >= 1);
+
+	const auto max_global_connections = 1000;
+
+	auto server_socket = std::make_shared<nano::server_socket> (*node, listen_endpoint, max_global_connections);
+	boost::system::error_code ec;
+	server_socket->start (ec);
+	ASSERT_FALSE (ec);
+
+	// successful incoming connections are stored in server_sockets to keep them alive (server side)
+	std::vector<std::shared_ptr<nano::socket>> server_sockets;
+	server_socket->on_connection ([&server_sockets] (std::shared_ptr<nano::socket> const & new_connection, boost::system::error_code const & ec_a) {
+		server_sockets.push_back (new_connection);
+		return true;
+	});
+
+	// client side connection tracking
+	std::atomic<size_t> connection_attempts = 0;
+	auto connect_handler = [&connection_attempts] (boost::system::error_code const & ec_a) {
+		ASSERT_EQ (ec_a.value (), 0);
+		++connection_attempts;
+	};
+
+	// start n clients, n-1 will persist but 1 will be dropped, where n == max_ip_connections
+	std::vector<std::shared_ptr<nano::socket>> client_list;
+	client_list.reserve (max_ip_connections + 1);
+
+	for (auto idx = 0; idx < max_ip_connections + 1; ++idx)
+	{
+		auto client = std::make_shared<nano::socket> (*node);
+		client->async_connect (dst_endpoint, connect_handler);
+		client_list.push_back (client);
+	}
+
+	auto get_tcp_max_per_ip = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_ip, nano::stat::dir::in);
+	};
+
+	auto get_tcp_accept_successes = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_accept_success, nano::stat::dir::in);
+	};
+
+	ASSERT_TIMELY (5s, get_tcp_accept_successes () == max_ip_connections);
+	ASSERT_TIMELY (5s, get_tcp_max_per_ip () == 1);
+	ASSERT_TIMELY (5s, connection_attempts == max_ip_connections + 1);
+
+	node->stop ();
+}
+
+TEST (socket, limited_subnet_address)
+{
+	auto address = boost::asio::ip::make_address ("a41d:b7b2:8298:cf45:672e:bd1a:e7fb:f713");
+	auto network = nano::socket_functions::get_ipv6_subnet_address (address.to_v6 (), 32); // network prefix = 32.
+	ASSERT_EQ ("a41d:b7b2:8298:cf45:672e:bd1a:e7fb:f713/32", network.to_string ());
+	ASSERT_EQ ("a41d:b7b2::/32", network.canonical ().to_string ());
+}
+
+TEST (socket, first_ipv6_subnet_address)
+{
+	auto address = boost::asio::ip::make_address ("a41d:b7b2:8298:cf45:672e:bd1a:e7fb:f713");
+	auto first_address = nano::socket_functions::first_ipv6_subnet_address (address.to_v6 (), 32); // network prefix = 32.
+	ASSERT_EQ ("a41d:b7b2::", first_address.to_string ());
+}
+
+TEST (socket, last_ipv6_subnet_address)
+{
+	auto address = boost::asio::ip::make_address ("a41d:b7b2:8298:cf45:672e:bd1a:e7fb:f713");
+	auto last_address = nano::socket_functions::last_ipv6_subnet_address (address.to_v6 (), 32); // network prefix = 32.
+	ASSERT_EQ ("a41d:b7b2:ffff:ffff:ffff:ffff:ffff:ffff", last_address.to_string ());
+}
+
+TEST (socket, count_subnetwork_connections)
+{
+	nano::system system;
+	auto node = system.add_node ();
+
+	auto address0 = boost::asio::ip::make_address ("a41d:b7b1:ffff:ffff:ffff:ffff:ffff:ffff"); // out of network prefix
+	auto address1 = boost::asio::ip::make_address ("a41d:b7b2:8298:cf45:672e:bd1a:e7fb:f713"); // referece address
+	auto address2 = boost::asio::ip::make_address ("a41d:b7b2::"); // start of the network range
+	auto address3 = boost::asio::ip::make_address ("a41d:b7b2::1");
+	auto address4 = boost::asio::ip::make_address ("a41d:b7b2:ffff:ffff:ffff:ffff:ffff:ffff"); // end of the network range
+	auto address5 = boost::asio::ip::make_address ("a41d:b7b3::"); // out of the network prefix
+	auto address6 = boost::asio::ip::make_address ("a41d:b7b3::1"); // out of the network prefix
+
+	auto connection0 = std::make_shared<nano::socket> (*node);
+	auto connection1 = std::make_shared<nano::socket> (*node);
+	auto connection2 = std::make_shared<nano::socket> (*node);
+	auto connection3 = std::make_shared<nano::socket> (*node);
+	auto connection4 = std::make_shared<nano::socket> (*node);
+	auto connection5 = std::make_shared<nano::socket> (*node);
+	auto connection6 = std::make_shared<nano::socket> (*node);
+
+	nano::address_socket_mmap connections_per_address;
+	connections_per_address.emplace (address0, connection0);
+	connections_per_address.emplace (address1, connection1);
+	connections_per_address.emplace (address2, connection2);
+	connections_per_address.emplace (address3, connection3);
+	connections_per_address.emplace (address4, connection4);
+	connections_per_address.emplace (address5, connection5);
+	connections_per_address.emplace (address6, connection6);
+
+	// Asserts it counts only the connections for the specified address and its network prefix.
+	ASSERT_EQ (4, nano::socket_functions::count_subnetwork_connections (connections_per_address, address1.to_v6 (), 32));
+}
+
+TEST (socket, max_connections_per_subnetwork)
+{
+	nano::system system;
+
+	nano::node_flags node_flags;
+	// disabling IP limit because it will be used the same IP address to check they come from the same subnetwork.
+	node_flags.disable_max_peers_per_ip = true;
+	node_flags.disable_max_peers_per_subnetwork = false;
+	auto node = system.add_node (node_flags);
+	ASSERT_TRUE (node->flags.disable_max_peers_per_ip);
+	ASSERT_FALSE (node->flags.disable_max_peers_per_subnetwork);
+
+	auto server_port = nano::get_available_port ();
+	boost::asio::ip::tcp::endpoint listen_endpoint{ boost::asio::ip::address_v6::any (), server_port };
+	boost::asio::ip::tcp::endpoint dst_endpoint{ boost::asio::ip::address_v6::loopback (), server_port };
+
+	const auto max_subnetwork_connections = node->network_params.network.max_peers_per_subnetwork;
+	ASSERT_TRUE (max_subnetwork_connections >= 1);
+
+	const auto max_global_connections = 1000;
+
+	auto server_socket = std::make_shared<nano::server_socket> (*node, listen_endpoint, max_global_connections);
+	boost::system::error_code ec;
+	server_socket->start (ec);
+	ASSERT_FALSE (ec);
+
+	// successful incoming connections are stored in server_sockets to keep them alive (server side)
+	std::vector<std::shared_ptr<nano::socket>> server_sockets;
+	server_socket->on_connection ([&server_sockets] (std::shared_ptr<nano::socket> const & new_connection, boost::system::error_code const & ec_a) {
+		server_sockets.push_back (new_connection);
+		return true;
+	});
+
+	// client side connection tracking
+	std::atomic<size_t> connection_attempts = 0;
+	auto connect_handler = [&connection_attempts] (boost::system::error_code const & ec_a) {
+		ASSERT_EQ (ec_a.value (), 0);
+		++connection_attempts;
+	};
+
+	// start n clients, n-1 will persist but 1 will be dropped, where n == max_subnetwork_connections
+	std::vector<std::shared_ptr<nano::socket>> client_list;
+	client_list.reserve (max_subnetwork_connections + 1);
+
+	for (auto idx = 0; idx < max_subnetwork_connections + 1; ++idx)
+	{
+		auto client = std::make_shared<nano::socket> (*node);
+		client->async_connect (dst_endpoint, connect_handler);
+		client_list.push_back (client);
+	}
+
+	auto get_tcp_max_per_subnetwork = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_subnetwork, nano::stat::dir::in);
+	};
+
+	auto get_tcp_accept_successes = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_accept_success, nano::stat::dir::in);
+	};
+
+	ASSERT_TIMELY (5s, get_tcp_accept_successes () == max_subnetwork_connections);
+	ASSERT_TIMELY (5s, get_tcp_max_per_subnetwork () == 1);
+	ASSERT_TIMELY (5s, connection_attempts == max_subnetwork_connections + 1);
+
+	node->stop ();
+}
+
+TEST (socket, disabled_max_peers_per_ip)
+{
+	nano::system system;
+
+	nano::node_flags node_flags;
+	node_flags.disable_max_peers_per_ip = true;
+	auto node = system.add_node (node_flags);
+	ASSERT_TRUE (node->flags.disable_max_peers_per_ip);
+
+	auto server_port = nano::get_available_port ();
+	boost::asio::ip::tcp::endpoint listen_endpoint{ boost::asio::ip::address_v6::any (), server_port };
+	boost::asio::ip::tcp::endpoint dst_endpoint{ boost::asio::ip::address_v6::loopback (), server_port };
+
+	const auto max_ip_connections = node->network_params.network.max_peers_per_ip;
+	ASSERT_TRUE (max_ip_connections >= 1);
+
+	const auto max_global_connections = 1000;
+
+	auto server_socket = std::make_shared<nano::server_socket> (*node, listen_endpoint, max_global_connections);
+	boost::system::error_code ec;
+	server_socket->start (ec);
+	ASSERT_FALSE (ec);
+
+	// successful incoming connections are stored in server_sockets to keep them alive (server side)
+	std::vector<std::shared_ptr<nano::socket>> server_sockets;
+	server_socket->on_connection ([&server_sockets] (std::shared_ptr<nano::socket> const & new_connection, boost::system::error_code const & ec_a) {
+		server_sockets.push_back (new_connection);
+		return true;
+	});
+
+	// client side connection tracking
+	std::atomic<size_t> connection_attempts = 0;
+	auto connect_handler = [&connection_attempts] (boost::system::error_code const & ec_a) {
+		ASSERT_EQ (ec_a.value (), 0);
+		++connection_attempts;
+	};
+
+	// start n clients, n-1 will persist but 1 will be dropped, where n == max_ip_connections
+	std::vector<std::shared_ptr<nano::socket>> client_list;
+	client_list.reserve (max_ip_connections + 1);
+
+	for (auto idx = 0; idx < max_ip_connections + 1; ++idx)
+	{
+		auto client = std::make_shared<nano::socket> (*node);
+		client->async_connect (dst_endpoint, connect_handler);
+		client_list.push_back (client);
+	}
+
+	auto get_tcp_max_per_ip = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_max_per_ip, nano::stat::dir::in);
+	};
+
+	auto get_tcp_accept_successes = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_accept_success, nano::stat::dir::in);
+	};
+
+	ASSERT_TIMELY (5s, get_tcp_accept_successes () == max_ip_connections + 1);
+	ASSERT_TIMELY (5s, get_tcp_max_per_ip () == 0);
+	ASSERT_TIMELY (5s, connection_attempts == max_ip_connections + 1);
+
+	node->stop ();
+}
+
+TEST (socket, disconnection_of_silent_connections)
+{
+	nano::system system;
+	auto node = system.add_node ();
+	auto socket = std::make_shared<nano::socket> (*node);
+	// Classify the socket type as real-time as the disconnections are done only for this connection type.
+	socket->type_set (nano::socket::type_t::realtime);
+	// Silent connections are connections open by external peers that don't contribute with any data.
+	socket->set_silent_connection_tolerance_time (std::chrono::seconds{ 5 });
+	auto bootstrap_endpoint = node->bootstrap.endpoint ();
+	std::atomic<bool> connected{ false };
+	// Opening a connection that will be closed because it remains silent during the tolerance time.
+	socket->async_connect (bootstrap_endpoint, [socket, &connected] (boost::system::error_code const & ec) {
+		ASSERT_FALSE (ec);
+		connected = true;
+	});
+	ASSERT_TIMELY (4s, connected);
+	// Checking the connection was closed.
+	ASSERT_TIMELY (10s, socket->is_closed ());
+
+	auto get_tcp_silent_connection_drops = [&node] () {
+		return node->stats.count (nano::stat::type::tcp, nano::stat::detail::tcp_silent_connection_drop, nano::stat::dir::in);
+	};
+	ASSERT_EQ (1, get_tcp_silent_connection_drops ());
+
+	node->stop ();
 }
 
 TEST (socket, drop_policy)
@@ -173,6 +446,8 @@ TEST (socket, concurrent_writes)
 {
 	auto node_flags = nano::inactive_node_flag_defaults ();
 	node_flags.read_only = false;
+	node_flags.disable_max_peers_per_ip = true;
+	node_flags.disable_max_peers_per_subnetwork = true;
 	nano::inactive_node inactivenode (nano::unique_path (), node_flags);
 	auto node = inactivenode.node;