hyper_http_proxy/
lib.rs

1//! A Proxy Connector crate for Hyper based applications
2//!
3//! # Example
4//! ```rust,no_run
5//! use hyper::{Request, Uri, body::Body};
6//! use hyper_util::client::legacy::Client;
7//! use hyper_util::client::legacy::connect::HttpConnector;
8//! use hyper_util::rt::TokioExecutor;
9//! use bytes::Bytes;
10//! use futures_util::{TryFutureExt, TryStreamExt};
11//! use http_body_util::{BodyExt, Empty};
12//! use hyper_http_proxy::{Proxy, ProxyConnector, Intercept};
13//! use headers::Authorization;
14//! use std::error::Error;
15//! use tokio::io::{stdout, AsyncWriteExt as _};
16//!
17//! #[tokio::main]
18//! async fn main() -> Result<(), Box<dyn Error>> {
19//! let proxy = {
20//!         let proxy_uri = "http://my-proxy:8080".parse().unwrap();
21//!         let mut proxy = Proxy::new(Intercept::All, proxy_uri);
22//!         proxy.set_authorization(Authorization::basic("John Doe", "Agent1234"));
23//!         let connector = HttpConnector::new();
24//!         # #[cfg(not(any(feature = "tls", feature = "rustls-base", feature = "openssl-tls")))]
25//!         # let proxy_connector = ProxyConnector::from_proxy_unsecured(connector, proxy);
26//!         # #[cfg(any(feature = "tls", feature = "rustls-base", feature = "openssl"))]
27//!         let proxy_connector = ProxyConnector::from_proxy(connector, proxy).unwrap();
28//!         proxy_connector
29//!     };
30//!
31//!     // Connecting to http will trigger regular GETs and POSTs.
32//!     // We need to manually append the relevant headers to the request
33//!     let uri: Uri = "http://my-remote-website.com".parse().unwrap();
34//!     let mut req = Request::get(uri.clone()).body(Empty::<Bytes>::new()).unwrap();
35//!
36//!     if let Some(headers) = proxy.http_headers(&uri) {
37//!         req.headers_mut().extend(headers.clone().into_iter());
38//!     }
39//!
40//!     let client = Client::builder(TokioExecutor::new()).build(proxy);
41//!     let mut resp = client.request(req).await?;
42//!     println!("Response: {}", resp.status());
43//!     while let Some(chunk) = resp.body_mut().collect().await.ok().map(|c| c.to_bytes()) {
44//!         stdout().write_all(&chunk).await?;
45//!     }
46//!
47//!     // Connecting to an https uri is straightforward (uses 'CONNECT' method underneath)
48//!     let uri = "https://my-remote-websitei-secured.com".parse().unwrap();
49//!     let mut resp = client.get(uri).await?;
50//!     println!("Response: {}", resp.status());
51//!     while let Some(chunk) = resp.body_mut().collect().await.ok().map(|c| c.to_bytes()) {
52//!         stdout().write_all(&chunk).await?;
53//!     }
54//!
55//!     Ok(())
56//! }
57//! ```
58
59#![allow(missing_docs)]
60
61mod rt;
62mod stream;
63mod tunnel;
64
65use std::{fmt, io, sync::Arc};
66use std::{
67    future::Future,
68    pin::Pin,
69    task::{Context, Poll},
70    time::Duration,
71};
72
73use futures_util::future::TryFutureExt;
74use headers::{authorization::Credentials, Authorization, HeaderMapExt, ProxyAuthorization};
75use http::header::{HeaderMap, HeaderName, HeaderValue};
76use hyper::rt::{Read, Write};
77use hyper::Uri;
78use tower_service::Service;
79
80pub use stream::ProxyStream;
81
82#[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
83use native_tls::TlsConnector as NativeTlsConnector;
84
85#[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
86use tokio_native_tls::TlsConnector;
87
88#[cfg(feature = "__rustls")]
89use hyper_rustls::ConfigBuilderExt;
90
91#[cfg(feature = "__rustls")]
92use tokio_rustls::TlsConnector;
93
94#[cfg(feature = "__rustls")]
95use tokio_rustls::rustls::pki_types::ServerName;
96
97type BoxError = Box<dyn std::error::Error + Send + Sync>;
98
99/// The Intercept enum to filter connections
100#[derive(Debug, Clone)]
101pub enum Intercept {
102    /// All incoming connection will go through proxy
103    All,
104    /// Only http connections will go through proxy
105    Http,
106    /// Only https connections will go through proxy
107    Https,
108    /// No connection will go through this proxy
109    None,
110    /// A custom intercept
111    Custom(Custom),
112}
113
114/// A trait for matching between Destination and Uri
115pub trait Dst {
116    /// Returns the connection scheme, e.g. "http" or "https"
117    fn scheme(&self) -> Option<&str>;
118    /// Returns the host of the connection
119    fn host(&self) -> Option<&str>;
120    /// Returns the port for the connection
121    fn port(&self) -> Option<u16>;
122}
123
124impl Dst for Uri {
125    fn scheme(&self) -> Option<&str> {
126        self.scheme_str()
127    }
128
129    fn host(&self) -> Option<&str> {
130        self.host()
131    }
132
133    fn port(&self) -> Option<u16> {
134        self.port_u16()
135    }
136}
137
138#[inline]
139pub(crate) fn io_err<E: Into<Box<dyn std::error::Error + Send + Sync>>>(e: E) -> io::Error {
140    io::Error::new(io::ErrorKind::Other, e)
141}
142
143pub type CustomProxyCallback =
144    dyn Fn(Option<&str>, Option<&str>, Option<u16>) -> bool + Send + Sync;
145
146/// A Custom struct to proxy custom uris
147#[derive(Clone)]
148pub struct Custom(Arc<CustomProxyCallback>);
149
150impl fmt::Debug for Custom {
151    fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
152        write!(f, "_")
153    }
154}
155
156impl<F: Fn(Option<&str>, Option<&str>, Option<u16>) -> bool + Send + Sync + 'static> From<F>
157    for Custom
158{
159    fn from(f: F) -> Custom {
160        Custom(Arc::new(f))
161    }
162}
163
164impl Intercept {
165    /// A function to check if given `Uri` is proxied
166    pub fn matches<D: Dst>(&self, uri: &D) -> bool {
167        match (self, uri.scheme()) {
168            (&Intercept::All, _)
169            | (&Intercept::Http, Some("http"))
170            | (&Intercept::Https, Some("https")) => true,
171            (&Intercept::Custom(Custom(ref f)), _) => f(uri.scheme(), uri.host(), uri.port()),
172            _ => false,
173        }
174    }
175}
176
177impl<F: Fn(Option<&str>, Option<&str>, Option<u16>) -> bool + Send + Sync + 'static> From<F>
178    for Intercept
179{
180    fn from(f: F) -> Intercept {
181        Intercept::Custom(f.into())
182    }
183}
184
185/// A Proxy struct
186#[derive(Clone, Debug)]
187pub struct Proxy {
188    intercept: Intercept,
189    force_connect: bool,
190    headers: HeaderMap,
191    uri: Uri,
192}
193
194impl Proxy {
195    /// Create a new `Proxy`
196    pub fn new<I: Into<Intercept>>(intercept: I, uri: Uri) -> Proxy {
197        let mut proxy = Proxy {
198            intercept: intercept.into(),
199            uri: uri.clone(),
200            headers: HeaderMap::new(),
201            force_connect: false,
202        };
203
204        if let Some((user, pass)) = extract_user_pass(&uri) {
205            proxy.set_authorization(Authorization::basic(user, pass));
206        }
207
208        proxy
209    }
210
211    /// Set `Proxy` authorization
212    pub fn set_authorization<C: Credentials + Clone>(&mut self, credentials: Authorization<C>) {
213        match self.intercept {
214            Intercept::Http => {
215                self.headers.typed_insert(Authorization(credentials.0));
216            }
217            Intercept::Https => {
218                self.headers.typed_insert(ProxyAuthorization(credentials.0));
219            }
220            _ => {
221                self.headers
222                    .typed_insert(Authorization(credentials.0.clone()));
223                self.headers.typed_insert(ProxyAuthorization(credentials.0));
224            }
225        }
226    }
227
228    /// Forces the use of the CONNECT method.
229    pub fn force_connect(&mut self) {
230        self.force_connect = true;
231    }
232
233    /// Set a custom header
234    pub fn set_header(&mut self, name: HeaderName, value: HeaderValue) {
235        self.headers.insert(name, value);
236    }
237
238    /// Get current intercept
239    pub fn intercept(&self) -> &Intercept {
240        &self.intercept
241    }
242
243    /// Get current `Headers` which must be sent to proxy
244    pub fn headers(&self) -> &HeaderMap {
245        &self.headers
246    }
247
248    /// Get proxy uri
249    pub fn uri(&self) -> &Uri {
250        &self.uri
251    }
252}
253
254/// A wrapper around `Proxy`s with a connector.
255#[derive(Clone)]
256pub struct ProxyConnector<C> {
257    proxies: Vec<Proxy>,
258    connector: C,
259    tls_handshake_timeout: Option<Duration>,
260
261    #[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
262    tls: Option<NativeTlsConnector>,
263
264    #[cfg(feature = "__rustls")]
265    tls: Option<TlsConnector>,
266
267    #[cfg(not(feature = "__tls"))]
268    tls: Option<()>,
269}
270
271impl<C: fmt::Debug> fmt::Debug for ProxyConnector<C> {
272    fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
273        write!(
274            f,
275            "ProxyConnector {}{{ proxies: {:?}, connector: {:?} }}",
276            if self.tls.is_some() {
277                ""
278            } else {
279                "(unsecured)"
280            },
281            self.proxies,
282            self.connector
283        )
284    }
285}
286
287impl<C> ProxyConnector<C> {
288    /// Create a new secured Proxies
289    #[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
290    pub fn new(connector: C) -> Result<Self, io::Error> {
291        let tls = NativeTlsConnector::builder()
292            .build()
293            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
294
295        Ok(ProxyConnector {
296            proxies: Vec::new(),
297            connector: connector,
298            tls_handshake_timeout: None,
299            tls: Some(tls),
300        })
301    }
302
303    /// Create a new secured Proxies
304    #[cfg(feature = "__rustls")]
305    pub fn new(connector: C) -> Result<Self, io::Error> {
306        let config = tokio_rustls::rustls::ClientConfig::builder();
307
308        #[cfg(feature = "rustls-tls-native-roots")]
309        let config = config.with_native_roots()?;
310
311        #[cfg(feature = "rustls-tls-webpki-roots")]
312        let config = config.with_webpki_roots();
313
314        let cfg = Arc::new(config.with_no_client_auth());
315        let tls = TlsConnector::from(cfg);
316
317        Ok(ProxyConnector {
318            proxies: Vec::new(),
319            connector,
320            tls_handshake_timeout: None,
321            tls: Some(tls),
322        })
323    }
324
325    /// Create a new unsecured Proxy
326    pub fn unsecured(connector: C) -> Self {
327        ProxyConnector {
328            proxies: Vec::new(),
329            connector,
330            tls_handshake_timeout: None,
331            tls: None,
332        }
333    }
334
335    /// Create a proxy connector and attach a particular proxy
336    #[cfg(feature = "__tls")]
337    pub fn from_proxy(connector: C, proxy: Proxy) -> Result<Self, io::Error> {
338        let mut c = ProxyConnector::new(connector)?;
339        c.proxies.push(proxy);
340        Ok(c)
341    }
342
343    /// Create a proxy connector and attach a particular proxy
344    pub fn from_proxy_unsecured(connector: C, proxy: Proxy) -> Self {
345        let mut c = ProxyConnector::unsecured(connector);
346        c.proxies.push(proxy);
347        c
348    }
349
350    /// Change proxy connector
351    pub fn with_connector<CC>(self, connector: CC) -> ProxyConnector<CC> {
352        ProxyConnector {
353            connector,
354            proxies: self.proxies,
355            tls_handshake_timeout: self.tls_handshake_timeout,
356            tls: self.tls,
357        }
358    }
359
360    /// Set or unset tls when tunneling
361    #[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
362    pub fn set_tls(&mut self, tls: Option<NativeTlsConnector>) {
363        self.tls = tls;
364    }
365
366    /// Set or unset tls when tunneling
367    #[cfg(feature = "__rustls")]
368    pub fn set_tls(&mut self, tls: Option<TlsConnector>) {
369        self.tls = tls;
370    }
371
372    /// Set or unset the timeout bounding the TLS handshake performed when tunneling through
373    /// this proxy to an HTTPS destination.
374    ///
375    /// `None` (the default) means the handshake is not bounded by a timeout.
376    pub fn set_tls_handshake_timeout(&mut self, timeout: Option<Duration>) {
377        self.tls_handshake_timeout = timeout;
378    }
379
380    /// Get the current proxies
381    pub fn proxies(&self) -> &[Proxy] {
382        &self.proxies
383    }
384
385    /// Add a new additional proxy
386    pub fn add_proxy(&mut self, proxy: Proxy) {
387        self.proxies.push(proxy);
388    }
389
390    /// Extend the list of proxies
391    pub fn extend_proxies<I: IntoIterator<Item = Proxy>>(&mut self, proxies: I) {
392        self.proxies.extend(proxies)
393    }
394
395    /// Get http headers for a matching uri
396    ///
397    /// These headers must be appended to the hyper Request for the proxy to work properly.
398    /// This is needed only for http requests.
399    pub fn http_headers(&self, uri: &Uri) -> Option<&HeaderMap> {
400        if uri.scheme_str() != Some("http") {
401            return None;
402        }
403
404        self.match_proxy(uri).map(|p| &p.headers)
405    }
406
407    fn match_proxy<D: Dst>(&self, uri: &D) -> Option<&Proxy> {
408        self.proxies.iter().find(|p| p.intercept.matches(uri))
409    }
410}
411
412macro_rules! mtry {
413    ($e:expr) => {
414        match $e {
415            Ok(v) => v,
416            Err(e) => break Err(e.into()),
417        }
418    };
419}
420
421impl<C> Service<Uri> for ProxyConnector<C>
422where
423    C: Service<Uri>,
424    C::Response: Read + Write + Send + Unpin + 'static,
425    C::Future: Send + 'static,
426    C::Error: Into<BoxError>,
427{
428    type Response = ProxyStream<C::Response>;
429    type Error = io::Error;
430    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
431
432    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
433        match self.connector.poll_ready(cx) {
434            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
435            Poll::Ready(Err(e)) => Poll::Ready(Err(io_err(e.into()))),
436            Poll::Pending => Poll::Pending,
437        }
438    }
439
440    fn call(&mut self, uri: Uri) -> Self::Future {
441        if let (Some(p), Some(host)) = (self.match_proxy(&uri), uri.host()) {
442            if uri.scheme() == Some(&http::uri::Scheme::HTTPS) || p.force_connect {
443                let host = host.to_owned();
444                let port =
445                    uri.port_u16()
446                        .unwrap_or(if uri.scheme() == Some(&http::uri::Scheme::HTTP) {
447                            80
448                        } else {
449                            443
450                        });
451
452                let tunnel = tunnel::new(&host, port, &p.headers);
453                let connection =
454                    proxy_dst(&uri, &p.uri).map(|proxy_url| self.connector.call(proxy_url));
455                let tls = if uri.scheme() == Some(&http::uri::Scheme::HTTPS) {
456                    self.tls.clone()
457                } else {
458                    None
459                };
460                #[cfg(feature = "__tls")]
461                let tls_handshake_timeout = self.tls_handshake_timeout;
462
463                Box::pin(async move {
464                    // this hack will gone once `try_blocks` will eventually stabilized
465                    #[allow(clippy::never_loop)]
466                    loop {
467                        let proxy_stream = mtry!(mtry!(connection).await.map_err(io_err));
468                        let tunnel_stream = mtry!(tunnel.with_stream(proxy_stream).await);
469
470                        break match tls {
471                            #[cfg(all(not(feature = "__rustls"), feature = "native-tls"))]
472                            Some(tls) => {
473                                use hyper_util::rt::TokioIo;
474                                let tls = TlsConnector::from(tls);
475                                let secure_stream = mtry!(mtry!(
476                                    with_optional_timeout(
477                                        tls_handshake_timeout,
478                                        tls.connect(&host, TokioIo::new(tunnel_stream))
479                                    )
480                                    .await
481                                )
482                                .map_err(io_err));
483
484                                Ok(ProxyStream::Secured(Box::new(TokioIo::new(secure_stream))))
485                            }
486
487                            #[cfg(feature = "__rustls")]
488                            Some(tls) => {
489                                use hyper_util::rt::TokioIo;
490                                let server_name =
491                                    mtry!(ServerName::try_from(host.to_string()).map_err(io_err));
492                                let secure_stream = mtry!(mtry!(
493                                    with_optional_timeout(
494                                        tls_handshake_timeout,
495                                        tls.connect(server_name, TokioIo::new(tunnel_stream))
496                                    )
497                                    .await
498                                )
499                                .map_err(io_err));
500
501                                Ok(ProxyStream::Secured(Box::new(TokioIo::new(secure_stream))))
502                            }
503
504                            #[cfg(not(feature = "__tls",))]
505                            Some(_) => panic!("hyper-proxy was not built with TLS support"),
506
507                            None => Ok(ProxyStream::Regular(tunnel_stream)),
508                        };
509                    }
510                })
511            } else {
512                match proxy_dst(&uri, &p.uri) {
513                    Ok(proxy_uri) => Box::pin(
514                        self.connector
515                            .call(proxy_uri)
516                            .map_ok(ProxyStream::Regular)
517                            .map_err(|err| io_err(err.into())),
518                    ),
519                    Err(err) => Box::pin(futures_util::future::err(io_err(err))),
520                }
521            }
522        } else {
523            Box::pin(
524                self.connector
525                    .call(uri)
526                    .map_ok(ProxyStream::NoProxy)
527                    .map_err(|err| io_err(err.into())),
528            )
529        }
530    }
531}
532
533/// Awaits `fut`, bounding it by `timeout` when set.
534///
535/// A `None` timeout awaits `fut` directly, so it never yields the outer timeout error.
536#[cfg(feature = "__tls")]
537async fn with_optional_timeout<F: Future>(
538    timeout: Option<Duration>,
539    fut: F,
540) -> io::Result<F::Output> {
541    match timeout {
542        Some(timeout) => tokio::time::timeout(timeout, fut)
543            .await
544            .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "TLS handshake timed out")),
545        None => Ok(fut.await),
546    }
547}
548
549fn proxy_dst(dst: &Uri, proxy: &Uri) -> io::Result<Uri> {
550    Uri::builder()
551        .scheme(
552            proxy
553                .scheme_str()
554                .ok_or_else(|| io_err(format!("proxy uri missing scheme: {}", proxy)))?,
555        )
556        .authority(
557            proxy
558                .authority()
559                .ok_or_else(|| io_err(format!("proxy uri missing host: {}", proxy)))?
560                .clone(),
561        )
562        .path_and_query(dst.path_and_query().unwrap().clone())
563        .build()
564        .map_err(|err| io_err(format!("other error: {}", err)))
565}
566
567/// Extracts the username and password from the URI
568fn extract_user_pass(uri: &Uri) -> Option<(&str, &str)> {
569    let authority = uri.authority()?.as_str();
570    let (userinfo, _) = authority.rsplit_once('@')?;
571    let (username, password) = userinfo.split_once(':')?;
572    Some((username, password))
573}
574
575#[cfg(test)]
576mod tests {
577    use http::Uri;
578
579    use crate::{Intercept, Proxy};
580
581    #[test]
582    fn test_new_proxy_with_authorization() {
583        let proxy = Proxy::new(
584            Intercept::All,
585            Uri::from_static("https://bob:secret@my-proxy:8080"),
586        );
587
588        assert_eq!(
589            proxy
590                .headers()
591                .get("authorization")
592                .unwrap()
593                .to_str()
594                .unwrap(),
595            "Basic Ym9iOnNlY3JldA=="
596        );
597    }
598
599    #[test]
600    fn test_new_proxy_without_authorization() {
601        let proxy = Proxy::new(Intercept::All, Uri::from_static("https://my-proxy:8080"));
602
603        assert_eq!(proxy.headers().get("authorization"), None);
604    }
605
606    #[cfg(feature = "__tls")]
607    #[tokio::test]
608    async fn tls_handshake_timeout_fires_against_a_stalled_proxy_tunnel() {
609        use std::time::Duration;
610
611        use hyper_util::client::legacy::connect::HttpConnector;
612        use tokio::io::{AsyncReadExt, AsyncWriteExt};
613        use tokio::net::TcpListener;
614        use tokio::time::timeout;
615        use tower_service::Service as _;
616
617        use crate::{Proxy, ProxyConnector};
618
619        #[cfg(feature = "__rustls")]
620        let _ = tokio_rustls::rustls::crypto::ring::default_provider().install_default();
621
622        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
623        let addr = listener.local_addr().unwrap();
624
625        // Emulates a proxy that completes the `CONNECT` tunnel but then never speaks TLS, so the
626        // client's handshake deadline is the only thing that can end the connection attempt.
627        let proxy_task = tokio::spawn(async move {
628            let (mut stream, _) = listener.accept().await.unwrap();
629            let mut buf = [0u8; 4096];
630            let _ = stream.read(&mut buf).await.unwrap();
631            stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").await.unwrap();
632            tokio::time::sleep(Duration::from_secs(30)).await;
633            drop(stream);
634        });
635
636        let proxy_uri: Uri = format!("http://{addr}").parse().unwrap();
637        let proxy = Proxy::new(Intercept::All, proxy_uri);
638
639        let mut proxy_connector = ProxyConnector::new(HttpConnector::new()).unwrap();
640        proxy_connector.add_proxy(proxy);
641        proxy_connector.set_tls_handshake_timeout(Some(Duration::from_millis(200)));
642
643        let dst: Uri = "https://example.invalid/".parse().unwrap();
644        let result = timeout(Duration::from_secs(5), proxy_connector.call(dst))
645            .await
646            .expect("request should not hit the outer test timeout");
647        let error = match result {
648            Ok(_) => panic!("handshake should time out before completing"),
649            Err(error) => error,
650        };
651
652        assert!(
653            error.to_string().contains("TLS handshake timed out"),
654            "expected a TLS handshake timeout, got: {error}"
655        );
656
657        proxy_task.abort();
658    }
659}
hyper_http_proxy/lib.rs

hyper_http_proxy/
lib.rs
