fixup: enable user/password auth, add tests

Author: tankyleo

Cargo.toml

@@ -67,4 +67,5 @@ check-cfg = [
     "cfg(require_route_graph_test)",
     "cfg(simple_close)",
     "cfg(peer_storage)",
+    "cfg(tor_socks5)",
 ]

lightning-net-tokio/src/lib.rs

@@ -477,12 +477,14 @@ where
 /// 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,
+	user_pass: Option<(&str, &str)>,
 ) -> 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()) };
+	let connect_fut = async {
+		socks5_connect(socks5_proxy_addr, addr, user_pass).await.map(|s| s.into_std().unwrap())
+	};
 	if let Ok(Ok(stream)) =
 		time::timeout(Duration::from_secs(SOCKS5_CONNECT_OUTBOUND_TIMEOUT), connect_fut).await
 	{
@@ -493,23 +495,28 @@ where
 }
 
 async fn socks5_connect(
-	socks5_proxy_addr: SocketAddr, addr: SocketAddress,
+	socks5_proxy_addr: SocketAddr, addr: SocketAddress, user_pass: Option<(&str, &str)>,
 ) -> Result<TcpStream, ()> {
 	use tokio::io::AsyncReadExt;
 	use tokio::io::AsyncWriteExt;
 
-	// Constants defined in RFC 1928
+	// Constants defined in RFC 1928 and RFC 1929
 	const VERSION: u8 = 5;
 	const NMETHODS: u8 = 1;
 	const NO_AUTH: u8 = 0;
-	const METHOD_SELECT_RES_LEN: usize = 2;
+	const USERNAME_PASSWORD_AUTH: u8 = 2;
+	const METHOD_SELECT_REPLY_LEN: usize = 2;
+	const USERNAME_PASSWORD_VERSION: u8 = 1;
+	const USERNAME_PASSWORD_REPLY_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;
+	const SUCCESS: u8 = 0;
 
+	const USERNAME_PASSWORD_REQUEST_MAX_LEN: usize = 1 /* VER */ + 1 /* ULEN */ + 255 /* UNAME max len */
+		+ 1 /* PLEN */ + 255 /* PASSWD max len */;
 	const SOCKS5_REQUEST_REPLY_MAX_LEN: usize = 1 /* VER */ + 1 /* CMD for request, REP for reply */ + 1 /* RSV */
 		+ 1 /* ATYP */ + 1 /* HOSTNAME len */ + 255 /* HOSTNAME */ + 2 /* PORT */;
 	const SOCKS5_REQUEST_REPLY_MIN_LEN: usize = 1 /* VER */ + 1 /* CMD for request, REP for reply */ + 1 /* RSV */
@@ -521,33 +528,62 @@ async fn socks5_connect(
 	const SOCKS5_HOSTNAME_REQUEST_STATIC_FIELDS_LEN: usize = 1 /* VER */ + 1 /* CMD */ + 1 /* RSV */ + 1 /* ATYP */
 		+ 1 /* HOSTNAME len */ + 2 /* DST.PORT */;
 
-	let method_selection_message = [VERSION, NMETHODS, NO_AUTH];
+	let selected_auth = if user_pass.is_some() { USERNAME_PASSWORD_AUTH } else { NO_AUTH };
+	let method_selection_request = [VERSION, NMETHODS, selected_auth];
 	let mut tcp_stream = TcpStream::connect(&socks5_proxy_addr).await.map_err(|_| ())?;
-	tcp_stream.write_all(&method_selection_message).await.map_err(|_| ())?;
+	tcp_stream.write_all(&method_selection_request).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] {
+	let mut method_selection_reply = [0u8; METHOD_SELECT_REPLY_LEN];
+	let n_read = tcp_stream.read_exact(&mut method_selection_reply).await.map_err(|_| ())?;
+	if n_read != METHOD_SELECT_REPLY_LEN || method_selection_reply != [VERSION, selected_auth] {
 		return Err(());
 	}
 
-	let mut socks5_request = [0u8; SOCKS5_REQUEST_REPLY_MAX_LEN];
-	let request_size;
+	if let Some((username, password)) = user_pass {
+		if username.len() > 255 || password.len() > 255 {
+			return Err(());
+		}
 
+		let mut username_password_request = [0u8; USERNAME_PASSWORD_REQUEST_MAX_LEN];
+		username_password_request[0] = USERNAME_PASSWORD_VERSION;
+		username_password_request[1] = username.len() as u8;
+		username_password_request[2..2 + username.len()].copy_from_slice(username.as_bytes());
+		let password_len_pos = 2 + username.len();
+		username_password_request[password_len_pos] = password.len() as u8;
+		let password_pos = password_len_pos + 1;
+		username_password_request[password_pos..password_pos + password.len()]
+			.copy_from_slice(password.as_bytes());
+		let username_password_request_len = password_pos + password.len();
+		tcp_stream
+			.write_all(&username_password_request[..username_password_request_len])
+			.await
+			.map_err(|_| ())?;
+
+		let mut username_password_reply = [0u8; USERNAME_PASSWORD_REPLY_LEN];
+		let n_read = tcp_stream.read_exact(&mut username_password_reply).await.map_err(|_| ())?;
+		if n_read != USERNAME_PASSWORD_REPLY_LEN
+			|| username_password_reply != [USERNAME_PASSWORD_VERSION, SUCCESS]
+		{
+			return Err(());
+		}
+	}
+
+	let mut socks5_request = [0u8; SOCKS5_REQUEST_REPLY_MAX_LEN];
+	let socks5_request_len;
 	socks5_request[0..3].copy_from_slice(&[VERSION, CMD_CONNECT, RSV]);
 
 	match addr {
 		SocketAddress::TcpIpV4 { addr, port } => {
 			socks5_request[3] = ATYP_IPV4;
 			socks5_request[4..8].copy_from_slice(&addr);
 			socks5_request[8..10].copy_from_slice(&port.to_be_bytes());
-			request_size = SOCKS5_IPV4_REQUEST_LEN;
+			socks5_request_len = SOCKS5_IPV4_REQUEST_LEN;
 		},
 		SocketAddress::TcpIpV6 { addr, port } => {
 			socks5_request[3] = ATYP_IPV6;
 			socks5_request[4..20].copy_from_slice(&addr);
 			socks5_request[20..22].copy_from_slice(&port.to_be_bytes());
-			request_size = SOCKS5_IPV6_REQUEST_LEN;
+			socks5_request_len = SOCKS5_IPV6_REQUEST_LEN;
 		},
 		ref onion_v3 @ SocketAddress::OnionV3 { port, .. } => {
 			let onion_v3_url = onion_v3.to_string();
@@ -557,47 +593,46 @@ async fn socks5_connect(
 			let port_index = 5 + hostname.len();
 			socks5_request[5..port_index].copy_from_slice(hostname);
 			socks5_request[port_index..port_index + 2].copy_from_slice(&port.to_be_bytes());
-			request_size = SOCKS5_HOSTNAME_REQUEST_STATIC_FIELDS_LEN + hostname.len();
+			socks5_request_len = SOCKS5_HOSTNAME_REQUEST_STATIC_FIELDS_LEN + hostname.len();
 		},
 		SocketAddress::Hostname { hostname, port } => {
 			socks5_request[3] = ATYP_DOMAINNAME;
 			socks5_request[4] = hostname.len();
 			let port_index = 5 + hostname.len() as usize;
 			socks5_request[5..port_index].copy_from_slice(hostname.as_bytes());
 			socks5_request[port_index..port_index + 2].copy_from_slice(&port.to_be_bytes());
-			request_size = SOCKS5_HOSTNAME_REQUEST_STATIC_FIELDS_LEN + hostname.len() as usize;
+			socks5_request_len =
+				SOCKS5_HOSTNAME_REQUEST_STATIC_FIELDS_LEN + hostname.len() as usize;
 		},
 		SocketAddress::OnionV2 { .. } => return Err(()),
 	};
 
-	tcp_stream.write_all(&socks5_request[..request_size]).await.map_err(|_| ())?;
+	tcp_stream.write_all(&socks5_request[..socks5_request_len]).await.map_err(|_| ())?;
 
 	let mut buffer = [0u8; SOCKS5_REQUEST_REPLY_MAX_LEN];
 	// First pull some bytes into the buffer
 	let mut total_read = tcp_stream.read(&mut buffer).await.map_err(|_| ())?;
 	// Then make sure we've reached EOF, otherwise keep appending until the max size of the buffer
 	while total_read < SOCKS5_REQUEST_REPLY_MAX_LEN {
 		let n_read = match tcp_stream.try_read(&mut buffer[total_read..]) {
+			Ok(0) => return Err(()), // The read half of the TcpStream has been closed
 			Ok(n) => n,
-			Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => 0,
+			Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
 			Err(_e) => return Err(()),
 		};
-		if n_read == 0 {
-			break;
-		}
+
 		total_read += n_read;
 	}
-	// If we've filled the buffer, return `Err` if we receive more bytes than the max allowed by the RFC
+
+	// If we've filled the buffer and we receive more bytes, return `Err` as these bytes are not expected per the RFC
 	if total_read == SOCKS5_REQUEST_REPLY_MAX_LEN {
 		match tcp_stream.try_read(&mut [0u8; 1]) {
-			Ok(0) => (),
-			Ok(1..) => return Err(()),
 			Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => (),
-			Err(_e) => return Err(()),
+			_ => return Err(()),
 		}
 	}
 
-	if total_read < SOCKS5_REQUEST_REPLY_MIN_LEN || buffer[..3] != [VERSION, SUCCEEDED, RSV] {
+	if total_read < SOCKS5_REQUEST_REPLY_MIN_LEN || buffer[..3] != [VERSION, SUCCESS, RSV] {
 		return Err(());
 	}
 
@@ -743,6 +778,9 @@ impl Hash for SocketDescriptor {
 
 #[cfg(test)]
 mod tests {
+	#[cfg(tor_socks5)]
+	use super::socks5_connect;
+
 	use bitcoin::constants::ChainHash;
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
 	use bitcoin::Network;
@@ -756,6 +794,8 @@ mod tests {
 	use tokio::sync::mpsc;
 
 	use std::mem;
+	#[cfg(tor_socks5)]
+	use std::net::SocketAddr;
 	use std::sync::atomic::{AtomicBool, Ordering};
 	use std::sync::{Arc, Mutex};
 	use std::time::Duration;
@@ -1076,4 +1116,66 @@ mod tests {
 	async fn unthreaded_race_disconnect_accept() {
 		race_disconnect_accept().await;
 	}
+
+	#[cfg(tor_socks5)]
+	#[tokio::test]
+	async fn test_socks5_connect() {
+		// Set TOR_SOCKS5_PROXY=127.0.0.1:9050
+		let socks5_proxy_addr: SocketAddr = std::env!("TOR_SOCKS5_PROXY").parse().unwrap();
+
+		// Success cases
+
+		for (addr_str, user_pass) in [
+			// google.com
+			("142.250.189.196:80", None),
+			// google.com
+			("[2607:f8b0:4005:813::2004]:80", None),
+			// torproject.org
+			("torproject.org:80", None),
+			// torproject.org
+			("2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion:80", None),
+			// Same vectors as above, with a username and password
+			("142.250.189.196:80", Some(("<torS0X>0", ""))),
+			("[2607:f8b0:4005:813::2004]:80", Some(("<torS0X>0", "123"))),
+			("torproject.org:80", Some(("<torS0X>1abc", ""))),
+			(
+				"2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion:80",
+				Some(("<torS0X>1abc", "123")),
+			),
+		] {
+			let addr: SocketAddress = addr_str.parse().unwrap();
+			let _tcp_stream = socks5_connect(socks5_proxy_addr, addr, user_pass).await.unwrap();
+		}
+
+		// Failure cases
+
+		for (addr_str, user_pass) in [
+			// google.com, with some invalid port
+			("142.250.189.196:1234", None),
+			// google.com, with some invalid port
+			("[2607:f8b0:4005:813::2004]:1234", None),
+			// torproject.org, with some invalid port
+			("torproject.org:1234", None),
+			// torproject.org, with a typo
+			("3gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion:80", None),
+			// Same vectors as above, with a username and password
+			("142.250.189.196:1234", Some(("<torS0X>0", ""))),
+			("[2607:f8b0:4005:813::2004]:1234", Some(("<torS0X>0", "123"))),
+			("torproject.org:1234", Some(("<torS0X>1abc", ""))),
+			(
+				"3gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion:80",
+				Some(("<torS0X>1abc", "123")),
+			),
+			/* TODO: Uncomment when format types 30 and 31 land in tor stable, see https://spec.torproject.org/socks-extensions.html,
+			   these are invalid usernames according to those standards.
+			("142.250.189.196:80", Some(("<torS0X>0abc", "123"))),
+			("[2607:f8b0:4005:813::2004]:80", Some(("<torS0X>1", "123"))),
+			("torproject.org:80", Some(("<torS0X>9", "123"))),
+			("2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion:80", Some(("<torS0X>", "123"))),
+			*/
+		] {
+			let addr: SocketAddress = addr_str.parse().unwrap();
+			assert!(socks5_connect(socks5_proxy_addr, addr, user_pass).await.is_err());
+		}
+	}
 }