ferron_common/http_proxy/
mod.rs

1mod builder;
2mod load_balancer;
3mod proxy_client;
4mod request_parts;
5mod send_net_io;
6mod send_request;
7
8use std::collections::HashMap;
9use std::error::Error;
10use std::net::IpAddr;
11use std::pin::Pin;
12use std::str::FromStr;
13use std::sync::atomic::AtomicUsize;
14use std::sync::Arc;
15use std::task::{Context, Poll, Waker};
16use std::time::Duration;
17
18use async_trait::async_trait;
19use bytes::Bytes;
20use connpool::{Item, Pool};
21use futures_util::FutureExt;
22use http_body_util::combinators::BoxBody;
23use hyper::header::{self, HeaderName};
24use hyper::{Request, StatusCode, Uri};
25#[cfg(feature = "runtime-monoio")]
26use monoio::net::TcpStream;
27#[cfg(all(feature = "runtime-monoio", unix))]
28use monoio::net::UnixStream;
29use rustls::client::WebPkiServerVerifier;
30use rustls_pki_types::ServerName;
31use rustls_platform_verifier::BuilderVerifierExt;
32use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
33#[cfg(feature = "runtime-tokio")]
34use tokio::net::TcpStream;
35#[cfg(all(feature = "runtime-tokio", unix))]
36use tokio::net::UnixStream;
37use tokio::sync::RwLock;
38use tokio_rustls::TlsConnector;
39#[cfg(feature = "runtime-vibeio")]
40use vibeio::net::TcpStream;
41#[cfg(all(feature = "runtime-vibeio", unix))]
42use vibeio::net::UnixStream;
43
44use crate::config::ServerConfiguration;
45use crate::http_proxy::send_request::SendRequestWrapper;
46use crate::logging::ErrorLogger;
47use crate::modules::{ModuleHandlers, ResponseData, SocketData};
48use crate::observability::{Metric, MetricAttributeValue, MetricType, MetricValue, MetricsMultiSender};
49use crate::util::{NoServerVerifier, TtlCache};
50
51pub use self::builder::ReverseProxyBuilder;
52#[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
53use self::send_net_io::{SendTcpStreamPoll, SendTcpStreamPollDropGuard};
54#[cfg(all(any(feature = "runtime-vibeio", feature = "runtime-monoio"), unix))]
55use self::send_net_io::{SendUnixStreamPoll, SendUnixStreamPollDropGuard};
56use self::{
57  load_balancer::{determine_proxy_to, resolve_upstreams},
58  proxy_client::{http_proxy, http_proxy_handshake},
59  request_parts::construct_proxy_request_parts,
60};
61
62type ConnectionsTrackState = Arc<RwLock<HashMap<UpstreamInner, Arc<()>>>>;
63
64enum LoadBalancerAlgorithmInner {
65  Random,
66  RoundRobin(Arc<AtomicUsize>),
67  LeastConnections(ConnectionsTrackState),
68  TwoRandomChoices(ConnectionsTrackState),
69}
70
71/// Backend selection strategy used when multiple upstreams are configured.
72#[derive(Clone, Copy, Hash, PartialEq, Eq)]
73pub enum LoadBalancerAlgorithm {
74  /// Selects a backend randomly for each request.
75  Random,
76  /// Cycles through backends in order.
77  RoundRobin,
78  /// Selects the backend with the least active tracked connections.
79  LeastConnections,
80  /// Chooses two random backends and picks the less loaded one.
81  TwoRandomChoices,
82}
83
84/// Proxy protocol version to prepend to upstream connections.
85#[derive(Clone, Copy)]
86pub enum ProxyHeader {
87  /// HAProxy PROXY protocol v1.
88  V1,
89  /// HAProxy PROXY protocol v2.
90  V2,
91}
92
93#[derive(Clone, Eq, PartialEq, Hash)]
94struct UpstreamInner {
95  proxy_to: String,
96  proxy_unix: Option<String>,
97}
98
99#[derive(Clone)]
100struct SrvUpstreamData {
101  to: String,
102  secondary_runtime_handle: tokio::runtime::Handle,
103  dns_resolver: Option<Arc<hickory_resolver::TokioResolver>>,
104}
105
106impl PartialEq for SrvUpstreamData {
107  fn eq(&self, other: &Self) -> bool {
108    self.to == other.to
109  }
110}
111
112impl Eq for SrvUpstreamData {}
113
114impl std::hash::Hash for SrvUpstreamData {
115  fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
116    self.to.hash(state);
117  }
118}
119
120#[derive(Clone, Eq, PartialEq, Hash)]
121enum Upstream {
122  Static(UpstreamInner),
123  Srv(SrvUpstreamData),
124}
125
126impl Upstream {
127  async fn resolve(
128    &self,
129    failed_backends: Arc<RwLock<TtlCache<UpstreamInner, u64>>>,
130    health_check_max_fails: u64,
131  ) -> Vec<UpstreamInner> {
132    match self {
133      Upstream::Static(inner) => vec![inner.clone()],
134      Upstream::Srv(srv_data) => {
135        let to = srv_data.to.clone();
136        let resolver = srv_data.dns_resolver.clone();
137        let failed_backends = failed_backends.clone();
138        srv_data
139          .secondary_runtime_handle
140          .spawn(async move {
141            let to_url = match Uri::from_str(&to) {
142              Ok(uri) => uri,
143              Err(_) => return vec![],
144            };
145            let to = match to_url.host() {
146              Some(host) => host.to_string(),
147              None => return vec![],
148            };
149            let resolver = match resolver {
150              Some(resolver) => resolver,
151              None => return vec![],
152            };
153
154            let srv_records = match resolver.srv_lookup(&to).await {
155              Ok(records) => records,
156              Err(_) => return vec![],
157            };
158
159            let failed_backends = failed_backends.read().await;
160            let srv_upstreams = srv_records
161              .answers()
162              .iter()
163              .filter_map(|record| {
164                let record = match &record.data {
165                  hickory_resolver::proto::rr::RData::SRV(srv) => srv,
166                  _ => return None,
167                };
168                let mut to_url_parts = to_url.clone().into_parts();
169                to_url_parts.authority = Some(format!("{}:{}", record.target, record.port).parse().ok()?);
170                let upstream_inner = UpstreamInner {
171                  proxy_to: Uri::from_parts(to_url_parts).ok()?.to_string(),
172                  proxy_unix: None,
173                };
174                if failed_backends
175                  .get(&upstream_inner)
176                  .is_some_and(|fails| fails > health_check_max_fails)
177                {
178                  // Backend is unhealthy, skip it
179                  None
180                } else {
181                  Some((upstream_inner, record.weight, record.priority))
182                }
183              })
184              .collect::<Vec<_>>();
185            let highest_priority = srv_upstreams
186              .iter()
187              .map(|(_, _, priority)| *priority)
188              .min()
189              .unwrap_or(0);
190            let filtered_srv_upstreams = srv_upstreams
191              .into_iter()
192              .filter(|(_, _, priority)| *priority == highest_priority)
193              .map(|(upstream, weight, _)| (upstream, weight))
194              .collect::<Vec<_>>();
195            let cumulative_weight: u64 = filtered_srv_upstreams.iter().map(|(_, weight)| *weight as u64).sum();
196            let mut random_weight = if cumulative_weight == 0 {
197              // Prevent empty range sampling panics
198              0
199            } else {
200              rand::random_range(0..cumulative_weight)
201            };
202            for upstream in filtered_srv_upstreams {
203              let weight = upstream.1;
204              if random_weight <= weight as u64 {
205                return vec![upstream.0];
206              }
207              random_weight -= weight as u64;
208            }
209            vec![]
210          })
211          .await
212          .unwrap_or(vec![])
213      }
214    }
215  }
216}
217
218type ProxyToKey = (Upstream, Option<usize>, Option<Duration>);
219type ProxyToKeyInner = (UpstreamInner, Option<usize>, Option<Duration>);
220
221type ConnectionPool = Arc<Pool<(UpstreamInner, Option<IpAddr>), SendRequestWrapper>>;
222type ConnectionPoolItem = Item<(UpstreamInner, Option<IpAddr>), SendRequestWrapper>;
223
224#[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
225#[allow(unused)]
226enum DropGuard {
227  Tcp(SendTcpStreamPollDropGuard),
228  #[cfg(unix)]
229  Unix(SendUnixStreamPollDropGuard),
230}
231
232enum Connection {
233  #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
234  Tcp(SendTcpStreamPoll),
235  #[cfg(not(any(feature = "runtime-vibeio", feature = "runtime-monoio")))]
236  Tcp(TcpStream),
237  #[cfg(all(any(feature = "runtime-vibeio", feature = "runtime-monoio"), unix))]
238  Unix(SendUnixStreamPoll),
239  #[cfg(all(not(any(feature = "runtime-vibeio", feature = "runtime-monoio")), unix))]
240  Unix(UnixStream),
241}
242
243#[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
244impl Connection {
245  unsafe fn get_drop_guard(&mut self) -> DropGuard {
246    match self {
247      Connection::Tcp(stream) => DropGuard::Tcp(stream.get_drop_guard()),
248      #[cfg(unix)]
249      Connection::Unix(stream) => DropGuard::Unix(stream.get_drop_guard()),
250    }
251  }
252}
253
254impl AsyncRead for Connection {
255  fn poll_read(
256    mut self: Pin<&mut Self>,
257    cx: &mut Context<'_>,
258    buf: &mut tokio::io::ReadBuf,
259  ) -> Poll<Result<(), std::io::Error>> {
260    match &mut *self {
261      Connection::Tcp(stream) => Pin::new(stream).poll_read(cx, buf),
262      #[cfg(unix)]
263      Connection::Unix(stream) => Pin::new(stream).poll_read(cx, buf),
264    }
265  }
266}
267
268impl AsyncWrite for Connection {
269  fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, std::io::Error>> {
270    match &mut *self {
271      Connection::Tcp(stream) => Pin::new(stream).poll_write(cx, buf),
272      #[cfg(unix)]
273      Connection::Unix(stream) => Pin::new(stream).poll_write(cx, buf),
274    }
275  }
276
277  fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
278    match &mut *self {
279      Connection::Tcp(stream) => Pin::new(stream).poll_flush(cx),
280      #[cfg(unix)]
281      Connection::Unix(stream) => Pin::new(stream).poll_flush(cx),
282    }
283  }
284
285  fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
286    match &mut *self {
287      Connection::Tcp(stream) => Pin::new(stream).poll_shutdown(cx),
288      #[cfg(unix)]
289      Connection::Unix(stream) => Pin::new(stream).poll_shutdown(cx),
290    }
291  }
292
293  fn is_write_vectored(&self) -> bool {
294    match self {
295      Connection::Tcp(stream) => stream.is_write_vectored(),
296      #[cfg(unix)]
297      Connection::Unix(stream) => stream.is_write_vectored(),
298    }
299  }
300
301  fn poll_write_vectored(
302    mut self: Pin<&mut Self>,
303    cx: &mut Context<'_>,
304    bufs: &[std::io::IoSlice<'_>],
305  ) -> Poll<Result<usize, std::io::Error>> {
306    match &mut *self {
307      Connection::Tcp(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
308      #[cfg(unix)]
309      Connection::Unix(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
310    }
311  }
312}
313
314/// Connection pool for reverse proxy
315pub struct Connections {
316  #[allow(clippy::type_complexity)]
317  load_balancer_cache: HashMap<
318    (
319      LoadBalancerAlgorithm,
320      Arc<Vec<(Upstream, Option<usize>, Option<Duration>)>>,
321    ),
322    Arc<LoadBalancerAlgorithmInner>,
323  >,
324  #[allow(clippy::type_complexity)]
325  failed_backend_cache: HashMap<
326    (Duration, u64, Arc<Vec<(Upstream, Option<usize>, Option<Duration>)>>),
327    Arc<RwLock<TtlCache<UpstreamInner, u64>>>,
328  >,
329  connections: ConnectionPool,
330  #[cfg(unix)]
331  unix_connections: ConnectionPool,
332}
333
334impl Connections {
335  /// Creates a connection pool without a global connection limit.
336  pub fn new() -> Self {
337    Self {
338      load_balancer_cache: HashMap::new(),
339      failed_backend_cache: HashMap::new(),
340      connections: Arc::new(Pool::new_unbounded()),
341      #[cfg(unix)]
342      unix_connections: Arc::new(Pool::new_unbounded()),
343    }
344  }
345
346  /// Creates a connection pool with a global TCP connection limit.
347  ///
348  /// Unix socket connections remain unbounded.
349  pub fn with_global_limit(global_limit: usize) -> Self {
350    Self {
351      load_balancer_cache: HashMap::new(),
352      failed_backend_cache: HashMap::new(),
353      connections: Arc::new(Pool::new(global_limit)),
354      #[cfg(unix)]
355      unix_connections: Arc::new(Pool::new_unbounded()),
356    }
357  }
358
359  /// Starts a reverse proxy builder using this connection pool.
360  pub fn get_builder<'a>(&'a mut self) -> ReverseProxyBuilder<'a> {
361    ReverseProxyBuilder {
362      connections: self,
363      upstreams: Vec::new(),
364      lb_algorithm: LoadBalancerAlgorithm::TwoRandomChoices,
365      lb_health_check_window: Duration::from_millis(5000),
366      lb_health_check_max_fails: 3,
367      lb_health_check: false,
368      proxy_no_verification: false,
369      proxy_intercept_errors: false,
370      lb_retry_connection: true,
371      proxy_http2_only: false,
372      proxy_http2: false,
373      proxy_keepalive: true,
374      proxy_proxy_header: None,
375      proxy_request_header: Vec::new(),
376      proxy_request_header_replace: Vec::new(),
377      proxy_request_header_remove: Vec::new(),
378      rewrite_host: false,
379    }
380  }
381}
382
383impl Default for Connections {
384  fn default() -> Self {
385    Self::new()
386  }
387}
388
389/// A reverse proxy
390pub struct ReverseProxy {
391  #[allow(clippy::type_complexity)]
392  failed_backends: Arc<RwLock<TtlCache<UpstreamInner, u64>>>,
393  load_balancer_algorithm: Arc<LoadBalancerAlgorithmInner>,
394  proxy_to: Arc<Vec<ProxyToKey>>,
395  health_check_max_fails: u64,
396  enable_health_check: bool,
397  disable_certificate_verification: bool,
398  proxy_intercept_errors: bool,
399  retry_connection: bool,
400  proxy_http2_only: bool,
401  proxy_http2: bool,
402  proxy_keepalive: bool,
403  proxy_header: Option<ProxyHeader>,
404  headers_to_add: Arc<Vec<(HeaderName, String)>>,
405  headers_to_replace: Arc<Vec<(HeaderName, String)>>,
406  headers_to_remove: Arc<Vec<HeaderName>>,
407  rewrite_host: bool,
408  connections: ConnectionPool,
409  #[cfg(unix)]
410  unix_connections: ConnectionPool,
411}
412
413impl ReverseProxy {
414  /// Creates a request handler instance with shared proxy state.
415  pub fn get_handler(&self) -> ReverseProxyHandler {
416    ReverseProxyHandler {
417      failed_backends: self.failed_backends.clone(),
418      load_balancer_algorithm: self.load_balancer_algorithm.clone(),
419      proxy_to: self.proxy_to.clone(),
420      health_check_max_fails: self.health_check_max_fails,
421      selected_backends_metrics: None,
422      unhealthy_backends_metrics: None,
423      connection_reused: false,
424      enable_health_check: self.enable_health_check,
425      disable_certificate_verification: self.disable_certificate_verification,
426      proxy_intercept_errors: self.proxy_intercept_errors,
427      retry_connection: self.retry_connection,
428      proxy_http2_only: self.proxy_http2_only,
429      proxy_http2: self.proxy_http2,
430      proxy_keepalive: self.proxy_keepalive,
431      proxy_header: self.proxy_header,
432      headers_to_add: self.headers_to_add.clone(),
433      headers_to_replace: self.headers_to_replace.clone(),
434      headers_to_remove: self.headers_to_remove.clone(),
435      rewrite_host: self.rewrite_host,
436      connections: self.connections.clone(),
437      #[cfg(unix)]
438      unix_connections: self.unix_connections.clone(),
439    }
440  }
441}
442
443/// Handlers for the reverse proxy module
444pub struct ReverseProxyHandler {
445  #[allow(clippy::type_complexity)]
446  failed_backends: Arc<RwLock<TtlCache<UpstreamInner, u64>>>,
447  load_balancer_algorithm: Arc<LoadBalancerAlgorithmInner>,
448  proxy_to: Arc<Vec<ProxyToKey>>,
449  health_check_max_fails: u64,
450  selected_backends_metrics: Option<Vec<UpstreamInner>>,
451  unhealthy_backends_metrics: Option<Vec<UpstreamInner>>,
452  connection_reused: bool,
453  enable_health_check: bool,
454  disable_certificate_verification: bool,
455  proxy_intercept_errors: bool,
456  retry_connection: bool,
457  proxy_http2_only: bool,
458  proxy_http2: bool,
459  proxy_keepalive: bool,
460  proxy_header: Option<ProxyHeader>,
461  headers_to_add: Arc<Vec<(HeaderName, String)>>,
462  headers_to_replace: Arc<Vec<(HeaderName, String)>>,
463  headers_to_remove: Arc<Vec<HeaderName>>,
464  rewrite_host: bool,
465  connections: ConnectionPool,
466  #[cfg(unix)]
467  unix_connections: ConnectionPool,
468}
469
470impl ReverseProxyHandler {
471  #[inline]
472  fn status_response(status_code: StatusCode) -> ResponseData {
473    ResponseData {
474      request: None,
475      response: None,
476      response_status: Some(status_code),
477      response_headers: None,
478      new_remote_address: None,
479    }
480  }
481
482  async fn mark_backend_failure(&mut self, upstream: &UpstreamInner) {
483    if !self.enable_health_check {
484      return;
485    }
486    if let Some(unhealthy_backends_metrics) = self.unhealthy_backends_metrics.as_mut() {
487      unhealthy_backends_metrics.push(upstream.clone());
488    }
489    let mut failed_backends_write = self.failed_backends.write().await;
490    let failed_attempts = failed_backends_write.get(upstream);
491    failed_backends_write.insert(upstream.clone(), failed_attempts.map_or(1, |x| x + 1));
492  }
493
494  async fn retry_or_respond(
495    &self,
496    error_logger: &ErrorLogger,
497    err: &dyn std::fmt::Display,
498    retry_connection: bool,
499    has_more_backends: bool,
500    status_code: StatusCode,
501    log_prefix: &str,
502  ) -> Option<ResponseData> {
503    if retry_connection && has_more_backends {
504      error_logger
505        .log(&format!("Failed to connect to backend, trying another backend: {err}"))
506        .await;
507      None
508    } else {
509      error_logger.log(&format!("{log_prefix}: {err}")).await;
510      Some(Self::status_response(status_code))
511    }
512  }
513
514  #[inline]
515  fn io_error_status(err: &std::io::Error) -> (StatusCode, &'static str) {
516    match err.kind() {
517      std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound | std::io::ErrorKind::HostUnreachable => {
518        (StatusCode::SERVICE_UNAVAILABLE, "Service unavailable")
519      }
520      std::io::ErrorKind::TimedOut => (StatusCode::GATEWAY_TIMEOUT, "Gateway timeout"),
521      _ => (StatusCode::BAD_GATEWAY, "Bad gateway"),
522    }
523  }
524}
525
526#[async_trait(?Send)]
527impl ModuleHandlers for ReverseProxyHandler {
528  /// Handles incoming HTTP requests and proxies them to the configured backend server(s)
529  ///
530  /// This handler:
531  /// 1. Determines which backend server to proxy to (supports load balancing)
532  /// 2. Transforms the request by:
533  ///    - Converting the URL to match the backend format
534  ///    - Setting appropriate headers (Host, X-Forwarded-*)
535  /// 3. Establishes a connection to the backend (HTTP or HTTPS)
536  /// 4. Forwards the request and returns the response
537  ///
538  /// The handler supports:
539  /// - Load balancing across multiple backends
540  /// - Connection pooling/reuse
541  /// - Health checking (marking failed backends)
542  /// - TLS/SSL for secure connections
543  /// - HTTP protocol upgrades (e.g., WebSockets)
544  async fn request_handler(
545    &mut self,
546    request: Request<BoxBody<Bytes, std::io::Error>>,
547    config: &ServerConfiguration,
548    socket_data: &SocketData,
549    error_logger: &ErrorLogger,
550  ) -> Result<ResponseData, Box<dyn Error + Send + Sync>> {
551    let enable_health_check = self.enable_health_check;
552    let health_check_max_fails = self.health_check_max_fails;
553    let disable_certificate_verification = self.disable_certificate_verification;
554    let proxy_intercept_errors = self.proxy_intercept_errors;
555    if self.proxy_to.is_empty() {
556      // No upstreams configured...
557      return Ok(ResponseData {
558        request: Some(request),
559        response: None,
560        response_status: None,
561        response_headers: None,
562        new_remote_address: None,
563      });
564    }
565    let mut proxy_to_vector = resolve_upstreams(
566      &self.proxy_to,
567      self.failed_backends.clone(),
568      self.health_check_max_fails,
569    )
570    .await;
571    let load_balancer_algorithm = self.load_balancer_algorithm.clone();
572    let connection_track = match &*load_balancer_algorithm {
573      LoadBalancerAlgorithmInner::LeastConnections(connection_track) => Some(connection_track),
574      LoadBalancerAlgorithmInner::TwoRandomChoices(connection_track) => Some(connection_track),
575      _ => None,
576    };
577    let retry_connection = self.retry_connection;
578    let (request_parts, request_body) = request.into_parts();
579    let mut request_parts = Some(request_parts);
580
581    loop {
582      if let Some((upstream, local_limit_index, keepalive_idle_timeout)) = determine_proxy_to(
583        &mut proxy_to_vector,
584        &self.failed_backends,
585        enable_health_check,
586        health_check_max_fails,
587        &load_balancer_algorithm,
588      )
589      .await
590      {
591        if let Some(selected_backends_metrics) = self.selected_backends_metrics.as_mut() {
592          selected_backends_metrics.push(upstream.clone());
593        }
594        let UpstreamInner { proxy_to, proxy_unix } = &upstream;
595        let proxy_request_url = proxy_to.parse::<hyper::Uri>()?;
596        let scheme_str = proxy_request_url.scheme_str();
597        let mut encrypted = false;
598
599        match scheme_str {
600          Some("http") => {
601            encrypted = false;
602          }
603          Some("https") => {
604            encrypted = true;
605          }
606          _ => Err(anyhow::anyhow!("Only HTTP and HTTPS reverse proxy URLs are supported."))?,
607        };
608
609        let host = match proxy_request_url.host() {
610          Some(host) => host,
611          None => Err(anyhow::anyhow!("The reverse proxy URL doesn't include the host"))?,
612        };
613
614        let port = proxy_request_url.port_u16().unwrap_or(match scheme_str {
615          Some("http") => 80,
616          Some("https") => 443,
617          _ => 80,
618        });
619
620        let addr = format!("{host}:{port}");
621
622        let request_parts_option = if proxy_to_vector.is_empty() {
623          request_parts.take()
624        } else {
625          request_parts.clone()
626        };
627        let request_parts = request_parts_option.ok_or(anyhow::anyhow!("Request parts not found"))?;
628        let proxy_request_parts = construct_proxy_request_parts(
629          request_parts,
630          config,
631          socket_data,
632          &proxy_request_url,
633          &self.headers_to_add,
634          &self.headers_to_replace,
635          &self.headers_to_remove,
636          self.rewrite_host,
637        )?;
638
639        let tracked_connection = if let Some(connection_track) = connection_track {
640          let connection_track_read = connection_track.read().await;
641          Some(if let Some(connection_count) = connection_track_read.get(&upstream) {
642            connection_count.clone()
643          } else {
644            let tracked_connection = Arc::new(());
645            drop(connection_track_read);
646            connection_track
647              .write()
648              .await
649              .insert(upstream.clone(), tracked_connection.clone());
650            tracked_connection
651          })
652        } else {
653          None
654        };
655
656        let proxy_header = self.proxy_header;
657
658        let is_http_upgrade = proxy_request_parts.headers.contains_key(header::UPGRADE);
659        let enable_http2_only_config = self.proxy_http2_only;
660        let enable_http2_config = self.proxy_http2;
661
662        let enable_keepalive =
663          (enable_http2_only_config || !enable_http2_config || !is_http_upgrade) && self.proxy_keepalive;
664        let connection_pool_item = {
665          #[cfg(unix)]
666          let connections = if proxy_unix.is_some() {
667            &self.unix_connections
668          } else {
669            &self.connections
670          };
671          #[cfg(not(unix))]
672          let connections = &self.connections;
673          let sender;
674          let mut send_request_items = Vec::new();
675          let proxy_client_ip = match proxy_header {
676            Some(ProxyHeader::V1) | Some(ProxyHeader::V2) => Some(socket_data.remote_addr.ip().to_canonical()),
677            _ => None,
678          };
679          loop {
680            let mut send_request_item = if send_request_items.is_empty() {
681              connections
682                .pull_with_wait_local_limit((upstream.clone(), proxy_client_ip), local_limit_index)
683                .await
684            } else if let Poll::Ready(send_request_item_option) = connections
685              .pull_with_wait_local_limit((upstream.clone(), proxy_client_ip), local_limit_index)
686              .boxed_local()
687              .poll_unpin(&mut Context::from_waker(Waker::noop()))
688            {
689              send_request_item_option
690            } else {
691              let send_request_items_taken = send_request_items;
692              send_request_items = Vec::new();
693              let fetch_nonready_send_request_fut = async {
694                let result = futures_util::future::select_ok(send_request_items_taken).await;
695                if let Ok((item, send_request_items_smaller)) = result {
696                  send_request_items = send_request_items_smaller;
697                  item
698                } else {
699                  futures_util::future::pending().await
700                }
701              };
702              crate::runtime::select! {
703                item = connections
704                  .pull_with_wait_local_limit((upstream.clone(), proxy_client_ip), local_limit_index)
705                => {
706                  item
707                },
708                item = fetch_nonready_send_request_fut => {
709                  item
710                }
711              }
712            };
713            if let Some(send_request) = send_request_item.inner_mut() {
714              match send_request.get(keepalive_idle_timeout) {
715                (Some(send_request), true) => {
716                  // Connection ready, send a request to it
717                  send_request_items.clear();
718                  self.connection_reused = true;
719                  let _ = send_request_item.inner_mut().take();
720                  let proxy_request = Request::from_parts(proxy_request_parts, request_body);
721                  let result = http_proxy(
722                    send_request,
723                    send_request_item,
724                    proxy_request,
725                    error_logger,
726                    proxy_intercept_errors,
727                    tracked_connection,
728                    true,
729                  )
730                  .await;
731                  return result;
732                }
733                (None, true) => {
734                  // Connection not ready
735                  send_request_items.push(Box::pin(async move {
736                    let inner_item = send_request_item.inner_mut();
737                    if let Some(inner_item_2) = inner_item {
738                      if !inner_item_2.wait_ready(keepalive_idle_timeout).await {
739                        // Connection closed or timed out
740                        inner_item.take();
741                        return Err(());
742                      }
743                      let _ = inner_item;
744                      Ok(send_request_item)
745                    } else {
746                      Err(())
747                    }
748                  }));
749                  continue;
750                }
751                (_, false) => {
752                  // Connection closed
753                  let _ = send_request_item.inner_mut().take();
754                  continue;
755                }
756              }
757            }
758            send_request_items.clear();
759            sender = send_request_item;
760            break;
761          }
762          sender
763        };
764
765        let stream = if let Some(proxy_unix_str) = &proxy_unix {
766          #[cfg(not(unix))]
767          {
768            let _ = proxy_unix_str; // Discard the variable to avoid unused variable warning
769            Err(anyhow::anyhow!("Unix sockets are not supported on this platform"))?
770          }
771
772          #[cfg(unix)]
773          {
774            let stream = match UnixStream::connect(proxy_unix_str).await {
775              Ok(stream) => stream,
776              Err(err) => {
777                self.mark_backend_failure(&upstream).await;
778                let (status_code, log_prefix) = Self::io_error_status(&err);
779                if let Some(response) = self
780                  .retry_or_respond(
781                    error_logger,
782                    &err,
783                    retry_connection,
784                    !proxy_to_vector.is_empty(),
785                    status_code,
786                    log_prefix,
787                  )
788                  .await
789                {
790                  return Ok(response);
791                }
792                continue;
793              }
794            };
795
796            #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
797            let stream = match SendUnixStreamPoll::new_comp_io(stream) {
798              Ok(stream) => stream,
799              Err(err) => {
800                self.mark_backend_failure(&upstream).await;
801                if let Some(response) = self
802                  .retry_or_respond(
803                    error_logger,
804                    &err,
805                    retry_connection,
806                    !proxy_to_vector.is_empty(),
807                    StatusCode::BAD_GATEWAY,
808                    "Bad gateway",
809                  )
810                  .await
811                {
812                  return Ok(response);
813                }
814                continue;
815              }
816            };
817
818            Connection::Unix(stream)
819          }
820        } else {
821          let stream = match TcpStream::connect(&addr).await {
822            Ok(stream) => stream,
823            Err(err) => {
824              self.mark_backend_failure(&upstream).await;
825              let (status_code, log_prefix) = Self::io_error_status(&err);
826              if let Some(response) = self
827                .retry_or_respond(
828                  error_logger,
829                  &err,
830                  retry_connection,
831                  !proxy_to_vector.is_empty(),
832                  status_code,
833                  log_prefix,
834                )
835                .await
836              {
837                return Ok(response);
838              }
839              continue;
840            }
841          };
842
843          if let Err(err) = stream.set_nodelay(true) {
844            self.mark_backend_failure(&upstream).await;
845            if let Some(response) = self
846              .retry_or_respond(
847                error_logger,
848                &err,
849                retry_connection,
850                !proxy_to_vector.is_empty(),
851                StatusCode::BAD_GATEWAY,
852                "Bad gateway",
853              )
854              .await
855            {
856              return Ok(response);
857            }
858            continue;
859          };
860
861          #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
862          let stream = match SendTcpStreamPoll::new_comp_io(stream) {
863            Ok(stream) => stream,
864            Err(err) => {
865              self.mark_backend_failure(&upstream).await;
866              if let Some(response) = self
867                .retry_or_respond(
868                  error_logger,
869                  &err,
870                  retry_connection,
871                  !proxy_to_vector.is_empty(),
872                  StatusCode::BAD_GATEWAY,
873                  "Bad gateway",
874                )
875                .await
876              {
877                return Ok(response);
878              }
879              continue;
880            }
881          };
882
883          Connection::Tcp(stream)
884        };
885
886        let proxy_header_to_write = match proxy_header {
887          Some(ProxyHeader::V1) => {
888            let is_ipv4 = socket_data.local_addr.ip().to_canonical().is_ipv4()
889              && socket_data.remote_addr.ip().to_canonical().is_ipv4();
890            let local_addr = if is_ipv4 {
891              match socket_data.local_addr.ip().to_canonical() {
892                IpAddr::V4(ip) => ip.to_string(),
893                IpAddr::V6(ip) => ip
894                  .to_ipv4_mapped()
895                  .ok_or(anyhow::anyhow!("Connection IP address type mismatch"))?
896                  .to_string(),
897              }
898            } else {
899              match socket_data.local_addr.ip().to_canonical() {
900                IpAddr::V4(ip) => ip
901                  .to_ipv6_mapped()
902                  .segments()
903                  .iter()
904                  .map(|seg| format!("{:04x}", seg))
905                  .collect::<Vec<_>>()
906                  .join(":"),
907                IpAddr::V6(ip) => ip
908                  .segments()
909                  .iter()
910                  .map(|seg| format!("{:04x}", seg))
911                  .collect::<Vec<_>>()
912                  .join(":"),
913              }
914            };
915            let remote_addr = if is_ipv4 {
916              match socket_data.remote_addr.ip().to_canonical() {
917                IpAddr::V4(ip) => ip.to_string(),
918                IpAddr::V6(ip) => ip
919                  .to_ipv4_mapped()
920                  .ok_or(anyhow::anyhow!("Connection IP address type mismatch"))?
921                  .to_string(),
922              }
923            } else {
924              match socket_data.remote_addr.ip().to_canonical() {
925                IpAddr::V4(ip) => ip
926                  .to_ipv6_mapped()
927                  .segments()
928                  .iter()
929                  .map(|seg| format!("{:04x}", seg))
930                  .collect::<Vec<_>>()
931                  .join(":"),
932                IpAddr::V6(ip) => ip
933                  .segments()
934                  .iter()
935                  .map(|seg| format!("{:04x}", seg))
936                  .collect::<Vec<_>>()
937                  .join(":"),
938              }
939            };
940            let local_port = socket_data.local_addr.port();
941            let remote_port = socket_data.remote_addr.port();
942            let header = format!(
943              "PROXY {} {} {} {} {}\r\n",
944              if is_ipv4 { "TCP4" } else { "TCP6" },
945              remote_addr,
946              local_addr,
947              remote_port,
948              local_port,
949            );
950            Some(header.into_bytes())
951          }
952          Some(ProxyHeader::V2) => {
953            let is_ipv4 = socket_data.local_addr.ip().to_canonical().is_ipv4()
954              && socket_data.remote_addr.ip().to_canonical().is_ipv4();
955            let addresses = if is_ipv4 {
956              ppp::v2::Addresses::IPv4(ppp::v2::IPv4::new(
957                match socket_data.remote_addr.ip().to_canonical() {
958                  IpAddr::V4(ip) => ip,
959                  IpAddr::V6(ip) => ip
960                    .to_ipv4_mapped()
961                    .ok_or(anyhow::anyhow!("Connection IP address type mismatch"))?,
962                },
963                match socket_data.local_addr.ip().to_canonical() {
964                  IpAddr::V4(ip) => ip,
965                  IpAddr::V6(ip) => ip
966                    .to_ipv4_mapped()
967                    .ok_or(anyhow::anyhow!("Connection IP address type mismatch"))?,
968                },
969                socket_data.remote_addr.port(),
970                socket_data.local_addr.port(),
971              ))
972            } else {
973              ppp::v2::Addresses::IPv6(ppp::v2::IPv6::new(
974                match socket_data.remote_addr.ip().to_canonical() {
975                  IpAddr::V4(ip) => ip.to_ipv6_mapped(),
976                  IpAddr::V6(ip) => ip,
977                },
978                match socket_data.local_addr.ip().to_canonical() {
979                  IpAddr::V4(ip) => ip.to_ipv6_mapped(),
980                  IpAddr::V6(ip) => ip,
981                },
982                socket_data.remote_addr.port(),
983                socket_data.local_addr.port(),
984              ))
985            };
986            let header_builder = ppp::v2::Builder::with_addresses(
987              ppp::v2::Version::Two | ppp::v2::Command::Proxy,
988              ppp::v2::Protocol::Stream,
989              addresses,
990            );
991            Some(header_builder.build()?)
992          }
993          _ => None,
994        };
995
996        let mut stream = stream; // Make the stream a mutable variable (to be able to write PROXY protocol header to it).
997
998        if let Some(proxy_header_to_write) = proxy_header_to_write {
999          if let Err(err) = stream.write_all(&proxy_header_to_write).await {
1000            self.mark_backend_failure(&upstream).await;
1001            if let Some(response) = self
1002              .retry_or_respond(
1003                error_logger,
1004                &err,
1005                retry_connection,
1006                !proxy_to_vector.is_empty(),
1007                StatusCode::BAD_GATEWAY,
1008                "Bad gateway",
1009              )
1010              .await
1011            {
1012              return Ok(response);
1013            }
1014            continue;
1015          }
1016        }
1017
1018        // Safety: the drop guard is dropped when the connection future is completed,
1019        // and after the underlying connection is moved across threads,
1020        // see the "http_proxy_handshake" function.
1021        #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
1022        let drop_guard = unsafe { stream.get_drop_guard() };
1023
1024        let sender = if !encrypted {
1025          let sender = match http_proxy_handshake(
1026            stream,
1027            enable_http2_only_config,
1028            #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
1029            drop_guard,
1030          )
1031          .await
1032          {
1033            Ok(sender) => sender,
1034            Err(err) => {
1035              self.mark_backend_failure(&upstream).await;
1036              if let Some(response) = self
1037                .retry_or_respond(
1038                  error_logger,
1039                  &err,
1040                  retry_connection,
1041                  !proxy_to_vector.is_empty(),
1042                  StatusCode::BAD_GATEWAY,
1043                  "Bad gateway",
1044                )
1045                .await
1046              {
1047                return Ok(response);
1048              }
1049              continue;
1050            }
1051          };
1052
1053          sender
1054        } else {
1055          let enable_http2_config = enable_http2_only_config || (enable_http2_config && !is_http_upgrade);
1056          let mut tls_client_config = (if disable_certificate_verification {
1057            rustls::ClientConfig::builder()
1058              .dangerous()
1059              .with_custom_certificate_verifier(Arc::new(NoServerVerifier::new()))
1060          } else if let Ok(client_config) = BuilderVerifierExt::with_platform_verifier(rustls::ClientConfig::builder())
1061          {
1062            client_config
1063          } else {
1064            rustls::ClientConfig::builder().with_webpki_verifier(
1065              WebPkiServerVerifier::builder(Arc::new(rustls::RootCertStore {
1066                roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(),
1067              }))
1068              .build()?,
1069            )
1070          })
1071          .with_no_client_auth();
1072          if enable_http2_only_config {
1073            tls_client_config.alpn_protocols = vec![b"h2".to_vec()];
1074          } else if enable_http2_config {
1075            tls_client_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
1076          } else {
1077            tls_client_config.alpn_protocols = vec![b"http/1.1".to_vec(), b"http/1.0".to_vec()];
1078          }
1079          let connector = TlsConnector::from(Arc::new(tls_client_config));
1080          let domain = ServerName::try_from(host)?.to_owned();
1081
1082          let tls_stream = match connector.connect(domain, stream).await {
1083            Ok(stream) => stream,
1084            Err(err) => {
1085              self.mark_backend_failure(&upstream).await;
1086              if let Some(response) = self
1087                .retry_or_respond(
1088                  error_logger,
1089                  &err,
1090                  retry_connection,
1091                  !proxy_to_vector.is_empty(),
1092                  StatusCode::BAD_GATEWAY,
1093                  "Bad gateway",
1094                )
1095                .await
1096              {
1097                return Ok(response);
1098              }
1099              continue;
1100            }
1101          };
1102
1103          // Enable HTTP/2 when the ALPN protocol is "h2"
1104          let enable_http2 = enable_http2_config && tls_stream.get_ref().1.alpn_protocol() == Some(b"h2");
1105
1106          let sender = match http_proxy_handshake(
1107            tls_stream,
1108            enable_http2,
1109            #[cfg(any(feature = "runtime-vibeio", feature = "runtime-monoio"))]
1110            drop_guard,
1111          )
1112          .await
1113          {
1114            Ok(sender) => sender,
1115            Err(err) => {
1116              self.mark_backend_failure(&upstream).await;
1117              if let Some(response) = self
1118                .retry_or_respond(
1119                  error_logger,
1120                  &err,
1121                  retry_connection,
1122                  !proxy_to_vector.is_empty(),
1123                  StatusCode::BAD_GATEWAY,
1124                  "Bad gateway",
1125                )
1126                .await
1127              {
1128                return Ok(response);
1129              }
1130              continue;
1131            }
1132          };
1133
1134          sender
1135        };
1136
1137        let proxy_request = Request::from_parts(proxy_request_parts, request_body);
1138
1139        return http_proxy(
1140          sender,
1141          connection_pool_item,
1142          proxy_request,
1143          error_logger,
1144          proxy_intercept_errors,
1145          tracked_connection,
1146          enable_keepalive,
1147        )
1148        .await;
1149      } else {
1150        let request_parts = request_parts.ok_or(anyhow::anyhow!("Request parts are missing"))?;
1151        error_logger.log("No upstreams available").await;
1152        return Ok(ResponseData {
1153          request: Some(Request::from_parts(request_parts, request_body)),
1154          response: None,
1155          response_status: Some(StatusCode::SERVICE_UNAVAILABLE), // No upstreams available
1156          response_headers: None,
1157          new_remote_address: None,
1158        });
1159      }
1160    }
1161  }
1162
1163  async fn metric_data_before_handler(
1164    &mut self,
1165    _request: &Request<BoxBody<Bytes, std::io::Error>>,
1166    _socket_data: &SocketData,
1167    _metrics_sender: &MetricsMultiSender,
1168  ) {
1169    self.selected_backends_metrics = Some(Vec::new());
1170    self.unhealthy_backends_metrics = Some(Vec::new());
1171  }
1172
1173  async fn metric_data_after_handler(&mut self, metrics_sender: &MetricsMultiSender) {
1174    if let Some(selected_backends_metrics) = self.selected_backends_metrics.take() {
1175      for selected_backend in selected_backends_metrics {
1176        let mut attributes = Vec::new();
1177        attributes.push((
1178          "ferron.proxy.backend_url",
1179          MetricAttributeValue::String(selected_backend.proxy_to),
1180        ));
1181        if let Some(backend_unix) = selected_backend.proxy_unix {
1182          attributes.push((
1183            "ferron.proxy.backend_unix_path",
1184            MetricAttributeValue::String(backend_unix),
1185          ));
1186        }
1187        metrics_sender
1188          .send(Metric::new(
1189            "ferron.proxy.backends.selected",
1190            attributes,
1191            MetricType::Counter,
1192            MetricValue::U64(1),
1193            Some("{backend}"),
1194            Some("Number of times a backend server was selected."),
1195          ))
1196          .await;
1197      }
1198    }
1199    if let Some(unhealthy_backends_metrics) = self.unhealthy_backends_metrics.take() {
1200      for unhealthy_backend in unhealthy_backends_metrics {
1201        let mut attributes = Vec::new();
1202        attributes.push((
1203          "ferron.proxy.backend_url",
1204          MetricAttributeValue::String(unhealthy_backend.proxy_to),
1205        ));
1206        if let Some(backend_unix) = unhealthy_backend.proxy_unix {
1207          attributes.push((
1208            "ferron.proxy.backend_unix_path",
1209            MetricAttributeValue::String(backend_unix),
1210          ));
1211        }
1212        metrics_sender
1213          .send(Metric::new(
1214            "ferron.proxy.backends.unhealthy",
1215            attributes,
1216            MetricType::Counter,
1217            MetricValue::U64(1),
1218            Some("{backend}"),
1219            Some("Number of health check failures for a backend server."),
1220          ))
1221          .await;
1222      }
1223    }
1224    metrics_sender
1225      .send(Metric::new(
1226        "ferron.proxy.requests",
1227        vec![(
1228          "ferron.proxy.connection_reused",
1229          MetricAttributeValue::Bool(self.connection_reused),
1230        )],
1231        MetricType::Counter,
1232        MetricValue::U64(1),
1233        Some("{request}"),
1234        Some("Number of reverse proxy requests."),
1235      ))
1236      .await;
1237  }
1238}
ferron_common/http_proxy/mod.rs

ferron_common/http_proxy/
mod.rs
