Finish impl

Author: ardatan

Cargo.lock

@@ -49,9 +49,10 @@ itoa = "1.0.15"
 ryu = "1.0.20"
 indexmap = "2.10.0"
 bumpalo = "3.19.0"
-aws-config = "1.8.8"
-aws-sigv4 = "1.3.5"
-aws-credential-types = "1.2.8"
+reqsign-aws-v4 = "2.0.1"
+reqsign-core = "2.0.1"
+reqsign-file-read-tokio = "2.0.1"
+reqsign-http-send-reqwest = "2.0.1"
 
 [dev-dependencies]
 subgraphs = { path = "../../bench/subgraphs" }

lib/executor/Cargo.toml

@@ -0,0 +1,49 @@
+use hive_router_config::aws_sig_v4::AwsSigV4SubgraphConfig;
+use reqsign_aws_v4::{
+    Credential, DefaultCredentialProvider, ProfileCredentialProvider, RequestSigner,
+    StaticCredentialProvider,
+};
+use reqsign_core::{Context, OsEnv, ProvideCredentialChain, Signer};
+use reqsign_file_read_tokio::TokioFileRead;
+use reqsign_http_send_reqwest::ReqwestHttpSend;
+
+pub fn create_awssigv4_signer(config: &AwsSigV4SubgraphConfig) -> Signer<Credential> {
+    let ctx = Context::new()
+        .with_file_read(TokioFileRead)
+        .with_http_send(ReqwestHttpSend::default())
+        .with_env(OsEnv);
+    let mut loader = ProvideCredentialChain::new();
+    match config {
+        AwsSigV4SubgraphConfig::DefaultChain(default_chain) => {
+            loader = loader.push(DefaultCredentialProvider::new());
+            if let Some(profile_name) = &default_chain.profile_name {
+                loader = loader.push(ProfileCredentialProvider::new().with_profile(profile_name));
+            }
+        }
+        AwsSigV4SubgraphConfig::HardCoded(hard_coded) => {
+            let mut provider = StaticCredentialProvider::new(
+                &hard_coded.access_key_id,
+                &hard_coded.secret_access_key,
+            );
+            if let Some(session_token) = &hard_coded.session_token {
+                provider = provider.with_session_token(session_token);
+            }
+            loader = loader.push(provider);
+        }
+    }
+    let service: &str = match config {
+        AwsSigV4SubgraphConfig::DefaultChain(default_chain) => {
+            default_chain.service.as_ref().map_or("s3", |v| v)
+        }
+        AwsSigV4SubgraphConfig::HardCoded(hard_coded) => hard_coded.service_name.as_str(),
+    };
+    let region: &str = match config {
+        AwsSigV4SubgraphConfig::DefaultChain(default_chain) => {
+            default_chain.region.as_ref().map_or("us-east-1", |v| v)
+        }
+        AwsSigV4SubgraphConfig::HardCoded(hard_coded) => hard_coded.region.as_str(),
+    };
+    let builder = RequestSigner::new(service, region);
+
+    Signer::new(ctx, loader, builder)
+}

lib/executor/src/execution/awssigv4.rs

@@ -1,3 +1,4 @@
+pub mod awssigv4;
 pub mod client_request_details;
 pub mod error;
 pub mod jwt_forward;

lib/executor/src/execution/mod.rs

@@ -77,7 +77,7 @@ impl SubgraphExecutorError {
             SubgraphExecutorError::RequestFailure(_, _) => "SUBGRAPH_REQUEST_FAILURE",
             SubgraphExecutorError::VariablesSerializationFailure(_, _) => {
                 "SUBGRAPH_VARIABLES_SERIALIZATION_FAILURE"
-            },
+            }
             SubgraphExecutorError::AwsSigV4SigningFailure(_, _) => {
                 "SUBGRAPH_AWS_SIGV4_SIGNING_FAILURE"
             }

lib/executor/src/executors/error.rs

@@ -3,11 +3,10 @@ use std::sync::Arc;
 use crate::executors::common::HttpExecutionResponse;
 use crate::executors::dedupe::{request_fingerprint, ABuildHasher, SharedResponse};
 use async_trait::async_trait;
-use aws_sigv4::http_request::{
-    sign, SignableBody, SignableRequest, SigningParams,
-};
 use dashmap::DashMap;
 use hive_router_config::HiveRouterConfig;
+use reqsign_aws_v4::Credential;
+use reqsign_core::Signer;
 use tokio::sync::OnceCell;
 
 use bytes::{BufMut, Bytes, BytesMut};
@@ -39,7 +38,7 @@ pub struct HTTPSubgraphExecutor {
     pub semaphore: Arc<Semaphore>,
     pub config: Arc<HiveRouterConfig>,
     pub in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
-    pub aws_signing_params: Option<Arc<SigningParams<'static>>>,
+    pub aws_sigv4_signer: Option<Signer<Credential>>,
 }
 
 const FIRST_VARIABLE_STR: &[u8] = b",\"variables\":{";
@@ -55,7 +54,7 @@ impl HTTPSubgraphExecutor {
         semaphore: Arc<Semaphore>,
         config: Arc<HiveRouterConfig>,
         in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
-        aws_signing_params: Option<Arc<SigningParams<'static>>>,
+        aws_sigv4_signer: Option<Signer<Credential>>,
     ) -> Self {
         let mut header_map = HeaderMap::new();
         header_map.insert(
@@ -75,7 +74,7 @@ impl HTTPSubgraphExecutor {
             semaphore,
             config,
             in_flight_requests,
-            aws_signing_params,
+            aws_sigv4_signer,
         }
     }
 
@@ -138,51 +137,6 @@ impl HTTPSubgraphExecutor {
         Ok(body)
     }
 
-    async fn sign_awssigv4<'a>(
-        &self,
-        req: &'a mut http::Request<Full<Bytes>>,
-        body: &'a [u8],
-    ) -> Result<(), SubgraphExecutorError> {
-        if let Some(signing_params) = &self.aws_signing_params {
-            let signable_request = SignableRequest::new(
-                req.method().as_str(),
-                req.uri().to_string(),
-                req.headers().iter().map(|(k, v)| {
-                    (
-                        k.as_str(),
-                        str::from_utf8(v.as_bytes())
-                            .map_err(|err| {
-                                SubgraphExecutorError::AwsSigV4SigningFailure(
-                                    self.subgraph_name.to_string(),
-                                    err.to_string(),
-                                )
-                            })
-                            .unwrap(),
-                    )
-                }),
-                SignableBody::Bytes(body),
-            )
-            .map_err(|err| {
-                SubgraphExecutorError::AwsSigV4SigningFailure(
-                    self.subgraph_name.to_string(),
-                    err.to_string(),
-                )
-            })?;
-
-            let (signing_instructions, _) = sign(signable_request, &signing_params)
-                .map_err(|err| {
-                    SubgraphExecutorError::AwsSigV4SigningFailure(
-                        self.subgraph_name.to_string(),
-                        err.to_string(),
-                    )
-                })?
-                .into_parts();
-
-            signing_instructions.apply_to_request_http1x(req);
-        }
-        Ok(())
-    }
-
     async fn _send_request(
         &self,
         body: Vec<u8>,
@@ -192,19 +146,26 @@ impl HTTPSubgraphExecutor {
             .method(http::Method::POST)
             .uri(&self.endpoint)
             .version(Version::HTTP_11)
-            .body(Default::default())
+            .body(Full::new(Bytes::from(body)))
             .map_err(|e| {
                 SubgraphExecutorError::RequestBuildFailure(self.endpoint.to_string(), e.to_string())
             })?;
 
         *req.headers_mut() = headers;
 
-        self.sign_awssigv4(&mut req, &body).await?;
-
-        *req.body_mut() = Full::new(Bytes::from(body));
-
         debug!("making http request to {}", self.endpoint.to_string());
 
+        if let Some(aws_sigv4_signer) = &self.aws_sigv4_signer {
+            let (mut parts, body) = req.into_parts();
+            aws_sigv4_signer.sign(&mut parts, None).await.map_err(|e| {
+                SubgraphExecutorError::AwsSigV4SigningFailure(
+                    self.endpoint.to_string(),
+                    e.to_string(),
+                )
+            })?;
+            req = http::Request::from_parts(parts, body);
+        }
+
         let res = self.http_client.request(req).await.map_err(|e| {
             SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
         })?;

lib/executor/src/executors/http.rs

@@ -27,7 +27,7 @@ use vrl::{
 };
 
 use crate::{
-    execution::client_request_details::ClientRequestDetails,
+    execution::{awssigv4::create_awssigv4_signer, client_request_details::ClientRequestDetails},
     executors::{
         common::{
             HttpExecutionRequest, HttpExecutionResponse, SubgraphExecutor, SubgraphExecutorBoxedArc,
@@ -317,14 +317,27 @@ impl SubgraphExecutorMap {
             .or_insert_with(|| Arc::new(Semaphore::new(self.max_connections_per_host)))
             .clone();
 
+        let aws_sigv4_signer = if self.config.aws_sig_v4.is_disabled() {
+            None
+        } else {
+            let aws_sigv4_subgraph_config = self
+                .config
+                .aws_sig_v4
+                .subgraphs
+                .get(subgraph_name)
+                .unwrap_or(&self.config.aws_sig_v4.all);
+
+            Some(create_awssigv4_signer(aws_sigv4_subgraph_config))
+        };
+
         let executor = HTTPSubgraphExecutor::new(
             subgraph_name.to_string(),
             endpoint_uri,
             self.client.clone(),
             semaphore,
             self.config.clone(),
             self.in_flight_requests.clone(),
-            None,
+            aws_sigv4_signer,
         );
 
         self.executors_by_subgraph

lib/executor/src/executors/map.rs

@@ -13,7 +13,7 @@ pub struct AwsSigV4Config {
 
     // configuration that will apply to all subgraphs
     pub all: AwsSigV4SubgraphConfig,
-    
+
     // per-subgraph configuration overrides
     #[serde(default)]
     pub subgraphs: HashMap<String, AwsSigV4SubgraphConfig>,
@@ -66,6 +66,7 @@ pub struct HardCodedConfig {
     pub secret_access_key: String,
     pub region: String,
     pub service_name: String,
+    pub session_token: Option<String>,
 }
 
 impl AwsSigV4Config {
@@ -83,4 +84,4 @@ pub struct AssumeRoleConfig {
     pub role_arn: String,
     pub session_name: Option<String>,
     pub external_id: Option<String>,
-}
+}

lib/router-config/src/aws_sig_v4.rs

@@ -1,3 +1,4 @@
+pub mod aws_sig_v4;
 pub mod cors;
 pub mod csrf;
 mod env_overrides;
@@ -12,7 +13,6 @@ pub mod primitives;
 pub mod query_planner;
 pub mod supergraph;
 pub mod traffic_shaping;
-pub mod aws_sig_v4;
 
 use config::{Config, File, FileFormat, FileSourceFile};
 use envconfig::Envconfig;
@@ -95,7 +95,10 @@ pub struct HiveRouterConfig {
     pub override_labels: OverrideLabelsConfig,
 
     /// Configuration for AWS SigV4 signing of requests to subgraphs.
-    #[serde(default, skip_serializing_if = "aws_sig_v4::AwsSigV4Config::is_disabled")]
+    #[serde(
+        default,
+        skip_serializing_if = "aws_sig_v4::AwsSigV4Config::is_disabled"
+    )]
     pub aws_sig_v4: aws_sig_v4::AwsSigV4Config,
 }