Convert authenticated client to reqwest middleware (#488)

While initially I wanted to create a number of traits for reqwest types,
as it turns out, there are better solutions.

`reqwest_middleware` provides a flexible way to insert middleware that
is activated either when the request is being built or just before it's
sent. `AuthenticatedClient` is converted to a middleware type that
handles auth.

This enables authentication with auth in rattler itself as well as in
`async_http_range_reader` through the same interface.

One consequence is removal of the blocking client and features, as
blocking comms are not supported in the middleware crate. I would argue
that in most places where blocking communication is preferred -- e. g.
simple programs performing a limited number of network requests -- it's
often enough to just spawn a tokio runtime to wait on the future,
because compared to network latency the cost will be negligible.
Applications doing intensive network i/o are either async already anyway
or likely should be.

This is not yet ready for merge -- I'm not sure what to do with Python
binding. Technically we could replicate this type using the middleware
client, but it's a bit weird that it's removed from Rust API but stays
in Python API.

I will update the docs once there's a view on how to proceed.

Author: vlad-ivanov-name

crates/async_http_range_reader/Cargo.toml

@@ -14,6 +14,7 @@ itertools = "0.11.0"
 bisection = "0.1.0"
 memmap2 = "0.9.0"
 reqwest = { version = "0.11.22", default-features = false, features = ["stream"] }
+reqwest-middleware = "0.2.4"
 tokio = { version = "1.33.0", default-features = false }
 tokio-stream = { version = "0.1.14", features = ["sync"] }
 tokio-util = "0.7.9"

crates/async_http_range_reader/src/async_http_range_reader.rs

@@ -21,7 +21,7 @@ use futures::{FutureExt, Stream, StreamExt};
 use http_content_range::{ContentRange, ContentRangeBytes};
 use memmap2::MmapMut;
 use reqwest::header::HeaderMap;
-use reqwest::{Client, Response, Url};
+use reqwest::{Response, Url};
 use std::{
     io::{self, ErrorKind, SeekFrom},
     ops::Range,
@@ -68,7 +68,7 @@ pub use crate::async_http_range_reader_error::AsyncHttpRangeReaderError;
 ///     if response.status() == reqwest::StatusCode::NOT_MODIFIED {
 ///         Ok(None)
 ///     } else {
-///         let reader = AsyncHttpRangeReader::from_head_response(client, response).await?;
+///         let reader = AsyncHttpRangeReader::from_head_response(client.into(), response).await?;
 ///         Ok(Some(reader))
 ///     }
 /// }
@@ -127,10 +127,16 @@ pub enum CheckSupportMethod {
     Head,
 }
 
+fn error_for_status(response: reqwest::Response) -> reqwest_middleware::Result<Response> {
+    response
+        .error_for_status()
+        .map_err(reqwest_middleware::Error::Reqwest)
+}
+
 impl AsyncHttpRangeReader {
     /// Construct a new `AsyncHttpRangeReader`.
     pub async fn new(
-        client: reqwest::Client,
+        client: reqwest_middleware::ClientWithMiddleware,
         url: reqwest::Url,
         check_method: CheckSupportMethod,
     ) -> Result<(Self, HeaderMap), AsyncHttpRangeReaderError> {
@@ -162,7 +168,7 @@ impl AsyncHttpRangeReader {
     /// requests. This will return a number of bytes from the end of the stream. Use the
     /// `initial_chunk_size` parameter to define how many bytes should be requested from the end.
     pub async fn initial_tail_request(
-        client: reqwest::Client,
+        client: reqwest_middleware::ClientWithMiddleware,
         url: reqwest::Url,
         initial_chunk_size: u64,
         extra_headers: HeaderMap,
@@ -176,7 +182,7 @@ impl AsyncHttpRangeReader {
             .headers(extra_headers)
             .send()
             .await
-            .and_then(Response::error_for_status)
+            .and_then(error_for_status)
             .map_err(Arc::new)
             .map_err(AsyncHttpRangeReaderError::HttpError)?;
         Ok(tail_response)
@@ -185,7 +191,7 @@ impl AsyncHttpRangeReader {
     /// Initialize the reader from [`AsyncHttpRangeReader::initial_tail_request`] (or a user
     /// provided response that also has a range of bytes from the end as body)
     pub async fn from_tail_response(
-        client: reqwest::Client,
+        client: reqwest_middleware::ClientWithMiddleware,
         tail_request_response: Response,
     ) -> Result<Self, AsyncHttpRangeReaderError> {
         // Get the size of the file from this initial request
@@ -260,7 +266,7 @@ impl AsyncHttpRangeReader {
     /// Send an initial range request to determine if the remote accepts range
     /// requests and get the content length
     pub async fn initial_head_request(
-        client: reqwest::Client,
+        client: reqwest_middleware::ClientWithMiddleware,
         url: reqwest::Url,
         extra_headers: HeaderMap,
     ) -> Result<Response, AsyncHttpRangeReaderError> {
@@ -270,7 +276,7 @@ impl AsyncHttpRangeReader {
             .headers(extra_headers)
             .send()
             .await
-            .and_then(Response::error_for_status)
+            .and_then(error_for_status)
             .map_err(Arc::new)
             .map_err(AsyncHttpRangeReaderError::HttpError)?;
         Ok(head_response)
@@ -279,7 +285,7 @@ impl AsyncHttpRangeReader {
     /// Initialize the reader from [`AsyncHttpRangeReader::initial_head_request`] (or a user
     /// provided response the)
     pub async fn from_head_response(
-        client: reqwest::Client,
+        client: reqwest_middleware::ClientWithMiddleware,
         head_response: Response,
     ) -> Result<Self, AsyncHttpRangeReaderError> {
         // Are range requests supported?
@@ -387,7 +393,7 @@ impl AsyncHttpRangeReader {
 /// become available.
 #[tracing::instrument(name = "fetch_ranges", skip_all, fields(url))]
 async fn run_streamer(
-    client: Client,
+    client: reqwest_middleware::ClientWithMiddleware,
     url: Url,
     initial_tail_response: Option<(Response, u64)>,
     mut memory_map: MmapMut,
@@ -447,7 +453,7 @@ async fn run_streamer(
                 .send()
                 .instrument(span)
                 .await
-                .and_then(Response::error_for_status)
+                .and_then(error_for_status)
                 .map_err(|e| std::io::Error::new(ErrorKind::Other, e))
             {
                 Err(e) => {
@@ -649,7 +655,7 @@ mod test {
 
         // Construct an AsyncRangeReader
         let (mut range, _) = AsyncHttpRangeReader::new(
-            Client::new(),
+            Client::new().into(),
             server.url().join("andes-1.8.3-pyhd8ed1ab_0.conda").unwrap(),
             check_method,
         )
@@ -743,7 +749,7 @@ mod test {
 
         // Construct an AsyncRangeReader
         let (mut range, _) = AsyncHttpRangeReader::new(
-            Client::new(),
+            Client::new().into(),
             server.url().join("andes-1.8.3-pyhd8ed1ab_0.conda").unwrap(),
             check_method,
         )
@@ -784,7 +790,7 @@ mod test {
     async fn test_not_found() {
         let server = StaticDirectoryServer::new(Path::new(env!("CARGO_MANIFEST_DIR")));
         let err = AsyncHttpRangeReader::new(
-            Client::new(),
+            Client::new().into(),
             server.url().join("not-found").unwrap(),
             CheckSupportMethod::Head,
         )

crates/async_http_range_reader/src/async_http_range_reader_error.rs

@@ -10,11 +10,11 @@ pub enum AsyncHttpRangeReaderError {
 
     /// Other HTTP error
     #[error(transparent)]
-    HttpError(#[from] Arc<reqwest::Error>),
+    HttpError(#[from] Arc<reqwest_middleware::Error>),
 
     /// An error occurred during transport
     #[error("an error occurred during transport: {0}")]
-    TransportError(#[source] Arc<reqwest::Error>),
+    TransportError(#[source] Arc<reqwest_middleware::Error>),
 
     /// An IO error occurred
     #[error("io error occurred: {0}")]
@@ -39,8 +39,14 @@ impl From<std::io::Error> for AsyncHttpRangeReaderError {
     }
 }
 
+impl From<reqwest_middleware::Error> for AsyncHttpRangeReaderError {
+    fn from(err: reqwest_middleware::Error) -> Self {
+        AsyncHttpRangeReaderError::TransportError(Arc::new(err))
+    }
+}
+
 impl From<reqwest::Error> for AsyncHttpRangeReaderError {
     fn from(err: reqwest::Error) -> Self {
-        AsyncHttpRangeReaderError::TransportError(Arc::new(err))
+        AsyncHttpRangeReaderError::TransportError(Arc::new(err.into()))
     }
 }

crates/rattler-bin/Cargo.toml

@@ -36,6 +36,7 @@ rattler_repodata_gateway = { version = "0.16.2", path = "../rattler_repodata_gat
 rattler_solve = { version = "0.16.2", path = "../rattler_solve", features = ["resolvo", "libsolv_c"] }
 rattler_virtual_packages = { version = "0.16.2", path = "../rattler_virtual_packages" }
 reqwest = { version = "0.11.22", default-features = false }
+reqwest-middleware = { version = "0.2.4" }
 tokio = { version = "1.32.0", features = ["rt-multi-thread", "macros"] }
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 

crates/rattler-bin/src/commands/create.rs

@@ -15,14 +15,15 @@ use rattler_conda_types::{
     PrefixRecord, RepoDataRecord, Version,
 };
 use rattler_networking::{
-    retry_policies::default_retry_policy, AuthenticatedClient, AuthenticationStorage,
+    retry_policies::default_retry_policy, AuthenticationMiddleware, AuthenticationStorage,
 };
 use rattler_repodata_gateway::fetch::{
     CacheResult, DownloadProgress, FetchRepoDataError, FetchRepoDataOptions,
 };
 use rattler_repodata_gateway::sparse::SparseRepoData;
 use rattler_solve::{libsolv_c, resolvo, SolverImpl, SolverTask};
 use reqwest::Client;
+use std::sync::Arc;
 use std::{
     borrow::Cow,
     env,
@@ -117,8 +118,12 @@ pub async fn create(opt: Opt) -> anyhow::Result<()> {
         .expect("failed to create client");
 
     let authentication_storage = AuthenticationStorage::default();
+    let download_client = reqwest_middleware::ClientBuilder::new(download_client)
+        .with_arc(Arc::new(AuthenticationMiddleware::new(
+            authentication_storage,
+        )))
+        .build();
 
-    let download_client = AuthenticatedClient::from_client(download_client, authentication_storage);
     let multi_progress = global_multi_progress();
 
     let repodata_cache_path = cache_dir.join("repodata");
@@ -297,7 +302,7 @@ async fn execute_transaction(
     transaction: Transaction<PrefixRecord, RepoDataRecord>,
     target_prefix: PathBuf,
     cache_dir: PathBuf,
-    download_client: AuthenticatedClient,
+    download_client: reqwest_middleware::ClientWithMiddleware,
 ) -> anyhow::Result<()> {
     // Open the package cache
     let package_cache = PackageCache::new(cache_dir.join("pkgs"));
@@ -385,7 +390,7 @@ async fn execute_transaction(
 #[allow(clippy::too_many_arguments)]
 async fn execute_operation(
     target_prefix: &Path,
-    download_client: AuthenticatedClient,
+    download_client: reqwest_middleware::ClientWithMiddleware,
     package_cache: &PackageCache,
     install_driver: &InstallDriver,
     download_pb: Option<&ProgressBar>,
@@ -551,7 +556,7 @@ async fn fetch_repo_data_records_with_progress(
     channel: Channel,
     platform: Platform,
     repodata_cache: &Path,
-    client: AuthenticatedClient,
+    client: reqwest_middleware::ClientWithMiddleware,
     multi_progress: indicatif::MultiProgress,
 ) -> Result<Option<SparseRepoData>, anyhow::Error> {
     // Create a progress bar

crates/rattler/Cargo.toml

@@ -36,10 +36,11 @@ pin-project-lite = "0.2.13"
 rattler_conda_types = { version = "0.16.2", path = "../rattler_conda_types" }
 rattler_digest = { version = "0.16.2", path = "../rattler_digest" }
 rattler_networking = { version = "0.16.2", path = "../rattler_networking", default-features = false }
-rattler_package_streaming = { version = "0.16.2", path = "../rattler_package_streaming", features = ["reqwest", "tokio"], default-features = false }
+rattler_package_streaming = { version = "0.16.2", path = "../rattler_package_streaming", features = ["reqwest"], default-features = false }
 reflink-copy = "0.1.14"
 regex = "1.9.6"
 reqwest = { version = "0.11.22", default-features = false, features = ["stream", "json", "gzip"] }
+reqwest-middleware = "0.2.4"
 serde = { version = "1.0.188", features = ["derive"] }
 serde_json = { version = "1.0.107", features = ["raw_value"] }
 serde_with = "3.3.0"

crates/rattler/src/install/clobber_registry.rs

@@ -278,7 +278,7 @@ mod tests {
         package::IndexJson, PackageRecord, Platform, PrefixRecord, RepoDataRecord,
     };
     use rattler_digest::{Md5, Sha256};
-    use rattler_networking::{retry_policies::default_retry_policy, AuthenticatedClient};
+    use rattler_networking::retry_policies::default_retry_policy;
     use rattler_package_streaming::seek::read_package_file;
     use transaction::{Transaction, TransactionOperation};
 
@@ -368,7 +368,7 @@ mod tests {
 
     async fn execute_operation(
         target_prefix: &Path,
-        download_client: &AuthenticatedClient,
+        download_client: &reqwest_middleware::ClientWithMiddleware,
         package_cache: &PackageCache,
         install_driver: &InstallDriver,
         op: TransactionOperation<PrefixRecord, RepoDataRecord>,
@@ -416,7 +416,7 @@ mod tests {
     async fn execute_transaction(
         transaction: Transaction<PrefixRecord, RepoDataRecord>,
         target_prefix: &Path,
-        download_client: &AuthenticatedClient,
+        download_client: &reqwest_middleware::ClientWithMiddleware,
         package_cache: &PackageCache,
         install_driver: &InstallDriver,
         install_options: &InstallOptions,
@@ -515,7 +515,7 @@ mod tests {
         execute_transaction(
             transaction,
             target_prefix.path(),
-            &AuthenticatedClient::default(),
+            &reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new()),
             &cache,
             &InstallDriver::default(),
             &InstallOptions::default(),
@@ -572,7 +572,7 @@ mod tests {
         execute_transaction(
             transaction,
             target_prefix.path(),
-            &AuthenticatedClient::default(),
+            &reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new()),
             &cache,
             &install_driver,
             &InstallOptions::default(),
@@ -637,7 +637,7 @@ mod tests {
             execute_transaction(
                 transaction,
                 target_prefix.path(),
-                &AuthenticatedClient::default(),
+                &reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new()),
                 &cache,
                 &InstallDriver::default(),
                 &InstallOptions::default(),
@@ -684,7 +684,7 @@ mod tests {
             execute_transaction(
                 transaction,
                 target_prefix.path(),
-                &AuthenticatedClient::default(),
+                &reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new()),
                 &cache,
                 &InstallDriver::default(),
                 &InstallOptions::default(),
@@ -735,7 +735,7 @@ mod tests {
             execute_transaction(
                 transaction,
                 target_prefix.path(),
-                &AuthenticatedClient::default(),
+                &reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new()),
                 &cache,
                 &install_driver,
                 &InstallOptions::default(),

crates/rattler/src/install/mod.rs

@@ -621,7 +621,6 @@ mod test {
     use rattler_conda_types::package::ArchiveIdentifier;
     use rattler_conda_types::{ExplicitEnvironmentSpec, Platform, Version};
     use rattler_lock::LockFile;
-    use rattler_networking::AuthenticatedClient;
 
     use std::env::temp_dir;
     use std::process::Command;
@@ -683,7 +682,7 @@ mod test {
         let package_cache = PackageCache::new(temp_dir().join("rattler").join(cache_name));
 
         // Create an HTTP client we can use to download packages
-        let client = AuthenticatedClient::default();
+        let client = reqwest_middleware::ClientWithMiddleware::from(reqwest::Client::new());
 
         // Specify python version
         let python_version =