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

wip messages creation & handling

Browse Source
This commit is contained in:
TIBERGHIEN corentin
2026-01-09 20:09:39 +01:00
parent 9fc33804d0
commit 489669b93d
7 changed files with 114 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
client-network/src/cryptographic_signature.rs
View File

@@ -108,11 +108,10 @@ pub fn sign_message(crypto_pair: &CryptographicSignature, message: &Vec<u8>) ->
let digest = Sha256::digest(&message[..7 + msg_length as usize]);
let signature = crypto_pair.priv_key.sign_prehash_recoverable(&digest);
let message_length = 12 + msg_length as usize + 32;
let message_length = 7 + msg_length as usize + 64;
let mut signed_message = Vec::with_capacity(message_length);
println!("{}", message_length);
signed_message.extend_from_slice(&message[..8 + msg_length as usize]);
signed_message.pop();
signed_message.extend_from_slice(&message[..7 + msg_length as usize]);
println!("signed_tmp:{:?}", signed_message);
match signature {
Ok(signature) => {
@@ -123,7 +122,7 @@ pub fn sign_message(crypto_pair: &CryptographicSignature, message: &Vec<u8>) ->
let s_bytes = s.to_bytes();
signed_message.extend_from_slice(&r_bytes[..32]);
signed_message.extend_from_slice(&s_bytes[..32]);
println!("signed:{:?}", signed_message);
println!("signed:{:?}, len: {}", signed_message, signed_message.len());
signed_message
}
Err(e) => {

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

@@ -187,6 +187,7 @@ pub fn start_p2p_executor(
start_receving_thread(
sd,
*first, // copie le SocketAddr (implémente Copy pour SocketAddr)
event_tx.clone(), //
);
register_ip_addresses(
sd.cryptopair_ref(),

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

@@ -7,8 +7,11 @@ use crate::{
messages_structure::construct_message,
registration,
};
use std::sync::{Arc, Mutex};
use std::{collections::HashMap, net::SocketAddr};
use std::{
net::IpAddr,
sync::{Arc, Mutex},
};
pub enum EventType {
ServerHelloReply,
@@ -42,6 +45,8 @@ pub fn handle_recevied_message(
socket_addr: &SocketAddr,
senders: &MultipleSenders,
server_name: &String,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
ip: SocketAddr,
) {
if recevied_message.len() < 4 {
return;
@@ -50,8 +55,31 @@ pub fn handle_recevied_message(
let message_id: [u8; 4] = recevied_message[0..4].try_into().expect("size error");
let id = i32::from_be_bytes(message_id);
parse_message(recevied_message, received_name, id, crypto_pair, cmd_tx)
//TODO
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 + EXTENSIONS + 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);
match resp {
None => {}
Some(resp_msg) => {
println!("msg_sent:{:?}", resp_msg);
senders.send_via(0, resp_msg, ip.to_string(), is_resp_to_server_handshake);
}
}
// Lock the mutex to access the HashMap
/*let list = messages_list.lock().unwrap();
@@ -119,32 +147,12 @@ pub fn handle_recevied_message(
}*/
}
pub fn ping(received_message: Vec<u8>, socket_addr: String) {}
pub fn ok(received_message: Vec<u8>, socket_addr: String) {}
pub fn error(received_message: Vec<u8>, socket_addr: String) {}
pub fn hello(received_message: Vec<u8>, socket_addr: String) {}
pub fn hello_reply(received_message: Vec<u8>, socket_addr: String) {}
pub fn root_request(received_message: Vec<u8>, socket_addr: String) {}
pub fn root_reply(received_message: Vec<u8>, socket_addr: String) {}
pub fn datum_request(received_message: Vec<u8>, socket_addr: String) {}
pub fn no_datum(received_message: Vec<u8>, socket_addr: String) {}
pub fn datum(received_message: Vec<u8>, socket_addr: String) {}
pub fn parse_message(
received_message: Vec<u8>,
received_name: String,
id: i32,
crypto_pair: &CryptographicSignature,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
ip: SocketAddr,
) -> Option<Vec<u8>> {
let cmd_tx_clone = cmd_tx.clone();
@@ -163,9 +171,15 @@ pub fn parse_message(
// verify signature
match msgtype {
HELLO | HELLOREPLY | ROOTREPLY | NODATUM | NATTRAVERSALREQUEST | NATTRAVERSALREQUEST2 => {
let ilength = u16::from_be_bytes(length_bytes);
let received_name =
&received_message[LENGTH + EXTENSIONS..LENGTH + EXTENSIONS + ilength as usize];
let received_username = String::from_utf8(received_name.to_vec());
match received_username {
Ok(username) => {
let peer_pubkey = tokio::runtime::Runtime::new()
.unwrap()
.block_on(get_peer_key(&received_name))
.block_on(get_peer_key(&username))
.expect("failed to retrieve public key");
let signature: [u8; SIGNATURE] = received_message
[LENGTH + msg_length..LENGTH + msg_length + SIGNATURE]
@@ -174,11 +188,17 @@ pub fn parse_message(
if !verify_signature(peer_pubkey, &received_message) {
println!(
"incorrect signature from given peer: {}, ignoring message of type {} with id {}",
&received_name, received_message[ID], id
&username, received_message[ID], id
);
return None;
}
}
Err(e) => {
println!("incorrect name: {}", e);
return None;
}
}
}
_ => {}
}
@@ -204,17 +224,10 @@ pub fn parse_message(
if let Ok(err_received) =
String::from_utf8(received_message[LENGTH..(msg_length + LENGTH)].to_vec())
{
let err_msg = format!(
"Error received from peer {} : {}",
String::from(received_name),
err_received
);
let err_msg = format!("Error received from peer {} : {}", ip, err_received);
let _ = cmd_tx_clone.send(NetworkEvent::Error(err_msg));
} else {
let err_msg = format!(
"Error received from peer {} : N/A",
String::from(received_name),
);
let err_msg = format!("Error received from peer {} : N/A", ip,);
let _ = cmd_tx_clone.send(NetworkEvent::Error(err_msg));
}
}

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

@@ -253,7 +253,11 @@ impl MultipleSenders {
});
}*/
pub fn start_receving_thread(shared_data: &P2PSharedData, socket_addr: SocketAddr) {
pub fn start_receving_thread(
shared_data: &P2PSharedData,
socket_addr: SocketAddr,
cmd_tx: crossbeam_channel::Sender<NetworkEvent>,
) {
let sock_clone = shared_data.socket();
let cryptopair_clone = shared_data.cryptopair();
let senders_clone = shared_data.senders();
@@ -274,6 +278,8 @@ pub fn start_receving_thread(shared_data: &P2PSharedData, socket_addr: SocketAdd
&socket_addr,
&senders_clone,
&servername_clone,
cmd_tx.clone(),
src,
);
}
Err(e) => eprintln!("Erreur de réception: {}", e),

4
client-network/src/messages_structure.rs
View File

@@ -39,7 +39,9 @@ pub fn construct_message(
match msgtype {
HELLO | HELLOREPLY => {
// length
message.extend_from_slice(&(EXTENSIONS + payload.len()).to_be_bytes());
let a = payload.len() as u16;
println!("payload size:{}", a);
message.extend_from_slice(&a.to_be_bytes());
message.extend_from_slice(&payload);
let signature = sign_message(crypto_pair, &message);
message.extend_from_slice(&signature);

39
client-network/src/peers_refresh.rs
View File

@@ -3,9 +3,12 @@
use std::{
collections::{HashMap, VecDeque},
net::SocketAddr,
net::{AddrParseError, SocketAddr},
ops::Add,
process::Command,
sync::{Arc, Mutex},
thread,
time::{self, SystemTime},
time::{self, Duration, SystemTime},
};
pub struct PeerInfo {
@@ -14,24 +17,36 @@ pub struct PeerInfo {
}
pub struct HandshakeHistory {
time_k_hash_v: HashMap<u64, u64>,
hash_k_peerinfo_v: HashMap<u64, PeerInfo>,
times_to_check: VecDeque<u64>,
time_k_ip_v: HashMap<u64, u64>,
ip_k_peerinfo_v: HashMap<u64, PeerInfo>,
}
impl HandshakeHistory {
pub fn new() -> HandshakeHistory {
HandshakeHistory {
time_k_hash_v: HashMap::new(),
hash_k_peerinfo_v: HashMap::new(),
times_to_check: VecDeque::new(),
time_k_ip_v: HashMap::new(),
ip_k_peerinfo_v: HashMap::new(),
}
}
pub fn update_handshake(&self) {
pub fn update_handshake(&mut self) {
thread::spawn(move || {
let mut times_to_check = VecDeque::new();
let current_time: u64 = SystemTime::now()
.duration_since(time::UNIX_EPOCH)
.expect("system time before UNIX EPOCH")
.add(Duration::from_secs(10))
.as_secs();
// adds 10 seconds in the queue every 10 seconds
self.times_to_check.insert(index, value);
loop {
let mut child = Command::new("sleep").arg("9").spawn().unwrap();
let _result = child.wait().unwrap();
for n in 0..9 {
// push 9 successive seconds
times_to_check.push_back(current_time + n);
// gestion d'erreur si verrou mort
}
}
});
}
@@ -41,9 +56,9 @@ impl HandshakeHistory {
.expect("system time before UNIX EPOCH")
.as_secs();
println!("time:{}", current_time);
self.time_k_hash_v.insert(current_time, hash);
/*self.time_k_hash_v.insert(current_time, hash);
self.hash_k_peerinfo_v
.insert(hash, PeerInfo { username, ip });
.insert(hash, PeerInfo { username, ip });*/
}
}

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

@@ -4,6 +4,7 @@ use getrandom::Error;
use crate::cryptographic_signature::{CryptographicSignature, formatPubKey, sign_message};
use crate::message_handling::EventType;
use crate::messages_channels::{Message, MultipleSenders};
use crate::messages_structure::construct_message;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::net::UdpSocket;
@@ -72,17 +73,17 @@ pub fn register_ip_addresses(
messages_list: &Mutex<HashMap<i32, EventType>>,
id: i32,
) {
let username_size = crypto_pair.username.len();
let hello_handshake = HandshakeMessage::hello(
id as u32,
username_size as u16 + 4,
crypto_pair.username.clone(),
);
//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, false);
let mut list = messages_list.lock().expect("Failed to lock messages_list");
let mut payload = Vec::new();
payload.extend_from_slice(&0u32.to_be_bytes());
payload.extend_from_slice(&crypto_pair.username.clone().as_bytes());
let hello_handshake = construct_message(1, payload, id, crypto_pair);
match hello_handshake {
Some(handshake_message) => {
senders.send_via(0, handshake_message, server_uri, false);
}
None => {}
}
/*let mut list = messages_list.lock().expect("Failed to lock messages_list");
match list.get(&id) {
Some(_) => {
list.remove(&id);
@@ -91,7 +92,7 @@ pub fn register_ip_addresses(
list.insert(id, EventType::ServerHelloReply);
}
}
println!("message sent: {}", &id);
println!("message sent: {}", &id);*/
// 3. Perform the insertion
/*let mut buf = [0u8; 1024];
socket.recv_from(&mut buf).expect("receive failed");