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();
269 let limit = MAX_HTTP_HEADER_BYTES as u64;
270 let mut reader = BufReader::new(stream.take(limit));
271 let mut line = String::new();
272
273 loop {
274 line.clear();
275 let n = reader.read_line(&mut line).await?;
276 if n == 0 {
277 return Err(ProxyError::HttpConnectMalformed);
280 }
281
282 response_buffer.extend_from_slice(line.as_bytes());
283
284 if line == "\r\n" || line == "\n" {
286 break;
287 }
288 }
289
290 if !reader.buffer().is_empty() {
291 return Err(ProxyError::HttpConnectMalformed);
299 }
300
301 let _status_code = parse_http_connect_response(&response_buffer)?;
303
304 Ok(reader.into_inner().into_inner())
306}
307
308#[derive(Clone, Debug, thiserror::Error)]
310#[non_exhaustive]
311pub enum ProxyError {
312 #[error("Problem while connecting to proxy")]
314 ProxyConnect(#[source] Arc<std::io::Error>),
315
316 #[error("Problem while communicating with proxy")]
318 ProxyIo(#[source] Arc<std::io::Error>),
319
320 #[error("SOCKS proxy does not support target address")]
322 InvalidSocksAddr(#[source] tor_socksproto::Error),
323
324 #[error("Got an address type we don't recognize")]
326 UnrecognizedAddr,
327
328 #[error("Tried to make an invalid SOCKS request")]
330 InvalidSocksRequest(#[source] tor_socksproto::Error),
331
332 #[error("Protocol error while communicating with SOCKS proxy")]
334 SocksProto(#[source] tor_socksproto::Error),
335
336 #[error("Internal error")]
338 Bug(#[from] tor_error::Bug),
339
340 #[error("Received unexpected early data from peer")]
351 UnexpectedData,
352
353 #[error("SOCKS proxy reported an error: {0}")]
355 SocksError(SocksStatus),
356
357 #[error("HTTP CONNECT proxy returned status: {0}")]
359 HttpConnectError(u16),
360
361 #[error("HTTP CONNECT proxy returned invalid response")]
363 HttpConnectMalformed,
364}
365
366impl From<std::io::Error> for ProxyError {
367 fn from(e: std::io::Error) -> Self {
368 ProxyError::ProxyIo(Arc::new(e))
369 }
370}
371
372impl From<ProxyError> for std::io::Error {
373 fn from(e: ProxyError) -> Self {
374 std::io::Error::other(e)
375 }
376}
377
378impl tor_error::HasKind for ProxyError {
379 fn kind(&self) -> tor_error::ErrorKind {
380 use ProxyError as E;
381 use tor_error::ErrorKind as EK;
382 match self {
383 E::ProxyConnect(_) | E::ProxyIo(_) => EK::LocalNetworkError,
384 E::InvalidSocksAddr(_) | E::InvalidSocksRequest(_) => EK::BadApiUsage,
385 E::UnrecognizedAddr => EK::NotImplemented,
386 E::SocksProto(_) => EK::LocalProtocolViolation,
387 E::Bug(e) => e.kind(),
388 E::UnexpectedData => EK::NotImplemented,
389 E::SocksError(_) => EK::LocalProtocolViolation,
390 E::HttpConnectError(_) | E::HttpConnectMalformed => EK::LocalProtocolViolation,
391 }
392 }
393}
394
395impl tor_error::HasRetryTime for ProxyError {
396 fn retry_time(&self) -> tor_error::RetryTime {
397 use ProxyError as E;
398 use SocksStatus as S;
399 use tor_error::RetryTime as RT;
400 match self {
401 E::ProxyConnect(_) | E::ProxyIo(_) => RT::AfterWaiting,
402 E::InvalidSocksAddr(_) => RT::Never,
403 E::UnrecognizedAddr => RT::Never,
404 E::InvalidSocksRequest(_) => RT::Never,
405 E::SocksProto(_) => RT::AfterWaiting,
406 E::Bug(_) => RT::Never,
407 E::UnexpectedData => RT::Never,
408 E::SocksError(e) => match *e {
409 S::CONNECTION_REFUSED
410 | S::GENERAL_FAILURE
411 | S::HOST_UNREACHABLE
412 | S::NETWORK_UNREACHABLE
413 | S::TTL_EXPIRED => RT::AfterWaiting,
414 _ => RT::Never,
415 },
416 E::HttpConnectError(code) => {
417 if *code == 502 || *code == 503 || *code == 504 {
419 RT::AfterWaiting
420 } else {
421 RT::Never
422 }
423 }
424 E::HttpConnectMalformed => RT::Never,
425 }
426 }
427}
428
429#[cfg(feature = "pt-client")]
430#[derive(Clone, Debug)]
433pub struct ExternalProxyPlugin<R> {
434 runtime: R,
436 proxy_addr: SocketAddr,
438 proxy_version: SocksVersion,
440}
441
442#[cfg(feature = "pt-client")]
443impl<R: NetStreamProvider + Send + Sync> ExternalProxyPlugin<R> {
444 pub fn new(rt: R, proxy_addr: SocketAddr, proxy_version: SocksVersion) -> Self {
446 Self {
447 runtime: rt,
448 proxy_addr,
449 proxy_version,
450 }
451 }
452}
453
454#[cfg(feature = "pt-client")]
455#[async_trait]
456impl<R: NetStreamProvider + Send + Sync> TransportImplHelper for ExternalProxyPlugin<R> {
457 type Stream = R::Stream;
458
459 async fn connect(&self, target: &OwnedChanTarget) -> crate::Result<(PeerAddr, R::Stream)> {
460 let pt_target = match target.chan_method() {
461 ChannelMethod::Direct(_) => {
462 return Err(crate::Error::UnusableTarget(bad_api_usage!(
463 "Used pluggable transport for a TCP connection."
464 )));
465 }
466 ChannelMethod::Pluggable(target) => target,
467 other => {
468 return Err(crate::Error::UnusableTarget(bad_api_usage!(
469 "Used unknown, unsupported, transport {:?} for a TCP connection.",
470 other,
471 )));
472 }
473 };
474
475 let into_err = |e: ProxyError| crate::Error::Connect {
476 addresses: vec![(sv(pt_target.to_string()), e.into())],
477 };
478 let protocol =
479 settings_to_protocol(self.proxy_version, encode_settings(pt_target.settings()))
480 .map_err(into_err)?;
481 let stream =
482 connect_via_proxy(&self.runtime, &self.proxy_addr, &protocol, pt_target.addr())
483 .await
484 .map_err(into_err)?;
485
486 Ok((pt_target.into(), stream))
487 }
488}
489
490#[cfg(feature = "pt-client")]
492fn encode_settings<'a, IT>(settings: IT) -> String
493where
494 IT: Iterator<Item = (&'a str, &'a str)>,
495{
496 enum EscChar {
500 Backslash(char),
502 Literal(char),
504 Done,
506 }
507 impl EscChar {
508 fn new(ch: char, in_key: bool) -> Self {
510 match ch {
511 '\\' | ';' => EscChar::Backslash(ch),
512 '=' if in_key => EscChar::Backslash(ch),
513 _ => EscChar::Literal(ch),
514 }
515 }
516 }
517 impl Iterator for EscChar {
518 type Item = char;
519
520 fn next(&mut self) -> Option<Self::Item> {
521 match *self {
522 EscChar::Backslash(ch) => {
523 *self = EscChar::Literal(ch);
524 Some('\\')
525 }
526 EscChar::Literal(ch) => {
527 *self = EscChar::Done;
528 Some(ch)
529 }
530 EscChar::Done => None,
531 }
532 }
533 }
534
535 fn esc(s: &str, in_key: bool) -> impl Iterator<Item = char> + '_ {
537 s.chars().flat_map(move |c| EscChar::new(c, in_key))
538 }
539
540 let mut result = String::new();
541 for (k, v) in settings {
542 result.extend(esc(k, true));
543 result.push('=');
544 result.extend(esc(v, false));
545 result.push(';');
546 }
547 result.pop(); result
550}
551
552#[cfg(feature = "pt-client")]
556pub fn settings_to_protocol(vers: SocksVersion, s: String) -> Result<Protocol, ProxyError> {
557 let mut bytes: Vec<_> = s.into();
558 Ok(if bytes.is_empty() {
559 Protocol::Socks(vers, SocksAuth::NoAuth)
560 } else if vers == SocksVersion::V4 {
561 if bytes.contains(&0) {
562 return Err(ProxyError::InvalidSocksRequest(
563 tor_socksproto::Error::NotImplemented(
564 "SOCKS 4 doesn't support internal NUL bytes (for PT settings list)".into(),
565 ),
566 ));
567 } else {
568 Protocol::Socks(SocksVersion::V4, SocksAuth::Socks4(bytes))
569 }
570 } else if bytes.len() <= 255 {
571 Protocol::Socks(SocksVersion::V5, SocksAuth::Username(bytes, vec![0]))
573 } else if bytes.len() <= (255 * 2) {
574 let password = bytes.split_off(255);
575 Protocol::Socks(SocksVersion::V5, SocksAuth::Username(bytes, password))
576 } else {
577 return Err(ProxyError::InvalidSocksRequest(
578 tor_socksproto::Error::NotImplemented("PT settings list too long for SOCKS 5".into()),
579 ));
580 })
581}
582
583#[cfg(test)]
584mod test {
585 #![allow(clippy::bool_assert_comparison)]
587 #![allow(clippy::clone_on_copy)]
588 #![allow(clippy::dbg_macro)]
589 #![allow(clippy::mixed_attributes_style)]
590 #![allow(clippy::print_stderr)]
591 #![allow(clippy::print_stdout)]
592 #![allow(clippy::single_char_pattern)]
593 #![allow(clippy::unwrap_used)]
594 #![allow(clippy::unchecked_time_subtraction)]
595 #![allow(clippy::useless_vec)]
596 #![allow(clippy::needless_pass_by_value)]
597 #![allow(clippy::string_slice)] #[allow(unused_imports)]
600 use super::*;
601
602 #[test]
603 fn protocol_debug_redacts_http_connect_auth() {
604 let proto = Protocol::HttpConnect {
605 auth: Some((
606 Sensitive::new("user_name".to_owned()),
607 Sensitive::new("pass_word".to_owned()),
608 )),
609 };
610
611 let formatted = format!("{proto:?}");
612 assert!(formatted.contains("HttpConnect"));
613 assert!(!formatted.contains("user_name"));
614 assert!(!formatted.contains("pass_word"));
615 }
616
617 #[cfg(feature = "pt-client")]
618 #[test]
619 fn setting_encoding() {
620 fn check(settings: Vec<(&str, &str)>, expected: &str) {
621 assert_eq!(encode_settings(settings.into_iter()), expected);
622 }
623
624 check(vec![], "");
626 check(vec![("hello", "world")], "hello=world");
627 check(
628 vec![("hey", "verden"), ("hello", "world")],
629 "hey=verden;hello=world",
630 );
631 check(
632 vec![("hey", "verden"), ("hello", "world"), ("selv", "tak")],
633 "hey=verden;hello=world;selv=tak",
634 );
635
636 check(
637 vec![("semi;colon", "equals=sign")],
638 r"semi\;colon=equals=sign",
639 );
640 check(
641 vec![("equals=sign", "semi;colon")],
642 r"equals\=sign=semi\;colon",
643 );
644 check(
645 vec![("semi;colon", "equals=sign"), ("also", "back\\slash")],
646 r"semi\;colon=equals=sign;also=back\\slash",
647 );
648 }
649
650 #[cfg(feature = "pt-client")]
651 #[test]
652 fn split_settings() {
653 use SocksVersion::*;
654 let long_string = "examplestrg".to_owned().repeat(50);
655 assert_eq!(long_string.len(), 550);
656 let sv = |v, a, b| settings_to_protocol(v, long_string[a..b].to_owned()).unwrap();
657 let s = |a, b| sv(V5, a, b);
658 let v = |a, b| long_string.as_bytes()[a..b].to_vec();
659
660 assert_eq!(s(0, 0), Protocol::Socks(V5, SocksAuth::NoAuth));
661 assert_eq!(
662 s(0, 50),
663 Protocol::Socks(V5, SocksAuth::Username(v(0, 50), vec![0]))
664 );
665 assert_eq!(
666 s(0, 255),
667 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), vec![0]))
668 );
669 assert_eq!(
670 s(0, 256),
671 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 256)))
672 );
673 assert_eq!(
674 s(0, 300),
675 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 300)))
676 );
677 assert_eq!(
678 s(0, 510),
679 Protocol::Socks(V5, SocksAuth::Username(v(0, 255), v(255, 510)))
680 );
681
682 assert_eq!(
684 sv(V4, 0, 511),
685 Protocol::Socks(V4, SocksAuth::Socks4(v(0, 511)))
686 );
687
688 assert_eq!(
690 settings_to_protocol(V5, "\0".to_owned()).unwrap(),
691 Protocol::Socks(V5, SocksAuth::Username(vec![0], vec![0]))
692 );
693 assert_eq!(
694 settings_to_protocol(V5, "\0".to_owned().repeat(510)).unwrap(),
695 Protocol::Socks(V5, SocksAuth::Username(vec![0; 255], vec![0; 255]))
696 );
697
698 assert!(settings_to_protocol(V5, "\0".to_owned().repeat(511)).is_err());
700
701 assert!(settings_to_protocol(V5, long_string[0..512].to_owned()).is_err());
703
704 assert!(settings_to_protocol(V4, "\0".to_owned()).is_err());
706 }
707
708 #[test]
709 fn parse_http_connect_200_ok() {
710 let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n";
711 assert_eq!(parse_http_connect_response(response).unwrap(), 200);
712 }
713
714 #[test]
715 fn parse_http_connect_407_auth_required() {
716 let response = b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n";
717 match parse_http_connect_response(response) {
718 Err(ProxyError::HttpConnectError(407)) => (), other => panic!("Expected 407 error, got {:?}", other),
720 }
721 }
722
723 #[test]
724 fn parse_http_connect_malformed_no_status() {
725 let response = b"INVALID HTTP";
726 assert!(matches!(
727 parse_http_connect_response(response),
728 Err(ProxyError::HttpConnectMalformed)
729 ));
730 }
731
732 #[test]
733 fn parse_http_connect_with_headers() {
734 let response = b"HTTP/1.1 200 Connection Established\r\nConnection: close\r\nProxy-Agent: Proxy/1.0\r\n\r\n";
735 assert_eq!(parse_http_connect_response(response).unwrap(), 200);
736 }
737
738 #[test]
739 fn parse_http_connect_rejects_pipelined_data() {
740 let response = b"HTTP/1.1 200 OK\r\n\r\nEXTRA_DATA";
741 assert!(matches!(
742 parse_http_connect_response(response),
743 Err(ProxyError::UnexpectedData)
744 ));
745 }
746
747 #[test]
748 fn parse_http_connect_oversized_headers() {
749 let huge_header = vec![b'X'; MAX_HTTP_HEADER_BYTES + 1];
750 assert!(matches!(
751 parse_http_connect_response(&huge_header),
752 Err(ProxyError::HttpConnectMalformed)
753 ));
754 }
755}