fix(net): robustness fixes for multiple issues

framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java

@@ -157,7 +157,7 @@ private void processMessage(PeerConnection peer, byte[] data) {
 
       peer.getPeerStatistics().messageStatistics.addTcpInMessage(msg);
       if (PeerConnection.needToLog(msg)) {
-        logger.info("Receive message from  peer: {}, {}", peer.getInetSocketAddress(), msg);
+        logger.info("Receive message from peer: {}, {}", peer.getInetSocketAddress(), msg);
       }
 
       switch (type) {
@@ -300,13 +300,8 @@ private void processException(PeerConnection peer, TronMessage msg, Exception ex
           code = Protocol.ReasonCode.UNKNOWN;
           break;
       }
-      if (type.equals(P2pException.TypeEnum.BAD_MESSAGE)) {
-        logger.error("Message from {} process failed, {} \n type: ({})",
-            peer.getInetSocketAddress(), msg, type, ex);
-      } else {
-        logger.warn("Message from {} process failed, {} \n type: ({}), detail: {}",
-            peer.getInetSocketAddress(), msg, type, ex.getMessage());
-      }
+      logger.warn("Message from {} process failed, {} \n type: ({}), detail: {}",
+          peer.getInetSocketAddress(), msg, type, ex.getMessage());
     } else {
       code = Protocol.ReasonCode.UNKNOWN;
       logger.warn("Message from {} process failed, {}",

framework/src/main/java/org/tron/core/net/message/handshake/HelloMessage.java

@@ -10,9 +10,11 @@
 import org.tron.core.ChainBaseManager;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.config.args.Args;
+import org.tron.core.exception.P2pException;
 import org.tron.core.net.message.MessageTypes;
 import org.tron.core.net.message.TronMessage;
 import org.tron.p2p.discover.Node;
+import org.tron.p2p.utils.NetUtil;
 import org.tron.program.Version;
 import org.tron.protos.Discover.Endpoint;
 import org.tron.protos.Protocol;
@@ -31,6 +33,9 @@ public HelloMessage(byte type, byte[] rawData) throws Exception {
   public HelloMessage(byte[] data) throws Exception {
     super(MessageTypes.P2P_HELLO.asByte(), data);
     this.helloMessage = Protocol.HelloMessage.parseFrom(data);
+    if (!valid()) {
+      throw new P2pException(P2pException.TypeEnum.BAD_MESSAGE, "invalid hello message");
+    }
   }
 
   public HelloMessage(Node from, long timestamp, ChainBaseManager chainBaseManager) {
@@ -124,7 +129,6 @@ public String toString() {
     StringBuilder builder = new StringBuilder();
 
     builder.append(super.toString())
-            .append("from: ").append(getFrom().getPreferInetSocketAddress()).append("\n")
             .append("timestamp: ").append(getTimestamp()).append("\n")
             .append("headBlockId: ").append(getHeadBlockId().getString()).append("\n")
             .append("nodeType: ").append(helloMessage.getNodeType()).append("\n")
@@ -156,6 +160,10 @@ public Protocol.HelloMessage getInstance() {
   }
 
   public boolean valid() {
+    if (helloMessage.hasFrom() && !NetUtil.validNode(getFrom())) {
+      return false;
+    }
+
     byte[] genesisBlockByte = this.helloMessage.getGenesisBlockId().getHash().toByteArray();
     if (genesisBlockByte.length != Sha256Hash.LENGTH) {
       return false;

framework/src/main/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandler.java

@@ -5,7 +5,6 @@
 import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.NoSuchElementException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -57,7 +56,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
     }
 
     while (!peer.getSyncBlockToFetch().isEmpty()) {
-      if (peer.getSyncBlockToFetch().peekLast().equals(blockIdWeGet.peekFirst())) {
+      if (blockIdWeGet.peekFirst().equals(peer.getSyncBlockToFetch().peekLast())) {
         break;
       }
       peer.getSyncBlockToFetch().pollLast();
@@ -69,22 +68,18 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
     peer.getSyncBlockToFetch().addAll(blockIdWeGet);
 
     synchronized (tronNetDelegate.getBlockLock()) {
-      try {
-        BlockId blockId = null;
-        while (!peer.getSyncBlockToFetch().isEmpty() && tronNetDelegate
-                .containBlock(peer.getSyncBlockToFetch().peek())) {
-          blockId = peer.getSyncBlockToFetch().pop();
+      Deque<BlockId> toFetch = peer.getSyncBlockToFetch();
+      BlockId blockId = null;
+      BlockId next;
+      while ((next = toFetch.peek()) != null && tronNetDelegate.containBlock(next)) {
+        if (toFetch.remove(next)) {
+          blockId = next;
           peer.setBlockBothHave(blockId);
         }
-        if (blockId != null) {
-          logger.info("Block {} from {} is processed",
-              blockId.getString(), peer.getInetAddress());
-        }
-      } catch (NoSuchElementException e) {
-        logger.warn("Process ChainInventoryMessage failed, peer {}, isDisconnect:{}",
-                peer.getInetAddress(), peer.isDisconnect());
-        peer.setFetchAble(true);
-        return;
+      }
+      if (blockId != null) {
+        logger.info("Block {} from {} is processed",
+            blockId.getString(), peer.getInetAddress());
       }
     }
 

framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java

@@ -138,7 +138,7 @@ private void check(PeerConnection peer, TransactionsMessage msg) throws P2pExcep
       Item item = new Item(id, InventoryType.TRX);
       if (!peer.getAdvInvRequest().containsKey(item)) {
         throw new P2pException(TypeEnum.BAD_MESSAGE,
-            "trx: " + msg.getMessageId() + " without request.");
+            "trx: " + id + " without request.");
       }
       if (trx.getRawData().getContractCount() < 1) {
         throw new P2pException(TypeEnum.BAD_TRX,
@@ -216,4 +216,4 @@ public TrxEvent(PeerConnection peer, TransactionMessage msg) {
       this.time = System.currentTimeMillis();
     }
   }
-}
+}

framework/src/main/java/org/tron/core/net/service/effective/EffectiveCheckService.java

@@ -127,8 +127,9 @@ private void findEffectiveNode() {
         // Connection attempt cancelled by user
         cur = null;
       } else if (!future.isSuccess()) {
-        // You might get a NullPointerException here because the future might not be completed yet.
-        logger.warn("Connect to chosen peer {} fail, cause:{}", cur, future.cause().getMessage());
+        Throwable cause = future.cause();
+        logger.warn("Connect to chosen peer {} fail, cause:{}", cur,
+            cause == null ? "unknown" : cause.getMessage());
         future.channel().close();
         cur = null;
         triggerNext();

framework/src/main/java/org/tron/core/net/service/handshake/HandshakeService.java

@@ -4,7 +4,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import org.tron.common.utils.ByteArray;
 import org.tron.core.ChainBaseManager;
 import org.tron.core.ChainBaseManager.NodeType;
 import org.tron.core.config.args.Args;
@@ -41,27 +40,15 @@ public void processHelloMessage(PeerConnection peer, HelloMessage msg) {
       return;
     }
 
-    TronNetService.getP2pService().updateNodeId(peer.getChannel(), msg.getFrom().getHexId());
+    if (msg.getHelloMessage().hasFrom()) {
+      TronNetService.getP2pService().updateNodeId(peer.getChannel(), msg.getFrom().getHexId());
+    }
     if (peer.isDisconnect()) {
       logger.info("Duplicate Peer {}", peer.getInetSocketAddress());
       peer.disconnect(ReasonCode.DUPLICATE_PEER);
       return;
     }
 
-    if (!msg.valid()) {
-      logger.warn("Peer {} invalid hello message parameters, GenesisBlockId: {}, SolidBlockId: {}, "
-              + "HeadBlockId: {}, address: {}, sig: {}, codeVersion: {}",
-          peer.getInetSocketAddress(),
-          ByteArray.toHexString(msg.getInstance().getGenesisBlockId().getHash().toByteArray()),
-          ByteArray.toHexString(msg.getInstance().getSolidBlockId().getHash().toByteArray()),
-          ByteArray.toHexString(msg.getInstance().getHeadBlockId().getHash().toByteArray()),
-          msg.getInstance().getAddress().toByteArray().length,
-          msg.getInstance().getSignature().toByteArray().length,
-          msg.getInstance().getCodeVersion().toByteArray().length);
-      peer.disconnect(ReasonCode.INCOMPATIBLE_PROTOCOL);
-      return;
-    }
-
     peer.setAddress(msg.getHelloMessage().getAddress());
 
     if (!relayService.checkHelloMessage(msg, peer.getChannel())) {

framework/src/main/java/org/tron/core/services/ratelimiter/strategy/QpsStrategy.java

@@ -22,7 +22,7 @@ public QpsStrategy(String paramString) {
   @Override
   protected Map<String, ParamItem> defaultParam() {
     Map<String, ParamItem> map = new HashMap<>();
-    map.put(STRATEGY_PARAM_QPS, new ParamItem(Double.class, DEFAULT_QPS));
+    map.put(STRATEGY_PARAM_QPS, new ParamItem(Double.class, (double) DEFAULT_QPS));
     return map;
   }
 
@@ -34,4 +34,4 @@ public boolean acquire() {
     rateLimiter.acquire();
     return true;
   }
-}
+}

framework/src/test/java/org/tron/core/net/P2pEventHandlerImplTest.java

@@ -18,6 +18,7 @@
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
+import org.tron.core.net.message.MessageTypes;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.adv.FetchInvDataMessage;
 import org.tron.core.net.message.adv.InventoryMessage;
@@ -34,6 +35,19 @@ public static void init() throws Exception {
         TestConstants.TEST_CONF);
   }
 
+  @Test
+  public void testInvalidHelloRejectedBeforeLoggingAndHandshake() throws Exception {
+    PeerConnection peer = mock(PeerConnection.class);
+    P2pEventHandlerImpl handler = new P2pEventHandlerImpl();
+    Method method = handler.getClass()
+        .getDeclaredMethod("processMessage", PeerConnection.class, byte[].class);
+    method.setAccessible(true);
+    method.invoke(handler, peer, new byte[]{MessageTypes.P2P_HELLO.asByte()});
+
+    verify(peer).disconnect(Protocol.ReasonCode.BAD_PROTOCOL);
+    Mockito.verify(peer, Mockito.never()).getPeerStatistics();
+  }
+
   @Test
   public void testProcessInventoryMessage() throws Exception {
     CommonParameter parameter = CommonParameter.getInstance();

framework/src/test/java/org/tron/core/net/messagehandler/ChainInventoryMsgHandlerTest.java

@@ -1,21 +1,27 @@
 package org.tron.core.net.messagehandler;
 
 import java.util.ArrayList;
+import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.tron.common.TestConstants;
 import org.tron.common.utils.Pair;
+import org.tron.common.utils.ReflectUtils;
+import org.tron.common.utils.Sha256Hash;
 import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.core.config.Parameter.NetConstants;
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
+import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.keepalive.PingMessage;
 import org.tron.core.net.message.sync.ChainInventoryMessage;
 import org.tron.core.net.peer.PeerConnection;
+import org.tron.core.net.service.sync.SyncService;
 
 public class ChainInventoryMsgHandlerTest {
 
@@ -34,6 +40,38 @@ public static void destroy() {
   private ChainInventoryMessage msg = new ChainInventoryMessage(new ArrayList<>(), 0L);
   private List<BlockId> blockIds = new ArrayList<>();
 
+  @Test
+  public void testQueueChangedDuringProcessing() throws Exception {
+    ChainInventoryMsgHandler handler = new ChainInventoryMsgHandler();
+    TronNetDelegate delegate = Mockito.mock(TronNetDelegate.class);
+    SyncService syncService = Mockito.mock(SyncService.class);
+    ReflectUtils.setFieldValue(handler, "tronNetDelegate", delegate);
+    ReflectUtils.setFieldValue(handler, "syncService", syncService);
+
+    BlockId parent = new BlockId(Sha256Hash.ZERO_HASH, 0);
+    BlockId next = new BlockId(Sha256Hash.ZERO_HASH, 1);
+    PeerConnection peer = Mockito.mock(PeerConnection.class);
+    Deque<BlockId> toFetch = Mockito.mock(Deque.class);
+    LinkedList<BlockId> requested = new LinkedList<>();
+    requested.add(parent);
+    Mockito.when(peer.getSyncChainRequested())
+        .thenReturn(new Pair<>(requested, System.currentTimeMillis()));
+    Mockito.when(peer.getSyncBlockToFetch()).thenReturn(toFetch);
+    Mockito.when(toFetch.peek()).thenReturn(next, (BlockId) null);
+    Mockito.when(toFetch.isEmpty()).thenReturn(false, true);
+    Mockito.when(toFetch.peekLast()).thenReturn(null);
+    Mockito.when(delegate.getHeadBlockId()).thenReturn(parent);
+    Mockito.when(delegate.getBlockLock()).thenReturn(new Object());
+    Mockito.when(delegate.containBlock(next)).thenReturn(true);
+    Mockito.when(toFetch.remove(next)).thenReturn(false);
+
+    handler.processMessage(peer, new ChainInventoryMessage(
+        java.util.Arrays.asList(parent, next), 0L));
+
+    Mockito.verify(toFetch).pollLast();
+    Mockito.verify(peer, Mockito.never()).setBlockBothHave(Mockito.any());
+  }
+
   @Test
   public void testProcessMessage() throws Exception {
     try {

framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java

@@ -337,6 +337,28 @@ public void testDuplicateTransactionRejected() throws Exception {
     }
   }
 
+  @Test
+  public void testUnrequestedTransactionReportsTransactionId() throws Exception {
+    TransactionsMsgHandler handler = new TransactionsMsgHandler();
+    try {
+      PeerConnection peer = Mockito.mock(PeerConnection.class);
+      Mockito.when(peer.getAdvInvRequest()).thenReturn(new ConcurrentHashMap<>());
+      TransactionsMessage msg = buildTransferMessage(1);
+      Protocol.Transaction trx = msg.getTransactions().getTransactions(0);
+      String transactionId = new TransactionMessage(trx).getMessageId().toString();
+
+      try {
+        handler.processMessage(peer, msg);
+        Assert.fail("Expected an unrequested transaction to be rejected");
+      } catch (P2pException e) {
+        Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
+        Assert.assertTrue(e.getMessage().contains(transactionId));
+      }
+    } finally {
+      handler.close();
+    }
+  }
+
   @Test
   public void testInvalidSigLength() throws Exception {
     TransactionsMsgHandler handler = new TransactionsMsgHandler();