diff --git a/clients/client-core/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs index 796680ad4a..db67cc6592 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!") // TODO: WTF is happening here? + panic!("trying to disconnect non-existent sender!") } guard.message_sender = None; } diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index d7247b3140..a5be64e596 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -44,7 +44,6 @@ 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 { @@ -259,8 +258,18 @@ impl NymClient { MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle()); } - fn start_listening(&self, listener: Box) { - listener.listen(self.config.get_listening_port(), 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); } /// EXPERIMENTAL DIRECT RUST API @@ -371,16 +380,10 @@ 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_listening(websocket); + self.start_websocket_listener(received_buffer_request_sender, input_sender) } SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode @@ -414,30 +417,3 @@ 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 bca5a1e708..24378f1fd5 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::{self, Runtime}; +use tokio::runtime::Runtime; pub struct NymClient { /// Client configuration options, including, among other things, packet sending rates, @@ -260,9 +260,35 @@ impl NymClient { MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle()); } - fn start_listening(&self, listener: Box) { + fn start_socks5_listener( + &self, + buffer_requester: ReceivedBufferRequestSender, + msg_input: InputMessageSender, + ) { info!("Starting socks5 listener..."); - listener.listen(self.config.get_listening_port(), self.runtime.handle()); + 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) } /// EXPERIMENTAL DIRECT RUST API @@ -374,17 +400,9 @@ 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_listening(socks); + self.start_socks5_listener(received_buffer_request_sender, input_sender) } SocketType::None => { // if we did not start the socket, it means we're running (supposedly) in the native mode @@ -418,56 +436,3 @@ 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 3fb7fa20c1..1118ad3cdd 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -18,8 +18,6 @@ pub struct SphinxSocksServer { authenticator: Authenticator, listening_address: SocketAddr, service_provider: Recipient, - input_sender: InputMessageSender, - buffer_requester: ReceivedBufferRequestSender, } impl SphinxSocksServer { @@ -29,22 +27,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) -> Result<(), SocksProxyError> { + pub(crate) async fn serve( + &mut self, + input_sender: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, + ) -> Result<(), SocksProxyError> { info!("Serving Connections..."); let mut listener = TcpListener::bind(self.listening_address).await.unwrap(); @@ -52,7 +50,7 @@ impl SphinxSocksServer { let active_streams = Arc::new(Mutex::new(HashMap::new())); let mut mixnet_response_listener = - MixnetResponseListener::new(self.buffer_requester.clone(), Arc::clone(&active_streams)); + MixnetResponseListener::new(buffer_requester, Arc::clone(&active_streams)); tokio::spawn(async move { mixnet_response_listener.run().await; @@ -64,7 +62,7 @@ impl SphinxSocksServer { let mut client = SocksClient::new( stream, self.authenticator.clone(), - self.input_sender.clone(), + input_sender.clone(), self.service_provider.clone(), Arc::clone(&active_streams), );