vector/
http.rs

1#![allow(missing_docs)]
2use std::{
3    collections::HashMap,
4    fmt,
5    net::SocketAddr,
6    task::{Context, Poll},
7    time::Duration,
8};
9
10use futures::future::BoxFuture;
11use headers::{Authorization, HeaderMapExt};
12use http::{
13    HeaderMap, Request, Response, Uri, Version, header::HeaderValue, request::Builder,
14    uri::InvalidUri,
15};
16use hyper::{
17    body::{Body, HttpBody},
18    client,
19    client::{Client, HttpConnector, connect::Connect},
20};
21use hyper_openssl::HttpsConnector;
22use hyper_proxy::ProxyConnector;
23use rand::RngExt;
24use serde_with::serde_as;
25use snafu::{ResultExt, Snafu};
26use tokio::time::Instant;
27use tower::{Layer, Service};
28use tower_http::{
29    classify::{ServerErrorsAsFailures, SharedClassifier},
30    trace::TraceLayer,
31};
32use tracing::{Instrument, Span};
33use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString};
34
35#[cfg(feature = "aws-core")]
36use crate::aws::AwsAuthentication;
37use crate::{
38    config::ProxyConfig,
39    internal_events::{HttpServerRequestReceived, HttpServerResponseSent, http_client},
40    tls::{MaybeTlsSettings, TlsError, tls_connector_builder},
41};
42
43pub mod status {
44    pub const FORBIDDEN: u16 = 403;
45    pub const NOT_FOUND: u16 = 404;
46    pub const TOO_MANY_REQUESTS: u16 = 429;
47}
48
49#[derive(Debug, Snafu)]
50#[snafu(visibility(pub(crate)))]
51pub enum HttpError {
52    #[snafu(display("Failed to build TLS connector: {}", source))]
53    BuildTlsConnector { source: TlsError },
54    #[snafu(display("Failed to build HTTPS connector: {}", source))]
55    MakeHttpsConnector { source: openssl::error::ErrorStack },
56    #[snafu(display("Failed to build Proxy connector: {}", source))]
57    MakeProxyConnector { source: InvalidUri },
58    #[snafu(display("Failed to make HTTP(S) request: {}", source))]
59    CallRequest { source: hyper::Error },
60    #[snafu(display("Failed to build HTTP request: {}", source))]
61    BuildRequest { source: http::Error },
62}
63
64impl HttpError {
65    pub const fn is_retriable(&self) -> bool {
66        match self {
67            HttpError::BuildRequest { .. } | HttpError::MakeProxyConnector { .. } => false,
68            HttpError::CallRequest { .. }
69            | HttpError::BuildTlsConnector { .. }
70            | HttpError::MakeHttpsConnector { .. } => true,
71        }
72    }
73}
74
75pub type HttpClientFuture = <HttpClient as Service<http::Request<Body>>>::Future;
76pub type HttpProxyConnector = ProxyConnector<HttpsConnector<HttpConnector>>;
77
78/// An HTTP client generic over the underlying connector `C`.
79///
80/// `C` defaults to [`HttpProxyConnector`], the proxy-aware TLS connector built from Vector's
81/// [`MaybeTlsSettings`]/[`ProxyConfig`]. Callers that need a custom connector can supply
82/// their own via [`HttpClient::new_with_connector`].
83pub struct HttpClient<B = Body, C = HttpProxyConnector> {
84    client: Client<C, B>,
85    user_agent: HeaderValue,
86    // Proxy-header injection only applies when the client owns a `ProxyConnector`. A custom
87    // connector supplied via `new_with_connector` sets this to `None`.
88    proxy_connector: Option<HttpProxyConnector>,
89}
90
91impl<B> HttpClient<B, HttpProxyConnector>
92where
93    B: fmt::Debug + HttpBody + Send + 'static,
94    B::Data: Send,
95    B::Error: Into<crate::Error>,
96{
97    pub fn new(
98        tls_settings: impl Into<MaybeTlsSettings>,
99        proxy_config: &ProxyConfig,
100    ) -> Result<Self, HttpError> {
101        HttpClient::new_with_custom_client(tls_settings, proxy_config, &mut Client::builder())
102    }
103
104    pub fn new_with_custom_client(
105        tls_settings: impl Into<MaybeTlsSettings>,
106        proxy_config: &ProxyConfig,
107        client_builder: &mut client::Builder,
108    ) -> Result<Self, HttpError> {
109        let proxy_connector = build_proxy_connector(tls_settings.into(), proxy_config)?;
110        let client = client_builder.build(proxy_connector.clone());
111
112        Ok(HttpClient {
113            client,
114            user_agent: default_user_agent(),
115            proxy_connector: Some(proxy_connector),
116        })
117    }
118}
119
120impl<B, C> HttpClient<B, C>
121where
122    B: fmt::Debug + HttpBody + Send + 'static,
123    B::Data: Send,
124    B::Error: Into<crate::Error>,
125    C: Connect + Clone + Send + Sync + 'static,
126{
127    /// Build an `HttpClient` from a pre-constructed connector.
128    ///
129    /// Proxy-header injection is disabled for custom connectors (`proxy_connector` is always `None`).
130    /// It only matters for the case of a plaintext-`http` request sent through an HTTP proxy, where
131    /// `Proxy-Authorization` must ride on the request itself. HTTPS-through-proxy is unaffected:
132    /// proxy auth happens during the CONNECT handshake inside the connector.
133    pub fn new_with_connector(connector: C, client_builder: &mut client::Builder) -> Self {
134        let client = client_builder.build(connector);
135
136        HttpClient {
137            client,
138            user_agent: default_user_agent(),
139            proxy_connector: None,
140        }
141    }
142
143    pub fn send(
144        &self,
145        mut request: Request<B>,
146    ) -> BoxFuture<'static, Result<http::Response<Body>, HttpError>> {
147        let span = tracing::info_span!("http");
148        let _enter = span.enter();
149
150        default_request_headers(&mut request, &self.user_agent);
151        self.maybe_add_proxy_headers(&mut request);
152
153        emit!(http_client::AboutToSendHttpRequest { request: &request });
154
155        let response = self.client.request(request);
156
157        let fut = async move {
158            // Capture the time right before we issue the request.
159            // Request doesn't start the processing until we start polling it.
160            let before = std::time::Instant::now();
161
162            // Send request and wait for the result.
163            let response_result = response.await;
164
165            // Compute the roundtrip time it took to send the request and get
166            // the response or error.
167            let roundtrip = before.elapsed();
168
169            // Handle the errors and extract the response.
170            let response = response_result
171                .inspect_err(|error| {
172                    // Emit the error into the internal events system.
173                    emit!(http_client::GotHttpWarning { error, roundtrip });
174                })
175                .context(CallRequestSnafu)?;
176
177            // Emit the response into the internal events system.
178            emit!(http_client::GotHttpResponse {
179                response: &response,
180                roundtrip
181            });
182            Ok(response)
183        }
184        .instrument(span.clone().or_current());
185
186        Box::pin(fut)
187    }
188
189    fn maybe_add_proxy_headers(&self, request: &mut Request<B>) {
190        let Some(proxy_connector) = &self.proxy_connector else {
191            return;
192        };
193        if let Some(proxy_headers) = proxy_connector.http_headers(request.uri()) {
194            for (k, v) in proxy_headers {
195                let request_headers = request.headers_mut();
196                if !request_headers.contains_key(k) {
197                    request_headers.insert(k, v.into());
198                }
199            }
200        }
201    }
202}
203
204fn default_user_agent() -> HeaderValue {
205    let app_name = crate::get_app_name();
206    let version = crate::get_version();
207    HeaderValue::from_str(&format!("{app_name}/{version}"))
208        .expect("Invalid header value for user-agent!")
209}
210
211pub fn build_proxy_connector(
212    tls_settings: MaybeTlsSettings,
213    proxy_config: &ProxyConfig,
214) -> Result<ProxyConnector<HttpsConnector<HttpConnector>>, HttpError> {
215    // The inner connector dials the proxy for proxied requests and the destination directly for
216    // `no_proxy` requests. The `server_name` override must apply only to the destination, so
217    // collect the authorities of proxies reached over TLS to exclude from it; otherwise an HTTPS
218    // proxy's own certificate would be verified against the destination name and fail with a
219    // hostname mismatch.
220    let proxy_authorities = if proxy_config.enabled {
221        TlsProxyAuthorities {
222            http: tls_proxy_authority(proxy_config.http.as_deref()),
223            https: tls_proxy_authority(proxy_config.https.as_deref()),
224        }
225    } else {
226        TlsProxyAuthorities::default()
227    };
228
229    // `server_name` still cannot be applied to the tunneled destination TLS of a proxied HTTPS
230    // request: `hyper-proxy` offers no per-connection callback there and verifies against the
231    // destination URL host. Warn when that combination is configured.
232    if proxy_config.enabled
233        && (proxy_config.http.is_some() || proxy_config.https.is_some())
234        && let Some(tls) = tls_settings.tls()
235        && tls.server_name().is_some()
236        && tls.verify_hostname()
237    {
238        warn!(
239            message = "`tls.server_name` is set with hostname verification enabled, but a proxy is configured. \
240                       `server_name` is not applied to proxied (tunneled) TLS connections, so certificate \
241                       verification may fail with a hostname mismatch."
242        );
243    }
244
245    // Create dedicated TLS connector for the proxied connection with user TLS settings.
246    let tls = tls_connector_builder(&tls_settings)
247        .context(BuildTlsConnectorSnafu)?
248        .build();
249    let https = build_https_connector(tls_settings, proxy_authorities)?;
250    let mut proxy = ProxyConnector::new(https).unwrap();
251    // Make proxy connector aware of user TLS settings by setting the TLS connector:
252    // https://github.com/vectordotdev/vector/issues/13683
253    proxy.set_tls(Some(tls));
254    proxy_config
255        .configure(&mut proxy)
256        .context(MakeProxyConnectorSnafu)?;
257    Ok(proxy)
258}
259
260pub fn build_tls_connector(
261    tls_settings: MaybeTlsSettings,
262) -> Result<HttpsConnector<HttpConnector>, HttpError> {
263    build_https_connector(tls_settings, TlsProxyAuthorities::default())
264}
265
266/// Authorities (host and optional port) of the configured forward proxies that are reached over TLS
267/// (i.e. `https://` proxy URLs). Used to skip the `tls.server_name` override for the connection to
268/// such a proxy, so its own certificate is verified against the proxy authority rather than the
269/// destination name. Plaintext (`http://`) proxies never trigger a TLS handshake to the proxy, so
270/// they are not tracked; matching on the full authority also avoids mistaking a direct connection
271/// to a destination that merely shares a host with a proxy (e.g. on a different port).
272#[derive(Clone, Default)]
273struct TlsProxyAuthorities {
274    http: Option<(String, Option<u16>)>,
275    https: Option<(String, Option<u16>)>,
276}
277
278impl TlsProxyAuthorities {
279    fn matches(&self, uri: &http::Uri) -> bool {
280        if self.http.is_none() && self.https.is_none() {
281            return false;
282        }
283        let target = (uri.host(), uri.port_u16());
284        let matches_one = |authority: &Option<(String, Option<u16>)>| {
285            authority
286                .as_ref()
287                .is_some_and(|(host, port)| target.0 == Some(host.as_str()) && target.1 == *port)
288        };
289        matches_one(&self.http) || matches_one(&self.https)
290    }
291}
292
293/// Extract the authority (host and optional port) of a proxy URL, but only when it is reached over
294/// TLS (an `https://` URL). Returns `None` for plaintext proxies or unparseable URLs.
295fn tls_proxy_authority(url: Option<&str>) -> Option<(String, Option<u16>)> {
296    let uri = url?.parse::<http::Uri>().ok()?;
297    if uri.scheme_str() != Some("https") {
298        return None;
299    }
300    Some((uri.host()?.to_owned(), uri.port_u16()))
301}
302
303/// Build an HTTPS connector, skipping the `tls.server_name` override for connections to one of
304/// `proxy_authorities`. The override must only apply to the upstream destination; applying it to a
305/// proxy connection would verify the proxy certificate against the destination name.
306fn build_https_connector(
307    tls_settings: MaybeTlsSettings,
308    proxy_authorities: TlsProxyAuthorities,
309) -> Result<HttpsConnector<HttpConnector>, HttpError> {
310    let mut http = HttpConnector::new();
311    http.enforce_http(false);
312
313    let tls = tls_connector_builder(&tls_settings).context(BuildTlsConnectorSnafu)?;
314    let mut https = HttpsConnector::with_connector(http, tls).context(MakeHttpsConnectorSnafu)?;
315
316    let settings = tls_settings.tls().cloned();
317    https.set_callback(move |c, uri| {
318        if let Some(settings) = &settings {
319            let skip_server_name = proxy_authorities.matches(uri);
320            settings.apply_connect_configuration(c, skip_server_name)
321        } else {
322            Ok(())
323        }
324    });
325    Ok(https)
326}
327
328fn default_request_headers<B>(request: &mut Request<B>, user_agent: &HeaderValue) {
329    if !request.headers().contains_key("User-Agent") {
330        request
331            .headers_mut()
332            .insert("User-Agent", user_agent.clone());
333    }
334
335    if !request.headers().contains_key("Accept-Encoding") {
336        // hardcoding until we support compressed responses:
337        // https://github.com/vectordotdev/vector/issues/5440
338        request
339            .headers_mut()
340            .insert("Accept-Encoding", HeaderValue::from_static("identity"));
341    }
342}
343
344impl<B, C> Service<Request<B>> for HttpClient<B, C>
345where
346    B: fmt::Debug + HttpBody + Send + 'static,
347    B::Data: Send,
348    B::Error: Into<crate::Error> + Send,
349    C: Connect + Clone + Send + Sync + 'static,
350{
351    type Response = http::Response<Body>;
352    type Error = HttpError;
353    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
354
355    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
356        Poll::Ready(Ok(()))
357    }
358
359    fn call(&mut self, request: Request<B>) -> Self::Future {
360        self.send(request)
361    }
362}
363
364impl<B, C: Clone> Clone for HttpClient<B, C> {
365    fn clone(&self) -> Self {
366        Self {
367            client: self.client.clone(),
368            user_agent: self.user_agent.clone(),
369            proxy_connector: self.proxy_connector.clone(),
370        }
371    }
372}
373
374impl<B, C> fmt::Debug for HttpClient<B, C> {
375    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
376        f.debug_struct("HttpClient")
377            .field("client", &self.client)
378            .field("user_agent", &self.user_agent)
379            .finish()
380    }
381}
382
383/// Configuration of the authentication strategy for HTTP requests.
384///
385/// HTTP authentication should be used with HTTPS only, as the authentication credentials are passed as an
386/// HTTP header without any additional encryption beyond what is provided by the transport itself.
387#[configurable_component]
388#[derive(Clone, Debug, Eq, PartialEq)]
389#[serde(deny_unknown_fields, rename_all = "snake_case", tag = "strategy")]
390#[configurable(metadata(docs::enum_tag_description = "The authentication strategy to use."))]
391pub enum Auth {
392    /// Basic authentication.
393    ///
394    /// The username and password are concatenated and encoded using [base64][base64].
395    ///
396    /// [base64]: https://en.wikipedia.org/wiki/Base64
397    Basic {
398        /// The basic authentication username.
399        #[configurable(metadata(docs::examples = "${USERNAME}"))]
400        #[configurable(metadata(docs::examples = "username"))]
401        user: String,
402
403        /// The basic authentication password.
404        #[configurable(metadata(docs::examples = "${PASSWORD}"))]
405        #[configurable(metadata(docs::examples = "password"))]
406        password: SensitiveString,
407    },
408
409    /// Bearer authentication.
410    ///
411    /// The bearer token value (OAuth2, JWT, etc.) is passed as-is.
412    Bearer {
413        /// The bearer authentication token.
414        token: SensitiveString,
415    },
416
417    #[cfg(feature = "aws-core")]
418    /// AWS authentication.
419    Aws {
420        /// The AWS authentication configuration.
421        auth: AwsAuthentication,
422
423        /// The AWS service name to use for signing.
424        service: String,
425    },
426
427    /// Custom Authorization Header Value, will be inserted into the headers as `Authorization: < value >`
428    Custom {
429        /// Custom string value of the Authorization header
430        #[configurable(metadata(docs::examples = "${AUTH_HEADER_VALUE}"))]
431        #[configurable(metadata(docs::examples = "CUSTOM_PREFIX ${TOKEN}"))]
432        value: String,
433    },
434}
435
436pub trait MaybeAuth: Sized {
437    fn choose_one(&self, other: &Self) -> crate::Result<Self>;
438}
439
440impl MaybeAuth for Option<Auth> {
441    fn choose_one(&self, other: &Self) -> crate::Result<Self> {
442        if self.is_some() && other.is_some() {
443            Err("Two authorization credentials was provided.".into())
444        } else {
445            Ok(self.clone().or_else(|| other.clone()))
446        }
447    }
448}
449
450impl Auth {
451    pub fn apply<B>(&self, req: &mut Request<B>) {
452        self.apply_headers_map(req.headers_mut())
453    }
454
455    pub fn apply_builder(&self, mut builder: Builder) -> Builder {
456        if let Some(map) = builder.headers_mut() {
457            self.apply_headers_map(map)
458        }
459        builder
460    }
461
462    pub fn apply_headers_map(&self, map: &mut HeaderMap) {
463        match &self {
464            Auth::Basic { user, password } => {
465                let auth = Authorization::basic(user.as_str(), password.inner());
466                map.typed_insert(auth);
467            }
468            Auth::Bearer { token } => match Authorization::bearer(token.inner()) {
469                Ok(auth) => map.typed_insert(auth),
470                Err(error) => error!(message = "Invalid bearer token.", token = %token, %error),
471            },
472            Auth::Custom { value } => {
473                // The value contains just the value for the Authorization header
474                // Expected format: "SSWS token123" or "Bearer token123", etc.
475                match HeaderValue::from_str(value) {
476                    Ok(header_val) => {
477                        map.insert(http::header::AUTHORIZATION, header_val);
478                    }
479                    Err(error) => {
480                        error!(message = "Invalid custom auth header value.", value = %value, %error)
481                    }
482                }
483            }
484            #[cfg(feature = "aws-core")]
485            _ => {}
486        }
487    }
488}
489
490pub fn get_http_scheme_from_uri(uri: &Uri) -> &'static str {
491    // If there's no scheme, we just use "http" since it provides the most semantic relevance without inadvertently
492    // implying things it can't know i.e. returning "https" when we're not actually sure HTTPS was used.
493    uri.scheme_str().map_or("http", |scheme| match scheme {
494        "http" => "http",
495        "https" => "https",
496        // `http::Uri` ensures that we always get "http" or "https" if the URI is created with a well-formed scheme, but
497        // it also supports arbitrary schemes, which is where we bomb out down here, since we can't generate a static
498        // string for an arbitrary input string... and anything other than "http" and "https" makes no sense for an HTTP
499        // client anyways.
500        s => panic!("invalid URI scheme for HTTP client: {s}"),
501    })
502}
503
504/// Builds a [TraceLayer] configured for a HTTP server.
505///
506/// This layer emits HTTP specific telemetry for requests received, responses sent, and handler duration.
507pub fn build_http_trace_layer<T, U>(
508    span: Span,
509) -> TraceLayer<
510    SharedClassifier<ServerErrorsAsFailures>,
511    impl Fn(&Request<T>) -> Span + Clone,
512    impl Fn(&Request<T>, &Span) + Clone,
513    impl Fn(&Response<U>, Duration, &Span) + Clone,
514    (),
515    (),
516    (),
517> {
518    TraceLayer::new_for_http()
519        .make_span_with(move |request: &Request<T>| {
520            // This is an error span so that the labels are always present for metrics.
521            error_span!(
522               parent: &span,
523               "http-request",
524               method = %request.method(),
525               path = %request.uri().path(),
526            )
527        })
528        .on_request(Box::new(|_request: &Request<T>, _span: &Span| {
529            emit!(HttpServerRequestReceived);
530        }))
531        .on_response(|response: &Response<U>, latency: Duration, _span: &Span| {
532            emit!(HttpServerResponseSent { response, latency });
533        })
534        .on_failure(())
535        .on_body_chunk(())
536        .on_eos(())
537}
538
539/// Configuration of HTTP server keepalive parameters.
540#[serde_as]
541#[configurable_component]
542#[derive(Clone, Debug, PartialEq)]
543#[serde(deny_unknown_fields)]
544pub struct KeepaliveConfig {
545    /// The maximum amount of time a connection may exist before it is closed by sending
546    /// a `Connection: close` header on the HTTP response. Set this to a large value like
547    /// `100000000` to "disable" this feature
548    ///
549    ///
550    /// Only applies to HTTP/0.9, HTTP/1.0, and HTTP/1.1 requests.
551    ///
552    /// A random jitter configured by `max_connection_age_jitter_factor` is added
553    /// to the specified duration to spread out connection storms.
554    #[serde(default = "default_max_connection_age")]
555    #[configurable(metadata(docs::examples = 600))]
556    #[configurable(metadata(docs::type_unit = "seconds"))]
557    #[configurable(metadata(docs::human_name = "Maximum Connection Age"))]
558    pub max_connection_age_secs: Option<u64>,
559
560    /// The factor by which to jitter the `max_connection_age_secs` value.
561    ///
562    /// A value of 0.1 means that the actual duration will be between 90% and 110% of the
563    /// specified maximum duration.
564    #[serde(default = "default_max_connection_age_jitter_factor")]
565    #[configurable(validation(range(min = 0.0, max = 1.0)))]
566    pub max_connection_age_jitter_factor: f64,
567}
568
569const fn default_max_connection_age() -> Option<u64> {
570    Some(300) // 5 minutes
571}
572
573const fn default_max_connection_age_jitter_factor() -> f64 {
574    0.1
575}
576
577impl Default for KeepaliveConfig {
578    fn default() -> Self {
579        Self {
580            max_connection_age_secs: default_max_connection_age(),
581            max_connection_age_jitter_factor: default_max_connection_age_jitter_factor(),
582        }
583    }
584}
585
586/// A layer that limits the maximum duration of a client connection. It does so by adding a
587/// `Connection: close` header to the response if `max_connection_duration` time has elapsed
588/// since `start_reference`.
589///
590/// **Notes:**
591/// - This is intended to be used in a Hyper server (or similar) that will automatically close
592///   the connection after a response with a `Connection: close` header is sent.
593/// - This layer assumes that it is instantiated once per connection, which is true within the
594///   Hyper framework.
595pub struct MaxConnectionAgeLayer {
596    start_reference: Instant,
597    max_connection_age: Duration,
598    peer_addr: SocketAddr,
599}
600
601impl MaxConnectionAgeLayer {
602    pub fn new(max_connection_age: Duration, jitter_factor: f64, peer_addr: SocketAddr) -> Self {
603        Self {
604            start_reference: Instant::now(),
605            max_connection_age: Self::jittered_duration(max_connection_age, jitter_factor),
606            peer_addr,
607        }
608    }
609
610    fn jittered_duration(duration: Duration, jitter_factor: f64) -> Duration {
611        // Ensure the jitter_factor is between 0.0 and 1.0
612        let jitter_factor = jitter_factor.clamp(0.0, 1.0);
613        // Generate a random jitter factor between `1 - jitter_factor`` and `1 + jitter_factor`.
614        let mut rng = rand::rng();
615        let random_jitter_factor = rng.random_range(-jitter_factor..=jitter_factor) + 1.;
616        duration.mul_f64(random_jitter_factor)
617    }
618}
619
620impl<S> Layer<S> for MaxConnectionAgeLayer
621where
622    S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
623    S::Future: Send + 'static,
624{
625    type Service = MaxConnectionAgeService<S>;
626
627    fn layer(&self, service: S) -> Self::Service {
628        MaxConnectionAgeService {
629            service,
630            start_reference: self.start_reference,
631            max_connection_age: self.max_connection_age,
632            peer_addr: self.peer_addr,
633        }
634    }
635}
636
637/// A service that limits the maximum age of a client connection. It does so by adding a
638/// `Connection: close` header to the response if `max_connection_age` time has elapsed
639/// since `start_reference`.
640///
641/// **Notes:**
642/// - This is intended to be used in a Hyper server (or similar) that will automatically close
643///   the connection after a response with a `Connection: close` header is sent.
644/// - This service assumes that it is instantiated once per connection, which is true within the
645///   Hyper framework.
646#[derive(Clone)]
647pub struct MaxConnectionAgeService<S> {
648    service: S,
649    start_reference: Instant,
650    max_connection_age: Duration,
651    peer_addr: SocketAddr,
652}
653
654impl<S, E> Service<Request<Body>> for MaxConnectionAgeService<S>
655where
656    S: Service<Request<Body>, Response = Response<Body>, Error = E> + Clone + Send + 'static,
657    S::Future: Send + 'static,
658{
659    type Response = S::Response;
660    type Error = E;
661    type Future = BoxFuture<'static, Result<Self::Response, E>>;
662
663    fn poll_ready(
664        &mut self,
665        cx: &mut std::task::Context<'_>,
666    ) -> std::task::Poll<Result<(), Self::Error>> {
667        self.service.poll_ready(cx)
668    }
669
670    fn call(&mut self, req: Request<Body>) -> Self::Future {
671        let start_reference = self.start_reference;
672        let max_connection_age = self.max_connection_age;
673        let peer_addr = self.peer_addr;
674        let version = req.version();
675        let future = self.service.call(req);
676        Box::pin(async move {
677            let mut response = future.await?;
678            match version {
679                Version::HTTP_09 | Version::HTTP_10 | Version::HTTP_11
680                    if start_reference.elapsed() >= max_connection_age =>
681                {
682                    debug!(
683                        message = "Closing connection due to max connection age.",
684                        ?max_connection_age,
685                        connection_age = ?start_reference.elapsed(),
686                        ?peer_addr,
687                    );
688                    // Tell the client to close this connection.
689                    // Hyper will automatically close the connection after the response is sent.
690                    response.headers_mut().insert(
691                        hyper::header::CONNECTION,
692                        hyper::header::HeaderValue::from_static("close"),
693                    );
694                }
695                Version::HTTP_09 | Version::HTTP_10 | Version::HTTP_11 => (),
696                // TODO need to send GOAWAY frame
697                Version::HTTP_2 => (),
698                // TODO need to send GOAWAY frame
699                Version::HTTP_3 => (),
700                _ => (),
701            }
702            Ok(response)
703        })
704    }
705}
706
707/// The type of a query parameter's value, determines if it's treated as a plain string or a VRL expression.
708#[configurable_component]
709#[derive(Clone, Debug, Default, Eq, PartialEq)]
710#[serde(rename_all = "snake_case")]
711pub enum ParamType {
712    /// The parameter value is a plain string.
713    #[default]
714    String,
715    /// The parameter value is a VRL expression that is evaluated before each request.
716    Vrl,
717}
718
719impl ParamType {
720    fn is_default(&self) -> bool {
721        *self == Self::default()
722    }
723}
724
725/// Represents a query parameter value, which can be a simple string or a typed object
726/// indicating whether the value is a string or a VRL expression.
727#[configurable_component]
728#[derive(Clone, Debug, Eq, PartialEq)]
729#[serde(untagged)]
730pub enum ParameterValue {
731    /// A simple string value. For backwards compatibility.
732    String(String),
733    /// A value with an explicit type.
734    Typed {
735        /// The raw value of the parameter.
736        value: String,
737        /// The parameter type, indicating how the `value` should be treated.
738        #[serde(
739            default,
740            skip_serializing_if = "ParamType::is_default",
741            rename = "type"
742        )]
743        r#type: ParamType,
744    },
745}
746
747impl ParameterValue {
748    /// Returns true if the parameter is a VRL expression.
749    pub const fn is_vrl(&self) -> bool {
750        match self {
751            ParameterValue::String(_) => false,
752            ParameterValue::Typed { r#type, .. } => matches!(r#type, ParamType::Vrl),
753        }
754    }
755
756    /// Returns the raw string value of the parameter.
757    pub const fn value(&self) -> &str {
758        match self {
759            ParameterValue::String(value) | ParameterValue::Typed { value, .. } => value.as_str(),
760        }
761    }
762
763    /// Consumes the `ParameterValue` and returns the owned raw string value.
764    pub fn into_value(self) -> String {
765        match self {
766            ParameterValue::String(s) => s,
767            ParameterValue::Typed { value, .. } => value,
768        }
769    }
770}
771
772/// Configuration of the query parameter value for HTTP requests.
773#[configurable_component]
774#[derive(Clone, Debug, Eq, PartialEq)]
775#[serde(untagged)]
776#[configurable(metadata(docs::enum_tag_description = "Query parameter value"))]
777pub enum QueryParameterValue {
778    /// Query parameter with single value
779    SingleParam(ParameterValue),
780    /// Query parameter with multiple values
781    MultiParams(Vec<ParameterValue>),
782}
783
784impl QueryParameterValue {
785    /// Returns an iterator over the contained `ParameterValue`s.
786    pub fn iter(&self) -> impl Iterator<Item = &ParameterValue> {
787        match self {
788            QueryParameterValue::SingleParam(param) => std::slice::from_ref(param).iter(),
789            QueryParameterValue::MultiParams(params) => params.iter(),
790        }
791    }
792
793    /// Convert to `Vec<ParameterValue>` for owned iteration.
794    fn into_vec(self) -> Vec<ParameterValue> {
795        match self {
796            QueryParameterValue::SingleParam(param) => vec![param],
797            QueryParameterValue::MultiParams(params) => params,
798        }
799    }
800}
801
802// Implement IntoIterator for owned QueryParameterValue
803impl IntoIterator for QueryParameterValue {
804    type Item = ParameterValue;
805    type IntoIter = std::vec::IntoIter<ParameterValue>;
806
807    fn into_iter(self) -> Self::IntoIter {
808        self.into_vec().into_iter()
809    }
810}
811
812pub type QueryParameters = HashMap<String, QueryParameterValue>;
813
814mod client_v1;
815
816#[cfg(test)]
817mod transport_tests;
818
819#[cfg(test)]
820mod tests {
821    use std::convert::Infallible;
822
823    use hyper::{Server, server::conn::AddrStream, service::make_service_fn};
824    use proptest::prelude::*;
825    use tower::ServiceBuilder;
826
827    use super::*;
828    use crate::test_util::addr::next_addr;
829
830    #[test]
831    fn tls_proxy_authority_only_tracks_https_proxies() {
832        assert_eq!(
833            tls_proxy_authority(Some("https://proxy.example:3128")),
834            Some(("proxy.example".to_owned(), Some(3128)))
835        );
836        assert_eq!(
837            tls_proxy_authority(Some("https://proxy.example")),
838            Some(("proxy.example".to_owned(), None))
839        );
840        // Plaintext proxies never trigger a TLS handshake to the proxy.
841        assert_eq!(tls_proxy_authority(Some("http://proxy.example:3128")), None);
842        assert_eq!(tls_proxy_authority(None), None);
843    }
844
845    #[test]
846    fn proxy_authorities_match_full_authority() {
847        let authorities = TlsProxyAuthorities {
848            http: None,
849            https: tls_proxy_authority(Some("https://proxy.example:3128")),
850        };
851        let uri = |s: &str| s.parse::<http::Uri>().unwrap();
852
853        // The proxy connection itself matches and skips the override.
854        assert!(authorities.matches(&uri("https://proxy.example:3128")));
855
856        // A direct destination sharing the host but on a different port must not match, so the
857        // `server_name` override still applies.
858        assert!(!authorities.matches(&uri("https://proxy.example:8443")));
859        // A different host does not match either.
860        assert!(!authorities.matches(&uri("https://destination.example:3128")));
861    }
862
863    #[test]
864    fn test_default_request_headers_defaults() {
865        let user_agent = HeaderValue::from_static("vector");
866        let mut request = Request::post("http://example.com").body(()).unwrap();
867        default_request_headers(&mut request, &user_agent);
868        assert_eq!(
869            request.headers().get("Accept-Encoding"),
870            Some(&HeaderValue::from_static("identity")),
871        );
872        assert_eq!(request.headers().get("User-Agent"), Some(&user_agent));
873    }
874
875    #[test]
876    fn test_default_request_headers_does_not_overwrite() {
877        let mut request = Request::post("http://example.com")
878            .header("Accept-Encoding", "gzip")
879            .header("User-Agent", "foo")
880            .body(())
881            .unwrap();
882        default_request_headers(&mut request, &HeaderValue::from_static("vector"));
883        assert_eq!(
884            request.headers().get("Accept-Encoding"),
885            Some(&HeaderValue::from_static("gzip")),
886        );
887        assert_eq!(
888            request.headers().get("User-Agent"),
889            Some(&HeaderValue::from_static("foo"))
890        );
891    }
892
893    proptest! {
894        #[test]
895        fn test_jittered_duration(duration_in_secs in 0u64..120, jitter_factor in 0.0..1.0) {
896            let duration = Duration::from_secs(duration_in_secs);
897            let jittered_duration = MaxConnectionAgeLayer::jittered_duration(duration, jitter_factor);
898
899            // Check properties based on the range of inputs
900            if jitter_factor == 0.0 {
901                // When jitter_factor is 0, jittered_duration should be equal to the original duration
902                prop_assert_eq!(
903                    jittered_duration,
904                    duration,
905                    "jittered_duration {:?} should be equal to duration {:?}",
906                    jittered_duration,
907                    duration,
908                );
909            } else if duration_in_secs > 0 {
910                // Check the bounds when duration is non-zero and jitter_factor is non-zero
911                let lower_bound = duration.mul_f64(1.0 - jitter_factor);
912                let upper_bound = duration.mul_f64(1.0 + jitter_factor);
913                prop_assert!(
914                    jittered_duration >= lower_bound && jittered_duration <= upper_bound,
915                    "jittered_duration {:?} should be between {:?} and {:?}",
916                    jittered_duration,
917                    lower_bound,
918                    upper_bound,
919                );
920            } else {
921                // When duration is zero, jittered_duration should also be zero
922                prop_assert_eq!(
923                    jittered_duration,
924                    Duration::from_secs(0),
925                    "jittered_duration {:?} should be equal to zero",
926                    jittered_duration,
927                );
928            }
929        }
930    }
931
932    #[tokio::test]
933    async fn test_max_connection_age_service() {
934        tokio::time::pause();
935
936        let start_reference = Instant::now();
937        let max_connection_age = Duration::from_secs(1);
938        let mut service = MaxConnectionAgeService {
939            service: tower::service_fn(|_req: Request<Body>| async {
940                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
941            }),
942            start_reference,
943            max_connection_age,
944            peer_addr: "1.2.3.4:1234".parse().unwrap(),
945        };
946
947        let req = Request::get("http://example.com")
948            .body(Body::empty())
949            .unwrap();
950        let response = service.call(req).await.unwrap();
951        assert_eq!(response.headers().get("Connection"), None);
952
953        tokio::time::advance(Duration::from_millis(500)).await;
954        let req = Request::get("http://example.com")
955            .body(Body::empty())
956            .unwrap();
957        let response = service.call(req).await.unwrap();
958        assert_eq!(response.headers().get("Connection"), None);
959
960        tokio::time::advance(Duration::from_millis(500)).await;
961        let req = Request::get("http://example.com")
962            .body(Body::empty())
963            .unwrap();
964        let response = service.call(req).await.unwrap();
965        assert_eq!(
966            response.headers().get("Connection"),
967            Some(&HeaderValue::from_static("close"))
968        );
969    }
970
971    #[tokio::test]
972    async fn test_max_connection_age_service_http2() {
973        tokio::time::pause();
974
975        let start_reference = Instant::now();
976        let max_connection_age = Duration::from_secs(0);
977        let mut service = MaxConnectionAgeService {
978            service: tower::service_fn(|_req: Request<Body>| async {
979                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
980            }),
981            start_reference,
982            max_connection_age,
983            peer_addr: "1.2.3.4:1234".parse().unwrap(),
984        };
985
986        let mut req = Request::get("http://example.com")
987            .body(Body::empty())
988            .unwrap();
989        *req.version_mut() = Version::HTTP_2;
990        let response = service.call(req).await.unwrap();
991        assert_eq!(response.headers().get("Connection"), None);
992    }
993
994    #[tokio::test]
995    async fn test_max_connection_age_service_http3() {
996        tokio::time::pause();
997
998        let start_reference = Instant::now();
999        let max_connection_age = Duration::from_secs(0);
1000        let mut service = MaxConnectionAgeService {
1001            service: tower::service_fn(|_req: Request<Body>| async {
1002                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
1003            }),
1004            start_reference,
1005            max_connection_age,
1006            peer_addr: "1.2.3.4:1234".parse().unwrap(),
1007        };
1008
1009        let mut req = Request::get("http://example.com")
1010            .body(Body::empty())
1011            .unwrap();
1012        *req.version_mut() = Version::HTTP_3;
1013        let response = service.call(req).await.unwrap();
1014        assert_eq!(response.headers().get("Connection"), None);
1015    }
1016
1017    #[tokio::test]
1018    async fn test_max_connection_age_service_zero_duration() {
1019        tokio::time::pause();
1020
1021        let start_reference = Instant::now();
1022        let max_connection_age = Duration::from_millis(0);
1023        let mut service = MaxConnectionAgeService {
1024            service: tower::service_fn(|_req: Request<Body>| async {
1025                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
1026            }),
1027            start_reference,
1028            max_connection_age,
1029            peer_addr: "1.2.3.4:1234".parse().unwrap(),
1030        };
1031
1032        let req = Request::get("http://example.com")
1033            .body(Body::empty())
1034            .unwrap();
1035        let response = service.call(req).await.unwrap();
1036        assert_eq!(
1037            response.headers().get("Connection"),
1038            Some(&HeaderValue::from_static("close"))
1039        );
1040    }
1041
1042    // Note that we unfortunately cannot mock the time in this test because the client calls
1043    // sleep internally, which advances the clock.  However, this test shouldn't be flakey given
1044    // the time bounds provided.
1045    #[tokio::test]
1046    async fn test_max_connection_age_service_with_hyper_server() {
1047        // Create a hyper server with the max connection age layer.
1048        let max_connection_age = Duration::from_secs(1);
1049        let (_guard, addr) = next_addr();
1050        let make_svc = make_service_fn(move |conn: &AddrStream| {
1051            let svc = ServiceBuilder::new()
1052                .layer(MaxConnectionAgeLayer::new(
1053                    max_connection_age,
1054                    0.,
1055                    conn.remote_addr(),
1056                ))
1057                .service(tower::service_fn(|_req: Request<Body>| async {
1058                    Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
1059                }));
1060            futures_util::future::ok::<_, Infallible>(svc)
1061        });
1062
1063        tokio::spawn(async move {
1064            Server::bind(&addr).serve(make_svc).await.unwrap();
1065        });
1066
1067        // Wait for the server to start.
1068        tokio::time::sleep(Duration::from_millis(10)).await;
1069
1070        // Create a client, which has its own connection pool.
1071        let client = HttpClient::new(None, &ProxyConfig::default()).unwrap();
1072
1073        // Responses generated before the client's max connection age has elapsed do not
1074        // include a `Connection: close` header in the response.
1075        let req = Request::get(format!("http://{addr}/"))
1076            .body(Body::empty())
1077            .unwrap();
1078        let response = client.send(req).await.unwrap();
1079        assert_eq!(response.headers().get("Connection"), None);
1080
1081        let req = Request::get(format!("http://{addr}/"))
1082            .body(Body::empty())
1083            .unwrap();
1084        let response = client.send(req).await.unwrap();
1085        assert_eq!(response.headers().get("Connection"), None);
1086
1087        // The first response generated after the client's max connection age has elapsed should
1088        // include the `Connection: close` header.
1089        tokio::time::sleep(Duration::from_secs(1)).await;
1090        let req = Request::get(format!("http://{addr}/"))
1091            .body(Body::empty())
1092            .unwrap();
1093        let response = client.send(req).await.unwrap();
1094        assert_eq!(
1095            response.headers().get("Connection"),
1096            Some(&HeaderValue::from_static("close")),
1097        );
1098
1099        // The next request should establish a new connection.
1100        // Importantly, this also confirms that each connection has its own independent
1101        // connection age timer.
1102        let req = Request::get(format!("http://{addr}/"))
1103            .body(Body::empty())
1104            .unwrap();
1105        let response = client.send(req).await.unwrap();
1106        assert_eq!(response.headers().get("Connection"), None);
1107    }
1108}
vector/http.rs

vector/
http.rs
