feat(l1): only use negotiated capabilities

crates/common/rlp/decode.rs

@@ -513,7 +513,7 @@ pub fn decode_bytes(data: &[u8]) -> Result<(&[u8], &[u8]), RLPDecodeError> {
 /// Pads a slice of bytes with zeros on the left to make it a fixed size slice.
 /// The size of the data must be less than or equal to the size of the output array.
 #[inline]
-pub fn static_left_pad<const N: usize>(data: &[u8]) -> Result<[u8; N], RLPDecodeError> {
+pub(crate) fn static_left_pad<const N: usize>(data: &[u8]) -> Result<[u8; N], RLPDecodeError> {
     let mut result = [0; N];
 
     if data.is_empty() {

crates/common/rlp/encode.rs

@@ -188,7 +188,7 @@ impl<T: RLPEncode> RLPEncode for Vec<T> {
     }
 }
 
-pub fn encode_length(total_len: usize, buf: &mut dyn BufMut) {
+pub(crate) fn encode_length(total_len: usize, buf: &mut dyn BufMut) {
     if total_len < 56 {
         buf.put_u8(0xc0 + total_len as u8);
     } else {

crates/common/rlp/error.rs

@@ -26,6 +26,10 @@ pub enum RLPDecodeError {
 pub enum RLPEncodeError {
     #[error("InvalidCompression")]
     InvalidCompression(#[from] snap::Error),
+    #[error("IncompatibleProtocol")]
+    IncompatibleProtocol,
+    #[error("MalformedData")]
+    MalformedData,
     #[error("{0}")]
     Custom(String),
 }

crates/networking/p2p/discv4/messages.rs

@@ -229,8 +229,6 @@ impl PingMessage {
         }
     }
 
-    // TODO: remove when used
-    #[allow(unused)]
     pub fn with_enr_seq(self, enr_seq: u64) -> Self {
         Self {
             enr_seq: Some(enr_seq),

crates/networking/p2p/peer_handler.rs

@@ -329,9 +329,15 @@ impl PeerHandler {
             if let Some(receipts) = tokio::time::timeout(PEER_REPLY_TIMEOUT, async move {
                 loop {
                     match receiver.recv().await {
-                        Some(RLPxMessage::Receipts(receipts)) => {
-                            if receipts.get_id() == request_id {
-                                return Some(receipts.get_receipts());
+                        Some(RLPxMessage::Receipts68(receipts)) => {
+                            if receipts.id == request_id {
+                                return Some(receipts.receipts);
+                            }
+                            return None;
+                        }
+                        Some(RLPxMessage::Receipts69(receipts)) => {
+                            if receipts.id == request_id {
+                                return Some(receipts.receipts);
                             }
                             return None;
                         }

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

@@ -1,4 +1,7 @@
+use crate::rlpx::connection::server::Capabilities;
 use crate::rlpx::{error::RLPxError, message as rlpx, utils::ecdh_xchng};
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use super::handshake::{LocalState, RemoteState};
 use aes::{
@@ -24,13 +27,15 @@ pub(crate) struct RLPxCodec {
     pub(crate) egress_mac: Keccak256,
     pub(crate) ingress_aes: Aes256Ctr64BE,
     pub(crate) egress_aes: Aes256Ctr64BE,
+    pub capabilities: Arc<Mutex<Capabilities>>,
 }
 
 impl RLPxCodec {
     pub(crate) fn new(
         local_state: &LocalState,
         remote_state: &RemoteState,
         hashed_nonces: [u8; 32],
+        capabilities: Arc<Mutex<Capabilities>>,
     ) -> Result<Self, RLPxError> {
         let ephemeral_key_secret = ecdh_xchng(
             &local_state.ephemeral_key,
@@ -67,8 +72,49 @@ impl RLPxCodec {
             egress_mac,
             ingress_aes,
             egress_aes,
+            capabilities,
         })
     }
+
+    // pub fn set_p2p_protocol(&mut self, cap: &Capability) -> Result<(), RLPxError> {
+    //     if !cap.is_p2p() {
+    //         return Err(RLPxError::InternalError(
+    //             "The protocol should be p2p".into(),
+    //         ));
+    //     }
+    //     self.p2p_protocol = Some(cap.clone());
+    //     Ok(())
+    // }
+
+    // pub fn set_eth_protocol(&mut self, cap: &Capability) -> Result<(), RLPxError> {
+    //     if !cap.is_eth() {
+    //         return Err(RLPxError::InternalError(
+    //             "The protocol should be eth".into(),
+    //         ));
+    //     }
+    //     if self.p2p_protocol.is_none() {
+    //         return Err(RLPxError::InternalError(
+    //             "p2p protocol should be established first".into(),
+    //         ));
+    //     }
+    //     self.eth_protocol = Some(cap.clone());
+    //     Ok(())
+    // }
+
+    // pub fn set_snap_protocol(&mut self, cap: &Capability) -> Result<(), RLPxError> {
+    //     if !cap.is_snap() {
+    //         return Err(RLPxError::InternalError(
+    //             "The protocol should be snap".into(),
+    //         ));
+    //     }
+    //     if self.eth_protocol.is_none() {
+    //         return Err(RLPxError::InternalError(
+    //             "Eth protocol should be established first".into(),
+    //         ));
+    //     }
+    //     self.snap_protocol = Some(cap.clone());
+    //     Ok(())
+    // }
 }
 
 impl Decoder for RLPxCodec {
@@ -193,7 +239,18 @@ impl Decoder for RLPxCodec {
         let (frame_data, _padding) = frame_ciphertext.split_at(frame_size);
 
         let (msg_id, msg_data): (u8, _) = RLPDecode::decode_unfinished(frame_data)?;
-        Ok(Some(rlpx::Message::decode(msg_id, msg_data)?))
+
+        // NOTE: this crashes since it is not possible to create a runtime within an async context.
+        // Find the right way to lock the capabilities mutex
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let capabilities = rt.block_on(async { self.capabilities.lock().await });
+        Ok(Some(rlpx::Message::decode(
+            msg_id,
+            msg_data,
+            &capabilities.p2p,
+            &capabilities.eth,
+            &capabilities.snap,
+        )?))
     }
 
     fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
@@ -222,7 +279,17 @@ impl Encoder<rlpx::Message> for RLPxCodec {
 
     fn encode(&mut self, message: rlpx::Message, buffer: &mut BytesMut) -> Result<(), Self::Error> {
         let mut frame_data = vec![];
-        message.encode(&mut frame_data)?;
+
+        // NOTE: this crashes since it is not possible to create a runtime within an async context.
+        // Find the right way to lock the capabilities mutex
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let capabilities = rt.block_on(async { self.capabilities.lock().await });
+        message.encode(
+            &mut frame_data,
+            &capabilities.p2p,
+            &capabilities.eth,
+            &capabilities.snap,
+        )?;
 
         let mac_aes_cipher = Aes256Enc::new_from_slice(&self.mac_key.0)?;
 

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

@@ -6,7 +6,7 @@ use std::{
 
 use crate::{
     rlpx::{
-        connection::server::{Established, InnerState},
+        connection::server::{Capabilities, Established, InnerState},
         error::RLPxError,
         utils::{
             compress_pubkey, decompress_pubkey, ecdh_xchng, kdf, log_peer_debug, sha256,
@@ -64,6 +64,7 @@ pub(crate) struct LocalState {
 pub(crate) async fn perform(
     state: InnerState,
 ) -> Result<(Established, SplitStream<Framed<TcpStream, RLPxCodec>>), RLPxError> {
+    let capabilities = Arc::new(Mutex::new(Capabilities::default()));
     let (context, node, framed, inbound) = match state {
         InnerState::Initiator(Initiator { context, node }) => {
             let addr = SocketAddr::new(node.ip, node.tcp_port);
@@ -81,7 +82,12 @@ pub(crate) async fn perform(
             // keccak256(nonce || initiator-nonce)
             let hashed_nonces: [u8; 32] =
                 Keccak256::digest([remote_state.nonce.0, local_state.nonce.0].concat()).into();
-            let codec = RLPxCodec::new(&local_state, &remote_state, hashed_nonces)?;
+            let codec = RLPxCodec::new(
+                &local_state,
+                &remote_state,
+                hashed_nonces,
+                capabilities.clone(),
+            )?;
             log_peer_debug(&node, "Completed handshake as initiator");
             (context, node, Framed::new(stream, codec), false)
         }
@@ -99,7 +105,12 @@ pub(crate) async fn perform(
             // keccak256(nonce || initiator-nonce)
             let hashed_nonces: [u8; 32] =
                 Keccak256::digest([local_state.nonce.0, remote_state.nonce.0].concat()).into();
-            let codec = RLPxCodec::new(&local_state, &remote_state, hashed_nonces)?;
+            let codec = RLPxCodec::new(
+                &local_state,
+                &remote_state,
+                hashed_nonces,
+                capabilities.clone(),
+            )?;
             let node = Node::new(
                 peer_addr.ip(),
                 peer_addr.port(),
@@ -122,8 +133,6 @@ pub(crate) async fn perform(
             storage: context.storage.clone(),
             blockchain: context.blockchain.clone(),
             capabilities: vec![],
-            negotiated_eth_capability: None,
-            negotiated_snap_capability: None,
             last_block_range_update_block: 0,
             broadcasted_txs: HashSet::new(),
             requested_pooled_txs: HashMap::new(),
@@ -132,6 +141,7 @@ pub(crate) async fn perform(
             table: context.table.clone(),
             backend_channel: None,
             inbound,
+            negotiated_capabilities: capabilities,
         },
         stream,
     ))