feat(net): add P2P message deduplication and length validation

wb · wb · commit 805f5b5d1e7f · 2026-05-09T11:12:01.000+08:00
- FetchInvDataMsgHandler: reject messages with duplicate hashes
- TransactionsMsgHandler: reject messages with duplicate transactions
- SyncBlockChainMsgHandler: reject blockIds list exceeding 30 entries
- Add MAX_SYNC_CHAIN_IDS = 30 constant to NetConstants
- Add unit tests covering duplicate rejection and boundary values

All violations throw P2pException(BAD_MESSAGE), triggering peer disconnect via existing P2pEventHandlerImpl error path.
diff --git a/common/src/main/java/org/tron/common/parameter/CommonParameter.java b/common/src/main/java/org/tron/common/parameter/CommonParameter.java
@@ -492,6 +492,9 @@ public class CommonParameter {
   public long pendingTransactionTimeout;
   @Getter
   @Setter
+  public int maxTrxCacheSize;
+  @Getter
+  @Setter
   public boolean nodeMetricsEnable = false;
   @Getter
   @Setter
diff --git a/common/src/main/java/org/tron/core/config/Parameter.java b/common/src/main/java/org/tron/core/config/Parameter.java
@@ -102,6 +102,7 @@ public class NetConstants {
     public static final int MSG_CACHE_DURATION_IN_BLOCKS = 5;
     public static final int MAX_BLOCK_FETCH_PER_PEER = 100;
     public static final int MAX_TRX_FETCH_PER_PEER = 1000;
+    public static final int MAX_SYNC_CHAIN_IDS = 30;
   }
 
   public class DatabaseConstants {
diff --git a/common/src/main/java/org/tron/core/config/args/NodeConfig.java b/common/src/main/java/org/tron/core/config/args/NodeConfig.java
@@ -85,6 +85,7 @@ public class NodeConfig {
   private ChannelConfig channel = new ChannelConfig();
   private int maxTransactionPendingSize = 2000;
   private long pendingTransactionTimeout = 60000;
+  private int maxTrxCacheSize = 50_000;
   private int agreeNodeCount = 0;
   private boolean openHistoryQueryWhenLiteFN = false;
   private boolean unsolidifiedBlockCheck = false;
@@ -498,6 +499,11 @@ private void postProcess() {
     if (dynamicConfig.checkInterval <= 0) {
       dynamicConfig.checkInterval = 600;
     }
+
+    // maxTrxCacheSize: minimum 2000
+    if (maxTrxCacheSize < 2000) {
+      maxTrxCacheSize = 2000;
+    }
   }
 
   // ===========================================================================
diff --git a/common/src/main/resources/reference.conf b/common/src/main/resources/reference.conf
@@ -346,6 +346,8 @@ node {
   receiveTcpMinDataLength = 2048
   maxTransactionPendingSize = 2000
   pendingTransactionTimeout = 60000
+  # total cached trx across handler queues + pending + rePush
+  maxTrxCacheSize = 50000
 
   # Consensus agreement
   agreeNodeCount = 0
diff --git a/framework/src/main/java/org/tron/core/config/args/Args.java b/framework/src/main/java/org/tron/core/config/args/Args.java
@@ -625,6 +625,7 @@ private static void applyNodeConfig(NodeConfig nc) {
 
     PARAMETER.maxTransactionPendingSize = nc.getMaxTransactionPendingSize();
     PARAMETER.pendingTransactionTimeout = nc.getPendingTransactionTimeout();
+    PARAMETER.maxTrxCacheSize = nc.getMaxTrxCacheSize();
 
     PARAMETER.validContractProtoThreadNum = nc.getValidContractProtoThreads();
 
diff --git a/framework/src/main/java/org/tron/core/db/Manager.java b/framework/src/main/java/org/tron/core/db/Manager.java
@@ -2085,9 +2085,13 @@ public NullifierStore getNullifierStore() {
     return chainBaseManager.getNullifierStore();
   }
 
+  public int getCachedTransactionSize() {
+    return pushTransactionQueue.size() + getPendingTransactions().size()
+        + getRePushTransactions().size();
+  }
+
   public boolean isTooManyPending() {
-    return getPendingTransactions().size() + getRePushTransactions().size()
-        > maxTransactionPendingSize;
+    return getCachedTransactionSize() > maxTransactionPendingSize;
   }
 
   private void preValidateTransactionSign(List<TransactionCapsule> txs)
diff --git a/framework/src/main/java/org/tron/core/net/TronNetDelegate.java b/framework/src/main/java/org/tron/core/net/TronNetDelegate.java
@@ -384,4 +384,8 @@ public boolean isBlockUnsolidified() {
     return headNum - solidNum >= maxUnsolidifiedBlocks;
   }
 
+  public int getCachedTransactionSize() {
+    return dbManager.getCachedTransactionSize();
+  }
+
 }
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandler.java
@@ -3,6 +3,7 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.collect.Lists;
+import java.util.HashSet;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
@@ -153,6 +154,12 @@ public boolean isAdvInv(PeerConnection peer, FetchInvDataMessage msg) {
 
   private void check(PeerConnection peer, FetchInvDataMessage fetchInvDataMsg,
                      boolean isAdv) throws P2pException {
+    List<Sha256Hash> hashList = fetchInvDataMsg.getHashList();
+    if (hashList.size() != new HashSet<>(hashList).size()) {
+      throw new P2pException(TypeEnum.BAD_MESSAGE,
+          "FetchInvData contains duplicate hashes, size: " + hashList.size());
+    }
+
     MessageTypes type = fetchInvDataMsg.getInvMessageType();
 
     if (type == MessageTypes.TRX) {
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java
@@ -1,10 +1,14 @@
 package org.tron.core.net.messagehandler;
 
+import java.util.HashSet;
+import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.config.args.Args;
+import org.tron.core.exception.P2pException;
+import org.tron.core.exception.P2pException.TypeEnum;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.adv.InventoryMessage;
@@ -27,7 +31,7 @@ public class InventoryMsgHandler implements TronMsgHandler {
   private TransactionsMsgHandler transactionsMsgHandler;
 
   @Override
-  public void processMessage(PeerConnection peer, TronMessage msg) {
+  public void processMessage(PeerConnection peer, TronMessage msg) throws P2pException {
     InventoryMessage inventoryMessage = (InventoryMessage) msg;
     InventoryType type = inventoryMessage.getInventoryType();
 
@@ -45,10 +49,17 @@ public void processMessage(PeerConnection peer, TronMessage msg) {
     }
   }
 
-  private boolean check(PeerConnection peer, InventoryMessage inventoryMessage) {
+  private boolean check(PeerConnection peer, InventoryMessage inventoryMessage)
+      throws P2pException {
+
+    List<Sha256Hash> hashList = inventoryMessage.getHashList();
+    if (hashList.size() != new HashSet<>(hashList).size()) {
+      throw new P2pException(TypeEnum.BAD_MESSAGE,
+          "Inventory contains duplicate hashes, size: " + hashList.size());
+    }
 
     InventoryType type = inventoryMessage.getInventoryType();
-    int size = inventoryMessage.getHashList().size();
+    int size = hashList.size();
 
     if (peer.isNeedSyncFromPeer() || peer.isNeedSyncFromUs()) {
       logger.warn("Drop inv: {} size: {} from Peer {}, syncFromUs: {}, syncFromPeer: {}",
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java
@@ -71,6 +71,12 @@ private boolean check(PeerConnection peer, SyncBlockChainMessage msg) throws P2p
       throw new P2pException(TypeEnum.BAD_MESSAGE, "SyncBlockChain blockIds is empty");
     }
 
+    if (blockIds.size() > NetConstants.MAX_SYNC_CHAIN_IDS) {
+      throw new P2pException(TypeEnum.BAD_MESSAGE,
+          "SyncBlockChain blockIds size " + blockIds.size()
+              + " exceeds limit " + NetConstants.MAX_SYNC_CHAIN_IDS);
+    }
+
     BlockId firstId = blockIds.get(0);
     if (!tronNetDelegate.containBlockInMainChain(firstId)) {
       logger.warn("Sync message from peer {} without the first block: {}",
diff --git a/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java b/framework/src/main/java/org/tron/core/net/messagehandler/TransactionsMsgHandler.java
@@ -1,5 +1,8 @@
 package org.tron.core.net.messagehandler;
 
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -11,6 +14,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.es.ExecutorServiceManager;
+import org.tron.common.utils.Sha256Hash;
 import org.tron.core.ChainBaseManager;
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
@@ -32,7 +36,6 @@
 @Component
 public class TransactionsMsgHandler implements TronMsgHandler {
 
-  private static int MAX_TRX_SIZE = 50_000;
   private static int MAX_SMART_CONTRACT_SUBMIT_SIZE = 100;
   @Autowired
   private TronNetDelegate tronNetDelegate;
@@ -41,7 +44,8 @@ public class TransactionsMsgHandler implements TronMsgHandler {
   @Autowired
   private ChainBaseManager chainBaseManager;
 
-  private BlockingQueue<TrxEvent> smartContractQueue = new LinkedBlockingQueue(MAX_TRX_SIZE);
+  private BlockingQueue<TrxEvent> smartContractQueue = new LinkedBlockingQueue(
+      Args.getInstance().getMaxTrxCacheSize());
 
   private BlockingQueue<Runnable> queue = new LinkedBlockingQueue();
 
@@ -71,7 +75,8 @@ public void close() {
   }
 
   public boolean isBusy() {
-    return queue.size() + smartContractQueue.size() > MAX_TRX_SIZE;
+    return queue.size() + smartContractQueue.size()
+        + tronNetDelegate.getCachedTransactionSize() > Args.getInstance().getMaxTrxCacheSize();
   }
 
   @Override
@@ -120,8 +125,15 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
   }
 
   private void check(PeerConnection peer, TransactionsMessage msg) throws P2pException {
-    for (Transaction trx : msg.getTransactions().getTransactionsList()) {
-      Item item = new Item(new TransactionMessage(trx).getMessageId(), InventoryType.TRX);
+    List<Transaction> list = msg.getTransactions().getTransactionsList();
+    Set<Sha256Hash> seen = new HashSet<>(list.size() * 2);
+    for (Transaction trx : list) {
+      Sha256Hash id = new TransactionMessage(trx).getMessageId();
+      if (!seen.add(id)) {
+        throw new P2pException(TypeEnum.BAD_MESSAGE,
+            "TransactionsMessage contains duplicate transaction: " + id);
+      }
+      Item item = new Item(id, InventoryType.TRX);
       if (!peer.getAdvInvRequest().containsKey(item)) {
         throw new P2pException(TypeEnum.BAD_MESSAGE,
             "trx: " + msg.getMessageId() + " without request.");
diff --git a/framework/src/main/resources/config.conf b/framework/src/main/resources/config.conf
@@ -160,6 +160,10 @@ node {
   # Range: [50, 2000], default: 500
   # maxPendingBlockSize = 500
 
+  # Maximum total number of cached transactions (handler queues + pending + rePush).
+  # When exceeded, the node stops accepting TRX INV messages from peers.
+  # maxTrxCacheSize = 50000
+
   # Number of validate sign thread, default availableProcessors
   # validateSignThreadNum = 16
 
diff --git a/framework/src/test/java/org/tron/core/db/ManagerTest.java b/framework/src/test/java/org/tron/core/db/ManagerTest.java
@@ -26,6 +26,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
@@ -1419,6 +1420,56 @@ public void testClearSolidityContractTriggerCache() throws Exception {
     }
   }
 
+  @Test
+  public void testGetCachedTransactionSize() throws Exception {
+    BlockingQueue<TransactionCapsule> pushQ = new LinkedBlockingQueue<>();
+    pushQ.add(new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    Field pushField = Manager.class.getDeclaredField("pushTransactionQueue");
+    pushField.setAccessible(true);
+    pushField.set(dbManager, pushQ);
+
+    dbManager.getPendingTransactions().clear();
+    dbManager.getPendingTransactions().add(
+        new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    dbManager.getPendingTransactions().add(
+        new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+
+    dbManager.getRePushTransactions().clear();
+
+    // 1 (push) + 2 (pending) + 0 (rePush) = 3
+    Assert.assertEquals(3, dbManager.getCachedTransactionSize());
+
+    // cleanup
+    pushQ.clear();
+    dbManager.getPendingTransactions().clear();
+  }
+
+  @Test
+  public void testIsTooManyPendingIncludesPushQueue() throws Exception {
+    int threshold = Args.getInstance().getMaxTransactionPendingSize();
+
+    BlockingQueue<TransactionCapsule> pushQ = new LinkedBlockingQueue<>();
+    Field pushField = Manager.class.getDeclaredField("pushTransactionQueue");
+    pushField.setAccessible(true);
+    pushField.set(dbManager, pushQ);
+
+    dbManager.getPendingTransactions().clear();
+    dbManager.getRePushTransactions().clear();
+
+    for (int i = 0; i < threshold; i++) {
+      dbManager.getPendingTransactions().add(
+          new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    }
+    Assert.assertFalse(dbManager.isTooManyPending());
+
+    pushQ.add(new TransactionCapsule(Protocol.Transaction.getDefaultInstance()));
+    Assert.assertTrue(dbManager.isTooManyPending());
+
+    // cleanup
+    dbManager.getPendingTransactions().clear();
+    pushQ.clear();
+  }
+
   public void adjustBalance(AccountStore accountStore, byte[] accountAddress, long amount)
       throws BalanceInsufficientException {
     Commons.adjustBalance(accountStore, accountAddress, amount,
diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandlerTest.java
@@ -15,6 +15,7 @@
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.config.Parameter;
+import org.tron.core.exception.P2pException;
 import org.tron.core.net.P2pRateLimiter;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.adv.BlockMessage;
@@ -124,12 +125,42 @@ public void testSyncFetchCheck() {
     Assert.assertEquals("minBlockNum: 16000, blockNum: 10000", e2.getMessage());
   }
 
+  @Test
+  public void testDuplicateHashRejected() throws Exception {
+    FetchInvDataMsgHandler handler = new FetchInvDataMsgHandler();
+    PeerConnection peer = Mockito.mock(PeerConnection.class);
+    AdvService advService = Mockito.mock(AdvService.class);
+    TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);
+
+    ReflectUtils.setFieldValue(handler, "advService", advService);
+    ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate);
+
+    Sha256Hash hash = Sha256Hash.ZERO_HASH;
+    List<Sha256Hash> hashList = new LinkedList<>();
+    hashList.add(hash);
+    hashList.add(hash); // duplicate
+
+    FetchInvDataMessage msg = new FetchInvDataMessage(hashList,
+        Protocol.Inventory.InventoryType.TRX);
+
+    Cache<Item, Long> advInvSpread = CacheBuilder.newBuilder()
+        .maximumSize(20000).expireAfterWrite(1, TimeUnit.HOURS).build();
+    advInvSpread.put(new Item(hash, Protocol.Inventory.InventoryType.TRX), 1L);
+    Mockito.when(peer.getAdvInvSpread()).thenReturn(advInvSpread);
+
+    try {
+      handler.processMessage(peer, msg);
+      Assert.fail("Expected P2pException for duplicate hash");
+    } catch (P2pException e) {
+      Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
+    }
+  }
+
   @Test
   public void testRateLimiter() {
-    BlockCapsule.BlockId blockId = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 10000L);
     List<Sha256Hash> blockIds = new LinkedList<>();
     for (int i = 0; i <= 100; i++) {
-      blockIds.add(blockId);
+      blockIds.add(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, (long) i));
     }
     FetchInvDataMessage msg =
         new FetchInvDataMessage(blockIds, Protocol.Inventory.InventoryType.BLOCK);
diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/InventoryMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/InventoryMsgHandlerTest.java
@@ -6,10 +6,14 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
+import java.util.Arrays;
+import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.tron.common.TestConstants;
+import org.tron.common.utils.Sha256Hash;
 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.adv.InventoryMessage;
 import org.tron.core.net.peer.PeerConnection;
@@ -52,6 +56,24 @@ public void testProcessMessage() throws Exception {
     Mockito.verify(tronNetDelegate, Mockito.atLeastOnce()).isBlockUnsolidified();
   }
 
+  @Test
+  public void testDuplicateHashesRejected() throws Exception {
+    InventoryMsgHandler handler = new InventoryMsgHandler();
+    Args.setParam(new String[] {}, TestConstants.TEST_CONF);
+
+    Sha256Hash hash = Sha256Hash.wrap(new byte[32]);
+    InventoryMessage msg = new InventoryMessage(Arrays.asList(hash, hash), InventoryType.TRX);
+    PeerConnection peer = new PeerConnection();
+    peer.setChannel(getChannel("1.0.0.4", 1000));
+
+    try {
+      handler.processMessage(peer, msg);
+      Assert.fail("Expected P2pException for duplicate hashes");
+    } catch (P2pException e) {
+      Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
+    }
+  }
+
   private Channel getChannel(String host, int port) throws Exception {
     Channel channel = new Channel();
     InetSocketAddress inetSocketAddress = new InetSocketAddress(host, port);
diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandlerTest.java
diff --git a/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java b/framework/src/test/java/org/tron/core/net/messagehandler/TransactionsMsgHandlerTest.java

Original file line number	Diff line number	Diff line change
`@@ -102,6 +102,7 @@ public class NetConstants {`
`102`	`102`	`public static final int MSG_CACHE_DURATION_IN_BLOCKS = 5;`
`103`	`103`	`public static final int MAX_BLOCK_FETCH_PER_PEER = 100;`
`104`	`104`	`public static final int MAX_TRX_FETCH_PER_PEER = 1000;`
	`105`	`+ public static final int MAX_SYNC_CHAIN_IDS = 30;`
`105`	`106`	`}`
`106`	`107`
`107`	`108`	`public class DatabaseConstants {`
Original file line number	Diff line number	Diff line change
`@@ -2085,9 +2085,13 @@ public NullifierStore getNullifierStore() {`
`2085`	`2085`	`return chainBaseManager.getNullifierStore();`
`2086`	`2086`	`}`
`2087`	`2087`
	`2088`	`+ public int getCachedTransactionSize() {`
	`2089`	`+ return pushTransactionQueue.size() + getPendingTransactions().size()`
	`2090`	`+ + getRePushTransactions().size();`
	`2091`	`+ }`
	`2092`	`+`
`2088`	`2093`	`public boolean isTooManyPending() {`
`2089`		`- return getPendingTransactions().size() + getRePushTransactions().size()`
`2090`		`- > maxTransactionPendingSize;`
	`2094`	`+ return getCachedTransactionSize() > maxTransactionPendingSize;`
`2091`	`2095`	`}`
`2092`	`2096`
`2093`	`2097`	`private void preValidateTransactionSign(List<TransactionCapsule> txs)`
Original file line number	Diff line number	Diff line change
`@@ -384,4 +384,8 @@ public boolean isBlockUnsolidified() {`
`384`	`384`	`return headNum - solidNum >= maxUnsolidifiedBlocks;`
`385`	`385`	`}`
`386`	`386`
	`387`	`+ public int getCachedTransactionSize() {`
	`388`	`+ return dbManager.getCachedTransactionSize();`
	`389`	`+ }`
	`390`	`+`
`387`	`391`	`}`