fix(l1): validate incoming pooled tx blobs

.github/workflows/pr-main_l1.yaml

@@ -150,7 +150,7 @@ jobs:
             artifact_prefix: rpc_compat
           - name: "Devp2p tests"
             simulation: devp2p
-            limit: discv4|eth|snap/Ping|Amplification|Status|StorageRanges|ByteCodes|GetBlockHeaders|SimultaneousRequests|SameRequestID|ZeroRequestID|GetBlockBodies|MaliciousHandshake|MaliciousStatus|Transaction|NewPooledTxs|GetBlockReceipts|LargeTxRequest|InvalidTxs|BlockRangeUpdate|AccountRange|GetTrieNodes|GetByteCodes|GetStorageRanges|Findnode
+            limit: discv4|eth|snap/Ping|Amplification|Status|StorageRanges|ByteCodes|GetBlockHeaders|SimultaneousRequests|SameRequestID|ZeroRequestID|GetBlockBodies|MaliciousHandshake|MaliciousStatus|Transaction|NewPooledTxs|GetBlockReceipts|LargeTxRequest|InvalidTxs|BlockRangeUpdate|AccountRange|GetTrieNodes|GetByteCodes|GetStorageRanges|Findnode|TestBlobTxWithMismatchedSidecar|TestBlobTxWithoutSidecar
             artifact_prefix: devp2p
           - name: "Engine Auth and EC tests"
             simulation: ethereum/engine

crates/common/types/blobs_bundle.rs

@@ -78,6 +78,10 @@ impl BlobsBundle {
         Self::default()
     }
 
+    pub fn is_empty(&self) -> bool {
+        self.blobs.is_empty() && self.commitments.is_empty() && self.proofs.is_empty()
+    }
+
     // In the future we might want to provide a new method that calculates the commitments and proofs using the following.
     #[cfg(feature = "c-kzg")]
     pub fn create_from_blobs(blobs: &Vec<Blob>) -> Result<Self, BlobsBundleError> {
@@ -169,6 +173,23 @@ impl BlobsBundle {
 
         Ok(())
     }
+
+    pub fn validate_blob_commitment_hashes(
+        &self,
+        blob_versioned_hashes: &[H256],
+    ) -> Result<(), BlobsBundleError> {
+        if self.commitments.len() != blob_versioned_hashes.len() {
+            return Err(BlobsBundleError::BlobVersionedHashesError);
+        }
+        for (commitment, blob_versioned_hash) in
+            self.commitments.iter().zip(blob_versioned_hashes.iter())
+        {
+            if *blob_versioned_hash != kzg_commitment_to_versioned_hash(commitment) {
+                return Err(BlobsBundleError::BlobVersionedHashesError);
+            }
+        }
+        Ok(())
+    }
 }
 
 impl RLPEncode for BlobsBundle {

crates/common/types/transaction.rs

@@ -140,7 +140,20 @@ impl RLPEncode for WrappedEIP4844Transaction {
 impl RLPDecode for WrappedEIP4844Transaction {
     fn decode_unfinished(rlp: &[u8]) -> Result<(WrappedEIP4844Transaction, &[u8]), RLPDecodeError> {
         let decoder = Decoder::new(rlp)?;
-        let (tx, decoder) = decoder.decode_field("tx")?;
+        let Ok((tx, decoder)) = decoder.decode_field("tx") else {
+            // Handle the case of blobless transaction
+            let (tx, rest) = EIP4844Transaction::decode_unfinished(rlp)?;
+            return Ok((
+                WrappedEIP4844Transaction {
+                    tx,
+                    wrapper_version: None,
+                    // Empty blobs bundles are not valid
+                    blobs_bundle: BlobsBundle::empty(),
+                },
+                rest,
+            ));
+        };
+
         let (wrapper_version, decoder) = decoder.decode_optional_field();
         let (blobs, decoder) = decoder.decode_field("blobs")?;
         let (commitments, decoder) = decoder.decode_field("commitments")?;

crates/networking/p2p/rlpx/connection/server.rs

@@ -37,9 +37,9 @@ use crate::{
     types::Node,
 };
 use ethrex_blockchain::Blockchain;
-use ethrex_common::types::MempoolTransaction;
 #[cfg(feature = "l2")]
 use ethrex_common::types::Transaction;
+use ethrex_common::types::{MempoolTransaction, P2PTransaction};
 use ethrex_storage::{Store, error::StoreError};
 use ethrex_trie::TrieError;
 use futures::{SinkExt as _, Stream, stream::SplitSink};
@@ -996,8 +996,26 @@ async fn handle_incoming_message(
             send(state, Message::PooledTransactions(response)).await?;
         }
         Message::PooledTransactions(msg) if peer_supports_eth => {
+            // If we receive a blob transaction without blobs or with blobs that don't match the versioned hashes we must disconnect from the peer
+            for tx in &msg.pooled_transactions {
+                if let P2PTransaction::EIP4844TransactionWithBlobs(itx) = tx
+                    && (itx.blobs_bundle.is_empty()
+                        || itx
+                            .blobs_bundle
+                            .validate_blob_commitment_hashes(&itx.tx.blob_versioned_hashes)
+                            .is_err())
+                {
+                    log_peer_warn!(
+                        &state.node,
+                        "disconnected from peer. Reason: Invalid/Missing Blobs",
+                    );
+                    send_disconnect_message(state, Some(DisconnectReason::SubprotocolError)).await;
+                    return Err(PeerConnectionError::DisconnectSent(
+                        DisconnectReason::SubprotocolError,
+                    ));
+                }
+            }
             if state.blockchain.is_synced() {
-                // TODO(#3745): disconnect from peers that send invalid blob sidecars
                 if let Some(requested) = state.requested_pooled_txs.get(&msg.id) {
                     let fork = state.blockchain.current_fork().await?;
                     if let Err(error) = msg.validate_requested(requested, fork).await {

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

@@ -224,7 +224,7 @@ pub 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
     pub id: u64,
-    pooled_transactions: Vec<P2PTransaction>,
+    pub pooled_transactions: Vec<P2PTransaction>,
 }
 
 impl PooledTransactions {