/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2024 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::{ io::{self, ErrorKind}, time::Duration, }; use async_trait::async_trait; use log::error; use smol::io::{AsyncRead, AsyncWrite}; use url::Url; /// TLS upgrade mechanism pub(crate) mod tls; /// SOCKS5 proxy client pub mod socks5; /// TCP transport pub(crate) mod tcp; #[cfg(feature = "p2p-tor")] /// Tor transport pub(crate) mod tor; #[cfg(feature = "p2p-nym")] /// Nym transport pub(crate) mod nym; /// Unix socket transport pub(crate) mod unix; /// Dialer variants #[derive(Debug, Clone)] pub enum DialerVariant { /// Plain TCP Tcp(tcp::TcpDialer), /// TCP with TLS TcpTls(tcp::TcpDialer), #[cfg(feature = "p2p-tor")] /// Tor Tor(tor::TorDialer), #[cfg(feature = "p2p-tor")] /// Tor with TLS TorTls(tor::TorDialer), #[cfg(feature = "p2p-nym")] /// Nym Nym(nym::NymDialer), #[cfg(feature = "p2p-nym")] /// Nym with TLS NymTls(nym::NymDialer), /// Unix socket Unix(unix::UnixDialer), } /// Listener variants #[derive(Debug, Clone)] pub enum ListenerVariant { /// Plain TCP Tcp(tcp::TcpListener), /// TCP with TLS TcpTls(tcp::TcpListener), #[cfg(feature = "p2p-tor")] /// Tor Tor(tor::TorListener), /// Unix socket Unix(unix::UnixListener), } /// A dialer that is able to transparently operate over arbitrary transports. pub struct Dialer { /// The endpoint to connect to endpoint: Url, /// The dialer variant (transport protocol) variant: DialerVariant, } macro_rules! enforce_hostport { ($endpoint:ident) => { if $endpoint.host_str().is_none() || $endpoint.port().is_none() { return Err(io::Error::from_raw_os_error(libc::ENETUNREACH)) } }; } macro_rules! enforce_abspath { ($endpoint:ident) => { if $endpoint.host_str().is_some() || $endpoint.port().is_some() { return Err(io::Error::from_raw_os_error(libc::ENETUNREACH)) } if $endpoint.to_file_path().is_err() { return Err(io::Error::from_raw_os_error(libc::ENETUNREACH)) } }; } impl Dialer { /// Instantiate a new [`Dialer`] with the given [`Url`] and datastore path. pub async fn new(endpoint: Url, datastore: Option) -> io::Result { match endpoint.scheme().to_lowercase().as_str() { "tcp" => { // Build a TCP dialer enforce_hostport!(endpoint); let variant = tcp::TcpDialer::new(None).await?; let variant = DialerVariant::Tcp(variant); Ok(Self { endpoint, variant }) } "tcp+tls" => { // Build a TCP dialer wrapped with TLS enforce_hostport!(endpoint); let variant = tcp::TcpDialer::new(None).await?; let variant = DialerVariant::TcpTls(variant); Ok(Self { endpoint, variant }) } #[cfg(feature = "p2p-tor")] "tor" => { // Build a Tor dialer enforce_hostport!(endpoint); let variant = tor::TorDialer::new(datastore).await?; let variant = DialerVariant::Tor(variant); Ok(Self { endpoint, variant }) } #[cfg(feature = "p2p-tor")] "tor+tls" => { // Build a Tor dialer wrapped with TLS enforce_hostport!(endpoint); let variant = tor::TorDialer::new(datastore).await?; let variant = DialerVariant::TorTls(variant); Ok(Self { endpoint, variant }) } #[cfg(feature = "p2p-nym")] "nym" => { // Build a Nym dialer enforce_hostport!(endpoint); let variant = nym::NymDialer::new().await?; let variant = DialerVariant::Nym(variant); Ok(Self { endpoint, variant }) } #[cfg(feature = "p2p-nym")] "nym+tls" => { // Build a Nym dialer wrapped with TLS enforce_hostport!(endpoint); let variant = nym::NymDialer::new().await?; let variant = DialerVariant::NymTls(variant); Ok(Self { endpoint, variant }) } "unix" => { // Build a Unix socket dialer enforce_abspath!(endpoint); let variant = unix::UnixDialer::new().await?; let variant = DialerVariant::Unix(variant); Ok(Self { endpoint, variant }) } x => { error!("[P2P] Requested unsupported transport: {}", x); Err(io::Error::from_raw_os_error(libc::ENETUNREACH)) } } } /// Dial an instantiated [`Dialer`]. This creates a connection and returns a stream. /// The Tor-based Dialer variants can panic: this is intended. There exists validation /// for hosts and ports in other parts of the codebase. A panic occurring here /// likely indicates a configuration issue on the part of the user. It is preferable /// in this case that the user is alerted to this problem via a panic. pub async fn dial(&self, timeout: Option) -> io::Result> { match &self.variant { DialerVariant::Tcp(dialer) => { // NOTE: sockaddr here is an array, can contain both ipv4 and ipv6 let sockaddr = self.endpoint.socket_addrs(|| None)?; let stream = dialer.do_dial(sockaddr[0], timeout).await?; Ok(Box::new(stream)) } DialerVariant::TcpTls(dialer) => { let sockaddr = self.endpoint.socket_addrs(|| None)?; let stream = dialer.do_dial(sockaddr[0], timeout).await?; let tlsupgrade = tls::TlsUpgrade::new().await; let stream = tlsupgrade.upgrade_dialer_tls(stream).await?; Ok(Box::new(stream)) } #[cfg(feature = "p2p-tor")] DialerVariant::Tor(dialer) => { let host = self.endpoint.host_str().unwrap(); let port = self.endpoint.port().unwrap(); let stream = dialer.do_dial(host, port, timeout).await?; Ok(Box::new(stream)) } #[cfg(feature = "p2p-tor")] DialerVariant::TorTls(dialer) => { let host = self.endpoint.host_str().unwrap(); let port = self.endpoint.port().unwrap(); let stream = dialer.do_dial(host, port, timeout).await?; let tlsupgrade = tls::TlsUpgrade::new().await; let stream = tlsupgrade.upgrade_dialer_tls(stream).await?; Ok(Box::new(stream)) } #[cfg(feature = "p2p-nym")] DialerVariant::Nym(_dialer) => { todo!(); } #[cfg(feature = "p2p-nym")] DialerVariant::NymTls(_dialer) => { todo!(); } DialerVariant::Unix(dialer) => { let path = match self.endpoint.to_file_path() { Ok(v) => v, Err(_) => return Err(io::Error::new(ErrorKind::Unsupported, "Invalid path")), }; let stream = dialer.do_dial(path).await?; Ok(Box::new(stream)) } } } /// Return a reference to the `Dialer` endpoint pub fn endpoint(&self) -> &Url { &self.endpoint } } /// A listener that is able to transparently listen over arbitrary transports. pub struct Listener { /// The address to open the listener on endpoint: Url, /// The listener variant (transport protocol) variant: ListenerVariant, } impl Listener { /// Instantiate a new [`Listener`] with the given [`Url`] and datastore path. /// Must contain a scheme, host string, and a port. pub async fn new(endpoint: Url, datastore: Option) -> io::Result { match endpoint.scheme().to_lowercase().as_str() { "tcp" => { // Build a TCP listener enforce_hostport!(endpoint); let variant = tcp::TcpListener::new(1024).await?; let variant = ListenerVariant::Tcp(variant); Ok(Self { endpoint, variant }) } "tcp+tls" => { // Build a TCP listener wrapped with TLS enforce_hostport!(endpoint); let variant = tcp::TcpListener::new(1024).await?; let variant = ListenerVariant::TcpTls(variant); Ok(Self { endpoint, variant }) } #[cfg(feature = "p2p-tor")] "tor" => { // Build a Tor Hidden Service listener enforce_hostport!(endpoint); let variant = tor::TorListener::new(datastore).await?; let variant = ListenerVariant::Tor(variant); Ok(Self { endpoint, variant }) } "unix" => { enforce_abspath!(endpoint); let variant = unix::UnixListener::new().await?; let variant = ListenerVariant::Unix(variant); Ok(Self { endpoint, variant }) } x => { error!("[P2P] Requested unsupported transport: {}", x); Err(io::Error::from_raw_os_error(libc::ENETUNREACH)) } } } /// Listen on an instantiated [`Listener`]. /// This will open a socket and return the listener. pub async fn listen(&self) -> io::Result> { match &self.variant { ListenerVariant::Tcp(listener) => { let sockaddr = self.endpoint.socket_addrs(|| None)?; let l = listener.do_listen(sockaddr[0]).await?; Ok(Box::new(l)) } ListenerVariant::TcpTls(listener) => { let sockaddr = self.endpoint.socket_addrs(|| None)?; let l = listener.do_listen(sockaddr[0]).await?; let tlsupgrade = tls::TlsUpgrade::new().await; let l = tlsupgrade.upgrade_listener_tcp_tls(l).await?; Ok(Box::new(l)) } #[cfg(feature = "p2p-tor")] ListenerVariant::Tor(listener) => { let port = self.endpoint.port().unwrap(); let l = listener.do_listen(port).await?; Ok(Box::new(l)) } ListenerVariant::Unix(listener) => { let path = match self.endpoint.to_file_path() { Ok(v) => v, Err(_) => return Err(io::Error::new(ErrorKind::Unsupported, "Invalid path")), }; let l = listener.do_listen(&path).await?; Ok(Box::new(l)) } } } pub async fn endpoint(&self) -> Url { match &self.variant { #[cfg(feature = "p2p-tor")] ListenerVariant::Tor(listener) => listener.endpoint.lock().await.clone().unwrap(), _ => self.endpoint.clone(), } } } /// Wrapper trait for async streams pub trait PtStream: AsyncRead + AsyncWrite + Unpin + Send {} impl PtStream for smol::net::TcpStream {} impl PtStream for futures_rustls::TlsStream {} #[cfg(feature = "p2p-tor")] impl PtStream for arti_client::DataStream {} #[cfg(feature = "p2p-tor")] impl PtStream for futures_rustls::TlsStream {} impl PtStream for smol::net::unix::UnixStream {} /// Wrapper trait for async listeners #[async_trait] pub trait PtListener: Send + Unpin { async fn next(&self) -> io::Result<(Box, Url)>; }