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

server handshake handling

Browse Source
This commit is contained in:
wikano
2026-01-05 02:48:58 +01:00
parent c748dfa71d
commit f51b8e999c
5 changed files with 96 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,10 +30,14 @@ pub struct P2PSharedData {
shared_cryptopair: Arc<CryptographicSignature>,
shared_messageslist: Arc<Mutex<HashMap<i32, EventType>>>,
shared_senders: Arc<MultipleSenders>,
server_name: Arc<Mutex<String>>,
}
impl P2PSharedData {
pub fn new(username: String) -> Result<P2PSharedData, Error> {
pub fn new(
username: String,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
) -> Result<P2PSharedData, Error> {
let messages_list = HashMap::<i32, EventType>::new();
let username = String::from(username);
let crypto_pair = CryptographicSignature::new(username);
@@ -42,13 +46,15 @@ impl P2PSharedData {
let shared_cryptopair = Arc::new(crypto_pair);
let shared_messageslist = Arc::new(Mutex::new(messages_list));
let senders = MultipleSenders::new(1, &shared_socket);
let senders = MultipleSenders::new(1, &shared_socket, cmd_tx);
let shared_senders = Arc::new(senders);
let server_name = Arc::new(Mutex::new("".to_string()));
Ok(P2PSharedData {
shared_socket: shared_socket,
shared_cryptopair: shared_cryptopair,
shared_messageslist: shared_messageslist,
shared_senders: shared_senders,
server_name: server_name,
})
}
pub fn socket(&self) -> Arc<UdpSocket> {
@@ -61,6 +67,14 @@ impl P2PSharedData {
pub fn messages_list(&self) -> Arc<Mutex<HashMap<i32, EventType>>> {
self.shared_messageslist.clone()
}
pub fn servername(&self) -> String {
let guard = self.server_name.lock().unwrap();
guard.to_string()
}
pub fn set_servername(&self, new: String) {
let mut guard = self.server_name.lock().unwrap();
*guard = new
}
pub fn senders(&self) -> Arc<MultipleSenders> {
self.shared_senders.clone()
}
@@ -98,6 +112,7 @@ pub enum NetworkCommand {
RequestDirectoryContent(String, String),
RequestChunk(String, String),
Disconnect(),
ResetServerPeer(),
// ...
}
@@ -112,6 +127,8 @@ pub enum NetworkEvent {
FileTreeReceived(String, Vec<MerkleNode>), // peer_id, content
DataReceived(String, MerkleNode),
FileTreeRootReceived(String, String),
HandshakeFailed(),
ServerHandshakeFailed(String),
// ...
}
@@ -155,7 +172,7 @@ pub fn start_p2p_executor(
NetworkCommand::ServerHandshake(username, ip) => {
if let Some(sd) = shared_data.as_ref() {
println!("username:{}, ip:{}", username, ip);
let server_addr_query = get_socket_address(username, ip);
let server_addr_query = get_socket_address(username.clone(), ip);
match server_addr_query.await {
Ok(sockaddr_bytes) => {
@@ -163,6 +180,7 @@ pub fn start_p2p_executor(
Ok(s) => {
let addresses = parse_addresses(&s);
if let Some(first) = addresses.first() {
sd.set_servername(username);
// first: &SocketAddr
start_receving_thread(
sd,
@@ -230,7 +248,7 @@ pub fn start_p2p_executor(
// Actual server connection
shared_data = match P2PSharedData::new(name.clone()) {
shared_data = match P2PSharedData::new(name.clone(), event_tx.clone()) {
Ok(sd) => Some(sd),
Err(e) => {
let mut err_msg = String::from("failed to initialize socket: ");
@@ -311,6 +329,13 @@ pub fn start_p2p_executor(
println!("no p2p data");
}
}
NetworkCommand::ResetServerPeer() => {
if let Some(sd) = shared_data.as_ref() {
sd.set_servername("".to_string());
} else {
println!("no p2p data");
}
}
}
}

10
client-network/src/message_handling.rs
View File

@@ -1,4 +1,5 @@
use crate::{
NetworkEvent,
cryptographic_signature::{
CryptographicSignature, get_peer_key, sign_message, verify_signature,
},
@@ -45,6 +46,7 @@ pub fn handle_recevied_message(
crypto_pair: &CryptographicSignature,
socket_addr: &SocketAddr,
senders: &MultipleSenders,
server_name: &String,
) {
if recevied_message.len() < 4 {
return;
@@ -73,6 +75,7 @@ pub fn handle_recevied_message(
// Handle handshake
if message_type == 1 {
let mut resp_to_serv = false;
println!("verify the signature");
let parsed_received_message = HandshakeMessage::parse(recevied_message.to_vec());
let received_name = String::from_utf8(parsed_received_message.name).expect("error");
@@ -81,12 +84,17 @@ pub fn handle_recevied_message(
.block_on(get_peer_key(&received_name))
.expect("failed to retrieve public key");
if received_name == server_name.to_string() {
resp_to_serv = true;
}
if !verify_signature(peer_pubkey, recevied_message) {
println!(
"incorrect signature from given peer: {}, ignoring message {}",
&received_name, id
);
} else {
// verify if this is a server handshake request
let username_size = crypto_pair.username.len();
let hello_handshake = HandshakeMessage::helloReply(
id as u32,
@@ -96,7 +104,7 @@ pub fn handle_recevied_message(
//HandshakeMessage::display(&hello_handshake);
let hello_handshake_serialized = hello_handshake.serialize();
let message_signed = sign_message(crypto_pair, &hello_handshake_serialized);
senders.send_via(0, message_signed, socket_addr.to_string());
senders.send_via(0, message_signed, socket_addr.to_string(), resp_to_serv);
let mut list = messages_list.lock().expect("Failed to lock messages_list");
match list.get(&id) {
Some(_) => {

45
client-network/src/messages_channels.rs
View File

@@ -13,13 +13,17 @@ use std::thread;
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use crate::NetworkEvent;
pub struct MultipleSenders {
senders: Vec<Sender<Message>>,
response_channel: crossbeam_channel::Sender<NetworkEvent>,
}
pub struct Message {
payload: Vec<u8>,
address: String,
is_resp_to_server_handshake: bool,
}
struct RetryMessage {
@@ -68,12 +72,17 @@ impl MultipleSenders {
MultipleSenders { senders }
}*/
pub fn new(num_channels: usize, socket: &Arc<UdpSocket>) -> Self {
pub fn new(
num_channels: usize,
socket: &Arc<UdpSocket>,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
) -> Self {
let mut senders = Vec::new();
for i in 0..num_channels {
let (tx, rx) = mpsc::channel::<Message>();
let sock_clone = Arc::clone(&socket);
let cmd_tx_clone = cmd_tx.clone();
senders.push(tx);
@@ -91,6 +100,10 @@ impl MultipleSenders {
let mut item = queue.pop_front().unwrap();
match sock_clone.send_to(&item.msg.payload, &item.msg.address) {
Ok(_) => {
if (&item).msg.is_resp_to_server_handshake {
let res =
cmd_tx_clone.send(NetworkEvent::ConnectedHandshake());
}
let message_id: [u8; 4] =
item.msg.payload[0..4].try_into().expect("size error");
let id = i32::from_be_bytes(message_id);
@@ -103,10 +116,14 @@ impl MultipleSenders {
Err(e) => {
item.attempts += 1;
if item.attempts >= max_attempts {
eprintln!(
let str = format!(
"Abandon du message après {} tentatives sur canal {}: {}, address: {}",
item.attempts, i, e, item.msg.address
);
if (&item).msg.is_resp_to_server_handshake {
let res = cmd_tx_clone
.send(NetworkEvent::ServerHandshakeFailed(str));
}
} else {
// Backoff exponentiel simple
let backoff = Duration::from_millis(
@@ -131,6 +148,10 @@ impl MultipleSenders {
// On tente d'envoyer immédiatement
match sock_clone.send_to(&msg.payload, &msg.address) {
Ok(_) => {
if msg.is_resp_to_server_handshake {
let res =
cmd_tx_clone.send(NetworkEvent::ConnectedHandshake());
}
let message_id: [u8; 4] =
msg.payload[0..4].try_into().expect("size error");
let id = i32::from_be_bytes(message_id);
@@ -171,15 +192,29 @@ impl MultipleSenders {
});
}
MultipleSenders { senders }
MultipleSenders {
senders,
response_channel: cmd_tx.clone(),
}
}
/// Envoie un message via un canal spécifique (round-robin ou index précis)
pub fn send_via(&self, channel_idx: usize, data: Vec<u8>, remote_addr: String) {
pub fn send_via(
&self,
channel_idx: usize,
data: Vec<u8>,
remote_addr: String,
is_resp_to_server_handshake: bool,
) {
println!(
"is_resp_to_server_handshake {}",
is_resp_to_server_handshake
);
if let Some(sender) = self.senders.get(channel_idx) {
let _ = sender.send(Message {
payload: data,
address: remote_addr,
is_resp_to_server_handshake,
});
}
}
@@ -223,6 +258,7 @@ pub fn start_receving_thread(shared_data: &P2PSharedData, socket_addr: SocketAdd
let cryptopair_clone = shared_data.cryptopair();
let senders_clone = shared_data.senders();
let messages_clone = shared_data.messages_list();
let servername_clone = shared_data.servername();
thread::spawn(move || {
let mut buf = [0u8; 1024];
loop {
@@ -237,6 +273,7 @@ pub fn start_receving_thread(shared_data: &P2PSharedData, socket_addr: SocketAdd
&cryptopair_clone,
&socket_addr,
&senders_clone,
&servername_clone,
);
}
Err(e) => eprintln!("Erreur de réception: {}", e),

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

@@ -82,7 +82,7 @@ pub fn register_ip_addresses(
//HandshakeMessage::display(&hello_handshake);
let hello_handshake_serialized = hello_handshake.serialize();
let message_signed = sign_message(crypto_pair, &hello_handshake_serialized);
senders.send_via(0, message_signed, server_uri);
senders.send_via(0, message_signed, server_uri, false);
let mut list = messages_list.lock().expect("Failed to lock messages_list");
match list.get(&id) {
Some(_) => {