bob/p2p
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

work

Browse Source
This commit is contained in:
Tiago Batista Cardoso
2026-01-21 22:48:05 +01:00
parent 271fdbbb4c
commit 424c11c5aa
2 changed files with 90 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
client-network/src/lib.rs
View File

@@ -15,16 +15,13 @@ use crate::{
cryptographic_signature::CryptographicSignature,
message_handling::EventType,
messages_channels::{MultipleSenders, start_receving_thread, start_retry_thread},
messages_structure::{
NATTRAVERSALREQUEST, NATTRAVERSALREQUEST2, PING, ROOTREQUEST, construct_message,
},
messages_structure::{NATTRAVERSALREQUEST, PING, ROOTREQUEST, construct_message},
peers_refresh::HandshakeHistory,
registration::{parse_addresses, perform_handshake, register_with_the_server},
server_communication::{generate_id, get_peer_list},
threads_handling::Worker,
};
use std::{
clone,
io::Error,
net::{IpAddr, Ipv4Addr, UdpSocket},
time::Duration,
@@ -40,12 +37,12 @@ pub struct P2PSharedData {
shared_messageslist: Arc<Mutex<HashMap<i32, EventType>>>,
shared_senders: Arc<MultipleSenders>,
server_name: Arc<Mutex<String>>,
server_address: Arc<Mutex<String>>,
handshake_peers: Arc<HandshakeHistory>,
threads: Vec<Worker>,
}
use bytes::Bytes;
use p256::pkcs8::der::pem::Base64Encoder;
use reqwest::Client;
impl P2PSharedData {
@@ -66,6 +63,7 @@ impl P2PSharedData {
let senders = MultipleSenders::new(1, &shared_socket, cmd_tx, &mut threads);
let shared_senders = Arc::new(senders);
let server_name = Arc::new(Mutex::new("".to_string()));
let server_address = Arc::new(Mutex::new("".to_string()));
let handhsake_peers = Arc::new(HandshakeHistory::new());
Ok(P2PSharedData {
@@ -74,6 +72,7 @@ impl P2PSharedData {
shared_messageslist: shared_messageslist,
shared_senders: shared_senders,
server_name: server_name,
server_address: server_address,
handshake_peers: handhsake_peers,
threads,
})
@@ -92,10 +91,18 @@ impl P2PSharedData {
let guard = self.server_name.lock().unwrap();
guard.to_string()
}
pub fn serveraddress(&self) -> String {
let guard = self.server_address.lock().unwrap();
guard.to_string()
}
pub fn set_servername(&self, new: String) {
let mut guard = self.server_name.lock().unwrap();
*guard = new
}
pub fn set_serveraddress(&self, new: String) {
let mut guard = self.server_address.lock().unwrap();
*guard = new
}
pub fn senders(&self) -> Arc<MultipleSenders> {
self.shared_senders.clone()
}
@@ -378,7 +385,8 @@ pub fn start_p2p_executor(
sd.add_message(id, EventType::Ping);
let pingrequest =
construct_message(PING, Vec::new(), id, sd.cryptopair_ref());
let peer_address = get_socket_address(str, ip).await;
let peer_address =
get_socket_address(str, ip, shared_data.as_ref()).await;
match peer_address {
Some(addr) => {
if let Some(ping) = pingrequest {
@@ -424,9 +432,16 @@ pub fn start_p2p_executor(
if let Some(sd) = shared_data.as_ref() {
println!("username:{}, ip:{}", username, ip);
// user server to send nattraversal request
let server_addr_query =
get_socket_address(sd.servername().clone(), ip.clone());
let peer_addr_query = get_socket_address(username.clone(), ip.clone());
let server_addr_query = get_socket_address(
sd.servername().clone(),
ip.clone(),
shared_data.as_ref(),
);
let peer_addr_query = get_socket_address(
username.clone(),
ip.clone(),
shared_data.as_ref(),
);
match server_addr_query.await {
Some(server_addr) => match peer_addr_query.await {
@@ -516,7 +531,13 @@ fn quick_ping(addr: &SocketAddr, timeout_ms: u64) -> bool {
/// sends a get request to the server to get the socket address of the given peer
///
pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAddr> {
pub async fn get_socket_address(
username: String,
ip: String,
shared_data: Option<&P2PSharedData>,
) -> Option<SocketAddr> {
let sd = shared_data.expect("No shared data");
let client = match Client::builder().timeout(Duration::from_secs(5)).build() {
Ok(c) => c,
Err(e) => {
@@ -562,14 +583,29 @@ pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAd
let addresses = parse_addresses(&s); // assumes parse_addresses: &str -> Vec<SocketAddr>
for addr in addresses {
if quick_ping(&addr, 300) {
println!("trying address : {}", addr);
if quick_ping(&addr, 1000) {
return Some(addr);
}
// TODO: nat_traversal(&addr).await;
// after NAT traversal attempt, ping again
if quick_ping(&addr, 300) {
let natreq = construct_message(
NATTRAVERSALREQUEST,
addr.to_string().into_bytes(),
generate_id(),
&sd.shared_cryptopair,
);
sd.senders_ref().send_dispatch(
natreq.expect("couldnt construct message nattraversalrequest2"),
sd.serveraddress(),
false,
sd.messages_list().clone(),
);
if quick_ping(&addr, 1500) {
return Some(addr);
}
}
@@ -577,32 +613,32 @@ pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAd
None
}
//pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAddr> {
// let client = Client::builder()
// .timeout(Duration::from_secs(5))
// .build()
// .expect("cannot create client");
// let uri = format!("{}/peers/{}/addresses", ip, username);
// let res = client.get(uri).send().await.expect("couldnt get response");
// if res.status().is_success() {
// println!("Successfully retreived the addresses. {}", res.status());
// } else {
// eprintln!(
// "Failed to get the peers addresses from the server. Status: {}",
// res.status()
// );
// }
// let body: Bytes = res.bytes().await.expect("couldnt get bytes");
//
// match String::from_utf8(body.to_vec()) {
// Ok(s) => {
// let addresses = parse_addresses(&s);
// if let Some(first) = addresses.first() {
// Some(first.clone())
// } else {
// None
// }
// }
// Err(_) => None,
// }
//}
pub async fn get_server_address(username: String, ip: String) -> Option<SocketAddr> {
let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.expect("cannot create client");
let uri = format!("{}/peers/{}/addresses", ip, username);
let res = client.get(uri).send().await.expect("couldnt get response");
if res.status().is_success() {
println!("Successfully retreived the addresses. {}", res.status());
} else {
eprintln!(
"Failed to get the peers addresses from the server. Status: {}",
res.status()
);
}
let body: Bytes = res.bytes().await.expect("couldnt get bytes");
match String::from_utf8(body.to_vec()) {
Ok(s) => {
let addresses = parse_addresses(&s);
if let Some(first) = addresses.first() {
Some(first.clone())
} else {
None
}
}
Err(_) => None,
}
}

27
client-network/src/registration.rs
View File

@@ -1,19 +1,14 @@
use bytes::Bytes;
use crate::NetworkEvent;
use crate::P2PSharedData;
use crate::cryptographic_signature::{CryptographicSignature, formatPubKey, sign_message};
use crate::get_socket_address;
use crate::cryptographic_signature::CryptographicSignature;
use crate::get_server_address;
use crate::message_handling::EventType;
use crate::messages_channels::MultipleSenders;
use crate::messages_structure::construct_message;
use crate::server_communication::generate_id;
use crossbeam_channel::{Receiver, Sender};
use std::collections::HashMap;
use crossbeam_channel::Sender;
use std::net::SocketAddr;
use std::net::UdpSocket;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::sync::Arc;
///
/// sends the cryptographic signature to the server using a PUT request over the HTTP API.
@@ -29,8 +24,13 @@ pub async fn register_with_the_server(
let pubkey_bytes_minus = pubkey_bytes[1..].to_vec();
let res = client.put(uri).body(pubkey_bytes_minus).send().await?;
let res = res.error_for_status()?;
println!("register ip adresses");
Ok(())
match res.error_for_status() {
Ok(_) => {
println!("register ip adresses");
Ok(())
}
Err(e) => Err(e),
}
}
pub fn parse_addresses(input: &String) -> Vec<SocketAddr> {
@@ -62,10 +62,11 @@ pub async fn perform_handshake(
let senders = sd.senders_ref();
let id = generate_id();
let server_addr_query = get_socket_address(username.clone(), ip.clone());
let server_addr_query = get_server_address(username.clone(), ip.clone());
match server_addr_query.await {
Some(sockaddr_bytes) => {
sd.set_servername(username);
sd.set_serveraddress(sockaddr_bytes.to_string());
// first: &SocketAddr
let mut payload = Vec::new();
payload.extend_from_slice(&0u32.to_be_bytes());
@@ -115,8 +116,6 @@ pub async fn perform_handshake(
#[cfg(test)]
mod tests {
// Note this useful idiom: importing names from outer (for mod tests) scope.
use super::*;
/*///
/// does the procedure to register with the server