Merge pull request #34 from nymtech/feature/client_registration

Feature/client registration

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<InputMessage>,
     // to be used by "send" function or socket, etc
     input_rx: mpsc::UnboundedReceiver<InputMessage>,
-
+    socket_listening_address: SocketAddr,
     is_local: bool,
+    auth_token: Option<AuthToken>
 }
 
 #[derive(Debug)]
 pub struct InputMessage(pub Destination, pub Vec<u8>);
 
 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<AuthToken>) -> Self {
         let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
 
         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<dyn std::error::Error>> {
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
         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(

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<io::Error> for ProviderClientError {
+    fn from(_: Error) -> Self {
+        use ProviderClientError::*;
+
+        NetworkError
+    }
+}
+
+impl From<ProviderResponseError> 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<AuthToken>,
+}
 
 impl ProviderClient {
-    pub fn new() -> Self {
-        ProviderClient {}
-    }
-
-    pub async fn retrieve_messages(
-        &self,
-        provider: SocketAddrV4,
-    ) -> Result<(), Box<dyn std::error::Error>> {
-        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<AuthToken>) -> 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<u8>) -> Result<Vec<u8>, 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<Vec<Vec<u8>>, 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<AuthToken, ProviderClientError> {
+        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)
     }
 }

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();
 }

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);

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();
 }