Merge branch 'hotfix/2.3.2' of https://github.com/LiskHQ/lisk-sdk

elements/lisk-elements/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "lisk-elements",
-	"version": "2.4.0",
+	"version": "2.4.1",
 	"description": "Elements for building blockchain applications in the Lisk network",
 	"author": "Lisk Foundation <admin@lisk.io>, lightcurve GmbH <admin@lightcurve.io>",
 	"license": "Apache-2.0",

elements/lisk-p2p/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@liskhq/lisk-p2p",
-	"version": "0.3.1",
+	"version": "0.3.2",
 	"description": "Unstructured P2P library for use with Lisk-related software",
 	"author": "Lisk Foundation <admin@lisk.io>, lightcurve GmbH <admin@lightcurve.io>",
 	"license": "Apache-2.0",
@@ -50,8 +50,8 @@
 		"@liskhq/lisk-cryptography": "2.3.0",
 		"lodash.shuffle": "4.2.0",
 		"semver": "5.6.0",
-		"socketcluster-client": "14.3.0",
-		"socketcluster-server": "14.5.0",
+		"socketcluster-client": "14.3.1",
+		"socketcluster-server": "14.6.0",
 		"validator": "10.11.0"
 	},
 	"devDependencies": {

elements/lisk-p2p/src/disconnect_status_codes.ts

@@ -35,8 +35,10 @@ export const INCOMPATIBLE_PEER_CODE = 4104;
 export const INCOMPATIBLE_PEER_UNKNOWN_REASON =
 	'Peer is incompatible with the node for unknown reasons';
 
-// First case to follow HTTP status codes
 export const FORBIDDEN_CONNECTION = 4403;
 export const FORBIDDEN_CONNECTION_REASON = 'Peer is not allowed to connect';
 
+export const DUPLICATE_CONNECTION = 4404;
+export const DUPLICATE_CONNECTION_REASON = 'Peer has a duplicate connection';
+
 export const EVICTED_PEER_CODE = 4418;

elements/lisk-p2p/src/p2p.ts

@@ -29,6 +29,8 @@ import { REMOTE_RPC_GET_PEERS_LIST } from './peer';
 import { PeerBook } from './peer_directory';
 
 import {
+	DUPLICATE_CONNECTION,
+	DUPLICATE_CONNECTION_REASON,
 	FORBIDDEN_CONNECTION,
 	FORBIDDEN_CONNECTION_REASON,
 	INCOMPATIBLE_PEER_CODE,
@@ -386,9 +388,10 @@ export class P2P extends EventEmitter {
 					}
 				} else {
 					this._peerBook.addPeer(detailedPeerInfo);
+					// Re-emit the message to allow it to bubble up the class hierarchy.
+					// Only emit event when a peer is discovered for the first time.
+					this.emit(EVENT_DISCOVERED_PEER, detailedPeerInfo);
 				}
-				// Re-emit the message to allow it to bubble up the class hierarchy.
-				this.emit(EVENT_DISCOVERED_PEER, detailedPeerInfo);
 			}
 		};
 
@@ -728,7 +731,13 @@ export class P2P extends EventEmitter {
 
 				const existingPeer = this._peerPool.getPeer(peerId);
 
-				if (!existingPeer) {
+				if (existingPeer) {
+					this._disconnectSocketDueToFailedHandshake(
+						socket,
+						DUPLICATE_CONNECTION,
+						DUPLICATE_CONNECTION_REASON,
+					);
+				} else {
 					this._peerPool.addInboundPeer(incomingPeerInfo, socket);
 					this.emit(EVENT_NEW_INBOUND_PEER, incomingPeerInfo);
 					this.emit(EVENT_NEW_PEER, incomingPeerInfo);

elements/lisk-p2p/src/peer_pool.ts

@@ -60,6 +60,7 @@ import {
 	InboundPeer,
 	OutboundPeer,
 	Peer,
+	PeerConfig,
 } from './peer';
 import { getUniquePeersbyIp } from './peer_selection';
 import { constructPeerIdFromPeerInfo } from './utils';
@@ -188,11 +189,24 @@ export class PeerPool extends EventEmitter {
 	private readonly _peerSelectForConnection: P2PPeerSelectionForConnectionFunction;
 	private readonly _sendPeerLimit: number;
 	private readonly _outboundShuffleIntervalId: NodeJS.Timer | undefined;
+	private readonly _peerConfig: PeerConfig;
 
 	public constructor(peerPoolConfig: PeerPoolConfig) {
 		super();
 		this._peerMap = new Map();
 		this._peerPoolConfig = peerPoolConfig;
+		this._peerConfig = {
+			connectTimeout: this._peerPoolConfig.connectTimeout,
+			ackTimeout: this._peerPoolConfig.ackTimeout,
+			wsMaxMessageRate: this._peerPoolConfig.wsMaxMessageRate,
+			wsMaxMessageRatePenalty: this._peerPoolConfig.wsMaxMessageRatePenalty,
+			maxPeerDiscoveryResponseLength: this._peerPoolConfig
+				.maxPeerDiscoveryResponseLength,
+			rateCalculationInterval: this._peerPoolConfig.rateCalculationInterval,
+			wsMaxPayload: this._peerPoolConfig.wsMaxPayload,
+			maxPeerInfoSize: this._peerPoolConfig.maxPeerInfoSize,
+			secret: this._peerPoolConfig.secret,
+		};
 		this._peerSelectForSend = peerPoolConfig.peerSelectionForSend;
 		this._peerSelectForRequest = peerPoolConfig.peerSelectionForRequest;
 		this._peerSelectForConnection = peerPoolConfig.peerSelectionForConnection;
@@ -310,6 +324,10 @@ export class PeerPool extends EventEmitter {
 		return this._nodeInfo;
 	}
 
+	public get peerConfig(): PeerConfig {
+		return { ...this._peerConfig };
+	}
+
 	public async request(packet: P2PRequestPacket): Promise<P2PResponsePacket> {
 		// This function can be customized so we should pass as much info as possible.
 		const selectedPeers = this._peerSelectForRequest({
@@ -428,18 +446,9 @@ export class PeerPool extends EventEmitter {
 			this._evictPeer(InboundPeer);
 		}
 
-		const peerConfig = {
-			connectTimeout: this._peerPoolConfig.connectTimeout,
-			ackTimeout: this._peerPoolConfig.ackTimeout,
-			wsMaxMessageRate: this._peerPoolConfig.wsMaxMessageRate,
-			wsMaxMessageRatePenalty: this._peerPoolConfig.wsMaxMessageRatePenalty,
-			maxPeerDiscoveryResponseLength: this._peerPoolConfig
-				.maxPeerDiscoveryResponseLength,
-			maxPeerInfoSize: this._peerPoolConfig.maxPeerInfoSize,
-			rateCalculationInterval: this._peerPoolConfig.rateCalculationInterval,
-			secret: this._peerPoolConfig.secret,
-		};
-		const peer = new InboundPeer(peerInfo, socket, peerConfig);
+		const peer = new InboundPeer(peerInfo, socket, {
+			...this._peerConfig,
+		});
 
 		// Throw an error because adding a peer multiple times is a common developer error which is very difficult to identify and debug.
 		if (this._peerMap.has(peer.id)) {
@@ -461,20 +470,7 @@ export class PeerPool extends EventEmitter {
 			return existingPeer;
 		}
 
-		const peerConfig = {
-			connectTimeout: this._peerPoolConfig.connectTimeout,
-			ackTimeout: this._peerPoolConfig.ackTimeout,
-			banTime: this._peerPoolConfig.peerBanTime,
-			wsMaxMessageRate: this._peerPoolConfig.wsMaxMessageRate,
-			wsMaxMessageRatePenalty: this._peerPoolConfig.wsMaxMessageRatePenalty,
-			maxPeerDiscoveryResponseLength: this._peerPoolConfig
-				.maxPeerDiscoveryResponseLength,
-			rateCalculationInterval: this._peerPoolConfig.rateCalculationInterval,
-			wsMaxPayload: this._peerPoolConfig.wsMaxPayload,
-			maxPeerInfoSize: this._peerPoolConfig.maxPeerInfoSize,
-			secret: this._peerPoolConfig.secret,
-		};
-		const peer = new OutboundPeer(peerInfo, peerConfig);
+		const peer = new OutboundPeer(peerInfo, { ...this._peerConfig });
 
 		this._peerMap.set(peer.id, peer);
 		this._bindHandlersToPeer(peer);

elements/lisk-p2p/src/validation.ts

@@ -92,11 +92,19 @@ export const validatePeerInfoSchema = (rawPeerInfo: unknown): P2PPeerInfo => {
 		!protocolPeer.wsPort ||
 		!validatePeerAddress(protocolPeer.ip, protocolPeer.wsPort)
 	) {
-		throw new InvalidPeerError(`Invalid peer ip or port`);
+		throw new InvalidPeerError(
+			`Invalid peer ip or port for peer with ip: ${
+				protocolPeer.ip
+			} and wsPort ${protocolPeer.wsPort}`,
+		);
 	}
 
 	if (!protocolPeer.version || !isValidVersion(protocolPeer.version)) {
-		throw new InvalidPeerError(`Invalid peer version`);
+		throw new InvalidPeerError(
+			`Invalid peer version for peer with ip: ${protocolPeer.ip}, wsPort ${
+				protocolPeer.wsPort
+			} and version ${protocolPeer.version}`,
+		);
 	}
 
 	const version = protocolPeer.version;

elements/lisk-p2p/test/integration/p2p.ts

@@ -14,7 +14,7 @@
  */
 
 import { expect } from 'chai';
-import { P2P, EVENT_REMOVE_PEER } from '../../src/index';
+import { P2P, EVENT_REMOVE_PEER, EVENT_CLOSE_OUTBOUND } from '../../src/index';
 import { wait } from '../utils/helpers';
 import { platform } from 'os';
 import {
@@ -25,7 +25,7 @@ import {
 	P2PPeerSelectionForRequestInput,
 	P2PPeerSelectionForConnectionInput,
 } from '../../src/p2p_types';
-import { InboundPeer } from '../../src/peer';
+import { InboundPeer, OutboundPeer, ConnectionState } from '../../src/peer';
 import { SCServerSocket } from 'socketcluster-server';
 import * as url from 'url';
 import cloneDeep = require('lodash.clonedeep');
@@ -121,9 +121,9 @@ describe('Integration tests for P2P library', () => {
 					seedPeers,
 					wsEngine: 'ws',
 					connectTimeout: 100,
-					ackTimeout: 200,
+					ackTimeout: 100,
 					peerBanTime: 100,
-					populatorInterval: POPULATOR_INTERVAL,
+					populatorInterval: 100,
 					maxOutboundConnections: DEFAULT_MAX_OUTBOUND_CONNECTIONS,
 					maxInboundConnections: DEFAULT_MAX_INBOUND_CONNECTIONS,
 					nodeInfo: {
@@ -148,7 +148,7 @@ describe('Integration tests for P2P library', () => {
 		describe('Peer discovery', () => {
 			it('should discover all peers in the network after a few cycles of discovery', async () => {
 				// Wait for a few cycles of discovery.
-				await wait(POPULATOR_INTERVAL * 10);
+				await wait(POPULATOR_INTERVAL * 15);
 
 				for (let p2p of p2pNodeList) {
 					const peerPorts = p2p
@@ -941,6 +941,64 @@ describe('Integration tests for P2P library', () => {
 				);
 			});
 		});
+
+		describe('Disconnect duplicate peers', () => {
+			let firstP2PNodeCloseEvents: Array<any> = [];
+			let firstPeerCloseEvents: Array<any> = [];
+			let firstPeerErrors: Array<any> = [];
+			let firstPeerDuplicate: OutboundPeer;
+			let firstP2PNode: P2P;
+			let existingPeer: InboundPeer;
+
+			beforeEach(async () => {
+				firstP2PNode = p2pNodeList[0];
+				firstPeerCloseEvents = [];
+				existingPeer = firstP2PNode['_peerPool'].getPeers(
+					InboundPeer,
+				)[0] as InboundPeer;
+				firstPeerDuplicate = new OutboundPeer(
+					existingPeer.peerInfo,
+					firstP2PNode['_peerPool'].peerConfig,
+				);
+
+				firstPeerDuplicate.on(EVENT_CLOSE_OUTBOUND, (event: any) => {
+					firstPeerCloseEvents.push(event);
+				});
+
+				try {
+					// This will create a connection.
+					await firstPeerDuplicate.applyNodeInfo(firstP2PNode.nodeInfo);
+				} catch (error) {
+					firstPeerErrors.push(error);
+				}
+
+				firstP2PNode.on(EVENT_CLOSE_OUTBOUND, event => {
+					firstP2PNodeCloseEvents.push(event);
+				});
+				await wait(100);
+			});
+			afterEach(() => {
+				firstPeerDuplicate.removeAllListeners(EVENT_CLOSE_OUTBOUND);
+				firstP2PNode.removeAllListeners(EVENT_CLOSE_OUTBOUND);
+				firstPeerDuplicate.disconnect();
+			});
+
+			// Simulate legacy behaviour where the node tries to connect back to an inbound peer.
+			it('should remove a peer if they try to connect but they are already connected', async () => {
+				expect(firstPeerErrors).to.have.length(1);
+				expect(firstPeerErrors[0])
+					.to.have.property('name')
+					.which.equals('BadConnectionError');
+				expect(firstPeerErrors[0])
+					.to.have.property('name')
+					.which.equals('BadConnectionError');
+				expect(firstPeerDuplicate)
+					.to.have.property('state')
+					.which.equals(ConnectionState.CLOSED);
+				// Disconnecting our new outbound socket should not cause the existing inbound peer instance to be removed.
+				expect(firstP2PNodeCloseEvents).to.be.empty;
+			});
+		});
 	});
 
 	describe('Fully connected network: Message rate checks', () => {