reth_eth_wire/
handshake.rs

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

reth_eth_wire/
handshake.rs
