Files
p2p/client-network/src/message_handling.rs
2026-01-25 03:46:19 +01:00

521 lines
20 KiB
Rust

use crate::{
NetworkEvent, NodeHash,
cryptographic_signature::{CryptographicSignature, get_peer_key, verify_signature},
datum_parsing::parse_received_datum,
messages_channels::MultipleSenders,
messages_structure::construct_message,
peers_refresh::HandshakeHistory,
server_communication::generate_id,
timestamp::Timestamp,
};
use std::{
collections::HashMap,
default,
net::{Ipv4Addr, SocketAddr},
};
use std::{
net::IpAddr,
sync::{Arc, Mutex},
};
// Types of messages that await for a response
#[derive(Debug, Clone)]
pub enum EventType {
HelloThenRootRequest,
Hello,
RootRequest,
Ping,
NatTraversal,
DatumRequest,
DatumRequestBig,
Unknown,
}
impl EventType {
pub fn to_string(&self) -> String {
match self {
EventType::HelloThenRootRequest => "HelloThenRootRequest".to_owned(),
EventType::Hello => "Hello".to_owned(),
EventType::RootRequest => "RootRequest".to_owned(),
EventType::Ping => "Ping".to_owned(),
EventType::NatTraversal => "NatTraversal".to_owned(),
EventType::DatumRequest => "DatumRequest".to_owned(),
EventType::Unknown => "Unknown".to_owned(),
EventType::DatumRequestBig => "DatumRequestBig".to_owned(),
}
}
pub fn from_msgtype(msgtype: u8) -> EventType {
match msgtype {
PING => EventType::Ping,
HELLO => EventType::Hello,
ROOTREQUEST => EventType::RootRequest,
NATTRAVERSALREQUEST => EventType::NatTraversal,
DATUMREQUEST => EventType::DatumRequest,
_ => EventType::Unknown,
}
}
}
const ID: usize = 4;
const TYPE: usize = 5;
const LENGTH: usize = 7;
const EXTENSIONS: usize = 4;
const SIGNATURE: usize = 64;
pub const PING: u8 = 0;
const OK: u8 = 128;
const ERROR: u8 = 129;
const HELLO: u8 = 1;
const HELLOREPLY: u8 = 130;
const ROOTREQUEST: u8 = 2;
const ROOTREPLY: u8 = 131;
const DATUMREQUEST: u8 = 3;
const NODATUM: u8 = 133;
const DATUM: u8 = 132;
const NATTRAVERSALREQUEST: u8 = 4;
const NATTRAVERSALREQUEST2: u8 = 5;
pub fn handle_recevied_message(
messages_list: &Arc<Mutex<HashMap<i32, EventType>>>,
messages_received: &Arc<Mutex<HashMap<String, (EventType, Timestamp)>>>,
recevied_message: &Vec<u8>,
crypto_pair: &CryptographicSignature,
//socket_addr: &SocketAddr,
senders: &MultipleSenders,
server_name: &String,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
ip: SocketAddr,
handhsake_history: Arc<HandshakeHistory>,
) {
if recevied_message.len() < 4 {
return;
} // Basic safety check
let message_id: [u8; 4] = recevied_message[0..4].try_into().expect("size error");
let id = i32::from_be_bytes(message_id);
let mut is_resp_to_server_handshake = false;
if recevied_message[4] == HELLO {
let length_bytes: [u8; 2] = recevied_message[TYPE..LENGTH]
.try_into()
.expect("Taille incorrecte");
let msg_length = u16::from_be_bytes(length_bytes) as usize;
let ilength = u16::from_be_bytes(length_bytes);
let received_name = &recevied_message[LENGTH + EXTENSIONS..LENGTH + ilength as usize];
let name = String::from_utf8(received_name.to_vec()).expect("wrong name");
if name.clone() == server_name.clone() {
is_resp_to_server_handshake = true;
}
}
let resp = parse_message(
recevied_message.to_vec(),
id,
crypto_pair,
cmd_tx,
ip,
messages_list,
messages_received,
handhsake_history,
senders,
);
match resp {
None => {}
Some(resp_msg) => {
println!("msg_sent:{:?}", resp_msg);
senders.send_dispatch(
resp_msg,
ip.to_string(),
is_resp_to_server_handshake,
messages_list.clone(),
);
}
}
}
pub fn parse_message(
received_message: Vec<u8>,
id: i32,
crypto_pair: &CryptographicSignature,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
ip: SocketAddr,
messages_list: &Arc<Mutex<HashMap<i32, EventType>>>,
messages_received: &Arc<Mutex<HashMap<String, (EventType, Timestamp)>>>,
handhsake_history: Arc<HandshakeHistory>,
senders: &MultipleSenders,
) -> Option<Vec<u8>> {
let cmd_tx_clone = cmd_tx.clone();
let id_bytes: [u8; 4] = received_message[0..ID]
.try_into()
.expect("Taille incorrecte");
let msgtype = received_message[ID];
messages_received
.lock()
.expect("couldnt lock received map")
.insert(
ip.to_string(),
(EventType::from_msgtype(msgtype), Timestamp::now()),
);
let length_bytes: [u8; 2] = received_message[TYPE..LENGTH]
.try_into()
.expect("Taille incorrecte");
let msg_length = u16::from_be_bytes(length_bytes) as usize;
// verify signature
match msgtype {
HELLO | HELLOREPLY => {
let ilength = u16::from_be_bytes(length_bytes);
println!("hello");
println!("name received length: {}", ilength);
let received_name = &received_message[LENGTH + EXTENSIONS..LENGTH + ilength as usize];
let received_username = String::from_utf8(received_name.to_vec());
match received_username {
Ok(username) => {
let peer_pubkey =
match handhsake_history.get_peer_info_username(username.clone()) {
Some(peerinfo) => peerinfo.pubkey,
_ => tokio::runtime::Runtime::new()
.unwrap()
.block_on(get_peer_key(&username))
.expect("failed to retrieve public key"),
};
match msgtype {
HELLOREPLY => {
handhsake_history.add_new_handshake(peer_pubkey, "".to_string(), ip);
}
_ => {
println!("no handshake added");
}
}
if !verify_signature(peer_pubkey, &received_message) {
println!(
"incorrect signature from given peer: {}, ignoring message of type {} with id {}",
&username, received_message[ID], id
);
return None;
}
}
Err(e) => {
println!("incorrect name: {}", e);
return None;
}
}
}
ROOTREPLY | NODATUM | NATTRAVERSALREQUEST | NATTRAVERSALREQUEST2 => {
let ilength = u16::from_be_bytes(length_bytes);
println!("name received length: {}", ilength);
if let Some(peerinfo) = handhsake_history.get_peer_info_ip(ip.to_string()) {
if !verify_signature(peerinfo.pubkey, &received_message) {
println!(
"incorrect signature from given peer: {}, ignoring message of type {} with id {}",
&peerinfo.username, received_message[ID], id
);
return None;
} else {
println!("signature verified");
}
}
}
_ => {}
}
// Message handling
let mut constructed_message: Option<Vec<u8>> = None;
match msgtype {
PING => {
constructed_message = construct_message(OK, Vec::new(), id, crypto_pair);
}
OK => {
let mut guard = messages_list.lock().unwrap();
let res = guard.get(&id);
match res {
Some(ev) => {
println!("{:?}", ev);
let _ = &guard.remove_entry(&id);
println!("message {} retiré de la liste", id);
}
None => {
println!("ping non trouvé");
}
}
}
NATTRAVERSALREQUEST => {
// send ok & send nattraversalrequest2 to peer
constructed_message = construct_message(OK, Vec::new(), id, crypto_pair);
let ilength = u16::from_be_bytes(length_bytes);
let received_address =
&received_message[LENGTH + EXTENSIONS..LENGTH + ilength as usize];
let address = String::from_utf8(received_address.to_vec()).expect("wrong name");
let natreq2 = construct_message(
NATTRAVERSALREQUEST2,
ip.to_string().into_bytes(),
id,
crypto_pair,
);
senders.send_dispatch(
natreq2.expect("couldnt construct message nattraversalrequest2"),
address,
false,
messages_list.clone(),
);
}
NATTRAVERSALREQUEST2 => {
// send ok & send ping to peer
constructed_message = construct_message(OK, Vec::new(), id, crypto_pair);
let ilength = u16::from_be_bytes(length_bytes);
let received_address = &received_message[LENGTH..LENGTH + ilength as usize];
println!("received_address:{:?}", received_message);
//let addressv4 = IpAddr::V4(Ipv4Addr::from_octets(
// received_address[0..4].try_into().expect("incorrect size"),
//));
let bytes: [u8; 4] = received_address[0..4].try_into().expect("incorrect size");
let addr_v4 = Ipv4Addr::from(bytes);
let addressv4 = IpAddr::V4(addr_v4);
let address = SocketAddr::new(
addressv4,
u16::from_be_bytes(received_address[4..6].try_into().expect("incorrect size")),
);
println!("ip: {}", address);
let pingreq = construct_message(PING, Vec::new(), id, crypto_pair);
senders.send_dispatch(
constructed_message.expect("couldnt construct message ping request"),
ip.to_string(),
false,
messages_list.clone(),
);
senders.send_dispatch(
pingreq.expect("couldnt construct message ping request"),
address.to_string(),
false,
messages_list.clone(),
);
constructed_message = None;
}
ERROR => {
if let Ok(err_received) =
String::from_utf8(received_message[LENGTH..(msg_length + LENGTH)].to_vec())
{
let err_msg = format!("Error received from peer {} : {}", ip, err_received);
let _ = cmd_tx_clone.send(NetworkEvent::Error(err_msg, "".to_owned()));
} else {
let err_msg = format!("Error received from peer {} : N/A", ip,);
let _ = cmd_tx_clone.send(NetworkEvent::Error(err_msg, "".to_owned()));
}
}
HELLO => {
let mut payload = Vec::new();
let received_length = u16::from_be_bytes(
received_message[TYPE..LENGTH]
.try_into()
.expect("incorrect size"),
);
let received_username =
&received_message[LENGTH + EXTENSIONS..LENGTH + received_length as usize];
handhsake_history.update_peer_info(
ip.to_string(),
String::from_utf8(received_username.to_vec()).expect("invalid conversion"),
);
payload.extend_from_slice(&0u32.to_be_bytes());
payload.extend_from_slice(&crypto_pair.username.clone().as_bytes());
let helloreply = construct_message(HELLOREPLY, payload, id, crypto_pair);
return helloreply;
}
HELLOREPLY => {
// ajoute l'username a la liste des peers handshake
let received_length = u16::from_be_bytes(
received_message[TYPE..LENGTH]
.try_into()
.expect("incorrect size"),
);
let received_username =
&received_message[LENGTH + EXTENSIONS..LENGTH + received_length as usize];
handhsake_history.update_peer_info(
ip.to_string(),
String::from_utf8(received_username.to_vec()).expect("invalid conversion"),
);
// verifie s'il faut renvoyer un root request
let mut guard = messages_list.lock().expect("Échec du verrouillage");
let res = guard.get(&id);
match res {
Some(ev) => {
match ev {
EventType::HelloThenRootRequest => {
// envoyer la root request
let _ = &guard.remove_entry(&id);
println!("message {} retiré de la liste", id);
let new_id = generate_id();
let rootrequest =
construct_message(ROOTREQUEST, Vec::new(), new_id, crypto_pair);
let _ = &guard.insert(new_id, EventType::RootRequest);
println!("root requesst sent");
return rootrequest;
}
EventType::Hello => {
let _ = &guard.remove_entry(&id);
println!("message {} retiré de la liste", id);
}
_ => {}
}
}
None => {}
}
}
ROOTREPLY => {
// recuperer le pseudo du peers ayant repondu
println!("root reply received");
let peers_exist = handhsake_history.get_peer_info_ip(ip.to_string());
match peers_exist {
Some(peerinfo) => {
let mut guard = messages_list.lock().expect("Échec du verrouillage");
let res = guard.get(&id);
match res {
Some(ev) => {
match ev {
EventType::RootRequest => {
// envoyer la root request
let _ = &guard.remove_entry(&id);
println!("message {} retiré de la liste", id);
// envoyer le hash a la gui
let received_hash: NodeHash = received_message
[LENGTH..(32 + LENGTH)]
.try_into()
.expect("incorrect size");
let res =
cmd_tx_clone.send(NetworkEvent::FileTreeRootReceived(
peerinfo.username.clone(),
received_hash,
));
println!("file tree sent");
// envoyer un datum
let mut payload = Vec::new();
payload.extend_from_slice(&received_hash);
let new_id = generate_id();
let datumreqest = construct_message(
DATUMREQUEST,
payload,
new_id,
crypto_pair,
);
constructed_message = datumreqest;
guard.insert(new_id, EventType::DatumRequest);
}
_ => {
println!("event not prensent");
}
}
}
None => {}
}
}
None => {
eprintln!("no peers found");
}
}
}
DATUM => {
let mut guard = messages_list.lock().expect("Échec du verrouillage");
let res = guard.get(&id);
match res {
Some(ev) => match ev {
EventType::DatumRequest => {
let _ = &guard.remove_entry(&id);
println!("message {} retiré de la liste", id);
let received_length = u16::from_be_bytes(
received_message[TYPE..LENGTH]
.try_into()
.expect("incorrect size"),
);
let received_datum = &received_message[LENGTH..];
let parsed_node =
parse_received_datum(received_datum.to_vec(), received_length as usize);
match parsed_node {
Some(tuple) => {
let _ = cmd_tx.send(NetworkEvent::FileTreeReceived(
tuple.0,
tuple.1,
ip.to_string(),
));
}
None => {}
}
}
EventType::DatumRequestBig => {
println!("message {} retiré de la liste", id);
let received_length = u16::from_be_bytes(
received_message[TYPE..LENGTH]
.try_into()
.expect("incorrect size"),
);
println!("received length:{}", received_length);
let received_datum = &received_message[LENGTH..];
let parsed_node =
parse_received_datum(received_datum.to_vec(), received_length as usize);
match parsed_node {
Some(tuple) => {
let _ = &guard.remove_entry(&id);
let _ = cmd_tx.send(NetworkEvent::DataReceived(
tuple.0,
tuple.1,
ip.to_string(),
));
println!("datareceived event sent");
}
None => {
println!("message corrompu, nouvelle tentative");
}
}
}
_ => {}
},
None => {}
}
}
ROOTREQUEST => {
println!("root request received");
let _ = cmd_tx.send(NetworkEvent::RootRequest(ip.to_string()));
}
DATUMREQUEST => {
let received_length = u16::from_be_bytes(
received_message[TYPE..LENGTH]
.try_into()
.expect("incorrect size"),
);
let received_hash = &received_message[LENGTH..LENGTH + received_length as usize];
let _ = cmd_tx.send(NetworkEvent::DatumRequest(
received_hash.try_into().expect("incorrect size"),
ip.to_string(),
));
}
_ => return None,
};
constructed_message
}