feat: connect to bootnodes and request peer data (#118)

ricomateo · MegaRedHand · web-flow · commit e54f466faca3 · 2024-07-08T15:34:09.000-03:00
**Motivation** Build the functionality to join the network **Description** Closes #86 --------- Co-authored-by: Tomás Grüner <47506558+MegaRedHand@users.noreply.github.com>
diff --git a/crates/net/src/discv4.rs b/crates/net/src/discv4.rs
@@ -6,7 +6,7 @@ use ethereum_rust_core::rlp::{
     structs::{self, Decoder, Encoder},
 };
 use ethereum_rust_core::{H256, H512};
-use k256::ecdsa::{signature::Signer, SigningKey};
+use k256::ecdsa::SigningKey;
 use std::net::IpAddr;
 
 #[derive(Debug, Eq, PartialEq)]
@@ -25,7 +25,7 @@ pub(crate) enum Message {
 }
 
 impl Message {
-    pub fn encode_with_header(&self, buf: &mut dyn BufMut, node_signer: SigningKey) {
+    pub fn encode_with_header(&self, buf: &mut dyn BufMut, node_signer: &SigningKey) {
         let signature_size = 65_usize;
         let mut data: Vec<u8> = Vec::with_capacity(signature_size.next_power_of_two());
         data.resize(signature_size, 0);
@@ -34,7 +34,9 @@ impl Message {
 
         let digest = keccak_hash::keccak_buffer(&mut &data[signature_size..]).unwrap();
 
-        let (signature, recovery_id) = node_signer.try_sign(&digest.0).expect("failed to sign");
+        let (signature, recovery_id) = node_signer
+            .sign_prehash_recoverable(&digest.0)
+            .expect("failed to sign");
         let b = signature.to_bytes();
 
         data[..signature_size - 1].copy_from_slice(&b);
@@ -91,6 +93,11 @@ impl Message {
             _ => Err(RLPDecodeError::MalformedData),
         }
     }
+    // TODO: we can put the hash, signature and the message (as payload) inside a new struct
+    pub fn get_hash(encoded_msg: &[u8]) -> &[u8] {
+        let hash_len = 32;
+        &encoded_msg[..hash_len]
+    }
 
     fn packet_type(&self) -> u8 {
         match self {
@@ -142,7 +149,7 @@ pub(crate) struct PingMessage {
     /// The Ping message version. Should be set to 4, but mustn't be enforced.
     version: u8,
     /// The endpoint of the sender.
-    from: Endpoint,
+    pub from: Endpoint,
     /// The endpoint of the receiver.
     to: Endpoint,
     /// The expiration time of the message. If the message is older than this time,
@@ -346,14 +353,13 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer);
+        msg.encode_with_header(&mut buf, &signer);
         let result = to_hex(&buf);
-        let hash = "d9b83d9701c6481a99db908b19551c6b082bcb28d5bef44cfa55256bc7977500";
-        let signature = "f0bff907b5c432e623ba5d3803d6a405bdbaffdfc0373499ac2a243ef3ab52de3a5312c0a9a96593979b746a4cd37ebdf21cf6971cf8c10c94f4d45c1a0f90dd00";
+        let hash = "d2821841963050aa505c00d8e4fd2d016f95eff739b784e0e26587a58226738e";
+        let signature = "8a73f13d613c0ba5148787bb52fd04eb984c3dae486bac19433adf658d29bbb352f3acf2d55f2bdae3afff5298723114581e3f34c37815b32b9195a3326dd68700";
         let pkt_type = "01";
         let encoded_message = "dd04cb840102030482064d8218dbc984ffff0205820bf780850400e78bba";
         let expected = [hash, signature, pkt_type, encoded_message].concat();
-
         assert_eq!(result, expected);
     }
 
@@ -370,17 +376,18 @@ mod tests {
                 .unwrap();
         let enr_seq = 1704896740573;
         let msg = Message::Pong(PongMessage::new(to, ping_hash, expiration).with_enr_seq(enr_seq));
+
         let key_bytes =
             H256::from_str("577d8278cc7748fad214b5378669b420f8221afb45ce930b7f22da49cbc545f3")
                 .unwrap();
         let signer = SigningKey::from_slice(key_bytes.as_bytes()).unwrap();
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer);
+        msg.encode_with_header(&mut buf, &signer);
         let result = to_hex(&buf);
-        let hash = "852ef38c2087413400cb33215709a8cfa6f274929e91704ec27a1ae4d226f85d";
-        let signature = "a7ab61ec963f779d10918c9bc3c3243c05f45eabbd078e90bf78313904e1c91201a03e78a133c2676e1c2686601e70ab1ec7aa602ad7f65bb468e52367d7123c00";
+        let hash = "9657e4e2db33b51cbbeb503bd195efcf081d6a83befbb42b4be95d0f7bf27ffe";
+        let signature = "b1a91caa6105b941d3ecce052dcfea5e4f4290c9e6a89ff72707a8b5116ee87a1ea3fa0086990cd862a8a2347f346f1b118122a28bf2ed2ca371d2c493a86bde01";
         let pkt_type = "02";
         let msg = "f7c984bebfbc3982765f80a03e1bf98f025f98d54ed2f61bbef63b6b46f50e12d7b937d6bdea19afd640be2384667d9af086018cf3c3bcdd";
         let expected = [hash, signature, pkt_type, msg].concat();
@@ -407,14 +414,13 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer);
+        msg.encode_with_header(&mut buf, &signer);
         let result = to_hex(&buf);
-        let hash = "cccfa9bf8e49603f8cc5381579d435bd322d386091732e3da7f6b7df13172b92";
-        let signature = "b1caebcd4d754552be21df4a100bd4ccd85e9d95b2e29b29db2df681c17c370068e410ea31e7106081c2ed39489c1762125cbd34477b41d940d230d1d3888a4101";
+        let hash = "58a1d0ea66afd9617c198b60a7417637ae27b847b004dbebc1e29d4067327e35";
+        let signature = "e1988832d7d7b73925ec584ff818ff3a7bffe1a84fe3835923c3ab17af40071f7c9263176203c80c6ed77f0586479b78884e9e47fdb3287d2aafa92348e5c16700";
         let pkt_type = "02";
         let msg = "f0c984bebfbc3982765f80a03e1bf98f025f98d54ed2f61bbef63b6b46f50e12d7b937d6bdea19afd640be2384667d9af0";
         let expected = [hash, signature, pkt_type, msg].concat();
-
         assert_eq!(result, expected);
     }
 
@@ -432,13 +438,14 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer);
+        msg.encode_with_header(&mut buf, &signer);
         let result = to_hex(&buf);
-        let hash = "dabe1b1a4dc26324120594091bb94dbdd4aa98326cbfecb60a4a67778ebb0e67";
-        let signature = "28cf71e9a929fa29f4e528f5fdc96e25a3086a777c8967710ddc1ef5f456bae40808baf0840d0449ea5a17ba11bbb012719e679cf3cf8d137106884c393ef15b00";
+        let hash = "23770430fc208bdc78bc77052bf7ec2e928b38c13c085b87491c15ebebb2050f";
+        let signature = "7c98bb4759569117031a9fbbeb00314d018eba55135c65ee98dbf6871aaebe61225f36b36e4f60da5b5d6c917e3589dd235acfacc6de4dade116c4bb851b884b01";
         let pkt_type = "03";
         let encoded_message = "f848b840d860a01f9722d78051619d1e2351aba3f43f943f6f00718d1b9baa4101932a1f5011f16bb2b1bb35db20d6fe28fa0bf09636d26a87d31de9ec6203eeedb1f666850400e78bba";
         let expected = [hash, signature, pkt_type, encoded_message].concat();
+
         assert_eq!(result, expected);
     }
 
@@ -526,7 +533,7 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer.clone());
+        msg.encode_with_header(&mut buf, &signer);
         let result = Message::decode_with_header(&buf).expect("Failed decoding PingMessage");
         assert_eq!(result, msg);
     }
@@ -556,7 +563,7 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer.clone());
+        msg.encode_with_header(&mut buf, &signer);
         let result = Message::decode_with_header(&buf).expect("Failed decoding PingMessage");
         assert_eq!(result, msg);
     }
@@ -574,7 +581,7 @@ mod tests {
 
         let mut buf = Vec::new();
 
-        msg.encode_with_header(&mut buf, signer);
+        msg.encode_with_header(&mut buf, &signer);
         let result = Message::decode_with_header(&buf).unwrap();
         assert_eq!(result, msg);
     }
diff --git a/crates/net/src/lib.rs b/crates/net/src/lib.rs
@@ -1,53 +1,79 @@
 pub(crate) mod discv4;
-
+use discv4::{Endpoint, FindNodeMessage, Message, PingMessage, PongMessage};
+use ethereum_rust_core::H512;
+use k256::elliptic_curve::sec1::ToEncodedPoint;
+use k256::elliptic_curve::PublicKey;
+use k256::{ecdsa::SigningKey, elliptic_curve::rand_core::OsRng};
+use keccak_hash::H256;
 use std::{
     net::SocketAddr,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
-
-use discv4::{Endpoint, Message, PingMessage};
-use k256::{ecdsa::SigningKey, elliptic_curve::rand_core::OsRng};
 use tokio::{
     net::{TcpSocket, UdpSocket},
     try_join,
 };
 use tracing::info;
+use types::BootNode;
 pub mod types;
 
 const MAX_DISC_PACKET_SIZE: usize = 1280;
 
-pub async fn start_network(udp_addr: SocketAddr, tcp_addr: SocketAddr) {
+pub async fn start_network(udp_addr: SocketAddr, tcp_addr: SocketAddr, bootnodes: Vec<BootNode>) {
     info!("Starting discovery service at {udp_addr}");
     info!("Listening for requests at {tcp_addr}");
 
-    let discovery_handle = tokio::spawn(discover_peers(udp_addr));
+    let discovery_handle = tokio::spawn(discover_peers(udp_addr, bootnodes));
     let server_handle = tokio::spawn(serve_requests(tcp_addr));
     try_join!(discovery_handle, server_handle).unwrap();
 }
 
-async fn discover_peers(udp_addr: SocketAddr) {
+async fn discover_peers(udp_addr: SocketAddr, bootnodes: Vec<BootNode>) {
     let udp_socket = UdpSocket::bind(udp_addr).await.unwrap();
-    // This is just a placeholder example. The address is a known bootnode.
-    let receiver_addr: SocketAddr = ("138.197.51.181:30303").parse().unwrap();
-    let mut buf = vec![0; MAX_DISC_PACKET_SIZE];
+    let signer = SigningKey::random(&mut OsRng);
+    let bootnode = match bootnodes.first() {
+        Some(b) => b,
+        None => {
+            return;
+        }
+    };
 
-    ping(&udp_socket, udp_addr, receiver_addr).await;
+    ping(&udp_socket, udp_addr, bootnode.socket_address, &signer).await;
 
-    let (read, from) = udp_socket.recv_from(&mut buf).await.unwrap();
-    info!("Received {read} bytes from {from}");
-    let msg = Message::decode_with_header(&buf[..read]);
-    info!("Message: {:?}", msg);
+    let mut buf = vec![0; MAX_DISC_PACKET_SIZE];
+    loop {
+        let (read, from) = udp_socket.recv_from(&mut buf).await.unwrap();
+        let packet_type = buf[32 + 65];
+        match packet_type {
+            0x04 => {
+                info!("Received NEIGHBORS message from {from}");
+            }
+            _ => {
+                let msg = Message::decode_with_header(&buf[..read]).unwrap();
+                info!("Received {read} bytes from {from}");
+                info!("Message: {:?}", msg);
+                if let Message::Ping(_) = msg {
+                    let ping_hash = H256::from_slice(Message::get_hash(&buf[..read]));
+                    pong(&udp_socket, from, ping_hash, &signer).await;
+                    find_node(&udp_socket, from, &signer).await;
+                }
+            }
+        }
+    }
 }
 
-async fn ping(socket: &UdpSocket, local_addr: SocketAddr, to_addr: SocketAddr) {
+async fn ping(
+    socket: &UdpSocket,
+    local_addr: SocketAddr,
+    to_addr: SocketAddr,
+    signer: &SigningKey,
+) {
     let mut buf = Vec::new();
 
     let expiration: u64 = (SystemTime::now() + Duration::from_secs(10))
         .duration_since(UNIX_EPOCH)
         .unwrap()
-        .as_millis()
-        .try_into()
-        .unwrap();
+        .as_secs();
 
     // TODO: this should send our advertised TCP port
     let from = Endpoint {
@@ -61,10 +87,49 @@ async fn ping(socket: &UdpSocket, local_addr: SocketAddr, to_addr: SocketAddr) {
         tcp_port: 0,
     };
 
-    let msg: discv4::Message = discv4::Message::Ping(PingMessage::new(from, to, expiration));
-    let signer = SigningKey::random(&mut OsRng);
+    let ping: discv4::Message = discv4::Message::Ping(PingMessage::new(from, to, expiration));
+
+    ping.encode_with_header(&mut buf, signer);
+    socket.send_to(&buf, to_addr).await.unwrap();
+}
+
+async fn find_node(socket: &UdpSocket, to_addr: SocketAddr, signer: &SigningKey) {
+    let public_key = PublicKey::from(signer.verifying_key());
+    let encoded = public_key.to_encoded_point(false);
+    let bytes = encoded.as_bytes();
+    debug_assert_eq!(bytes[0], 4);
+
+    let target = H512::from_slice(&bytes[1..]);
+
+    let expiration: u64 = (SystemTime::now() + Duration::from_secs(10))
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
 
+    let msg: discv4::Message = discv4::Message::FindNode(FindNodeMessage::new(target, expiration));
+
+    let mut buf = Vec::new();
     msg.encode_with_header(&mut buf, signer);
+
+    socket.send_to(&buf, to_addr).await.unwrap();
+}
+
+async fn pong(socket: &UdpSocket, to_addr: SocketAddr, ping_hash: H256, signer: &SigningKey) {
+    let mut buf = Vec::new();
+
+    let expiration: u64 = (SystemTime::now() + Duration::from_secs(10))
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+
+    let to = Endpoint {
+        ip: to_addr.ip(),
+        udp_port: to_addr.port(),
+        tcp_port: 0,
+    };
+    let pong: discv4::Message = discv4::Message::Pong(PongMessage::new(to, ping_hash, expiration));
+
+    pong.encode_with_header(&mut buf, signer);
     socket.send_to(&buf, to_addr).await.unwrap();
 }
 
diff --git a/crates/net/src/types/bootnode.rs b/crates/net/src/types/bootnode.rs
@@ -1,7 +1,7 @@
 use ethereum_rust_core::H512;
 use std::{net::SocketAddr, num::ParseIntError, str::FromStr};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Copy)]
 pub struct BootNode {
     pub node_id: H512,
     pub socket_address: SocketAddr,
diff --git a/ethereum_rust/src/cli.rs b/ethereum_rust/src/cli.rs
@@ -1,4 +1,5 @@
 use clap::{Arg, ArgAction, Command};
+use ethereum_rust_net::types::BootNode;
 
 pub fn cli() -> Command {
     Command::new("ethereum_rust")
@@ -71,6 +72,7 @@ pub fn cli() -> Command {
             Arg::new("bootnodes")
                 .long("bootnodes")
                 .value_name("BOOTNODE_LIST")
+                .value_parser(clap::value_parser!(BootNode))
                 .value_delimiter(',')
                 .num_args(1..)
                 .action(ArgAction::Set),
diff --git a/ethereum_rust/src/main.rs b/ethereum_rust/src/main.rs
@@ -5,7 +5,6 @@ use std::{
     net::{SocketAddr, ToSocketAddrs},
 };
 use tokio::try_join;
-
 use tracing::{warn, Level};
 use tracing_subscriber::FmtSubscriber;
 mod cli;
@@ -50,8 +49,9 @@ async fn main() {
         .get_one::<String>("network")
         .expect("network is required");
 
-    let bootnodes: Vec<&BootNode> = matches
+    let bootnodes: Vec<BootNode> = matches
         .get_many("bootnodes")
+        .map(Iterator::copied)
         .map(Iterator::collect)
         .unwrap_or_default();
 
@@ -72,7 +72,7 @@ async fn main() {
     let _genesis = read_genesis_file(genesis_file_path);
 
     let rpc_api = ethereum_rust_rpc::start_api(http_socket_addr, authrpc_socket_addr);
-    let networking = ethereum_rust_net::start_network(udp_socket_addr, tcp_socket_addr);
+    let networking = ethereum_rust_net::start_network(udp_socket_addr, tcp_socket_addr, bootnodes);
 
     try_join!(tokio::spawn(rpc_api), tokio::spawn(networking)).unwrap();
 }
