From 4d444ceb07af2d9ac888a4cb0c4d5e67caf0754a Mon Sep 17 00:00:00 2001 From: Dave Hrycyszyn Date: Thu, 13 Aug 2020 19:10:34 +0100 Subject: [PATCH] Starting to unify the slightly different interfaces of NymClient --- .../client-core/src/client/received_buffer.rs | 2 +- clients/native/src/client/mod.rs | 50 +++++++--- clients/socks5/src/client/mod.rs | 95 +++++++++++++------ clients/socks5/src/socks/server.rs | 16 ++-- 4 files changed, 112 insertions(+), 51 deletions(-) diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index db67cc6592..796680ad4a 100644 --- a/clients/client-core/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -150,7 +150,7 @@ impl ReceivedMessagesBuffer { if guard.message_sender.is_none() { // in theory we could just ignore it, but that situation should have never happened // in the first place, so this way we at least know we have an important bug to fix - panic!("trying to disconnect non-existent sender!") + // panic!("trying to disconnect non-existent sender!") // TODO: WTF is happening here? } guard.message_sender = None; } diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index a5be64e596..d7247b3140 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -44,6 +44,7 @@ use nymsphinx::addressing::clients::Recipient; use nymsphinx::addressing::nodes::NodeIdentity; use nymsphinx::anonymous_replies::ReplySURB; use nymsphinx::receiver::ReconstructedMessage; +use tokio::runtime; use tokio::runtime::Runtime; pub struct NymClient { @@ -258,18 +259,8 @@ impl NymClient { MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle()); } - fn start_websocket_listener( - &self, - buffer_requester: ReceivedBufferRequestSender, - msg_input: InputMessageSender, - ) { - info!("Starting websocket listener..."); - - let websocket_handler = - websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient()); - - websocket::Listener::new(self.config.get_listening_port()) - .start(self.runtime.handle(), websocket_handler); + fn start_listening(&self, listener: Box) { + listener.listen(self.config.get_listening_port(), self.runtime.handle()); } /// EXPERIMENTAL DIRECT RUST API @@ -380,10 +371,16 @@ impl NymClient { sphinx_message_sender.clone(), ); self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); + let recipient = self.as_mix_recipient(); + let websocket = Box::new(WebsocketListener::new( + received_buffer_request_sender.clone(), + input_sender.clone(), + recipient, + )); match self.config.get_socket_type() { SocketType::WebSocket => { - self.start_websocket_listener(received_buffer_request_sender, input_sender) + self.start_listening(websocket); } SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode @@ -417,3 +414,30 @@ impl NymClient { ); } } + +pub(crate) trait ClientListener { + fn listen(&self, port: u16, runtime_handle: &runtime::Handle); +} + +struct WebsocketListener { + handler: websocket::Handler, +} + +impl ClientListener for WebsocketListener { + fn listen(&self, port: u16, runtime_handle: &runtime::Handle) { + info!("Starting websocket listener..."); + websocket::Listener::new(port).start(runtime_handle, self.handler.clone()); + } +} + +impl WebsocketListener { + fn new( + received_buffer_request_sender: ReceivedBufferRequestSender, + input_sender: InputMessageSender, + recipient: Recipient, + ) -> WebsocketListener { + let handler = + websocket::Handler::new(input_sender, received_buffer_request_sender, recipient); + WebsocketListener { handler } + } +} diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 24378f1fd5..bca5a1e708 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -46,7 +46,7 @@ use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipi use nymsphinx::addressing::nodes::NodeIdentity; use nymsphinx::anonymous_replies::ReplySURB; use nymsphinx::receiver::ReconstructedMessage; -use tokio::runtime::Runtime; +use tokio::runtime::{self, Runtime}; pub struct NymClient { /// Client configuration options, including, among other things, packet sending rates, @@ -260,35 +260,9 @@ impl NymClient { MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle()); } - fn start_socks5_listener( - &self, - buffer_requester: ReceivedBufferRequestSender, - msg_input: InputMessageSender, - ) { + fn start_listening(&self, listener: Box) { info!("Starting socks5 listener..."); - let mut auth_methods: Vec = Vec::new(); - auth_methods.push(AuthenticationMethods::NoAuth as u8); - let allowed_users: Vec = Vec::new(); - - let authenticator = Authenticator::new(auth_methods, allowed_users); - let recipient = self.load_socks5_service_provider(); - let mut sphinx_socks = SphinxSocksServer::new(1080, "127.0.0.1", authenticator, recipient); - self.runtime - .spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await }); - } - // TODO: make this configurable in the client config file - // TODO: Talk to JS about where I can easily find these. - fn load_socks5_service_provider(&self) -> Recipient { - // load from file here, or better yet, inject it - let identity = "7qBXQor8nHXUXDAUM4aLvqDJ2MECFKn3AJ3brQsu5qz8"; - let encryption_key = "4uzn7m3vPEy5MhPHLqzFCJHk2BCnvbCsGdrYzng7jnor"; - let gateway_key = "e3vUAo6YhB7zq3GH8B4k3iiGT4H2USjdd5ZMZoUsHdF"; - - let client_identity = ClientIdentity::from_base58_string(identity).unwrap(); - let client_encryption_key = - ClientEncryptionKey::from_base58_string(encryption_key).unwrap(); - let gateway = NodeIdentity::from_base58_string(gateway_key).unwrap(); - Recipient::new(client_identity, client_encryption_key, gateway) + listener.listen(self.config.get_listening_port(), self.runtime.handle()); } /// EXPERIMENTAL DIRECT RUST API @@ -400,9 +374,17 @@ impl NymClient { ); self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); + let recipient = load_socks5_service_provider(); + let socks = Box::new(SocksListener::new( + received_buffer_request_sender.clone(), + input_sender.clone(), + recipient, + )); + // let socks = Box::new(listener); + match self.config.get_socket_type() { SocketType::WebSocket => { - self.start_socks5_listener(received_buffer_request_sender, input_sender) + self.start_listening(socks); } SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode @@ -436,3 +418,56 @@ impl NymClient { ); } } + +// TODO: make this configurable in the client config file +// TODO: Talk to JS about where I can easily find these. +fn load_socks5_service_provider() -> Recipient { + // load from file here, or better yet, inject it + let identity = "7qBXQor8nHXUXDAUM4aLvqDJ2MECFKn3AJ3brQsu5qz8"; + let encryption_key = "4uzn7m3vPEy5MhPHLqzFCJHk2BCnvbCsGdrYzng7jnor"; + let gateway_key = "e3vUAo6YhB7zq3GH8B4k3iiGT4H2USjdd5ZMZoUsHdF"; + + let client_identity = ClientIdentity::from_base58_string(identity).unwrap(); + let client_encryption_key = ClientEncryptionKey::from_base58_string(encryption_key).unwrap(); + let gateway = NodeIdentity::from_base58_string(gateway_key).unwrap(); + Recipient::new(client_identity, client_encryption_key, gateway) +} + +trait ClientListener { + fn listen(&mut self, port: u16, runtime_handle: &runtime::Handle); +} + +struct SocksListener { + server: SphinxSocksServer, +} + +impl ClientListener for SocksListener { + fn listen(&mut self, port: u16, runtime_handle: &runtime::Handle) { + info!("Starting socks listener..."); + + runtime_handle.spawn(async move { self.server.serve().await }); + } +} + +impl SocksListener { + fn new( + received_buffer_request_sender: ReceivedBufferRequestSender, + input_sender: InputMessageSender, + recipient: Recipient, + ) -> SocksListener { + let mut auth_methods: Vec = Vec::new(); + auth_methods.push(AuthenticationMethods::NoAuth as u8); + let allowed_users: Vec = Vec::new(); + + let authenticator = Authenticator::new(auth_methods, allowed_users); + let server = SphinxSocksServer::new( + 1080, + "127.0.0.1", + authenticator, + recipient, + input_sender, + received_buffer_request_sender, + ); + SocksListener { server } + } +} diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index 1118ad3cdd..3fb7fa20c1 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -18,6 +18,8 @@ pub struct SphinxSocksServer { authenticator: Authenticator, listening_address: SocketAddr, service_provider: Recipient, + input_sender: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, } impl SphinxSocksServer { @@ -27,22 +29,22 @@ impl SphinxSocksServer { ip: &str, authenticator: Authenticator, service_provider: Recipient, + input_sender: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, ) -> Self { info!("Listening on {}:{}", ip, port); SphinxSocksServer { authenticator, + buffer_requester, listening_address: format!("{}:{}", ip, port).parse().unwrap(), service_provider, + input_sender, } } /// Set up the listener and initiate connection handling when something /// connects to the server. - pub(crate) async fn serve( - &mut self, - input_sender: InputMessageSender, - buffer_requester: ReceivedBufferRequestSender, - ) -> Result<(), SocksProxyError> { + pub(crate) async fn serve(&mut self) -> Result<(), SocksProxyError> { info!("Serving Connections..."); let mut listener = TcpListener::bind(self.listening_address).await.unwrap(); @@ -50,7 +52,7 @@ impl SphinxSocksServer { let active_streams = Arc::new(Mutex::new(HashMap::new())); let mut mixnet_response_listener = - MixnetResponseListener::new(buffer_requester, Arc::clone(&active_streams)); + MixnetResponseListener::new(self.buffer_requester.clone(), Arc::clone(&active_streams)); tokio::spawn(async move { mixnet_response_listener.run().await; @@ -62,7 +64,7 @@ impl SphinxSocksServer { let mut client = SocksClient::new( stream, self.authenticator.clone(), - input_sender.clone(), + self.input_sender.clone(), self.service_provider.clone(), Arc::clone(&active_streams), );