Skip to content

Commit af8a84d

Browse files
xxo1shinexxo1shine
authored
feat(net): add P2P message deduplication and length validation (#6712)
- 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.
1 parent a2364fd commit af8a84d

9 files changed

Lines changed: 197 additions & 9 deletions

File tree

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@ public class NetConstants {
102102
public static final int MSG_CACHE_DURATION_IN_BLOCKS = 5;
103103
public static final int MAX_BLOCK_FETCH_PER_PEER = 100;
104104
public static final int MAX_TRX_FETCH_PER_PEER = 1000;
105+
public static final int MAX_SYNC_CHAIN_IDS = 30;
105106
}
106107

107108
public class DatabaseConstants {

‎framework/src/main/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandler.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import com.google.common.cache.Cache;
44
import com.google.common.cache.CacheBuilder;
55
import com.google.common.collect.Lists;
6+
import java.util.HashSet;
67
import java.util.List;
78
import java.util.concurrent.TimeUnit;
89
import lombok.extern.slf4j.Slf4j;
@@ -153,6 +154,12 @@ public boolean isAdvInv(PeerConnection peer, FetchInvDataMessage msg) {
153154

154155
private void check(PeerConnection peer, FetchInvDataMessage fetchInvDataMsg,
155156
boolean isAdv) throws P2pException {
157+
List<Sha256Hash> hashList = fetchInvDataMsg.getHashList();
158+
if (hashList.size() != new HashSet<>(hashList).size()) {
159+
throw new P2pException(TypeEnum.BAD_MESSAGE,
160+
"FetchInvData contains duplicate hashes, size: " + hashList.size());
161+
}
162+
156163
MessageTypes type = fetchInvDataMsg.getInvMessageType();
157164

158165
if (type == MessageTypes.TRX) {

‎framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,14 @@
11
package org.tron.core.net.messagehandler;
22

3+
import java.util.HashSet;
4+
import java.util.List;
35
import lombok.extern.slf4j.Slf4j;
46
import org.springframework.beans.factory.annotation.Autowired;
57
import org.springframework.stereotype.Component;
68
import org.tron.common.utils.Sha256Hash;
79
import org.tron.core.config.args.Args;
10+
import org.tron.core.exception.P2pException;
11+
import org.tron.core.exception.P2pException.TypeEnum;
812
import org.tron.core.net.TronNetDelegate;
913
import org.tron.core.net.message.TronMessage;
1014
import org.tron.core.net.message.adv.InventoryMessage;
@@ -27,7 +31,7 @@ public class InventoryMsgHandler implements TronMsgHandler {
2731
private TransactionsMsgHandler transactionsMsgHandler;
2832

2933
@Override
30-
public void processMessage(PeerConnection peer, TronMessage msg) {
34+
public void processMessage(PeerConnection peer, TronMessage msg) throws P2pException {
3135
InventoryMessage inventoryMessage = (InventoryMessage) msg;
3236
InventoryType type = inventoryMessage.getInventoryType();
3337

@@ -45,10 +49,17 @@ public void processMessage(PeerConnection peer, TronMessage msg) {
4549
}
4650
}
4751

48-
private boolean check(PeerConnection peer, InventoryMessage inventoryMessage) {
52+
private boolean check(PeerConnection peer, InventoryMessage inventoryMessage)
53+
throws P2pException {
54+
55+
List<Sha256Hash> hashList = inventoryMessage.getHashList();
56+
if (hashList.size() != new HashSet<>(hashList).size()) {
57+
throw new P2pException(TypeEnum.BAD_MESSAGE,
58+
"Inventory contains duplicate hashes, size: " + hashList.size());
59+
}
4960

5061
InventoryType type = inventoryMessage.getInventoryType();
51-
int size = inventoryMessage.getHashList().size();
62+
int size = hashList.size();
5263

5364
if (peer.isNeedSyncFromPeer() || peer.isNeedSyncFromUs()) {
5465
logger.warn("Drop inv: {} size: {} from Peer {}, syncFromUs: {}, syncFromPeer: {}",

‎framework/src/main/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandler.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,12 @@ private boolean check(PeerConnection peer, SyncBlockChainMessage msg) throws P2p
7171
throw new P2pException(TypeEnum.BAD_MESSAGE, "SyncBlockChain blockIds is empty");
7272
}
7373

74+
if (blockIds.size() > NetConstants.MAX_SYNC_CHAIN_IDS) {
75+
throw new P2pException(TypeEnum.BAD_MESSAGE,
76+
"SyncBlockChain blockIds size " + blockIds.size()
77+
+ " exceeds limit " + NetConstants.MAX_SYNC_CHAIN_IDS);
78+
}
79+
7480
BlockId firstId = blockIds.get(0);
7581
if (!tronNetDelegate.containBlockInMainChain(firstId)) {
7682
logger.warn("Sync message from peer {} without the first block: {}",

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

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
package org.tron.core.net.messagehandler;
22

3+
import java.util.HashSet;
4+
import java.util.List;
5+
import java.util.Set;
36
import java.util.concurrent.BlockingQueue;
47
import java.util.concurrent.ExecutorService;
58
import java.util.concurrent.LinkedBlockingQueue;
@@ -11,6 +14,7 @@
1114
import org.springframework.beans.factory.annotation.Autowired;
1215
import org.springframework.stereotype.Component;
1316
import org.tron.common.es.ExecutorServiceManager;
17+
import org.tron.common.utils.Sha256Hash;
1418
import org.tron.core.ChainBaseManager;
1519
import org.tron.core.config.args.Args;
1620
import org.tron.core.exception.P2pException;
@@ -121,8 +125,15 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
121125
}
122126

123127
private void check(PeerConnection peer, TransactionsMessage msg) throws P2pException {
124-
for (Transaction trx : msg.getTransactions().getTransactionsList()) {
125-
Item item = new Item(new TransactionMessage(trx).getMessageId(), InventoryType.TRX);
128+
List<Transaction> list = msg.getTransactions().getTransactionsList();
129+
Set<Sha256Hash> seen = new HashSet<>(list.size() * 2);
130+
for (Transaction trx : list) {
131+
Sha256Hash id = new TransactionMessage(trx).getMessageId();
132+
if (!seen.add(id)) {
133+
throw new P2pException(TypeEnum.BAD_MESSAGE,
134+
"TransactionsMessage contains duplicate transaction: " + id);
135+
}
136+
Item item = new Item(id, InventoryType.TRX);
126137
if (!peer.getAdvInvRequest().containsKey(item)) {
127138
throw new P2pException(TypeEnum.BAD_MESSAGE,
128139
"trx: " + msg.getMessageId() + " without request.");

‎framework/src/test/java/org/tron/core/net/messagehandler/FetchInvDataMsgHandlerTest.java‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.tron.common.utils.Sha256Hash;
1616
import org.tron.core.capsule.BlockCapsule;
1717
import org.tron.core.config.Parameter;
18+
import org.tron.core.exception.P2pException;
1819
import org.tron.core.net.P2pRateLimiter;
1920
import org.tron.core.net.TronNetDelegate;
2021
import org.tron.core.net.message.adv.BlockMessage;
@@ -124,12 +125,42 @@ public void testSyncFetchCheck() {
124125
Assert.assertEquals("minBlockNum: 16000, blockNum: 10000", e2.getMessage());
125126
}
126127

128+
@Test
129+
public void testDuplicateHashRejected() throws Exception {
130+
FetchInvDataMsgHandler handler = new FetchInvDataMsgHandler();
131+
PeerConnection peer = Mockito.mock(PeerConnection.class);
132+
AdvService advService = Mockito.mock(AdvService.class);
133+
TronNetDelegate tronNetDelegate = Mockito.mock(TronNetDelegate.class);
134+
135+
ReflectUtils.setFieldValue(handler, "advService", advService);
136+
ReflectUtils.setFieldValue(handler, "tronNetDelegate", tronNetDelegate);
137+
138+
Sha256Hash hash = Sha256Hash.ZERO_HASH;
139+
List<Sha256Hash> hashList = new LinkedList<>();
140+
hashList.add(hash);
141+
hashList.add(hash); // duplicate
142+
143+
FetchInvDataMessage msg = new FetchInvDataMessage(hashList,
144+
Protocol.Inventory.InventoryType.TRX);
145+
146+
Cache<Item, Long> advInvSpread = CacheBuilder.newBuilder()
147+
.maximumSize(20000).expireAfterWrite(1, TimeUnit.HOURS).build();
148+
advInvSpread.put(new Item(hash, Protocol.Inventory.InventoryType.TRX), 1L);
149+
Mockito.when(peer.getAdvInvSpread()).thenReturn(advInvSpread);
150+
151+
try {
152+
handler.processMessage(peer, msg);
153+
Assert.fail("Expected P2pException for duplicate hash");
154+
} catch (P2pException e) {
155+
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
156+
}
157+
}
158+
127159
@Test
128160
public void testRateLimiter() {
129-
BlockCapsule.BlockId blockId = new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, 10000L);
130161
List<Sha256Hash> blockIds = new LinkedList<>();
131162
for (int i = 0; i <= 100; i++) {
132-
blockIds.add(blockId);
163+
blockIds.add(new BlockCapsule.BlockId(Sha256Hash.ZERO_HASH, (long) i));
133164
}
134165
FetchInvDataMessage msg =
135166
new FetchInvDataMessage(blockIds, Protocol.Inventory.InventoryType.BLOCK);

‎framework/src/test/java/org/tron/core/net/messagehandler/InventoryMsgHandlerTest.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,14 @@
66
import java.net.InetAddress;
77
import java.net.InetSocketAddress;
88
import java.util.ArrayList;
9+
import java.util.Arrays;
10+
import org.junit.Assert;
911
import org.junit.Test;
1012
import org.mockito.Mockito;
1113
import org.tron.common.TestConstants;
14+
import org.tron.common.utils.Sha256Hash;
1215
import org.tron.core.config.args.Args;
16+
import org.tron.core.exception.P2pException;
1317
import org.tron.core.net.TronNetDelegate;
1418
import org.tron.core.net.message.adv.InventoryMessage;
1519
import org.tron.core.net.peer.PeerConnection;
@@ -52,6 +56,24 @@ public void testProcessMessage() throws Exception {
5256
Mockito.verify(tronNetDelegate, Mockito.atLeastOnce()).isBlockUnsolidified();
5357
}
5458

59+
@Test
60+
public void testDuplicateHashesRejected() throws Exception {
61+
InventoryMsgHandler handler = new InventoryMsgHandler();
62+
Args.setParam(new String[] {}, TestConstants.TEST_CONF);
63+
64+
Sha256Hash hash = Sha256Hash.wrap(new byte[32]);
65+
InventoryMessage msg = new InventoryMessage(Arrays.asList(hash, hash), InventoryType.TRX);
66+
PeerConnection peer = new PeerConnection();
67+
peer.setChannel(getChannel("1.0.0.4", 1000));
68+
69+
try {
70+
handler.processMessage(peer, msg);
71+
Assert.fail("Expected P2pException for duplicate hashes");
72+
} catch (P2pException e) {
73+
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
74+
}
75+
}
76+
5577
private Channel getChannel(String host, int port) throws Exception {
5678
Channel channel = new Channel();
5779
InetSocketAddress inetSocketAddress = new InetSocketAddress(host, port);

‎framework/src/test/java/org/tron/core/net/messagehandler/SyncBlockChainMsgHandlerTest.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,13 @@
1616
import org.junit.rules.TemporaryFolder;
1717
import org.tron.common.TestConstants;
1818
import org.tron.common.application.TronApplicationContext;
19+
import org.tron.common.utils.Sha256Hash;
1920
import org.tron.core.capsule.BlockCapsule;
2021
import org.tron.core.capsule.BlockCapsule.BlockId;
2122
import org.tron.core.config.DefaultConfig;
2223
import org.tron.core.config.args.Args;
2324
import org.tron.core.exception.P2pException;
25+
import org.tron.core.net.TronNetDelegate;
2426
import org.tron.core.net.message.sync.BlockInventoryMessage;
2527
import org.tron.core.net.message.sync.SyncBlockChainMessage;
2628
import org.tron.core.net.peer.PeerConnection;
@@ -108,6 +110,56 @@ public void testProcessMessage() throws Exception {
108110
Assert.assertEquals(1, list.size());
109111
}
110112

113+
@Test
114+
public void testBlockIdsExceedsLimit() throws Exception {
115+
List<BlockId> blockIds = new ArrayList<>();
116+
// genesis block as first (in main chain), then 30 more = 31 total → exceeds limit
117+
BlockId genesis = context.getBean(
118+
TronNetDelegate.class).getGenesisBlockId();
119+
blockIds.add(genesis);
120+
for (int i = 1; i <= 30; i++) {
121+
blockIds.add(new BlockId(Sha256Hash.ZERO_HASH, i));
122+
}
123+
SyncBlockChainMessage msg = new SyncBlockChainMessage(blockIds);
124+
125+
try {
126+
Method checkMethod = SyncBlockChainMsgHandler.class
127+
.getDeclaredMethod("check", PeerConnection.class, SyncBlockChainMessage.class);
128+
checkMethod.setAccessible(true);
129+
checkMethod.invoke(handler, peer, msg);
130+
Assert.fail("Expected P2pException for oversized blockIds");
131+
} catch (InvocationTargetException e) {
132+
Assert.assertTrue(e.getCause() instanceof P2pException);
133+
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE,
134+
((P2pException) e.getCause()).getType());
135+
}
136+
}
137+
138+
@Test
139+
public void testBlockIdsAtLimit() throws Exception {
140+
List<BlockId> blockIds = new ArrayList<>();
141+
BlockId genesis = context.getBean(
142+
TronNetDelegate.class).getGenesisBlockId();
143+
blockIds.add(genesis);
144+
for (int i = 1; i < 30; i++) {
145+
blockIds.add(new BlockId(Sha256Hash.ZERO_HASH, i));
146+
}
147+
// exactly 30 → should not throw for length check
148+
SyncBlockChainMessage msg = new SyncBlockChainMessage(blockIds);
149+
150+
Method checkMethod = SyncBlockChainMsgHandler.class
151+
.getDeclaredMethod("check", PeerConnection.class, SyncBlockChainMessage.class);
152+
checkMethod.setAccessible(true);
153+
// does not throw P2pException due to length (may return false for other checks — that's fine)
154+
try {
155+
checkMethod.invoke(handler, peer, msg);
156+
} catch (InvocationTargetException e) {
157+
Assert.assertFalse("Should not fail with BAD_MESSAGE for length at limit",
158+
e.getCause() instanceof P2pException
159+
&& ((P2pException) e.getCause()).getMessage().contains("exceeds limit"));
160+
}
161+
}
162+
111163
@AfterClass
112164
public static void destroy() {
113165
for (PeerConnection p : PeerManager.getPeers()) {

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

Lines changed: 49 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.tron.common.TestConstants;
2424
import org.tron.common.runtime.TvmTestUtils;
2525
import org.tron.common.utils.ByteArray;
26+
import org.tron.common.utils.ReflectUtils;
2627
import org.tron.core.ChainBaseManager;
2728
import org.tron.core.config.args.Args;
2829
import org.tron.core.exception.P2pException;
@@ -142,10 +143,10 @@ public void testProcessMessageAfterClose() throws Exception {
142143
TransactionsMsgHandler handler = new TransactionsMsgHandler();
143144
handler.init();
144145
handler.close();
145-
146+
146147
PeerConnection peer = Mockito.mock(PeerConnection.class);
147148
TransactionsMessage msg = Mockito.mock(TransactionsMessage.class);
148-
149+
149150
handler.processMessage(peer, msg);
150151

151152
Mockito.verify(msg, Mockito.never()).getTransactions();
@@ -290,6 +291,52 @@ public void testHandleTransaction() throws Exception {
290291
}
291292
}
292293

294+
@Test
295+
public void testDuplicateTransactionRejected() throws Exception {
296+
TransactionsMsgHandler handler = new TransactionsMsgHandler();
297+
handler.init();
298+
try {
299+
PeerConnection peer = Mockito.mock(PeerConnection.class);
300+
301+
// Build a transaction
302+
BalanceContract.TransferContract transferContract = BalanceContract.TransferContract
303+
.newBuilder()
304+
.setAmount(10)
305+
.setOwnerAddress(ByteString.copyFrom(ByteArray.fromHexString("121212a9cf")))
306+
.setToAddress(ByteString.copyFrom(ByteArray.fromHexString("232323a9cf")))
307+
.build();
308+
Protocol.Transaction trx = Protocol.Transaction.newBuilder()
309+
.setRawData(Protocol.Transaction.raw.newBuilder()
310+
.addContract(Protocol.Transaction.Contract.newBuilder()
311+
.setType(Protocol.Transaction.Contract.ContractType.TransferContract)
312+
.setParameter(Any.pack(transferContract)).build())
313+
.build())
314+
.build();
315+
316+
// Same trx twice → duplicate
317+
Protocol.Transactions transactions = Protocol.Transactions.newBuilder()
318+
.addTransactions(trx)
319+
.addTransactions(trx)
320+
.build();
321+
TransactionsMessage msg = new TransactionsMessage(transactions.getTransactionsList());
322+
323+
TransactionMessage trxMsg = new TransactionMessage(trx);
324+
Item item = new Item(trxMsg.getMessageId(), Protocol.Inventory.InventoryType.TRX);
325+
Map<Item, Long> advInvRequest = new ConcurrentHashMap<>();
326+
advInvRequest.put(item, System.currentTimeMillis());
327+
Mockito.when(peer.getAdvInvRequest()).thenReturn(advInvRequest);
328+
329+
try {
330+
handler.processMessage(peer, msg);
331+
Assert.fail("Expected P2pException for duplicate transaction");
332+
} catch (P2pException e) {
333+
Assert.assertEquals(P2pException.TypeEnum.BAD_MESSAGE, e.getType());
334+
}
335+
} finally {
336+
handler.close();
337+
}
338+
}
339+
293340
@Test
294341
public void testIsBusyWithCachedTransactions() throws Exception {
295342
TransactionsMsgHandler handler = new TransactionsMsgHandler();

0 commit comments

Comments
 (0)