reth_eth_wire/
handshake.rs

1use crate::{
2    errors::{EthHandshakeError, EthStreamError, P2PStreamError},
3    message::MAX_MESSAGE_SIZE,
4    p2pstream::log_prefix,
5    CanDisconnect,
6};
7use alloy_primitives::hex;
8use bytes::{Bytes, BytesMut};
9use futures::{Sink, SinkExt, Stream};
10use reth_eth_wire_types::{
11    DisconnectReason, EthMessage, EthNetworkPrimitives, ProtocolMessage, StatusMessage,
12    UnifiedStatus,
13};
14use reth_ethereum_forks::ForkFilter;
15use reth_primitives_traits::GotExpected;
16use std::{fmt::Debug, future::Future, pin::Pin, time::Duration};
17use tokio::time::timeout;
18use tokio_stream::StreamExt;
19use tracing::{debug, trace};
20
21/// A trait that knows how to perform the P2P handshake.
22pub trait EthRlpxHandshake: Debug + Send + Sync + 'static {
23    /// Perform the P2P handshake for the `eth` protocol.
24    fn handshake<'a>(
25        &'a self,
26        unauth: &'a mut dyn UnauthEth,
27        status: UnifiedStatus,
28        fork_filter: ForkFilter,
29        timeout_limit: Duration,
30    ) -> Pin<Box<dyn Future<Output = Result<UnifiedStatus, EthStreamError>> + 'a + Send>>;
31}
32
33/// An unauthenticated stream that can send and receive messages.
34pub trait UnauthEth:
35    Stream<Item = Result<BytesMut, P2PStreamError>>
36    + Sink<Bytes, Error = P2PStreamError>
37    + CanDisconnect<Bytes>
38    + Send
39{
40}
41
42impl<T> UnauthEth for T where
43    T: Stream<Item = Result<BytesMut, P2PStreamError>>
44        + Sink<Bytes, Error = P2PStreamError>
45        + CanDisconnect<Bytes>
46        + Send
47{
48}
49
50/// The Ethereum P2P handshake.
51///
52/// This performs the regular ethereum `eth` rlpx handshake.
53#[derive(Debug, Default, Clone)]
54#[non_exhaustive]
55pub struct EthHandshake;
56
57impl EthRlpxHandshake for EthHandshake {
58    fn handshake<'a>(
59        &'a self,
60        unauth: &'a mut dyn UnauthEth,
61        status: UnifiedStatus,
62        fork_filter: ForkFilter,
63        timeout_limit: Duration,
64    ) -> Pin<Box<dyn Future<Output = Result<UnifiedStatus, EthStreamError>> + 'a + Send>> {
65        Box::pin(async move {
66            timeout(timeout_limit, EthereumEthHandshake(unauth).eth_handshake(status, fork_filter))
67                .await
68                .map_err(|_| EthStreamError::StreamTimeout)?
69        })
70    }
71}
72
73/// A type that performs the ethereum specific `eth` protocol handshake.
74#[derive(Debug)]
75pub struct EthereumEthHandshake<'a, S: ?Sized>(pub &'a mut S);
76
77impl<S, E> EthereumEthHandshake<'_, S>
78where
79    S: ?Sized + Stream<Item = Result<BytesMut, E>> + CanDisconnect<Bytes> + Send,
80    EthStreamError: From<E> + From<<S as Sink<Bytes>>::Error>,
81{
82    /// Performs the `eth` rlpx protocol handshake using the given input stream.
83    pub async fn eth_handshake(
84        self,
85        unified_status: UnifiedStatus,
86        fork_filter: ForkFilter,
87    ) -> Result<UnifiedStatus, EthStreamError> {
88        let unauth = self.0;
89
90        let status = unified_status.into_message();
91
92        // Send our status message
93        let status_msg = alloy_rlp::encode(ProtocolMessage::<EthNetworkPrimitives>::from(
94            EthMessage::Status(status),
95        ))
96        .into();
97        unauth.send(status_msg).await?;
98
99        // Receive peer's response
100        let their_msg_res = unauth.next().await;
101        let their_msg = match their_msg_res {
102            Some(Ok(msg)) => msg,
103            Some(Err(e)) => return Err(EthStreamError::from(e)),
104            None => {
105                unauth.disconnect(DisconnectReason::DisconnectRequested).await?;
106                return Err(EthStreamError::EthHandshakeError(EthHandshakeError::NoResponse));
107            }
108        };
109
110        if their_msg.len() > MAX_MESSAGE_SIZE {
111            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
112            return Err(EthStreamError::MessageTooBig(their_msg.len()));
113        }
114
115        let version = status.version();
116        let their_status_message = match ProtocolMessage::<EthNetworkPrimitives>::decode_status(
117            version,
118            &mut their_msg.as_ref(),
119        ) {
120            Ok(status) => status,
121            Err(err) => {
122                debug!(
123                    %err,
124                    len = their_msg.len(),
125                    msg = %hex::encode(log_prefix(&their_msg)),
126                    "decode error in eth handshake"
127                );
128                unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
129                return Err(EthStreamError::InvalidMessage(err));
130            }
131        };
132
133        trace!("Validating incoming ETH status from peer");
134
135        if status.genesis() != their_status_message.genesis() {
136            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
137            return Err(EthHandshakeError::MismatchedGenesis(
138                GotExpected { expected: status.genesis(), got: their_status_message.genesis() }
139                    .into(),
140            )
141            .into());
142        }
143
144        if status.version() != their_status_message.version() {
145            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
146            return Err(EthHandshakeError::MismatchedProtocolVersion(GotExpected {
147                got: their_status_message.version(),
148                expected: status.version(),
149            })
150            .into());
151        }
152
153        if *status.chain() != *their_status_message.chain() {
154            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
155            return Err(EthHandshakeError::MismatchedChain(GotExpected {
156                got: *their_status_message.chain(),
157                expected: *status.chain(),
158            })
159            .into());
160        }
161
162        // Ensure peer's total difficulty is reasonable
163        if let StatusMessage::Legacy(s) = &their_status_message &&
164            s.total_difficulty.bit_len() > 160
165        {
166            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
167            return Err(EthHandshakeError::TotalDifficultyBitLenTooLarge {
168                got: s.total_difficulty.bit_len(),
169                maximum: 160,
170            }
171            .into());
172        }
173
174        // Fork validation
175        if let Err(err) = fork_filter
176            .validate(their_status_message.forkid())
177            .map_err(EthHandshakeError::InvalidFork)
178        {
179            unauth.disconnect(DisconnectReason::ProtocolBreach).await?;
180            return Err(err.into());
181        }
182
183        if let StatusMessage::Eth69(s) = &their_status_message {
184            if s.earliest > s.latest {
185                return Err(EthHandshakeError::EarliestBlockGreaterThanLatestBlock {
186                    got: s.earliest,
187                    latest: s.latest,
188                }
189                .into());
190            }
191
192            if s.blockhash.is_zero() {
193                return Err(EthHandshakeError::BlockhashZero.into());
194            }
195        }
196
197        Ok(UnifiedStatus::from_message(their_status_message))
198    }
199}
reth_eth_wire/handshake.rs

reth_eth_wire/
handshake.rs
