| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| //! Generic utilities for network based [`ObjectStore`] implementations |
| //! |
| //! [`ObjectStore`]: crate::ObjectStore |
| |
| pub(crate) mod backoff; |
| |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| mod dns; |
| |
| #[cfg(not(target_arch = "wasm32"))] |
| #[cfg(test)] |
| pub(crate) mod mock_server; |
| |
| pub(crate) mod retry; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub(crate) mod pagination; |
| |
| pub(crate) mod get; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub(crate) mod list; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub(crate) mod token; |
| |
| pub(crate) mod header; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base"))] |
| pub(crate) mod s3; |
| |
| pub(crate) mod builder; |
| mod http; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub(crate) mod parts; |
| pub use http::*; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| mod crypto; |
| |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub use crypto::*; |
| |
| use ::http::header::{HeaderMap, HeaderValue}; |
| use async_trait::async_trait; |
| use serde::{Deserialize, Serialize}; |
| use std::collections::HashMap; |
| use std::str::FromStr; |
| use std::sync::Arc; |
| use std::time::Duration; |
| |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| use reqwest::{NoProxy, Proxy}; |
| |
| use crate::config::{ConfigValue, fmt_duration}; |
| use crate::path::Path; |
| use crate::{GetOptions, Result}; |
| |
| #[cfg(feature = "reqwest")] |
| fn map_client_error(e: reqwest::Error) -> super::Error { |
| super::Error::Generic { |
| store: "HTTP client", |
| source: Box::new(e), |
| } |
| } |
| |
| #[cfg(feature = "reqwest")] |
| static DEFAULT_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),); |
| |
| /// Configuration keys for [`ClientOptions`] |
| #[derive(PartialEq, Eq, Hash, Clone, Debug, Copy, Deserialize, Serialize)] |
| #[non_exhaustive] |
| pub enum ClientConfigKey { |
| /// Allow non-TLS, i.e. non-HTTPS connections |
| /// |
| /// Supported keys: |
| /// - `allow_http` |
| AllowHttp, |
| /// Skip certificate validation on https connections. |
| /// |
| /// <div class="warning"> |
| /// |
| /// **Warning** |
| /// |
| /// You should think very carefully before using this method. If |
| /// invalid certificates are trusted, *any* certificate for *any* site |
| /// will be trusted for use. This includes expired certificates. This |
| /// introduces significant vulnerabilities, and should only be used |
| /// as a last resort or for testing |
| /// |
| /// </div> |
| /// |
| /// Supported keys: |
| /// - `allow_invalid_certificates` |
| AllowInvalidCertificates, |
| /// Disable certificate validation using the operating system's certificate facilities. |
| /// |
| /// See [`ClientOptions::with_no_system_certificates`] |
| /// |
| /// Supported keys: |
| /// |
| /// - `disable_system_certificates` |
| NoSystemCertificates, |
| /// Timeout for only the connect phase of a Client |
| /// |
| /// Supported keys: |
| /// - `connect_timeout` |
| ConnectTimeout, |
| /// default [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for uploads |
| /// |
| /// Supported keys: |
| /// - `default_content_type` |
| DefaultContentType, |
| /// Only use HTTP/1 connections |
| /// |
| /// Supported keys: |
| /// - `http1_only` |
| Http1Only, |
| /// Interval for HTTP/2 Ping frames should be sent to keep a connection alive. |
| /// |
| /// Supported keys: |
| /// - `http2_keep_alive_interval` |
| Http2KeepAliveInterval, |
| /// Timeout for receiving an acknowledgement of the keep-alive ping. |
| /// |
| /// Supported keys: |
| /// - `http2_keep_alive_timeout` |
| Http2KeepAliveTimeout, |
| /// Enable HTTP/2 keep alive pings for idle connections |
| /// |
| /// Supported keys: |
| /// - `http2_keep_alive_while_idle` |
| Http2KeepAliveWhileIdle, |
| /// Sets the maximum frame size to use for HTTP/2. |
| /// |
| /// Supported keys: |
| /// - `http2_max_frame_size` |
| Http2MaxFrameSize, |
| /// Only use HTTP/2 connections |
| /// |
| /// Supported keys: |
| /// - `http2_only` |
| Http2Only, |
| /// The pool max idle timeout |
| /// |
| /// This is the length of time an idle connection will be kept alive |
| /// |
| /// Supported keys: |
| /// - `pool_idle_timeout` |
| PoolIdleTimeout, |
| /// maximum number of idle connections per host |
| /// |
| /// Supported keys: |
| /// - `pool_max_idle_per_host` |
| PoolMaxIdlePerHost, |
| /// HTTP proxy to use for requests |
| /// |
| /// Supported keys: |
| /// - `proxy_url` |
| ProxyUrl, |
| /// PEM-formatted CA certificate for proxy connections |
| /// |
| /// Supported keys: |
| /// - `proxy_ca_certificate` |
| ProxyCaCertificate, |
| /// List of hosts that bypass proxy |
| /// |
| /// Supported keys: |
| /// - `proxy_excludes` |
| ProxyExcludes, |
| /// Randomize order addresses that the DNS resolution yields. |
| /// |
| /// This will spread the connections across more servers. |
| /// |
| /// <div class="warning"> |
| /// |
| /// **Warning** |
| /// |
| /// This will override the DNS resolver configured by [`reqwest`]. |
| /// |
| /// </div> |
| /// |
| /// Supported keys: |
| /// - `randomize_addresses` |
| RandomizeAddresses, |
| /// Read timeout |
| /// |
| /// The timeout applies to each read operation, and resets after a |
| /// successful read. This is useful for detecting stalled connections |
| /// when the size of the response is not known beforehand. |
| /// |
| /// Supported keys: |
| /// - `read_timeout` |
| ReadTimeout, |
| /// Request timeout |
| /// |
| /// The timeout is applied from when the request starts connecting until the |
| /// response body has finished |
| /// |
| /// Supported keys: |
| /// - `timeout` |
| Timeout, |
| /// User-Agent header to be used by this client |
| /// |
| /// Supported keys: |
| /// - `user_agent` |
| UserAgent, |
| } |
| |
| impl AsRef<str> for ClientConfigKey { |
| fn as_ref(&self) -> &str { |
| match self { |
| Self::AllowHttp => "allow_http", |
| Self::AllowInvalidCertificates => "allow_invalid_certificates", |
| Self::NoSystemCertificates => "disable_system_certificates", |
| Self::ConnectTimeout => "connect_timeout", |
| Self::DefaultContentType => "default_content_type", |
| Self::Http1Only => "http1_only", |
| Self::Http2Only => "http2_only", |
| Self::Http2KeepAliveInterval => "http2_keep_alive_interval", |
| Self::Http2KeepAliveTimeout => "http2_keep_alive_timeout", |
| Self::Http2KeepAliveWhileIdle => "http2_keep_alive_while_idle", |
| Self::Http2MaxFrameSize => "http2_max_frame_size", |
| Self::PoolIdleTimeout => "pool_idle_timeout", |
| Self::PoolMaxIdlePerHost => "pool_max_idle_per_host", |
| Self::ProxyUrl => "proxy_url", |
| Self::ProxyCaCertificate => "proxy_ca_certificate", |
| Self::ProxyExcludes => "proxy_excludes", |
| Self::RandomizeAddresses => "randomize_addresses", |
| Self::ReadTimeout => "read_timeout", |
| Self::Timeout => "timeout", |
| Self::UserAgent => "user_agent", |
| } |
| } |
| } |
| |
| impl FromStr for ClientConfigKey { |
| type Err = super::Error; |
| |
| fn from_str(s: &str) -> Result<Self, Self::Err> { |
| match s { |
| "allow_http" => Ok(Self::AllowHttp), |
| "allow_invalid_certificates" => Ok(Self::AllowInvalidCertificates), |
| "disable_system_certificates" => Ok(Self::NoSystemCertificates), |
| "connect_timeout" => Ok(Self::ConnectTimeout), |
| "default_content_type" => Ok(Self::DefaultContentType), |
| "http1_only" => Ok(Self::Http1Only), |
| "http2_only" => Ok(Self::Http2Only), |
| "http2_keep_alive_interval" => Ok(Self::Http2KeepAliveInterval), |
| "http2_keep_alive_timeout" => Ok(Self::Http2KeepAliveTimeout), |
| "http2_keep_alive_while_idle" => Ok(Self::Http2KeepAliveWhileIdle), |
| "http2_max_frame_size" => Ok(Self::Http2MaxFrameSize), |
| "pool_idle_timeout" => Ok(Self::PoolIdleTimeout), |
| "pool_max_idle_per_host" => Ok(Self::PoolMaxIdlePerHost), |
| "proxy_url" => Ok(Self::ProxyUrl), |
| "proxy_ca_certificate" => Ok(Self::ProxyCaCertificate), |
| "proxy_excludes" => Ok(Self::ProxyExcludes), |
| "randomize_addresses" => Ok(Self::RandomizeAddresses), |
| "read_timeout" => Ok(Self::ReadTimeout), |
| "timeout" => Ok(Self::Timeout), |
| "user_agent" => Ok(Self::UserAgent), |
| _ => Err(super::Error::UnknownConfigurationKey { |
| store: "HTTP", |
| key: s.into(), |
| }), |
| } |
| } |
| } |
| |
| /// Represents a CA certificate provided by the user. |
| /// |
| /// This is used to configure the client to trust a specific certificate. See |
| /// [Self::from_pem] for an example |
| #[derive(Debug, Clone)] |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| pub struct Certificate(reqwest::tls::Certificate); |
| |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| impl Certificate { |
| /// Create a `Certificate` from a PEM encoded certificate. |
| /// |
| /// # Example from a PEM file |
| /// |
| /// ```no_run |
| /// # use object_store::Certificate; |
| /// # use std::fs::File; |
| /// # use std::io::Read; |
| /// let mut buf = Vec::new(); |
| /// File::open("my_cert.pem").unwrap() |
| /// .read_to_end(&mut buf).unwrap(); |
| /// let cert = Certificate::from_pem(&buf).unwrap(); |
| /// |
| /// ``` |
| pub fn from_pem(pem: &[u8]) -> Result<Self> { |
| Ok(Self( |
| reqwest::tls::Certificate::from_pem(pem).map_err(map_client_error)?, |
| )) |
| } |
| |
| /// Create a collection of `Certificate` from a PEM encoded certificate |
| /// bundle. |
| /// |
| /// Files that contain such collections have extensions such as `.crt`, |
| /// `.cer` and `.pem` files. |
| pub fn from_pem_bundle(pem_bundle: &[u8]) -> Result<Vec<Self>> { |
| Ok(reqwest::tls::Certificate::from_pem_bundle(pem_bundle) |
| .map_err(map_client_error)? |
| .into_iter() |
| .map(Self) |
| .collect()) |
| } |
| |
| /// Create a `Certificate` from a binary DER encoded certificate. |
| pub fn from_der(der: &[u8]) -> Result<Self> { |
| Ok(Self( |
| reqwest::tls::Certificate::from_der(der).map_err(map_client_error)?, |
| )) |
| } |
| } |
| |
| /// HTTP client configuration for remote object stores |
| #[derive(Debug, Clone)] |
| pub struct ClientOptions { |
| user_agent: Option<ConfigValue<HeaderValue>>, |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| root_certificates: Vec<Certificate>, |
| no_system_certificates: ConfigValue<bool>, |
| content_type_map: HashMap<String, String>, |
| default_content_type: Option<String>, |
| default_headers: Option<HeaderMap>, |
| proxy_url: Option<String>, |
| proxy_ca_certificate: Option<String>, |
| proxy_excludes: Option<String>, |
| allow_http: ConfigValue<bool>, |
| allow_invalid_certificates: ConfigValue<bool>, |
| timeout: Option<ConfigValue<Duration>>, |
| connect_timeout: Option<ConfigValue<Duration>>, |
| read_timeout: Option<ConfigValue<Duration>>, |
| pool_idle_timeout: Option<ConfigValue<Duration>>, |
| pool_max_idle_per_host: Option<ConfigValue<usize>>, |
| http2_keep_alive_interval: Option<ConfigValue<Duration>>, |
| http2_keep_alive_timeout: Option<ConfigValue<Duration>>, |
| http2_keep_alive_while_idle: ConfigValue<bool>, |
| http2_max_frame_size: Option<ConfigValue<u32>>, |
| http1_only: ConfigValue<bool>, |
| http2_only: ConfigValue<bool>, |
| randomize_addresses: ConfigValue<bool>, |
| } |
| |
| impl Default for ClientOptions { |
| fn default() -> Self { |
| // Defaults based on |
| // <https://docs.aws.amazon.com/sdkref/latest/guide/feature-smart-config-defaults.html> |
| // <https://docs.aws.amazon.com/whitepapers/latest/s3-optimizing-performance-best-practices/timeouts-and-retries-for-latency-sensitive-applications.html> |
| // Which recommend a connection timeout of 3.1s and a request timeout of 2s |
| // |
| // As object store requests may involve the transfer of non-trivial volumes of data |
| // we opt for a slightly higher default timeout of 30 seconds |
| Self { |
| user_agent: None, |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| root_certificates: Default::default(), |
| no_system_certificates: false.into(), |
| content_type_map: Default::default(), |
| default_content_type: None, |
| default_headers: None, |
| proxy_url: None, |
| proxy_ca_certificate: None, |
| proxy_excludes: None, |
| allow_http: Default::default(), |
| allow_invalid_certificates: Default::default(), |
| timeout: Some(Duration::from_secs(30).into()), |
| connect_timeout: Some(Duration::from_secs(5).into()), |
| read_timeout: None, |
| pool_idle_timeout: None, |
| pool_max_idle_per_host: None, |
| http2_keep_alive_interval: None, |
| http2_keep_alive_timeout: None, |
| http2_keep_alive_while_idle: Default::default(), |
| http2_max_frame_size: None, |
| // HTTP/2 is known to be significantly slower than HTTP/1, so we default |
| // to HTTP/1 for now. |
| // https://github.com/apache/arrow-rs/issues/5194 |
| http1_only: true.into(), |
| http2_only: Default::default(), |
| randomize_addresses: true.into(), |
| } |
| } |
| } |
| |
| impl ClientOptions { |
| /// Create a new [`ClientOptions`] with default values |
| pub fn new() -> Self { |
| Default::default() |
| } |
| |
| /// Set an option by key |
| pub fn with_config(mut self, key: ClientConfigKey, value: impl Into<String>) -> Self { |
| match key { |
| ClientConfigKey::AllowHttp => self.allow_http.parse(value), |
| ClientConfigKey::AllowInvalidCertificates => { |
| self.allow_invalid_certificates.parse(value) |
| } |
| ClientConfigKey::NoSystemCertificates => self.no_system_certificates.parse(value), |
| ClientConfigKey::ConnectTimeout => { |
| self.connect_timeout = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::ReadTimeout => { |
| self.read_timeout = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::DefaultContentType => self.default_content_type = Some(value.into()), |
| ClientConfigKey::Http1Only => self.http1_only.parse(value), |
| ClientConfigKey::Http2Only => self.http2_only.parse(value), |
| ClientConfigKey::Http2KeepAliveInterval => { |
| self.http2_keep_alive_interval = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::Http2KeepAliveTimeout => { |
| self.http2_keep_alive_timeout = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::Http2KeepAliveWhileIdle => { |
| self.http2_keep_alive_while_idle.parse(value) |
| } |
| ClientConfigKey::Http2MaxFrameSize => { |
| self.http2_max_frame_size = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::PoolIdleTimeout => { |
| self.pool_idle_timeout = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::PoolMaxIdlePerHost => { |
| self.pool_max_idle_per_host = Some(ConfigValue::Deferred(value.into())) |
| } |
| ClientConfigKey::ProxyUrl => self.proxy_url = Some(value.into()), |
| ClientConfigKey::ProxyCaCertificate => self.proxy_ca_certificate = Some(value.into()), |
| ClientConfigKey::ProxyExcludes => self.proxy_excludes = Some(value.into()), |
| ClientConfigKey::RandomizeAddresses => { |
| self.randomize_addresses.parse(value); |
| } |
| ClientConfigKey::Timeout => self.timeout = Some(ConfigValue::Deferred(value.into())), |
| ClientConfigKey::UserAgent => { |
| self.user_agent = Some(ConfigValue::Deferred(value.into())) |
| } |
| } |
| self |
| } |
| |
| /// Get an option by key |
| pub fn get_config_value(&self, key: &ClientConfigKey) -> Option<String> { |
| match key { |
| ClientConfigKey::AllowHttp => Some(self.allow_http.to_string()), |
| ClientConfigKey::AllowInvalidCertificates => { |
| Some(self.allow_invalid_certificates.to_string()) |
| } |
| ClientConfigKey::NoSystemCertificates => Some(self.no_system_certificates.to_string()), |
| ClientConfigKey::ConnectTimeout => self.connect_timeout.as_ref().map(fmt_duration), |
| ClientConfigKey::ReadTimeout => self.read_timeout.as_ref().map(fmt_duration), |
| ClientConfigKey::DefaultContentType => self.default_content_type.clone(), |
| ClientConfigKey::Http1Only => Some(self.http1_only.to_string()), |
| ClientConfigKey::Http2KeepAliveInterval => { |
| self.http2_keep_alive_interval.as_ref().map(fmt_duration) |
| } |
| ClientConfigKey::Http2KeepAliveTimeout => { |
| self.http2_keep_alive_timeout.as_ref().map(fmt_duration) |
| } |
| ClientConfigKey::Http2KeepAliveWhileIdle => { |
| Some(self.http2_keep_alive_while_idle.to_string()) |
| } |
| ClientConfigKey::Http2MaxFrameSize => { |
| self.http2_max_frame_size.as_ref().map(|v| v.to_string()) |
| } |
| ClientConfigKey::Http2Only => Some(self.http2_only.to_string()), |
| ClientConfigKey::PoolIdleTimeout => self.pool_idle_timeout.as_ref().map(fmt_duration), |
| ClientConfigKey::PoolMaxIdlePerHost => { |
| self.pool_max_idle_per_host.as_ref().map(|v| v.to_string()) |
| } |
| ClientConfigKey::ProxyUrl => self.proxy_url.clone(), |
| ClientConfigKey::ProxyCaCertificate => self.proxy_ca_certificate.clone(), |
| ClientConfigKey::ProxyExcludes => self.proxy_excludes.clone(), |
| ClientConfigKey::RandomizeAddresses => Some(self.randomize_addresses.to_string()), |
| ClientConfigKey::Timeout => self.timeout.as_ref().map(fmt_duration), |
| ClientConfigKey::UserAgent => self |
| .user_agent |
| .as_ref() |
| .and_then(|v| v.get().ok()) |
| .and_then(|v| v.to_str().ok().map(|s| s.to_string())), |
| } |
| } |
| |
| /// Sets the [`User-Agent`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/User-Agent) header to be used by this client |
| /// |
| /// Default is based on the version of this crate |
| pub fn with_user_agent(mut self, agent: HeaderValue) -> Self { |
| self.user_agent = Some(agent.into()); |
| self |
| } |
| |
| /// Add a custom root certificate. |
| /// |
| /// This can be used to connect to a server that has a self-signed |
| /// certificate for example. |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| pub fn with_root_certificate(mut self, certificate: Certificate) -> Self { |
| self.root_certificates.push(certificate); |
| self |
| } |
| |
| /// Set the default [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for uploads |
| pub fn with_default_content_type(mut self, mime: impl Into<String>) -> Self { |
| self.default_content_type = Some(mime.into()); |
| self |
| } |
| |
| /// Set the [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Content-Type) for a given file extension |
| pub fn with_content_type_for_suffix( |
| mut self, |
| extension: impl Into<String>, |
| mime: impl Into<String>, |
| ) -> Self { |
| self.content_type_map.insert(extension.into(), mime.into()); |
| self |
| } |
| |
| /// Sets the default headers for every request |
| pub fn with_default_headers(mut self, headers: HeaderMap) -> Self { |
| self.default_headers = Some(headers); |
| self |
| } |
| |
| /// Sets what protocol is allowed. |
| /// |
| /// If `allow_http` is : |
| /// * `false` (default): Only HTTPS is allowed |
| /// * `true`: HTTP and HTTPS are allowed |
| pub fn with_allow_http(mut self, allow_http: bool) -> Self { |
| self.allow_http = allow_http.into(); |
| self |
| } |
| |
| /// Allows connections to invalid SSL certificates |
| /// |
| /// If `allow_invalid_certificates` is : |
| /// * `false` (default): Only valid HTTPS certificates are allowed |
| /// * `true`: All HTTPS certificates are allowed |
| /// |
| /// <div class="warning"> |
| /// |
| /// **Warning** |
| /// |
| /// You should think very carefully before using this method. If |
| /// invalid certificates are trusted, *any* certificate for *any* site |
| /// will be trusted for use. This includes expired certificates. This |
| /// introduces significant vulnerabilities, and should only be used |
| /// as a last resort or for testing |
| /// |
| /// </div> |
| pub fn with_allow_invalid_certificates(mut self, allow_invalid_certificates: bool) -> Self { |
| self.allow_invalid_certificates = allow_invalid_certificates.into(); |
| self |
| } |
| |
| /// Disable certificates provided by the system |
| /// |
| /// By default TLS certificates are validated using [`rustls-platform-verifier`], |
| /// which makes use of the system's trust store, in addition to any certificates |
| /// registered using [`Self::with_root_certificate`]. If disabled, instead [`rustls-webpki`] |
| /// is used with only the certificates registered using [`Self::with_root_certificate`]. |
| /// |
| /// [`rustls-platform-verifier`]: https://crates.io/crates/rustls-platform-verifier |
| /// [`rustls-webpki`]: https://crates.io/crates/rustls-webpki |
| pub fn with_no_system_certificates(mut self, no_certs: bool) -> Self { |
| self.no_system_certificates = no_certs.into(); |
| self |
| } |
| |
| /// Only use HTTP/1 connections (default) |
| /// |
| /// # See Also |
| /// * [`Self::with_http2_only`] if you only want to use HTTP/2 |
| /// * [`Self::with_allow_http2`] if you want to use HTTP/1 or HTTP/2 |
| /// |
| /// <div class="warning"> |
| /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104) |
| /// </div> |
| pub fn with_http1_only(mut self) -> Self { |
| self.http2_only = false.into(); |
| self.http1_only = true.into(); |
| self |
| } |
| |
| /// Only use HTTP/2 connections |
| /// |
| /// # See Also |
| /// * [`Self::with_http1_only`] if you only want to use HTTP/1 |
| /// * [`Self::with_allow_http2`] if you want to use HTTP/1 or HTTP/2 |
| /// |
| /// <div class="warning"> |
| /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104) |
| /// </div> |
| pub fn with_http2_only(mut self) -> Self { |
| self.http1_only = false.into(); |
| self.http2_only = true.into(); |
| self |
| } |
| |
| /// Use HTTP/2 if supported, otherwise use HTTP/1. |
| /// |
| /// # See Also |
| /// * [`Self::with_http1_only`] if you only want to use HTTP/1 |
| /// * [`Self::with_http2_only`] if you only want to use HTTP/2 |
| /// |
| /// <div class="warning"> |
| /// HTTP/2 is not used by default. See details [#104](https://github.com/apache/arrow-rs-object-store/issues/104) |
| /// </div> |
| pub fn with_allow_http2(mut self) -> Self { |
| self.http1_only = false.into(); |
| self.http2_only = false.into(); |
| self |
| } |
| |
| /// Set a proxy URL to use for requests |
| pub fn with_proxy_url(mut self, proxy_url: impl Into<String>) -> Self { |
| self.proxy_url = Some(proxy_url.into()); |
| self |
| } |
| |
| /// Set a trusted proxy CA certificate |
| pub fn with_proxy_ca_certificate(mut self, proxy_ca_certificate: impl Into<String>) -> Self { |
| self.proxy_ca_certificate = Some(proxy_ca_certificate.into()); |
| self |
| } |
| |
| /// Set a list of hosts to exclude from proxy connections |
| pub fn with_proxy_excludes(mut self, proxy_excludes: impl Into<String>) -> Self { |
| self.proxy_excludes = Some(proxy_excludes.into()); |
| self |
| } |
| |
| /// Set timeout for the overall request |
| /// |
| /// The timeout starts from when the request starts connecting until the |
| /// response body has finished. If the request does not complete within the |
| /// timeout, the client returns a timeout error. |
| /// |
| /// Timeout errors are retried, subject to the [`RetryConfig`] |
| /// |
| /// Default is 30 seconds |
| /// |
| /// # See Also |
| /// * [`Self::with_timeout_disabled`] to disable the timeout |
| /// * [`Self::with_connect_timeout`] to set a timeout for the connect phase |
| /// |
| /// [`RetryConfig`]: crate::RetryConfig |
| pub fn with_timeout(mut self, timeout: Duration) -> Self { |
| self.timeout = Some(ConfigValue::Parsed(timeout)); |
| self |
| } |
| |
| /// Disables the request timeout |
| /// |
| /// # See Also |
| /// * [`Self::with_timeout`] |
| pub fn with_timeout_disabled(mut self) -> Self { |
| self.timeout = None; |
| self |
| } |
| |
| /// Set a timeout for only the connect phase of a Client |
| /// |
| /// This is the time allowed for the client to establish a connection |
| /// and if the connection is not established within this time, |
| /// the client returns a timeout error. |
| /// |
| /// Timeout errors are retried, subject to the [`RetryConfig`] |
| /// |
| /// Default is 5 seconds |
| /// |
| /// # See Also |
| /// * [`Self::with_timeout`] to set a timeout for the overall request |
| /// * [`Self::with_connect_timeout_disabled`] to disable the connect timeout |
| /// |
| /// [`RetryConfig`]: crate::RetryConfig |
| pub fn with_connect_timeout(mut self, timeout: Duration) -> Self { |
| self.connect_timeout = Some(ConfigValue::Parsed(timeout)); |
| self |
| } |
| |
| /// Disables the connection timeout |
| /// |
| /// # See Also |
| /// * [`Self::with_connect_timeout`] |
| pub fn with_connect_timeout_disabled(mut self) -> Self { |
| self.connect_timeout = None; |
| self |
| } |
| |
| /// Set a read timeout |
| /// |
| /// The timeout applies to each read operation, and resets after a |
| /// successful read. This is useful for detecting stalled connections |
| /// when the size of the response is not known beforehand. |
| /// |
| /// Timeout errors are retried, subject to the [`RetryConfig`] |
| /// |
| /// Default is disabled (no read timeout) |
| /// |
| /// # See Also |
| /// * [`Self::with_read_timeout_disabled`] to disable the read timeout |
| /// * [`Self::with_timeout`] to set a timeout for the overall request |
| /// * [`Self::with_connect_timeout`] to set a timeout for the connect phase |
| /// |
| /// [`RetryConfig`]: crate::RetryConfig |
| pub fn with_read_timeout(mut self, timeout: Duration) -> Self { |
| self.read_timeout = Some(ConfigValue::Parsed(timeout)); |
| self |
| } |
| |
| /// Disables the read timeout |
| /// |
| /// # See Also |
| /// * [`Self::with_read_timeout`] |
| pub fn with_read_timeout_disabled(mut self) -> Self { |
| self.read_timeout = None; |
| self |
| } |
| |
| /// Set the pool max idle timeout |
| /// |
| /// This is the length of time an idle connection will be kept alive |
| /// |
| /// Default is 90 seconds enforced by reqwest |
| pub fn with_pool_idle_timeout(mut self, timeout: Duration) -> Self { |
| self.pool_idle_timeout = Some(ConfigValue::Parsed(timeout)); |
| self |
| } |
| |
| /// Set the maximum number of idle connections per host |
| /// |
| /// Default is no limit enforced by reqwest |
| pub fn with_pool_max_idle_per_host(mut self, max: usize) -> Self { |
| self.pool_max_idle_per_host = Some(max.into()); |
| self |
| } |
| |
| /// Sets an interval for HTTP/2 Ping frames should be sent to keep a connection alive. |
| /// |
| /// Default is disabled enforced by reqwest |
| pub fn with_http2_keep_alive_interval(mut self, interval: Duration) -> Self { |
| self.http2_keep_alive_interval = Some(ConfigValue::Parsed(interval)); |
| self |
| } |
| |
| /// Sets a timeout for receiving an acknowledgement of the keep-alive ping. |
| /// |
| /// If the ping is not acknowledged within the timeout, the connection will be closed. |
| /// Does nothing if `http2_keep_alive_interval` is disabled. |
| /// |
| /// Default is disabled enforced by reqwest |
| pub fn with_http2_keep_alive_timeout(mut self, interval: Duration) -> Self { |
| self.http2_keep_alive_timeout = Some(ConfigValue::Parsed(interval)); |
| self |
| } |
| |
| /// Enable HTTP/2 keep alive pings for idle connections |
| /// |
| /// If disabled, keep-alive pings are only sent while there are open request/response |
| /// streams. If enabled, pings are also sent when no streams are active |
| /// |
| /// Default is disabled enforced by reqwest |
| pub fn with_http2_keep_alive_while_idle(mut self) -> Self { |
| self.http2_keep_alive_while_idle = true.into(); |
| self |
| } |
| |
| /// Sets the maximum frame size to use for HTTP/2. |
| /// |
| /// Default is currently 16,384 but may change internally to optimize for common uses. |
| pub fn with_http2_max_frame_size(mut self, sz: u32) -> Self { |
| self.http2_max_frame_size = Some(ConfigValue::Parsed(sz)); |
| self |
| } |
| |
| /// Get the default headers defined through `ClientOptions::with_default_headers` |
| pub fn get_default_headers(&self) -> Option<&HeaderMap> { |
| self.default_headers.as_ref() |
| } |
| |
| /// Get the mime type for the file in `path` to be uploaded |
| /// |
| /// Gets the file extension from `path`, and returns the |
| /// mime type if it was defined initially through |
| /// `ClientOptions::with_content_type_for_suffix` |
| /// |
| /// Otherwise, returns the default mime type if it was defined |
| /// earlier through `ClientOptions::with_default_content_type` |
| pub fn get_content_type(&self, path: &Path) -> Option<&str> { |
| match path.extension() { |
| Some(extension) => match self.content_type_map.get(extension) { |
| Some(ct) => Some(ct.as_str()), |
| None => self.default_content_type.as_deref(), |
| }, |
| None => self.default_content_type.as_deref(), |
| } |
| } |
| |
| /// Returns a copy of this [`ClientOptions`] with overrides necessary for metadata endpoint access |
| /// |
| /// In particular: |
| /// * Allows HTTP as metadata endpoints do not use TLS |
| /// * Configures a low connection timeout to provide quick feedback if not present |
| #[cfg(any(feature = "aws-base", feature = "gcp-base", feature = "azure-base"))] |
| pub(crate) fn metadata_options(&self) -> Self { |
| self.clone() |
| .with_allow_http(true) |
| .with_connect_timeout(Duration::from_secs(1)) |
| } |
| |
| #[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))] |
| pub(crate) fn client(&self) -> Result<reqwest::Client> { |
| let mut builder = reqwest::ClientBuilder::new(); |
| |
| match &self.user_agent { |
| Some(user_agent) => builder = builder.user_agent(user_agent.get()?), |
| None => builder = builder.user_agent(DEFAULT_USER_AGENT), |
| } |
| |
| if let Some(headers) = &self.default_headers { |
| builder = builder.default_headers(headers.clone()) |
| } |
| |
| if let Some(proxy) = &self.proxy_url { |
| let mut proxy = Proxy::all(proxy).map_err(map_client_error)?; |
| |
| if let Some(certificate) = &self.proxy_ca_certificate { |
| let certificate = reqwest::tls::Certificate::from_pem(certificate.as_bytes()) |
| .map_err(map_client_error)?; |
| |
| builder = builder.tls_certs_merge(std::iter::once(certificate)); |
| } |
| |
| if let Some(proxy_excludes) = &self.proxy_excludes { |
| let no_proxy = NoProxy::from_string(proxy_excludes); |
| |
| proxy = proxy.no_proxy(no_proxy); |
| } |
| |
| builder = builder.proxy(proxy); |
| } |
| |
| let certs = self |
| .root_certificates |
| .iter() |
| .map(|certificate| certificate.0.clone()); |
| |
| if self.no_system_certificates.get()? { |
| builder = builder.tls_certs_only(certs); |
| } else { |
| builder = builder.tls_certs_merge(certs); |
| } |
| |
| if let Some(timeout) = &self.timeout { |
| builder = builder.timeout(timeout.get()?) |
| } |
| |
| if let Some(timeout) = &self.connect_timeout { |
| builder = builder.connect_timeout(timeout.get()?) |
| } |
| |
| if let Some(timeout) = &self.read_timeout { |
| builder = builder.read_timeout(timeout.get()?) |
| } |
| |
| if let Some(timeout) = &self.pool_idle_timeout { |
| builder = builder.pool_idle_timeout(timeout.get()?) |
| } |
| |
| if let Some(max) = &self.pool_max_idle_per_host { |
| builder = builder.pool_max_idle_per_host(max.get()?) |
| } |
| |
| if let Some(interval) = &self.http2_keep_alive_interval { |
| builder = builder.http2_keep_alive_interval(interval.get()?) |
| } |
| |
| if let Some(interval) = &self.http2_keep_alive_timeout { |
| builder = builder.http2_keep_alive_timeout(interval.get()?) |
| } |
| |
| if self.http2_keep_alive_while_idle.get()? { |
| builder = builder.http2_keep_alive_while_idle(true) |
| } |
| |
| if let Some(sz) = &self.http2_max_frame_size { |
| builder = builder.http2_max_frame_size(Some(sz.get()?)) |
| } |
| |
| if self.http1_only.get()? { |
| builder = builder.http1_only() |
| } |
| |
| if self.http2_only.get()? { |
| builder = builder.http2_prior_knowledge() |
| } |
| |
| if self.allow_invalid_certificates.get()? { |
| builder = builder.danger_accept_invalid_certs(true) |
| } |
| |
| // Explicitly disable compression, since it may be automatically enabled |
| // when certain reqwest features are enabled. Compression interferes |
| // with the `Content-Length` header, which is used to determine the |
| // size of objects. |
| builder = builder.no_gzip().no_brotli().no_zstd().no_deflate(); |
| |
| if self.randomize_addresses.get()? { |
| builder = builder.dns_resolver(Arc::new(dns::ShuffleResolver)); |
| } |
| |
| builder |
| .https_only(!self.allow_http.get()?) |
| .build() |
| .map_err(map_client_error) |
| } |
| |
| #[cfg(all(feature = "reqwest", target_arch = "wasm32", target_os = "unknown"))] |
| pub(crate) fn client(&self) -> Result<reqwest::Client> { |
| let mut builder = reqwest::ClientBuilder::new(); |
| |
| match &self.user_agent { |
| Some(user_agent) => builder = builder.user_agent(user_agent.get()?), |
| None => builder = builder.user_agent(DEFAULT_USER_AGENT), |
| } |
| |
| if let Some(headers) = &self.default_headers { |
| builder = builder.default_headers(headers.clone()) |
| } |
| |
| builder.build().map_err(map_client_error) |
| } |
| } |
| |
| pub(crate) trait GetOptionsExt { |
| fn with_get_options(self, options: GetOptions) -> Self; |
| } |
| |
| impl GetOptionsExt for HttpRequestBuilder { |
| fn with_get_options(mut self, options: GetOptions) -> Self { |
| use hyper::header::*; |
| |
| let GetOptions { |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| range, |
| version: _, |
| head: _, |
| extensions, |
| } = options; |
| |
| if let Some(range) = range { |
| self = self.header(RANGE, range.to_string()); |
| } |
| |
| if let Some(tag) = if_match { |
| self = self.header(IF_MATCH, tag); |
| } |
| |
| if let Some(tag) = if_none_match { |
| self = self.header(IF_NONE_MATCH, tag); |
| } |
| |
| const DATE_FORMAT: &str = "%a, %d %b %Y %H:%M:%S GMT"; |
| if let Some(date) = if_unmodified_since { |
| self = self.header(IF_UNMODIFIED_SINCE, date.format(DATE_FORMAT).to_string()); |
| } |
| |
| if let Some(date) = if_modified_since { |
| self = self.header(IF_MODIFIED_SINCE, date.format(DATE_FORMAT).to_string()); |
| } |
| |
| self = self.extensions(extensions); |
| |
| self |
| } |
| } |
| |
| /// Provides credentials for use when signing requests |
| #[async_trait] |
| pub trait CredentialProvider: std::fmt::Debug + Send + Sync { |
| /// The type of credential returned by this provider |
| type Credential; |
| |
| /// Return a credential |
| async fn get_credential(&self) -> Result<Arc<Self::Credential>>; |
| } |
| |
| /// A static set of credentials |
| #[derive(Debug)] |
| pub struct StaticCredentialProvider<T> { |
| credential: Arc<T>, |
| } |
| |
| impl<T> StaticCredentialProvider<T> { |
| /// A [`CredentialProvider`] for a static credential of type `T` |
| pub fn new(credential: T) -> Self { |
| Self { |
| credential: Arc::new(credential), |
| } |
| } |
| } |
| |
| #[async_trait] |
| impl<T> CredentialProvider for StaticCredentialProvider<T> |
| where |
| T: std::fmt::Debug + Send + Sync, |
| { |
| type Credential = T; |
| |
| async fn get_credential(&self) -> Result<Arc<T>> { |
| Ok(Arc::clone(&self.credential)) |
| } |
| } |
| |
| #[cfg(any(feature = "aws-base", feature = "azure-base", feature = "gcp-base"))] |
| mod cloud { |
| use super::*; |
| use crate::RetryConfig; |
| use crate::client::token::{TemporaryToken, TokenCache}; |
| |
| /// A [`CredentialProvider`] that uses [`HttpClient`] to fetch temporary tokens |
| #[derive(Debug)] |
| pub(crate) struct TokenCredentialProvider<T: TokenProvider> { |
| inner: T, |
| client: HttpClient, |
| retry: RetryConfig, |
| cache: TokenCache<Arc<T::Credential>>, |
| } |
| |
| impl<T: TokenProvider> TokenCredentialProvider<T> { |
| pub(crate) fn new(inner: T, client: HttpClient, retry: RetryConfig) -> Self { |
| Self { |
| inner, |
| client, |
| retry, |
| cache: Default::default(), |
| } |
| } |
| |
| /// Override the minimum remaining TTL for a cached token to be used |
| #[cfg(any(feature = "aws-base", feature = "gcp-base"))] |
| pub(crate) fn with_min_ttl(mut self, min_ttl: Duration) -> Self { |
| self.cache = self.cache.with_min_ttl(min_ttl); |
| self |
| } |
| } |
| |
| #[async_trait] |
| impl<T: TokenProvider> CredentialProvider for TokenCredentialProvider<T> { |
| type Credential = T::Credential; |
| |
| async fn get_credential(&self) -> Result<Arc<Self::Credential>> { |
| self.cache |
| .get_or_insert_with(|| self.inner.fetch_token(&self.client, &self.retry)) |
| .await |
| } |
| } |
| |
| #[async_trait] |
| pub(crate) trait TokenProvider: std::fmt::Debug + Send + Sync { |
| type Credential: std::fmt::Debug + Send + Sync; |
| |
| async fn fetch_token( |
| &self, |
| client: &HttpClient, |
| retry: &RetryConfig, |
| ) -> Result<TemporaryToken<Arc<Self::Credential>>>; |
| } |
| } |
| |
| use crate::client::builder::HttpRequestBuilder; |
| #[cfg(any(feature = "aws-base", feature = "azure-base", feature = "gcp-base"))] |
| pub(crate) use cloud::*; |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| use std::collections::HashMap; |
| |
| #[test] |
| fn client_test_config_from_map() { |
| let allow_http = "true".to_string(); |
| let allow_invalid_certificates = "false".to_string(); |
| let connect_timeout = "90 seconds".to_string(); |
| let default_content_type = "object_store:fake_default_content_type".to_string(); |
| let http1_only = "true".to_string(); |
| let http2_only = "false".to_string(); |
| let http2_keep_alive_interval = "90 seconds".to_string(); |
| let http2_keep_alive_timeout = "91 seconds".to_string(); |
| let http2_keep_alive_while_idle = "92 seconds".to_string(); |
| let http2_max_frame_size = "1337".to_string(); |
| let no_system_certificates = "true".to_string(); |
| let pool_idle_timeout = "93 seconds".to_string(); |
| let pool_max_idle_per_host = "94".to_string(); |
| let proxy_url = "https://fake_proxy_url".to_string(); |
| let read_timeout = "45 seconds".to_string(); |
| let timeout = "95 seconds".to_string(); |
| let user_agent = "object_store:fake_user_agent".to_string(); |
| |
| let options = HashMap::from([ |
| ("allow_http", allow_http.clone()), |
| ( |
| "allow_invalid_certificates", |
| allow_invalid_certificates.clone(), |
| ), |
| ("connect_timeout", connect_timeout.clone()), |
| ("default_content_type", default_content_type.clone()), |
| ("http1_only", http1_only.clone()), |
| ("http2_only", http2_only.clone()), |
| ( |
| "http2_keep_alive_interval", |
| http2_keep_alive_interval.clone(), |
| ), |
| ("http2_keep_alive_timeout", http2_keep_alive_timeout.clone()), |
| ( |
| "http2_keep_alive_while_idle", |
| http2_keep_alive_while_idle.clone(), |
| ), |
| ("http2_max_frame_size", http2_max_frame_size.clone()), |
| ( |
| "disable_system_certificates", |
| no_system_certificates.clone(), |
| ), |
| ("pool_idle_timeout", pool_idle_timeout.clone()), |
| ("pool_max_idle_per_host", pool_max_idle_per_host.clone()), |
| ("proxy_url", proxy_url.clone()), |
| ("read_timeout", read_timeout.clone()), |
| ("timeout", timeout.clone()), |
| ("user_agent", user_agent.clone()), |
| ]); |
| |
| let builder = options |
| .into_iter() |
| .fold(ClientOptions::new(), |builder, (key, value)| { |
| builder.with_config(key.parse().unwrap(), value) |
| }); |
| |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::AllowHttp) |
| .unwrap(), |
| allow_http |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::AllowInvalidCertificates) |
| .unwrap(), |
| allow_invalid_certificates |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::ConnectTimeout) |
| .unwrap(), |
| connect_timeout |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::DefaultContentType) |
| .unwrap(), |
| default_content_type |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http1Only) |
| .unwrap(), |
| http1_only |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http2Only) |
| .unwrap(), |
| http2_only |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http2KeepAliveInterval) |
| .unwrap(), |
| http2_keep_alive_interval |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http2KeepAliveTimeout) |
| .unwrap(), |
| http2_keep_alive_timeout |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http2KeepAliveWhileIdle) |
| .unwrap(), |
| http2_keep_alive_while_idle |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::Http2MaxFrameSize) |
| .unwrap(), |
| http2_max_frame_size |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::NoSystemCertificates) |
| .unwrap(), |
| no_system_certificates |
| ); |
| |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::PoolIdleTimeout) |
| .unwrap(), |
| pool_idle_timeout |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::PoolMaxIdlePerHost) |
| .unwrap(), |
| pool_max_idle_per_host |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::ProxyUrl) |
| .unwrap(), |
| proxy_url |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::ReadTimeout) |
| .unwrap(), |
| read_timeout |
| ); |
| assert_eq!( |
| builder.get_config_value(&ClientConfigKey::Timeout).unwrap(), |
| timeout |
| ); |
| assert_eq!( |
| builder |
| .get_config_value(&ClientConfigKey::UserAgent) |
| .unwrap(), |
| user_agent |
| ); |
| } |
| } |