net-tokio: add fn socks5_connect_outbound

Author: tankyleo

lightning-net-tokio/Cargo.toml

@@ -19,7 +19,7 @@ rustdoc-args = ["--cfg", "docsrs"]
 [dependencies]
 bitcoin = "0.32.2"
 lightning = { version = "0.3.0", path = "../lightning" }
-tokio = { version = "1.35", features = [ "rt", "sync", "net", "time" ] }
+tokio = { version = "1.35", features = [ "rt", "sync", "net", "time", "io-util" ] }
 
 [dev-dependencies]
 tokio = { version = "1.35", features = [ "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }

lightning-net-tokio/src/lib.rs

@@ -51,6 +51,9 @@ use std::time::Duration;
 
 static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
 
+const CONNECT_OUTBOUND_TIMEOUT: u64 = 10;
+const SOCKS5_CONNECT_OUTBOUND_TIMEOUT: u64 = 30;
+
 // We only need to select over multiple futures in one place, and taking on the full `tokio/macros`
 // dependency tree in order to do so (which has broken our MSRV before) is excessive. Instead, we
 // define a trivial two- and three- select macro with the specific types we need and just use that.
@@ -462,13 +465,118 @@ where
 	PM::Target: APeerManager<Descriptor = SocketDescriptor>,
 {
 	let connect_fut = async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) };
-	if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), connect_fut).await {
+	if let Ok(Ok(stream)) =
+		time::timeout(Duration::from_secs(CONNECT_OUTBOUND_TIMEOUT), connect_fut).await
+	{
+		Some(setup_outbound(peer_manager, their_node_id, stream))
+	} else {
+		None
+	}
+}
+
+/// Same as [`connect_outbound`], using a SOCKS5 proxy
+pub async fn socks5_connect_outbound<PM: Deref + 'static + Send + Sync + Clone>(
+	peer_manager: PM, their_node_id: PublicKey, socks5_proxy_addr: SocketAddr, addr: SocketAddress,
+) -> Option<impl std::future::Future<Output = ()>>
+where
+	PM::Target: APeerManager<Descriptor = SocketDescriptor>,
+{
+	let connect_fut =
+		async { socks5_connect(socks5_proxy_addr, addr).await.map(|s| s.into_std().unwrap()) };
+	if let Ok(Ok(stream)) =
+		time::timeout(Duration::from_secs(SOCKS5_CONNECT_OUTBOUND_TIMEOUT), connect_fut).await
+	{
 		Some(setup_outbound(peer_manager, their_node_id, stream))
 	} else {
 		None
 	}
 }
 
+async fn socks5_connect(
+	socks5_proxy_addr: SocketAddr, addr: SocketAddress,
+) -> Result<TcpStream, ()> {
+	use tokio::io::AsyncReadExt;
+	use tokio::io::AsyncWriteExt;
+
+	// Constants defined in RFC 1928
+	const VERSION: u8 = 5;
+	const NMETHODS: u8 = 1;
+	const NO_AUTH: u8 = 0;
+	const METHOD_SELECT_RES_LEN: usize = 2;
+	const CMD_CONNECT: u8 = 1;
+	const RSV: u8 = 0;
+	const ATYP_IPV4: u8 = 1;
+	const ATYP_DOMAINNAME: u8 = 3;
+	const ATYP_IPV6: u8 = 4;
+	const SUCCEEDED: u8 = 0;
+
+	// The size of a socks5 request for an onion V3 address
+	const SOCKS5_REQUEST_CAPACITY: usize = 69;
+	// The minimum size of a socks5 response, also the size of the response from a Tor proxy
+	const SOCKS5_REPLY_MIN_LEN: usize = 10;
+
+	if let SocketAddress::OnionV2 { .. } = addr {
+		return Err(());
+	}
+
+	let method_selection_message = [VERSION, NMETHODS, NO_AUTH];
+	let mut tcp_stream = TcpStream::connect(&socks5_proxy_addr).await.map_err(|_| ())?;
+	tcp_stream.write_all(&method_selection_message).await.map_err(|_| ())?;
+
+	let mut method_selection_response = [0u8; METHOD_SELECT_RES_LEN];
+	let n_read = tcp_stream.read_exact(&mut method_selection_response).await.map_err(|_| ())?;
+	if n_read != METHOD_SELECT_RES_LEN || method_selection_response != [VERSION, NO_AUTH] {
+		return Err(());
+	}
+
+	let mut socks5_request: Vec<u8> = Vec::with_capacity(SOCKS5_REQUEST_CAPACITY);
+
+	socks5_request.push(VERSION);
+	socks5_request.push(CMD_CONNECT);
+	socks5_request.push(RSV);
+
+	socks5_request.push(match addr {
+		SocketAddress::TcpIpV4 { .. } => ATYP_IPV4,
+		SocketAddress::TcpIpV6 { .. } => ATYP_IPV6,
+		SocketAddress::OnionV3 { .. } | SocketAddress::Hostname { .. } => ATYP_DOMAINNAME,
+		SocketAddress::OnionV2 { .. } => unreachable!(),
+	});
+
+	match addr {
+		SocketAddress::TcpIpV4 { addr, port } => {
+			socks5_request.extend_from_slice(&addr);
+			socks5_request.extend_from_slice(&port.to_be_bytes());
+		},
+		SocketAddress::TcpIpV6 { addr, port } => {
+			socks5_request.extend_from_slice(&addr);
+			socks5_request.extend_from_slice(&port.to_be_bytes());
+		},
+		onion_v3 @ SocketAddress::OnionV3 { port, .. } => {
+			let onion_v3_url = onion_v3.to_string();
+			let hostname = onion_v3_url.split_once(':').ok_or(())?.0.as_bytes();
+			socks5_request.extend_from_slice(&[hostname.len() as u8]);
+			socks5_request.extend_from_slice(hostname);
+			socks5_request.extend_from_slice(&port.to_be_bytes());
+		},
+		SocketAddress::Hostname { hostname, port } => {
+			socks5_request.extend_from_slice(&[hostname.len() as u8]);
+			socks5_request.extend_from_slice(hostname.as_bytes());
+			socks5_request.extend_from_slice(&port.to_be_bytes());
+		},
+		SocketAddress::OnionV2 { .. } => unreachable!(),
+	};
+
+	tcp_stream.write_all(&socks5_request).await.map_err(|_| ())?;
+
+	let mut buffer: Vec<u8> = Vec::with_capacity(SOCKS5_REPLY_MIN_LEN);
+	let n_read = tcp_stream.read_to_end(&mut buffer).await.map_err(|_| ())?;
+	if n_read < SOCKS5_REPLY_MIN_LEN || buffer[..3] != [VERSION, SUCCEEDED, RSV] {
+		return Err(());
+	}
+
+	Ok(tcp_stream)
+}
+
 const SOCK_WAKER_VTABLE: task::RawWakerVTable = task::RawWakerVTable::new(
 	clone_socket_waker,
 	wake_socket_waker,