object_store/client/
mod.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Generic utilities for network based [`ObjectStore`] implementations
19//!
20//! [`ObjectStore`]: crate::ObjectStore
21
22pub(crate) mod backoff;
23
24#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
25mod dns;
26
27#[cfg(not(target_arch = "wasm32"))]
28#[cfg(test)]
29pub(crate) mod mock_server;
30
31pub(crate) mod retry;
32
33#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
34pub(crate) mod pagination;
35
36pub(crate) mod get;
37
38#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
39pub(crate) mod list;
40
41#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
42pub(crate) mod token;
43
44pub(crate) mod header;
45
46#[cfg(any(feature = "aws-base", feature = "gcp-base"))]
47pub(crate) mod s3;
48
49pub(crate) mod builder;
50mod http;
51
52#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
53pub(crate) mod parts;
54pub use http::*;
55
56#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
57mod crypto;
58
59#[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
60pub use crypto::*;
61
62use ::http::header::{HeaderMap, HeaderValue};
63use async_trait::async_trait;
64use serde::{Deserialize, Serialize};
65use std::collections::HashMap;
66use std::str::FromStr;
67use std::sync::Arc;
68use std::time::Duration;
69
70#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
71use reqwest::{NoProxy, Proxy};
72
73use crate::config::{ConfigValue, fmt_duration};
74use crate::path::Path;
75use crate::{GetOptions, Result};
76
77#[cfg(feature = "reqwest")]
78fn map_client_error(e: reqwest::Error) -> super::Error {
79    super::Error::Generic {
80        store: "HTTP client",
81        source: Box::new(e),
82    }
83}
84
85#[cfg(feature = "reqwest")]
86static DEFAULT_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
87
88/// Configuration keys for [`ClientOptions`]
89#[derive(PartialEq, Eq, Hash, Clone, Debug, Copy, Deserialize, Serialize)]
90#[non_exhaustive]
91pub enum ClientConfigKey {
92    /// Allow non-TLS, i.e. non-HTTPS connections
93    ///
94    /// Supported keys:
95    /// - `allow_http`
96    AllowHttp,
97    /// Skip certificate validation on https connections.
98    ///
99    /// <div class="warning">
100    ///
101    /// **Warning**
102    ///
103    /// You should think very carefully before using this method. If
104    /// invalid certificates are trusted, *any* certificate for *any* site
105    /// will be trusted for use. This includes expired certificates. This
106    /// introduces significant vulnerabilities, and should only be used
107    /// as a last resort or for testing
108    ///
109    /// </div>
110    ///
111    /// Supported keys:
112    /// - `allow_invalid_certificates`
113    AllowInvalidCertificates,
114    /// Disable certificate validation using the operating system's certificate facilities.
115    ///
116    /// See [`ClientOptions::with_no_system_certificates`]
117    ///
118    /// Supported keys:
119    ///
120    /// - `disable_system_certificates`
121    NoSystemCertificates,
122    /// Timeout for only the connect phase of a Client
123    ///
124    /// Supported keys:
125    /// - `connect_timeout`
126    ConnectTimeout,
127    /// default [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for uploads
128    ///
129    /// Supported keys:
130    /// - `default_content_type`
131    DefaultContentType,
132    /// Only use HTTP/1 connections
133    ///
134    /// Supported keys:
135    /// - `http1_only`
136    Http1Only,
137    /// Interval for HTTP/2 Ping frames should be sent to keep a connection alive.
138    ///
139    /// Supported keys:
140    /// - `http2_keep_alive_interval`
141    Http2KeepAliveInterval,
142    /// Timeout for receiving an acknowledgement of the keep-alive ping.
143    ///
144    /// Supported keys:
145    /// - `http2_keep_alive_timeout`
146    Http2KeepAliveTimeout,
147    /// Enable HTTP/2 keep alive pings for idle connections
148    ///
149    /// Supported keys:
150    /// - `http2_keep_alive_while_idle`
151    Http2KeepAliveWhileIdle,
152    /// Sets the maximum frame size to use for HTTP/2.
153    ///
154    /// Supported keys:
155    /// - `http2_max_frame_size`
156    Http2MaxFrameSize,
157    /// Only use HTTP/2 connections
158    ///
159    /// Supported keys:
160    /// - `http2_only`
161    Http2Only,
162    /// The pool max idle timeout
163    ///
164    /// This is the length of time an idle connection will be kept alive
165    ///
166    /// Supported keys:
167    /// - `pool_idle_timeout`
168    PoolIdleTimeout,
169    /// maximum number of idle connections per host
170    ///
171    /// Supported keys:
172    /// - `pool_max_idle_per_host`
173    PoolMaxIdlePerHost,
174    /// HTTP proxy to use for requests
175    ///
176    /// Supported keys:
177    /// - `proxy_url`
178    ProxyUrl,
179    /// PEM-formatted CA certificate for proxy connections
180    ///
181    /// Supported keys:
182    /// - `proxy_ca_certificate`
183    ProxyCaCertificate,
184    /// List of hosts that bypass proxy
185    ///
186    /// Supported keys:
187    /// - `proxy_excludes`
188    ProxyExcludes,
189    /// Randomize order addresses that the DNS resolution yields.
190    ///
191    /// This will spread the connections across more servers.
192    ///
193    /// <div class="warning">
194    ///
195    /// **Warning**
196    ///
197    /// This will override the DNS resolver configured by [`reqwest`].
198    ///
199    /// </div>
200    ///
201    /// Supported keys:
202    /// - `randomize_addresses`
203    RandomizeAddresses,
204    /// Read timeout
205    ///
206    /// The timeout applies to each read operation, and resets after a
207    /// successful read. This is useful for detecting stalled connections
208    /// when the size of the response is not known beforehand.
209    ///
210    /// Supported keys:
211    /// - `read_timeout`
212    ReadTimeout,
213    /// Request timeout
214    ///
215    /// The timeout is applied from when the request starts connecting until the
216    /// response body has finished
217    ///
218    /// Supported keys:
219    /// - `timeout`
220    Timeout,
221    /// User-Agent header to be used by this client
222    ///
223    /// Supported keys:
224    /// - `user_agent`
225    UserAgent,
226}
227
228impl AsRef<str> for ClientConfigKey {
229    fn as_ref(&self) -> &str {
230        match self {
231            Self::AllowHttp => "allow_http",
232            Self::AllowInvalidCertificates => "allow_invalid_certificates",
233            Self::NoSystemCertificates => "disable_system_certificates",
234            Self::ConnectTimeout => "connect_timeout",
235            Self::DefaultContentType => "default_content_type",
236            Self::Http1Only => "http1_only",
237            Self::Http2Only => "http2_only",
238            Self::Http2KeepAliveInterval => "http2_keep_alive_interval",
239            Self::Http2KeepAliveTimeout => "http2_keep_alive_timeout",
240            Self::Http2KeepAliveWhileIdle => "http2_keep_alive_while_idle",
241            Self::Http2MaxFrameSize => "http2_max_frame_size",
242            Self::PoolIdleTimeout => "pool_idle_timeout",
243            Self::PoolMaxIdlePerHost => "pool_max_idle_per_host",
244            Self::ProxyUrl => "proxy_url",
245            Self::ProxyCaCertificate => "proxy_ca_certificate",
246            Self::ProxyExcludes => "proxy_excludes",
247            Self::RandomizeAddresses => "randomize_addresses",
248            Self::ReadTimeout => "read_timeout",
249            Self::Timeout => "timeout",
250            Self::UserAgent => "user_agent",
251        }
252    }
253}
254
255impl FromStr for ClientConfigKey {
256    type Err = super::Error;
257
258    fn from_str(s: &str) -> Result<Self, Self::Err> {
259        match s {
260            "allow_http" => Ok(Self::AllowHttp),
261            "allow_invalid_certificates" => Ok(Self::AllowInvalidCertificates),
262            "disable_system_certificates" => Ok(Self::NoSystemCertificates),
263            "connect_timeout" => Ok(Self::ConnectTimeout),
264            "default_content_type" => Ok(Self::DefaultContentType),
265            "http1_only" => Ok(Self::Http1Only),
266            "http2_only" => Ok(Self::Http2Only),
267            "http2_keep_alive_interval" => Ok(Self::Http2KeepAliveInterval),
268            "http2_keep_alive_timeout" => Ok(Self::Http2KeepAliveTimeout),
269            "http2_keep_alive_while_idle" => Ok(Self::Http2KeepAliveWhileIdle),
270            "http2_max_frame_size" => Ok(Self::Http2MaxFrameSize),
271            "pool_idle_timeout" => Ok(Self::PoolIdleTimeout),
272            "pool_max_idle_per_host" => Ok(Self::PoolMaxIdlePerHost),
273            "proxy_url" => Ok(Self::ProxyUrl),
274            "proxy_ca_certificate" => Ok(Self::ProxyCaCertificate),
275            "proxy_excludes" => Ok(Self::ProxyExcludes),
276            "randomize_addresses" => Ok(Self::RandomizeAddresses),
277            "read_timeout" => Ok(Self::ReadTimeout),
278            "timeout" => Ok(Self::Timeout),
279            "user_agent" => Ok(Self::UserAgent),
280            _ => Err(super::Error::UnknownConfigurationKey {
281                store: "HTTP",
282                key: s.into(),
283            }),
284        }
285    }
286}
287
288/// Represents a CA certificate provided by the user.
289///
290/// This is used to configure the client to trust a specific certificate. See
291/// [Self::from_pem] for an example
292#[derive(Debug, Clone)]
293#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
294pub struct Certificate(reqwest::tls::Certificate);
295
296#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
297impl Certificate {
298    /// Create a `Certificate` from a PEM encoded certificate.
299    ///
300    /// # Example from a PEM file
301    ///
302    /// ```no_run
303    /// # use object_store::Certificate;
304    /// # use std::fs::File;
305    /// # use std::io::Read;
306    /// let mut buf = Vec::new();
307    /// File::open("my_cert.pem").unwrap()
308    ///   .read_to_end(&mut buf).unwrap();
309    /// let cert = Certificate::from_pem(&buf).unwrap();
310    ///
311    /// ```
312    pub fn from_pem(pem: &[u8]) -> Result<Self> {
313        Ok(Self(
314            reqwest::tls::Certificate::from_pem(pem).map_err(map_client_error)?,
315        ))
316    }
317
318    /// Create a collection of `Certificate` from a PEM encoded certificate
319    /// bundle.
320    ///
321    /// Files that contain such collections have extensions such as `.crt`,
322    /// `.cer` and `.pem` files.
323    pub fn from_pem_bundle(pem_bundle: &[u8]) -> Result<Vec<Self>> {
324        Ok(reqwest::tls::Certificate::from_pem_bundle(pem_bundle)
325            .map_err(map_client_error)?
326            .into_iter()
327            .map(Self)
328            .collect())
329    }
330
331    /// Create a `Certificate` from a binary DER encoded certificate.
332    pub fn from_der(der: &[u8]) -> Result<Self> {
333        Ok(Self(
334            reqwest::tls::Certificate::from_der(der).map_err(map_client_error)?,
335        ))
336    }
337}
338
339/// HTTP client configuration for remote object stores
340#[derive(Debug, Clone)]
341pub struct ClientOptions {
342    user_agent: Option<ConfigValue<HeaderValue>>,
343    #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
344    root_certificates: Vec<Certificate>,
345    no_system_certificates: ConfigValue<bool>,
346    content_type_map: HashMap<String, String>,
347    default_content_type: Option<String>,
348    default_headers: Option<HeaderMap>,
349    proxy_url: Option<String>,
350    proxy_ca_certificate: Option<String>,
351    proxy_excludes: Option<String>,
352    allow_http: ConfigValue<bool>,
353    allow_invalid_certificates: ConfigValue<bool>,
354    timeout: Option<ConfigValue<Duration>>,
355    connect_timeout: Option<ConfigValue<Duration>>,
356    read_timeout: Option<ConfigValue<Duration>>,
357    pool_idle_timeout: Option<ConfigValue<Duration>>,
358    pool_max_idle_per_host: Option<ConfigValue<usize>>,
359    http2_keep_alive_interval: Option<ConfigValue<Duration>>,
360    http2_keep_alive_timeout: Option<ConfigValue<Duration>>,
361    http2_keep_alive_while_idle: ConfigValue<bool>,
362    http2_max_frame_size: Option<ConfigValue<u32>>,
363    http1_only: ConfigValue<bool>,
364    http2_only: ConfigValue<bool>,
365    randomize_addresses: ConfigValue<bool>,
366}
367
368impl Default for ClientOptions {
369    fn default() -> Self {
370        // Defaults based on
371        // <https://docs.aws.amazon.com/sdkref/latest/guide/feature-smart-config-defaults.html>
372        // <https://docs.aws.amazon.com/whitepapers/latest/s3-optimizing-performance-best-practices/timeouts-and-retries-for-latency-sensitive-applications.html>
373        // Which recommend a connection timeout of 3.1s and a request timeout of 2s
374        //
375        // As object store requests may involve the transfer of non-trivial volumes of data
376        // we opt for a slightly higher default timeout of 30 seconds
377        Self {
378            user_agent: None,
379            #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
380            root_certificates: Default::default(),
381            no_system_certificates: false.into(),
382            content_type_map: Default::default(),
383            default_content_type: None,
384            default_headers: None,
385            proxy_url: None,
386            proxy_ca_certificate: None,
387            proxy_excludes: None,
388            allow_http: Default::default(),
389            allow_invalid_certificates: Default::default(),
390            timeout: Some(Duration::from_secs(30).into()),
391            connect_timeout: Some(Duration::from_secs(5).into()),
392            read_timeout: None,
393            pool_idle_timeout: None,
394            pool_max_idle_per_host: None,
395            http2_keep_alive_interval: None,
396            http2_keep_alive_timeout: None,
397            http2_keep_alive_while_idle: Default::default(),
398            http2_max_frame_size: None,
399            // HTTP/2 is known to be significantly slower than HTTP/1, so we default
400            // to HTTP/1 for now.
401            // https://github.com/apache/arrow-rs/issues/5194
402            http1_only: true.into(),
403            http2_only: Default::default(),
404            randomize_addresses: true.into(),
405        }
406    }
407}
408
409impl ClientOptions {
410    /// Create a new [`ClientOptions`] with default values
411    pub fn new() -> Self {
412        Default::default()
413    }
414
415    /// Set an option by key
416    pub fn with_config(mut self, key: ClientConfigKey, value: impl Into<String>) -> Self {
417        match key {
418            ClientConfigKey::AllowHttp => self.allow_http.parse(value),
419            ClientConfigKey::AllowInvalidCertificates => {
420                self.allow_invalid_certificates.parse(value)
421            }
422            ClientConfigKey::NoSystemCertificates => self.no_system_certificates.parse(value),
423            ClientConfigKey::ConnectTimeout => {
424                self.connect_timeout = Some(ConfigValue::Deferred(value.into()))
425            }
426            ClientConfigKey::ReadTimeout => {
427                self.read_timeout = Some(ConfigValue::Deferred(value.into()))
428            }
429            ClientConfigKey::DefaultContentType => self.default_content_type = Some(value.into()),
430            ClientConfigKey::Http1Only => self.http1_only.parse(value),
431            ClientConfigKey::Http2Only => self.http2_only.parse(value),
432            ClientConfigKey::Http2KeepAliveInterval => {
433                self.http2_keep_alive_interval = Some(ConfigValue::Deferred(value.into()))
434            }
435            ClientConfigKey::Http2KeepAliveTimeout => {
436                self.http2_keep_alive_timeout = Some(ConfigValue::Deferred(value.into()))
437            }
438            ClientConfigKey::Http2KeepAliveWhileIdle => {
439                self.http2_keep_alive_while_idle.parse(value)
440            }
441            ClientConfigKey::Http2MaxFrameSize => {
442                self.http2_max_frame_size = Some(ConfigValue::Deferred(value.into()))
443            }
444            ClientConfigKey::PoolIdleTimeout => {
445                self.pool_idle_timeout = Some(ConfigValue::Deferred(value.into()))
446            }
447            ClientConfigKey::PoolMaxIdlePerHost => {
448                self.pool_max_idle_per_host = Some(ConfigValue::Deferred(value.into()))
449            }
450            ClientConfigKey::ProxyUrl => self.proxy_url = Some(value.into()),
451            ClientConfigKey::ProxyCaCertificate => self.proxy_ca_certificate = Some(value.into()),
452            ClientConfigKey::ProxyExcludes => self.proxy_excludes = Some(value.into()),
453            ClientConfigKey::RandomizeAddresses => {
454                self.randomize_addresses.parse(value);
455            }
456            ClientConfigKey::Timeout => self.timeout = Some(ConfigValue::Deferred(value.into())),
457            ClientConfigKey::UserAgent => {
458                self.user_agent = Some(ConfigValue::Deferred(value.into()))
459            }
460        }
461        self
462    }
463
464    /// Get an option by key
465    pub fn get_config_value(&self, key: &ClientConfigKey) -> Option<String> {
466        match key {
467            ClientConfigKey::AllowHttp => Some(self.allow_http.to_string()),
468            ClientConfigKey::AllowInvalidCertificates => {
469                Some(self.allow_invalid_certificates.to_string())
470            }
471            ClientConfigKey::NoSystemCertificates => Some(self.no_system_certificates.to_string()),
472            ClientConfigKey::ConnectTimeout => self.connect_timeout.as_ref().map(fmt_duration),
473            ClientConfigKey::ReadTimeout => self.read_timeout.as_ref().map(fmt_duration),
474            ClientConfigKey::DefaultContentType => self.default_content_type.clone(),
475            ClientConfigKey::Http1Only => Some(self.http1_only.to_string()),
476            ClientConfigKey::Http2KeepAliveInterval => {
477                self.http2_keep_alive_interval.as_ref().map(fmt_duration)
478            }
479            ClientConfigKey::Http2KeepAliveTimeout => {
480                self.http2_keep_alive_timeout.as_ref().map(fmt_duration)
481            }
482            ClientConfigKey::Http2KeepAliveWhileIdle => {
483                Some(self.http2_keep_alive_while_idle.to_string())
484            }
485            ClientConfigKey::Http2MaxFrameSize => {
486                self.http2_max_frame_size.as_ref().map(|v| v.to_string())
487            }
488            ClientConfigKey::Http2Only => Some(self.http2_only.to_string()),
489            ClientConfigKey::PoolIdleTimeout => self.pool_idle_timeout.as_ref().map(fmt_duration),
490            ClientConfigKey::PoolMaxIdlePerHost => {
491                self.pool_max_idle_per_host.as_ref().map(|v| v.to_string())
492            }
493            ClientConfigKey::ProxyUrl => self.proxy_url.clone(),
494            ClientConfigKey::ProxyCaCertificate => self.proxy_ca_certificate.clone(),
495            ClientConfigKey::ProxyExcludes => self.proxy_excludes.clone(),
496            ClientConfigKey::RandomizeAddresses => Some(self.randomize_addresses.to_string()),
497            ClientConfigKey::Timeout => self.timeout.as_ref().map(fmt_duration),
498            ClientConfigKey::UserAgent => self
499                .user_agent
500                .as_ref()
501                .and_then(|v| v.get().ok())
502                .and_then(|v| v.to_str().ok().map(|s| s.to_string())),
503        }
504    }
505
506    /// Sets the [`User-Agent`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/User-Agent) header to be used by this client
507    ///
508    /// Default is based on the version of this crate
509    pub fn with_user_agent(mut self, agent: HeaderValue) -> Self {
510        self.user_agent = Some(agent.into());
511        self
512    }
513
514    /// Add a custom root certificate.
515    ///
516    /// This can be used to connect to a server that has a self-signed
517    /// certificate for example.
518    #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
519    pub fn with_root_certificate(mut self, certificate: Certificate) -> Self {
520        self.root_certificates.push(certificate);
521        self
522    }
523
524    /// Set the default [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for uploads
525    pub fn with_default_content_type(mut self, mime: impl Into<String>) -> Self {
526        self.default_content_type = Some(mime.into());
527        self
528    }
529
530    /// Set the [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for a given file extension
531    pub fn with_content_type_for_suffix(
532        mut self,
533        extension: impl Into<String>,
534        mime: impl Into<String>,
535    ) -> Self {
536        self.content_type_map.insert(extension.into(), mime.into());
537        self
538    }
539
540    /// Sets the default headers for every request
541    pub fn with_default_headers(mut self, headers: HeaderMap) -> Self {
542        self.default_headers = Some(headers);
543        self
544    }
545
546    /// Sets what protocol is allowed.
547    ///
548    /// If `allow_http` is :
549    /// * `false` (default):  Only HTTPS is allowed
550    /// * `true`:  HTTP and HTTPS are allowed
551    pub fn with_allow_http(mut self, allow_http: bool) -> Self {
552        self.allow_http = allow_http.into();
553        self
554    }
555
556    /// Allows connections to invalid SSL certificates
557    ///
558    /// If `allow_invalid_certificates` is :
559    /// * `false` (default):  Only valid HTTPS certificates are allowed
560    /// * `true`:  All HTTPS certificates are allowed
561    ///
562    /// <div class="warning">
563    ///
564    /// **Warning**
565    ///
566    /// You should think very carefully before using this method. If
567    /// invalid certificates are trusted, *any* certificate for *any* site
568    /// will be trusted for use. This includes expired certificates. This
569    /// introduces significant vulnerabilities, and should only be used
570    /// as a last resort or for testing
571    ///
572    /// </div>
573    pub fn with_allow_invalid_certificates(mut self, allow_invalid_certificates: bool) -> Self {
574        self.allow_invalid_certificates = allow_invalid_certificates.into();
575        self
576    }
577
578    /// Disable certificates provided by the system
579    ///
580    /// By default TLS certificates are validated using [`rustls-platform-verifier`],
581    /// which makes use of the system's trust store, in addition to any certificates
582    /// registered using [`Self::with_root_certificate`]. If disabled, instead [`rustls-webpki`]
583    /// is used with only the certificates registered using [`Self::with_root_certificate`].
584    ///
585    /// [`rustls-platform-verifier`]: https://crates.io/crates/rustls-platform-verifier
586    /// [`rustls-webpki`]: https://crates.io/crates/rustls-webpki
587    pub fn with_no_system_certificates(mut self, no_certs: bool) -> Self {
588        self.no_system_certificates = no_certs.into();
589        self
590    }
591
592    /// Only use HTTP/1 connections (default)
593    ///
594    /// # See Also
595    /// * [`Self::with_http2_only`] if you only want to use HTTP/2
596    /// * [`Self::with_allow_http2`] if you want to use HTTP/1 or HTTP/2
597    ///
598    /// <div class="warning">
599    /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104)
600    /// </div>
601    pub fn with_http1_only(mut self) -> Self {
602        self.http2_only = false.into();
603        self.http1_only = true.into();
604        self
605    }
606
607    /// Only use HTTP/2 connections
608    ///
609    /// # See Also
610    /// * [`Self::with_http1_only`] if you only want to use HTTP/1
611    /// * [`Self::with_allow_http2`] if you want to use HTTP/1 or HTTP/2
612    ///
613    /// <div class="warning">
614    /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104)
615    /// </div>
616    pub fn with_http2_only(mut self) -> Self {
617        self.http1_only = false.into();
618        self.http2_only = true.into();
619        self
620    }
621
622    /// Use HTTP/2 if supported, otherwise use HTTP/1.
623    ///
624    /// # See Also
625    /// * [`Self::with_http1_only`] if you only want to use HTTP/1
626    /// * [`Self::with_http2_only`] if you only want to use HTTP/2
627    ///
628    /// <div class="warning">
629    /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104)
630    /// </div>
631    pub fn with_allow_http2(mut self) -> Self {
632        self.http1_only = false.into();
633        self.http2_only = false.into();
634        self
635    }
636
637    /// Set a proxy URL to use for requests
638    pub fn with_proxy_url(mut self, proxy_url: impl Into<String>) -> Self {
639        self.proxy_url = Some(proxy_url.into());
640        self
641    }
642
643    /// Set a trusted proxy CA certificate
644    pub fn with_proxy_ca_certificate(mut self, proxy_ca_certificate: impl Into<String>) -> Self {
645        self.proxy_ca_certificate = Some(proxy_ca_certificate.into());
646        self
647    }
648
649    /// Set a list of hosts to exclude from proxy connections
650    pub fn with_proxy_excludes(mut self, proxy_excludes: impl Into<String>) -> Self {
651        self.proxy_excludes = Some(proxy_excludes.into());
652        self
653    }
654
655    /// Set timeout for the overall request
656    ///
657    /// The timeout starts from when the request starts connecting until the
658    /// response body has finished. If the request does not complete within the
659    /// timeout, the client returns a timeout error.
660    ///
661    /// Timeout errors are retried, subject to the [`RetryConfig`]
662    ///
663    /// Default is 30 seconds
664    ///
665    /// # See Also
666    /// * [`Self::with_timeout_disabled`] to disable the timeout
667    /// * [`Self::with_connect_timeout`] to set a timeout for the connect phase
668    ///
669    /// [`RetryConfig`]: crate::RetryConfig
670    pub fn with_timeout(mut self, timeout: Duration) -> Self {
671        self.timeout = Some(ConfigValue::Parsed(timeout));
672        self
673    }
674
675    /// Disables the request timeout
676    ///
677    /// # See Also
678    /// * [`Self::with_timeout`]
679    pub fn with_timeout_disabled(mut self) -> Self {
680        self.timeout = None;
681        self
682    }
683
684    /// Set a timeout for only the connect phase of a Client
685    ///
686    /// This is the time allowed for the client to establish a connection
687    /// and if the connection is not established within this time,
688    /// the client returns a timeout error.
689    ///
690    /// Timeout errors are retried, subject to the [`RetryConfig`]
691    ///
692    /// Default is 5 seconds
693    ///
694    /// # See Also
695    /// * [`Self::with_timeout`] to set a timeout for the overall request
696    /// * [`Self::with_connect_timeout_disabled`] to disable the connect timeout
697    ///
698    /// [`RetryConfig`]: crate::RetryConfig
699    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
700        self.connect_timeout = Some(ConfigValue::Parsed(timeout));
701        self
702    }
703
704    /// Disables the connection timeout
705    ///
706    /// # See Also
707    /// * [`Self::with_connect_timeout`]
708    pub fn with_connect_timeout_disabled(mut self) -> Self {
709        self.connect_timeout = None;
710        self
711    }
712
713    /// Set a read timeout
714    ///
715    /// The timeout applies to each read operation, and resets after a
716    /// successful read. This is useful for detecting stalled connections
717    /// when the size of the response is not known beforehand.
718    ///
719    /// Timeout errors are retried, subject to the [`RetryConfig`]
720    ///
721    /// Default is disabled (no read timeout)
722    ///
723    /// # See Also
724    /// * [`Self::with_read_timeout_disabled`] to disable the read timeout
725    /// * [`Self::with_timeout`] to set a timeout for the overall request
726    /// * [`Self::with_connect_timeout`] to set a timeout for the connect phase
727    ///
728    /// [`RetryConfig`]: crate::RetryConfig
729    pub fn with_read_timeout(mut self, timeout: Duration) -> Self {
730        self.read_timeout = Some(ConfigValue::Parsed(timeout));
731        self
732    }
733
734    /// Disables the read timeout
735    ///
736    /// # See Also
737    /// * [`Self::with_read_timeout`]
738    pub fn with_read_timeout_disabled(mut self) -> Self {
739        self.read_timeout = None;
740        self
741    }
742
743    /// Set the pool max idle timeout
744    ///
745    /// This is the length of time an idle connection will be kept alive
746    ///
747    /// Default is 90 seconds enforced by reqwest
748    pub fn with_pool_idle_timeout(mut self, timeout: Duration) -> Self {
749        self.pool_idle_timeout = Some(ConfigValue::Parsed(timeout));
750        self
751    }
752
753    /// Set the maximum number of idle connections per host
754    ///
755    /// Default is no limit enforced by reqwest
756    pub fn with_pool_max_idle_per_host(mut self, max: usize) -> Self {
757        self.pool_max_idle_per_host = Some(max.into());
758        self
759    }
760
761    /// Sets an interval for HTTP/2 Ping frames should be sent to keep a connection alive.
762    ///
763    /// Default is disabled enforced by reqwest
764    pub fn with_http2_keep_alive_interval(mut self, interval: Duration) -> Self {
765        self.http2_keep_alive_interval = Some(ConfigValue::Parsed(interval));
766        self
767    }
768
769    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
770    ///
771    /// If the ping is not acknowledged within the timeout, the connection will be closed.
772    /// Does nothing if `http2_keep_alive_interval` is disabled.
773    ///
774    /// Default is disabled enforced by reqwest
775    pub fn with_http2_keep_alive_timeout(mut self, interval: Duration) -> Self {
776        self.http2_keep_alive_timeout = Some(ConfigValue::Parsed(interval));
777        self
778    }
779
780    /// Enable HTTP/2 keep alive pings for idle connections
781    ///
782    /// If disabled, keep-alive pings are only sent while there are open request/response
783    /// streams. If enabled, pings are also sent when no streams are active
784    ///
785    /// Default is disabled enforced by reqwest
786    pub fn with_http2_keep_alive_while_idle(mut self) -> Self {
787        self.http2_keep_alive_while_idle = true.into();
788        self
789    }
790
791    /// Sets the maximum frame size to use for HTTP/2.
792    ///
793    /// Default is currently 16,384 but may change internally to optimize for common uses.
794    pub fn with_http2_max_frame_size(mut self, sz: u32) -> Self {
795        self.http2_max_frame_size = Some(ConfigValue::Parsed(sz));
796        self
797    }
798
799    /// Get the default headers defined through `ClientOptions::with_default_headers`
800    pub fn get_default_headers(&self) -> Option<&HeaderMap> {
801        self.default_headers.as_ref()
802    }
803
804    /// Get the mime type for the file in `path` to be uploaded
805    ///
806    /// Gets the file extension from `path`, and returns the
807    /// mime type if it was defined initially through
808    /// `ClientOptions::with_content_type_for_suffix`
809    ///
810    /// Otherwise, returns the default mime type if it was defined
811    /// earlier through `ClientOptions::with_default_content_type`
812    pub fn get_content_type(&self, path: &Path) -> Option<&str> {
813        match path.extension() {
814            Some(extension) => match self.content_type_map.get(extension) {
815                Some(ct) => Some(ct.as_str()),
816                None => self.default_content_type.as_deref(),
817            },
818            None => self.default_content_type.as_deref(),
819        }
820    }
821
822    /// Returns a copy of this [`ClientOptions`] with overrides necessary for metadata endpoint access
823    ///
824    /// In particular:
825    /// * Allows HTTP as metadata endpoints do not use TLS
826    /// * Configures a low connection timeout to provide quick feedback if not present
827    #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))]
828    pub(crate) fn metadata_options(&self) -> Self {
829        self.clone()
830            .with_allow_http(true)
831            .with_connect_timeout(Duration::from_secs(1))
832    }
833
834    #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
835    pub(crate) fn client(&self) -> Result<reqwest::Client> {
836        let mut builder = reqwest::ClientBuilder::new();
837
838        match &self.user_agent {
839            Some(user_agent) => builder = builder.user_agent(user_agent.get()?),
840            None => builder = builder.user_agent(DEFAULT_USER_AGENT),
841        }
842
843        if let Some(headers) = &self.default_headers {
844            builder = builder.default_headers(headers.clone())
845        }
846
847        if let Some(proxy) = &self.proxy_url {
848            let mut proxy = Proxy::all(proxy).map_err(map_client_error)?;
849
850            if let Some(certificate) = &self.proxy_ca_certificate {
851                let certificate = reqwest::tls::Certificate::from_pem(certificate.as_bytes())
852                    .map_err(map_client_error)?;
853
854                builder = builder.tls_certs_merge(std::iter::once(certificate));
855            }
856
857            if let Some(proxy_excludes) = &self.proxy_excludes {
858                let no_proxy = NoProxy::from_string(proxy_excludes);
859
860                proxy = proxy.no_proxy(no_proxy);
861            }
862
863            builder = builder.proxy(proxy);
864        }
865
866        let certs = self
867            .root_certificates
868            .iter()
869            .map(|certificate| certificate.0.clone());
870
871        if self.no_system_certificates.get()? {
872            builder = builder.tls_certs_only(certs);
873        } else {
874            builder = builder.tls_certs_merge(certs);
875        }
876
877        if let Some(timeout) = &self.timeout {
878            builder = builder.timeout(timeout.get()?)
879        }
880
881        if let Some(timeout) = &self.connect_timeout {
882            builder = builder.connect_timeout(timeout.get()?)
883        }
884
885        if let Some(timeout) = &self.read_timeout {
886            builder = builder.read_timeout(timeout.get()?)
887        }
888
889        if let Some(timeout) = &self.pool_idle_timeout {
890            builder = builder.pool_idle_timeout(timeout.get()?)
891        }
892
893        if let Some(max) = &self.pool_max_idle_per_host {
894            builder = builder.pool_max_idle_per_host(max.get()?)
895        }
896
897        if let Some(interval) = &self.http2_keep_alive_interval {
898            builder = builder.http2_keep_alive_interval(interval.get()?)
899        }
900
901        if let Some(interval) = &self.http2_keep_alive_timeout {
902            builder = builder.http2_keep_alive_timeout(interval.get()?)
903        }
904
905        if self.http2_keep_alive_while_idle.get()? {
906            builder = builder.http2_keep_alive_while_idle(true)
907        }
908
909        if let Some(sz) = &self.http2_max_frame_size {
910            builder = builder.http2_max_frame_size(Some(sz.get()?))
911        }
912
913        if self.http1_only.get()? {
914            builder = builder.http1_only()
915        }
916
917        if self.http2_only.get()? {
918            builder = builder.http2_prior_knowledge()
919        }
920
921        if self.allow_invalid_certificates.get()? {
922            builder = builder.danger_accept_invalid_certs(true)
923        }
924
925        // Explicitly disable compression, since it may be automatically enabled
926        // when certain reqwest features are enabled. Compression interferes
927        // with the `Content-Length` header, which is used to determine the
928        // size of objects.
929        builder = builder.no_gzip().no_brotli().no_zstd().no_deflate();
930
931        if self.randomize_addresses.get()? {
932            builder = builder.dns_resolver(Arc::new(dns::ShuffleResolver));
933        }
934
935        builder
936            .https_only(!self.allow_http.get()?)
937            .build()
938            .map_err(map_client_error)
939    }
940
941    #[cfg(all(feature = "reqwest", target_arch = "wasm32", target_os = "unknown"))]
942    pub(crate) fn client(&self) -> Result<reqwest::Client> {
943        let mut builder = reqwest::ClientBuilder::new();
944
945        match &self.user_agent {
946            Some(user_agent) => builder = builder.user_agent(user_agent.get()?),
947            None => builder = builder.user_agent(DEFAULT_USER_AGENT),
948        }
949
950        if let Some(headers) = &self.default_headers {
951            builder = builder.default_headers(headers.clone())
952        }
953
954        builder.build().map_err(map_client_error)
955    }
956}
957
958pub(crate) trait GetOptionsExt {
959    fn with_get_options(self, options: GetOptions) -> Self;
960}
961
962impl GetOptionsExt for HttpRequestBuilder {
963    fn with_get_options(mut self, options: GetOptions) -> Self {
964        use hyper::header::*;
965
966        let GetOptions {
967            if_match,
968            if_none_match,
969            if_modified_since,
970            if_unmodified_since,
971            range,
972            version: _,
973            head: _,
974            extensions,
975        } = options;
976
977        if let Some(range) = range {
978            self = self.header(RANGE, range.to_string());
979        }
980
981        if let Some(tag) = if_match {
982            self = self.header(IF_MATCH, tag);
983        }
984
985        if let Some(tag) = if_none_match {
986            self = self.header(IF_NONE_MATCH, tag);
987        }
988
989        const DATE_FORMAT: &str = "%a, %d %b %Y %H:%M:%S GMT";
990        if let Some(date) = if_unmodified_since {
991            self = self.header(IF_UNMODIFIED_SINCE, date.format(DATE_FORMAT).to_string());
992        }
993
994        if let Some(date) = if_modified_since {
995            self = self.header(IF_MODIFIED_SINCE, date.format(DATE_FORMAT).to_string());
996        }
997
998        self = self.extensions(extensions);
999
1000        self
1001    }
1002}
1003
1004/// Provides credentials for use when signing requests
1005#[async_trait]
1006pub trait CredentialProvider: std::fmt::Debug + Send + Sync {
1007    /// The type of credential returned by this provider
1008    type Credential;
1009
1010    /// Return a credential
1011    async fn get_credential(&self) -> Result<Arc<Self::Credential>>;
1012}
1013
1014/// A static set of credentials
1015#[derive(Debug)]
1016pub struct StaticCredentialProvider<T> {
1017    credential: Arc<T>,
1018}
1019
1020impl<T> StaticCredentialProvider<T> {
1021    /// A [`CredentialProvider`] for a static credential of type `T`
1022    pub fn new(credential: T) -> Self {
1023        Self {
1024            credential: Arc::new(credential),
1025        }
1026    }
1027}
1028
1029#[async_trait]
1030impl<T> CredentialProvider for StaticCredentialProvider<T>
1031where
1032    T: std::fmt::Debug + Send + Sync,
1033{
1034    type Credential = T;
1035
1036    async fn get_credential(&self) -> Result<Arc<T>> {
1037        Ok(Arc::clone(&self.credential))
1038    }
1039}
1040
1041#[cfg(any(feature = "aws-base", feature = "azure-base", feature = "gcp-base"))]
1042mod cloud {
1043    use super::*;
1044    use crate::RetryConfig;
1045    use crate::client::token::{TemporaryToken, TokenCache};
1046
1047    /// A [`CredentialProvider`] that uses [`HttpClient`] to fetch temporary tokens
1048    #[derive(Debug)]
1049    pub(crate) struct TokenCredentialProvider<T: TokenProvider> {
1050        inner: T,
1051        client: HttpClient,
1052        retry: RetryConfig,
1053        cache: TokenCache<Arc<T::Credential>>,
1054    }
1055
1056    impl<T: TokenProvider> TokenCredentialProvider<T> {
1057        pub(crate) fn new(inner: T, client: HttpClient, retry: RetryConfig) -> Self {
1058            Self {
1059                inner,
1060                client,
1061                retry,
1062                cache: Default::default(),
1063            }
1064        }
1065
1066        /// Override the minimum remaining TTL for a cached token to be used
1067        #[cfg(any(feature = "aws-base", feature = "gcp-base"))]
1068        pub(crate) fn with_min_ttl(mut self, min_ttl: Duration) -> Self {
1069            self.cache = self.cache.with_min_ttl(min_ttl);
1070            self
1071        }
1072    }
1073
1074    #[async_trait]
1075    impl<T: TokenProvider> CredentialProvider for TokenCredentialProvider<T> {
1076        type Credential = T::Credential;
1077
1078        async fn get_credential(&self) -> Result<Arc<Self::Credential>> {
1079            self.cache
1080                .get_or_insert_with(|| self.inner.fetch_token(&self.client, &self.retry))
1081                .await
1082        }
1083    }
1084
1085    #[async_trait]
1086    pub(crate) trait TokenProvider: std::fmt::Debug + Send + Sync {
1087        type Credential: std::fmt::Debug + Send + Sync;
1088
1089        async fn fetch_token(
1090            &self,
1091            client: &HttpClient,
1092            retry: &RetryConfig,
1093        ) -> Result<TemporaryToken<Arc<Self::Credential>>>;
1094    }
1095}
1096
1097use crate::client::builder::HttpRequestBuilder;
1098#[cfg(any(feature = "aws-base", feature = "azure-base", feature = "gcp-base"))]
1099pub(crate) use cloud::*;
1100
1101#[cfg(test)]
1102mod tests {
1103    use super::*;
1104    use std::collections::HashMap;
1105
1106    #[test]
1107    fn client_test_config_from_map() {
1108        let allow_http = "true".to_string();
1109        let allow_invalid_certificates = "false".to_string();
1110        let connect_timeout = "90 seconds".to_string();
1111        let default_content_type = "object_store:fake_default_content_type".to_string();
1112        let http1_only = "true".to_string();
1113        let http2_only = "false".to_string();
1114        let http2_keep_alive_interval = "90 seconds".to_string();
1115        let http2_keep_alive_timeout = "91 seconds".to_string();
1116        let http2_keep_alive_while_idle = "92 seconds".to_string();
1117        let http2_max_frame_size = "1337".to_string();
1118        let no_system_certificates = "true".to_string();
1119        let pool_idle_timeout = "93 seconds".to_string();
1120        let pool_max_idle_per_host = "94".to_string();
1121        let proxy_url = "https://fake_proxy_url".to_string();
1122        let read_timeout = "45 seconds".to_string();
1123        let timeout = "95 seconds".to_string();
1124        let user_agent = "object_store:fake_user_agent".to_string();
1125
1126        let options = HashMap::from([
1127            ("allow_http", allow_http.clone()),
1128            (
1129                "allow_invalid_certificates",
1130                allow_invalid_certificates.clone(),
1131            ),
1132            ("connect_timeout", connect_timeout.clone()),
1133            ("default_content_type", default_content_type.clone()),
1134            ("http1_only", http1_only.clone()),
1135            ("http2_only", http2_only.clone()),
1136            (
1137                "http2_keep_alive_interval",
1138                http2_keep_alive_interval.clone(),
1139            ),
1140            ("http2_keep_alive_timeout", http2_keep_alive_timeout.clone()),
1141            (
1142                "http2_keep_alive_while_idle",
1143                http2_keep_alive_while_idle.clone(),
1144            ),
1145            ("http2_max_frame_size", http2_max_frame_size.clone()),
1146            (
1147                "disable_system_certificates",
1148                no_system_certificates.clone(),
1149            ),
1150            ("pool_idle_timeout", pool_idle_timeout.clone()),
1151            ("pool_max_idle_per_host", pool_max_idle_per_host.clone()),
1152            ("proxy_url", proxy_url.clone()),
1153            ("read_timeout", read_timeout.clone()),
1154            ("timeout", timeout.clone()),
1155            ("user_agent", user_agent.clone()),
1156        ]);
1157
1158        let builder = options
1159            .into_iter()
1160            .fold(ClientOptions::new(), |builder, (key, value)| {
1161                builder.with_config(key.parse().unwrap(), value)
1162            });
1163
1164        assert_eq!(
1165            builder
1166                .get_config_value(&ClientConfigKey::AllowHttp)
1167                .unwrap(),
1168            allow_http
1169        );
1170        assert_eq!(
1171            builder
1172                .get_config_value(&ClientConfigKey::AllowInvalidCertificates)
1173                .unwrap(),
1174            allow_invalid_certificates
1175        );
1176        assert_eq!(
1177            builder
1178                .get_config_value(&ClientConfigKey::ConnectTimeout)
1179                .unwrap(),
1180            connect_timeout
1181        );
1182        assert_eq!(
1183            builder
1184                .get_config_value(&ClientConfigKey::DefaultContentType)
1185                .unwrap(),
1186            default_content_type
1187        );
1188        assert_eq!(
1189            builder
1190                .get_config_value(&ClientConfigKey::Http1Only)
1191                .unwrap(),
1192            http1_only
1193        );
1194        assert_eq!(
1195            builder
1196                .get_config_value(&ClientConfigKey::Http2Only)
1197                .unwrap(),
1198            http2_only
1199        );
1200        assert_eq!(
1201            builder
1202                .get_config_value(&ClientConfigKey::Http2KeepAliveInterval)
1203                .unwrap(),
1204            http2_keep_alive_interval
1205        );
1206        assert_eq!(
1207            builder
1208                .get_config_value(&ClientConfigKey::Http2KeepAliveTimeout)
1209                .unwrap(),
1210            http2_keep_alive_timeout
1211        );
1212        assert_eq!(
1213            builder
1214                .get_config_value(&ClientConfigKey::Http2KeepAliveWhileIdle)
1215                .unwrap(),
1216            http2_keep_alive_while_idle
1217        );
1218        assert_eq!(
1219            builder
1220                .get_config_value(&ClientConfigKey::Http2MaxFrameSize)
1221                .unwrap(),
1222            http2_max_frame_size
1223        );
1224        assert_eq!(
1225            builder
1226                .get_config_value(&ClientConfigKey::NoSystemCertificates)
1227                .unwrap(),
1228            no_system_certificates
1229        );
1230
1231        assert_eq!(
1232            builder
1233                .get_config_value(&ClientConfigKey::PoolIdleTimeout)
1234                .unwrap(),
1235            pool_idle_timeout
1236        );
1237        assert_eq!(
1238            builder
1239                .get_config_value(&ClientConfigKey::PoolMaxIdlePerHost)
1240                .unwrap(),
1241            pool_max_idle_per_host
1242        );
1243        assert_eq!(
1244            builder
1245                .get_config_value(&ClientConfigKey::ProxyUrl)
1246                .unwrap(),
1247            proxy_url
1248        );
1249        assert_eq!(
1250            builder
1251                .get_config_value(&ClientConfigKey::ReadTimeout)
1252                .unwrap(),
1253            read_timeout
1254        );
1255        assert_eq!(
1256            builder.get_config_value(&ClientConfigKey::Timeout).unwrap(),
1257            timeout
1258        );
1259        assert_eq!(
1260            builder
1261                .get_config_value(&ClientConfigKey::UserAgent)
1262                .unwrap(),
1263            user_agent
1264        );
1265    }
1266}
object_store/client/mod.rs

object_store/client/
mod.rs
