WIP: switch to reqwest

Author: sandhose

Cargo.lock

@@ -182,6 +182,12 @@ version = "0.3.0"
 [workspace.dependencies.rand]
 version = "0.8.5"
 
+# High-level HTTP client
+[workspace.dependencies.reqwest]
+version = "0.12.8"
+default-features = false
+features = ["http2", "rustls-tls-manual-roots", "charset", "json"]
+
 # TLS stack
 [workspace.dependencies.rustls]
 version = "0.23.15"

Cargo.toml

@@ -41,6 +41,9 @@ aide.workspace = true
 async-graphql.workspace = true
 schemars.workspace = true
 
+# HTTP client
+reqwest.workspace = true
+
 # Emails
 lettre.workspace = true
 

crates/handlers/Cargo.toml

@@ -312,6 +312,7 @@ where
     MetadataCache: FromRef<S>,
     SiteConfig: FromRef<S>,
     Limiter: FromRef<S>,
+    reqwest::Client: FromRef<S>,
     BoxHomeserverConnection: FromRef<S>,
     BoxClock: FromRequestParts<S>,
     BoxRng: FromRequestParts<S>,

crates/handlers/src/lib.rs

@@ -109,6 +109,7 @@ pub(crate) struct TestState {
     pub limiter: Limiter,
     pub clock: Arc<MockClock>,
     pub rng: Arc<Mutex<ChaChaRng>>,
+    pub http_client: reqwest::Client,
 
     #[allow(dead_code)] // It is used, as it will cancel the CancellationToken when dropped
     cancellation_drop_guard: Arc<DropGuard>,
@@ -169,6 +170,8 @@ impl TestState {
         )
         .await?;
 
+        let http_client = mas_http::reqwest_client();
+
         // TODO: add more test keys to the store
         let rsa =
             PrivateKey::load_pem(include_str!("../../keystore/tests/keys/rsa.pkcs1.pem")).unwrap();
@@ -241,6 +244,7 @@ impl TestState {
             limiter,
             clock,
             rng,
+            http_client,
             cancellation_drop_guard: Arc::new(shutdown_token.drop_guard()),
         })
     }
@@ -494,6 +498,12 @@ impl FromRef<TestState> for Limiter {
     }
 }
 
+impl FromRef<TestState> for reqwest::Client {
+    fn from_ref(input: &TestState) -> Self {
+        input.http_client.clone()
+    }
+}
+
 #[async_trait]
 impl FromRequestParts<TestState> for ActivityTracker {
     type Rejection = Infallible;

crates/handlers/src/test_utils.rs

@@ -62,10 +62,10 @@ impl IntoResponse for RouteError {
 pub(crate) async fn get(
     mut rng: BoxRng,
     clock: BoxClock,
-    State(http_client_factory): State<HttpClientFactory>,
     State(metadata_cache): State<MetadataCache>,
     mut repo: BoxRepository,
     State(url_builder): State<UrlBuilder>,
+    State(http_client): State<reqwest::Client>,
     cookie_jar: CookieJar,
     Path(provider_id): Path<Ulid>,
     Query(query): Query<OptionalPostAuthAction>,
@@ -77,12 +77,10 @@ pub(crate) async fn get(
         .filter(UpstreamOAuthProvider::enabled)
         .ok_or(RouteError::ProviderNotFound)?;
 
-    let http_service = http_client_factory.http_service("upstream_oauth2.authorize");
-
     // First, discover the provider
     // This is done lazyly according to provider.discovery_mode and the various
     // endpoint overrides
-    let mut lazy_metadata = LazyProviderInfos::new(&metadata_cache, &provider, &http_service);
+    let mut lazy_metadata = LazyProviderInfos::new(&metadata_cache, &provider, &http_client);
     lazy_metadata.maybe_discover().await?;
 
     let redirect_uri = url_builder.upstream_oauth_callback(provider.id);

crates/handlers/src/upstream_oauth2/authorize.rs

@@ -9,7 +9,6 @@ use std::{collections::HashMap, sync::Arc};
 use mas_data_model::{
     UpstreamOAuthProvider, UpstreamOAuthProviderDiscoveryMode, UpstreamOAuthProviderPkceMode,
 };
-use mas_http::HttpService;
 use mas_iana::oauth::PkceCodeChallengeMethod;
 use mas_oidc_client::error::DiscoveryError;
 use mas_storage::{upstream_oauth2::UpstreamOAuthProviderRepository, RepositoryAccess};
@@ -22,20 +21,20 @@ use url::Url;
 pub struct LazyProviderInfos<'a> {
     cache: &'a MetadataCache,
     provider: &'a UpstreamOAuthProvider,
-    http_service: &'a HttpService,
+    client: &'a reqwest::Client,
     loaded_metadata: Option<Arc<VerifiedProviderMetadata>>,
 }
 
 impl<'a> LazyProviderInfos<'a> {
     pub fn new(
         cache: &'a MetadataCache,
         provider: &'a UpstreamOAuthProvider,
-        http_service: &'a HttpService,
+        client: &'a reqwest::Client,
     ) -> Self {
         Self {
             cache,
             provider,
-            http_service,
+            client,
             loaded_metadata: None,
         }
     }
@@ -64,7 +63,7 @@ impl<'a> LazyProviderInfos<'a> {
 
             let metadata = self
                 .cache
-                .get(self.http_service, &self.provider.issuer, verify)
+                .get(self.client, &self.provider.issuer, verify)
                 .await?;
 
             self.loaded_metadata = Some(metadata);
@@ -155,7 +154,7 @@ impl MetadataCache {
     #[tracing::instrument(name = "metadata_cache.warm_up_and_run", skip_all, err)]
     pub async fn warm_up_and_run<R: RepositoryAccess>(
         &self,
-        http_service: HttpService,
+        client: &reqwest::Client,
         interval: std::time::Duration,
         repository: &mut R,
     ) -> Result<tokio::task::JoinHandle<()>, R::Error> {
@@ -168,32 +167,32 @@ impl MetadataCache {
                 UpstreamOAuthProviderDiscoveryMode::Disabled => continue,
             };
 
-            if let Err(e) = self.fetch(&http_service, &provider.issuer, verify).await {
+            if let Err(e) = self.fetch(client, &provider.issuer, verify).await {
                 tracing::error!(issuer = %provider.issuer, error = &e as &dyn std::error::Error, "Failed to fetch provider metadata");
             }
         }
 
         // Spawn a background task to refresh the cache regularly
         let cache = self.clone();
+        let client = client.clone();
         Ok(tokio::spawn(async move {
             loop {
                 // Re-fetch the known metadata at the given interval
                 tokio::time::sleep(interval).await;
-                cache.refresh_all(&http_service).await;
+                cache.refresh_all(&client).await;
             }
         }))
     }
 
     #[tracing::instrument(name = "metadata_cache.fetch", fields(%issuer), skip_all, err)]
     async fn fetch(
         &self,
-        http_service: &HttpService,
+        client: &reqwest::Client,
         issuer: &str,
         verify: bool,
     ) -> Result<Arc<VerifiedProviderMetadata>, DiscoveryError> {
         if verify {
-            let metadata =
-                mas_oidc_client::requests::discovery::discover(http_service, issuer).await?;
+            let metadata = mas_oidc_client::requests::discovery::discover(client, issuer).await?;
             let metadata = Arc::new(metadata);
 
             self.cache
@@ -204,8 +203,7 @@ impl MetadataCache {
             Ok(metadata)
         } else {
             let metadata =
-                mas_oidc_client::requests::discovery::insecure_discover(http_service, issuer)
-                    .await?;
+                mas_oidc_client::requests::discovery::insecure_discover(client, issuer).await?;
             let metadata = Arc::new(metadata);
 
             self.insecure_cache
@@ -221,7 +219,7 @@ impl MetadataCache {
     #[tracing::instrument(name = "metadata_cache.get", fields(%issuer), skip_all, err)]
     pub async fn get(
         &self,
-        http_service: &HttpService,
+        client: &reqwest::Client,
         issuer: &str,
         verify: bool,
     ) -> Result<Arc<VerifiedProviderMetadata>, DiscoveryError> {
@@ -237,20 +235,20 @@ impl MetadataCache {
         // Drop the cache guard so that we don't deadlock when we try to fetch
         drop(cache);
 
-        let metadata = self.fetch(http_service, issuer, verify).await?;
+        let metadata = self.fetch(client, issuer, verify).await?;
         Ok(metadata)
     }
 
     #[tracing::instrument(name = "metadata_cache.refresh_all", skip_all)]
-    async fn refresh_all(&self, http_service: &HttpService) {
+    async fn refresh_all(&self, client: &reqwest::Client) {
         // Grab all the keys first to avoid locking the cache for too long
         let keys: Vec<String> = {
             let cache = self.cache.read().await;
             cache.keys().cloned().collect()
         };
 
         for issuer in keys {
-            if let Err(e) = self.fetch(http_service, &issuer, true).await {
+            if let Err(e) = self.fetch(client, &issuer, true).await {
                 tracing::error!(issuer = %issuer, error = &e as &dyn std::error::Error, "Failed to refresh provider metadata");
             }
         }
@@ -262,13 +260,14 @@ impl MetadataCache {
         };
 
         for issuer in keys {
-            if let Err(e) = self.fetch(http_service, &issuer, false).await {
+            if let Err(e) = self.fetch(client, &issuer, false).await {
                 tracing::error!(issuer = %issuer, error = &e as &dyn std::error::Error, "Failed to refresh provider metadata");
             }
         }
     }
 }
 
+/* TODO: redo those tests
 #[cfg(test)]
 mod tests {
     #![allow(clippy::too_many_lines)]
@@ -619,3 +618,4 @@ mod tests {
         }
     }
 }
+*/

crates/handlers/src/upstream_oauth2/cache.rs

@@ -134,6 +134,7 @@ pub(crate) async fn get(
     State(url_builder): State<UrlBuilder>,
     State(encrypter): State<Encrypter>,
     State(keystore): State<Keystore>,
+    State(client): State<reqwest::Client>,
     cookie_jar: CookieJar,
     Path(provider_id): Path<Ulid>,
     Query(params): Query<QueryParams>,
@@ -186,12 +187,11 @@ pub(crate) async fn get(
         CodeOrError::Code { code } => code,
     };
 
-    let http_service = http_client_factory.http_service("upstream_oauth2.callback");
-    let mut lazy_metadata = LazyProviderInfos::new(&metadata_cache, &provider, &http_service);
+    let mut lazy_metadata = LazyProviderInfos::new(&metadata_cache, &provider, &client);
 
     // Fetch the JWKS
     let jwks =
-        mas_oidc_client::requests::jose::fetch_jwks(&http_service, lazy_metadata.jwks_uri().await?)
+        mas_oidc_client::requests::jose::fetch_jwks(&client, lazy_metadata.jwks_uri().await?)
             .await?;
 
     // Figure out the client credentials
@@ -222,7 +222,7 @@ pub(crate) async fn get(
 
     let (response, id_token) =
         mas_oidc_client::requests::authorization_code::access_token_with_authorization_code(
-            &http_service,
+            &client,
             client_credentials,
             lazy_metadata.token_endpoint().await?,
             code,

crates/handlers/src/upstream_oauth2/callback.rs

@@ -23,10 +23,12 @@ hyper.workspace = true
 hyper-util.workspace = true
 hyper-rustls = { workspace = true, optional = true }
 opentelemetry.workspace = true
+opentelemetry-http.workspace = true
 opentelemetry-semantic-conventions.workspace = true
 rustls = { workspace = true, optional = true }
 rustls-platform-verifier = { workspace = true, optional = true }
 pin-project-lite = "0.2.14"
+reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_urlencoded = "0.7.1"

crates/http/Cargo.toml

@@ -13,6 +13,7 @@
 mod client;
 mod ext;
 mod layers;
+mod reqwest;
 mod service;
 
 #[cfg(feature = "client")]
@@ -33,6 +34,7 @@ pub use self::{
         json_request::{self, JsonRequest, JsonRequestLayer},
         json_response::{self, JsonResponse, JsonResponseLayer},
     },
+    reqwest::{client as reqwest_client, RequestBuilderExt},
     service::{BoxCloneSyncService, HttpService},
 };