diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 39975b339b..fd5de14d11 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -14,6 +14,7 @@ use std::net::SocketAddr; use std::net::SocketAddrV4; use std::time::Duration; use tokio::runtime::Runtime; +use sfw_provider_requests::AuthToken; pub mod directory; pub mod mix; @@ -63,22 +64,25 @@ pub struct NymClient { pub input_tx: mpsc::UnboundedSender, // to be used by "send" function or socket, etc input_rx: mpsc::UnboundedReceiver, - + socket_listening_address: SocketAddr, is_local: bool, + auth_token: Option } #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); impl NymClient { - pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self { + pub fn new(address: DestinationAddressBytes, socket_listening_address: SocketAddr, is_local: bool, auth_token: Option) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { address, input_tx, input_rx, + socket_listening_address, is_local, + auth_token, } } @@ -129,26 +133,29 @@ impl NymClient { } } - async fn start_provider_polling(provider_address: SocketAddrV4) { - let provider_client = ProviderClient::new(); + async fn start_provider_polling(provider_client: ProviderClient) { +// let provider_client = ProviderClient::new(provider_address); loop { println!("[FETCH MSG] - Polling provider..."); let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY); tokio::time::delay_for(delay_duration).await; - provider_client - .retrieve_messages(provider_address) + let messages = provider_client + .retrieve_messages() .await .unwrap(); + for message in messages { + println!("Retrieved: {:?}", String::from_utf8(message).unwrap()) + } } } - pub fn start(self, socket_address: SocketAddr) -> Result<(), Box> { + pub fn start(self) -> Result<(), Box> { println!("starting nym client"); - - let (mix_tx, mix_rx) = mpsc::unbounded(); let mut rt = Runtime::new()?; + let topology = get_topology(self.is_local); + // this is temporary and assumes there exists only a single provider. let provider_address: SocketAddrV4 = topology .mix_provider_nodes .first() @@ -157,6 +164,22 @@ impl NymClient { .parse() .unwrap(); + let mut provider_client = ProviderClient::new(provider_address, self.address, self.auth_token); + + rt.block_on(async { + match self.auth_token { + None => { + let auth_token = provider_client.register().await.unwrap(); + provider_client.update_token(auth_token); + println!("Obtained new token! - {:?}", auth_token); + }, + Some(token) => println!("Already got the token! - {:?}", token), + } + }); + + + let (mix_tx, mix_rx) = mpsc::unbounded(); + let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream( mix_tx.clone(), @@ -171,8 +194,8 @@ impl NymClient { topology.clone(), )); - let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_address)); - let websocket_future = rt.spawn(ws::start_websocket(socket_address, self.input_tx)); + let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_client)); + let websocket_future = rt.spawn(ws::start_websocket(self.socket_listening_address, self.input_tx)); rt.block_on(async { let future_results = join5( diff --git a/src/clients/provider.rs b/src/clients/provider.rs index ff2fdb6bac..eabed21ef4 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,50 +1,115 @@ -use sfw_provider_requests::requests::{ProviderRequest, PullRequest}; -use sfw_provider_requests::responses::{ProviderResponse, PullResponse}; +use sfw_provider_requests::requests::{ProviderRequest, PullRequest, RegisterRequest}; +use sfw_provider_requests::responses::{ProviderResponse, PullResponse, RegisterResponse, ProviderResponseError}; use std::net::Shutdown; use std::net::SocketAddrV4; use tokio::prelude::*; -use tokio::time::Duration; +use std::time::Duration; +use sphinx::route::DestinationAddressBytes; +use futures::io::Error; +use sfw_provider_requests::AuthToken; -pub struct ProviderClient {} +#[derive(Debug)] +pub enum ProviderClientError { + ClientAlreadyRegisteredError, + EmptyAuthTokenError, + NetworkError, + + InvalidRequestError, + InvalidResponseError, + InvalidResponseLengthError, +} + +impl From for ProviderClientError { + fn from(_: Error) -> Self { + use ProviderClientError::*; + + NetworkError + } +} + +impl From for ProviderClientError { + fn from(err: ProviderResponseError) -> Self { + use ProviderClientError::*; + match err { + ProviderResponseError::MarshalError => InvalidRequestError, + ProviderResponseError::UnmarshalError => InvalidResponseError, + ProviderResponseError::UnmarshalErrorInvalidLength => InvalidResponseLengthError, + } + } +} + +pub struct ProviderClient { + provider_network_address: SocketAddrV4, + our_address: DestinationAddressBytes, + auth_token: Option, +} impl ProviderClient { - pub fn new() -> Self { - ProviderClient {} - } - - pub async fn retrieve_messages( - &self, - provider: SocketAddrV4, - ) -> Result<(), Box> { - let address = [42; 32]; - let pull_request = PullRequest::new(address); - let bytes = pull_request.to_bytes(); - + pub fn new(provider_network_address: SocketAddrV4, our_address: DestinationAddressBytes, auth_token: Option) -> Self { // DH temporary: the provider's client port is not in the topology, but we can't change that // right now without messing up the existing Go mixnet. So I'm going to hardcode this // for the moment until the Go mixnet goes away. - let provider_socket = SocketAddrV4::new(*provider.ip(), 9000); - println!("Provider: {:?}", provider_socket); + let provider_socket = SocketAddrV4::new(*provider_network_address.ip(), 9000); - let mut socket = tokio::net::TcpStream::connect(provider_socket).await?; + ProviderClient { + provider_network_address: provider_socket, + our_address, + auth_token, + } + } + + pub fn update_token(&mut self, auth_token: AuthToken) { + self.auth_token = Some(auth_token) + } + + pub async fn send_request(&self, bytes: Vec) -> Result, ProviderClientError> { + let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?; println!("keep alive: {:?}", socket.keepalive()); socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); socket.write_all(&bytes[..]).await?; - if let Err(e) = socket.shutdown(Shutdown::Write) { - eprintln!("failed to close write part of the socket; err = {:?}", e) + if let Err(_e) = socket.shutdown(Shutdown::Write) { + // TODO: make it a silent log once we have a proper logging library +// eprintln!("failed to close write part of the socket; err = {:?}", e) } let mut response = Vec::new(); socket.read_to_end(&mut response).await?; - if let Err(e) = socket.shutdown(Shutdown::Read) { - eprintln!("failed to close read part of the socket; err = {:?}", e) + if let Err(_e) = socket.shutdown(Shutdown::Read) { + // TODO: make it a silent log once we have a proper logging library +// eprintln!("failed to close read part of the socket; err = {:?}", e) } - println!("Received the following response: {:?}", response); - let parsed_response = PullResponse::from_bytes(&response).unwrap(); - for message in parsed_response.messages { - println!("Received: {:?}", String::from_utf8(message).unwrap()) + Ok(response) + } + + pub async fn retrieve_messages( + &self, + ) -> Result>, ProviderClientError> { + if self.auth_token.is_none() { + return Err(ProviderClientError::EmptyAuthTokenError); } - Ok(()) + + let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap()); + let bytes = pull_request.to_bytes(); + + let response = self.send_request(bytes).await?; + println!("Received the following response: {:?}", response); + + let parsed_response = PullResponse::from_bytes(&response)?; + Ok(parsed_response.messages) + } + + pub async fn register(&self) -> Result { + if self.auth_token.is_some() { + return Err(ProviderClientError::ClientAlreadyRegisteredError); + } + + let register_request = RegisterRequest::new(self.our_address); + let bytes = register_request.to_bytes(); + + let response = self.send_request(bytes).await?; + let parsed_response = RegisterResponse::from_bytes(&response)?; + + Ok(parsed_response.auth_token) } } diff --git a/src/commands/run.rs b/src/commands/run.rs index 552a6469a6..932da4a190 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -1,16 +1,17 @@ use crate::banner; -use crate::clients::NymClient; -use crate::persistence::pemstore; +//use crate::clients::NymClient; +//use crate::persistence::pemstore; use clap::ArgMatches; -pub fn execute(matches: &ArgMatches) { +pub fn execute(_matches: &ArgMatches) { println!("{}", banner()); - - let is_local = matches.is_present("local"); - let id = matches.value_of("id").unwrap().to_string(); - println!("Starting client..."); - - let keypair = pemstore::read_keypair_from_disk(id); - let client = NymClient::new(keypair.public_bytes(), is_local); - client.start("127.0.0.1:9000".parse().unwrap()).unwrap(); + panic!("For time being this command is deprecated! Please use 'websocket' instead"); + // + // let is_local = matches.is_present("local"); + // let id = matches.value_of("id").unwrap().to_string(); + // println!("Starting client..."); + // + // let keypair = pemstore::read_keypair_from_disk(id); + // let client = NymClient::new(keypair.public_bytes(), is_local); + // client.start("127.0.0.1:9000".parse().unwrap()).unwrap(); } diff --git a/src/commands/tcpsocket.rs b/src/commands/tcpsocket.rs index 1e3a21f1d1..4ba652bcd6 100644 --- a/src/commands/tcpsocket.rs +++ b/src/commands/tcpsocket.rs @@ -26,7 +26,12 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let _client = NymClient::new(keypair.public_bytes(), is_local); + let _client = NymClient::new( + keypair.public_bytes(), + socket_address.clone(), + is_local, + None, + ); // Question: should we be passing the client into the TCP socket somehow next? tcp::start(socket_address); diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index d26ba5c988..36a88ea9ed 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -25,7 +25,9 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let client = NymClient::new(keypair.public_bytes(), is_local); + // TODO: reading auth_token from disk (if exists); + let auth_token = None; + let client = NymClient::new(keypair.public_bytes(), socket_address, is_local, auth_token); - client.start(socket_address).unwrap(); + client.start().unwrap(); }