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 65447912bf
commit 732daf0578
2 changed files with 89 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -27,7 +27,6 @@ use crate::{
}; };
use std::collections::HashSet; use std::collections::HashSet;
use std::{ use std::{
clone,
io::Error, io::Error,
net::{IpAddr, Ipv4Addr, UdpSocket}, net::{IpAddr, Ipv4Addr, UdpSocket},
time::Duration, time::Duration,
@@ -43,12 +42,12 @@ pub struct P2PSharedData {
shared_messageslist: Arc<Mutex<HashMap<i32, EventType>>>, shared_messageslist: Arc<Mutex<HashMap<i32, EventType>>>,
shared_senders: Arc<MultipleSenders>, shared_senders: Arc<MultipleSenders>,
server_name: Arc<Mutex<String>>, server_name: Arc<Mutex<String>>,
server_address: Arc<Mutex<String>>,
handshake_peers: Arc<HandshakeHistory>, handshake_peers: Arc<HandshakeHistory>,
threads: Vec<Worker>, threads: Vec<Worker>,
} }
use bytes::Bytes; use bytes::Bytes;
use p256::pkcs8::der::pem::Base64Encoder;
use reqwest::Client; use reqwest::Client;
impl P2PSharedData { impl P2PSharedData {
@@ -69,6 +68,7 @@ impl P2PSharedData {
let senders = MultipleSenders::new(1, &shared_socket, cmd_tx, &mut threads); let senders = MultipleSenders::new(1, &shared_socket, cmd_tx, &mut threads);
let shared_senders = Arc::new(senders); let shared_senders = Arc::new(senders);
let server_name = Arc::new(Mutex::new("".to_string())); 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()); let handhsake_peers = Arc::new(HandshakeHistory::new());
Ok(P2PSharedData { Ok(P2PSharedData {
@@ -77,6 +77,7 @@ impl P2PSharedData {
shared_messageslist: shared_messageslist, shared_messageslist: shared_messageslist,
shared_senders: shared_senders, shared_senders: shared_senders,
server_name: server_name, server_name: server_name,
server_address: server_address,
handshake_peers: handhsake_peers, handshake_peers: handhsake_peers,
threads, threads,
}) })
@@ -95,10 +96,18 @@ impl P2PSharedData {
let guard = self.server_name.lock().unwrap(); let guard = self.server_name.lock().unwrap();
guard.to_string() 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) { pub fn set_servername(&self, new: String) {
let mut guard = self.server_name.lock().unwrap(); let mut guard = self.server_name.lock().unwrap();
*guard = new *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> { pub fn senders(&self) -> Arc<MultipleSenders> {
self.shared_senders.clone() self.shared_senders.clone()
} }
@@ -496,7 +505,8 @@ pub fn start_p2p_executor(
sd.add_message(id, EventType::Ping); sd.add_message(id, EventType::Ping);
let pingrequest = let pingrequest =
construct_message(PING, Vec::new(), id, sd.cryptopair_ref()); 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 { match peer_address {
Some(addr) => { Some(addr) => {
if let Some(ping) = pingrequest { if let Some(ping) = pingrequest {
@@ -542,9 +552,16 @@ pub fn start_p2p_executor(
if let Some(sd) = shared_data.as_ref() { if let Some(sd) = shared_data.as_ref() {
println!("username:{}, ip:{}", username, ip); println!("username:{}, ip:{}", username, ip);
// user server to send nattraversal request // user server to send nattraversal request
let server_addr_query = let server_addr_query = get_socket_address(
get_socket_address(sd.servername().clone(), ip.clone()); sd.servername().clone(),
let peer_addr_query = get_socket_address(username.clone(), ip.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 { match server_addr_query.await {
Some(server_addr) => match peer_addr_query.await { Some(server_addr) => match peer_addr_query.await {
@@ -634,7 +651,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 /// 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() { let client = match Client::builder().timeout(Duration::from_secs(5)).build() {
Ok(c) => c, Ok(c) => c,
Err(e) => { Err(e) => {
@@ -680,14 +703,29 @@ pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAd
let addresses = parse_addresses(&s); // assumes parse_addresses: &str -> Vec<SocketAddr> let addresses = parse_addresses(&s); // assumes parse_addresses: &str -> Vec<SocketAddr>
for addr in addresses { for addr in addresses {
if quick_ping(&addr, 300) { println!("trying address : {}", addr);
if quick_ping(&addr, 1000) {
return Some(addr); return Some(addr);
} }
// TODO: nat_traversal(&addr).await; // TODO: nat_traversal(&addr).await;
// after NAT traversal attempt, ping again // 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); return Some(addr);
} }
} }
@@ -695,32 +733,32 @@ pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAd
None None
} }
//pub async fn get_socket_address(username: String, ip: String) -> Option<SocketAddr> { pub async fn get_server_address(username: String, ip: String) -> Option<SocketAddr> {
// let client = Client::builder() let client = Client::builder()
// .timeout(Duration::from_secs(5)) .timeout(Duration::from_secs(5))
// .build() .build()
// .expect("cannot create client"); .expect("cannot create client");
// let uri = format!("{}/peers/{}/addresses", ip, username); let uri = format!("{}/peers/{}/addresses", ip, username);
// let res = client.get(uri).send().await.expect("couldnt get response"); let res = client.get(uri).send().await.expect("couldnt get response");
// if res.status().is_success() { if res.status().is_success() {
// println!("Successfully retreived the addresses. {}", res.status()); println!("Successfully retreived the addresses. {}", res.status());
// } else { } else {
// eprintln!( eprintln!(
// "Failed to get the peers addresses from the server. Status: {}", "Failed to get the peers addresses from the server. Status: {}",
// res.status() res.status()
// ); );
// } }
// let body: Bytes = res.bytes().await.expect("couldnt get bytes"); let body: Bytes = res.bytes().await.expect("couldnt get bytes");
//
// match String::from_utf8(body.to_vec()) { match String::from_utf8(body.to_vec()) {
// Ok(s) => { Ok(s) => {
// let addresses = parse_addresses(&s); let addresses = parse_addresses(&s);
// if let Some(first) = addresses.first() { if let Some(first) = addresses.first() {
// Some(first.clone()) Some(first.clone())
// } else { } else {
// None None
// } }
// } }
// Err(_) => None, Err(_) => None,
// } }
//} }

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

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