diff --git a/Cargo.lock b/Cargo.lock index 8ce58ed2d0..ec472a6b3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3022,12 +3022,16 @@ dependencies = [ "simple-socks5-requests", "tokio 0.2.16", "tokio-tungstenite", + "utils", "websocket-requests", ] [[package]] name = "simple-socks5-requests" version = "0.1.0" +dependencies = [ + "nymsphinx-addressing", +] [[package]] name = "siphasher" @@ -3157,6 +3161,7 @@ dependencies = [ "tokio-tungstenite", "topology", "url 2.1.1", + "utils", ] [[package]] @@ -3474,6 +3479,17 @@ dependencies = [ "tokio-reactor", ] +[[package]] +name = "tokio-test" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed0049c119b6d505c4447f5c64873636c7af6c75ab0d45fd9f618d82acb8016d" +dependencies = [ + "bytes 0.5.4", + "futures-core", + "tokio 0.2.16", +] + [[package]] name = "tokio-threadpool" version = "0.1.18" @@ -3746,6 +3762,15 @@ version = "0.7.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05e42f7c18b8f902290b009cde6d651262f956c98bc51bca4cd1d511c9cd85c7" +[[package]] +name = "utils" +version = "0.1.0" +dependencies = [ + "bytes 0.5.4", + "tokio 0.2.16", + "tokio-test", +] + [[package]] name = "uuid" version = "0.7.4" diff --git a/Cargo.toml b/Cargo.toml index 4e039a6ff9..9257dc53f7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,9 @@ panic = "abort" opt-level = "s" +[profile.dev] +panic = "abort" + [workspace] members = [ @@ -28,6 +31,7 @@ members = [ "common/nymsphinx/types", "common/pemstore", "common/topology", + "common/utils", "gateway", "gateway/gateway-requests", "service-providers/simple-socks5", diff --git a/clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs index ed25032438..e8a97e497f 100644 --- a/clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs @@ -67,11 +67,18 @@ where async fn on_retransmission_request(&mut self, frag_id: FragmentIdentifier) { let pending_acks_map_read_guard = self.pending_acks.read().await; - // if the unwrap failed here, we have some weird bug somewhere - honestly, I'm not sure - // if it's even possible for it to happen - let unreceived_ack_fragment = pending_acks_map_read_guard - .get(&frag_id) - .expect("wanted to retransmit ack'd fragment"); + + let unreceived_ack_fragment = match pending_acks_map_read_guard.get(&frag_id) { + Some(pending_ack) => pending_ack, + // this can actually happen when ack retransmission times out while `on_ack` is being processed + // 1. `retransmission_sender.unbounded_send(frag_id).unwrap()` happens thus triggering this function + // 2. at the same time ack is received and fully processed (which takes pending_acks *WRITE* lock!!) -> ack is removed from the map + `self.pending_acks.read()` blocks + // 3. `on_retransmission_request` manages to get read lock, but the entry was already removed + None => { + info!("wanted to retransmit ack'd fragment"); + return; + } + }; let packet_recipient = unreceived_ack_fragment.recipient.clone(); let chunk_clone = unreceived_ack_fragment.message_chunk.clone(); @@ -101,22 +108,30 @@ where // waiting for the write lock on `pending_acks` drop(topology_permit); - self.pending_acks - .write() - .await - .get_mut(&frag_id) - .expect( - "on_retransmission_request: somehow we already received an ack for this packet?", - ) - .update_delay(prepared_fragment.total_delay); + // for this to actually return a None, the following sequence of events needs to happen: + // 0. recall that up until this point we're holding a READ lock, so nobody else can WRITE + // 1. `on_retransmission_request` is called - processing takes a while (we need to create SPHINX packet, etc.) + // 2. at the same time we receive DELAYED (i.e. post timeout) ack for the packet we are about to retransmit + // 3. the procedure to remove the pending ack waits for the WRITE lock and acquires it in the tiny window + // between when READ lock is dropped and WRITE lock is reacquired in this method + // 4. the pending ack is removed and when WRITE lock is acquired here, `None` is returned - self.real_message_sender - .unbounded_send(RealMessage::new( - prepared_fragment.first_hop_address, - prepared_fragment.sphinx_packet, - frag_id, - )) - .unwrap(); + // TODO: benchmark whether it wouldn't be potentially more efficient to acquire WRITE lock at the very beginning of the method + // one major drawback: nobody else could READ while we're preparing two sphinx packets, encrypting data, etc. + if let Some(pending_ack) = self.pending_acks.write().await.get_mut(&frag_id) { + pending_ack.update_delay(prepared_fragment.total_delay); + + self.real_message_sender + .unbounded_send(RealMessage::new( + prepared_fragment.first_hop_address, + prepared_fragment.sphinx_packet, + frag_id, + )) + .unwrap(); + } else { + // later on we will want this to be decreased to 'debug' (or maybe not?) + info!("received an ack after timeout, but before retransmission went through") + } } pub(super) async fn run(&mut self) { diff --git a/clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs index 30efb7cc81..23d15f6ff6 100644 --- a/clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs +++ b/clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs @@ -65,9 +65,13 @@ impl SentNotificationListener { // load that `on_sent_message()` is not called (and we do not receive the read permit) // until we already received and processed an ack for the packet // but this seems extremely unrealistic, but perhaps we should guard against that? - let pending_ack_data = pending_acks_map_read_guard - .get(&frag_id) - .expect("on_sent_message: somehow we already received an ack for this packet?"); + let pending_ack_data = match pending_acks_map_read_guard.get(&frag_id) { + Some(pending_ack) => pending_ack, + None => { + info!("on_sent_message: somehow we already received an ack for this packet?"); + return; + } + }; // if this assertion ever fails, we have some bug due to some unintended leak. // the only reason I see it could happen if the `tokio::select` in the spawned diff --git a/clients/client-core/src/config/mod.rs b/clients/client-core/src/config/mod.rs index 4ccb1606f5..3a0a9b2e43 100644 --- a/clients/client-core/src/config/mod.rs +++ b/clients/client-core/src/config/mod.rs @@ -28,10 +28,14 @@ const DEFAULT_DIRECTORY_SERVER: &str = "https://directory.nymtech.net"; // where applicable, the below are defined in milliseconds const DEFAULT_ACK_WAIT_MULTIPLIER: f64 = 1.5; const DEFAULT_ACK_WAIT_ADDITION: u64 = 800; -const DEFAULT_LOOP_COVER_STREAM_AVERAGE_DELAY: u64 = 1000; // 1s -const DEFAULT_MESSAGE_STREAM_AVERAGE_DELAY: u64 = 500; // 0.5s -const DEFAULT_AVERAGE_PACKET_DELAY: u64 = 200; // 0.2s -const DEFAULT_TOPOLOGY_REFRESH_RATE: u64 = 30_000; // 30s +const DEFAULT_LOOP_COVER_STREAM_AVERAGE_DELAY: u64 = 1000; +// 1s +const DEFAULT_MESSAGE_STREAM_AVERAGE_DELAY: u64 = 500; +// 0.5s +const DEFAULT_AVERAGE_PACKET_DELAY: u64 = 200; +// 0.2s +const DEFAULT_TOPOLOGY_REFRESH_RATE: u64 = 30_000; +// 30s const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: u64 = 5_000; // 5s const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: u64 = 1_500; // 1.5s diff --git a/clients/socks5/Cargo.toml b/clients/socks5/Cargo.toml index 808ebcf096..afeb744a0d 100644 --- a/clients/socks5/Cargo.toml +++ b/clients/socks5/Cargo.toml @@ -39,6 +39,7 @@ nymsphinx = { path = "../../common/nymsphinx" } pemstore = { path = "../../common/pemstore" } simple-socks5-requests = { path = "../../service-providers/simple-socks5/simple-socks5-requests" } topology = { path = "../../common/topology" } +utils = { path = "../../common/utils" } [build-dependencies] built = "0.3.2" diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 24378f1fd5..1b5d1bedc4 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -272,7 +272,13 @@ impl NymClient { 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); + let mut sphinx_socks = SphinxSocksServer::new( + 1080, + "127.0.0.1", + authenticator, + recipient, + self.as_mix_recipient(), + ); self.runtime .spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await }); } @@ -280,9 +286,9 @@ impl NymClient { // 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 identity = "G3WSu7S3Jdm5HHQwZ3bT7NKoEBNSGr8JY2jngbUigqfz"; + let encryption_key = "6KUoz9gexdFziLrrh6M2gXpAW4Fdn3HBHss11WJDYKEL"; + let gateway_key = "5QAR66H9aMqaEo9y9G4hxAsvenJe13wJSRkwQ8qjsF6C"; let client_identity = ClientIdentity::from_base58_string(identity).unwrap(); let client_encryption_key = diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index 44acb25777..73b1c5bd08 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -32,6 +32,7 @@ pub(crate) struct SocksClient { input_sender: InputMessageSender, connection_id: ConnectionId, service_provider: Recipient, + self_address: Recipient, } type StreamResponseSender = oneshot::Sender>; @@ -52,6 +53,7 @@ impl SocksClient { input_sender: InputMessageSender, service_provider: Recipient, active_streams: ActiveStreams, + self_address: Recipient, ) -> Self { let connection_id = Self::generate_random(); SocksClient { @@ -63,6 +65,7 @@ impl SocksClient { authenticator, input_sender, service_provider, + self_address, } } @@ -138,6 +141,7 @@ impl SocksClient { let request = SocksRequest::from_stream(&mut self.stream).await?; let remote_address = request.to_string(); + let client_address = self.stream.peer_addr().unwrap().to_string(); match request.command { // Use the Proxy to connect to the specified addr/port SocksCommand::Connect => { @@ -145,11 +149,12 @@ impl SocksClient { self.acknowledge_socks5().await; let request_data_bytes = - SocksRequest::try_read_request_data(&mut self.stream).await?; + SocksRequest::try_read_request_data(&mut self.stream, &client_address).await?; let socks_provider_request = Request::new_connect( self.connection_id, remote_address.clone(), request_data_bytes, + self.self_address.clone(), ); let response_data = self .send_request_to_mixnet_and_get_response(socks_provider_request) @@ -158,7 +163,7 @@ impl SocksClient { loop { if let Ok(request_data_bytes) = - SocksRequest::try_read_request_data(&mut self.stream).await + SocksRequest::try_read_request_data(&mut self.stream, &client_address).await { if request_data_bytes.is_empty() { break; @@ -168,7 +173,13 @@ impl SocksClient { let response_data = self .send_request_to_mixnet_and_get_response(socks_provider_request) .await; - self.stream.write_all(&response_data).await.unwrap(); + if let Err(err) = self.stream.write_all(&response_data).await { + error!( + "tried to write to (presumably) closed connection - {:?}", + err + ); + break; + } } else { break; } diff --git a/clients/socks5/src/socks/request.rs b/clients/socks5/src/socks/request.rs index 7ff1dd11fe..3e81097af1 100644 --- a/clients/socks5/src/socks/request.rs +++ b/clients/socks5/src/socks/request.rs @@ -1,9 +1,10 @@ use super::types::{AddrType, ResponseCode, SocksProxyError}; -use super::{utils, SOCKS_VERSION}; +use super::{utils as socks_utils, SOCKS_VERSION}; use log::*; use std::net::SocketAddr; use tokio::net::TcpStream; use tokio::prelude::*; +use utils::read_delay_loop::try_read_data; /// A Socks5 request hitting the proxy. pub(crate) struct SocksRequest { @@ -100,44 +101,23 @@ impl SocksRequest { /// Print out the address and port to a String. /// This might return domain:port, ipv6:port, or ipv4:port. pub(crate) fn to_string(&self) -> String { - let address = utils::pretty_print_addr(&self.addr_type, &self.addr); + let address = socks_utils::pretty_print_addr(&self.addr_type, &self.addr); format!("{}:{}", address, self.port) } /// Convert the request object to a SocketAddr pub(crate) fn to_socket(&self) -> Result, SocksProxyError> { - utils::addr_to_socket(&self.addr_type, &self.addr, self.port) + socks_utils::addr_to_socket(&self.addr_type, &self.addr, self.port) } /// Attempts to read data from the Socks5 request stream. Times out and /// returns what it's got if no data is read for the timeout_duration pub(crate) async fn try_read_request_data( reader: &mut R, + remote_address: &str, ) -> io::Result> { let timeout_duration = std::time::Duration::from_millis(500); - let mut data = Vec::new(); - let mut timeout = tokio::time::delay_for(timeout_duration); - loop { - let mut buf = [0u8; 1024]; - tokio::select! { - _ = &mut timeout => { - println!("we timed out!"); - return Ok(data) - } - read_data = reader.read(&mut buf) => { - match read_data { - Err(err) => return Err(err), - Ok(0) => return Ok(data), - Ok(n) => { - let now = timeout.deadline(); - let next = now + timeout_duration; - timeout.reset(next); - data.extend_from_slice(&buf[..n]) - } - } - } - } - } + try_read_data(timeout_duration, reader, remote_address).await } } diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index 1118ad3cdd..44adb70d96 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -18,6 +18,7 @@ pub struct SphinxSocksServer { authenticator: Authenticator, listening_address: SocketAddr, service_provider: Recipient, + self_address: Recipient, } impl SphinxSocksServer { @@ -27,12 +28,14 @@ impl SphinxSocksServer { ip: &str, authenticator: Authenticator, service_provider: Recipient, + self_address: Recipient, ) -> Self { info!("Listening on {}:{}", ip, port); SphinxSocksServer { authenticator, listening_address: format!("{}:{}", ip, port).parse().unwrap(), service_provider, + self_address, } } @@ -65,6 +68,7 @@ impl SphinxSocksServer { input_sender.clone(), self.service_provider.clone(), Arc::clone(&active_streams), + self.self_address.clone(), ); tokio::spawn(async move { diff --git a/common/crypto/src/asymmetric/encryption/mod.rs b/common/crypto/src/asymmetric/encryption/mod.rs index 55295d5902..b321cda538 100644 --- a/common/crypto/src/asymmetric/encryption/mod.rs +++ b/common/crypto/src/asymmetric/encryption/mod.rs @@ -25,7 +25,7 @@ pub const PUBLIC_KEY_SIZE: usize = 32; /// Size of a X25519 shared secret pub const SHARED_SECRET_SIZE: usize = 32; -#[derive(Debug)] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug)] pub enum EncryptionKeyError { InvalidPublicKey, InvalidPrivateKey, diff --git a/common/nymsphinx/addressing/src/clients.rs b/common/nymsphinx/addressing/src/clients.rs index 1de9964c5d..82d86e8a62 100644 --- a/common/nymsphinx/addressing/src/clients.rs +++ b/common/nymsphinx/addressing/src/clients.rs @@ -19,7 +19,7 @@ const CLIENT_ENCRYPTION_KEY_SIZE: usize = encryption::PUBLIC_KEY_SIZE; pub type ClientIdentity = identity::PublicKey; const CLIENT_IDENTITY_SIZE: usize = identity::PUBLIC_KEY_LENGTH; -#[derive(Debug)] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug)] pub enum RecipientFormattingError { MalformedRecipientError, MalformedIdentityError(identity::SignatureError), diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index c1c06f0ba5..89ecb46e63 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -33,8 +33,10 @@ pub struct InvalidPacketSize; #[repr(u8)] #[derive(Clone, Copy, Debug)] pub enum PacketSize { - RegularPacket = 1, // for example instant messaging use case - ACKPacket = 2, // for sending SURB-ACKs + RegularPacket = 1, + // for example instant messaging use case + ACKPacket = 2, + // for sending SURB-ACKs ExtendedPacket = 3, // for example for streaming fast and furious in uncompressed 10bit 4K HDR quality } diff --git a/common/utils/Cargo.toml b/common/utils/Cargo.toml new file mode 100644 index 0000000000..3b5c7ea2d6 --- /dev/null +++ b/common/utils/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "utils" +version = "0.1.0" +authors = ["Jedrzej Stuczynski "] +edition = "2018" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +bytes = "0.5" +# no need for any features as `AsyncRead` is always available +tokio = { version = "0.2", features = [] } + +[dev-dependencies] +tokio = { version = "0.2", features = ["rt-threaded", "macros"] } +tokio-test = "0.2" \ No newline at end of file diff --git a/common/utils/src/available_reader.rs b/common/utils/src/available_reader.rs new file mode 100644 index 0000000000..630f970b37 --- /dev/null +++ b/common/utils/src/available_reader.rs @@ -0,0 +1,140 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use bytes::{BufMut, Bytes, BytesMut}; +use std::cell::RefCell; +use std::future::Future; +use std::io; +use std::ops::DerefMut; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::AsyncRead; + +pub struct AvailableReader<'a, R: AsyncRead + Unpin> { + // TODO: come up with a way to avoid using RefCell (not sure if possible though) + buf: RefCell, + inner: RefCell<&'a mut R>, +} + +impl<'a, R> AvailableReader<'a, R> +where + R: AsyncRead + Unpin, +{ + const BUF_INCREMENT: usize = 4096; + + pub fn new(reader: &'a mut R) -> Self { + AvailableReader { + buf: RefCell::new(BytesMut::with_capacity(Self::BUF_INCREMENT)), + inner: RefCell::new(reader), + } + } +} + +impl<'a, R: AsyncRead + Unpin> Future for AvailableReader<'a, R> { + type Output = io::Result; + + // this SHOULD stay mutable, because we rely on runtime checks inside the method + #[allow(unused_mut)] + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + // if we have no space in buffer left - expand it + if !self.buf.borrow().has_remaining_mut() { + self.buf.borrow_mut().reserve(Self::BUF_INCREMENT); + } + + // note: poll_read_buf calls `buf.advance_mut(n)` + let poll_res = Pin::new(self.inner.borrow_mut().deref_mut()) + .poll_read_buf(cx, self.buf.borrow_mut().deref_mut()); + + match poll_res { + Poll::Pending => { + // there's nothing for us here, just return whatever we have (assuming we read anything!) + if self.buf.borrow().is_empty() { + Poll::Pending + } else { + let buf = self.buf.replace(BytesMut::new()); + Poll::Ready(Ok(buf.freeze())) + } + } + Poll::Ready(Err(err)) => Poll::Ready(Err(err)), + Poll::Ready(Ok(n)) => { + // if we read a non-0 amount, we're not done yet! + if n == 0 { + let buf = self.buf.replace(BytesMut::new()); + Poll::Ready(Ok(buf.freeze())) + } else { + // tell the waker we should be polled again! + cx.waker().wake_by_ref(); + Poll::Pending + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + use std::time::Duration; + + #[tokio::test] + async fn available_reader_reads_all_available_data_smaller_than_its_buf() { + let data = vec![42u8; 100]; + let mut reader = Cursor::new(data.clone()); + + let available_reader = AvailableReader::new(&mut reader); + + assert_eq!(available_reader.await.unwrap(), data) + } + + #[tokio::test] + async fn available_reader_reads_all_available_data_bigger_than_its_buf() { + let data = vec![42u8; AvailableReader::>>::BUF_INCREMENT + 100]; + let mut reader = Cursor::new(data.clone()); + + let available_reader = AvailableReader::new(&mut reader); + + assert_eq!(available_reader.await.unwrap(), data) + } + + #[tokio::test] + async fn available_reader_will_not_wait_for_more_data_if_it_already_has_some() { + let first_data_chunk = vec![42u8; 100]; + let second_data_chunk = vec![123u8; 100]; + + let mut reader_mock = tokio_test::io::Builder::new() + .read(&first_data_chunk) + .wait(Duration::from_millis(100)) // delay is irrelevant, what matters is that we don't get everything immediately + .read(&second_data_chunk) + .build(); + + let available_reader = AvailableReader::new(&mut reader_mock); + + assert_eq!(available_reader.await.unwrap(), first_data_chunk); + } + + #[tokio::test] + async fn available_reader_will_wait_for_more_data_if_it_doesnt_have_anything() { + let data = vec![42u8; 100]; + + let mut reader_mock = tokio_test::io::Builder::new() + .wait(Duration::from_millis(100)) + .read(&data) + .build(); + + let available_reader = AvailableReader::new(&mut reader_mock); + + assert_eq!(available_reader.await.unwrap(), data); + } +} diff --git a/common/utils/src/lib.rs b/common/utils/src/lib.rs new file mode 100644 index 0000000000..15f5844db7 --- /dev/null +++ b/common/utils/src/lib.rs @@ -0,0 +1,16 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +pub mod available_reader; +pub mod read_delay_loop; diff --git a/common/utils/src/read_delay_loop.rs b/common/utils/src/read_delay_loop.rs new file mode 100644 index 0000000000..020f3ba82c --- /dev/null +++ b/common/utils/src/read_delay_loop.rs @@ -0,0 +1,70 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// The only reason this exists is to remove duplicate code from +// nym\service-providers\simple-socks5\src\connection.rs::try_read_response_data +// and +// nym\clients\socks5\src\socks\request.rs::try_read_request_data + +// once those use sequence numbers, this code should be removed!! + +use crate::available_reader::AvailableReader; +use std::io; +use tokio::io::AsyncRead; +use tokio::time::Duration; + +pub async fn try_read_data( + timeout: Duration, + mut reader: R, + address: &str, +) -> io::Result> +where + R: AsyncRead + Unpin, +{ + let mut data = Vec::new(); + let mut delay = tokio::time::delay_for(timeout); + + let mut available_reader = AvailableReader::new(&mut reader); + + loop { + tokio::select! { + _ = &mut delay => { + println!("Timed out. returning {} bytes received from {}", data.len(), address); + return Ok(data) // we return all response data on timeout + } + read_data = &mut available_reader => { + match read_data { + Err(err) => { + return Err(err); + } + Ok(bytes) => { + if bytes.len() == 0 { + println!("Connection is closed! Returning {} bytes received from {}", data.len(), address); + // we return all we managed to read because + // we know no more stuff is coming + return Ok(data) + } + let now = tokio::time::Instant::now(); + let next = now + timeout; + delay.reset(next); + println!("Received {} bytes from {}. Waiting for more...", bytes.len(), address); + + // temporarily this is fine... (this loop will go away anyway) + data.extend_from_slice(&bytes) + } + } + } + } + } +} diff --git a/scripts/start_local_network.sh b/scripts/start_local_network.sh index 3e4929af0c..288256f19d 100755 --- a/scripts/start_local_network.sh +++ b/scripts/start_local_network.sh @@ -38,7 +38,7 @@ export RUST_LOG=warning # NOTE: If we wanted to suppress stdout and stderr, replace `&` with `> /dev/null 2>&1 &` in the `run` # cargo run --bin nym-gateway -- init --id gateway-local --mix-host 127.0.0.1:10000 --clients-host 127.0.0.1:10001 --directory $DIR -cargo run --bin nym-gateway -- run --id gateway-local & +cargo run --release --bin nym-gateway -- run --id gateway-local & sleep 1 @@ -46,8 +46,8 @@ sleep 1 # Will make it later either configurable by flags or config file. for (( j=0; j<$NUMMIXES; j++ )); do let layer=j%MAX_LAYERS+1 - cargo run --bin nym-mixnode -- init --id mix-local$j --host 127.0.0.1 --port $((9980+$j)) --layer $layer --directory $DIR - cargo run --bin nym-mixnode -- run --id mix-local$j & + cargo run --release --bin nym-mixnode -- init --id mix-local$j --host 127.0.0.1 --port $((9980+$j)) --layer $layer --directory $DIR + cargo run --release --bin nym-mixnode -- run --id mix-local$j & sleep 1 done diff --git a/service-providers/simple-socks5/Cargo.toml b/service-providers/simple-socks5/Cargo.toml index e47c96540b..079fc9d2d3 100644 --- a/service-providers/simple-socks5/Cargo.toml +++ b/service-providers/simple-socks5/Cargo.toml @@ -10,7 +10,7 @@ edition = "2018" bs58 = "0.3.0" clap = "2.33.0" futures = "0.3" -futures-util = { version = "0.3", default-features = false, features = ["async-await", "sink", "std"] } +futures-util = { version = "0.3", default-features = false, features = ["async-await", "sink", "std"] } # why ? log = "0.4" pretty_env_logger = "0.3" tokio = { version = "0.2", features = ["full"] } @@ -18,4 +18,5 @@ tokio-tungstenite = "0.11.0" nymsphinx = { path = "../../common/nymsphinx" } simple-socks5-requests = { path = "simple-socks5-requests" } -websocket-requests = { path = "../../clients/native/websocket-requests" } \ No newline at end of file +utils = { path = "../../common/utils" } +websocket-requests = { path = "../../clients/native/websocket-requests" } diff --git a/service-providers/simple-socks5/simple-socks5-requests/Cargo.toml b/service-providers/simple-socks5/simple-socks5-requests/Cargo.toml index bb416f5d42..d55f9cb1f8 100644 --- a/service-providers/simple-socks5/simple-socks5-requests/Cargo.toml +++ b/service-providers/simple-socks5/simple-socks5-requests/Cargo.toml @@ -7,3 +7,4 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +nymsphinx-addressing = { path = "../../../common/nymsphinx/addressing" } diff --git a/service-providers/simple-socks5/simple-socks5-requests/src/request.rs b/service-providers/simple-socks5/simple-socks5-requests/src/request.rs index 007ccd4839..43df046dad 100644 --- a/service-providers/simple-socks5/simple-socks5-requests/src/request.rs +++ b/service-providers/simple-socks5/simple-socks5-requests/src/request.rs @@ -1,3 +1,4 @@ +use nymsphinx_addressing::clients::{Recipient, RecipientFormattingError}; use std::convert::TryFrom; pub type ConnectionId = u64; @@ -18,6 +19,17 @@ pub enum RequestError { ConnectionIdTooShort, NoData, UnknownRequestFlag, + ReturnAddressTooShort, + MalformedReturnAddress(RecipientFormattingError), +} + +impl RequestError { + pub fn is_malformed_return(&self) -> bool { + match self { + RequestError::MalformedReturnAddress(_) => true, + _ => false, + } + } } impl TryFrom for RequestFlag { @@ -39,7 +51,13 @@ impl TryFrom for RequestFlag { pub enum Request { /// Start a new TCP connection to the specified `RemoteAddress` and send /// the request data up the connection. - Connect(ConnectionId, RemoteAddress, Vec), + /// All responses produced on this `ConnectionId` should come back to the specified `Recipient` + Connect { + conn_id: ConnectionId, + remote_addr: RemoteAddress, + data: Vec, + return_address: Recipient, + }, /// Re-use an existing TCP connection, sending more request data up it. Send(ConnectionId, Vec), @@ -54,8 +72,14 @@ impl Request { conn_id: ConnectionId, remote_addr: RemoteAddress, data: Vec, + return_address: Recipient, ) -> Request { - Request::Connect(conn_id, remote_addr, data) + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } } /// Construct a new Request::Send instance @@ -113,12 +137,23 @@ impl Request { let address_bytes = &connect_request_bytes[address_start..address_end]; let remote_address = String::from_utf8_lossy(&address_bytes).to_string(); - let request_data = &connect_request_bytes[address_end..]; - Ok(Request::Connect( - connection_id, - remote_address, - request_data.to_vec(), - )) + // just a temporary reference to mid-slice for ease of use + let recipient_data_bytes = &connect_request_bytes[address_end..]; + if recipient_data_bytes.len() < Recipient::LEN { + return Err(RequestError::ReturnAddressTooShort); + } + + let mut return_bytes = [0u8; Recipient::LEN]; + return_bytes.copy_from_slice(&recipient_data_bytes[..Recipient::LEN]); + let return_address = Recipient::try_from_bytes(return_bytes) + .map_err(|err| RequestError::MalformedReturnAddress(err))?; + + Ok(Request::Connect { + conn_id: connection_id, + remote_addr: remote_address, + data: recipient_data_bytes[Recipient::LEN..].to_vec(), + return_address, + }) } RequestFlag::Send => Ok(Request::Send(connection_id, b[9..].as_ref().to_vec())), RequestFlag::Close => Ok(Request::Close(connection_id)), @@ -130,13 +165,21 @@ impl Request { /// service provider which will make the request. pub fn into_bytes(self) -> Vec { match self { - Request::Connect(conn_id, remote_address, data) => { - let remote_address_bytes = remote_address.into_bytes(); + // connect is: CONN_FLAG || CONN_ID || REMOTE_LEN || REMOTE || RETURN || DATA + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } => { + let remote_address_bytes = remote_addr.into_bytes(); let remote_address_bytes_len = remote_address_bytes.len() as u16; + std::iter::once(RequestFlag::Connect as u8) .chain(conn_id.to_be_bytes().iter().cloned()) .chain(remote_address_bytes_len.to_be_bytes().iter().cloned()) .chain(remote_address_bytes.into_iter()) + .chain(return_address.to_bytes().iter().cloned()) .chain(data.into_iter()) .collect() } @@ -206,6 +249,87 @@ mod request_deserialization_tests { ); } + #[test] + fn returns_error_for_when_return_address_is_too_short() { + // this one has "foo.com" remote address and correct 8 bytes of connection_id + let request_bytes_prefix = [ + RequestFlag::Connect as u8, + 1, + 2, + 3, + 4, + 5, + 6, + 7, + 8, + 0, + 7, + 102, + 111, + 111, + 46, + 99, + 111, + 109, + ]; + + let recipient = Recipient::try_from_base58_string("CytBseW6yFXUMzz4SGAKdNLGR7q3sJLLYxyBGvutNEQV.4QXYyEVc5fUDjmmi8PrHN9tdUFV4PCvSJE1278cHyvoe@4sBbL1ngf1vtNqykydQKTFh26sQCw888GpUqvPvyNB4f").unwrap(); + let recipient_bytes = recipient.to_bytes(); + + // take only part of actual recipient + let request_bytes: Vec<_> = request_bytes_prefix + .iter() + .cloned() + .chain(recipient_bytes.iter().take(40).cloned()) + .collect(); + assert_eq!( + RequestError::ReturnAddressTooShort, + Request::try_from_bytes(&request_bytes).unwrap_err() + ); + } + + #[test] + fn returns_error_for_when_return_address_is_malformed() { + // this one has "foo.com" remote address and correct 8 bytes of connection_id + let request_bytes_prefix = [ + RequestFlag::Connect as u8, + 1, + 2, + 3, + 4, + 5, + 6, + 7, + 8, + 0, + 7, + 102, + 111, + 111, + 46, + 99, + 111, + 109, + ]; + + let recipient = Recipient::try_from_base58_string("CytBseW6yFXUMzz4SGAKdNLGR7q3sJLLYxyBGvutNEQV.4QXYyEVc5fUDjmmi8PrHN9tdUFV4PCvSJE1278cHyvoe@4sBbL1ngf1vtNqykydQKTFh26sQCw888GpUqvPvyNB4f").unwrap(); + let mut recipient_bytes = recipient.to_bytes(); + + // mess up few bytes + recipient_bytes[0] = 255; + recipient_bytes[15] ^= 1; + recipient_bytes[31] ^= 1; + + let request_bytes: Vec<_> = request_bytes_prefix + .iter() + .cloned() + .chain(recipient_bytes.iter().cloned()) + .collect(); + assert!(Request::try_from_bytes(&request_bytes) + .unwrap_err() + .is_malformed_return()); + } + #[test] fn works_when_request_is_sized_properly_even_without_data() { // this one has "foo.com" remote address, correct 8 bytes of connection_id, and 0 bytes request data @@ -230,11 +354,29 @@ mod request_deserialization_tests { 109, ] .to_vec(); + + let recipient = Recipient::try_from_base58_string("CytBseW6yFXUMzz4SGAKdNLGR7q3sJLLYxyBGvutNEQV.4QXYyEVc5fUDjmmi8PrHN9tdUFV4PCvSJE1278cHyvoe@4sBbL1ngf1vtNqykydQKTFh26sQCw888GpUqvPvyNB4f").unwrap(); + let recipient_bytes = recipient.to_bytes(); + + let request_bytes: Vec<_> = request_bytes + .into_iter() + .chain(recipient_bytes.iter().cloned()) + .collect(); + let request = Request::try_from_bytes(&request_bytes).unwrap(); match request { - Request::Connect(conn_id, remote_address, data) => { - assert_eq!("foo.com".to_string(), remote_address); + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } => { + assert_eq!("foo.com".to_string(), remote_addr); assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); + assert_eq!( + return_address.to_bytes().to_vec(), + recipient.to_bytes().to_vec() + ); assert_eq!(Vec::::new(), data); } _ => unreachable!(), @@ -243,7 +385,7 @@ mod request_deserialization_tests { #[test] fn works_when_request_is_sized_properly_and_has_data() { - // this one has a 1-byte remote address, correct 16 bytes of connection_id, and 3 bytes request data + // this one has a 1-byte remote address, correct 8 bytes of connection_id, and 3 bytes request data let request_bytes = [ RequestFlag::Connect as u8, 1, @@ -263,17 +405,32 @@ mod request_deserialization_tests { 99, 111, 109, - 255, - 255, - 255, ] .to_vec(); + let recipient = Recipient::try_from_base58_string("CytBseW6yFXUMzz4SGAKdNLGR7q3sJLLYxyBGvutNEQV.4QXYyEVc5fUDjmmi8PrHN9tdUFV4PCvSJE1278cHyvoe@4sBbL1ngf1vtNqykydQKTFh26sQCw888GpUqvPvyNB4f").unwrap(); + let recipient_bytes = recipient.to_bytes(); + + let request_bytes: Vec<_> = request_bytes + .into_iter() + .chain(recipient_bytes.iter().cloned()) + .chain(vec![255, 255, 255].into_iter()) + .collect(); + let request = Request::try_from_bytes(&request_bytes).unwrap(); match request { - Request::Connect(conn_id, remote_address, data) => { - assert_eq!("foo.com".to_string(), remote_address); + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } => { + assert_eq!("foo.com".to_string(), remote_addr); assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); + assert_eq!( + return_address.to_bytes().to_vec(), + recipient.to_bytes().to_vec() + ); assert_eq!(vec![255, 255, 255], data); } _ => unreachable!(), diff --git a/service-providers/simple-socks5/src/connection.rs b/service-providers/simple-socks5/src/connection.rs index 6a0bf31e01..f6c53ddc52 100644 --- a/service-providers/simple-socks5/src/connection.rs +++ b/service-providers/simple-socks5/src/connection.rs @@ -1,6 +1,8 @@ +use nymsphinx::addressing::clients::Recipient; use simple_socks5_requests::{ConnectionId, RemoteAddress}; use tokio::net::TcpStream; use tokio::prelude::*; +use utils::read_delay_loop::try_read_data; /// A TCP connection between the Socks5 service provider, which makes /// outbound requests on behalf of users and returns the responses through @@ -10,6 +12,7 @@ pub(crate) struct Connection { id: ConnectionId, address: RemoteAddress, conn: TcpStream, + return_address: Recipient, } impl Connection { @@ -17,14 +20,31 @@ impl Connection { id: ConnectionId, address: RemoteAddress, initial_data: &[u8], + return_address: Recipient, ) -> io::Result { - let conn = TcpStream::connect(&address).await?; - let mut connection = Connection { id, address, conn }; + let conn = match TcpStream::connect(&address).await { + Ok(conn) => conn, + Err(err) => { + eprintln!("error while connecting to {:?} ! - {:?}", address, err); + return Err(err); + } + }; + let mut connection = Connection { + id, + address, + conn, + return_address, + }; connection.send_data(&initial_data).await?; Ok(connection) } + pub(crate) fn return_address(&self) -> Recipient { + self.return_address.clone() + } + pub(crate) async fn send_data(&mut self, data: &[u8]) -> io::Result<()> { + println!("Sending {} bytes to {}", data.len(), self.address); self.conn.write_all(&data).await } @@ -32,27 +52,6 @@ impl Connection { /// remote server. Returns once it times out or the connection closes. pub(crate) async fn try_read_response_data(&mut self) -> io::Result> { let timeout_duration = std::time::Duration::from_millis(500); - let mut data = Vec::new(); - let mut timeout = tokio::time::delay_for(timeout_duration); - loop { - let mut buf = [0u8; 1024]; - tokio::select! { - _ = &mut timeout => { - return Ok(data) // we return all response data on timeout - } - read_data = self.conn.read(&mut buf) => { - match read_data { - Err(err) => return Err(err), - Ok(0) => return Ok(data), - Ok(n) => { - let now = timeout.deadline(); - let next = now + timeout_duration; - timeout.reset(next); - data.extend_from_slice(&buf[..n]) - } - } - } - } - } + try_read_data(timeout_duration, &mut self.conn, &self.address).await } } diff --git a/service-providers/simple-socks5/src/controller.rs b/service-providers/simple-socks5/src/controller.rs index 2f98aa5e97..cb791bf563 100644 --- a/service-providers/simple-socks5/src/controller.rs +++ b/service-providers/simple-socks5/src/controller.rs @@ -1,7 +1,10 @@ use crate::connection::Connection; +use futures::lock::Mutex; +use nymsphinx::addressing::clients::Recipient; use simple_socks5_requests::{ConnectionId, RemoteAddress, Request, Response}; use std::collections::HashMap; use std::io; +use std::sync::Arc; #[derive(Debug)] pub enum ConnectionError { @@ -15,49 +18,66 @@ impl From for ConnectionError { } } +// clone is fine here because HashMap is never cloned, the pointer is +#[derive(Clone)] pub(crate) struct Controller { - open_connections: HashMap, + open_connections: Arc>>, } impl Controller { pub(crate) fn new() -> Self { Controller { - open_connections: HashMap::new(), + open_connections: Arc::new(Mutex::new(HashMap::new())), } } pub(crate) async fn process_request( &mut self, request: Request, - ) -> Result, ConnectionError> { + ) -> Result, ConnectionError> { match request { - Request::Connect(conn_id, remote_addr, data) => { + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } => { let response = self - .create_new_connection(conn_id, remote_addr, data) + .create_new_connection(conn_id, remote_addr, data, return_address.clone()) .await?; - Ok(Some(response)) + Ok(Some((response, return_address))) } Request::Send(conn_id, data) => { - let response = self.send_to_connection(conn_id, data).await?; - Ok(Some(response)) + let (response, return_address) = self.send_to_connection(conn_id, data).await?; + Ok(Some((response, return_address))) } Request::Close(conn_id) => { - self.close_connection(conn_id)?; + self.close_connection(conn_id).await?; Ok(None) } } } + async fn insert_connection( + &mut self, + conn_id: ConnectionId, + conn: Connection, + ) -> Option { + self.open_connections.lock().await.insert(conn_id, conn) + } + async fn create_new_connection( &mut self, conn_id: ConnectionId, remote_addr: RemoteAddress, init_data: Vec, + return_address: Recipient, ) -> Result { - let mut connection = Connection::new(conn_id, remote_addr, &init_data).await?; - + println!("Connecting {} to remote {}", conn_id, remote_addr); + let mut connection = + Connection::new(conn_id, remote_addr, &init_data, return_address).await?; let response_data = connection.try_read_response_data().await?; - self.open_connections.insert(conn_id, connection); + self.insert_connection(conn_id, connection).await; Ok(Response::new(conn_id, response_data)) } @@ -65,19 +85,28 @@ impl Controller { &mut self, conn_id: ConnectionId, data: Vec, - ) -> Result { - let connection = self - .open_connections - .get_mut(&conn_id) + ) -> Result<(Response, Recipient), ConnectionError> { + let mut open_connections_guard = self.open_connections.lock().await; + + // TODO: is it possible to do it more nicely than getting lock -> removing connection -> + // processing -> reacquiring lock and putting connection back? + + let mut connection = open_connections_guard + .remove(&conn_id) .ok_or_else(|| ConnectionError::MissingConnection)?; connection.send_data(&data).await?; + let return_address = connection.return_address(); + + drop(open_connections_guard); let response_data = connection.try_read_response_data().await?; - Ok(Response::new(conn_id, response_data)) + self.insert_connection(conn_id, connection).await; + + Ok((Response::new(conn_id, response_data), return_address)) } - fn close_connection(&mut self, conn_id: ConnectionId) -> Result<(), ConnectionError> { - match self.open_connections.remove(&conn_id) { + async fn close_connection(&mut self, conn_id: ConnectionId) -> Result<(), ConnectionError> { + match self.open_connections.lock().await.remove(&conn_id) { // I *think* connection is closed implicitly on drop, but I'm not 100% sure! Some(_conn) => (), None => log::error!("tried to close non-existent connection - {}", conn_id), diff --git a/service-providers/simple-socks5/src/core.rs b/service-providers/simple-socks5/src/core.rs index 2ca19b7eb8..c0007681a9 100644 --- a/service-providers/simple-socks5/src/core.rs +++ b/service-providers/simple-socks5/src/core.rs @@ -1,87 +1,95 @@ use crate::{controller::Controller, websocket}; +use futures::channel::mpsc; use futures::SinkExt; use futures_util::StreamExt; use nymsphinx::addressing::clients::Recipient; -use simple_socks5_requests::Request; +use simple_socks5_requests::{Request, Response}; use tokio::net::TcpStream; -use tokio::runtime::Runtime; use tokio_tungstenite::tungstenite::protocol::Message; use tokio_tungstenite::WebSocketStream; use websocket::WebsocketConnectionError; use websocket_requests::{requests::ClientRequest, responses::ServerResponse}; -pub struct ServiceProvider { - runtime: Runtime, -} + +pub struct ServiceProvider {} impl ServiceProvider { pub fn new() -> ServiceProvider { - let runtime = Runtime::new().unwrap(); - ServiceProvider { runtime } + ServiceProvider {} } /// Start all subsystems - pub fn start(&mut self) { - let websocket_stream = self.connect_websocket("ws://localhost:1977"); + pub async fn run(&mut self) { + let websocket_stream = self.connect_websocket("ws://localhost:1977").await; let (mut websocket_writer, mut websocket_reader) = websocket_stream.split(); - let mut controller = Controller::new(); + let controller = Controller::new(); - self.runtime.block_on(async { - println!("\nAll systems go. Press CTRL-C to stop the server."); - while let Some(msg) = websocket_reader.next().await { - let data = msg.unwrap().into_data(); - let received = match ServerResponse::deserialize(&data).expect("todo: error handling") { - ServerResponse::Received(received) => received, - ServerResponse::Error(err) => { - panic!("received error from native client! - {}", err) - }, - _ => unimplemented!("probably should never be reached?") - }; - - let raw_message = received.message; - let request = Request::try_from_bytes(&raw_message).unwrap(); - let response = match controller.process_request(request).await.unwrap() { - None => continue, // restart the loop if we got nothing back - Some(response) => response, - }; - - // TODO: wire SURBs in here once they're available - let return_address = "7tVXwePpo6SM99sqM1xEp6S4T1TSpxYx97fTpEdvmF7i.GgrN8998SmwvQghNEvqtPPZCgMQqJovWBrzspMnBESsE@e3vUAo6YhB7zq3GH8B4k3iiGT4H2USjdd5ZMZoUsHdF"; - let recipient = Recipient::try_from_base58_string(return_address).unwrap(); + let (sender, mut reader) = mpsc::unbounded::<(Response, Recipient)>(); + let response_reader_future = async move { + // TODO: wire SURBs in here once they're available + while let Some((response, return_address)) = reader.next().await { // make 'request' to native-websocket client let response_message = ClientRequest::Send { - recipient, + recipient: return_address, message: response.into_bytes(), - with_reply_surb: false + with_reply_surb: false, }; let message = Message::Binary(response_message.serialize()); websocket_writer.send(message).await.unwrap(); } - }); + }; + tokio::spawn(response_reader_future); + + println!("\nAll systems go. Press CTRL-C to stop the server."); + while let Some(msg) = websocket_reader.next().await { + let data = msg.unwrap().into_data(); + let received = match ServerResponse::deserialize(&data).expect("todo: error handling") { + ServerResponse::Received(received) => received, + ServerResponse::Error(err) => { + panic!("received error from native client! - {}", err) + } + _ => unimplemented!("probably should never be reached?"), + }; + + let raw_message = received.message; + let request = Request::try_from_bytes(&raw_message).unwrap(); + + let mut controller_local_pointer = controller.clone(); + let response_sender_clone = sender.clone(); + tokio::spawn(async move { + if let Ok(response_option) = controller_local_pointer.process_request(request).await + { + if let Some((response, return_address)) = response_option { + // if we have an actual response - send it through the mixnet! + response_sender_clone + .unbounded_send((response, return_address)) + .expect("channel got closed?"); + } + }; + }); + } } - /// Keep running until the user hits CTRL-C. - pub fn run_forever(&mut self) { - if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) { - println!("Stopping with error: {:?}", e); - } - println!("\nStopping..."); - } + // /// Keep running until the user hits CTRL-C. + // pub fn run_forever(&mut self) { + // if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) { + // println!("Stopping with error: {:?}", e); + // } + // println!("\nStopping..."); + // } // Make the websocket connection so we can receive incoming Mixnet messages. - fn connect_websocket(&mut self, uri: &str) -> WebSocketStream { - self.runtime.block_on(async { - let ws_stream = match websocket::Connection::new(uri).connect().await { - Ok(ws_stream) => { - println!("* connected to local websocket server at {}", uri); - ws_stream - } - Err(WebsocketConnectionError::ConnectionNotEstablished) => { - panic!("Error: websocket connection attempt failed, is the Nym client running?") - } - }; - return ws_stream; - }) + async fn connect_websocket(&mut self, uri: &str) -> WebSocketStream { + let ws_stream = match websocket::Connection::new(uri).connect().await { + Ok(ws_stream) => { + println!("* connected to local websocket server at {}", uri); + ws_stream + } + Err(WebsocketConnectionError::ConnectionNotEstablished) => { + panic!("Error: websocket connection attempt failed, is the Nym client running?") + } + }; + return ws_stream; } } diff --git a/service-providers/simple-socks5/src/main.rs b/service-providers/simple-socks5/src/main.rs index b11257b2ec..63ef762c3a 100644 --- a/service-providers/simple-socks5/src/main.rs +++ b/service-providers/simple-socks5/src/main.rs @@ -3,9 +3,10 @@ mod controller; mod core; mod websocket; -fn main() { +#[tokio::main] +async fn main() { println!("Starting socks5 service provider:"); let mut server = core::ServiceProvider::new(); - server.start(); - server.run_forever(); + server.run().await; + // server.run_forever(); }