feat(l1): track new pooled transaction hashes (lambdaclass#3158)

**Motivation**

We weren't keeping track of the received NewPooledTransationHashes
messages, we were just propagating the incoming msg without validating
them.
<!-- Why does this pull request exist? What are its goals? -->

The PR fixes the following hivetests:
- BlobViolations
This test sends some invalid blob tx announcements and expects the node
to disconnect.

- TestBlobTxWithoutSidecar
This test checks that a blob transaction first advertised/transmitted
without blobs will result in the sending peer being disconnected, and
the full transaction should be successfully retrieved from another peer.

- TestBlobTxWithMismatchedSidecar
This test checks that a blob transaction first advertised/transmitted
without blobs, whose commitment don't correspond to the
blob_versioned_hashes in the transaction, will result in the sending
peer being disconnected, and the full transaction should be successfully
retrieved from another peer.


### Sequence diagram of TestBlobTxWithoutSidecar &
TestBlobTxWithMismatchedSidecar
```mermaid
sequenceDiagram
    participant Node
    participant BadPeer
    participant GoodPeer

    %% Setup: Forkchoice Update
    Node->>Node: sendForkchoiceUpdated()

    %% Stage 1: Bad Peer announces transaction
    BadPeer->>Node: NewPooledTransactionHashesMsg<br>(badTx Hash, BlobTxType, Size)
    Node->>BadPeer: GetPooledTransactionsPacket<br>(Request badTx)
    Note right of BadPeer: stage1.Done()

    %% Stage 2: Good Peer announces transaction
    Note right of GoodPeer: stage1.Wait()
    GoodPeer->>Node: NewPooledTransactionHashesMsg<br>(tx Hash, BlobTxType, Size)
    Note right of GoodPeer: stage2.Done()

    %% Stage 3: Bad Peer sends invalid transaction and is disconnected
    Note right of BadPeer: stage2.Wait()
    BadPeer->>Node: PooledTransactionsMsg<br>(badTx, RequestId)
    Node->>BadPeer: Disconnect
    Note right of BadPeer: stage3.Done()

    %% Good Peer sends correct transaction
    Note right of GoodPeer: stage3.Wait()
    Node->>GoodPeer: GetPooledTransactionsPacket<br>(Request tx)
    GoodPeer->>Node: PooledTransactionsMsg<br>(tx, RequestId)
    Note right of GoodPeer: No Disconnect<br>close(errc)

    %% Synchronization Points
    Note over BadPeer,GoodPeer: stage1: Bad Peer announces<br>stage2: Good Peer announces<br>stage3: Bad Peer disconnected
```
You can check the tests code
[here](https://github.com/ethereum/go-ethereum/blob/72d92698a474059f3a73798c6312699c1f210497/cmd/devp2p/internal/ethtest/suite.go#L917)

**Description**
A new field `requested_pooled_txs` was added to the RLPxConnection
struct. In that struct, the request_id and the requested hashed are
stored to check later, when the response is received. By doing these, we
can assure that the received Pooled Transactions match the expected
income.

Some new error types were added.

The fn `get_p2p_transaction` was moved to RLPxConnection.

Can be tested using:
```bash
make run-hive SIMULATION=devp2p TEST_PATTERN='eth/blob' HIVE_BRANCH=mecha/update-devp2p-testfiles SIM_LOG_LEVEL=4
```

The tests can't be activated in the CI yet. We need to update the Hive
fork first. [PR](lambdaclass/hive#35)
<!-- A clear and concise general description of the changes this PR
introduces -->

<!-- Link to issues: Resolves lambdaclass#111, Resolves lambdaclass#222 -->

Closes: 
lambdaclass#1781 
lambdaclass#3122
lambdaclass#3123 
lambdaclass#1416

Author: Mechanix97

crates/blockchain/blockchain.rs

@@ -12,16 +12,16 @@ use constants::{MAX_INITCODE_SIZE, MAX_TRANSACTION_DATA_SIZE};
 use error::MempoolError;
 use error::{ChainError, InvalidBlockError};
 use ethrex_common::constants::{GAS_PER_BLOB, MIN_BASE_FEE_PER_BLOB_GAS};
-use ethrex_common::types::ELASTICITY_MULTIPLIER;
 use ethrex_common::types::MempoolTransaction;
 use ethrex_common::types::block_execution_witness::ExecutionWitnessResult;
 use ethrex_common::types::requests::{EncodedRequests, Requests, compute_requests_hash};
 use ethrex_common::types::{
     AccountUpdate, Block, BlockHash, BlockHeader, BlockNumber, ChainConfig, EIP4844Transaction,
-    Receipt, Transaction, compute_receipts_root, validate_block_header,
+    Receipt, Transaction, WrappedEIP4844Transaction, compute_receipts_root, validate_block_header,
     validate_cancun_header_fields, validate_prague_header_fields,
     validate_pre_cancun_header_fields,
 };
+use ethrex_common::types::{ELASTICITY_MULTIPLIER, P2PTransaction};
 use ethrex_common::{Address, H256, TrieLogger};
 use ethrex_metrics::metrics;
 use ethrex_storage::{Store, UpdateBatch, error::StoreError, hash_address, hash_key};
@@ -737,6 +737,39 @@ impl Blockchain {
     pub fn is_synced(&self) -> bool {
         self.is_synced.load(Ordering::Relaxed)
     }
+
+    pub fn get_p2p_transaction_by_hash(&self, hash: &H256) -> Result<P2PTransaction, StoreError> {
+        let Some(tx) = self.mempool.get_transaction_by_hash(*hash)? else {
+            return Err(StoreError::Custom(format!(
+                "Hash {} not found in the mempool",
+                hash
+            )));
+        };
+        let result = match tx {
+            Transaction::LegacyTransaction(itx) => P2PTransaction::LegacyTransaction(itx),
+            Transaction::EIP2930Transaction(itx) => P2PTransaction::EIP2930Transaction(itx),
+            Transaction::EIP1559Transaction(itx) => P2PTransaction::EIP1559Transaction(itx),
+            Transaction::EIP4844Transaction(itx) => {
+                let Some(bundle) = self.mempool.get_blobs_bundle(*hash)? else {
+                    return Err(StoreError::Custom(format!(
+                        "Blob transaction present without its bundle: hash {}",
+                        hash
+                    )));
+                };
+
+                P2PTransaction::EIP4844TransactionWithBlobs(WrappedEIP4844Transaction {
+                    tx: itx,
+                    blobs_bundle: bundle,
+                })
+            }
+            Transaction::EIP7702Transaction(itx) => P2PTransaction::EIP7702Transaction(itx),
+            Transaction::PrivilegedL2Transaction(itx) => {
+                P2PTransaction::PrivilegedL2Transaction(itx)
+            }
+        };
+
+        Ok(result)
+    }
 }
 
 pub fn validate_requests_hash(

crates/blockchain/error.rs

@@ -84,6 +84,12 @@ pub enum MempoolError {
     NotEnoughBalance,
     #[error("Transaction gas fields are invalid")]
     InvalidTxGasvalues,
+    #[error("Invalid pooled TxType, expected: {0}")]
+    InvalidPooledTxType(u8),
+    #[error("Invalid pooled transaction size, differs from expected")]
+    InvalidPooledTxSize,
+    #[error("Requested pooled transaction was not received")]
+    RequestedPooledTxNotFound,
 }
 
 #[derive(Debug)]

crates/common/types/transaction.rs

@@ -1504,6 +1504,29 @@ mod canonic_encoding {
             self.encode_canonical(&mut buf);
             buf
         }
+
+        pub fn compute_hash(&self) -> H256 {
+            match self {
+                P2PTransaction::LegacyTransaction(t) => {
+                    Transaction::LegacyTransaction(t.clone()).compute_hash()
+                }
+                P2PTransaction::EIP2930Transaction(t) => {
+                    Transaction::EIP2930Transaction(t.clone()).compute_hash()
+                }
+                P2PTransaction::EIP1559Transaction(t) => {
+                    Transaction::EIP1559Transaction(t.clone()).compute_hash()
+                }
+                P2PTransaction::EIP4844TransactionWithBlobs(t) => {
+                    Transaction::EIP4844Transaction(t.tx.clone()).compute_hash()
+                }
+                P2PTransaction::EIP7702Transaction(t) => {
+                    Transaction::EIP7702Transaction(t.clone()).compute_hash()
+                }
+                P2PTransaction::PrivilegedL2Transaction(t) => {
+                    Transaction::PrivilegedL2Transaction(t.clone()).compute_hash()
+                }
+            }
+        }
     }
 }
 

crates/networking/p2p/rlpx/connection.rs

@@ -1,3 +1,8 @@
+use super::{
+    eth::{transactions::NewPooledTransactionHashes, update::BlockRangeUpdate},
+    p2p::DisconnectReason,
+    utils::log_peer_warn,
+};
 use crate::{
     kademlia::PeerChannels,
     rlpx::{
@@ -32,7 +37,10 @@ use ethrex_storage::Store;
 use futures::SinkExt;
 use k256::{PublicKey, SecretKey, ecdsa::SigningKey};
 use rand::random;
-use std::{collections::HashSet, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 use tokio::{
     io::{AsyncRead, AsyncWrite},
     sync::{
@@ -47,12 +55,6 @@ use tokio_stream::StreamExt;
 use tokio_util::codec::Framed;
 use tracing::debug;
 
-use super::{
-    eth::{transactions::NewPooledTransactionHashes, update::BlockRangeUpdate},
-    p2p::DisconnectReason,
-    utils::log_peer_warn,
-};
-
 const PERIODIC_PING_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);
 const PERIODIC_TX_BROADCAST_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500);
 const PERIODIC_TASKS_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500);
@@ -92,6 +94,7 @@ pub(crate) struct RLPxConnection<S> {
     next_block_range_update: Instant,
     last_block_range_update_block: u64,
     broadcasted_txs: HashSet<H256>,
+    requested_pooled_txs: HashMap<u64, NewPooledTransactionHashes>,
     client_version: String,
     /// Send end of the channel used to broadcast messages
     /// to other connected peers, is ok to have it here,
@@ -130,6 +133,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
             next_block_range_update: Instant::now() + PERIODIC_BLOCK_RANGE_UPDATE_INTERVAL,
             last_block_range_update_block: 0,
             broadcasted_txs: HashSet::new(),
+            requested_pooled_txs: HashMap::new(),
             client_version,
             connection_broadcast_send: connection_broadcast,
         }
@@ -568,12 +572,14 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
             Message::NewPooledTransactionHashes(new_pooled_transaction_hashes)
                 if peer_supports_eth =>
             {
-                //TODO(#1415): evaluate keeping track of requests to avoid sending the same twice.
                 let hashes =
                     new_pooled_transaction_hashes.get_transactions_to_request(&self.blockchain)?;
 
-                //TODO(#1416): Evaluate keeping track of the request-id.
-                let request = GetPooledTransactions::new(random(), hashes);
+                let request_id = random();
+                self.requested_pooled_txs
+                    .insert(request_id, new_pooled_transaction_hashes);
+
+                let request = GetPooledTransactions::new(request_id, hashes);
                 self.send(Message::GetPooledTransactions(request)).await?;
             }
             Message::GetPooledTransactions(msg) => {
@@ -582,6 +588,21 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
             }
             Message::PooledTransactions(msg) if peer_supports_eth => {
                 if self.blockchain.is_synced() {
+                    if let Some(requested) = self.requested_pooled_txs.get(&msg.id) {
+                        if let Err(error) = msg.validate_requested(requested).await {
+                            log_peer_warn(
+                                &self.node,
+                                &format!("disconnected from peer. Reason: {}", error),
+                            );
+                            self.send_disconnect_message(Some(DisconnectReason::SubprotocolError))
+                                .await;
+                            return Err(RLPxError::DisconnectSent(
+                                DisconnectReason::SubprotocolError,
+                            ));
+                        } else {
+                            self.requested_pooled_txs.remove(&msg.id);
+                        }
+                    }
                     msg.handle(&self.node, &self.blockchain).await?;
                 }
             }

crates/networking/p2p/rlpx/eth/transactions.rs

@@ -4,8 +4,8 @@ use ethrex_blockchain::Blockchain;
 use ethrex_blockchain::error::MempoolError;
 use ethrex_common::types::BlobsBundle;
 use ethrex_common::types::P2PTransaction;
-use ethrex_common::types::WrappedEIP4844Transaction;
 use ethrex_common::{H256, types::Transaction};
+use ethrex_rlp::encode::RLPEncode;
 use ethrex_rlp::{
     error::{RLPDecodeError, RLPEncodeError},
     structs::{Decoder, Encoder},
@@ -65,11 +65,11 @@ impl RLPxMessage for Transactions {
 
 // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#newpooledtransactionhashes-0x08
 // Broadcast message
-#[derive(Debug)]
+#[derive(Debug, Eq, PartialEq)]
 pub(crate) struct NewPooledTransactionHashes {
     transaction_types: Bytes,
     transaction_sizes: Vec<usize>,
-    transaction_hashes: Vec<H256>,
+    pub(crate) transaction_hashes: Vec<H256>,
 }
 
 impl NewPooledTransactionHashes {
@@ -165,8 +165,8 @@ impl RLPxMessage for NewPooledTransactionHashes {
 pub(crate) struct GetPooledTransactions {
     // id is a u64 chosen by the requesting peer, the responding peer must mirror the value for the response
     // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#protocol-messages
-    id: u64,
-    transaction_hashes: Vec<H256>,
+    pub(crate) id: u64,
+    pub(crate) transaction_hashes: Vec<H256>,
 }
 
 impl GetPooledTransactions {
@@ -182,54 +182,19 @@ impl GetPooledTransactions {
         let txs = self
             .transaction_hashes
             .iter()
-            .map(|hash| Self::get_p2p_transaction(hash, blockchain))
+            .map(|hash| blockchain.get_p2p_transaction_by_hash(hash))
             // Return an error in case anything failed.
             .collect::<Result<Vec<_>, _>>()?
             .into_iter()
             // As per the spec, Nones are perfectly acceptable, for example if a transaction was
             // taken out of the mempool due to payload building after being advertised.
-            .flatten()
             .collect();
 
         Ok(PooledTransactions {
             id: self.id,
             pooled_transactions: txs,
         })
     }
-
-    /// Gets a p2p transaction given a hash.
-    fn get_p2p_transaction(
-        hash: &H256,
-        blockchain: &Blockchain,
-    ) -> Result<Option<P2PTransaction>, StoreError> {
-        let Some(tx) = blockchain.mempool.get_transaction_by_hash(*hash)? else {
-            return Ok(None);
-        };
-        let result = match tx {
-            Transaction::LegacyTransaction(itx) => P2PTransaction::LegacyTransaction(itx),
-            Transaction::EIP2930Transaction(itx) => P2PTransaction::EIP2930Transaction(itx),
-            Transaction::EIP1559Transaction(itx) => P2PTransaction::EIP1559Transaction(itx),
-            Transaction::EIP4844Transaction(itx) => {
-                let Some(bundle) = blockchain.mempool.get_blobs_bundle(*hash)? else {
-                    return Err(StoreError::Custom(format!(
-                        "Blob transaction present without its bundle: hash {}",
-                        hash
-                    )));
-                };
-
-                P2PTransaction::EIP4844TransactionWithBlobs(WrappedEIP4844Transaction {
-                    tx: itx,
-                    blobs_bundle: bundle,
-                })
-            }
-            Transaction::EIP7702Transaction(itx) => P2PTransaction::EIP7702Transaction(itx),
-            Transaction::PrivilegedL2Transaction(itx) => {
-                P2PTransaction::PrivilegedL2Transaction(itx)
-            }
-        };
-
-        Ok(Some(result))
-    }
 }
 
 impl RLPxMessage for GetPooledTransactions {
@@ -261,7 +226,7 @@ impl RLPxMessage for GetPooledTransactions {
 pub(crate) struct PooledTransactions {
     // id is a u64 chosen by the requesting peer, the responding peer must mirror the value for the response
     // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#protocol-messages
-    id: u64,
+    pub(crate) id: u64,
     pooled_transactions: Vec<P2PTransaction>,
 }
 
@@ -273,6 +238,37 @@ impl PooledTransactions {
         }
     }
 
+    /// validates if the received TXs match the request
+    pub async fn validate_requested(
+        &self,
+        requested: &NewPooledTransactionHashes,
+    ) -> Result<(), MempoolError> {
+        for tx in &self.pooled_transactions {
+            if let P2PTransaction::EIP4844TransactionWithBlobs(itx) = tx {
+                itx.blobs_bundle.validate(&itx.tx)?;
+            }
+            let tx_hash = tx.compute_hash();
+            let Some(pos) = requested
+                .transaction_hashes
+                .iter()
+                .position(|&hash| hash == tx_hash)
+            else {
+                return Err(MempoolError::RequestedPooledTxNotFound);
+            };
+
+            let expected_type = requested.transaction_types[pos];
+            let expected_size = requested.transaction_sizes[pos];
+            if tx.tx_type() as u8 != expected_type {
+                return Err(MempoolError::InvalidPooledTxType(expected_type));
+            }
+            // remove the code from the encoding (-4)
+            if tx.encode_to_vec().len() - 4 != expected_size {
+                return Err(MempoolError::InvalidPooledTxSize);
+            }
+        }
+        Ok(())
+    }
+
     /// Saves every incoming pooled transaction to the mempool.
     pub async fn handle(self, node: &Node, blockchain: &Blockchain) -> Result<(), MempoolError> {
         for tx in self.pooled_transactions {