feat(p2p): add connection limit

packages/constants/source/environment-variables.ts

@@ -23,6 +23,7 @@ export const EnvironmentVariableNames = [
 	"MAINSAIL_P2P_DEVELOPMENT_MODE_ENABLED",
 	"MAINSAIL_P2P_HOST",
 	"MAINSAIL_P2P_LOG_LEVEL",
+	"MAINSAIL_P2P_MAX_CONNECTIONS",
 	"MAINSAIL_P2P_MAX_PEERS_BROADCAST",
 	"MAINSAIL_P2P_MAX_PEERS_SAME_SUBNET",
 	"MAINSAIL_P2P_MIN_NETWORK_REACH",

packages/p2p/source/defaults.ts

@@ -10,6 +10,7 @@ export const defaults = {
 		enabled: Environment.isTrue(EnvironmentVariables.MAINSAIL_P2P_DEVELOPMENT_MODE_ENABLED),
 	},
 	getBlocksTimeout: 30_000,
+	maxConnections: Environment.get(EnvironmentVariables.MAINSAIL_P2P_MAX_CONNECTIONS, 500),
 	maxPeersBroadcast: Environment.get(EnvironmentVariables.MAINSAIL_P2P_MAX_PEERS_BROADCAST, 4),
 	maxSameSubnetPeers: Environment.get(EnvironmentVariables.MAINSAIL_P2P_MAX_PEERS_SAME_SUBNET, 5),
 	minimumNetworkReach: Environment.get(EnvironmentVariables.MAINSAIL_P2P_MIN_NETWORK_REACH, 20),

packages/p2p/source/hapi-nes/listener.test.ts

@@ -66,6 +66,43 @@ describeSkip("Listener", ({ it, spy, beforeEach, assert, nock, each }) => {
 		await server.stop();
 	});
 
+	it("limits number of connections per ip", async () => {
+		const server = Hapi.server();
+		await server.register({ plugin: plugin, options: { maxConnectionsPerIp: 1 } });
+
+		await server.start();
+		const client = new Client("http://127.0.0.1:" + server.info.port);
+		await client.connect();
+
+		const client2 = new Client("http://127.0.0.1:" + server.info.port);
+		client2.onError = Hoek.ignore;
+
+		await assert.rejects(() => client2.connect());
+
+		await client.disconnect();
+		await client2.disconnect();
+		await server.stop();
+	});
+
+	it("exempts whitelisted ips from the per-ip connection limit", async () => {
+		const server = Hapi.server();
+		await server.register({
+			plugin: plugin,
+			options: { maxConnectionsPerIp: 1, maxConnectionsPerIpWhitelist: ["127.0.0.1", "::ffff:127.0.0.1", "::1"] },
+		});
+
+		await server.start();
+		const client = new Client("http://127.0.0.1:" + server.info.port);
+		await client.connect();
+
+		const client2 = new Client("http://127.0.0.1:" + server.info.port);
+		await client2.connect();
+
+		await client.disconnect();
+		await client2.disconnect();
+		await server.stop();
+	});
+
 	it("rejects unknown origin", async () => {
 		const server = Hapi.server();
 		await server.register({ plugin: plugin, options: { origin: ["http://127.0.0.1:12345"] } });

packages/p2p/source/hapi-nes/listener.ts

@@ -22,6 +22,7 @@ export class Listener {
 	private _heartbeat;
 	private _beatTimeout;
 	private _wss;
+	private _connectionsPerIp;
 
 	public constructor(server, settings) {
 		this._server = server;
@@ -31,6 +32,7 @@ export class Listener {
 		this._heartbeat = null;
 		this._beatTimeout = null;
 		this._stopped = false;
+		this._connectionsPerIp = new Set();
 
 		// WebSocket listener
 
@@ -50,7 +52,8 @@ export class Listener {
 
 			if (
 				this._stopped ||
-				(this._settings.maxConnections && this._sockets.length() >= this._settings.maxConnections)
+				(this._settings.maxConnections && this._sockets.length() >= this._settings.maxConnections) ||
+				this._hasExceededMaxConnectionsPerIp(req.socket.remoteAddress)
 			) {
 				return ws.close();
 			}
@@ -127,9 +130,11 @@ export class Listener {
 		const socket = new Socket(ws, req, this);
 
 		this._sockets.add(socket);
+		this._trackConnection(socket.info.remoteAddress);
 
 		ws.once("close", async (code, message) => {
 			this._sockets.remove(socket);
+			this._untrackConnection(socket.info.remoteAddress);
 
 			if (this._settings.onDisconnection) {
 				this._settings.onDisconnection(socket);
@@ -138,6 +143,26 @@ export class Listener {
 			socket._removed.attend();
 		});
 	}
+
+	private _hasExceededMaxConnectionsPerIp(ip) {
+		return this._connectionsPerIp.has(ip);
+	}
+
+	private _trackConnection(ip) {
+		if (ip === undefined) {
+			return;
+		}
+
+		this._connectionsPerIp.set(ip);
+	}
+
+	private _untrackConnection(ip) {
+		if (ip === undefined) {
+			return;
+		}
+
+		this._connectionsPerIp.delete(ip);
+	}
 }
 
 // Sockets manager

packages/p2p/source/service-provider.test.ts

@@ -124,6 +124,7 @@ describe<{
 
 		assert.array(result.value.blacklist);
 		assert.number(result.value.getBlocksTimeout);
+		assert.number(result.value.maxConnections);
 		assert.number(result.value.maxPeersBroadcast);
 		assert.number(result.value.maxSameSubnetPeers);
 		assert.number(result.value.minimumNetworkReach);
@@ -243,6 +244,24 @@ describe<{
 		assert.equal(result.error?.message, '"maxSameSubnetPeers" must be a number');
 	});
 
+	it("should parse process.env.MAINSAIL_P2P_MAX_CONNECTIONS", async ({ serviceProvider }) => {
+		process.env.MAINSAIL_P2P_MAX_CONNECTIONS = "2500";
+
+		const result = serviceProvider.configSchema().validate(await importDefaults());
+
+		assert.undefined(result.error);
+		assert.equal(result.value.maxConnections, 2500);
+	});
+
+	it("should throw if process.env.MAINSAIL_P2P_MAX_CONNECTIONS is not number", async ({ serviceProvider }) => {
+		process.env.MAINSAIL_P2P_MAX_CONNECTIONS = "dummy";
+
+		const result = serviceProvider.configSchema().validate(await importDefaults());
+
+		assert.defined(result.error);
+		assert.equal(result.error?.message, '"maxConnections" must be a number');
+	});
+
 	it("should parse process.env.MAINSAIL_P2P_MAX_PEERS_BROADCAST", async ({ serviceProvider }) => {
 		process.env.MAINSAIL_P2P_MAX_PEERS_BROADCAST = "10";
 

packages/p2p/source/service-provider.ts

@@ -86,6 +86,7 @@ export class ServiceProvider extends Providers.ServiceProvider {
 			disableDiscovery: Joi.bool(),
 			getBlocksTimeout: Joi.number().integer().min(0).required(),
 			ignoreMinimumNetworkReach: Joi.bool(),
+			maxConnections: Joi.number().integer().min(1).required(),
 			maxPeersBroadcast: Joi.number().integer().min(0).required(),
 			maxSameSubnetPeers: Joi.number().integer().min(0).required(),
 			minimumNetworkReach: Joi.number().integer().min(0).required(),

packages/p2p/source/socket-server/server.ts

@@ -2,7 +2,7 @@ import type { Contracts } from "@mainsail/contracts";
 
 import { Plugin, Server as HapiServer, ServerInjectOptions, ServerInjectResponse, ServerRoute } from "@hapi/hapi";
 import { Identifiers } from "@mainsail/constants";
-import { inject, injectable, multiInject } from "@mainsail/container";
+import { inject, injectable, multiInject, tagged } from "@mainsail/container";
 import { ensureError } from "@mainsail/utils";
 
 import { constants } from "../constants.js";
@@ -26,6 +26,10 @@ export class Server implements Contracts.P2P.Server {
 	@inject(Identifiers.Services.Log.Service)
 	private readonly logger!: Contracts.Kernel.Logger;
 
+	@inject(Identifiers.ServiceProvider.Configuration)
+	@tagged("plugin", "p2p")
+	private readonly configuration!: Contracts.Kernel.PluginConfiguration;
+
 	@multiInject(Identifiers.P2P.Routes)
 	private readonly routes!: Contracts.P2P.Route[];
 
@@ -43,6 +47,7 @@ export class Server implements Contracts.P2P.Server {
 		this.server.app = this.app;
 		await this.server.register({
 			options: {
+				maxConnections: this.configuration.getRequired<number | false>("maxConnections"),
 				maxPayload: constants.MAX_PAYLOAD_SERVER,
 			},
 			plugin: hapiNesPlugin,