1#![allow(dead_code)]
14
15use std::{
16 net::{IpAddr, SocketAddr},
17 sync::Arc,
18};
19
20use base64ct::{Base64, Encoding};
21use futures::io::{AsyncBufReadExt, BufReader};
22use futures::{AsyncReadExt, AsyncWriteExt};
23use httparse;
24use safelog::Sensitive;
25use tor_linkspec::PtTargetAddr;
26use tor_rtcompat::NetStreamProvider;
27use tor_socksproto::{
28 Handshake as _, SocksAddr, SocksAuth, SocksClientHandshake, SocksCmd, SocksRequest,
29 SocksStatus, SocksVersion,
30};
31use tracing::trace;
32
33#[cfg(feature = "pt-client")]
34use super::TransportImplHelper;
35#[cfg(feature = "pt-client")]
36use async_trait::async_trait;
37#[cfg(feature = "pt-client")]
38use safelog::sensitive as sv;
39#[cfg(feature = "pt-client")]
40use tor_error::bad_api_usage;
41#[cfg(feature = "pt-client")]
42use tor_linkspec::{ChannelMethod, HasChanMethod, OwnedChanTarget};
43#[cfg(feature = "pt-client")]
44use tor_proto::peer::PeerAddr;
45
46#[derive(Clone, Debug, Eq, PartialEq)]
48#[non_exhaustive]
49pub enum Protocol {
50 Socks(SocksVersion, SocksAuth),
52 HttpConnect {
54 auth: Option<(Sensitive<String>, Sensitive<String>)>,
56 },
57}
58
59const NO_ADDR: IpAddr = IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 1));
61const MAX_HTTP_HEADER_BYTES: usize = 16 * 1024;
63
64pub(crate) async fn connect_via_proxy<R: NetStreamProvider + Send + Sync>(
81 runtime: &R,
82 proxy: &SocketAddr,
83 protocol: &Protocol,
84 target: &PtTargetAddr,
85) -> Result<R::Stream, ProxyError> {
86 trace!(
87 "Launching a proxied connection to {} via proxy at {} using {:?}",
88 target, proxy, protocol
89 );
90 let connect_options = Default::default();
92 let stream = runtime
93 .connect(proxy, &connect_options)
94 .await
95 .map_err(|e| ProxyError::ProxyConnect(Arc::new(e)))?;
96
97 match protocol {
98 Protocol::Socks(version, auth) => {
99 do_socks_handshake::<R>(stream, version, auth, target).await
100 }
101 Protocol::HttpConnect { auth } => {
102 do_http_connect_handshake::<R>(stream, auth, target).await
103 }
104 }
105}
106
107async fn do_socks_handshake<R: NetStreamProvider + Send + Sync>(
109 mut stream: R::Stream,
110 version: &SocksVersion,
111 auth: &SocksAuth,
112 target: &PtTargetAddr,
113) -> Result<R::Stream, ProxyError> {
114 let (target_addr, target_port): (SocksAddr, u16) = match target {
115 PtTargetAddr::IpPort(a) => (SocksAddr::Ip(a.ip()), a.port()),
116 #[cfg(feature = "pt-client")]
117 PtTargetAddr::HostPort(host, port) => (
118 SocksAddr::Hostname(
119 host.clone()
120 .try_into()
121 .map_err(ProxyError::InvalidSocksAddr)?,
122 ),
123 *port,
124 ),
125 #[cfg(feature = "pt-client")]
126 PtTargetAddr::None => (SocksAddr::Ip(NO_ADDR), 1),
127 _ => return Err(ProxyError::UnrecognizedAddr),
128 };
129
130 let request = SocksRequest::new(
131 *version,
132 SocksCmd::CONNECT,
133 target_addr,
134 target_port,
135 auth.clone(),
136 )
137 .map_err(ProxyError::InvalidSocksRequest)?;
138 let mut handshake = SocksClientHandshake::new(request);
139
140 let mut buf = tor_socksproto::Buffer::new();
141 let reply = loop {
142 use tor_socksproto::NextStep as NS;
143 match handshake.step(&mut buf).map_err(ProxyError::SocksProto)? {
144 NS::Send(send) => {
145 stream.write_all(&send).await?;
146 stream.flush().await?;
147 }
148 NS::Finished(fin) => {
149 break fin
150 .into_output_forbid_pipelining()
151 .map_err(ProxyError::SocksProto)?;
152 }
153 NS::Recv(mut recv) => {
154 let n = stream.read(recv.buf()).await?;
155 recv.note_received(n).map_err(ProxyError::SocksProto)?;
156 }
157 }
158 };
159
160 let status = reply.status();
161 trace!("SOCKS handshake succeeded, status {:?}", status);
162
163 if status != SocksStatus::SUCCEEDED {
164 return Err(ProxyError::SocksError(status));
165 }
166
167 Ok(stream)
168}
169
170fn format_connect_target(target: &PtTargetAddr) -> Result<String, ProxyError> {
172 match target {
173 PtTargetAddr::IpPort(a) => {
174 let host = match a.ip() {
175 IpAddr::V4(ip) => ip.to_string(),
176 IpAddr::V6(ip) => format!("[{}]", ip),
177 };
178 Ok(format!("{}:{}", host, a.port()))
179 }
180 #[cfg(feature = "pt-client")]
181 PtTargetAddr::HostPort(host, port) => Ok(format!("{}:{}", host, port)),
182 #[cfg(feature = "pt-client")]
183 PtTargetAddr::None => Err(ProxyError::UnrecognizedAddr),
184 _ => Err(ProxyError::UnrecognizedAddr),
185 }
186}
187
188fn build_http_connect_request(
190 target_str: &str,
191 auth: &Option<(Sensitive<String>, Sensitive<String>)>,
192) -> String {
193 let mut request = format!(
195 "CONNECT {} HTTP/1.1\r\nHost: {}\r\n",
196 target_str, target_str
197 );
198
199 if let Some((user, pass)) = auth {
200 let credentials = format!("{}:{}", user.as_ref(), pass.as_ref());
202 let encoded = Base64::encode_string(credentials.as_bytes());
203 request.push_str(&format!("Proxy-Authorization: Basic {}\r\n", encoded));
204 }
205
206 request.push_str("\r\n");
207 request
208}
209
210fn parse_http_connect_response(response_bytes: &[u8]) -> Result<u16, ProxyError> {
216 if response_bytes.len() > MAX_HTTP_HEADER_BYTES {
217 return Err(ProxyError::HttpConnectMalformed);
218 }
219
220 let mut headers = [httparse::EMPTY_HEADER; 64];
221 let mut resp = httparse::Response::new(&mut headers);
222
223 match resp.parse(response_bytes) {
224 Ok(httparse::Status::Complete(header_end)) => {
225 let status = resp.code.ok_or(ProxyError::HttpConnectMalformed)?;
226
227 if !(200..300).contains(&status) {
228 return Err(ProxyError::HttpConnectError(status));
229 }
230
231 if header_end < response_bytes.len() {
233 return Err(ProxyError::UnexpectedData);
234 }
235
236 trace!("HTTP CONNECT successful, status {}", status);
237 Ok(status)
238 }
239 Ok(httparse::Status::Partial) => Err(ProxyError::HttpConnectMalformed),
240 Err(_) => Err(ProxyError::HttpConnectMalformed),
241 }
242}
243
244async fn send_http_connect_request<R: NetStreamProvider + Send + Sync>(
246 stream: &mut R::Stream,
247 auth: &Option<(Sensitive<String>, Sensitive<String>)>,
248 target_str: &str,
249) -> Result<(), ProxyError> {
250 let request = build_http_connect_request(target_str, auth);
251 trace!("Sending HTTP CONNECT request for {}", target_str);
252 stream.write_all(request.as_bytes()).await?;
253 stream.flush().await?;
254 Ok(())
255}
256
257async fn do_http_connect_handshake<R: NetStreamProvider + Send + Sync>(
259 mut stream: R::Stream,
260 auth: &Option<(Sensitive<String>, Sensitive<String>)>,
261 target: &PtTargetAddr,
262) -> Result<R::Stream, ProxyError> {
263 let target_str = format_connect_target(target)?;
264 send_http_connect_request::<R>(&mut stream, auth, &target_str).await?;
265
266 let mut response_buffer = Vec::new();
268 let mut reader = BufReader::new(stream);
269 let mut line = String::new();
270
271 loop {
272 line.clear();
273 let n = reader.read_line(&mut line).await?;
274 if n == 0 {
275 return Err(ProxyError::HttpConnectMalformed);
276 }
277
278 response_buffer.extend_from_slice(line.as_bytes());
279 if response_buffer.len() > MAX_HTTP_HEADER_BYTES {
280 return Err(ProxyError::HttpConnectMalformed);
281 }
282
283 if line == "\r\n" || line == "\n" {
285 break;
286 }
287 }
288
289 let _status_code = parse_http_connect_response(&response_buffer)?;
291
292 Ok(reader.into_inner())
294}
295
296#[derive(Clone, Debug, thiserror::Error)]
298#[non_exhaustive]
299pub enum ProxyError {
300 #[error("Problem while connecting to proxy")]
302 ProxyConnect(#[source] Arc<std::io::Error>),
303
304 #[error("Problem while communicating with proxy")]
306 ProxyIo(#[source] Arc<std::io::Error>),
307
308 #[error("SOCKS proxy does not support target address")]
310 InvalidSocksAddr(#[source] tor_socksproto::Error),
311
312 #[error("Got an address type we don't recognize")]
314 UnrecognizedAddr,
315
316 #[error("Tried to make an invalid SOCKS request")]
318 InvalidSocksRequest(#[source] tor_socksproto::Error),
319
320 #[error("Protocol error while communicating with SOCKS proxy")]
322 SocksProto(#[source] tor_socksproto::Error),
323
324 #[error("Internal error")]
326 Bug(#[from] tor_error::Bug),
327
328 #[error("Received unexpected early data from peer")]
339 UnexpectedData,
340
341 #[error("SOCKS proxy reported an error: {0}")]
343 SocksError(SocksStatus),
344
345 #[error("HTTP CONNECT proxy returned status: {0}")]
347 HttpConnectError(u16),
348
349 #[error("HTTP CONNECT proxy returned invalid response")]
351 HttpConnectMalformed,
352}
353
354impl From<std::io::Error> for ProxyError {
355 fn from(e: std::io::Error) -> Self {
356 ProxyError::ProxyIo(Arc::new(e))
357 }
358}
359
360impl From<ProxyError> for std::io::Error {
361 fn from(e: ProxyError) -> Self {
362 std::io::Error::other(e)
363 }
364}
365
366impl tor_error::HasKind for ProxyError {
367 fn kind(&self) -> tor_error::ErrorKind {
368 use ProxyError as E;
369 use tor_error::ErrorKind as EK;
370 match self {
371 E::ProxyConnect(_) | E::ProxyIo(_) => EK::LocalNetworkError,
372 E::InvalidSocksAddr(_) | E::InvalidSocksRequest(_) => EK::BadApiUsage,
373 E::UnrecognizedAddr => EK::NotImplemented,
374 E::SocksProto(_) => EK::LocalProtocolViolation,
375 E::Bug(e) => e.kind(),
376 E::UnexpectedData => EK::NotImplemented,
377 E::SocksError(_) => EK::LocalProtocolViolation,
378 E::HttpConnectError(_) | E::HttpConnectMalformed => EK::LocalProtocolViolation,
379 }
380 }
381}
382
383impl tor_error::HasRetryTime for ProxyError {
384 fn retry_time(&self) -> tor_error::RetryTime {
385 use ProxyError as E;
386 use SocksStatus as S;
387 use tor_error::RetryTime as RT;
388 match self {
389 E::ProxyConnect(_) | E::ProxyIo(_) => RT::AfterWaiting,
390 E::InvalidSocksAddr(_) => RT::Never,
391 E::UnrecognizedAddr => RT::Never,
392 E::InvalidSocksRequest(_) => RT::Never,
393 E::SocksProto(_) => RT::AfterWaiting,
394 E::Bug(_) => RT::Never,
395 E::UnexpectedData => RT::Never,
396 E::SocksError(e) => match *e {
397 S::CONNECTION_REFUSED
398 | S::GENERAL_FAILURE
399 | S::HOST_UNREACHABLE
400 | S::NETWORK_UNREACHABLE
401 | S::TTL_EXPIRED => RT::AfterWaiting,
402 _ => RT::Never,
403 },
404 E::HttpConnectError(code) => {
405 if *code == 502 || *code == 503 || *code == 504 {
407 RT::AfterWaiting
408 } else {
409 RT::Never
410 }
411 }
412 E::HttpConnectMalformed => RT::Never,
413 }
414 }
415}
416
417#[cfg(feature = "pt-client")]
418#[derive(Clone, Debug)]
421pub struct ExternalProxyPlugin<R> {
422 runtime: R,
424 proxy_addr: SocketAddr,
426 proxy_version: SocksVersion,
428}
429
430#[cfg(feature = "pt-client")]
431impl<R: NetStreamProvider + Send + Sync> ExternalProxyPlugin<R> {
432 pub fn new(rt: R, proxy_addr: SocketAddr, proxy_version: SocksVersion) -> Self {
434 Self {
435 runtime: rt,
436 proxy_addr,
437 proxy_version,
438 }
439 }
440}
441
442#[cfg(feature = "pt-client")]
443#[async_trait]
444impl<R: NetStreamProvider + Send + Sync> TransportImplHelper for ExternalProxyPlugin<R> {
445 type Stream = R::Stream;
446
447 async fn connect(&self, target: &OwnedChanTarget) -> crate::Result<(PeerAddr, R::Stream)> {
448 let pt_target = match target.chan_method() {
449 ChannelMethod::Direct(_) => {
450 return Err(crate::Error::UnusableTarget(bad_api_usage!(
451 "Used pluggable transport for a TCP connection."
452 )));
453 }
454 ChannelMethod::Pluggable(target) => target,
455 other => {
456 return Err(crate::Error::UnusableTarget(bad_api_usage!(
457 "Used unknown, unsupported, transport {:?} for a TCP connection.",
458 other,
459 )));
460 }
461 };
462
463 let into_err = |e: ProxyError| crate::Error::Connect {
464 addresses: vec![(sv(pt_target.to_string()), e.into())],
465 };
466 let protocol =
467 settings_to_protocol(self.proxy_version, encode_settings(pt_target.settings()))
468 .map_err(into_err)?;
469 let stream =
470 connect_via_proxy(&self.runtime, &self.proxy_addr, &protocol, pt_target.addr())
471 .await
472 .map_err(into_err)?;
473
474 Ok((pt_target.into(), stream))
475 }
476}
477
478#[cfg(feature = "pt-client")]
480fn encode_settings<'a, IT>(settings: IT) -> String
481where
482 IT: Iterator<Item = (&'a str, &'a str)>,
483{
484 enum EscChar {
488 Backslash(char),
490 Literal(char),
492 Done,
494 }
495 impl EscChar {
496 fn new(ch: char, in_key: bool) -> Self {
498 match ch {
499 '\\' | ';' => EscChar::Backslash(ch),
500 '=' if in_key => EscChar::Backslash(ch),
501 _ => EscChar::Literal(ch),
502 }
503 }
504 }
505 impl Iterator for EscChar {
506 type Item = char;
507
508 fn next(&mut self) -> Option<Self::Item> {
509 match *self {
510 EscChar::Backslash(ch) => {
511 *self = EscChar::Literal(ch);
512 Some('\\')
513 }
514 EscChar::Literal(ch) => {
515 *self = EscChar::Done;
516 Some(ch)
517 }
518 EscChar::Done => None,
519 }
520 }
521 }
522
523 fn esc(s: &str, in_key: bool) -> impl Iterator<Item = char> + '_ {
525 s.chars().flat_map(move |c| EscChar::new(c, in_key))
526 }
527
528 let mut result = String::new();
529 for (k, v) in settings {
530 result.extend(esc(k, true));
531 result.push('=');
532 result.extend(esc(v, false));
533 result.push(';');
534 }
535 result.pop(); result
538}
539
540#[cfg(feature = "pt-client")]
544pub fn settings_to_protocol(vers: SocksVersion, s: String) -> Result<Protocol, ProxyError> {
545 let mut bytes: Vec<_> = s.into();
546 Ok(if bytes.is_empty() {
547 Protocol::Socks(vers, SocksAuth::NoAuth)
548 } else if vers == SocksVersion::V4 {
549 if bytes.contains(&0) {
550 return Err(ProxyError::InvalidSocksRequest(
551 tor_socksproto::Error::NotImplemented(
552 "SOCKS 4 doesn't support internal NUL bytes (for PT settings list)".into(),
553 ),
554 ));
555 } else {
556 Protocol::Socks(SocksVersion::V4, SocksAuth::Socks4(bytes))
557 }
558 } else if bytes.len() <= 255 {
559 Protocol::Socks(SocksVersion::V5, SocksAuth::Username(bytes, vec![0]))
561 } else if bytes.len() <= (255 * 2) {
562 let password = bytes.split_off(255);
563 Protocol::Socks(SocksVersion::V5, SocksAuth::Username(bytes, password))
564 } else {
565 return Err(ProxyError::InvalidSocksRequest(
566 tor_socksproto::Error::NotImplemented("PT settings list too long for SOCKS 5".into()),
567 ));
568 })
569}
570
571#[cfg(test)]
572mod test {
573 #![allow(clippy::bool_assert_comparison)]
575 #![allow(clippy::clone_on_copy)]
576 #![allow(clippy::dbg_macro)]
577 #![allow(clippy::mixed_attributes_style)]
578 #![allow(clippy::print_stderr)]
579 #![allow(clippy::print_stdout)]
580 #![allow(clippy::single_char_pattern)]
581 #![allow(clippy::unwrap_used)]
582 #![allow(clippy::unchecked_time_subtraction)]
583 #![allow(clippy::useless_vec)]
584 #![allow(clippy::needless_pass_by_value)]
585 #![allow(clippy::string_slice)] #[allow(unused_imports)]
588 use super::*;
589
590 #[test]
591 fn protocol_debug_redacts_http_connect_auth() {
592 let proto = Protocol::HttpConnect {
593 auth: Some((
594 Sensitive::new("user_name".to_owned()),
595 Sensitive::new("pass_word".to_owned()),
596 )),
597 };
598
599 let formatted = format!("{proto:?}");
600 assert!(formatted.contains("HttpConnect"));
601 assert!(!formatted.contains("user_name"));
602 assert!(!formatted.contains("pass_word"));
603 }
604
605 #[cfg(feature = "pt-client")]
606 #[test]
607 fn setting_encoding() {
608 fn check(settings: Vec<(&str, &str)>, expected: &str) {
609 assert_eq!(encode_settings(settings.into_iter()), expected);
610 }
611
612 check(vec![], "");
614 check(vec![("hello", "world")], "hello=world");
615 check(
616 vec![("hey", "verden"), ("hello", "world")],
617 "hey=verden;hello=world",
618 );
619 check(
620 vec![("hey", "verden"), ("hello", "world"), ("selv", "tak")],
621 "hey=verden;hello=world;selv=tak",
622 );
623
624 check(
625 vec![("semi;colon", "equals=sign")],
626 r"semi\;colon=equals=sign",
627 );
628 check(
629 vec![("equals=sign", "semi;colon")],
630 r"equals\=sign=semi\;colon",
631 );
632 check(
633 vec![("semi;colon", "equals=sign"), ("also", "back\\slash")],
634 r"semi\;colon=equals=sign;also=back\\slash",
635 );
636 }
637
638 #[cfg(feature = "pt-client")]
639 #[test]
640 fn split_settings() {
641 use SocksVersion::*;
642 let long_string = "examplestrg".to_owned().repeat(50);
643 assert_eq!(long_string.len(), 550);
644 let sv = |v, a, b| settings_to_protocol(v, long_string[a..b].to_owned()).unwrap();
645 let s = |a, b| sv(V5, a, b);
646 let v = |a, b| long_string.as_bytes()[a..b].to_vec();
647
648 assert_eq!(s(0, 0), Protocol::Socks(V5, SocksAuth::NoAuth));
649 assert_eq!(
650 s(0, 50),
651 Protocol::Socks(V5, SocksAuth::Username(v(0, 50), vec![0]))
652 );
653 assert_eq!(
654 s(0, 255),
655 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), vec![0]))
656 );
657 assert_eq!(
658 s(0, 256),
659 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 256)))
660 );
661 assert_eq!(
662 s(0, 300),
663 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 300)))
664 );
665 assert_eq!(
666 s(0, 510),
667 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 510)))
668 );
669
670 assert_eq!(
672 sv(V4, 0, 511),
673 Protocol::Socks(V4, SocksAuth::Socks4(v(0, 511)))
674 );
675
676 assert_eq!(
678 settings_to_protocol(V5, "\0".to_owned()).unwrap(),
679 Protocol::Socks(V5, SocksAuth::Username(vec![0], vec![0]))
680 );
681 assert_eq!(
682 settings_to_protocol(V5, "\0".to_owned().repeat(510)).unwrap(),
683 Protocol::Socks(V5, SocksAuth::Username(vec![0; 255], vec![0; 255]))
684 );
685
686 assert!(settings_to_protocol(V5, "\0".to_owned().repeat(511)).is_err());
688
689 assert!(settings_to_protocol(V5, long_string[0..512].to_owned()).is_err());
691
692 assert!(settings_to_protocol(V4, "\0".to_owned()).is_err());
694 }
695
696 #[test]
697 fn parse_http_connect_200_ok() {
698 let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
699 assert_eq!(parse_http_connect_response(response).unwrap(), 200);
700 }
701
702 #[test]
703 fn parse_http_connect_407_auth_required() {
704 let response = b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n";
705 match parse_http_connect_response(response) {
706 Err(ProxyError::HttpConnectError(407)) => (), other => panic!("Expected 407 error, got {:?}", other),
708 }
709 }
710
711 #[test]
712 fn parse_http_connect_malformed_no_status() {
713 let response = b"INVALID HTTP";
714 assert!(matches!(
715 parse_http_connect_response(response),
716 Err(ProxyError::HttpConnectMalformed)
717 ));
718 }
719
720 #[test]
721 fn parse_http_connect_with_headers() {
722 let response = b"HTTP/1.1 200 Connection Established\r\nConnection: close\r\nProxy-Agent: Proxy/1.0\r\n\r\n";
723 assert_eq!(parse_http_connect_response(response).unwrap(), 200);
724 }
725
726 #[test]
727 fn parse_http_connect_rejects_pipelined_data() {
728 let response = b"HTTP/1.1 200 OK\r\n\r\nEXTRA_DATA";
729 assert!(matches!(
730 parse_http_connect_response(response),
731 Err(ProxyError::UnexpectedData)
732 ));
733 }
734
735 #[test]
736 fn parse_http_connect_oversized_headers() {
737 let huge_header = vec![b'X'; MAX_HTTP_HEADER_BYTES + 1];
738 assert!(matches!(
739 parse_http_connect_response(&huge_header),
740 Err(ProxyError::HttpConnectMalformed)
741 ));
742 }
743}