andy/nexus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

refactor: one blocking SOCKS5/HTTP CONNECT proxy module shared by all TCP clients

Browse Source 
SSH's async SOCKS5 and HTTP CONNECT code moves to clients::proxy as a
blocking dialer (SSH runs it on a blocking thread), so RDP and others can
use the same implementation. The proxy handshake now has a read timeout,
and new tests drive both proxy types against fake proxies, including
SOCKS5 username/password auth and a refused HTTP CONNECT (407).
This commit is contained in:
Andy Dixon
2026-10-12 10:15:36 +01:00
parent 4bd12bfb03
commit 6a4d7381d9
3 changed files with 236 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/core/src/clients/mod.rs
+1
View File
@@ -37,6 +37,7 @@ impl Callbacks {
}
}
pub mod proxy;
pub mod raw;
pub mod serial;
pub mod telnet;
rust/core/src/clients/proxy.rs
+224
View File
@@ -0,0 +1,224 @@
//! Outbound proxies (SOCKS5, HTTP CONNECT) for the clients that dial TCP themselves: SSH and RDP.
//! Blocking; async callers run it on a blocking thread.
use crate::Result;
use base64::Engine;
use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::Duration;
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Proxy {
/// "socks5" or "http"; anything else (including "none" and "") dials directly.
pub kind: String,
pub host: String,
pub port: u16,
pub username: String,
pub password: String,
}
impl Proxy {
pub fn is_direct(&self) -> bool {
!matches!(self.kind.to_lowercase().as_str(), "socks5" | "http")
}
/// A TCP stream to `host:port`, through the proxy when one is configured.
pub fn connect(&self, host: &str, port: u16, timeout: Duration) -> Result<TcpStream> {
match self.kind.to_lowercase().as_str() {
"socks5" => self.socks5(host, port, timeout),
"http" => self.http(host, port, timeout),
_ => tcp(host, port, timeout).map_err(|e| format!("dial tcp {host}:{port}: {e}").into()),
}
}
fn open(&self, default_port: u16, what: &str, timeout: Duration) -> Result<TcpStream> {
let port = if self.port == 0 { default_port } else { self.port };
let s = tcp(&self.host, port, timeout).map_err(|e| format!("{what} proxy connection failed: {e}"))?;
// The handshake must not hang on a silent proxy either.
s.set_read_timeout(Some(timeout))?;
Ok(s)
}
fn socks5(&self, host: &str, port: u16, timeout: Duration) -> Result<TcpStream> {
let mut s = self.open(1080, "SOCKS5", timeout)?;
let auth = !self.username.is_empty();
s.write_all(if auth { &[5, 2, 0, 2] } else { &[5, 1, 0] })?;
let mut r = [0u8; 2];
s.read_exact(&mut r)?;
match r[1] {
0 => {}
2 if auth => {
s.write_all(&socks5_auth_request(&self.username, &self.password)?)?;
s.read_exact(&mut r)?;
if r[1] != 0 {
return Err("SOCKS5 proxy: authentication failed".into());
}
}
_ => return Err("SOCKS5 proxy: no acceptable authentication method".into()),
}
s.write_all(&socks5_connect_request(host, port)?)?;
let mut head = [0u8; 4];
s.read_exact(&mut head)?;
if head[1] != 0 {
return Err(format!("SOCKS5 proxy connection to {host}:{port} failed: reply code {}", head[1]).into());
}
let addr_len = match head[3] {
1 => 4,
4 => 16,
3 => {
let mut n = [0u8; 1];
s.read_exact(&mut n)?;
n[0] as usize
}
_ => 0,
};
let mut rest = vec![0u8; addr_len + 2];
s.read_exact(&mut rest)?;
s.set_read_timeout(None)?;
Ok(s)
}
fn http(&self, host: &str, port: u16, timeout: Duration) -> Result<TcpStream> {
let mut s = self.open(8080, "HTTP", timeout)?;
let mut req = format!("CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n");
if !self.username.is_empty() {
let cred = base64::engine::general_purpose::STANDARD.encode(format!("{}:{}", self.username, self.password));
req += &format!("Proxy-Authorization: Basic {cred}\r\n");
}
req += "\r\n";
s.write_all(req.as_bytes())?;
// Read the response head byte by byte so no tunnelled bytes are consumed.
let mut head = Vec::new();
let mut b = [0u8; 1];
while !head.ends_with(b"\r\n\r\n") {
s.read_exact(&mut b).map_err(|e| format!("HTTP CONNECT response read failed: {e}"))?;
head.push(b[0]);
if head.len() > 16 * 1024 {
return Err("HTTP CONNECT response too large".into());
}
}
let status = String::from_utf8_lossy(&head).lines().next().unwrap_or("").to_string();
if status.split_whitespace().nth(1) != Some("200") {
return Err(format!("HTTP CONNECT proxy returned {status}").into());
}
s.set_read_timeout(None)?;
Ok(s)
}
}
fn tcp(host: &str, port: u16, timeout: Duration) -> std::io::Result<TcpStream> {
use std::net::ToSocketAddrs;
let mut last = std::io::Error::new(std::io::ErrorKind::NotFound, "no addresses found");
for sa in (host, port).to_socket_addrs()? {
match TcpStream::connect_timeout(&sa, timeout) {
Ok(s) => return Ok(s),
Err(e) => last = e,
}
}
Err(last)
}
/// RFC 1929 username/password sub-negotiation. Both fields carry a one-byte length, so longer
/// values are rejected rather than silently truncated into a malformed request.
fn socks5_auth_request(user: &str, pass: &str) -> Result<Vec<u8>> {
let ulen = u8::try_from(user.len()).map_err(|_| "SOCKS5 proxy: username longer than 255 bytes")?;
let plen = u8::try_from(pass.len()).map_err(|_| "SOCKS5 proxy: password longer than 255 bytes")?;
let mut m = vec![1, ulen];
m.extend(user.as_bytes());
m.push(plen);
m.extend(pass.as_bytes());
Ok(m)
}
/// RFC 1928 CONNECT with a domain-name address (one-byte length).
fn socks5_connect_request(host: &str, port: u16) -> Result<Vec<u8>> {
let hlen = u8::try_from(host.len()).map_err(|_| "SOCKS5 proxy: host name longer than 255 bytes")?;
let mut req = vec![5, 1, 0, 3, hlen];
req.extend(host.as_bytes());
req.extend(port.to_be_bytes());
Ok(req)
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::TcpListener;
#[test]
fn socks5_encoding() {
assert_eq!(socks5_connect_request("example.com", 22).unwrap(), [&[5u8, 1, 0, 3, 11][..], b"example.com", &[0, 22]].concat());
assert!(socks5_connect_request(&"h".repeat(256), 22).is_err());
assert_eq!(socks5_auth_request("u", "pw").unwrap(), [1, 1, b'u', 2, b'p', b'w']);
assert!(socks5_auth_request(&"u".repeat(256), "").is_err());
}
/// A one-shot proxy that checks the handshake, then echoes.
fn fake_proxy(handshake: impl FnOnce(&mut TcpStream) + Send + 'static) -> u16 {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
let (mut s, _) = l.accept().unwrap();
handshake(&mut s);
let mut buf = [0u8; 4];
s.read_exact(&mut buf).unwrap();
s.write_all(&buf).unwrap();
});
port
}
fn echo_through(p: &Proxy) {
let mut s = p.connect("target.example", 3389, Duration::from_secs(5)).unwrap();
s.write_all(b"ping").unwrap();
let mut buf = [0u8; 4];
s.read_exact(&mut buf).unwrap();
assert_eq!(&buf, b"ping");
}
#[test]
fn socks5_with_auth_reaches_the_target() {
let port = fake_proxy(|s| {
let mut b = [0u8; 4];
s.read_exact(&mut b).unwrap();
assert_eq!(b, [5, 2, 0, 2]);
s.write_all(&[5, 2]).unwrap();
let mut auth = [0u8; 7];
s.read_exact(&mut auth).unwrap();
assert_eq!(auth, [1, 1, b'u', 3, b'p', b'w', b'd']);
s.write_all(&[1, 0]).unwrap();
let mut req = vec![0u8; 5 + 14 + 2];
s.read_exact(&mut req).unwrap();
assert_eq!(&req[5..19], b"target.example");
assert_eq!(&req[19..], &3389u16.to_be_bytes());
s.write_all(&[5, 0, 0, 1, 127, 0, 0, 1, 0, 0]).unwrap();
});
echo_through(&Proxy { kind: "socks5".into(), host: "127.0.0.1".into(), port, username: "u".into(), password: "pwd".into() });
}
#[test]
fn http_connect_sends_basic_auth_and_needs_200() {
let port = fake_proxy(|s| {
let mut head = Vec::new();
let mut b = [0u8; 1];
while !head.ends_with(b"\r\n\r\n") {
s.read_exact(&mut b).unwrap();
head.push(b[0]);
}
let head = String::from_utf8(head).unwrap();
assert!(head.starts_with("CONNECT target.example:3389 HTTP/1.1\r\n"), "{head}");
assert!(head.contains("Proxy-Authorization: Basic dTpwd2Q=\r\n"), "{head}");
s.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").unwrap();
});
echo_through(&Proxy { kind: "http".into(), host: "127.0.0.1".into(), port, username: "u".into(), password: "pwd".into() });
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
let (mut s, _) = l.accept().unwrap();
let mut buf = [0u8; 256];
let _ = s.read(&mut buf);
s.write_all(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n").unwrap();
});
let err = Proxy { kind: "http".into(), host: "127.0.0.1".into(), port, ..Default::default() }.connect("t", 1, Duration::from_secs(5)).unwrap_err();
assert!(err.to_string().contains("407"), "{err}");
}
}
rust/core/src/clients/ssh.rs
+11 -98
View File
@@ -13,7 +13,7 @@ use russh::{Channel, ChannelMsg, Pty};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::mpsc;
use zeroize::Zeroizing;
@@ -226,95 +226,17 @@ async fn blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) ->
}
async fn dial(sc: &SshConfig, proxy_password: &str, host: &str, port: u16, timeout: Duration) -> Result<tokio::net::TcpStream> {
let fut = async {
match sc.proxy_type.to_lowercase().as_str() {
"socks5" => socks5(sc, proxy_password, host, port).await,
"http" => http_connect(sc, proxy_password, host, port).await,
_ => Ok(tokio::net::TcpStream::connect((host, port)).await?),
}
let proxy = super::proxy::Proxy {
kind: sc.proxy_type.clone(),
host: sc.proxy_host.clone(),
port: sc.proxy_port as u16,
username: sc.proxy_username.clone(),
password: proxy_password.into(),
};
tokio::time::timeout(timeout, fut).await.map_err(|_| format!("dial tcp {host}:{port}: i/o timeout"))?
}
async fn socks5(sc: &SshConfig, proxy_password: &str, host: &str, port: u16) -> Result<tokio::net::TcpStream> {
let pport = if sc.proxy_port == 0 { 1080 } else { sc.proxy_port as u16 };
let mut s = tokio::net::TcpStream::connect((sc.proxy_host.as_str(), pport)).await.map_err(|e| format!("SOCKS5 proxy setup failed: {e}"))?;
let auth = !sc.proxy_username.is_empty();
s.write_all(if auth { &[5, 2, 0, 2] } else { &[5, 1, 0] }).await?;
let mut r = [0u8; 2];
s.read_exact(&mut r).await?;
match r[1] {
0 => {}
2 if auth => {
s.write_all(&socks5_auth_request(&sc.proxy_username, proxy_password)?).await?;
s.read_exact(&mut r).await?;
if r[1] != 0 {
return Err("SOCKS5 proxy: authentication failed".into());
}
}
_ => return Err("SOCKS5 proxy: no acceptable authentication method".into()),
}
s.write_all(&socks5_connect_request(host, port)?).await?;
let mut head = [0u8; 4];
s.read_exact(&mut head).await?;
if head[1] != 0 {
return Err(format!("SOCKS5 proxy connection to {host}:{port} failed: reply code {}", head[1]).into());
}
let skip = match head[3] {
1 => 4,
4 => 16,
3 => s.read_u8().await? as usize,
_ => 0,
} + 2;
let mut rest = vec![0u8; skip];
s.read_exact(&mut rest).await?;
Ok(s)
}
/// RFC 1929 username/password sub-negotiation. Both fields carry a one-byte length, so longer
/// values are rejected rather than silently truncated into a malformed request.
fn socks5_auth_request(user: &str, pass: &str) -> Result<Vec<u8>> {
let ulen = u8::try_from(user.len()).map_err(|_| "SOCKS5 proxy: username longer than 255 bytes")?;
let plen = u8::try_from(pass.len()).map_err(|_| "SOCKS5 proxy: password longer than 255 bytes")?;
let mut m = vec![1, ulen];
m.extend(user.as_bytes());
m.push(plen);
m.extend(pass.as_bytes());
Ok(m)
}
/// RFC 1928 CONNECT with a domain-name address (one-byte length).
fn socks5_connect_request(host: &str, port: u16) -> Result<Vec<u8>> {
let hlen = u8::try_from(host.len()).map_err(|_| "SOCKS5 proxy: host name longer than 255 bytes")?;
let mut req = vec![5, 1, 0, 3, hlen];
req.extend(host.as_bytes());
req.extend(port.to_be_bytes());
Ok(req)
}
async fn http_connect(sc: &SshConfig, proxy_password: &str, host: &str, port: u16) -> Result<tokio::net::TcpStream> {
let pport = if sc.proxy_port == 0 { 8080 } else { sc.proxy_port as u16 };
let mut s = tokio::net::TcpStream::connect((sc.proxy_host.as_str(), pport)).await.map_err(|e| format!("HTTP proxy connection failed: {e}"))?;
let mut req = format!("CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n");
if !sc.proxy_username.is_empty() {
let cred = base64::engine::general_purpose::STANDARD.encode(format!("{}:{proxy_password}", sc.proxy_username));
req += &format!("Proxy-Authorization: Basic {cred}\r\n");
}
req += "\r\n";
s.write_all(req.as_bytes()).await?;
// Read the response head byte by byte so no tunnelled bytes are consumed.
let mut head = Vec::new();
while !head.ends_with(b"\r\n\r\n") {
head.push(s.read_u8().await.map_err(|e| format!("HTTP CONNECT response read failed: {e}"))?);
if head.len() > 16 * 1024 {
return Err("HTTP CONNECT response too large".into());
}
}
let status = String::from_utf8_lossy(&head).lines().next().unwrap_or("").to_string();
if status.split_whitespace().nth(1) != Some("200") {
return Err(format!("HTTP CONNECT proxy returned {status}").into());
}
Ok(s)
let host = host.to_string();
let s = blocking(move || proxy.connect(&host, port, timeout)).await?;
s.set_nonblocking(true)?;
Ok(tokio::net::TcpStream::from_std(s)?)
}
fn config(sc: &SshConfig) -> Arc<client::Config> {
@@ -912,15 +834,6 @@ mod tests {
assert_eq!(c.local_addr().unwrap(), s.peer_addr().unwrap());
}
#[test]
fn socks5_requests_reject_overlong_fields() {
assert_eq!(socks5_connect_request("example.com", 22).unwrap(), [&[5u8, 1, 0, 3, 11][..], b"example.com", &[0, 22]].concat());
assert!(socks5_connect_request(&"h".repeat(256), 22).is_err());
assert_eq!(socks5_auth_request("u", "pw").unwrap(), vec![1, 1, b'u', 2, b'p', b'w']);
assert!(socks5_auth_request("u", &"p".repeat(256)).is_err());
assert!(socks5_auth_request(&"u".repeat(256), "p").is_err());
}
#[test]
fn generates_encrypted_key_pairs_in_process() {
let dir = std::env::temp_dir().join(format!("nexus-keygen-{}", std::process::id()));