net/transport add tor transport

Cargo.lock

@@ -1591,6 +1591,7 @@ dependencies = [
  "pasta_curves",
  "rand 0.8.5",
  "rcgen",
+ "regex",
  "rocksdb",
  "rustls-pemfile 0.3.0",
  "serde",

Cargo.toml

@@ -79,6 +79,7 @@ itertools = {version = "0.10.3", optional = true}
 darkfi-derive = {path = "src/util/derive", optional = true}
 darkfi-derive-internal = {path = "src/util/derive-internal", optional = true}
 chrono = {version = "0.4.19", optional = true}
+regex = {version = "1.1.9", optional = true}
 
 # Misc
 termion = {version = "1.5.6", optional = true}
@@ -218,8 +219,10 @@ net = [
     "fxhash",
     "socket2",
     "futures-rustls",
+    "fast-socks5",
     "ed25519-compact",
     "rcgen",
+    "regex",
     "rustls-pemfile",
 
     "util",

src/net/transport.rs

@@ -5,9 +5,11 @@ use url::Url;
 
 mod tcp;
 mod tls;
+mod tor;
 
 pub use tcp::TcpTransport;
 pub use tls::TlsTransport;
+pub use tor::{TorController, TorTransport};
 
 pub trait Transport {
     type Acceptor;
@@ -35,6 +37,9 @@ pub enum TransportError<TErr> {
     #[error("Transport IO Error: {0}")]
     IoError(#[from] std::io::Error),
 
+    #[error("Socks5 Error: {0}")]
+    Socks5Error(#[from] fast_socks5::SocksError),
+
     #[error("{0}")]
     Other(TErr),
 }

src/net/transport/tor.rs

@@ -0,0 +1,199 @@
+use super::{Transport, TransportError};
+use async_std::{
+    net::{TcpListener, TcpStream},
+    sync::Arc,
+};
+use fast_socks5::{
+    client::{Config, Socks5Stream},
+    Result, SocksError,
+};
+use futures::prelude::*;
+
+use regex::Regex;
+use socket2::{Domain, Socket, Type};
+use std::{
+    io,
+    io::{BufRead, BufReader, Write},
+    net::SocketAddr,
+    pin::Pin,
+    time::Duration,
+};
+use url::Url;
+
+/// Implements communication through the tor proxy service
+#[derive(Clone)]
+pub struct TorTransport {
+    socks_url: Url,
+    tor_controller: Option<TorController>,
+}
+
+/// Represents information needed to communicate with the Tor control socket
+#[derive(Clone)]
+pub struct TorController {
+    socket: Arc<Socket>, // Need to hold this socket open as long as the tor trasport is alive, so ephemeral services are dropped when TorTransport is dropped
+    auth: String,
+}
+
+/// Wraps the errors, because dialing and listening use different communication
+#[derive(Debug, thiserror::Error)]
+pub enum TorError {
+    #[error("Transport IO Error: {0}")]
+    IoError(#[from] io::Error),
+    #[error("Socks: {0}")]
+    Socks5Error(#[from] SocksError),
+    #[error("Url parse error: {0}")]
+    UrlParseError(#[from] url::ParseError),
+    #[error("Regex parse error: {0}")]
+    RegexError(#[from] regex::Error),
+    #[error("Unexpected response from tor: {0}")]
+    TorError(String),
+}
+
+/// Contains the configuration to communicate with the Tor Controler
+/// As long as none of its clones are dropped, the hidden services created remain open
+impl TorController {
+    /// Creates a new TorTransport
+    ///
+    /// # Arguments
+    ///
+    /// * `url` - url to connect to the tor control. For example tcp://127.0.0.1:9051
+    ///
+    /// * `auth` - either authentication cookie bytes (32 byres) as hex in a string: assert_eq!(auth,"886b9177aec471965abd34b6a846dc32cf617dcff0625cba7a414e31dd4b75a0"), or a password as a quoted string: assert_eq!(auth,"\"mypassword\"")
+    pub fn new(url: Url, auth: String) -> Result<Self, io::Error> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+
+        match socket.connect(&socket_addr.into()) {
+            Ok(()) => {}
+            Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            Err(err) => return Err(err),
+        };
+        Ok(Self { socket: Arc::new(socket), auth })
+    }
+    /// Creates an ephemeral hidden service listening on port, returns onion address
+    pub fn create_ehs(&self, url: Url) -> Result<Url, TorError> {
+        let local_socket = self.socket.try_clone()?;
+        let mut stream = std::net::TcpStream::from(local_socket);
+
+        stream.set_write_timeout(Some(Duration::from_secs(2)))?;
+        let host =
+            url.host().ok_or(TorError::TorError("No host on url for listening".to_string()))?;
+        let port =
+            url.port().ok_or(TorError::TorError("No port on url for listening".to_string()))?;
+        let payload = format!(
+            "AUTHENTICATE {a}\r\nADD_ONION NEW:BEST Flags=DiscardPK Port={p},{h}:{p}\r\n",
+            a = self.auth,
+            p = port,
+            h = host
+        );
+        stream.write_all(payload.as_bytes())?;
+        stream.set_read_timeout(Some(Duration::from_secs(1)))?; // Maybe a bit too much. Gives tor time to reply
+        let mut reader = BufReader::new(stream);
+        let mut repl = String::new();
+        while let Ok(nbytes) = reader.read_line(&mut repl) {
+            if nbytes == 0 {
+                break
+            }
+        }
+        let re = Regex::new(r"250-ServiceID=(\w+*)")?;
+        let cap: Result<regex::Captures<'_>, TorError> =
+            re.captures(&repl).ok_or(TorError::TorError(repl.clone()));
+        let hurl = cap?.get(1).map_or(Err(TorError::TorError(repl.clone())), |m| Ok(m.as_str()))?;
+        let hurl = format!("tcp://{}.onion:{}", &hurl, port);
+        Ok(Url::parse(&hurl)?)
+    }
+}
+
+impl TorTransport {
+    /// Creates a new TorTransport
+    ///
+    /// # Arguments
+    ///
+    /// * `socks_url` - url to connect to the tor service. For example socks5://127.0.0.1:9050
+    ///
+    /// * `control_info` - Possibility to open a control connection to create ephemeral hidden services that live as long as the TorTransport. Is a tuple of control url and authentication cookie as string (represented in hex)
+    pub fn new(socks_url: Url, control_info: Option<(Url, String)>) -> Result<Self, TorError> {
+        match control_info {
+            Some(info) => {
+                let (url, auth) = info;
+                let tor_controller = Some(TorController::new(url, auth)?);
+                Ok(Self { socks_url, tor_controller })
+            }
+            None => Ok(Self { socks_url, tor_controller: None }),
+        }
+    }
+
+    pub fn create_ehs(&self, url: Url) -> Result<Url, TorError> {
+        self.tor_controller
+            .as_ref()
+            .ok_or(TorError::TorError("No controller configured for this transport".to_string()))?
+            .create_ehs(url)
+    }
+
+    pub async fn do_dial(self, url: Url) -> Result<Socks5Stream<TcpStream>, TorError> {
+        let socks_url_str = self.socks_url.socket_addrs(|| None)?[0].to_string();
+        let host = url.host().unwrap().to_string();
+        let port = url.port().unwrap_or_else(|| 80);
+        let config = Config::default();
+        let stream = if !self.socks_url.username().is_empty() && self.socks_url.password().is_some()
+        {
+            Socks5Stream::connect_with_password(
+                socks_url_str,
+                host,
+                port,
+                self.socks_url.username().to_string(),
+                self.socks_url.password().unwrap().to_string(),
+                config,
+            )
+            .await?
+        } else {
+            Socks5Stream::connect(socks_url_str, host, port, config).await?
+        };
+        Ok(stream)
+    }
+
+    fn create_socket(&self, socket_addr: SocketAddr) -> io::Result<Socket> {
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+        Ok(socket)
+    }
+
+    pub async fn do_listen(self, url: Url) -> Result<TcpListener, TorError> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        socket.listen(1024)?;
+        socket.set_nonblocking(true)?;
+        Ok(TcpListener::from(std::net::TcpListener::from(socket)))
+    }
+}
+
+impl Transport for TorTransport {
+    type Acceptor = TcpListener;
+    type Connector = Socks5Stream<TcpStream>;
+
+    type Error = TorError;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+        Ok(Box::pin(self.do_listen(url)))
+    }
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
+        Ok(Box::pin(self.do_dial(url)))
+    }
+}

tests/network_transports.rs

@@ -4,9 +4,11 @@ use async_std::{
     stream::StreamExt,
     task,
 };
-use url::Url;
 
-use darkfi::net::transport::{TcpTransport, TlsTransport, Transport};
+use darkfi::net::transport::{TcpTransport, TlsTransport, TorTransport, Transport};
+
+use std::{env::var, fs};
+use url::Url;
 
 #[async_std::test]
 async fn tcp_transport() {
@@ -60,3 +62,102 @@ async fn tls_transport() {
 
     assert_eq!(buf, payload);
 }
+
+#[async_std::test]
+#[ignore]
+async fn tor_transport_no_control() {
+    let url = Url::parse("socks5://127.0.0.1:9050").unwrap();
+    let hurl = var("DARKFI_TOR_LOCAL_ADDRESS")
+        .expect("Please set the env var DARKFI_TOR_LOCAL_ADDRESS to the configured local address in hidden service. \
+        For example: \'export DARKFI_TOR_LOCAL_ADDRESS=\"tcp://127.0.0.1:8080\"\'");
+    let hurl = Url::parse(&hurl).unwrap();
+
+    let onion = var("DARKFI_TOR_PUBLIC_ADDRESS").expect(
+        "Please set the env var DARKFI_TOR_PUBLIC_ADDRESS to the configured onion address. \
+        For example: \'export DARKFI_TOR_PUBLIC_ADDRESS=\"tor://abcdefghij234567.onion\"\'",
+    );
+
+    let tor = TorTransport::new(url, None).unwrap();
+    let listener = tor.clone().listen_on(hurl).unwrap().await.unwrap();
+
+    let _ = task::spawn(async move {
+        let mut incoming = listener.incoming();
+        while let Some(stream) = incoming.next().await {
+            let stream = stream.unwrap();
+            let (reader, writer) = &mut (&stream, &stream);
+            io::copy(reader, writer).await.unwrap();
+        }
+    });
+
+    let payload = b"ohai tor";
+    let url = Url::parse(&onion).unwrap();
+    let mut client = tor.dial(url).unwrap().await.unwrap();
+    client.write_all(payload).await.unwrap();
+    let mut buf = vec![0_u8; 8];
+    client.read_exact(&mut buf).await.unwrap();
+    assert_eq!(buf, payload);
+}
+
+#[async_std::test]
+#[ignore]
+async fn tor_transport_with_control() {
+    let auth_cookie = var("DARKFI_TOR_COOKIE").expect(
+        "Please set the env var DARKFI_TOR_COOKIE to the configured tor cookie file. \
+        For example: \'export DARKFI_TOR_COOKIE=\"/var/lib/tor/control_auth_cookie\"\'",
+    );
+    let auth_cookie = hex::encode(&fs::read(auth_cookie).unwrap());
+    let socks_url = Url::parse("socks5://127.0.0.1:9050").unwrap();
+    let torc_url = Url::parse("tcp://127.0.0.1:9051").unwrap();
+    let local_url = Url::parse("tcp://127.0.0.1:8787").unwrap();
+
+    let tor = TorTransport::new(socks_url, Some((torc_url, auth_cookie))).unwrap();
+    // generate EHS pointing to local address
+    let hurl = tor.create_ehs(local_url.clone()).unwrap();
+
+    let listener = tor.clone().listen_on(local_url).unwrap().await.unwrap();
+
+    let _ = task::spawn(async move {
+        let mut incoming = listener.incoming();
+        while let Some(stream) = incoming.next().await {
+            let stream = stream.unwrap();
+            let (reader, writer) = &mut (&stream, &stream);
+            io::copy(reader, writer).await.unwrap();
+        }
+    });
+
+    let payload = b"ohai tor";
+
+    let mut client = tor.dial(hurl).unwrap().await.unwrap();
+    client.write_all(payload).await.unwrap();
+    let mut buf = vec![0_u8; 8];
+    client.read_exact(&mut buf).await.unwrap();
+    assert_eq!(buf, payload);
+}
+
+#[async_std::test]
+#[should_panic(expected = "Socks5Error(ReplyError(HostUnreachable))")]
+#[ignore]
+async fn tor_transport_with_control_dropped() {
+    let auth_cookie = var("DARKFI_TOR_COOKIE").expect(
+        "Please set the env var DARKFI_TOR_COOKIE to the configured tor cookie file. \
+        For example: \'export DARKFI_TOR_COOKIE=\"/var/lib/tor/control_auth_cookie\"\'",
+    );
+    let auth_cookie = hex::encode(&fs::read(auth_cookie).unwrap());
+    let socks_url = Url::parse("socks5://127.0.0.1:9050").unwrap();
+    let torc_url = Url::parse("tcp://127.0.0.1:9051").unwrap();
+    let local_url = Url::parse("tcp://127.0.0.1:8787").unwrap();
+    let hurl;
+    // We create a new scope for the Transport, to see if when we drop it, the host is still reachable
+    {
+        let tor = TorTransport::new(socks_url.clone(), Some((torc_url, auth_cookie))).unwrap();
+        // generate EHS pointing to local address
+        hurl = tor.create_ehs(local_url.clone()).unwrap();
+        // Nothing is listening, but the host is reachable.
+        // In this case, dialing should fail with Socks5Error(ReplyError(GeneralFailure));
+        // And not with Socks5Error(ReplyError(HostUnreachable))
+    }
+
+    let tor_client = TorTransport::new(socks_url, None).unwrap();
+    // Try to reach the host
+    let _client = tor_client.dial(hurl).unwrap().await.unwrap();
+}