From 10f4a34ae417a2102cde87ed654f5d4cdf891e1d Mon Sep 17 00:00:00 2001 From: michel Date: Fri, 13 Sep 2024 20:58:04 +0200 Subject: [PATCH] feat: add missing message handlers --- .gp.md | 215 -------- Cargo.lock | 9 - Cargo.toml | 3 - src/client.rs | 147 +++++- src/dispatcher.rs | 65 +-- src/main.rs | 21 +- src/message/factory.rs | 60 --- src/message/handlers.rs | 56 +++ src/message/handlers/login.rs | 24 - src/message/handlers/mod.rs | 4 - src/message/handlers/privileged_users.rs | 11 - src/message/handlers/room_list.rs | 60 --- src/message/message.rs | 203 ++++++-- src/message/{reader.rs => message_reader.rs} | 67 +-- src/message/mod.rs | 7 +- src/message/server/connect_to_peer.rs | 40 ++ .../excluded_search_phrases.rs | 13 +- src/message/server/file_search.rs | 24 + src/message/server/login.rs | 41 ++ src/message/server/message_factory.rs | 119 +++++ src/message/server/message_user.rs | 65 +++ src/message/server/mod.rs | 18 + src/message/server/parent_min_speed.rs | 20 + src/message/server/parent_speed_ratio.rs | 20 + src/message/server/privileged_users.rs | 19 + src/message/server/room_list.rs | 62 +++ src/message/server/wish_list_interval.rs | 21 + src/peer/default_peer.rs | 142 ++++++ src/peer/listen.rs | 45 ++ src/peer/mod.rs | 3 + src/peer/peer.rs | 153 ++++++ src/server.rs | 222 +++++++-- src/utils/md5.rs | 471 ++++++++++++++++++ src/utils/mod.rs | 3 + 34 files changed, 1889 insertions(+), 564 deletions(-) delete mode 100644 .gp.md delete mode 100644 src/message/factory.rs create mode 100644 src/message/handlers.rs delete mode 100644 src/message/handlers/login.rs delete mode 100644 src/message/handlers/mod.rs delete mode 100644 src/message/handlers/privileged_users.rs delete mode 100644 src/message/handlers/room_list.rs rename src/message/{reader.rs => message_reader.rs} (60%) create mode 100644 src/message/server/connect_to_peer.rs rename src/message/{handlers => server}/excluded_search_phrases.rs (66%) create mode 100644 src/message/server/file_search.rs create mode 100644 src/message/server/login.rs create mode 100644 src/message/server/message_factory.rs create mode 100644 src/message/server/message_user.rs create mode 100644 src/message/server/mod.rs create mode 100644 src/message/server/parent_min_speed.rs create mode 100644 src/message/server/parent_speed_ratio.rs create mode 100644 src/message/server/privileged_users.rs create mode 100644 src/message/server/room_list.rs create mode 100644 src/message/server/wish_list_interval.rs create mode 100644 src/peer/default_peer.rs create mode 100644 src/peer/listen.rs create mode 100644 src/peer/mod.rs create mode 100644 src/peer/peer.rs create mode 100644 src/utils/md5.rs create mode 100644 src/utils/mod.rs diff --git a/.gp.md b/.gp.md deleted file mode 100644 index 9d200f1..0000000 --- a/.gp.md +++ /dev/null @@ -1,215 +0,0 @@ -Additional context is provided below. - -I have the following from soulseek_download/src/main.rs: - -```rust -use std::collections::VecDeque; -use std::io::{self, Write}; -use std::net::{TcpStream, ToSocketAddrs}; -use std::sync::{Arc, Barrier, Condvar, Mutex}; -use std::thread::{self, sleep}; -use std::time::Duration; - -mod buffered_reader; -mod message; -mod message_factory; - -#[derive(Debug, Clone)] -struct ServerAddress { - host: String, - port: u16, -} - -struct Client { - address: ServerAddress, - username: String, - password: String, -} - -struct Server { - address: ServerAddress, - stream: Arc>, - message_queue: Arc<(Mutex>>, Condvar)>, -} - -impl Server { - /// Create a new instance of Server, returning a Result - fn new(address: ServerAddress) -> Result { - println!("Connecting to server at {}:{}", address.host, address.port); - - // Convert the host and port into a socket address - let socket_address = format!("{}:{}", address.host, address.port) - .to_socket_addrs()? - .next() - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?; - - // Establish the connection with a timeout - let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?; - - // Set the stream to non-blocking mode if needed or configure timeouts - stream.set_read_timeout(Some(Duration::from_secs(5)))?; - stream.set_write_timeout(Some(Duration::from_secs(5)))?; - - let message_queue = Arc::new((Mutex::new(VecDeque::new()), Condvar::new())); // Initialize the message queue with Condvar - - let server = Self { - address, - stream: Arc::new(Mutex::new(stream)), - message_queue, - }; - - server.start_read_write_loops(); - - Ok(server) - } - /// Start reading and writing loops in separate threads - fn start_read_write_loops(&self) { - // Create a channel to send messages to the write thread - let read_stream = Arc::clone(&self.stream); - let write_stream = Arc::clone(&self.stream); - - let barrier = Arc::new(Barrier::new(2)); - let read_barrier = barrier.clone(); - let write_barrier = barrier.clone(); - let message_queue = self.message_queue.clone(); - - // Spawn a thread to handle reading from the server - thread::spawn(move || { - read_barrier.wait(); - let mut buffered_reader = buffered_reader::BufferedReader::new(); - - loop { - match buffered_reader.read_from_socket(&mut read_stream.lock().unwrap()) { - Ok(_) => {} - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, - Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { - println!("Read operation timed out"); - continue; - } - Err(e) => { - eprintln!("Error reading from server: {}", e); - break; - } - } - - match buffered_reader.extract_message() { - Ok(Some(mut message)) => { - println!("Received message: {:?}", message.get_data()); - self.handle_message(&mut message); - } - Err(e) => { - println!("Error extracting message: {}", e) - } - Ok(None) => continue, - } - } - }); - thread::spawn(move || { - write_barrier.wait(); - - loop { - let (lock, cvar) = &*message_queue; - let mut queue = lock.lock().unwrap(); - - // Wait for a message to be added to the queue - while queue.is_empty() { - queue = cvar.wait(queue).unwrap(); - } - - // Pop and send the message - if let Some(message) = queue.pop_front() { - match write_stream.lock().unwrap().write_all(&message) { - Ok(_) => println!("Sent buffered message: {:?}", message), - Err(e) => { - eprintln!("Failed to send buffered message: {}", e); - // Handle error appropriately - break; - } - } - } - } - }); - } - - fn queue_message(&self, message: Vec) { - let (lock, cvar) = &*self.message_queue; - let mut queue = lock.lock().unwrap(); - queue.push_back(message); - cvar.notify_one(); - } - - fn login(&mut self, username: &str, password: &str) -> Result<(), std::io::Error> { - self.queue_message(message_factory::build_init_message().get_data()); - self.queue_message(message_factory::build_login_message(username, password).get_data()); - Ok(()) - } - - fn handle_message(&mut self, message: &mut message::Message) { - let code = message.get_code(); - match code { - 1 => { - println!("Received login response"); - let response = message.read_int8(); - if response == 1 { - println!("Login successful"); - } else { - println!("Login failed"); - } - } - _ => { - println!("Received unknown message with code {}", code); - } - } - } -} - -impl Client { - fn new(address: ServerAddress, username: String, password: String) -> Self { - Self { - address, - username, - password, - } - } - fn connect(&self) { - // Attempt to create a server instance - match Server::new(self.address.clone()) { - Ok(mut server) => { - println!( - "Connected to server at {}:{}", - server.address.host, server.address.port - ); - // Attempt to login - println!("Logging in as {}", self.username); - - if let Err(e) = server.login(&self.username, &self.password) { - eprintln!("Error during login: {}", e); - } - } - Err(e) => eprintln!("Error connecting to server: {}", e), - } - } -} -// ServerAddress { -// host: String::from("server.slsknet.org"), -// port: 2242, -// }, - -fn main() { - let client = Client::new( - ServerAddress { - // host: String::from("server.slsknet.org"), - host: String::from("localhost"), - port: 2242, - }, - String::from("insane_in_the_brain2"), - String::from("invalid"), - // String::from("13375137"), - ); - client.connect(); - loop { - thread::park(); - } -} -``` diff --git a/Cargo.lock b/Cargo.lock index 894466a..a415a73 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,15 +2,6 @@ # It is not intended for manual editing. version = 3 -[[package]] -name = "md5" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771" - [[package]] name = "soulseek_download" version = "0.1.0" -dependencies = [ - "md5", -] diff --git a/Cargo.toml b/Cargo.toml index d8afbc7..58b24d6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,3 @@ name = "soulseek_download" version = "0.1.0" edition = "2021" - -[dependencies] -md5 = "0.7.0" diff --git a/src/client.rs b/src/client.rs index b7d3039..97deb52 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,36 +1,159 @@ -use crate::server::{Server, ServerAddress}; - +use crate::{ + peer::{default_peer::DefaultPeer, peer::Peer}, + server::{PeerAddress, Server}, + utils::md5, +}; +use std::{ + collections::HashMap, + sync::{ + mpsc::{Receiver, Sender}, + Mutex, + }, + thread, +}; +use std::{ + sync::{mpsc, Arc}, + time::{Duration, Instant}, +}; +pub enum ClientOperation { + ConnectToPeer(Peer), +} +struct ClientContext { + peers: HashMap, + sender: Option>, +} +impl ClientContext { + pub fn new() -> Self { + Self { + peers: HashMap::new(), + sender: None, + } + } +} pub struct Client { - address: ServerAddress, + address: PeerAddress, username: String, password: String, + server: Option, + context: Arc>, } impl Client { - pub fn new(address: ServerAddress, username: String, password: String) -> Self { + pub fn new(address: PeerAddress, username: String, password: String) -> Self { Self { address, username, password, + server: None, + context: Arc::new(Mutex::new(ClientContext::new())), } } - pub fn connect(&self) { - match Server::new(self.address.clone()) { - Ok(mut server) => { + pub fn connect(&mut self) { + let (sender, message_reader): (Sender, Receiver) = + mpsc::channel(); + + self.context.lock().unwrap().sender = Some(sender.clone()); + + // self.read_form_channel(message_reader); + self.server = match Server::new(self.address.clone(), sender) { + Ok(server) => { println!( "Connected to server at {}:{}", server.get_address().get_host(), server.get_address().get_port() ); - // Attempt to login - println!("Logging in as {}", self.username); - if let Err(e) = server.login(&self.username, &self.password) { - eprintln!("Error during login: {}", e); + Self::listen_to_client_operations(message_reader, self.context.clone()); + Some(server) + } + Err(e) => { + eprintln!("Error connecting to server: {}", e); + None + } + }; + } + + pub fn login(&self) -> Result { + // Attempt to login + println!("Logging in as {}", self.username); + if let Some(server) = &self.server { + let result = server.login(&self.username, &self.password); + if result.unwrap() == true { + return Ok(true); + } else { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "Error logging in", + )); + } + } else { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "Not connected to server", + )); + } + } + + pub fn search(&self, query: &str, timeout: Duration) { + println!("Searching for {}", query); + if let Some(server) = &self.server { + let hash = md5::md5(query); + let token = i32::from_str_radix(&hash[0..8].to_string(), 16).unwrap(); + server.file_search(token, &query); + } else { + eprintln!("Not connected to server"); + } + + let start = Instant::now(); + loop { + if start.elapsed() >= timeout { + break; + } + } + if let Some(server) = &self.server { + println!("server {:?}", server) + } + } + + fn listen_to_client_operations( + reader: Receiver, + client_context: Arc>, + ) { + thread::spawn(move || loop { + if let Ok(operation) = reader.recv() { + match operation { + ClientOperation::ConnectToPeer(peer) => { + Self::connect_to_peer(peer, client_context.clone()) + } } } - Err(e) => eprintln!("Error connecting to server: {}", e), + }); + } + + fn connect_to_peer(peer: Peer, client_context: Arc>) { + let context = client_context.clone(); + let unlocked_context = context.lock().unwrap(); + + if let Some(sender) = &unlocked_context.sender { + if unlocked_context.peers.get(&peer.username).is_none() { + let peer_clone = peer.clone(); + let sender_clone = sender.clone(); + thread::spawn(move || { + let default_peer = DefaultPeer::new(peer_clone, sender_clone); + match default_peer.connect() { + Ok(p) => { + let mut context = client_context.lock().unwrap(); + context.peers.insert(peer.username, p); + } + Err(_e) => { + // eprintln!("Error connecting to peer: {:?}", e) + } + } + }); + } + } else { + eprintln!("No sender found"); } } } diff --git a/src/dispatcher.rs b/src/dispatcher.rs index 8e6cd1c..bcf76f5 100644 --- a/src/dispatcher.rs +++ b/src/dispatcher.rs @@ -1,59 +1,32 @@ -use crate::message::handlers::excluded_search_phrases::ExcludedSearchPhrasesHandler; -use crate::message::handlers::login::LoginHandler; -use crate::message::handlers::privileged_users::PrivilegedUsersHandler; -use crate::message::handlers::room_list::RoomListHandler; -use crate::{ - message::{Message, MessageType}, - server::Context, -}; -use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use std::sync::mpsc::Sender; -pub trait MessageHandler { - fn handle(&self, message: &mut Message, context: Arc>); -} +use crate::{ + message::{handlers::Handlers, message::Message}, + server::ServerOperation, +}; pub struct MessageDispatcher { - handlers: HashMap>, - context: Arc>, + sender: Sender, + handlers: Handlers, } impl MessageDispatcher { - pub fn new(context: Arc>) -> Self { - let mut dispatcher = Self { - handlers: HashMap::new(), - context, - }; - dispatcher.register_default_handlers(); + pub fn new(sender: Sender, handlers: Handlers) -> Self { + let dispatcher = Self { handlers, sender }; dispatcher } - pub fn register_handler( - &mut self, - msg_type: MessageType, - handler: H, - ) -> &mut Self { - self.handlers.insert(msg_type, Box::new(handler)); - self - } - pub fn dispatch(&self, message: &mut Message) { - if let Some(handler) = self.handlers.get(&message.get_message_type()) { - message.set_pointer(8); - return handler.handle(message, self.context.clone()); + let code = message.get_message_code(); + // println!("message with code: {}", code); + match self.handlers.get_handler(code.clone()) { + Some(handler) => { + message.set_pointer(8); + handler.handle(message, self.sender.clone()); + } + None => { + println!("No handler found for message code: {:?}", code); + } } - println!( - "No handler found for message type: {:?}", - message.get_message_type() - ); - } - pub fn register_default_handlers(&mut self) { - self.register_handler(MessageType::Login, LoginHandler); - self.register_handler(MessageType::PrivilegedUsers, PrivilegedUsersHandler); - self.register_handler( - MessageType::ExcludedSearchPhrases, - ExcludedSearchPhrasesHandler, - ); - self.register_handler(MessageType::RoomList, RoomListHandler); } } diff --git a/src/main.rs b/src/main.rs index d1c5b62..904ad02 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,23 +1,30 @@ mod client; mod dispatcher; mod message; +mod peer; mod server; - -use std::thread; +mod utils; use client::Client; -use server::ServerAddress; +use server::PeerAddress; +use std::time::Duration; fn main() { - let client = Client::new( + let mut client = Client::new( // ServerAddress::new(String::from("localhost"), 2416), - ServerAddress::new(String::from("server.slsknet.org"), 2416), + PeerAddress::new(String::from("server.slsknet.org"), 2416), String::from("insane_in_the_brain2"), // String::from("invalid"), String::from("13375137"), ); + client.connect(); - loop { - thread::park(); + match client.login() { + Ok(_) => { + client.search("the weekend", Duration::from_secs(20)); + } + Err(e) => { + println!("Failed to login: {}", e); + } } } diff --git a/src/message/factory.rs b/src/message/factory.rs deleted file mode 100644 index 52cf210..0000000 --- a/src/message/factory.rs +++ /dev/null @@ -1,60 +0,0 @@ -use super::Message; - -pub fn build_init_message() -> Message { - let mut message = Message::new(); - message.write_int32(81); // need to send as first message? - message -} - -pub fn build_login_message(username: &str, password: &str) -> Message { - let mut hasher = md5::Context::new(); - - hasher.consume(username); - hasher.consume(password); - - let hash = format!("{:x}", hasher.compute()); - let hash_str = hash.as_str(); - - let mut message = Message::new(); - - message - .write_int32(1) - .write_string(username) - .write_string(password) - .write_int32(160) // version - .write_string(hash_str) - .write_int32(17); - - message -} - -pub fn build_shared_folders_message(folder_count: i32, file_count: i32) -> Message { - let mut message = Message::new(); - message - .write_int32(35) - .write_int32(folder_count) - .write_int32(file_count); - - message -} - -#[test] -fn test_build_init_message() { - let message = build_init_message(); - let expect: Vec = [81, 0, 0, 0].to_vec(); - assert_eq!(expect, message.get_data()) -} - -#[test] -fn test_build_login_message() { - let message = build_login_message("username", "password"); - - let expect: Vec = [ - 1, 0, 0, 0, 8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101, 8, 0, 0, 0, 112, 97, 115, - 115, 119, 111, 114, 100, 160, 0, 0, 0, 32, 0, 0, 0, 100, 53, 49, 99, 57, 97, 55, 101, 57, - 51, 53, 51, 55, 52, 54, 97, 54, 48, 50, 48, 102, 57, 54, 48, 50, 100, 52, 53, 50, 57, 50, - 57, 17, 0, 0, 0, - ] - .to_vec(); - assert_eq!(expect, message.get_data()) -} diff --git a/src/message/handlers.rs b/src/message/handlers.rs new file mode 100644 index 0000000..6ff5e29 --- /dev/null +++ b/src/message/handlers.rs @@ -0,0 +1,56 @@ +use std::collections::HashMap; + +use crate::{message::Message, server::ServerOperation}; +use std::sync::mpsc::Sender; + +use super::server::{ + ConnectToPeerHandler, ExcludedSearchPhrasesHandler, FileSearch, LoginHandler, MessageUser, + ParentMinSpeedHandler, ParentSpeedRatioHandler, PrivilegedUsersHandler, RoomListHandler, + WishListIntervalHandler, +}; + +pub trait MessageHandler { + fn get_code(&self) -> u8; + fn handle(&self, message: &mut Message, sender: Sender); +} +pub struct Handlers { + handlers: HashMap>, +} + +impl Handlers { + pub fn new() -> Self { + Self { + handlers: HashMap::new(), + } + } + pub fn new_with_server_handlers() -> Self { + let mut handlers = Self::new(); + handlers.add_server_handlers(); + handlers + } + + fn add_server_handlers(&mut self) { + self.register_handler(LoginHandler); + self.register_handler(RoomListHandler); + self.register_handler(ExcludedSearchPhrasesHandler); + self.register_handler(PrivilegedUsersHandler); + self.register_handler(MessageUser); + self.register_handler(WishListIntervalHandler); + self.register_handler(ParentMinSpeedHandler); + self.register_handler(ParentSpeedRatioHandler); + self.register_handler(PrivilegedUsersHandler); + self.register_handler(FileSearch); + self.register_handler(ConnectToPeerHandler); + } + + pub fn register_handler( + &mut self, + handler: H, + ) -> &mut Self { + self.handlers.insert(handler.get_code(), Box::new(handler)); + self + } + pub fn get_handler(&self, code: u8) -> Option<&Box> { + self.handlers.get(&code) + } +} diff --git a/src/message/handlers/login.rs b/src/message/handlers/login.rs deleted file mode 100644 index aca7080..0000000 --- a/src/message/handlers/login.rs +++ /dev/null @@ -1,24 +0,0 @@ -use std::sync::{Arc, Mutex}; - -use crate::{ - dispatcher::MessageHandler, - message::{factory::build_shared_folders_message, Message}, - server::Context, -}; -pub struct LoginHandler; - -impl MessageHandler for LoginHandler { - fn handle(&self, message: &mut Message, context: Arc>) { - let response = message.read_int8(); - if response == 1 { - println!("Login successful"); - let greeting = message.read_string(); - println!("Server geeting: {:?}", greeting); - // Build the shared folders message and queue it - let shared_message = build_shared_folders_message(1, 1); - context.lock().unwrap().queue_message(shared_message); - } else { - println!("Login failed"); - } - } -} diff --git a/src/message/handlers/mod.rs b/src/message/handlers/mod.rs deleted file mode 100644 index 52526f6..0000000 --- a/src/message/handlers/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -pub mod excluded_search_phrases; -pub mod login; -pub mod privileged_users; -pub mod room_list; diff --git a/src/message/handlers/privileged_users.rs b/src/message/handlers/privileged_users.rs deleted file mode 100644 index 0b9a429..0000000 --- a/src/message/handlers/privileged_users.rs +++ /dev/null @@ -1,11 +0,0 @@ -use std::sync::{Arc, Mutex}; - -use crate::{dispatcher::MessageHandler, message::Message, server::Context}; -pub struct PrivilegedUsersHandler; - -impl MessageHandler for PrivilegedUsersHandler { - fn handle(&self, message: &mut Message, _context: Arc>) { - let number = message.read_int32(); - println!("Number of privilaged users: {}", number); - } -} diff --git a/src/message/handlers/room_list.rs b/src/message/handlers/room_list.rs deleted file mode 100644 index 852028e..0000000 --- a/src/message/handlers/room_list.rs +++ /dev/null @@ -1,60 +0,0 @@ -use crate::{ - dispatcher::MessageHandler, - message::Message, - server::{Context, Room}, -}; -use std::sync::{Arc, Mutex}; -pub struct RoomListHandler; - -impl MessageHandler for RoomListHandler { - fn handle(&self, message: &mut Message, context: Arc>) { - let mut binding = context.lock().unwrap(); - let rooms = binding.get_rooms(); - - let num_public_rooms = message.read_int32(); - for _ in 0..num_public_rooms { - rooms.public_rooms.push(Room::new(message.read_string(), 0)); - } - - let num_public_rooms = message.read_int32(); - for i in 0..num_public_rooms { - rooms.public_rooms[i as usize].set_number_of_users(message.read_int32()); - } - - let num_owned_private_rooms = message.read_int32(); - for _ in 0..num_owned_private_rooms { - rooms - .owned_private_rooms - .push(Room::new(message.read_string(), 0)); - } - - let num_owned_private_rooms = message.read_int32(); - for i in 0..num_owned_private_rooms { - rooms.owned_private_rooms[i as usize].set_number_of_users(message.read_int32()); - } - - let num_private_rooms = message.read_int32(); - for _ in 0..num_private_rooms { - rooms - .private_rooms - .push(Room::new(message.read_string(), 0)); - } - - let num_private_rooms = message.read_int32(); - for i in 0..num_private_rooms { - rooms.private_rooms[i as usize].set_number_of_users(message.read_int32()); - } - - let num_operated_private_rooms = message.read_int32(); - for _ in 0..num_operated_private_rooms { - rooms - .operated_private_rooms - .push(Room::new(message.read_string(), 0)); - } - let num_operated_private_rooms = message.read_int32(); - for i in 0..num_operated_private_rooms { - rooms.operated_private_rooms[i as usize].set_number_of_users(message.read_int32()); - } - rooms.print(); - } -} diff --git a/src/message/message.rs b/src/message/message.rs index aec5ccc..e6b71e4 100644 --- a/src/message/message.rs +++ b/src/message/message.rs @@ -1,25 +1,5 @@ use std::str; -#[derive(Debug, Clone, Eq, PartialEq, Hash)] -pub enum MessageType { - Login, - PrivilegedUsers, - ExcludedSearchPhrases, - RoomList, - Unknown(u8), -} -impl From for MessageType { - fn from(value: u8) -> Self { - match value { - 1 => MessageType::Login, - 69 => MessageType::PrivilegedUsers, - 160 => MessageType::ExcludedSearchPhrases, - 64 => MessageType::RoomList, - _ => MessageType::Unknown(value), - } - } -} - #[derive(Debug, PartialEq, Clone)] pub struct Message { data: Vec, @@ -34,8 +14,99 @@ impl Message { } } - pub fn get_message_type(&self) -> MessageType { - MessageType::from(self.data[4]) + // pub fn print_hex(&self) { + // let data = &self.data; + // const BYTES_PER_LINE: usize = 16; + // + // let chunks = data.chunks(BYTES_PER_LINE); + // for (i, chunk) in chunks.enumerate() { + // // Print the offset + // print!("{:04x} ", i * BYTES_PER_LINE); + // + // // Print the hexadecimal part + // for j in 0..BYTES_PER_LINE { + // if j < chunk.len() { + // print!("{:02x} ", chunk[j]); + // } else { + // print!(" "); + // } + // + // // Add extra space in the middle + // if j == 7 { + // print!(" "); + // } + // } + // + // print!(" "); + // + // // Print the ASCII part + // let mut i = 0; + // for &byte in chunk { + // i = i + 1; + // if byte.is_ascii_graphic() || byte.is_ascii_whitespace() { + // print!("{}", byte as char); + // } else { + // print!("."); + // } + // + // if i == 8 { + // print!(" "); + // } + // } + // + // println!(); + // } + // } + // + // pub fn print_hex2(&self) -> String { + // let data = &self.data; + // let mut out = String::from(""); + // const BYTES_PER_LINE: usize = 16; + // + // let chunks = data.chunks(BYTES_PER_LINE); + // for (i, chunk) in chunks.enumerate() { + // // Print the offset + // // out += &format!("{:04x} ", i * BYTES_PER_LINE); + // + // // Print the hexadecimal part + // for j in 0..BYTES_PER_LINE { + // if j < chunk.len() { + // out += &format!("{:02x} ", chunk[j]); + // } else { + // // out += &format!(" "); + // } + // + // // Add extra space in the middle + // // if j == 7 { + // // out += &format!(" "); + // // } + // } + // + // // // Print the ASCII part + // // let mut i = 0; + // // for &byte in chunk { + // // i = i + 1; + // // if byte.is_ascii_graphic() || byte.is_ascii_whitespace() { + // // print!("{}", byte as char); + // // } else {messa + // // print!("."); + // // } + // // + // // if i == 8 { + // // print!(" "); + // // } + // // } + // // + // } + // println!("{:?}", out.trim()); + // return out.trim().into(); + // } + + pub fn get_message_code_u32(&self) -> u32 { + u32::from_le_bytes(self.data[4..8].try_into().unwrap()) + } + pub fn get_message_code(&self) -> u8 { + self.data[4] } pub fn new_with_data(data: Vec) -> Self { @@ -48,10 +119,21 @@ impl Message { self.pointer = pointer; } + // get's the message data pub fn get_data(&self) -> Vec { self.data.clone() } + /// gets buffer with the message length prepended + pub fn get_buffer(&self) -> Vec { + let mut b = vec![0u8; 4]; + let length = self.data.len() as u32; + b[0..4].copy_from_slice(&length.to_le_bytes()); + let mut combined = b; + combined.extend(&self.data); + combined + } + pub fn read_string(&mut self) -> String { let size = u32::from_le_bytes([ self.data[self.pointer], @@ -66,6 +148,14 @@ impl Message { String::from_utf8(data.to_vec()).expect("Failed to read string") } + pub fn read_raw_hex_str(&mut self, size: usize) -> String { + let hex_str = self.data[self.pointer..self.pointer + size] + .iter() + .map(|byte| format!("{:02x}", byte)) + .collect::(); + self.pointer += size; + return hex_str; + } pub fn read_int8(&mut self) -> i8 { let val = self.data[self.pointer] as i8; @@ -73,6 +163,7 @@ impl Message { val } + #[allow(dead_code)] pub fn read_int64(&mut self) -> i64 { let val = i64::from_le_bytes([ self.data[self.pointer], @@ -103,6 +194,12 @@ impl Message { val } + pub fn read_bool(&mut self) -> bool { + let val = self.data[self.pointer] == 1; + self.pointer += 1; + val + } + pub fn write_string(&mut self, val: &str) -> &mut Self { let length = val.len() as u32; self.data.extend_from_slice(&length.to_le_bytes()); @@ -114,18 +211,66 @@ impl Message { self.data.extend_from_slice(&value.to_le_bytes()); self } + pub fn write_raw_bytes(&mut self, value: Vec) -> &mut Self { + self.data.extend_from_slice(&value); + self + } + pub fn write_raw_hex_string(&mut self, val: &str) -> &mut Self { + let mut b = Vec::new(); + for i in (0..val.len()).step_by(2) { + let byte_str = &val[i..i + 2]; + let byte = u8::from_str_radix(byte_str, 16).expect("Invalid hex string"); + b.push(byte); + } + self.data.extend_from_slice(&b); + self.pointer += b.len(); + self + } // pub fn decode(&self) { - // let size = - // u32::from_le_bytes([self.data[0], self.data[1], self.data[2], self.data[3]]) as usize; - // println!("Size: {}", size); - // if size >= 4 { - // let code = u32::from_le_bytes([self.data[4], self.data[5], self.data[6], self.data[7]]); - // println!("Code: {}", code); - // } + // println!("{:?}", self.data); // } } +#[test] +fn test_get_buffer() { + let message = Message::new_with_data( + [ + 26, 0, 0, 0, 219, 178, 47, 28, 11, 0, 0, 0, 116, 104, 101, 32, 119, 101, 101, 107, 101, + 110, 100, + ] + .to_vec(), + ); + assert_eq!( + message.get_buffer(), + [ + 23, 0, 0, 0, 26, 0, 0, 0, 219, 178, 47, 28, 11, 0, 0, 0, 116, 104, 101, 32, 119, 101, + 101, 107, 101, 110, 100, + ] + .to_vec() + ); + + let message = Message::new_with_data( + [ + 1, 0, 0, 0, 20, 0, 0, 0, 105, 110, 115, 97, 110, 101, 95, 105, 110, 95, 116, 104, 101, + 95, 98, 114, 97, 105, 110, 50, 8, 0, 0, 0, 49, 51, 51, 55, 53, 49, 51, 55, 160, 0, 0, + 0, 32, 0, 0, 0, 50, 101, 100, 102, 53, 49, 100, 48, 51, 55, 57, 52, 51, 55, 56, 102, + 56, 98, 98, 54, 51, 49, 48, 100, 52, 54, 48, 99, 50, 50, 98, 49, 17, 0, 0, 0, + ] + .to_vec(), + ); + assert_eq!( + message.get_buffer(), + [ + 84, 0, 0, 0, 1, 0, 0, 0, 20, 0, 0, 0, 105, 110, 115, 97, 110, 101, 95, 105, 110, 95, + 116, 104, 101, 95, 98, 114, 97, 105, 110, 50, 8, 0, 0, 0, 49, 51, 51, 55, 53, 49, 51, + 55, 160, 0, 0, 0, 32, 0, 0, 0, 50, 101, 100, 102, 53, 49, 100, 48, 51, 55, 57, 52, 51, + 55, 56, 102, 56, 98, 98, 54, 51, 49, 48, 100, 52, 54, 48, 99, 50, 50, 98, 49, 17, 0, 0, + 0, + ] + .to_vec() + ); +} #[test] fn test_read_string() { let data = vec![ diff --git a/src/message/reader.rs b/src/message/message_reader.rs similarity index 60% rename from src/message/reader.rs rename to src/message/message_reader.rs index 0457252..e1a7df8 100644 --- a/src/message/reader.rs +++ b/src/message/message_reader.rs @@ -34,6 +34,7 @@ impl MessageReader { } } + #[cfg(test)] pub fn new_with_buffer(buffer: Vec) -> Self { Self { buffer: buffer.into(), @@ -43,7 +44,9 @@ impl MessageReader { pub fn read_from_socket(&mut self, stream: &mut TcpStream) -> io::Result<()> { let mut temp_buffer = [0; 1024]; // Temporary buffer for reading from the socket let bytes_read = stream.read(&mut temp_buffer)?; - // println!("Read {} bytes from the socket", bytes_read); + if bytes_read == 0 { + return Ok(()); + } // Add the read bytes to the internal buffer self.buffer.extend(&temp_buffer[..bytes_read]); @@ -75,35 +78,41 @@ impl MessageReader { } } -#[test] -fn test_extract_message() { - let buffer: Vec = [ - 8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101, 8, 0, 0, 0, 112, 97, 115, 115, 119, 111, - 114, 100, 160, 0, 0, 0, 32, 0, 0, 0, 100, 53, 49, 99, 57, 97, 55, 101, 57, 51, 53, 51, 55, - 52, 54, 97, 54, 48, 50, 48, 102, 57, 54, 48, 50, 100, 52, 53, 50, 57, 50, 57, 17, 0, 0, 0, - ] - .to_vec(); - let mut buffered_reader = MessageReader::new_with_buffer(buffer); - let mut message = buffered_reader.extract_message().unwrap().unwrap(); - assert_eq!( - message.get_data(), - vec![8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101] - ); - assert_eq!(message.read_string(), "username"); -} -#[test] -fn test_extract_message_incomplete_message() { - let incomplete_buffer = vec![1, 2, 3]; - let mut buffered_reader = MessageReader::new_with_buffer(incomplete_buffer); +#[cfg(test)] +mod tests { + use crate::message::MessageReader; - let result = buffered_reader.extract_message(); - assert_eq!(None, result.unwrap()); + #[test] + fn test_extract_message() { + let buffer: Vec = [ + 8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101, 8, 0, 0, 0, 112, 97, 115, 115, 119, + 111, 114, 100, 160, 0, 0, 0, 32, 0, 0, 0, 100, 53, 49, 99, 57, 97, 55, 101, 57, 51, 53, + 51, 55, 52, 54, 97, 54, 48, 50, 48, 102, 57, 54, 48, 50, 100, 52, 53, 50, 57, 50, 57, + 17, 0, 0, 0, + ] + .to_vec(); + let mut buffered_reader = MessageReader::new_with_buffer(buffer); + let mut message = buffered_reader.extract_message().unwrap().unwrap(); + assert_eq!( + message.get_data(), + vec![8, 0, 0, 0, 117, 115, 101, 114, 110, 97, 109, 101] + ); + assert_eq!(message.read_string(), "username"); + } + #[test] + fn test_extract_message_incomplete_message() { + let incomplete_buffer = vec![1, 2, 3]; + let mut buffered_reader = MessageReader::new_with_buffer(incomplete_buffer); - let rest: Vec = buffered_reader - .buffer - .drain(..buffered_reader.buffer.len()) - .collect(); + let result = buffered_reader.extract_message(); + assert_eq!(None, result.unwrap()); - assert!(buffered_reader.buffer.is_empty()); - assert_eq!(vec![1, 2, 3], rest); + let rest: Vec = buffered_reader + .buffer + .drain(..buffered_reader.buffer.len()) + .collect(); + + assert!(buffered_reader.buffer.is_empty()); + assert_eq!(vec![1, 2, 3], rest); + } } diff --git a/src/message/mod.rs b/src/message/mod.rs index 3fcba0e..bcf33d2 100644 --- a/src/message/mod.rs +++ b/src/message/mod.rs @@ -1,7 +1,8 @@ -pub mod factory; pub mod handlers; pub mod message; -pub mod reader; +pub mod message_reader; +pub mod server; +pub use handlers::*; pub use message::*; -pub use reader::*; +pub use message_reader::*; diff --git a/src/message/server/connect_to_peer.rs b/src/message/server/connect_to_peer.rs new file mode 100644 index 0000000..197cadb --- /dev/null +++ b/src/message/server/connect_to_peer.rs @@ -0,0 +1,40 @@ +use crate::message::{Message, MessageHandler}; +use crate::server::ServerOperation; +use std::sync::mpsc::Sender; +pub struct ConnectToPeerHandler; +impl MessageHandler for ConnectToPeerHandler { + fn get_code(&self) -> u8 { + 18 + } + fn handle(&self, _message: &mut Message, _sender: Sender) { + println!("Handling ConnectToPeer message"); + // let username = message.read_string(); + // let connection_type = message.read_string(); + // let mut ip: Vec = vec![]; + // for _ in 0..4 { + // ip.push(message.read_int8()); + // } + // let host: String = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]); + // + // let (port, token, privileged, unknown, obfuscated_port) = ( + // message.read_int32(), + // message.read_int32(), + // message.read_int8(), + // message.read_int8(), + // message.read_int8(), + // ); + // + // let peer = Peer::new( + // username, + // connection_type, + // host, + // port, + // token, + // privileged, + // unknown, + // obfuscated_port, + // ); + // + // sender.send(ServerOperation::ConnectToPeer(peer)).unwrap(); + } +} diff --git a/src/message/handlers/excluded_search_phrases.rs b/src/message/server/excluded_search_phrases.rs similarity index 66% rename from src/message/handlers/excluded_search_phrases.rs rename to src/message/server/excluded_search_phrases.rs index 4e1fa63..8fe0fba 100644 --- a/src/message/handlers/excluded_search_phrases.rs +++ b/src/message/server/excluded_search_phrases.rs @@ -1,10 +1,17 @@ -use std::sync::{Arc, Mutex}; +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; +use std::sync::mpsc::Sender; -use crate::{dispatcher::MessageHandler, message::Message, server::Context}; pub struct ExcludedSearchPhrasesHandler; impl MessageHandler for ExcludedSearchPhrasesHandler { - fn handle(&self, message: &mut Message, _context: Arc>) { + fn get_code(&self) -> u8 { + 160 + } + + fn handle(&self, message: &mut Message, _sender: Sender) { let item_count = message.read_int32(); let mut exluded_phrases: Vec = Vec::new(); diff --git a/src/message/server/file_search.rs b/src/message/server/file_search.rs new file mode 100644 index 0000000..c33fc61 --- /dev/null +++ b/src/message/server/file_search.rs @@ -0,0 +1,24 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{handlers::MessageHandler, message::Message}, + server::ServerOperation, +}; + +pub struct FileSearch; + +impl MessageHandler for FileSearch { + fn get_code(&self) -> u8 { + 26 + } + fn handle(&self, message: &mut Message, _sender: Sender) { + println!("Handling file search message"); + let username = message.read_string(); + let token = message.read_int32(); + let query = message.read_string(); + println!( + "Message search username:{}, token: {}, query: {}", + username, token, query + ); + } +} diff --git a/src/message/server/login.rs b/src/message/server/login.rs new file mode 100644 index 0000000..4ac64e4 --- /dev/null +++ b/src/message/server/login.rs @@ -0,0 +1,41 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; + +pub struct LoginHandler; + +impl MessageHandler for LoginHandler { + fn get_code(&self) -> u8 { + 1 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let response = message.read_int8(); + + if response != 1 { + return sender.send(ServerOperation::LoginStatus(false)).unwrap(); + } + + println!("Login successful"); + let greeting = message.read_string(); + println!("Server geeting: {:?}", greeting); + + sender.send(ServerOperation::LoginStatus(true)).unwrap(); + } +} + +// fn build_login_response_message() -> Message { +// return Message::new_with_data([ +// 50, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 81, 170, 162, 77, 32, 0, 0, 0, 101, 102, 99, 97, +// 51, 52, 102, 99, 52, 99, 56, 98, 101, 56, 98, 55, 101, 102, 51, 56, 97, 102, 50, 54, 50, +// 52, 100, 101, 53, 52, 54, 52, 0, +// ]); +// } +// #[test] +// fn test_can_handle() { +// assert_eq!(true, LoginHandler.can_handle(build_login_response_message()); +// ); +// } diff --git a/src/message/server/message_factory.rs b/src/message/server/message_factory.rs new file mode 100644 index 0000000..89076d9 --- /dev/null +++ b/src/message/server/message_factory.rs @@ -0,0 +1,119 @@ +use crate::{message::message::Message, utils::md5}; + +pub struct MessageFactory; +impl MessageFactory { + pub fn build_login_message(username: &str, password: &str) -> Message { + // Message::new_with_data( + // [ + // 1, 0, 0, 0, 20, 0, 0, 0, 105, 110, 115, 97, 110, 101, 95, 105, 110, 95, 116, 104, 101, + // 95, 98, 114, 97, 105, 110, 50, 8, 0, 0, 0, 49, 51, 51, 55, 53, 49, 51, 55, 160, 0, 0, + // 0, 32, 0, 0, 0, 50, 101, 100, 102, 53, 49, 100, 48, 51, 55, 57, 52, 51, 55, 56, 102, + // 56, 98, 98, 54, 51, 49, 48, 100, 52, 54, 48, 99, 50, 50, 98, 49, 17, 0, 0, + // 0, + // //0, // 84, 0, 0, 0, 1, 0, 0, 0, 20, 0, 0, 0, 105, 110, 115, 97, 110, 101, 95, 105, 110, 95, + // // 116, 104, 101, 95, 98, 114, 97, 105, 110, 50, 8, 0, 0, 0, 49, 51, 51, 55, 53, 49, 51, + // // 55, 160, 0, 0, 0, 32, 0, 0, 0, 50, 101, 100, 102, 53, 49, 100, 48, 51, 55, 57, 52, 51, + // // 55, 56, 102, 56, 98, 98, 54, 51, 49, 48, 100, 52, 54, 48, 99, 50, 50, 98, 49, 17, 0, 0, + // // 0, + // ] + // .to_vec(), + // ) + // .clone() + let hash = md5([username, password].join("").as_str()); + + let mut message = Message::new(); + + message + .write_int32(1) + .write_string(username) + .write_string(password) + .write_int32(157) // version + .write_string(&hash) + .write_int32(100); //minor version + + message + } + + pub fn build_shared_folders_message(folder_count: i32, file_count: i32) -> Message { + Message::new() + .write_int32(35) + .write_int32(folder_count) + .write_int32(file_count) + .clone() + } + pub fn build_file_search_message(token: i32, query: &str) -> Message { + // Message::new_with_data( + // [ + // 26, 0, 0, 0, 219, 178, 47, 28, 11, 0, 0, 0, 116, 104, 101, 32, 119, 101, 101, 107, 101, + // 110, 100, 0, 0, + // ] + // .to_vec(), + // ) + // .clone() + Message::new() + .write_int32(26) + .write_int32(token) + .write_string(query) + .clone() + } + pub fn build_set_status_message(status_code: i32) -> Message { + Message::new() + .write_int32(28) + .write_int32(status_code) + .clone() + } + pub fn build_no_parent_message() -> Message { + Message::new().write_int32(71).write_int32(1).clone() + } + pub fn build_set_wait_port_message() -> Message { + Message::new().write_int32(2).write_int32(2234).clone() + } + pub fn build_watch_user(token: &str) -> Message { + Message::new() + .write_raw_bytes([5, 0, 0, 0, 0].to_vec()) + .write_raw_hex_string(&token) + .clone() + } +} + +#[test] +fn test_build_watch_user() { + let token = "5b581500"; + let message = MessageFactory::build_watch_user(token); + let expect: Vec = [5, 0, 0, 0, 0, 91, 88, 21, 0].to_vec(); + + assert_eq!(expect, message.get_data()) +} + +#[test] +fn test_build_login_message() { + let message = MessageFactory::build_login_message("insane_in_the_brain2", "13375137"); + + let expect: Vec = [ + 1, 0, 0, 0, 20, 0, 0, 0, 105, 110, 115, 97, 110, 101, 95, 105, 110, 95, 116, 104, 101, 95, + 98, 114, 97, 105, 110, 50, 8, 0, 0, 0, 49, 51, 51, 55, 53, 49, 51, 55, 160, 0, 0, 0, 32, 0, + 0, 0, 50, 101, 100, 102, 53, 49, 100, 48, 51, 55, 57, 52, 51, 55, 56, 102, 56, 98, 98, 54, + 51, 49, 48, 100, 52, 54, 48, 99, 50, 50, 98, 49, 17, 0, 0, 0, + ] + .to_vec(); + + // println!("{:?}", print_hex(message.get_data())); + // assert_eq!(expect, message.get_data()); + assert_eq!(expect, message.get_data()) +} + +#[test] +fn test_build_file_search_message() { + let message = MessageFactory::build_file_search_message(12, "trance wax"); + let expect: Vec = [ + 26, 0, 0, 0, 199, 19, 251, 66, 10, 0, 0, 0, 116, 114, 97, 110, 99, 101, 32, 119, 97, 120, + ] + .to_vec(); + + // [ + // 26, 0, 0, 0, 116, 111, 107, 101, 110, 113, 117, 101, 114, 121, 0, 0, 0, 113, 117, 101, 114, + // 121, + // ] + // .to_vec(); + assert_eq!(expect, message.get_data()) +} diff --git a/src/message/server/message_user.rs b/src/message/server/message_user.rs new file mode 100644 index 0000000..d52b31e --- /dev/null +++ b/src/message/server/message_user.rs @@ -0,0 +1,65 @@ +use crate::{ + message::{Message, MessageHandler}, + server::{ServerOperation, UserMessage}, +}; +use std::sync::mpsc::Sender; + +pub struct MessageUser; + +impl MessageHandler for MessageUser { + fn get_code(&self) -> u8 { + 22 + } + + fn handle(&self, message: &mut Message, _sender: Sender) { + let id = message.read_int32(); + let timestamp = message.read_int32(); + let username = message.read_string(); + let message_content = message.read_string(); + let new_message = message.read_bool(); + let user_message = UserMessage::new( + id, + timestamp, + username.clone(), + message_content, + new_message, + ); + + println!("User message received:",); + user_message.print() + } +} + +// #[cfg(test)] +// use std::sync::mpsc; +// #[cfg(test)] +// use std::sync::mpsc::{Receiver, Sender}; +// +// #[test] +// fn test_handle() { +// let (message_sender, _message_reader): (Sender, Receiver) = mpsc::channel(); +// +// let context = Arc::new(Mutex::new(Context::new(Arc::new(Mutex::new( +// message_sender, +// ))))); +// let mut message = Message::new_with_data(vec![ +// 254, 0, 0, 0, 22, 0, 0, 0, 153, 102, 47, 0, 58, 65, 228, 102, 6, 0, 0, 0, 115, 101, 114, +// 118, 101, 114, 227, 0, 0, 0, 89, 111, 117, 114, 32, 99, 111, 110, 110, 101, 99, 116, 105, +// 111, 110, 32, 105, 115, 32, 114, 101, 115, 116, 114, 105, 99, 116, 101, 100, 58, 32, 89, +// 111, 117, 32, 99, 97, 110, 110, 111, 116, 32, 115, 101, 97, 114, 99, 104, 32, 111, 114, 32, +// 99, 104, 97, 116, 46, 32, 89, 111, 117, 114, 32, 99, 108, 105, 101, 110, 116, 32, 118, 101, +// 114, 115, 105, 111, 110, 32, 105, 115, 32, 116, 111, 111, 32, 111, 108, 100, 46, 32, 89, +// 111, 117, 32, 110, 101, 101, 100, 32, 116, 111, 32, 117, 112, 103, 114, 97, 100, 101, 32, +// 116, 111, 32, 116, 104, 101, 32, 108, 97, 116, 101, 115, 116, 32, 118, 101, 114, 115, 105, +// 111, 110, 46, 32, 67, 108, 111, 115, 101, 32, 116, 104, 105, 115, 32, 99, 108, 105, 101, +// 110, 116, 44, 32, 100, 111, 119, 110, 108, 111, 97, 100, 32, 110, 101, 119, 32, 118, 101, +// 114, 115, 105, 111, 110, 32, 102, 114, 111, 109, 32, 104, 116, 116, 112, 58, 47, 47, 119, +// 119, 119, 46, 115, 108, 115, 107, 110, 101, 116, 46, 111, 114, 103, 44, 32, 105, 110, 115, +// 116, 97, 108, 108, 32, 105, 116, 32, 97, 110, 100, 32, 114, 101, 99, 111, 110, 110, 101, +// 99, 116, 46, 1, +// ]); +// message.set_pointer(8); +// MessageUser.handle(&mut message.clone(), context.clone()); +// assert_eq!(1, context.lock().unwrap().get_user_messages().len()); +// } +// // } diff --git a/src/message/server/mod.rs b/src/message/server/mod.rs new file mode 100644 index 0000000..59d0458 --- /dev/null +++ b/src/message/server/mod.rs @@ -0,0 +1,18 @@ +pub mod connect_to_peer; +pub mod excluded_search_phrases; +pub mod file_search; +pub mod login; +pub mod message_factory; +pub mod message_user; +pub mod parent_min_speed; +pub mod parent_speed_ratio; +pub mod privileged_users; +pub mod room_list; +pub mod wish_list_interval; + +pub use self::{ + connect_to_peer::*, excluded_search_phrases::*, file_search::*, login::*, message_factory::*, + message_user::*, parent_min_speed::*, parent_speed_ratio::*, privileged_users::*, room_list::*, + wish_list_interval::*, +}; + diff --git a/src/message/server/parent_min_speed.rs b/src/message/server/parent_min_speed.rs new file mode 100644 index 0000000..cfd80ac --- /dev/null +++ b/src/message/server/parent_min_speed.rs @@ -0,0 +1,20 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; + +pub struct ParentMinSpeedHandler; + +impl MessageHandler for ParentMinSpeedHandler { + fn get_code(&self) -> u8 { + 83 + } + + fn handle(&self, message: &mut Message, sender: Sender) { + let _ = sender; + let number = message.read_int32(); + println!("Parent min speed: {}", number); + } +} diff --git a/src/message/server/parent_speed_ratio.rs b/src/message/server/parent_speed_ratio.rs new file mode 100644 index 0000000..b55caef --- /dev/null +++ b/src/message/server/parent_speed_ratio.rs @@ -0,0 +1,20 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; + +pub struct ParentSpeedRatioHandler; + +// The server sends us a speed ratio determining the number of children we can have in the distributed network. The maximum number of children is our upload speed divided by the speed ratio. +impl MessageHandler for ParentSpeedRatioHandler { + fn get_code(&self) -> u8 { + 84 + } + + fn handle(&self, message: &mut Message, _sender: Sender) { + let number = message.read_int32(); + println!("Parent min speed: {}", number); + } +} diff --git a/src/message/server/privileged_users.rs b/src/message/server/privileged_users.rs new file mode 100644 index 0000000..dd48e6c --- /dev/null +++ b/src/message/server/privileged_users.rs @@ -0,0 +1,19 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; + +pub struct PrivilegedUsersHandler; + +impl MessageHandler for PrivilegedUsersHandler { + fn get_code(&self) -> u8 { + 69 + } + + fn handle(&self, message: &mut Message, _sender: Sender) { + let number = message.read_int32(); + println!("Number of privilaged users: {}", number); + } +} diff --git a/src/message/server/room_list.rs b/src/message/server/room_list.rs new file mode 100644 index 0000000..cf15212 --- /dev/null +++ b/src/message/server/room_list.rs @@ -0,0 +1,62 @@ +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; +use std::sync::mpsc::Sender; + +pub struct RoomListHandler; + +impl MessageHandler for RoomListHandler { + fn get_code(&self) -> u8 { + 64 + } + + fn handle(&self, _message: &mut Message, _sender: Sender) { + // let num_public_rooms = message.read_int32(); + // for _ in 0..num_public_rooms { + // rooms.public_rooms.push(Room::new(message.read_string(), 0)); + // } + // + // let num_public_rooms = message.read_int32(); + // for i in 0..num_public_rooms { + // rooms.public_rooms[i as usize].set_number_of_users(message.read_int32()); + // } + // + // let num_owned_private_rooms = message.read_int32(); + // for _ in 0..num_owned_private_rooms { + // rooms + // .owned_private_rooms + // .push(Room::new(message.read_string(), 0)); + // } + // + // let num_owned_private_rooms = message.read_int32(); + // for i in 0..num_owned_private_rooms { + // rooms.owned_private_rooms[i as usize].set_number_of_users(message.read_int32()); + // } + // + // let num_private_rooms = message.read_int32(); + // for _ in 0..num_private_rooms { + // rooms + // .private_rooms + // .push(Room::new(message.read_string(), 0)); + // } + // + // let num_private_rooms = message.read_int32(); + // for i in 0..num_private_rooms { + // rooms.private_rooms[i as usize].set_number_of_users(message.read_int32()); + // } + // + // let num_operated_private_rooms = message.read_int32(); + // for _ in 0..num_operated_private_rooms { + // rooms + // .operated_private_rooms + // .push(Room::new(message.read_string(), 0)); + // } + // let num_operated_private_rooms = message.read_int32(); + // for i in 0..num_operated_private_rooms { + // rooms.operated_private_rooms[i as usize].set_number_of_users(message.read_int32()); + // } + // cvat.notify_all(); + // rooms.print(); + } +} diff --git a/src/message/server/wish_list_interval.rs b/src/message/server/wish_list_interval.rs new file mode 100644 index 0000000..9e9eb58 --- /dev/null +++ b/src/message/server/wish_list_interval.rs @@ -0,0 +1,21 @@ +use std::sync::mpsc::Sender; + +use crate::{ + message::{Message, MessageHandler}, + server::ServerOperation, +}; + +pub struct WishListIntervalHandler; + +// The server tells us the wishlist search interval. +// This interval is almost always 12 minutes, or 2 minutes for privileged users. +impl MessageHandler for WishListIntervalHandler { + fn get_code(&self) -> u8 { + 104 + } + + fn handle(&self, message: &mut Message, _sender: Sender) { + let number = message.read_int32(); + println!("Wishlist search interval: {} in seconds", number); + } +} diff --git a/src/peer/default_peer.rs b/src/peer/default_peer.rs new file mode 100644 index 0000000..2e3c5aa --- /dev/null +++ b/src/peer/default_peer.rs @@ -0,0 +1,142 @@ +use crate::message::server::MessageFactory; +use crate::message::{Message, MessageReader}; +use std::sync::mpsc; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, Barrier}; +use std::thread::{self}; + +use super::peer::Peer; +use crate::client::ClientOperation; +use std::io::{self, Write}; +use std::net::TcpStream; +use std::net::ToSocketAddrs; +use std::time::Duration; + +pub struct DefaultPeer { + peer: Peer, + // client_channel: Sender, + peer_channel: Option>, +} + +#[allow(dead_code)] +pub enum PeerOperation { + SendMessage(Message), +} +impl DefaultPeer { + pub fn new(peer: Peer, _client_channel: Sender) -> Self { + Self { + peer, + // client_channel, + peer_channel: None, + } + } + pub fn connect(mut self) -> Result { + // println!("Connecting to peer {:?}", self.peer); + let socket_address = format!("{}:{}", self.peer.host, self.peer.port) + .to_socket_addrs()? + .next() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?; + + let mut stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?; + + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + stream.set_write_timeout(Some(Duration::from_secs(5)))?; + if let Some(token) = self.peer.token.clone() { + stream + .write_all(&MessageFactory::build_watch_user(&token).get_data()) + .unwrap(); + // self.queue_message(build_watch_user(&token)); + } + self.start_read_write_loops(stream).unwrap(); + + Ok(self) + } + + fn start_read_write_loops(&mut self, stream: TcpStream) -> Result<(), io::Error> { + let (peer_sender, peer_reader): (Sender, Receiver) = + mpsc::channel(); + self.peer_channel = Some(peer_sender); + + let barrier = Arc::new(Barrier::new(3)); + let read_barrier = barrier.clone(); + let write_barrier = barrier.clone(); + let done_barrier = barrier.clone(); + + let mut read_stream = stream.try_clone()?; + let mut write_stream = stream.try_clone()?; + + let peer = self.peer.clone(); + thread::spawn(move || { + read_barrier.wait(); + let mut buffered_reader = MessageReader::new(); + loop { + match buffered_reader.read_from_socket(&mut read_stream) { + Ok(_) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + println!( + "Read operation timed out in default peer {:}:{:}", + peer.host, peer.port + ); + continue; + } + Err(e) => { + eprintln!("Error reading from peer: {}", e); + break; + } + } + + match buffered_reader.extract_message() { + Ok(Some(message)) => { + println!( + "Received message in default peer: {:?}", + message.get_message_code_u32() + ); + } + Err(e) => { + println!("Error extracting message in default peer: {}", e) + } + Ok(None) => continue, + } + } + }); + + // thread::spawn(move || loop { + // thread::sleep(Duration::from_secs(1)); + // // println!( + // // "Default peer ({}) - stream state: connected={}", + // // px.host, + // // stream.peer_addr().is_ok() + // // ); + // }); + thread::spawn(move || { + write_barrier.wait(); + loop { + if let Ok(operation) = peer_reader.recv() { + match operation { + PeerOperation::SendMessage(message) => { + match write_stream.write_all(&message.get_buffer()) { + Ok(_) => {} + Err(e) => { + eprintln!("Error writing message to stream : {}", e); + break; + } + } + } + } + } + } + }); + done_barrier.wait(); + Ok(()) + } + + // pub fn queue_message(&self, message: Message) { + // if let Some(sender) = self.peer_channel.clone() { + // if let Err(e) = sender.send(PeerOperation::SendMessage(message)) { + // eprintln!("Failed to send: {}", e); + // } + // } + // } +} diff --git a/src/peer/listen.rs b/src/peer/listen.rs new file mode 100644 index 0000000..c4b9e06 --- /dev/null +++ b/src/peer/listen.rs @@ -0,0 +1,45 @@ +use std::{io, net::TcpListener}; + +use crate::message::MessageReader; + +pub struct Listen {} + +impl Listen { + pub fn new(port: u32) { + println!("starting listener on port {port}"); + let listener = TcpListener::bind(format!("0.0.0.0:{port}")).unwrap(); + for stream in listener.incoming() { + let mut read_stream = stream.unwrap(); + + let mut buffered_reader = MessageReader::new(); + loop { + match buffered_reader.read_from_socket(&mut read_stream) { + Ok(_) => {} + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, + Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { + println!("Read operation timed out"); + continue; + } + Err(e) => { + eprintln!("Error reading from server: {}", e); + break; + } + } + + match buffered_reader.extract_message() { + Ok(Some(message)) => { + println!("Received message: {:?}", message.get_message_code_u32()); + println!("{:?}", message.get_data()); + } + Err(e) => { + println!("Error extracting message: {}", e) + } + Ok(None) => continue, + } + } + + println!("Connection established!"); + } + } +} diff --git a/src/peer/mod.rs b/src/peer/mod.rs new file mode 100644 index 0000000..61b578d --- /dev/null +++ b/src/peer/mod.rs @@ -0,0 +1,3 @@ +pub mod default_peer; +pub mod listen; +pub mod peer; diff --git a/src/peer/peer.rs b/src/peer/peer.rs new file mode 100644 index 0000000..a31fdc2 --- /dev/null +++ b/src/peer/peer.rs @@ -0,0 +1,153 @@ +use core::fmt; +use std::str::FromStr; + +use crate::message::Message; + +#[derive(Debug, Clone)] +pub enum ConnectionType { + P, + F, + D, +} + +// Define the error type +#[derive(Debug, Clone)] +pub struct ParseConnectionTypeError; + +impl fmt::Display for ParseConnectionTypeError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "invalid connection type") + } +} + +impl std::error::Error for ParseConnectionTypeError {} + +// Implement FromStr for ConnectionType +impl FromStr for ConnectionType { + type Err = ParseConnectionTypeError; + + fn from_str(s: &str) -> Result { + match s { + "P" => Ok(ConnectionType::P), + "F" => Ok(ConnectionType::F), + "D" => Ok(ConnectionType::D), + _ => Err(ParseConnectionTypeError), + } + } +} + +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct Peer { + pub username: String, + pub connection_type: ConnectionType, + pub host: String, + pub port: i32, + pub token: Option, + pub privileged: i8, + pub unknown: i8, + pub obfuscated_port: i8, +} +impl Peer { + #[allow(dead_code)] + pub fn new( + username: String, + connection_type: ConnectionType, + host: String, + port: i32, + token: String, + privileged: i8, + unknown: i8, + obfuscated_port: i8, + ) -> Self { + Self { + username, + connection_type, + host, + port, + token: Some(token), + privileged, + unknown, + obfuscated_port, + } + } + #[allow(dead_code)] + pub fn new_from_message(message: &mut Message) -> Self { + let username = message.read_string(); + let connection_type: ConnectionType = message.read_string().parse().unwrap(); + + let mut ip: Vec = vec![]; + for _ in 0..4 { + ip.push(message.read_int8().into()); + } + let host: String = format!( + "{}.{}.{}.{}", + ip[3].abs(), + ip[2].abs(), + ip[1].abs(), + ip[0].abs() + ); + + let (port, token, privileged, unknown, obfuscated_port) = ( + message.read_int32(), + message.read_raw_hex_str(4), + message.read_int8(), + message.read_int8(), + message.read_int8(), + ); + + Self { + username, + connection_type, + host, + port, + token: Some(token), + privileged, + unknown, + obfuscated_port, + } + } +} +#[test] +fn test_new_from_message() { + let data: Vec = [ + 36, 0, 0, 0, 18, 0, 0, 0, 2, 0, 0, 0, 100, 112, 1, 0, 0, 0, 80, 27, 231, 37, 45, 186, 8, 0, + 0, 178, 78, 25, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + ] + .to_vec(); + let mut message = Message::new_with_data(data); + message.set_pointer(8); + + let peer = Peer::new_from_message(&mut message); + + assert_eq!(peer.username, "dp"); + assert!(matches!(peer.connection_type, ConnectionType::P)); + assert_eq!(peer.host, "45.37.25.27"); + assert_eq!(peer.port, 2234); + assert_eq!(peer.token, Some("b24e1900".to_string())); + assert_eq!(peer.privileged, 0); + assert_eq!(peer.unknown, 0); + assert_eq!(peer.obfuscated_port, 0); +} + +#[test] +fn test_new_from_message2() { + let data: Vec = [ + 42, 0, 0, 0, 18, 0, 0, 0, 8, 0, 0, 0, 103, 114, 97, 110, 100, 112, 97, 103, 1, 0, 0, 0, 80, + 137, 128, 193, 68, 187, 8, 0, 0, 58, 16, 0, 0, 0, 1, 0, 0, 0, 188, 8, 0, 0, + ] + .to_vec(); + let mut message = Message::new_with_data(data); + message.set_pointer(8); + + let peer = Peer::new_from_message(&mut message); + + assert_eq!(peer.username, "grandpag"); + assert!(matches!(peer.connection_type, ConnectionType::P)); + assert_eq!(peer.host, "68.63.128.119"); + assert_eq!(peer.port, 2235); + assert_eq!(peer.token, Some("3a100000".to_string())); + assert_eq!(peer.privileged, 0); + assert_eq!(peer.unknown, 1); + assert_eq!(peer.obfuscated_port, 0); +} diff --git a/src/server.rs b/src/server.rs index c685976..20738b5 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,20 +1,27 @@ +use crate::client::ClientOperation; use crate::dispatcher::MessageDispatcher; -use crate::message::factory::{build_init_message, build_login_message}; -use crate::message::{Message, MessageReader}; +use crate::message::server::MessageFactory; +use crate::message::Handlers; +use crate::message::Message; +use crate::message::MessageReader; +use crate::peer::listen::Listen; +use crate::peer::peer::Peer; + use std::io::{self, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::sync::mpsc; use std::sync::mpsc::{Receiver, Sender}; use std::sync::{Arc, Barrier, Mutex}; use std::thread::{self}; -use std::time::Duration; +use std::time::{Duration, Instant}; + #[derive(Debug, Clone)] -pub struct ServerAddress { +pub struct PeerAddress { host: String, port: u16, } -impl ServerAddress { +impl PeerAddress { pub fn new(host: String, port: u16) -> Self { Self { host, port } } @@ -30,26 +37,61 @@ impl ServerAddress { #[derive(Debug)] pub struct Context { - message_sender: Arc>>, + pub logged_in: Option, + #[allow(dead_code)] rooms: Rooms, } impl Context { - pub fn new(message_sender: Arc>>) -> Self { + pub fn new() -> Self { Self { - message_sender, + #[allow(dead_code)] rooms: Rooms::new(), + logged_in: Option::None, } } - pub fn get_rooms(&mut self) -> &mut Rooms { - &mut self.rooms - } - pub fn queue_message(&self, message: Message) { - self.message_sender.lock().unwrap().send(message).unwrap(); + // pub fn get_messages_for_user(&self, username: String) -> Option<&UserMessage> { + // self.user_messages.get(&username) + // } + // + // pub fn get_rooms(&mut self) -> &mut Rooms { + // &mut self.rooms + // } +} +#[derive(Debug, Clone)] +pub struct UserMessage { + id: i32, + timestamp: i32, + username: String, + message: String, + new_message: bool, +} +impl UserMessage { + pub fn new( + id: i32, + timestamp: i32, + username: String, + message: String, + new_message: bool, + ) -> Self { + Self { + id, + timestamp, + username, + message, + new_message, + } + } + pub fn print(&self) { + println!( + "Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}", + self.timestamp, self.username, self.id, self.new_message, self.message + ); } } +#[allow(dead_code)] #[derive(Debug)] pub struct Rooms { pub public_rooms: Vec, @@ -66,6 +108,8 @@ impl Rooms { operated_private_rooms: Vec::new(), } } + + #[allow(dead_code)] pub fn print(&self) { println!("Public rooms ({}):", self.public_rooms.len()); for room in &self.public_rooms { @@ -94,12 +138,14 @@ pub struct Room { number_of_users: i32, } impl Room { + #[allow(dead_code)] pub fn new(name: String, number_of_users: i32) -> Self { Self { name, number_of_users, } } + #[allow(dead_code)] pub fn set_number_of_users(&mut self, number_of_users: i32) { self.number_of_users = number_of_users; } @@ -109,42 +155,52 @@ impl Room { self.name, self.number_of_users ); } - - // pub fn get_name(&self) -> &str { - // &self.name - // } - // pub fn get_number_of_users(&self) -> i32 { - // self.number_of_users - // } } +pub enum ServerOperation { + LoginStatus(bool), + SendMessage(Message), + #[allow(dead_code)] + ConnectToPeer(Peer), +} + +#[derive(Debug)] pub struct Server { - address: ServerAddress, + address: PeerAddress, + sender: Sender, context: Arc>, } impl Server { /// Create a new instance of Server, returning a Result - pub fn new(address: ServerAddress) -> Result { - let (message_sender, message_reader): (Sender, Receiver) = + pub fn new( + address: PeerAddress, + client_channel: Sender, + ) -> Result { + let (sender, server_channel): (Sender, Receiver) = mpsc::channel(); - let context = Arc::new(Mutex::new(Context::new(Arc::new(Mutex::new( - message_sender, - ))))); + let context = Arc::new(Mutex::new(Context::new())); - let server = Self { address, context }; - - server.start_read_write_loops(message_reader)?; + let mut server = Self { + address, + context, + sender, + }; + server.start_read_write_loops(server_channel, client_channel)?; Ok(server) } - pub fn get_address(&self) -> &ServerAddress { + pub fn get_address(&self) -> &PeerAddress { &self.address } /// Start reading and writing loops in separate threads - fn start_read_write_loops(&self, message_reader: Receiver) -> Result<(), io::Error> { + fn start_read_write_loops( + &mut self, + server_channel: Receiver, + client_channel: Sender, + ) -> Result<(), io::Error> { let socket_address = format!("{}:{}", self.address.host, self.address.port) .to_socket_addrs()? .next() @@ -163,15 +219,16 @@ impl Server { let mut read_stream = stream.try_clone()?; let mut write_stream = stream.try_clone()?; - let barrier = Arc::new(Barrier::new(2)); + let barrier = Arc::new(Barrier::new(3)); let read_barrier = barrier.clone(); let write_barrier = barrier.clone(); - - let dispatcher = MessageDispatcher::new(Arc::clone(&self.context)); + let done_barrier = barrier.clone(); + let sender = self.sender.clone(); thread::spawn(move || { read_barrier.wait(); - + let message_handlers = Handlers::new_with_server_handlers(); + let dispatcher = MessageDispatcher::new(sender, message_handlers); let mut buffered_reader = MessageReader::new(); loop { match buffered_reader.read_from_socket(&mut read_stream) { @@ -190,7 +247,9 @@ impl Server { match buffered_reader.extract_message() { Ok(Some(mut message)) => { - println!("Received message: {:?}", message.get_data()); + // println!("Received message: {:?}", message.get_message_code_u32()); + // message.print_hex(); + dispatcher.dispatch(&mut message) } Err(e) => { @@ -201,31 +260,96 @@ impl Server { } }); + // let monitor_handle = thread::spawn(move || loop { + // thread::sleep(Duration::from_secs(1)); + // println!("Stream state: connected={}", stream.peer_addr().is_ok()); + // }); + let context = self.context.clone(); thread::spawn(move || { write_barrier.wait(); - loop { - let message = message_reader.recv().unwrap(); - println!("Sending message from queue: {:?}", message); // Debug log when sending - match write_stream.write_all(&message.get_data()) { - Ok(_) => println!("Sent buffered message: {:?}", message), - Err(e) => { - eprintln!("Failed to send buffered message: {}", e); - break; + if let Ok(operation) = server_channel.recv() { + match operation { + ServerOperation::ConnectToPeer(peer) => { + match client_channel.send(ClientOperation::ConnectToPeer(peer)) { + Ok(_) => {} + Err(e) => { + println!("Error sending message: {}", e); + } + } + } + ServerOperation::LoginStatus(message) => { + context.lock().unwrap().logged_in = Some(message); + } + ServerOperation::SendMessage(message) => { + // message.decode(); + // message.print_hex2(); + match write_stream.write_all(&message.get_buffer()) { + Ok(_) => {} + Err(e) => { + eprintln!("Error writing message to stream : {}", e); + break; + } + } + } } } } }); + done_barrier.wait(); Ok(()) } fn queue_message(&self, message: Message) { - self.context.lock().unwrap().queue_message(message); + match self.sender.send(ServerOperation::SendMessage(message)) { + Ok(_) => {} + Err(e) => println!("Failed to send: {}", e), + } + } + fn start_listener(&self) { + thread::spawn(move || Listen::new(2234)); } - pub fn login(&mut self, username: &str, password: &str) -> Result<(), std::io::Error> { - self.queue_message(build_init_message()); - self.queue_message(build_login_message(username, password)); - Ok(()) + pub fn login(&self, username: &str, password: &str) -> Result { + self.start_listener(); + // Send the login message + self.queue_message(MessageFactory::build_login_message(username, password)); + let context = self.context.clone(); + let mut logged_in; + let timeout = Duration::from_secs(5); + let start = Instant::now(); + + // wait till server says your logged in or not + loop { + if start.elapsed() > timeout { + println!("Timeout waiting for login response"); + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "Timeout waiting for login response", + )); + } + + { + logged_in = context.lock().unwrap().logged_in.clone() + } + + if !logged_in.is_none() { + break; + } + } + + if logged_in.unwrap() { + println!("Logged in as {}", username); + self.queue_message(MessageFactory::build_set_wait_port_message()); + self.queue_message(MessageFactory::build_shared_folders_message(1, 1)); + self.queue_message(MessageFactory::build_no_parent_message()); + self.queue_message(MessageFactory::build_set_status_message(2)); + } + + Ok(logged_in.unwrap()) + } + + pub fn file_search(&self, token: i32, query: &str) { + self.queue_message(MessageFactory::build_file_search_message(token, query)); } } diff --git a/src/utils/md5.rs b/src/utils/md5.rs new file mode 100644 index 0000000..71277b4 --- /dev/null +++ b/src/utils/md5.rs @@ -0,0 +1,471 @@ +use std::convert::TryInto; +/** +* +* The MD5 Message-Digest Algorithm, written in Rust. +* Implemented by michael dimovich (@mdimovich) +* +* The purpose of this code is to act as a learning resource +* for the md5 message digest algorithm. I wanted to provide +* a straightforward walk-through of the md5 algorithm using +* Rust. +* +* This code was implemented using the original md5 message digest +* RFC, written by R. Rivest in April 1992. +* +* For more information about the md5 algorithm, or the original C +* source code implementation, please reference RFC 1321. +* +* (source: https://datatracker.ietf.org/doc/html/rfc1321) +* +*/ + +/** +* This function is specified in the rfc as a way to generate the +* table that we use in rounds 1-4 as part of the hashing operation. +* The function is specified as: +* K[i] = floor(2^32 * abs(sin(i+1))) +*/ +fn table_construction_function(i: u32) -> u32 { + let x: f64 = i as f64; + let sin_eval = x.sin().abs(); + // note: 4294967296 == 2^32 + return (4294967296.0 * sin_eval) as u32; +} + +/** +* 4 auxiliary functions that take 3 32-bit +* integers as input and return 1 32 bit integer as output. +* +* these bitwise operations are the crux of the md5 hashing +* algorithm, and they are performed a number of times to end +* up with our final result. +*/ +fn f(x: u32, y: u32, z: u32) -> u32 { + x & y | !x & z +} + +fn g(x: u32, y: u32, z: u32) -> u32 { + x & z | y & !z +} + +fn h(x: u32, y: u32, z: u32) -> u32 { + x ^ y ^ z +} + +fn i(x: u32, y: u32, z: u32) -> u32 { + y ^ (x | !z) +} + +// utility function to convert a vector of type T with size N into an array of type T with size N. +fn vec_to_array(v: Vec) -> [T; N] { + v.try_into() + .unwrap_or_else(|_v: Vec| panic!("error converting vector to array - sizes don't match")) +} + +/** +* Round 1 function is specified as 16 operations using the following: +* Let [abcd, k s i] denote the operation - +* a = b + ((a + f(b,c,d) + x[k] + table[i]) <<< s); +* where x[k] is a 32 bit chunk of our original message, +* table[i] is an entry generated by our sine function, s is an unsigned +* integer, and a,b,c,d are our 32 bit registers we perform operations on. +*/ +fn round_one_operations( + mut a: u32, + mut b: u32, + mut c: u32, + mut d: u32, + table: &Vec, + x: &Vec, +) -> [u32; 4] { + macro_rules! round1 { + ( $a:ident, $b:ident, $c:ident, $d:ident, $k:expr, $s:expr, $i: expr ) => { + $a = $b.wrapping_add( + ($a.wrapping_add(f($b, $c, $d)) + .wrapping_add(x[$k]) + .wrapping_add(table[$i])) + .rotate_left($s), + ) + }; + } + + round1!(a, b, c, d, 0, 7, 1); + round1!(d, a, b, c, 1, 12, 2); + round1!(c, d, a, b, 2, 17, 3); + round1!(b, c, d, a, 3, 22, 4); + + round1!(a, b, c, d, 4, 7, 5); + round1!(d, a, b, c, 5, 12, 6); + round1!(c, d, a, b, 6, 17, 7); + round1!(b, c, d, a, 7, 22, 8); + + round1!(a, b, c, d, 8, 7, 9); + round1!(d, a, b, c, 9, 12, 10); + round1!(c, d, a, b, 10, 17, 11); + round1!(b, c, d, a, 11, 22, 12); + + round1!(a, b, c, d, 12, 7, 13); + round1!(d, a, b, c, 13, 12, 14); + round1!(c, d, a, b, 14, 17, 15); + round1!(b, c, d, a, 15, 22, 16); + + return [a, b, c, d]; +} + +/** +* Round 2 is equivalent to round 1, except we swap out our f function for g as such: +* a = b + ((a + g(b,c,d) + x[k] + table[i]) <<< s); +*/ +fn round_two_operations( + mut a: u32, + mut b: u32, + mut c: u32, + mut d: u32, + table: &Vec, + x: &Vec, +) -> [u32; 4] { + macro_rules! round2 { + ( $a:ident, $b:ident, $c:ident, $d:ident, $k:expr, $s:expr, $i:expr) => { + $a = $b.wrapping_add( + ($a.wrapping_add(g($b, $c, $d)) + .wrapping_add(x[$k]) + .wrapping_add(table[$i])) + .rotate_left($s), + ) + }; + } + + round2!(a, b, c, d, 1, 5, 17); + round2!(d, a, b, c, 6, 9, 18); + round2!(c, d, a, b, 11, 14, 19); + round2!(b, c, d, a, 0, 20, 20); + + round2!(a, b, c, d, 5, 5, 21); + round2!(d, a, b, c, 10, 9, 22); + round2!(c, d, a, b, 15, 14, 23); + round2!(b, c, d, a, 4, 20, 24); + + round2!(a, b, c, d, 9, 5, 25); + round2!(d, a, b, c, 14, 9, 26); + round2!(c, d, a, b, 3, 14, 27); + round2!(b, c, d, a, 8, 20, 28); + + round2!(a, b, c, d, 13, 5, 29); + round2!(d, a, b, c, 2, 9, 30); + round2!(c, d, a, b, 7, 14, 31); + round2!(b, c, d, a, 12, 20, 32); + + return [a, b, c, d]; +} + +/** +* Round 3 is equivalent to round 1, except we swap out our f function for h as such: +* a = b + ((a + h(b,c,d) + x[k] + table[i]) <<< s); +*/ +fn round_three_operations( + mut a: u32, + mut b: u32, + mut c: u32, + mut d: u32, + table: &Vec, + x: &Vec, +) -> [u32; 4] { + macro_rules! round3 { + ( $a:ident, $b:ident, $c:ident, $d:ident, $k:expr, $s:expr, $i:expr ) => { + $a = $b.wrapping_add( + ($a.wrapping_add(h($b, $c, $d)) + .wrapping_add(x[$k]) + .wrapping_add(table[$i])) + .rotate_left($s), + ) + }; + } + + round3!(a, b, c, d, 5, 4, 33); + round3!(d, a, b, c, 8, 11, 34); + round3!(c, d, a, b, 11, 16, 35); + round3!(b, c, d, a, 14, 23, 36); + + round3!(a, b, c, d, 1, 4, 37); + round3!(d, a, b, c, 4, 11, 38); + round3!(c, d, a, b, 7, 16, 39); + round3!(b, c, d, a, 10, 23, 40); + + round3!(a, b, c, d, 13, 4, 41); + round3!(d, a, b, c, 0, 11, 42); + round3!(c, d, a, b, 3, 16, 43); + round3!(b, c, d, a, 6, 23, 44); + + round3!(a, b, c, d, 9, 4, 45); + round3!(d, a, b, c, 12, 11, 46); + round3!(c, d, a, b, 15, 16, 47); + round3!(b, c, d, a, 2, 23, 48); + + return [a, b, c, d]; +} + +/** +* Round 4 is equivalent to round 1, except we swap out our f function for h as such: +* a = b + ((a + i(b,c,d) + x[k] + table[i]) <<< s); +*/ +fn round_four_operations( + mut a: u32, + mut b: u32, + mut c: u32, + mut d: u32, + table: &Vec, + x: &Vec, +) -> [u32; 4] { + macro_rules! round4 { + ( $a:ident, $b:ident, $c:ident, $d:ident, $k:expr, $s:expr, $i:expr ) => { + $a = $b.wrapping_add( + ($a.wrapping_add(i($b, $c, $d)) + .wrapping_add(x[$k]) + .wrapping_add(table[$i])) + .rotate_left($s), + ) + }; + } + + round4!(a, b, c, d, 0, 6, 49); + round4!(d, a, b, c, 7, 10, 50); + round4!(c, d, a, b, 14, 15, 51); + round4!(b, c, d, a, 5, 21, 52); + + round4!(a, b, c, d, 12, 6, 53); + round4!(d, a, b, c, 3, 10, 54); + round4!(c, d, a, b, 10, 15, 55); + round4!(b, c, d, a, 1, 21, 56); + + round4!(a, b, c, d, 8, 6, 57); + round4!(d, a, b, c, 15, 10, 58); + round4!(c, d, a, b, 6, 15, 59); + round4!(b, c, d, a, 13, 21, 60); + + round4!(a, b, c, d, 4, 6, 61); + round4!(d, a, b, c, 11, 10, 62); + round4!(c, d, a, b, 2, 15, 63); + round4!(b, c, d, a, 9, 21, 64); + + return [a, b, c, d]; +} + +/** +* utility function to iterate over our slice of u8 ints +* and convert into a vector of unsigned 32 bit ints +*/ +fn convert_u8_chunk_to_u32(chunk: &mut [u8]) -> Vec { + let mut x: Vec = Vec::new(); + + let mut count = 0; + let mut temporary_vec: Vec = Vec::new(); + // iterate over our block and take + // our 8 bit ints and convert them to + // 32 bit integers + for i in 0..chunk.len() { + temporary_vec.push(chunk[i]); + count += 1; + if count == 4 { + let temp_arr: [u8; 4] = vec_to_array(temporary_vec.clone()); + let value = u32::from_ne_bytes(temp_arr); + x.push(value); + count = 0; + temporary_vec.clear(); + } + } + return x; +} + +fn compute_md5_digest(mut v: Vec) -> String { + // as described in the rfc, + // 4 32-bit words initialized as fixed constants. + let mut word_a = 0x67452301u32; + let mut word_b = 0xefcdab89u32; + let mut word_c = 0x98badcfeu32; + let mut word_d = 0x10325476u32; + + // construct the 64 element constant table. + let table = construct_value_table(); + + // let M[0 .. N-1] = words of resulting message, where N is multiple of 16 + for chunk in v.chunks_exact_mut(64) { + let x = convert_u8_chunk_to_u32(chunk); + + // set all values of a,b,c,d to aa,bb,cc, and dd respectively. + // this is to save the initial values. + let word_aa = word_a; + let word_bb = word_b; + let word_cc = word_c; + let word_dd = word_d; + + // execute round 1 + let result = round_one_operations(word_a, word_b, word_c, word_d, &table, &x); + word_a = result[0]; + word_b = result[1]; + word_c = result[2]; + word_d = result[3]; + + // execute round 2 + let result = round_two_operations(word_a, word_b, word_c, word_d, &table, &x); + + word_a = result[0]; + word_b = result[1]; + word_c = result[2]; + word_d = result[3]; + + // execute round 3 + let result = round_three_operations(word_a, word_b, word_c, word_d, &table, &x); + word_a = result[0]; + word_b = result[1]; + word_c = result[2]; + word_d = result[3]; + + // execute round 4 + let result = round_four_operations(word_a, word_b, word_c, word_d, &table, &x); + word_a = result[0]; + word_b = result[1]; + word_c = result[2]; + word_d = result[3]; + + // at end of loop iteration, add original word + // to the current word value + word_a = word_a.wrapping_add(word_aa); + word_b = word_b.wrapping_add(word_bb); + word_c = word_c.wrapping_add(word_cc); + word_d = word_d.wrapping_add(word_dd); + } + + // format and return the final result, which + // is a 128-bit digest string. + let message_digest = format!( + "{:08x}{:08x}{:08x}{:08x}", + word_a.swap_bytes(), + word_b.swap_bytes(), + word_c.swap_bytes(), + word_d.swap_bytes() + ); + return message_digest; +} + +/* +* BIT PADDING STEP +* 1. Append a 1 bit to the end of original message. +* 2. Until the bit length of the data is 64 bits +* short of being a multiple of 512, append 0's +* to the data. +* 3. With the last 64 bits, append the length in 64 bits +* (in lower-order bits first). +*/ +fn bit_padding(input: &str) -> Vec { + let mut input_vector: Vec = convert_str_to_vec(input); + let bit_length: u64 = (input.len() as u64) * 8u64; // todo - add support for > 2^64 bit size + + // 128_u8 is the equivalent of padding 1 as an unsigned 8-bit integer + // with lower-order bits first + input_vector.push(128_u8); + //check if bit length % 512 is 448 (64 less than 512) + while (input_vector.len() * 8) % 512 != 448 { + input_vector.push(0_u8); // push in another 8-bit 0 padded value until the correct + // result is reached; + } + + let length_bits_as_u8_array = split_u64_to_u8_array(bit_length); + input_vector.extend(length_bits_as_u8_array); + + return input_vector; +} + +fn split_u64_to_u8_array(s: u64) -> [u8; 8] { + let u8_array = [ + s as u8, + (s >> 8) as u8, + (s >> 16) as u8, + (s >> 24) as u8, + (s >> 32) as u8, + (s >> 40) as u8, + (s >> 48) as u8, + (s >> 56) as u8, + ]; + return u8_array; +} + +fn construct_value_table() -> Vec { + let mut t: Vec = Vec::new(); + t.push(0x00000000); + for i in 1..=64 { + t.push(table_construction_function(i)); + } + return t; +} + +// this should only work with utf-8 encoding and not full unicode support +// due to multi-byte unicode chars +fn convert_str_to_vec(input: &str) -> Vec { + let mut byte_vec: Vec = Vec::new(); + byte_vec.extend(input.as_bytes()); + return byte_vec; +} + +/** +* A Basic Overview of this MD5 Implementation: +* 1. Take in a command line string and convert into +* a vector of unsigned 8 bit ints (Vec) +* 2. Pad the message with a single 1, followed by n 0's, +* such that the message length in bits % 512 is equal to 448 +* 3. Pad the last 64 bits with a little-endian representation +* of the original message length. +* 4. Convert our vector of unsigned 8-bit integers into +* blocks of 32 bit unsigned ints (u32). +* 5. Using the 4 auxiliary functions described in rfc 1321, +* begin a series of 64 bitwise operations on each 512 bit +* block of our message. +* 6. Once we are done, combine our 4 32 bit registers used to +* during our bitwise operations to retrieve our message digest. +* (This is a simplification, but describes the core functionality) +*/ +pub fn md5(input: &str) -> String { + let input_vec = bit_padding(input); + return compute_md5_digest(input_vec); +} +#[test] +fn a_correct_hash() { + assert_eq!("0cc175b9c0f1b6a831c399e269772661", md5("a")); +} + +#[test] +fn empty_string_correct_hash() { + assert_eq!("d41d8cd98f00b204e9800998ecf8427e", md5("")); +} + +#[test] +fn abc_correct_hash() { + assert_eq!("900150983cd24fb0d6963f7d28e17f72", md5("abc")); +} + +#[test] +fn message_digest_correct_hash() { + assert_eq!("f96b697d7cb7938d525a2f31aaf161d0", md5("message digest")); +} + +#[test] +fn lowercase_alphabet_correct_hash() { + assert_eq!( + "c3fcd3d76192e4007dfb496cca67e13b", + md5("abcdefghijklmnopqrstuvwxyz") + ) +} + +#[test] +fn all_alphanumeric_correct_hash() { + let alpha_num_str = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"; + + assert_eq!("d174ab98d277d9f5a5611c2c9f419d9f", md5(alpha_num_str)); +} + +#[test] +fn repeated_numbers_correct_hash() { + let repeated_number_str = + "12345678901234567890123456789012345678901234567890123456789012345678901234567890"; + + assert_eq!("57edf4a22be3c955ac49da2e2107b67a", md5(repeated_number_str)); +} diff --git a/src/utils/mod.rs b/src/utils/mod.rs new file mode 100644 index 0000000..4c3a128 --- /dev/null +++ b/src/utils/mod.rs @@ -0,0 +1,3 @@ +pub mod md5; + +pub use md5::*;