use std::io::{self, Read, Write};
use std::net::{Shutdown, SocketAddr, TcpStream};
use std::sync::Arc;
use std::time::Duration;
use crate::error::{Error, Result};
use crate::net::socks;
use crate::net::stream::NetStream;
use crate::net::Resolver;
use crate::url::percent_decode;
#[derive(Debug, Clone)]
pub struct HttpProxyIntent {
pub auth: Option<(String, String)>,
}
pub trait Connector: Send + Sync + std::fmt::Debug {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>>;
fn http_forward_proxy(&self) -> Option<HttpProxyIntent> {
None
}
fn is_direct(&self) -> bool {
false
}
fn udp_proxy(&self) -> crate::net::udp::UdpProxy {
crate::net::udp::UdpProxy::Unsupported
}
}
fn open_tcp(host: &str, port: u16, timeout: Option<Duration>) -> Result<TcpStream> {
let addrs = crate::net::StdResolver.resolve(host, port)?;
Ok(crate::net::connect_any(&addrs, timeout)?)
}
#[derive(Debug, Default, Clone)]
pub struct DirectConnector;
impl Connector for DirectConnector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
Ok(Box::new(open_tcp(host, port, timeout)?))
}
fn is_direct(&self) -> bool {
true
}
fn udp_proxy(&self) -> crate::net::udp::UdpProxy {
crate::net::udp::UdpProxy::Direct
}
}
#[derive(Debug, Clone)]
pub struct Socks4Connector {
pub host: String,
pub port: u16,
pub user: String,
pub remote_dns: bool,
}
impl Connector for Socks4Connector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let mut s = open_tcp(&self.host, self.port, timeout)?;
apply_handshake_timeout(&s, timeout)?;
socks::socks4_connect(&mut s, host, port, &self.user, self.remote_dns)?;
clear_handshake_timeout(&s)?;
Ok(Box::new(s))
}
}
#[derive(Debug, Clone)]
pub struct Socks5Connector {
pub host: String,
pub port: u16,
pub auth: Option<(String, String)>,
pub remote_dns: bool,
}
impl Connector for Socks5Connector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let mut s = open_tcp(&self.host, self.port, timeout)?;
apply_handshake_timeout(&s, timeout)?;
let auth = self.auth.as_ref().map(|(u, p)| (u.as_str(), p.as_str()));
socks::socks5_connect(&mut s, host, port, auth, self.remote_dns)?;
clear_handshake_timeout(&s)?;
Ok(Box::new(s))
}
fn udp_proxy(&self) -> crate::net::udp::UdpProxy {
crate::net::udp::UdpProxy::Socks5 {
host: self.host.clone(),
port: self.port,
auth: self.auth.clone(),
}
}
}
#[derive(Debug, Clone)]
pub struct HttpProxyConnector {
pub host: String,
pub port: u16,
pub auth: Option<(String, String)>,
}
impl Connector for HttpProxyConnector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let mut s = open_tcp(&self.host, self.port, timeout)?;
apply_handshake_timeout(&s, timeout)?;
http_connect(&mut s, host, port, self.auth.as_ref())?;
clear_handshake_timeout(&s)?;
Ok(Box::new(s))
}
fn http_forward_proxy(&self) -> Option<HttpProxyIntent> {
Some(HttpProxyIntent {
auth: self.auth.clone(),
})
}
}
#[derive(Debug, Clone)]
pub struct HttpsProxyConnector {
pub host: String,
pub port: u16,
pub auth: Option<(String, String)>,
}
impl Connector for HttpsProxyConnector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let tls = crate::tls::TlsSettings::default();
https_proxy_connect(
&self.host,
self.port,
self.auth.as_ref(),
&tls,
host,
port,
timeout,
)
}
fn http_forward_proxy(&self) -> Option<HttpProxyIntent> {
Some(HttpProxyIntent {
auth: self.auth.clone(),
})
}
}
#[derive(Debug, Clone)]
pub(crate) struct HttpsProxyTlsConnector {
host: String,
port: u16,
auth: Option<(String, String)>,
tls: crate::tls::TlsSettings,
}
impl Connector for HttpsProxyTlsConnector {
fn connect(
&self,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
https_proxy_connect(
&self.host,
self.port,
self.auth.as_ref(),
&self.tls,
host,
port,
timeout,
)
}
fn http_forward_proxy(&self) -> Option<HttpProxyIntent> {
Some(HttpProxyIntent {
auth: self.auth.clone(),
})
}
}
fn https_proxy_connect(
proxy_host: &str,
proxy_port: u16,
auth: Option<&(String, String)>,
tls: &crate::tls::TlsSettings,
host: &str,
port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let opts = tls.to_opts(&[])?;
let tcp = open_tcp(proxy_host, proxy_port, timeout)?;
let ctl = tcp.try_clone()?;
apply_handshake_timeout(&tcp, timeout)?;
let mut tls = crate::tls::connect_over_tls(tcp, proxy_host, opts)?;
http_connect(&mut tls, host, port, auth)?;
clear_handshake_timeout(&ctl)?;
Ok(Box::new(TlsProxyStream { tls, ctl }))
}
struct TlsProxyStream {
tls: crate::tls::TlsStream<TcpStream>,
ctl: TcpStream,
}
impl Read for TlsProxyStream {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.tls.read(buf)
}
}
impl Write for TlsProxyStream {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.tls.write(buf)
}
fn flush(&mut self) -> io::Result<()> {
self.tls.flush()
}
}
impl NetStream for TlsProxyStream {
fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
self.ctl.set_read_timeout(dur)
}
fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
self.ctl.set_write_timeout(dur)
}
fn peer_addr(&self) -> io::Result<SocketAddr> {
self.ctl.peer_addr()
}
fn local_addr(&self) -> io::Result<SocketAddr> {
self.ctl.local_addr()
}
fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.ctl.shutdown(how)
}
fn try_clone_box(&self) -> io::Result<Box<dyn NetStream>> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"cannot clone an https-proxy stream",
))
}
}
fn http_connect<S: Read + Write>(
stream: &mut S,
host: &str,
port: u16,
auth: Option<&(String, String)>,
) -> Result<()> {
const MAX_HEADER_BYTES: usize = 64 * 1024;
let host_port = crate::url::authority(host, port);
let mut buf = Vec::with_capacity(128);
write!(
&mut buf,
"CONNECT {host_port} HTTP/1.1\r\nHost: {host_port}\r\nProxy-Connection: Keep-Alive\r\n"
)?;
if let Some((user, pass)) = auth {
let creds = crate::websocket::base64_encode(format!("{user}:{pass}").as_bytes());
write!(&mut buf, "Proxy-Authorization: Basic {creds}\r\n")?;
}
buf.extend_from_slice(b"\r\n");
stream.write_all(&buf)?;
stream.flush()?;
let mut status: Option<String> = None;
let mut line: Vec<u8> = Vec::with_capacity(128);
let mut byte = [0u8; 1];
let mut total = 0usize;
loop {
if total > MAX_HEADER_BYTES {
return Err(Error::BadResponse(
"CONNECT response headers exceed 64 KiB".into(),
));
}
let n = stream.read(&mut byte)?;
if n == 0 {
return Err(Error::UnexpectedEof);
}
total += 1;
if byte[0] == b'\n' {
let trimmed = line.strip_suffix(b"\r").unwrap_or(&line);
if status.is_none() {
status = Some(String::from_utf8_lossy(trimmed).into_owned());
}
if trimmed.is_empty() {
break;
}
line.clear();
} else {
line.push(byte[0]);
}
}
let status = status.ok_or_else(|| Error::BadResponse("CONNECT: no status line".into()))?;
let Some(code_str) = status.split(' ').nth(1) else {
return Err(Error::BadResponse(format!(
"CONNECT: malformed status line {status:?}"
)));
};
let code: u16 = code_str
.parse()
.map_err(|_| Error::BadResponse(format!("CONNECT: bad status {code_str:?}")))?;
if !(200..300).contains(&code) {
return Err(Error::BadResponse(format!(
"CONNECT to {host_port} failed: {status}"
)));
}
Ok(())
}
fn apply_handshake_timeout(s: &TcpStream, timeout: Option<Duration>) -> Result<()> {
if let Some(t) = timeout {
s.set_read_timeout(Some(t))?;
s.set_write_timeout(Some(t))?;
}
Ok(())
}
fn clear_handshake_timeout(s: &TcpStream) -> Result<()> {
s.set_read_timeout(None)?;
s.set_write_timeout(None)?;
Ok(())
}
#[cfg(unix)]
#[derive(Debug, Clone)]
pub struct UnixConnector {
pub path: std::path::PathBuf,
}
#[cfg(unix)]
impl Connector for UnixConnector {
fn connect(
&self,
_host: &str,
_port: u16,
timeout: Option<Duration>,
) -> Result<Box<dyn NetStream>> {
let s = std::os::unix::net::UnixStream::connect(&self.path)?;
if let Some(t) = timeout {
s.set_read_timeout(Some(t))?;
s.set_write_timeout(Some(t))?;
}
Ok(Box::new(s))
}
}
pub fn connector_from_proxy_url(spec: &str) -> Result<Arc<dyn Connector>> {
Ok(ProxySpec::parse(spec)?.connector(&crate::tls::TlsSettings::default()))
}
#[derive(Debug, Clone)]
pub(crate) struct ProxySpec {
scheme: String,
auth: Option<(String, String)>,
host: String,
port: u16,
}
impl ProxySpec {
pub(crate) fn parse(spec: &str) -> Result<ProxySpec> {
let p = parse_proxy_spec(spec)?;
match p.scheme.as_str() {
"http" | "https" | "socks4" | "socks4a" | "socks5" | "socks5h" => Ok(p),
other => Err(Error::UnsupportedScheme(format!(
"proxy scheme {other:?} not supported (use http/https/socks4/socks4a/socks5/socks5h)"
))),
}
}
pub(crate) fn http_config(&self) -> Option<crate::http::ProxyConfig> {
(self.scheme == "http").then(|| crate::http::ProxyConfig {
host: self.host.clone(),
port: self.port,
auth: self.auth.clone(),
})
}
pub(crate) fn connector(&self, tls: &crate::tls::TlsSettings) -> Arc<dyn Connector> {
let host = self.host.clone();
let port = self.port;
match self.scheme.as_str() {
"https" => Arc::new(HttpsProxyTlsConnector {
host,
port,
auth: self.auth.clone(),
tls: tls.clone(),
}),
"socks4" | "socks4a" => Arc::new(Socks4Connector {
user: self
.auth
.as_ref()
.map(|(u, _)| u.clone())
.unwrap_or_default(),
remote_dns: self.scheme == "socks4a",
host,
port,
}),
"socks5" | "socks5h" => Arc::new(Socks5Connector {
remote_dns: self.scheme == "socks5h",
host,
port,
auth: self.auth.clone(),
}),
_ => Arc::new(HttpProxyConnector {
host,
port,
auth: self.auth.clone(),
}),
}
}
}
fn parse_proxy_spec(spec: &str) -> Result<ProxySpec> {
let (scheme, rest) = match spec.split_once("://") {
Some((s, r)) => (s.to_ascii_lowercase(), r),
None => ("http".to_string(), spec),
};
let rest = match rest.find(['/', '?', '#']) {
Some(i) => &rest[..i],
None => rest,
};
let (userinfo, hostport) = match rest.rfind('@') {
Some(i) => (Some(&rest[..i]), &rest[i + 1..]),
None => (None, rest),
};
let auth = userinfo.map(|info| match info.split_once(':') {
Some((u, p)) => (percent_decode(u), percent_decode(p)),
None => (percent_decode(info), String::new()),
});
let (host, port) = parse_hostport(hostport)?;
Ok(ProxySpec {
scheme,
auth,
host,
port,
})
}
fn parse_hostport(hp: &str) -> Result<(String, u16)> {
const DEFAULT_PROXY_PORT: u16 = 1080;
let bad = |what: &str| Error::InvalidUrl(format!("proxy: {what} in {hp:?}"));
if let Some(after_bracket) = hp.strip_prefix('[') {
let close = after_bracket
.find(']')
.ok_or_else(|| bad("unterminated IPv6"))?;
let host = after_bracket[..close].to_string();
let tail = &after_bracket[close + 1..];
let port = if tail.is_empty() {
DEFAULT_PROXY_PORT
} else if let Some(p) = tail.strip_prefix(':') {
p.parse().map_err(|_| bad("bad port"))?
} else {
return Err(bad("junk after IPv6 host"));
};
if host.is_empty() {
return Err(bad("empty host"));
}
Ok((host, port))
} else if let Some(i) = hp.rfind(':') {
let host = hp[..i].to_string();
let port = hp[i + 1..].parse().map_err(|_| bad("bad port"))?;
if host.is_empty() {
return Err(bad("empty host"));
}
Ok((host, port))
} else if hp.is_empty() {
Err(bad("empty host"))
} else {
Ok((hp.to_string(), DEFAULT_PROXY_PORT))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn factory_dispatches_schemes() {
assert!(connector_from_proxy_url("http://p:8080").is_ok());
assert!(connector_from_proxy_url("https://p:443").is_ok());
assert!(connector_from_proxy_url("socks4://p:1080").is_ok());
assert!(connector_from_proxy_url("socks4a://p:1080").is_ok());
assert!(connector_from_proxy_url("socks5://p:1080").is_ok());
assert!(connector_from_proxy_url("socks5h://p").is_ok()); assert!(matches!(
connector_from_proxy_url("ftp://p:21"),
Err(Error::UnsupportedScheme(_))
));
}
#[test]
fn factory_parses_auth_and_default_port() {
let p = parse_proxy_spec("socks5h://alice:secret@proxy.local").unwrap();
assert_eq!(p.scheme, "socks5h");
assert_eq!(p.host, "proxy.local");
assert_eq!(p.port, 1080);
assert_eq!(p.auth, Some(("alice".into(), "secret".into())));
}
#[test]
fn factory_bare_hostport_is_http() {
let p = parse_proxy_spec("proxy:3128").unwrap();
assert_eq!(p.scheme, "http");
assert_eq!(p.host, "proxy");
assert_eq!(p.port, 3128);
assert!(p.auth.is_none());
}
#[test]
fn factory_ipv6_hostport() {
let p = parse_proxy_spec("socks5://[::1]:1080").unwrap();
assert_eq!(p.host, "::1");
assert_eq!(p.port, 1080);
let p2 = parse_proxy_spec("socks5://[fe80::1]").unwrap();
assert_eq!(p2.host, "fe80::1");
assert_eq!(p2.port, 1080);
}
#[test]
fn proxy_spec_ignores_trailing_path() {
let p = parse_proxy_spec("http://proxy:3128/").unwrap();
assert_eq!((p.host.as_str(), p.port), ("proxy", 3128));
let p = parse_proxy_spec("socks5h://u:p@proxy:1080/some/path?x#y").unwrap();
assert_eq!((p.host.as_str(), p.port), ("proxy", 1080));
assert_eq!(p.auth, Some(("u".into(), "p".into())));
let p = parse_proxy_spec("http://[::1]:8080/").unwrap();
assert_eq!((p.host.as_str(), p.port), ("::1", 8080));
}
#[test]
fn proxy_spec_percent_decodes_credentials() {
let p = parse_proxy_spec("http://us%65r:p%40ss%3Aword@proxy:3128").unwrap();
assert_eq!(p.auth, Some(("user".into(), "p@ss:word".into())));
}
#[test]
fn http_config_only_for_http_scheme() {
let cfg = |s: &str| ProxySpec::parse(s).unwrap().http_config();
let c = cfg("http://u:p%21@proxy:3128/").unwrap();
assert_eq!((c.host.as_str(), c.port), ("proxy", 3128));
assert_eq!(c.auth, Some(("u".into(), "p!".into())));
assert!(cfg("proxy:8080").is_some());
assert!(cfg("socks5://proxy:1080").is_none());
assert!(cfg("https://proxy:443").is_none());
}
#[test]
fn open_tcp_dials_ipv6_and_ipv4_literals() {
let l4 = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let p4 = l4.local_addr().unwrap().port();
assert!(open_tcp("127.0.0.1", p4, Some(Duration::from_secs(2))).is_ok());
if let Ok(l6) = std::net::TcpListener::bind("[::1]:0") {
let p6 = l6.local_addr().unwrap().port();
assert!(open_tcp("::1", p6, Some(Duration::from_secs(2))).is_ok());
assert!(open_tcp("[::1]", p6, Some(Duration::from_secs(2))).is_ok());
}
}
#[test]
fn http_connect_brackets_ipv6_target() {
struct Mock {
reply: std::io::Cursor<Vec<u8>>,
written: Vec<u8>,
}
impl Read for Mock {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.reply.read(buf)
}
}
impl Write for Mock {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.written.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
for host in ["::1", "[::1]"] {
let mut m = Mock {
reply: std::io::Cursor::new(b"HTTP/1.1 200 OK\r\n\r\n".to_vec()),
written: Vec::new(),
};
http_connect(&mut m, host, 443, None).unwrap();
let w = String::from_utf8(m.written).unwrap();
assert!(w.starts_with("CONNECT [::1]:443 HTTP/1.1\r\n"), "{w}");
}
let mut m = Mock {
reply: std::io::Cursor::new(b"HTTP/1.1 407 Auth\r\n\r\n".to_vec()),
written: Vec::new(),
};
assert!(http_connect(&mut m, "h", 443, None).is_err());
}
#[test]
fn direct_connector_is_direct() {
assert!(DirectConnector.is_direct());
assert!(DirectConnector.http_forward_proxy().is_none());
}
#[test]
fn http_proxy_connector_signals_forward_intent() {
let c = HttpProxyConnector {
host: "p".into(),
port: 8080,
auth: Some(("u".into(), "p".into())),
};
let intent = c.http_forward_proxy().expect("forward intent");
assert_eq!(intent.auth, Some(("u".into(), "p".into())));
assert!(!c.is_direct());
}
}