fix(l1): validate incoming pooled tx blobs (#4963)

**Motivation**
Enable hive p2p tests `TestBlobTxWithMismatchedSidecar ` &
`TestBlobTxWithoutSidecar`
In order to pass these tests we have to:
* Be able to decode pooled transactions without blobs (aka plain eip4844
transactions instead of a wrapped eip4844 with its blobsbundle)
* Disconnect from peers that send transactions without blobs or blob
bundles that don't match the versioned hashes
<!-- Why does this pull request exist? What are its goals? -->

**Description**
* Handle the case of plain `Eip4844` transaction when RLP-decoding
`WrappedEip4844` transactions
* Disconnect from peers that send empty/mismatched blobs
<!-- A clear and concise general description of the changes this PR
introduces -->

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

Closes #3745, part of #4941

---------

Co-authored-by: Martin Paulucci <martin.c.paulucci@gmail.com>

Author: 2 people

.github/workflows/pr-main_l1.yaml

@@ -16,27 +16,27 @@ concurrency:
 
 jobs:
   detect-changes:
-      runs-on: ubuntu-latest
-      outputs:
-        run_tests: ${{ steps.finish.outputs.run_tests }}
-      steps:
-        - uses: actions/checkout@v4
-        - uses: dorny/paths-filter@v3
-          id: filter
-          with:
-            filters: |
-              run_tests:
-                - '!crates/l2/**'
-        - name: finish
-          id: finish
-          run: |
-            if [[ "${GITHUB_EVENT_NAME}" != "pull_request" ]]; then
-                echo "run_tests=true" >> "$GITHUB_OUTPUT"
-            else
-                echo "run_tests=${{ steps.filter.outputs.run_tests }}" >> "$GITHUB_OUTPUT"
-            fi
-        - name: Print result
-          run: echo "run_tests=${{ steps.finish.outputs.run_tests }}"
+    runs-on: ubuntu-latest
+    outputs:
+      run_tests: ${{ steps.finish.outputs.run_tests }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dorny/paths-filter@v3
+        id: filter
+        with:
+          filters: |
+            run_tests:
+              - '!crates/l2/**'
+      - name: finish
+        id: finish
+        run: |
+          if [[ "${GITHUB_EVENT_NAME}" != "pull_request" ]]; then
+              echo "run_tests=true" >> "$GITHUB_OUTPUT"
+          else
+              echo "run_tests=${{ steps.filter.outputs.run_tests }}" >> "$GITHUB_OUTPUT"
+          fi
+      - name: Print result
+        run: echo "run_tests=${{ steps.finish.outputs.run_tests }}"
 
   lint:
     # "Lint" is a required check, don't change the name
@@ -187,7 +187,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|BlobViolations
+            limit: discv4|eth|snap
             artifact_prefix: devp2p
           - name: "Engine Auth and EC tests"
             simulation: ethereum/engine

crates/common/types/blobs_bundle.rs

@@ -80,6 +80,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(
@@ -179,6 +183,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

@@ -133,7 +133,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

@@ -36,9 +36,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};
@@ -1027,8 +1027,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())
+                {
+                    warn!(
+                        peer=%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 {