fix(aws): Include default headers in signature calculation (#484) (#636)

* fix(aws): Include default headers in signature calculation (#484)

When using `ClientOptions::with_default_headers()` to set S3 metadata
headers (like `x-amz-meta-*` or `x-amz-tagging`), these headers were not
included in the AWS SigV4 signature calculation, causing S3 to reject
requests with "headers present in the request which were not signed".

This fix adds default headers to the request before signing in all three
S3 request paths: `S3Client::request()`, `bulk_delete_request()`, and
`get_request()`.

* test(aws): Extend default headers signing test to cover all request paths

Add test coverage for bulk_delete_request and get_request to verify
default headers are included in signature calculation. Extract
assert_default_headers_signed and default_headers_config helpers to
reduce duplication.

* style(aws): Apply rustfmt formatting

---------

Co-authored-by: Saabir Singh <saabirs@amazon.com>

Author: singhsaabir

src/aws/client.rs

@@ -467,9 +467,13 @@ impl S3Client {
 
     pub(crate) fn request<'a>(&'a self, method: Method, path: &'a Path) -> Request<'a> {
         let url = self.config.path_url(path);
+        let mut builder = self.client.request(method, url);
+        if let Some(headers) = self.config.client_options.get_default_headers() {
+            builder = builder.headers(headers.clone());
+        }
         Request {
             path,
-            builder: self.client.request(method, url),
+            builder,
             payload: None,
             payload_sha256: None,
             config: &self.config,
@@ -535,6 +539,9 @@ impl S3Client {
         let body = Bytes::from(buffer);
 
         let mut builder = self.client.request(Method::POST, url);
+        if let Some(headers) = self.config.client_options.get_default_headers() {
+            builder = builder.headers(headers.clone());
+        }
 
         let digest = digest::digest(&digest::SHA256, &body);
         builder = builder.header(SHA256_CHECKSUM, BASE64_STANDARD.encode(digest));
@@ -863,6 +870,9 @@ impl GetClient for S3Client {
         };
 
         let mut builder = self.client.request(method, url);
+        if let Some(headers) = self.config.client_options.get_default_headers() {
+            builder = builder.headers(headers.clone());
+        }
         if self
             .config
             .encryption_headers
@@ -958,10 +968,15 @@ fn encode_path(path: &Path) -> PercentEncode<'_> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::GetOptions;
     use crate::client::HttpClient;
+    use crate::client::get::GetClient;
     use crate::client::mock_server::MockServer;
+    use crate::client::retry::RetryContext;
     use http::Response;
-    use http::header::CONTENT_LENGTH;
+    use http::header::{AUTHORIZATION, CONTENT_LENGTH};
+    use hyper::Request;
+    use hyper::body::Incoming;
 
     #[tokio::test]
     async fn test_create_multipart_has_content_length() {
@@ -1010,4 +1025,116 @@ mod tests {
         assert_eq!(result.unwrap(), "test-upload-id");
         mock.shutdown().await;
     }
+
+    fn assert_default_headers_signed(req: &Request<Incoming>) {
+        assert_eq!(req.headers().get("x-amz-meta-test").unwrap(), "test-value");
+        assert_eq!(req.headers().get("x-amz-tagging").unwrap(), "key=value");
+
+        let auth = req.headers().get(AUTHORIZATION).unwrap().to_str().unwrap();
+        assert!(
+            auth.contains("x-amz-meta-test"),
+            "x-amz-meta-test not in SignedHeaders: {auth}"
+        );
+        assert!(
+            auth.contains("x-amz-tagging"),
+            "x-amz-tagging not in SignedHeaders: {auth}"
+        );
+    }
+
+    fn default_headers_config(mock: &MockServer) -> S3Config {
+        let credential = AwsCredential {
+            key_id: "AKIAIOSFODNN7EXAMPLE".to_string(),
+            secret_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string(),
+            token: None,
+        };
+
+        let mut default_headers = HeaderMap::new();
+        default_headers.insert("x-amz-meta-test", "test-value".parse().unwrap());
+        default_headers.insert("x-amz-tagging", "key=value".parse().unwrap());
+
+        S3Config {
+            bucket_endpoint: mock.url().to_string(),
+            bucket: "test-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Arc::new(crate::StaticCredentialProvider::new(credential)),
+            client_options: ClientOptions::new()
+                .with_allow_http(true)
+                .with_default_headers(default_headers),
+            skip_signature: false,
+            session_provider: None,
+            retry_config: Default::default(),
+            sign_payload: false,
+            disable_tagging: false,
+            checksum: None,
+            copy_if_not_exists: None,
+            conditional_put: Default::default(),
+            encryption_headers: Default::default(),
+            request_payer: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn test_default_headers_signed_request() {
+        let mock = MockServer::new().await;
+        mock.push_fn(|req| {
+            assert_default_headers_signed(&req);
+            Response::builder()
+                .status(200)
+                .header("etag", "\"test-etag\"")
+                .body(String::new())
+                .unwrap()
+        });
+
+        let config = default_headers_config(&mock);
+        let client = S3Client::new(config, HttpClient::new(reqwest::Client::new()));
+        let result = client
+            .request(Method::PUT, &Path::from("test"))
+            .with_payload(PutPayload::default())
+            .do_put()
+            .await;
+
+        assert!(result.is_ok());
+        mock.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn test_default_headers_signed_bulk_delete() {
+        let mock = MockServer::new().await;
+        mock.push_fn(|req| {
+            assert_default_headers_signed(&req);
+            Response::builder()
+                .status(200)
+                .body("<DeleteResult><Deleted><Key>test</Key></Deleted></DeleteResult>".to_string())
+                .unwrap()
+        });
+
+        let config = default_headers_config(&mock);
+        let client = S3Client::new(config, HttpClient::new(reqwest::Client::new()));
+        let result = client.bulk_delete_request(vec![Path::from("test")]).await;
+
+        assert!(result.is_ok());
+        mock.shutdown().await;
+    }
+
+    #[tokio::test]
+    async fn test_default_headers_signed_get_request() {
+        let mock = MockServer::new().await;
+        mock.push_fn(|req| {
+            assert_default_headers_signed(&req);
+            Response::builder()
+                .status(200)
+                .body("test-body".to_string())
+                .unwrap()
+        });
+
+        let config = default_headers_config(&mock);
+        let client = S3Client::new(config, HttpClient::new(reqwest::Client::new()));
+        let mut ctx = RetryContext::new(&client.config.retry_config);
+        let result = client
+            .get_request(&mut ctx, &Path::from("test"), GetOptions::default())
+            .await;
+
+        assert!(result.is_ok());
+        mock.shutdown().await;
+    }
 }

src/client/mod.rs

@@ -700,6 +700,11 @@ impl ClientOptions {
         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