fix(net): gate peer traffic on hello validation and enforce timeout

common/src/main/java/org/tron/core/config/Parameter.java

@@ -100,6 +100,7 @@ public class NodeConstant {
   public class NetConstants {
 
     public static final long SYNC_FETCH_BATCH_NUM = 2000;
+    public static final long HELLO_TIME_OUT = 10000L;
     public static final long ADV_TIME_OUT = 20000L;
     public static final long SYNC_TIME_OUT = 5000L;
     public static final long NET_MAX_TRX_PER_SECOND = 700L;

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

@@ -127,6 +127,18 @@ public void onMessage(Channel c, byte[] data) {
       return;
     }
 
+    if (data == null || data.length == 0) {
+      peerConnection.disconnect(Protocol.ReasonCode.BAD_PROTOCOL);
+      return;
+    }
+
+    if (peerConnection.getHelloMessageReceive() == null
+        && data[0] != MessageTypes.P2P_HELLO.asByte()
+        && data[0] != MessageTypes.P2P_DISCONNECT.asByte()) {
+      peerConnection.disconnect(Protocol.ReasonCode.BAD_PROTOCOL);
+      return;
+    }
+
     if (MessageTypes.PBFT_MSG.asByte() == data[0]) {
       PbftMessage message = null;
       try {

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

@@ -11,14 +11,18 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.net.message.MessageTypes;
 import org.tron.core.net.message.TronMessage;
+import org.tron.p2p.base.Constant;
 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;
 import org.tron.protos.Protocol.HelloMessage.Builder;
 
 public class HelloMessage extends TronMessage {
 
+  private static final int MAX_BYTE_SIZE = 200;
+
   @Getter
   private Protocol.HelloMessage helloMessage;
 
@@ -120,6 +124,10 @@ public Class<?> getAnswerMessage() {
 
   @Override
   public String toString() {
+    if (!valid()) {
+      return "P2P_HELLO: invalid hello message";
+    }
+
     StringBuilder builder = new StringBuilder();
 
     builder.append(super.toString())
@@ -155,6 +163,10 @@ public Protocol.HelloMessage getInstance() {
   }
 
   public boolean valid() {
+    if (!validEndPoint()) {
+      return false;
+    }
+
     byte[] genesisBlockByte = this.helloMessage.getGenesisBlockId().getHash().toByteArray();
     if (genesisBlockByte.length != Sha256Hash.LENGTH) {
       return false;
@@ -170,25 +182,46 @@ public boolean valid() {
       return false;
     }
 
-    int maxByteSize = 200;
     ByteString address = this.helloMessage.getAddress();
-    if (!address.isEmpty() && address.toByteArray().length > maxByteSize) {
+    if (!address.isEmpty() && address.toByteArray().length > MAX_BYTE_SIZE) {
       return false;
     }
 
     ByteString sig = this.helloMessage.getSignature();
-    if (!sig.isEmpty() && sig.toByteArray().length > maxByteSize) {
+    if (!sig.isEmpty() && sig.toByteArray().length > MAX_BYTE_SIZE) {
       return false;
     }
 
     ByteString codeVersion = this.helloMessage.getCodeVersion();
-    if (!codeVersion.isEmpty() && codeVersion.toByteArray().length > maxByteSize) {
+    if (!codeVersion.isEmpty() && codeVersion.toByteArray().length > MAX_BYTE_SIZE) {
       return false;
     }
 
     return true;
   }
 
+  public boolean validEndPoint() {
+    Endpoint from = this.helloMessage.getFrom();
+    ByteString ipv4 = from.getAddress();
+    ByteString ipv6 = from.getAddressIpv6();
+    if (from.getNodeId().size() != Constant.NODE_ID_LEN
+        || from.getPort() <= 0 || from.getPort() > 0xFFFF
+        || ipv4.size() > MAX_BYTE_SIZE || ipv6.size() > MAX_BYTE_SIZE) {
+      return false;
+    }
+    if (ipv4.isEmpty() && ipv6.isEmpty()) {
+      return false;
+    }
+    // Validate raw literals before getFrom() constructs a Node during logging.
+    if (!ipv4.isEmpty() && !NetUtil.validIpV4(ByteArray.toStr(ipv4.toByteArray()))) {
+      return false;
+    }
+    if (!ipv6.isEmpty() && !NetUtil.validIpV6(ByteArray.toStr(ipv6.toByteArray()))) {
+      return false;
+    }
+    return true;
+  }
+
   public static Endpoint getEndpointFromNode(Node node) {
     Endpoint.Builder builder = Endpoint.newBuilder()
         .setPort(node.getPort());

framework/src/main/java/org/tron/core/net/peer/PeerConnection.java

@@ -128,7 +128,7 @@ public class PeerConnection {
 
   @Setter
   @Getter
-  private HelloMessage helloMessageReceive;
+  private volatile HelloMessage helloMessageReceive;
 
   @Setter
   @Getter

framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java

@@ -56,6 +56,14 @@ public void statusCheck() {
         isDisconnected = true;
       }
 
+      if (!isDisconnected) {
+        isDisconnected = peer.getHelloMessageReceive() == null
+            && peer.getChannel().getStartTime() <= now - NetConstants.HELLO_TIME_OUT;
+        if (isDisconnected) {
+          logger.warn("Peer {} hello message timeout", peer.getInetAddress());
+        }
+      }
+
       if (!isDisconnected) {
         isDisconnected = peer.getAdvInvRequest().values().stream()
             .anyMatch(time -> time < now - NetConstants.ADV_TIME_OUT);

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

@@ -5,7 +5,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.config.args.Args;
 import org.tron.core.net.TronNetService;
@@ -45,27 +44,29 @@ public void processHelloMessage(PeerConnection peer, HelloMessage msg) {
       return;
     }
 
+    if (!msg.valid()) {
+      logger.warn("Peer {} invalid hello message parameters, genesisHashLength: {}, "
+              + "solidHashLength: {}, headHashLength: {}, address: {}, sig: {}, codeVersion: {}, "
+              + "endpointValid: {}",
+          peer.getInetSocketAddress(),
+          msg.getInstance().getGenesisBlockId().getHash().size(),
+          msg.getInstance().getSolidBlockId().getHash().size(),
+          msg.getInstance().getHeadBlockId().getHash().size(),
+          msg.getInstance().getAddress().size(),
+          msg.getInstance().getSignature().size(),
+          msg.getInstance().getCodeVersion().size(),
+          msg.validEndPoint());
+      peer.disconnect(ReasonCode.INCOMPATIBLE_PROTOCOL);
+      return;
+    }
+
     if (!Objects.equals(peer.getChannel().getNodeId(), msg.getFrom().getHexId())) {
       logger.warn("Peer {} sent an application HELLO with a different node ID",
           peer.getInetSocketAddress());
       peer.disconnect(ReasonCode.UNEXPECTED_IDENTITY);
       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/test/java/org/tron/core/net/P2pHelloGateTest.java

@@ -0,0 +1,202 @@
+package org.tron.core.net;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.tron.common.utils.Sha256Hash;
+import org.tron.consensus.pbft.message.PbftMessage;
+import org.tron.core.ChainBaseManager;
+import org.tron.core.ChainBaseManager.NodeType;
+import org.tron.core.capsule.BlockCapsule.BlockId;
+import org.tron.core.net.message.MessageTypes;
+import org.tron.core.net.message.PbftMessageFactory;
+import org.tron.core.net.message.TronMessageFactory;
+import org.tron.core.net.message.adv.InventoryMessage;
+import org.tron.core.net.message.base.DisconnectMessage;
+import org.tron.core.net.message.handshake.HelloMessage;
+import org.tron.core.net.message.keepalive.PingMessage;
+import org.tron.core.net.messagehandler.BlockMsgHandler;
+import org.tron.core.net.messagehandler.ChainInventoryMsgHandler;
+import org.tron.core.net.messagehandler.FetchInvDataMsgHandler;
+import org.tron.core.net.messagehandler.InventoryMsgHandler;
+import org.tron.core.net.messagehandler.PbftDataSyncHandler;
+import org.tron.core.net.messagehandler.PbftMsgHandler;
+import org.tron.core.net.messagehandler.SyncBlockChainMsgHandler;
+import org.tron.core.net.messagehandler.TransactionsMsgHandler;
+import org.tron.core.net.peer.PeerConnection;
+import org.tron.core.net.peer.PeerManager;
+import org.tron.core.net.service.handshake.HandshakeService;
+import org.tron.core.net.service.keepalive.KeepAliveService;
+import org.tron.core.net.service.statistics.NodeStatistics;
+import org.tron.core.net.service.statistics.PeerStatistics;
+import org.tron.p2p.connection.Channel;
+import org.tron.p2p.discover.Node;
+import org.tron.protos.Protocol.Inventory.InventoryType;
+import org.tron.protos.Protocol.ReasonCode;
+
+public class P2pHelloGateTest {
+
+  @InjectMocks
+  private P2pEventHandlerImpl handler;
+  @Mock
+  private HandshakeService handshakeService;
+  @Mock
+  private KeepAliveService keepAliveService;
+  @Mock
+  private SyncBlockChainMsgHandler syncBlockChainMsgHandler;
+  @Mock
+  private ChainInventoryMsgHandler chainInventoryMsgHandler;
+  @Mock
+  private InventoryMsgHandler inventoryMsgHandler;
+  @Mock
+  private FetchInvDataMsgHandler fetchInvDataMsgHandler;
+  @Mock
+  private BlockMsgHandler blockMsgHandler;
+  @Mock
+  private TransactionsMsgHandler transactionsMsgHandler;
+  @Mock
+  private PbftDataSyncHandler pbftDataSyncHandler;
+  @Mock
+  private PbftMsgHandler pbftMsgHandler;
+  @Mock
+  private PeerConnection peer;
+  @Mock
+  private Channel channel;
+
+  private AutoCloseable mocks;
+  private MockedStatic<PeerManager> peerManager;
+
+  @Before
+  public void setUp() {
+    mocks = MockitoAnnotations.openMocks(this);
+    peerManager = Mockito.mockStatic(PeerManager.class);
+    peerManager.when(() -> PeerManager.getPeerConnection(channel)).thenReturn(peer);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    peerManager.close();
+    mocks.close();
+  }
+
+  @Test
+  public void testRejectAllOtherMessageTypesBeforeParsing() {
+    try (MockedStatic<TronMessageFactory> tronFactory =
+             Mockito.mockStatic(TronMessageFactory.class);
+         MockedStatic<PbftMessageFactory> pbftFactory =
+             Mockito.mockStatic(PbftMessageFactory.class)) {
+      int rejected = 0;
+      for (MessageTypes type : MessageTypes.values()) {
+        if (type != MessageTypes.P2P_HELLO && type != MessageTypes.P2P_DISCONNECT) {
+          handler.onMessage(channel, new byte[] {type.asByte()});
+          rejected++;
+        }
+      }
+      verify(peer, times(rejected)).disconnect(ReasonCode.BAD_PROTOCOL);
+      tronFactory.verifyNoInteractions();
+      pbftFactory.verifyNoInteractions();
+    }
+    verify(peer, never()).getPeerStatistics();
+    verifyNoInteractions(handshakeService, keepAliveService, syncBlockChainMsgHandler,
+        chainInventoryMsgHandler, inventoryMsgHandler, fetchInvDataMsgHandler, blockMsgHandler,
+        transactionsMsgHandler, pbftDataSyncHandler, pbftMsgHandler);
+  }
+
+  @Test
+  public void testEmptyFramesAreRejected() {
+    handler.onMessage(channel, new byte[0]);
+    handler.onMessage(channel, null);
+    verify(peer, times(2)).disconnect(ReasonCode.BAD_PROTOCOL);
+  }
+
+  @Test
+  public void testUnknownPeerIsIgnored() {
+    peerManager.when(() -> PeerManager.getPeerConnection(channel)).thenReturn(null);
+    handler.onMessage(channel, new PingMessage().getSendBytes());
+    verifyNoInteractions(peer, keepAliveService);
+  }
+
+  @Test
+  public void testHelloIsDispatchedBeforeHandshake() throws Exception {
+    ChainBaseManager chain = mock(ChainBaseManager.class);
+    when(chain.getGenesisBlockId()).thenReturn(new BlockId());
+    when(chain.getSolidBlockId()).thenReturn(new BlockId());
+    when(chain.getHeadBlockId()).thenReturn(new BlockId());
+    when(chain.getNodeType()).thenReturn(NodeType.FULL);
+    HelloMessage hello = new HelloMessage(
+        new Node(new byte[64], "127.0.0.1", null, 18888), 0, chain);
+    when(peer.getPeerStatistics()).thenReturn(new PeerStatistics());
+
+    handler.onMessage(channel, hello.getSendBytes());
+
+    verify(handshakeService).processHelloMessage(eq(peer), any(HelloMessage.class));
+    verify(peer, never()).disconnect(any());
+  }
+
+  @Test
+  public void testDisconnectIsDispatchedBeforeHandshake() {
+    NodeStatistics stats = mock(NodeStatistics.class);
+    when(peer.getPeerStatistics()).thenReturn(new PeerStatistics());
+    when(peer.getP2pRateLimiter()).thenReturn(new P2pRateLimiter());
+    when(peer.getChannel()).thenReturn(channel);
+    when(peer.getNodeStatistics()).thenReturn(stats);
+
+    handler.onMessage(channel, new DisconnectMessage(ReasonCode.PEER_QUITING).getSendBytes());
+
+    verify(channel).close();
+    verify(stats).nodeDisconnectedRemote(ReasonCode.PEER_QUITING);
+    verify(peer, never()).disconnect(any());
+  }
+
+  @Test
+  public void testKeepAliveIsDispatchedAfterHello() throws Exception {
+    when(peer.getHelloMessageReceive()).thenReturn(mock(HelloMessage.class));
+    when(peer.getPeerStatistics()).thenReturn(new PeerStatistics());
+
+    handler.onMessage(channel, new PingMessage().getSendBytes());
+
+    verify(keepAliveService).processMessage(eq(peer), any(PingMessage.class));
+    verify(peer, never()).disconnect(any());
+  }
+
+  @Test
+  public void testInventoryIsDispatchedAfterHello() throws Exception {
+    when(peer.getHelloMessageReceive()).thenReturn(mock(HelloMessage.class));
+    when(peer.getPeerStatistics()).thenReturn(new PeerStatistics());
+    InventoryMessage inventory = new InventoryMessage(
+        Collections.singletonList(Sha256Hash.ZERO_HASH), InventoryType.BLOCK);
+
+    handler.onMessage(channel, inventory.getSendBytes());
+
+    verify(inventoryMsgHandler).processMessage(eq(peer), any(InventoryMessage.class));
+    verify(peer, never()).disconnect(any());
+  }
+
+  @Test
+  public void testPbftIsDispatchedAfterHello() throws Exception {
+    when(peer.getHelloMessageReceive()).thenReturn(mock(HelloMessage.class));
+    byte[] data = new byte[] {MessageTypes.PBFT_MSG.asByte()};
+    PbftMessage message = mock(PbftMessage.class);
+    try (MockedStatic<PbftMessageFactory> factory = Mockito.mockStatic(PbftMessageFactory.class)) {
+      factory.when(() -> PbftMessageFactory.create(data)).thenReturn(message);
+      handler.onMessage(channel, data);
+      verify(pbftMsgHandler).processMessage(peer, message);
+      verify(peer, never()).disconnect(any());
+    }
+  }
+}