From 9efc1950464ea10fc368d94f18233fe3c3e3100f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 27 Aug 2020 13:59:37 +0100 Subject: [PATCH] Explicit proxy runner + closing local connection if remote is over (#314) * Updated built dependency So that it wouldn't fail on nwindows * Concurrent RW client-side * Draft of a ProxyRunner * Super experimental proxy runner for provider Very much WIP * Extracting shared code + cleanup * Logging cleanup * Removed redundant close request --- Cargo.lock | 119 ++++----- clients/client-core/Cargo.toml | 2 +- clients/native/Cargo.toml | 2 +- clients/socks5/Cargo.toml | 3 +- clients/socks5/src/socks/authentication.rs | 2 +- clients/socks5/src/socks/client.rs | 226 +++++++++++------ clients/socks5/src/socks/mixnet_responses.rs | 31 +-- clients/socks5/src/socks/request.rs | 17 +- clients/socks5/src/socks/server.rs | 18 +- clients/socks5/src/socks/types.rs | 14 +- .../crypto/src/asymmetric/encryption/mod.rs | 2 +- common/nymsphinx/addressing/src/clients.rs | 2 +- common/utils/Cargo.toml | 9 +- common/utils/src/connection_controller.rs | 109 +++++++++ common/utils/src/lib.rs | 2 + common/utils/src/proxy_runner.rs | 229 ++++++++++++++++++ common/utils/src/read_delay_loop.rs | 7 +- gateway/Cargo.toml | 2 +- mixnode/Cargo.toml | 2 +- service-providers/simple-socks5/Cargo.toml | 6 +- .../simple-socks5-requests/src/request.rs | 83 +++---- .../simple-socks5-requests/src/response.rs | 49 ++-- .../simple-socks5/src/connection.rs | 63 ++--- .../simple-socks5/src/controller.rs | 117 --------- service-providers/simple-socks5/src/core.rs | 178 ++++++++++---- service-providers/simple-socks5/src/main.rs | 24 +- .../simple-socks5/src/websocket.rs | 6 +- validator/Cargo.toml | 2 +- 28 files changed, 861 insertions(+), 465 deletions(-) create mode 100644 common/utils/src/connection_controller.rs create mode 100644 common/utils/src/proxy_runner.rs delete mode 100644 service-providers/simple-socks5/src/controller.rs diff --git a/Cargo.lock b/Cargo.lock index 404989ea0b..da9bc5ab8b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -325,14 +325,11 @@ checksum = "b170cd256a3f9fa6b9edae3e44a7dfdfc77e8124dbc3e2612d75f9c3e2396dae" [[package]] name = "built" -version = "0.3.2" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d2315cfb416f86e05360edc950b1d7d25ecfb00f7f8eba60dbd7882a0f2e944" +checksum = "3fa7899958f4aa3c40edc1b033d0e956763319e398924abb80a0034dda5bb198" dependencies = [ - "chrono", - "git2", - "semver", - "toml", + "cargo-lock", ] [[package]] @@ -369,14 +366,23 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "130aac562c0dd69c56b3b1cc8ffd2e17be31d0b6c25b61c96b76231aa23e39e1" +[[package]] +name = "cargo-lock" +version = "4.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8504b63dd1249fd1745b7b4ef9b6f7b107ddeb3c95370043c7dbcc38653a2679" +dependencies = [ + "semver", + "serde", + "toml", + "url 2.1.1", +] + [[package]] name = "cc" version = "1.0.50" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "95e28fa049fda1c330bcf9d723be7663a899c4679724b34c81e9f5a326aab8cd" -dependencies = [ - "jobserver", -] [[package]] name = "cfg-if" @@ -448,7 +454,7 @@ dependencies = [ "serde", "sled 0.33.0", "tempfile", - "tokio 0.2.16", + "tokio 0.2.22", "topology", "url 2.1.1", ] @@ -692,7 +698,7 @@ dependencies = [ "pretty_env_logger", "reqwest", "serde", - "tokio 0.2.16", + "tokio 0.2.22", ] [[package]] @@ -977,7 +983,7 @@ dependencies = [ "gateway-requests", "log 0.4.8", "nymsphinx", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tungstenite", ] @@ -1033,19 +1039,6 @@ version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aaf91faf136cb47367fa430cd46e37a788775e7fa104f8b4bcb3861dc389b724" -[[package]] -name = "git2" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c1af51ea8a906616af45a4ce78eacf25860f7a13ae7bf8a814693f0f4037a26" -dependencies = [ - "bitflags", - "libc", - "libgit2-sys", - "log 0.4.8", - "url 2.1.1", -] - [[package]] name = "h2" version = "0.2.5" @@ -1061,7 +1054,7 @@ dependencies = [ "indexmap", "log 0.4.8", "slab", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-util", ] @@ -1202,7 +1195,7 @@ dependencies = [ "net2", "pin-project", "time", - "tokio 0.2.16", + "tokio 0.2.22", "tower-service", "want", ] @@ -1216,7 +1209,7 @@ dependencies = [ "bytes 0.5.4", "hyper 0.13.5", "native-tls", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tls", ] @@ -1303,15 +1296,6 @@ version = "0.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8b7a7c0c47db5545ed3fef7468ee7bb5b74691498139e4b3f6a20685dc6dd8e" -[[package]] -name = "jobserver" -version = "0.1.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c71313ebb9439f74b00d9d2dcec36440beaf57a6aa0623068441dd7cd81a7f2" -dependencies = [ - "libc", -] - [[package]] name = "js-sys" version = "0.3.44" @@ -1355,30 +1339,6 @@ version = "0.2.72" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9f8082297d534141b30c8d39e9b1773713ab50fdbe4ff30f750d063b3bfd701" -[[package]] -name = "libgit2-sys" -version = "0.9.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4870c781f6063efb83150cd22c1ddf6ecf58531419e7570cdcced46970f64a16" -dependencies = [ - "cc", - "libc", - "libz-sys", - "pkg-config", -] - -[[package]] -name = "libz-sys" -version = "1.0.25" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2eb5e43362e38e2bca2fd5f5134c4d4564a23a5c28e9b95411652021a8675ebe" -dependencies = [ - "cc", - "libc", - "pkg-config", - "vcpkg", -] - [[package]] name = "lioness" version = "0.1.2" @@ -1583,7 +1543,7 @@ dependencies = [ "futures 0.3.4", "log 0.4.8", "nymsphinx", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-util", ] @@ -1693,7 +1653,7 @@ dependencies = [ "serde_json", "sled 0.33.0", "tempfile", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tungstenite", "topology", "url 2.1.1", @@ -1740,7 +1700,7 @@ dependencies = [ "serde", "sled 0.31.0", "tempfile", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tungstenite", "tokio-util", "tungstenite 0.10.1", @@ -1767,7 +1727,7 @@ dependencies = [ "pretty_env_logger", "serde", "tempfile", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-util", "topology", ] @@ -1792,7 +1752,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "tokio 0.2.16", + "tokio 0.2.22", ] [[package]] @@ -2553,7 +2513,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "time", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tls", "url 2.1.1", "wasm-bindgen", @@ -2670,6 +2630,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403" dependencies = [ "semver-parser", + "serde", ] [[package]] @@ -2772,15 +2733,13 @@ dependencies = [ name = "simple-socks5" version = "0.1.0" dependencies = [ - "bs58", "clap", "futures 0.3.4", - "futures-util", "log 0.4.8", "nymsphinx", "pretty_env_logger", "simple-socks5-requests", - "tokio 0.2.16", + "tokio 0.2.22", "tokio-tungstenite", "utils", "websocket-requests", @@ -2903,11 +2862,12 @@ dependencies = [ "gateway-requests", "log 0.4.8", "nymsphinx", + "pin-project", "pretty_env_logger", "rand 0.7.3", "simple-socks5-requests", "snafu", - "tokio 0.2.16", + "tokio 0.2.22", "topology", "url 2.1.1", "utils", @@ -3090,9 +3050,9 @@ dependencies = [ [[package]] name = "tokio" -version = "0.2.16" +version = "0.2.22" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee5a0dd887e37d37390c13ff8ac830f992307fe30a1fff0ab8427af67211ba28" +checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd" dependencies = [ "bytes 0.5.4", "fnv", @@ -3216,7 +3176,7 @@ checksum = "ed0049c119b6d505c4447f5c64873636c7af6c75ab0d45fd9f618d82acb8016d" dependencies = [ "bytes 0.5.4", "futures-core", - "tokio 0.2.16", + "tokio 0.2.22", ] [[package]] @@ -3255,7 +3215,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343" dependencies = [ "native-tls", - "tokio 0.2.16", + "tokio 0.2.22", ] [[package]] @@ -3267,7 +3227,7 @@ dependencies = [ "futures-util", "log 0.4.8", "pin-project", - "tokio 0.2.16", + "tokio 0.2.22", "tungstenite 0.11.1", ] @@ -3282,7 +3242,7 @@ dependencies = [ "futures-sink", "log 0.4.8", "pin-project-lite", - "tokio 0.2.16", + "tokio 0.2.22", ] [[package]] @@ -3487,7 +3447,10 @@ name = "utils" version = "0.1.0" dependencies = [ "bytes 0.5.4", - "tokio 0.2.16", + "futures 0.3.4", + "log 0.4.8", + "simple-socks5-requests", + "tokio 0.2.22", "tokio-test", ] diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml index c3aee9d5db..00e983236c 100644 --- a/clients/client-core/Cargo.toml +++ b/clients/client-core/Cargo.toml @@ -28,7 +28,7 @@ pemstore = { path = "../../common/pemstore" } topology = { path = "../../common/topology" } [build-dependencies] -built = "0.3.2" +built = "0.4.3" [dev-dependencies] tempfile = "3.1.0" \ No newline at end of file diff --git a/clients/native/Cargo.toml b/clients/native/Cargo.toml index 710adab148..1ecf9ec9b2 100644 --- a/clients/native/Cargo.toml +++ b/clients/native/Cargo.toml @@ -43,7 +43,7 @@ topology = { path = "../../common/topology" } websocket-requests = { path = "websocket-requests" } [build-dependencies] -built = "0.3.2" +built = "0.4.3" [dev-dependencies] tempfile = "3.1.0" diff --git a/clients/socks5/Cargo.toml b/clients/socks5/Cargo.toml index efc00de9fa..b6679a97d5 100644 --- a/clients/socks5/Cargo.toml +++ b/clients/socks5/Cargo.toml @@ -14,6 +14,7 @@ clap = "2.33.0" dotenv = "0.15.0" futures = "0.3" log = "0.4" +pin-project = "0.4" pretty_env_logger = "0.3" rand = { version = "0.7.3", features = ["wasm-bindgen"] } snafu = "0.4.1" @@ -33,4 +34,4 @@ topology = { path = "../../common/topology" } utils = { path = "../../common/utils" } [build-dependencies] -built = "0.3.2" +built = "0.4.3" diff --git a/clients/socks5/src/socks/authentication.rs b/clients/socks5/src/socks/authentication.rs index 7d1afaee78..6a564689f2 100644 --- a/clients/socks5/src/socks/authentication.rs +++ b/clients/socks5/src/socks/authentication.rs @@ -2,7 +2,7 @@ pub(crate) enum AuthenticationMethods { /// No Authentication NoAuth = 0x00, - // GssApi = 0x01, + // GssApi = 0x01, // question to DH: why is this commented? /// Authenticate with a username / password UserPass = 0x02, /// Cannot authenticate diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index 73b1c5bd08..fa27d52762 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -1,31 +1,134 @@ #![forbid(unsafe_code)] -use rand::RngCore; -use std::{collections::HashMap, net::Shutdown, sync::Arc}; - -use log::*; -use tokio::prelude::*; -use tokio::{self, net::TcpStream}; - -use nymsphinx::addressing::clients::Recipient; - -use client_core::client::inbound_messages::InputMessage; -use client_core::client::inbound_messages::InputMessageSender; -use futures::{channel::oneshot, lock::Mutex}; - use super::authentication::{AuthenticationMethods, Authenticator, User}; use super::request::{SocksCommand, SocksRequest}; use super::types::{ResponseCode, SocksProxyError}; use super::{RESERVED, SOCKS_VERSION}; +use client_core::client::inbound_messages::InputMessage; +use client_core::client::inbound_messages::InputMessageSender; +use futures::channel::mpsc; +use futures::core_reexport::pin::Pin; +use futures::task::{Context, Poll}; +use log::*; +use nymsphinx::addressing::clients::Recipient; +use pin_project::pin_project; +use rand::RngCore; use simple_socks5_requests::{ConnectionId, Request}; +use std::net::{Shutdown, SocketAddr}; +use tokio::prelude::*; +use tokio::{self, net::TcpStream}; +use utils::connection_controller::{ConnectionReceiver, ControllerCommand, ControllerSender}; +use utils::proxy_runner::ProxyRunner; + +#[pin_project(project = StateProject)] +enum StreamState { + Available(TcpStream), + RunningProxy, +} + +impl StreamState { + fn finish_proxy(&mut self, stream: TcpStream) { + match self { + StreamState::RunningProxy => *self = StreamState::Available(stream), + StreamState::Available(_) => panic!("invalid state!"), + } + } + + fn run_proxy(&mut self) -> TcpStream { + // It's not the nicest way to do it, but it works + #[allow(unused_assignments)] + let mut stream = None; + *self = match std::mem::replace(self, StreamState::RunningProxy) { + StreamState::Available(inner_stream) => { + stream = Some(inner_stream); + StreamState::RunningProxy + } + StreamState::RunningProxy => panic!("invalid state"), + }; + stream.unwrap() + } + + /// Returns the remote address that this stream is connected to. + fn peer_addr(&self) -> io::Result { + match self { + StreamState::RunningProxy => Err(io::Error::new( + io::ErrorKind::NotFound, + "stream is being used to run the proxy", + )), + StreamState::Available(ref stream) => stream.peer_addr(), + } + } + + fn shutdown(&self, how: Shutdown) -> io::Result<()> { + // shutdown should only be called if proxy is not being run. If it is, there's some bug + // somewhere + match self { + StreamState::RunningProxy => panic!("Tried to shutdown stream while proxy is running"), + StreamState::Available(ref stream) => TcpStream::shutdown(stream, how), + } + } +} + +// convenience implementations +impl AsyncRead for StreamState { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + match self.project() { + StateProject::RunningProxy => Poll::Ready(Err(io::Error::new( + io::ErrorKind::NotFound, + "stream is being used to run the proxy", + ))), + StateProject::Available(ref mut stream) => Pin::new(stream).poll_read(cx, buf), + } + } +} + +impl AsyncWrite for StreamState { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + match self.project() { + StateProject::RunningProxy => Poll::Ready(Err(io::Error::new( + io::ErrorKind::NotFound, + "stream is being used to run the proxy", + ))), + StateProject::Available(ref mut stream) => Pin::new(stream).poll_write(cx, buf), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.project() { + StateProject::RunningProxy => Poll::Ready(Err(io::Error::new( + io::ErrorKind::NotFound, + "stream is being used to run the proxy", + ))), + StateProject::Available(ref mut stream) => Pin::new(stream).poll_flush(cx), + } + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.project() { + StateProject::RunningProxy => Poll::Ready(Err(io::Error::new( + io::ErrorKind::NotFound, + "stream is being used to run the proxy", + ))), + StateProject::Available(ref mut stream) => Pin::new(stream).poll_shutdown(cx), + } + } +} /// A client connecting to the Socks proxy server, because /// it wants to make a Nym-protected outbound request. Typically, this is /// something like e.g. a wallet app running on your laptop connecting to /// SphinxSocksServer. pub(crate) struct SocksClient { - active_streams: ActiveStreams, - stream: TcpStream, + controller_sender: ControllerSender, + stream: StreamState, auth_nmethods: u8, authenticator: Authenticator, socks_version: u8, @@ -35,13 +138,14 @@ pub(crate) struct SocksClient { self_address: Recipient, } -type StreamResponseSender = oneshot::Sender>; - -pub(crate) type ActiveStreams = Arc>>; - impl Drop for SocksClient { fn drop(&mut self) { - println!("socksclient is going out of scope - the stream is getting dropped!") + // TODO: decrease to debug/trace + info!("socksclient is going out of scope - the stream is getting dropped!"); + info!("Connection {} is getting closed", self.connection_id); + self.controller_sender + .unbounded_send(ControllerCommand::Remove(self.connection_id)) + .unwrap(); } } @@ -52,14 +156,14 @@ impl SocksClient { authenticator: Authenticator, input_sender: InputMessageSender, service_provider: Recipient, - active_streams: ActiveStreams, + controller_sender: ControllerSender, self_address: Recipient, ) -> Self { let connection_id = Self::generate_random(); SocksClient { - active_streams, + controller_sender, connection_id, - stream, + stream: StreamState::Available(stream), auth_nmethods: 0, socks_version: 0, authenticator, @@ -82,7 +186,7 @@ impl SocksClient { /// Shutdown the TcpStream to the client and end the session pub fn shutdown(&mut self) -> Result<(), SocksProxyError> { - println!("client is shutting down its TCP stream"); + info!("client is shutting down its TCP stream"); self.stream.shutdown(Shutdown::Both)?; Ok(()) } @@ -118,20 +222,22 @@ impl SocksClient { self.send_to_mixnet(request.into_bytes()).await; } - async fn send_request_to_mixnet_and_get_response(&mut self, request: Request) -> Vec { - self.send_to_mixnet(request.into_bytes()).await; + async fn run_proxy(&mut self, conn_receiver: ConnectionReceiver) { + let stream = self.stream.run_proxy(); + let connection_id = self.connection_id; - // refactor idea: crossbeam oneshot channels are faster - let (sender, receiver) = oneshot::channel(); - let mut active_streams_guard = self.active_streams.lock().await; - if active_streams_guard - .insert(self.connection_id, sender) - .is_some() - { - panic!("there is already an active request with the same id present - it's probably a bug!") - }; - drop(active_streams_guard); - receiver.await.unwrap() + let input_sender = self.input_sender.clone(); + + let recipient = self.service_provider.clone(); + let (stream, _) = ProxyRunner::new(stream, conn_receiver, input_sender, connection_id) + .run(move |conn_id, read_data, socket_closed| { + let provider_request = Request::new_send(conn_id, read_data, socket_closed); + InputMessage::new_fresh(recipient, provider_request.into_bytes(), false) + }) + .await + .into_inner(); + // recover stream from the proxy + self.stream.finish_proxy(stream) } /// Handles a client request. @@ -140,15 +246,22 @@ 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(); + + // setup for receiving from the mixnet + let (mix_sender, mix_receiver) = mpsc::unbounded(); + self.controller_sender + .unbounded_send(ControllerCommand::Insert(self.connection_id, mix_sender)) + .unwrap(); + match request.command { // Use the Proxy to connect to the specified addr/port SocksCommand::Connect => { trace!("Connecting to: {:?}", request.to_socket()); self.acknowledge_socks5().await; - let request_data_bytes = + // 'connect' needs to be handled manually due to different structure + let (request_data_bytes, _) = SocksRequest::try_read_request_data(&mut self.stream, &client_address).await?; let socks_provider_request = Request::new_connect( self.connection_id, @@ -156,42 +269,17 @@ impl SocksClient { request_data_bytes, self.self_address.clone(), ); - let response_data = self - .send_request_to_mixnet_and_get_response(socks_provider_request) - .await; - self.stream.write_all(&response_data).await.unwrap(); - loop { - if let Ok(request_data_bytes) = - SocksRequest::try_read_request_data(&mut self.stream, &client_address).await - { - if request_data_bytes.is_empty() { - break; - } - let socks_provider_request = - Request::new_send(self.connection_id, request_data_bytes); - let response_data = self - .send_request_to_mixnet_and_get_response(socks_provider_request) - .await; - if let Err(err) = self.stream.write_all(&response_data).await { - error!( - "tried to write to (presumably) closed connection - {:?}", - err - ); - break; - } - } else { - break; - } - } - let socks_provider_request = Request::new_close(self.connection_id); self.send_request_to_mixnet(socks_provider_request).await; - // TODO: where is connection removed from active connection?? + info!("Starting proxy for {}", remote_address.clone()); + self.run_proxy(mix_receiver).await; + info!("Proxy for {} is finished", remote_address); } SocksCommand::Bind => unimplemented!(), // not handled SocksCommand::UdpAssociate => unimplemented!(), // not handled }; + Ok(()) } @@ -235,7 +323,7 @@ impl SocksClient { /// read-a-byte-or-two. The bytes being extracted look like this: /// /// +----+------+----------+------+------------+ - // |ver | ulen | uname | plen | password | + /// |ver | ulen | uname | plen | password | /// +----+------+----------+------+------------+ /// | 1 | 1 | 1 to 255 | 1 | 1 to 255 | /// +----+------+----------+------+------------+ diff --git a/clients/socks5/src/socks/mixnet_responses.rs b/clients/socks5/src/socks/mixnet_responses.rs index 6714ab0263..c222177e11 100644 --- a/clients/socks5/src/socks/mixnet_responses.rs +++ b/clients/socks5/src/socks/mixnet_responses.rs @@ -1,4 +1,3 @@ -use super::client::ActiveStreams; use client_core::client::received_buffer::ReconstructedMessagesReceiver; use client_core::client::received_buffer::{ReceivedBufferMessage, ReceivedBufferRequestSender}; use futures::channel::mpsc; @@ -6,11 +5,12 @@ use futures::StreamExt; use log::*; use nymsphinx::receiver::ReconstructedMessage; use simple_socks5_requests::Response; +use utils::connection_controller::{ControllerCommand, ControllerSender}; pub(crate) struct MixnetResponseListener { buffer_requester: ReceivedBufferRequestSender, mix_response_receiver: ReconstructedMessagesReceiver, - active_streams: ActiveStreams, + controller_sender: ControllerSender, } impl Drop for MixnetResponseListener { @@ -24,7 +24,7 @@ impl Drop for MixnetResponseListener { impl MixnetResponseListener { pub(crate) fn new( buffer_requester: ReceivedBufferRequestSender, - active_streams: ActiveStreams, + controller_sender: ControllerSender, ) -> Self { let (mix_response_sender, mix_response_receiver) = mpsc::unbounded(); buffer_requester @@ -32,7 +32,7 @@ impl MixnetResponseListener { .unwrap(); MixnetResponseListener { - active_streams, + controller_sender, buffer_requester, mix_response_receiver, } @@ -41,7 +41,7 @@ impl MixnetResponseListener { async fn on_message(&self, reconstructed_message: ReconstructedMessage) { let raw_message = reconstructed_message.message; if reconstructed_message.reply_SURB.is_some() { - println!("this message had a surb - we didn't do anything with it"); + warn!("this message had a surb - we didn't do anything with it"); } let response = match Response::try_from_bytes(&raw_message) { @@ -52,18 +52,13 @@ impl MixnetResponseListener { Ok(data) => data, }; - let mut active_streams_guard = self.active_streams.lock().await; - - // `remove` gives back the entry (assuming it exists). There's no reason - // for it to persist after we send data back - if let Some(stream_receiver) = active_streams_guard.remove(&response.connection_id) { - stream_receiver.send(response.data).unwrap() - } else { - warn!( - "no connection_id exists with id: {:?}", - &response.connection_id - ) - } + self.controller_sender + .unbounded_send(ControllerCommand::Send( + response.connection_id, + response.data, + response.is_closed, + )) + .unwrap(); } pub(crate) async fn run(&mut self) { @@ -72,6 +67,6 @@ impl MixnetResponseListener { self.on_message(reconstructed_message).await; } } - println!("We should never see this message"); + error!("We should never see this message"); } } diff --git a/clients/socks5/src/socks/request.rs b/clients/socks5/src/socks/request.rs index 3e81097af1..e5798ffb05 100644 --- a/clients/socks5/src/socks/request.rs +++ b/clients/socks5/src/socks/request.rs @@ -2,7 +2,6 @@ use super::types::{AddrType, ResponseCode, SocksProxyError}; 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; @@ -17,14 +16,17 @@ pub(crate) struct SocksRequest { impl SocksRequest { /// Parse a SOCKS5 request from a TcpStream - pub async fn from_stream(stream: &mut TcpStream) -> Result { + pub async fn from_stream(stream: &mut R) -> Result + where + R: AsyncRead + Unpin, + { let mut packet = [0u8; 4]; // Read a byte from the stream and determine the version being requested stream.read_exact(&mut packet).await?; if packet[0] != SOCKS_VERSION { warn!("from_stream Unsupported version: SOCKS{}", packet[0]); - stream.shutdown(); + return Err(SocksProxyError::UnsupportedProxyVersion(packet[0])); } // Get command @@ -36,7 +38,6 @@ impl SocksRequest { } None => { warn!("Invalid Command"); - stream.shutdown(); Err(ResponseCode::CommandNotSupported) } }?; @@ -51,7 +52,6 @@ impl SocksRequest { } None => { error!("No Addr"); - stream.shutdown(); Err(ResponseCode::AddrTypeNotSupported) } }?; @@ -112,10 +112,13 @@ impl SocksRequest { /// 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( + pub(crate) async fn try_read_request_data( reader: &mut R, remote_address: &str, - ) -> io::Result> { + ) -> io::Result<(Vec, bool)> + where + R: AsyncRead + Unpin, + { let timeout_duration = std::time::Duration::from_millis(500); 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 44adb70d96..12d6141457 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -7,11 +7,11 @@ use super::{ use client_core::client::{ inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender, }; -use futures::lock::Mutex; use log::*; use nymsphinx::addressing::clients::Recipient; -use std::{collections::HashMap, net::SocketAddr, sync::Arc}; +use std::net::SocketAddr; use tokio::net::TcpListener; +use utils::connection_controller::Controller; /// A Socks5 server that listens for connections. pub struct SphinxSocksServer { @@ -46,14 +46,18 @@ impl SphinxSocksServer { input_sender: InputMessageSender, buffer_requester: ReceivedBufferRequestSender, ) -> Result<(), SocksProxyError> { - info!("Serving Connections..."); let mut listener = TcpListener::bind(self.listening_address).await.unwrap(); + info!("Serving Connections..."); - // todo: probably just create a wrapper type for this guy - let active_streams = Arc::new(Mutex::new(HashMap::new())); + // controller for managing all active connections + let (mut active_streams_controller, controller_sender) = Controller::new(); + tokio::spawn(async move { + active_streams_controller.run().await; + }); + // listener for mix messages let mut mixnet_response_listener = - MixnetResponseListener::new(buffer_requester, Arc::clone(&active_streams)); + MixnetResponseListener::new(buffer_requester, controller_sender.clone()); tokio::spawn(async move { mixnet_response_listener.run().await; @@ -67,7 +71,7 @@ impl SphinxSocksServer { self.authenticator.clone(), input_sender.clone(), self.service_provider.clone(), - Arc::clone(&active_streams), + controller_sender.clone(), self.self_address.clone(), ); diff --git a/clients/socks5/src/socks/types.rs b/clients/socks5/src/socks/types.rs index cb3c9af5ef..b043da5ccc 100644 --- a/clients/socks5/src/socks/types.rs +++ b/clients/socks5/src/socks/types.rs @@ -23,21 +23,27 @@ pub(crate) enum ResponseCode { #[derive(Debug)] pub enum SocksProxyError { - GenericError(String), + GenericError(Box), + UnsupportedProxyVersion(u8), } impl std::fmt::Display for SocksProxyError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "foomp") + match self { + SocksProxyError::GenericError(err) => write!(f, "GenericError - {}", err), + SocksProxyError::UnsupportedProxyVersion(version) => { + write!(f, "Unsupported proxy version {}", version) + } + } } } impl From for SocksProxyError where - E: std::error::Error, + E: std::error::Error + Send + Sync + 'static, { fn from(err: E) -> Self { - SocksProxyError::GenericError(err.to_string()) + SocksProxyError::GenericError(Box::new(err)) } } diff --git a/common/crypto/src/asymmetric/encryption/mod.rs b/common/crypto/src/asymmetric/encryption/mod.rs index b321cda538..b8433efcd5 100644 --- a/common/crypto/src/asymmetric/encryption/mod.rs +++ b/common/crypto/src/asymmetric/encryption/mod.rs @@ -100,7 +100,7 @@ impl PemStorableKeyPair for KeyPair { } } -#[derive(Debug, Clone)] +#[derive(Debug, Copy, Clone)] pub struct PublicKey(x25519_dalek::PublicKey); impl PublicKey { diff --git a/common/nymsphinx/addressing/src/clients.rs b/common/nymsphinx/addressing/src/clients.rs index 82d86e8a62..6fa6e21c6b 100644 --- a/common/nymsphinx/addressing/src/clients.rs +++ b/common/nymsphinx/addressing/src/clients.rs @@ -58,7 +58,7 @@ impl From for RecipientFormattingError { } // TODO: this should a different home... somewhere, but where? -#[derive(Clone, Debug)] +#[derive(Clone, Copy, Debug)] pub struct Recipient { client_identity: ClientIdentity, client_encryption_key: ClientEncryptionKey, diff --git a/common/utils/Cargo.toml b/common/utils/Cargo.toml index 3b5c7ea2d6..edbf3ed5cd 100644 --- a/common/utils/Cargo.toml +++ b/common/utils/Cargo.toml @@ -8,9 +8,12 @@ edition = "2018" [dependencies] bytes = "0.5" -# no need for any features as `AsyncRead` is always available -tokio = { version = "0.2", features = [] } +# TODO: "time" feature is only required for the delay loop which is going to go away very soon! +tokio = { version = "0.2", features = [ "tcp", "io-util", "sync", "macros", "time" ] } +futures = "0.3" +log = "0.4" +simple-socks5-requests = { path = "../../service-providers/simple-socks5/simple-socks5-requests" } [dev-dependencies] -tokio = { version = "0.2", features = ["rt-threaded", "macros"] } +tokio = { version = "0.2", features = ["rt-threaded"] } tokio-test = "0.2" \ No newline at end of file diff --git a/common/utils/src/connection_controller.rs b/common/utils/src/connection_controller.rs new file mode 100644 index 0000000000..26a8287807 --- /dev/null +++ b/common/utils/src/connection_controller.rs @@ -0,0 +1,109 @@ +// 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 futures::channel::mpsc; +use log::*; +use simple_socks5_requests::ConnectionId; +use std::collections::HashMap; +use tokio::stream::StreamExt; + +/// A generic message produced after reading from a socket/connection. It includes data that was +/// actually read alongside boolean indicating whether the connection got closed so that +/// remote could act accordingly. +#[derive(Debug)] +pub struct ConnectionMessage { + pub payload: Vec, + pub socket_closed: bool, +} + +/// Channel responsible for sending data that was received from mix network into particular connection. +/// Data includes the actual payload that is to be written onto the connection +/// alongside boolean indicating whether the remote connection was closed after producing this message, +/// so that the local connection should also shut down. +pub type ConnectionSender = mpsc::UnboundedSender; + +/// Receiver part of the [`ConnectionSender`] +pub type ConnectionReceiver = mpsc::UnboundedReceiver; + +pub type ControllerSender = mpsc::UnboundedSender; +pub type ControllerReceiver = mpsc::UnboundedReceiver; + +pub enum ControllerCommand { + Insert(ConnectionId, ConnectionSender), + Remove(ConnectionId), + Send(ConnectionId, Vec, bool), +} + +/// Controller represents a way of managing multiple open connections that are used for socks5 +/// proxy. +pub struct Controller { + active_connections: HashMap, + receiver: ControllerReceiver, +} + +impl Controller { + pub fn new() -> (Self, ControllerSender) { + let (sender, receiver) = mpsc::unbounded(); + ( + Controller { + active_connections: HashMap::new(), + receiver, + }, + sender, + ) + } + + fn insert_connection(&mut self, conn_id: ConnectionId, sender: ConnectionSender) { + if self.active_connections.insert(conn_id, sender).is_some() { + panic!("there is already an active request with the same id present - it's probably a bug!") + } + } + + fn remove_connection(&mut self, conn_id: ConnectionId) { + debug!("Removing {} from controller", conn_id); + if self.active_connections.remove(&conn_id).is_none() { + error!( + "tried to remove non-existing connection with id: {:?}", + conn_id + ) + } + } + + fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec, is_closed: bool) { + if let Some(sender) = self.active_connections.get_mut(&conn_id) { + sender + .unbounded_send(ConnectionMessage { + payload, + socket_closed: is_closed, + }) + .unwrap() + } else { + error!("no connection exists with id: {:?}", conn_id); + } + } + + pub async fn run(&mut self) { + while let Some(command) = self.receiver.next().await { + match command { + ControllerCommand::Send(conn_id, data, is_closed) => { + self.send_to_connection(conn_id, data, is_closed) + } + ControllerCommand::Insert(conn_id, sender) => { + self.insert_connection(conn_id, sender) + } + ControllerCommand::Remove(conn_id) => self.remove_connection(conn_id), + } + } + } +} diff --git a/common/utils/src/lib.rs b/common/utils/src/lib.rs index 15f5844db7..be9e60e6b4 100644 --- a/common/utils/src/lib.rs +++ b/common/utils/src/lib.rs @@ -13,4 +13,6 @@ // limitations under the License. pub mod available_reader; +pub mod connection_controller; +pub mod proxy_runner; pub mod read_delay_loop; diff --git a/common/utils/src/proxy_runner.rs b/common/utils/src/proxy_runner.rs new file mode 100644 index 0000000000..180fa498d7 --- /dev/null +++ b/common/utils/src/proxy_runner.rs @@ -0,0 +1,229 @@ +// 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 super::read_delay_loop::try_read_data; +use crate::connection_controller::ConnectionReceiver; +use futures::channel::mpsc; +use log::*; +use simple_socks5_requests::ConnectionId; +use std::sync::Arc; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::TcpStream; +use tokio::prelude::*; +use tokio::stream::StreamExt; +use tokio::sync::Notify; + +#[derive(Debug)] +pub struct ProxyMessage { + pub data: Vec, + pub socket_closed: bool, +} + +impl Into for (Vec, bool) { + fn into(self) -> ProxyMessage { + ProxyMessage { + data: self.0, + socket_closed: self.1, + } + } +} + +pub type MixProxySender = mpsc::UnboundedSender; + +#[derive(Debug)] +pub struct ProxyRunner { + /// receives data from the mix network and sends that into the socket + mix_receiver: Option, + + /// sends whatever was read from the socket into the mix network + mix_sender: MixProxySender, + + socket: Option, + connection_id: ConnectionId, +} + +impl ProxyRunner +where + S: Send + 'static, +{ + pub fn new( + socket: TcpStream, + mix_receiver: ConnectionReceiver, + mix_sender: MixProxySender, + connection_id: ConnectionId, + ) -> Self { + ProxyRunner { + mix_receiver: Some(mix_receiver), + mix_sender, + socket: Some(socket), + connection_id, + } + } + + async fn run_inbound( + mut reader: OwnedReadHalf, + notify_closed: Arc, + connection_id: ConnectionId, + mix_sender: MixProxySender, + adapter_fn: F, + ) -> OwnedReadHalf + where + F: Fn(ConnectionId, Vec, bool) -> S + Send + 'static, + { + // TODO: to be removed with sequence numbers... + let socket_read_timeout_duration = std::time::Duration::from_millis(500); + + let address = reader.as_ref().peer_addr().unwrap().to_string(); + loop { + tokio::select! { + _ = notify_closed.notified() => { + // the remote socket is closed, so there's no point + // in reading anything more because we won't be able to write to remote anyway! + break + } + // try to read from local socket and push everything to mixnet to the remote + reading_result = try_read_data(socket_read_timeout_duration, &mut reader, &address) => { + let (read_data, timed_out) = match reading_result { + Ok(data) => data, + Err(err) => { + error!("failed to read request from the socket - {}", err); + break; + } + }; + + if read_data.is_empty() && timed_out { + // no point in writing empty data on each timeout + continue + } + + info!( + "Going to send {} bytes via mixnet to remote {}. Is local closed: {}", + read_data.len(), + connection_id, + !timed_out + ); + + mix_sender.unbounded_send(adapter_fn(connection_id, read_data, !timed_out)).unwrap(); + + if !timed_out { + // technically we already informed it when we sent the message to mixnet above + info!("The local socket is closed - won't receive any more data. Informing remote about that..."); + // no point in reading from mixnet if connection is closed! + notify_closed.notify(); + break; + } + } + } + } + + reader + } + + async fn run_outbound( + mut writer: OwnedWriteHalf, + notify_closed: Arc, + mut mix_receiver: ConnectionReceiver, + connection_id: ConnectionId, + ) -> (OwnedWriteHalf, ConnectionReceiver) { + loop { + tokio::select! { + _ = notify_closed.notified() => { + // no need to read from mixnet as we won't be able to send to socket + // anyway + break + } + mix_data = mix_receiver.next() => { + if mix_data.is_none() { + warn!("mix receiver is none so we already got removed somewhere. This isn't really a warning, but shouldn't happen to begin with, so please say if you see this message"); + // we already got closed + // not sure if we HAVE TO notify the other task, but might as well + notify_closed.notify(); + break + } + let connection_message = mix_data.unwrap(); + + info!( + "Going to write {} bytes received from mixnet to connection {}. Is remote closed: {}", + connection_message.payload.len(), + connection_id, + connection_message.socket_closed + ); + + if let Err(err) = writer.write_all(&connection_message.payload).await { + // the other half is probably going to blow up too (if not, this task also needs to notify the other one!!) + error!("failed to write response back to the socket - {}", err); + break; + } + if connection_message.socket_closed { + info!("Remote socket got closed - closing the local socket too"); + notify_closed.notify(); + break + } + } + } + } + + (writer, mix_receiver) + } + + // The `adapter_fn` is used to transform whatever was read into appropriate + // request/response as required by entity running particular side of the proxy. + pub async fn run(mut self, adapter_fn: F) -> Self + where + F: Fn(ConnectionId, Vec, bool) -> S + Send + 'static, + { + let notify_closed = Arc::new(Notify::new()); + let notify_clone = Arc::clone(¬ify_closed); + + let (read_half, write_half) = self.socket.take().unwrap().into_split(); + let mix_receiver = self.mix_receiver.take().unwrap(); + + // should run until either inbound closes or is notified from outbound + let inbound_future = Self::run_inbound( + read_half, + notify_closed, + self.connection_id, + self.mix_sender.clone(), + adapter_fn, + ); + let outbound_future = + Self::run_outbound(write_half, notify_clone, mix_receiver, self.connection_id); + + // TODO: this shouldn't really have to spawn tasks inside "library" code, but + // if we used join directly, stuff would have been executed on the same thread + // (it's not bad, but an unnecessary slowdown) + let handle_inbound = tokio::spawn(inbound_future); + let handle_outbound = tokio::spawn(outbound_future); + + let (inbound_result, outbound_result) = + futures::future::join(handle_inbound, handle_outbound).await; + + if inbound_result.is_err() || outbound_result.is_err() { + panic!("TODO: some future error?") + } + + let (write_half, mix_receiver) = outbound_result.unwrap(); + + self.socket = Some(write_half.reunite(inbound_result.unwrap()).unwrap()); + self.mix_receiver = Some(mix_receiver); + self + } + + pub fn into_inner(mut self) -> (TcpStream, ConnectionReceiver) { + ( + self.socket.take().unwrap(), + self.mix_receiver.take().unwrap(), + ) + } +} diff --git a/common/utils/src/read_delay_loop.rs b/common/utils/src/read_delay_loop.rs index 020f3ba82c..0cd536ac1a 100644 --- a/common/utils/src/read_delay_loop.rs +++ b/common/utils/src/read_delay_loop.rs @@ -24,11 +24,12 @@ use std::io; use tokio::io::AsyncRead; use tokio::time::Duration; +// It returns data alognside information whether it timed out while reading from the socket pub async fn try_read_data( timeout: Duration, mut reader: R, address: &str, -) -> io::Result> +) -> io::Result<(Vec, bool)> where R: AsyncRead + Unpin, { @@ -41,7 +42,7 @@ where tokio::select! { _ = &mut delay => { println!("Timed out. returning {} bytes received from {}", data.len(), address); - return Ok(data) // we return all response data on timeout + return Ok((data, true)) // we return all response data on timeout } read_data = &mut available_reader => { match read_data { @@ -53,7 +54,7 @@ where 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) + return Ok((data, false)) } let now = tokio::time::Instant::now(); let next = now + timeout; diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 1b2c262ab0..65d36b7d90 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -35,7 +35,7 @@ version = "0.10.0" default-features = false [build-dependencies] -built = "0.3.2" +built = "0.4.3" [dev-dependencies] tempfile = "3.1.0" \ No newline at end of file diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index 28b811eecc..9c0b730501 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -30,7 +30,7 @@ pemstore = {path = "../common/pemstore"} topology = {path = "../common/topology"} [build-dependencies] -built = "0.3.2" +built = "0.4.3" [dev-dependencies] tempfile = "3.1.0" \ No newline at end of file diff --git a/service-providers/simple-socks5/Cargo.toml b/service-providers/simple-socks5/Cargo.toml index 079fc9d2d3..19de49a494 100644 --- a/service-providers/simple-socks5/Cargo.toml +++ b/service-providers/simple-socks5/Cargo.toml @@ -1,19 +1,17 @@ [package] name = "simple-socks5" version = "0.1.0" -authors = ["Dave Hrycyszyn "] +authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -bs58 = "0.3.0" clap = "2.33.0" futures = "0.3" -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"] } +tokio = { version = "0.2", features = ["stream", "tcp", "rt-threaded", "macros"] } tokio-tungstenite = "0.11.0" nymsphinx = { path = "../../common/nymsphinx" } 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 43df046dad..e8b733da20 100644 --- a/service-providers/simple-socks5/simple-socks5-requests/src/request.rs +++ b/service-providers/simple-socks5/simple-socks5-requests/src/request.rs @@ -1,5 +1,6 @@ use nymsphinx_addressing::clients::{Recipient, RecipientFormattingError}; use std::convert::TryFrom; +use std::fmt::{self}; pub type ConnectionId = u64; pub type RemoteAddress = String; @@ -9,7 +10,6 @@ pub type RemoteAddress = String; pub enum RequestFlag { Connect = 0, Send = 1, - Close = 2, } #[derive(Debug, PartialEq)] @@ -23,6 +23,28 @@ pub enum RequestError { MalformedReturnAddress(RecipientFormattingError), } +impl fmt::Display for RequestError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> fmt::Result { + match self { + RequestError::AddressLengthTooShort => { + write!(f, "not enough bytes to recover the lenght of the address") + } + RequestError::AddressTooShort => write!(f, "not enough bytes to recover the address"), + RequestError::ConnectionIdTooShort => { + write!(f, "not enough bytes to recover the connection id") + } + RequestError::NoData => write!(f, "no data provided"), + RequestError::UnknownRequestFlag => write!(f, "request of unknown type"), + RequestError::ReturnAddressTooShort => write!(f, "too short return address"), + RequestError::MalformedReturnAddress(recipient_err) => { + write!(f, "malformed return address - {}", recipient_err) + } + } + } +} + +impl std::error::Error for RequestError {} + impl RequestError { pub fn is_malformed_return(&self) -> bool { match self { @@ -39,7 +61,6 @@ impl TryFrom for RequestFlag { match value { _ if value == (RequestFlag::Connect as u8) => Ok(Self::Connect), _ if value == (RequestFlag::Send as u8) => Ok(Self::Send), - _ if value == (RequestFlag::Close as u8) => Ok(Self::Close), _ => Err(RequestError::UnknownRequestFlag), } } @@ -60,10 +81,7 @@ pub enum Request { }, /// Re-use an existing TCP connection, sending more request data up it. - Send(ConnectionId, Vec), - - /// Close an existing TCP connection. - Close(ConnectionId), + Send(ConnectionId, Vec, bool), } impl Request { @@ -83,13 +101,8 @@ impl Request { } /// Construct a new Request::Send instance - pub fn new_send(conn_id: ConnectionId, data: Vec) -> Request { - Request::Send(conn_id, data) - } - - /// Construct a new Request::Close instance - pub fn new_close(conn_id: ConnectionId) -> Request { - Request::Close(conn_id) + pub fn new_send(conn_id: ConnectionId, data: Vec, local_closed: bool) -> Request { + Request::Send(conn_id, data, local_closed) } /// Deserialize the request type, connection id, destination address and port, @@ -155,8 +168,12 @@ impl Request { return_address, }) } - RequestFlag::Send => Ok(Request::Send(connection_id, b[9..].as_ref().to_vec())), - RequestFlag::Close => Ok(Request::Close(connection_id)), + RequestFlag::Send => { + let local_closed = b[9] != 0; + let data = b[10..].as_ref().to_vec(); + + Ok(Request::Send(connection_id, data, local_closed)) + } } } @@ -183,13 +200,11 @@ impl Request { .chain(data.into_iter()) .collect() } - Request::Send(conn_id, data) => std::iter::once(RequestFlag::Send as u8) + Request::Send(conn_id, data, local_closed) => std::iter::once(RequestFlag::Send as u8) .chain(conn_id.to_be_bytes().iter().cloned()) + .chain(std::iter::once(local_closed as u8)) .chain(data.into_iter()) .collect(), - Request::Close(conn_id) => std::iter::once(RequestFlag::Close as u8) - .chain(conn_id.to_be_bytes().iter().cloned()) - .collect(), } } } @@ -444,13 +459,14 @@ mod request_deserialization_tests { #[test] fn works_when_request_is_sized_properly_even_without_data() { - // correct 8 bytes of connection_id, and 0 bytes request data - let request_bytes = [RequestFlag::Send as u8, 1, 2, 3, 4, 5, 6, 7, 8].to_vec(); + // correct 8 bytes of connection_id, 1 byte of local_closed and 0 bytes request data + let request_bytes = [RequestFlag::Send as u8, 1, 2, 3, 4, 5, 6, 7, 8, 0].to_vec(); let request = Request::try_from_bytes(&request_bytes).unwrap(); match request { - Request::Send(conn_id, data) => { + Request::Send(conn_id, data, local_closed) => { assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); assert_eq!(Vec::::new(), data); + assert!(!local_closed) } _ => unreachable!(), } @@ -458,7 +474,7 @@ mod request_deserialization_tests { #[test] fn works_when_request_is_sized_properly_and_has_data() { - // correct 8 bytes of connection_id, and 3 bytes request data (all 255) + // correct 8 bytes of connection_id, 1 byte of local_closed and 3 bytes request data (all 255) let request_bytes = [ RequestFlag::Send as u8, 1, @@ -469,6 +485,7 @@ mod request_deserialization_tests { 6, 7, 8, + 0, 255, 255, 255, @@ -477,26 +494,10 @@ mod request_deserialization_tests { let request = Request::try_from_bytes(&request_bytes).unwrap(); match request { - Request::Send(conn_id, data) => { + Request::Send(conn_id, data, local_closed) => { assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); assert_eq!(vec![255, 255, 255], data); - } - _ => unreachable!(), - } - } - } - - #[cfg(test)] - mod connection_close_requests { - use super::*; - - #[test] - fn can_be_constructed() { - let request_bytes = [RequestFlag::Close as u8, 1, 2, 3, 4, 5, 6, 7, 8].to_vec(); - let request = Request::try_from_bytes(&request_bytes).unwrap(); - match request { - Request::Close(conn_id) => { - assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id); + assert!(!local_closed) } _ => unreachable!(), } diff --git a/service-providers/simple-socks5/simple-socks5-requests/src/response.rs b/service-providers/simple-socks5/simple-socks5-requests/src/response.rs index 3f0fa23909..52fdf83f8d 100644 --- a/service-providers/simple-socks5/simple-socks5-requests/src/response.rs +++ b/service-providers/simple-socks5/simple-socks5-requests/src/response.rs @@ -12,14 +12,16 @@ pub enum ResponseError { pub struct Response { pub data: Vec, pub connection_id: ConnectionId, + pub is_closed: bool, } impl Response { /// Constructor for responses - pub fn new(connection_id: ConnectionId, data: Vec) -> Self { + pub fn new(connection_id: ConnectionId, data: Vec, is_closed: bool) -> Self { Response { connection_id, data, + is_closed, } } @@ -28,15 +30,16 @@ impl Response { return Err(ResponseError::NoData); } - if b.len() < 8 { + let is_closed = b[0] != 0; + + if b.len() < 9 { return Err(ResponseError::ConnectionIdTooShort); } let mut connection_id_bytes = b.to_vec(); - let data = connection_id_bytes.split_off(8); + let data = connection_id_bytes.split_off(9); let connection_id = u64::from_be_bytes([ - connection_id_bytes[0], connection_id_bytes[1], connection_id_bytes[2], connection_id_bytes[3], @@ -44,19 +47,18 @@ impl Response { connection_id_bytes[5], connection_id_bytes[6], connection_id_bytes[7], + connection_id_bytes[8], ]); - let response = Response::new(connection_id, data); + let response = Response::new(connection_id, data, is_closed); Ok(response) } /// Serializes the response into bytes so that it can be sent back through /// the mixnet to the requesting application. pub fn into_bytes(self) -> Vec { - self.connection_id - .to_be_bytes() - .iter() - .cloned() + std::iter::once(self.is_closed as u8) + .chain(self.connection_id.to_be_bytes().iter().cloned()) .chain(self.data.into_iter()) .collect() } @@ -87,22 +89,29 @@ mod constructing_socks5_responses_from_bytes { #[test] fn works_when_there_is_no_data() { - let response_bytes = vec![0, 1, 2, 3, 4, 5, 6, 7]; - let expected = Response::new(u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), Vec::new()); - let actual = Response::try_from_bytes(&response_bytes).unwrap(); - assert_eq!(expected.connection_id, actual.connection_id); - assert_eq!(expected.data, actual.data); - } - - #[test] - fn works_when_there_is_data() { - let response_bytes = vec![0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255]; + let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7]; let expected = Response::new( u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), - vec![255, 255, 255], + Vec::new(), + false, ); let actual = Response::try_from_bytes(&response_bytes).unwrap(); assert_eq!(expected.connection_id, actual.connection_id); assert_eq!(expected.data, actual.data); + assert_eq!(expected.is_closed, actual.is_closed); + } + + #[test] + fn works_when_there_is_data() { + let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255]; + let expected = Response::new( + u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]), + vec![255, 255, 255], + false, + ); + let actual = Response::try_from_bytes(&response_bytes).unwrap(); + assert_eq!(expected.connection_id, actual.connection_id); + assert_eq!(expected.data, actual.data); + assert_eq!(expected.is_closed, actual.is_closed); } } diff --git a/service-providers/simple-socks5/src/connection.rs b/service-providers/simple-socks5/src/connection.rs index f6c53ddc52..81c7b6c776 100644 --- a/service-providers/simple-socks5/src/connection.rs +++ b/service-providers/simple-socks5/src/connection.rs @@ -1,8 +1,11 @@ +use futures::channel::mpsc; +use log::*; use nymsphinx::addressing::clients::Recipient; -use simple_socks5_requests::{ConnectionId, RemoteAddress}; +use simple_socks5_requests::{ConnectionId, RemoteAddress, Response}; use tokio::net::TcpStream; use tokio::prelude::*; -use utils::read_delay_loop::try_read_data; +use utils::connection_controller::ConnectionReceiver; +use utils::proxy_runner::ProxyRunner; /// A TCP connection between the Socks5 service provider, which makes /// outbound requests on behalf of users and returns the responses through @@ -11,7 +14,7 @@ use utils::read_delay_loop::try_read_data; pub(crate) struct Connection { id: ConnectionId, address: RemoteAddress, - conn: TcpStream, + conn: Option, return_address: Recipient, } @@ -22,36 +25,38 @@ impl Connection { initial_data: &[u8], return_address: Recipient, ) -> io::Result { - 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 { + let mut conn = TcpStream::connect(&address).await?; + + // write the initial data to the connection before continuing + info!( + "Sending initial {} bytes to {}", + initial_data.len(), + address + ); + conn.write_all(initial_data).await?; + + Ok(Connection { id, address, - conn, + conn: Some(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 - } - - /// Read response data by looping, waiting for anything we get back from the - /// 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); - try_read_data(timeout_duration, &mut self.conn, &self.address).await + pub(crate) async fn run_proxy( + &mut self, + mix_receiver: ConnectionReceiver, + mix_sender: mpsc::UnboundedSender<(Response, Recipient)>, + ) { + let stream = self.conn.take().unwrap(); + let connection_id = self.id; + let recipient = self.return_address; + let (stream, _) = ProxyRunner::new(stream, mix_receiver, mix_sender, connection_id) + .run(move |conn_id, read_data, socket_closed| { + (Response::new(conn_id, read_data, socket_closed), recipient) + }) + .await + .into_inner(); + self.conn = Some(stream); } } diff --git a/service-providers/simple-socks5/src/controller.rs b/service-providers/simple-socks5/src/controller.rs deleted file mode 100644 index cb791bf563..0000000000 --- a/service-providers/simple-socks5/src/controller.rs +++ /dev/null @@ -1,117 +0,0 @@ -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 { - ConnectionFailed(io::Error), - MissingConnection, -} - -impl From for ConnectionError { - fn from(e: io::Error) -> Self { - ConnectionError::ConnectionFailed(e) - } -} - -// clone is fine here because HashMap is never cloned, the pointer is -#[derive(Clone)] -pub(crate) struct Controller { - open_connections: Arc>>, -} - -impl Controller { - pub(crate) fn new() -> Self { - Controller { - open_connections: Arc::new(Mutex::new(HashMap::new())), - } - } - - pub(crate) async fn process_request( - &mut self, - request: Request, - ) -> Result, ConnectionError> { - match request { - Request::Connect { - conn_id, - remote_addr, - data, - return_address, - } => { - let response = self - .create_new_connection(conn_id, remote_addr, data, return_address.clone()) - .await?; - Ok(Some((response, return_address))) - } - Request::Send(conn_id, data) => { - 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).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 { - 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.insert_connection(conn_id, connection).await; - Ok(Response::new(conn_id, response_data)) - } - - async fn send_to_connection( - &mut self, - conn_id: ConnectionId, - data: Vec, - ) -> 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?; - self.insert_connection(conn_id, connection).await; - - Ok((Response::new(conn_id, response_data), return_address)) - } - - 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), - } - - Ok(()) - } -} diff --git a/service-providers/simple-socks5/src/core.rs b/service-providers/simple-socks5/src/core.rs index c0007681a9..08c7c6562e 100644 --- a/service-providers/simple-socks5/src/core.rs +++ b/service-providers/simple-socks5/src/core.rs @@ -1,50 +1,90 @@ -use crate::{controller::Controller, websocket}; +use crate::connection::Connection; +use crate::websocket; use futures::channel::mpsc; -use futures::SinkExt; -use futures_util::StreamExt; +use futures::stream::SplitSink; +use futures::{SinkExt, StreamExt}; +use log::*; use nymsphinx::addressing::clients::Recipient; use simple_socks5_requests::{Request, Response}; use tokio::net::TcpStream; use tokio_tungstenite::tungstenite::protocol::Message; use tokio_tungstenite::WebSocketStream; +use utils::connection_controller::{Controller, ControllerCommand}; use websocket::WebsocketConnectionError; use websocket_requests::{requests::ClientRequest, responses::ServerResponse}; -pub struct ServiceProvider {} +pub struct ServiceProvider { + listening_address: String, +} impl ServiceProvider { - pub fn new() -> ServiceProvider { - ServiceProvider {} + pub fn new(listening_address: String) -> ServiceProvider { + ServiceProvider { listening_address } + } + + /// Listens for any messages from `mix_reader` that should be written back to the mix network + /// via the `websocket_writer`. + async fn mixnet_response_listener( + mut websocket_writer: SplitSink, Message>, + mut mix_reader: mpsc::UnboundedReceiver<(Response, Recipient)>, + ) { + // TODO: wire SURBs in here once they're available + while let Some((response, return_address)) = mix_reader.next().await { + // make 'request' to native-websocket client + let response_message = ClientRequest::Send { + recipient: return_address, + message: response.into_bytes(), + with_reply_surb: false, + }; + + let message = Message::Binary(response_message.serialize()); + websocket_writer.send(message).await.unwrap(); + } } /// Start all subsystems 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 controller = Controller::new(); + let websocket_stream = self.connect_websocket(&self.listening_address).await; - let (sender, mut reader) = mpsc::unbounded::<(Response, Recipient)>(); + // split the websocket so that we could read and write from separate threads + let (websocket_writer, mut websocket_reader) = websocket_stream.split(); - 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: return_address, - message: response.into_bytes(), - with_reply_surb: false, - }; + // channels responsible for managing messages that are to be sent to the mix network. The receiver is + // going to be used by `mixnet_response_listener` + let (mix_input_sender, mix_input_receiver) = mpsc::unbounded::<(Response, Recipient)>(); - let message = Message::Binary(response_message.serialize()); - websocket_writer.send(message).await.unwrap(); - } - }; - tokio::spawn(response_reader_future); + // controller for managing all active connections + let (mut active_connections_controller, controller_sender) = Controller::new(); + tokio::spawn(async move { + active_connections_controller.run().await; + }); + + // start the listener for mix messages + tokio::spawn(async move { + Self::mixnet_response_listener(websocket_writer, mix_input_receiver).await; + }); println!("\nAll systems go. Press CTRL-C to stop the server."); + + // for each incoming message from the websocket... (which in 99.99% cases is going to be a mix message) while let Some(msg) = websocket_reader.next().await { - let data = msg.unwrap().into_data(); - let received = match ServerResponse::deserialize(&data).expect("todo: error handling") { + let data = msg + .expect("we failed to read from the websocket!") + .into_data(); + + // try to recover the actual message from the mix network... + let deserialized_message = match ServerResponse::deserialize(&data) { + Ok(deserialized) => deserialized, + Err(err) => { + error!( + "Failed to deserialize received websocket message! - {}", + err + ); + continue; + } + }; + + let received = match deserialized_message { ServerResponse::Received(received) => received, ServerResponse::Error(err) => { panic!("received error from native client! - {}", err) @@ -53,37 +93,75 @@ impl ServiceProvider { }; let raw_message = received.message; - let request = Request::try_from_bytes(&raw_message).unwrap(); + // TODO: here be potential SURB (i.e. received.reply_SURB) - 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?"); - } - }; - }); + // try to treat each received mix message as a service provider request + let deserialized_request = match Request::try_from_bytes(&raw_message) { + Ok(request) => request, + Err(err) => { + error!("Failed to deserialized received request! - {}", err); + continue; + } + }; + + match deserialized_request { + Request::Connect { + conn_id, + remote_addr, + data, + return_address, + } => { + // Connect implies it's a fresh connection - register it with our controller + let (mix_sender, mix_receiver) = mpsc::unbounded(); + controller_sender + .unbounded_send(ControllerCommand::Insert(conn_id, mix_sender)) + .unwrap(); + + let controller_sender_clone = controller_sender.clone(); + // and start the proxy for this connection + let mix_input_sender_clone = mix_input_sender.clone(); + tokio::spawn(async move { + let mut conn = match Connection::new( + conn_id, + remote_addr.clone(), + &data, + return_address, + ) + .await + { + Ok(conn) => conn, + Err(err) => { + error!( + "error while connecting to {:?} ! - {:?}", + remote_addr.clone(), + err + ); + return; + } + }; + + info!("Starting proxy for {}", remote_addr.clone()); + conn.run_proxy(mix_receiver, mix_input_sender_clone).await; + // proxy is done - remove the access channel from the controller + controller_sender_clone + .unbounded_send(ControllerCommand::Remove(conn_id)) + .unwrap(); + info!("Proxy for {} is finished", remote_addr); + }); + } + // on send just tell the controller to send that data to the correct connection + Request::Send(conn_id, data, closed) => controller_sender + .unbounded_send(ControllerCommand::Send(conn_id, data, closed)) + .unwrap(), + } } } - // /// 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. - async fn connect_websocket(&mut self, uri: &str) -> WebSocketStream { + async fn connect_websocket(&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); + info!("* connected to local websocket server at {}", uri); ws_stream } Err(WebsocketConnectionError::ConnectionNotEstablished) => { diff --git a/service-providers/simple-socks5/src/main.rs b/service-providers/simple-socks5/src/main.rs index 63ef762c3a..0532cb725a 100644 --- a/service-providers/simple-socks5/src/main.rs +++ b/service-providers/simple-socks5/src/main.rs @@ -1,12 +1,30 @@ mod connection; -mod controller; mod core; mod websocket; +fn setup_logging() { + let mut log_builder = pretty_env_logger::formatted_timed_builder(); + if let Ok(s) = ::std::env::var("RUST_LOG") { + log_builder.parse_filters(&s); + } else { + // default to 'Info' + log_builder.filter(None, log::LevelFilter::Info); + } + + log_builder + .filter_module("hyper", log::LevelFilter::Warn) + .filter_module("tokio_reactor", log::LevelFilter::Warn) + .filter_module("reqwest", log::LevelFilter::Warn) + .filter_module("mio", log::LevelFilter::Warn) + .filter_module("want", log::LevelFilter::Warn) + .init(); +} + #[tokio::main] async fn main() { + setup_logging(); + let uri = "ws://localhost:1977"; println!("Starting socks5 service provider:"); - let mut server = core::ServiceProvider::new(); + let mut server = core::ServiceProvider::new(uri.into()); server.run().await; - // server.run_forever(); } diff --git a/service-providers/simple-socks5/src/websocket.rs b/service-providers/simple-socks5/src/websocket.rs index e38152e4cc..eec1162c6e 100644 --- a/service-providers/simple-socks5/src/websocket.rs +++ b/service-providers/simple-socks5/src/websocket.rs @@ -15,9 +15,9 @@ impl Connection { pub async fn connect(&self) -> Result, WebsocketConnectionError> { match connect_async(&self.uri).await { - Ok((ws_stream, _)) => return Ok(ws_stream), - Err(_e) => return Err(WebsocketConnectionError::ConnectionNotEstablished), - }; + Ok((ws_stream, _)) => Ok(ws_stream), + Err(_e) => Err(WebsocketConnectionError::ConnectionNotEstablished), + } } } diff --git a/validator/Cargo.toml b/validator/Cargo.toml index 50c0e6fe8e..7e2f65678c 100644 --- a/validator/Cargo.toml +++ b/validator/Cargo.toml @@ -27,7 +27,7 @@ tokio = { version = "0.2", features = ["full"] } config = { path = "../common/config" } [build-dependencies] -built = "0.3.2" +built = "0.4.3" [dev-dependencies] tempfile = "3.1.0" \ No newline at end of file