enhancement(aws provider): Add ability to disable request signing

changelog.d/allow_disabling_aws_request_signing.enhancement.md

@@ -0,0 +1,2 @@
+AWS components can now have request signing disabled by setting `auth.sign: false` on AWS
+components. This can be useful, for example, when sending anonymous requests to AWS S3.

src/aws/auth.rs

@@ -175,6 +175,14 @@ pub enum AwsAuthentication {
         /// [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints
         #[configurable(metadata(docs::examples = "us-west-2"))]
         region: Option<String>,
+
+        /// Whether to sign requests.
+        ///
+        /// Setting this to false can be useful when writing to an AWS S3 bucket that allows
+        /// anonymous requests.
+        #[serde(default = "crate::serde::default_true")]
+        #[derivative(Default(value = "true"))]
+        sign: bool,
     },
 }
 
@@ -239,7 +247,7 @@ impl AwsAuthentication {
         service_region: Region,
         proxy: &ProxyConfig,
         tls_options: &Option<TlsConfig>,
-    ) -> crate::Result<SharedCredentialsProvider> {
+    ) -> crate::Result<Option<SharedCredentialsProvider>> {
         match self {
             Self::AccessKey {
                 access_key_id,
@@ -265,9 +273,9 @@ impl AwsAuthentication {
 
                     let provider = builder.build_from_provider(provider).await;
 
-                    return Ok(SharedCredentialsProvider::new(provider));
+                    return Ok(Some(SharedCredentialsProvider::new(provider)));
                 }
-                Ok(provider)
+                Ok(Some(provider))
             }
             AwsAuthentication::File {
                 credentials_file,
@@ -288,7 +296,7 @@ impl AwsAuthentication {
                     .profile_name(profile)
                     .configure(&provider_config)
                     .build();
-                Ok(SharedCredentialsProvider::new(profile_provider))
+                Ok(Some(SharedCredentialsProvider::new(profile_provider)))
             }
             AwsAuthentication::Role {
                 assume_role,
@@ -313,17 +321,23 @@ impl AwsAuthentication {
                     )
                     .await;
 
-                Ok(SharedCredentialsProvider::new(provider))
+                Ok(Some(SharedCredentialsProvider::new(provider)))
             }
-            AwsAuthentication::Default { imds, region, .. } => Ok(SharedCredentialsProvider::new(
+            AwsAuthentication::Default { sign: false, .. } => Ok(None),
+            AwsAuthentication::Default {
+                sign: true,
+                imds,
+                region,
+                ..
+            } => Ok(Some(SharedCredentialsProvider::new(
                 default_credentials_provider(
                     region.clone().map(Region::new).unwrap_or(service_region),
                     proxy,
                     tls_options,
                     *imds,
                 )
                 .await?,
-            )),
+            ))),
         }
     }
 
@@ -411,6 +425,7 @@ mod tests {
                 load_timeout_secs: Some(10),
                 imds: ImdsAuthentication { .. },
                 region: None,
+                sign: true,
             }
         ));
     }
@@ -453,6 +468,7 @@ mod tests {
                     connect_timeout: CONNECT_TIMEOUT,
                     read_timeout: READ_TIMEOUT,
                 },
+                sign: true,
             }
         ));
     }
@@ -676,4 +692,21 @@ mod tests {
             _ => panic!(),
         }
     }
+
+    #[test]
+    fn parsing_sign_false() {
+        let config = toml::from_str::<ComponentConfig>(
+            r#"
+            auth.sign = false
+        "#,
+        )
+        .unwrap();
+
+        match config.auth {
+            AwsAuthentication::Default { sign, .. } => {
+                assert!(!sign);
+            }
+            _ => panic!(),
+        }
+    }
 }

src/aws/mod.rs

@@ -24,7 +24,7 @@ use aws_smithy_runtime_api::client::{
     runtime_components::RuntimeComponents,
 };
 use aws_smithy_types::body::SdkBody;
-use aws_types::sdk_config::SharedHttpClient;
+use aws_types::sdk_config::{SharedAsyncSleep, SharedHttpClient};
 use bytes::Bytes;
 use futures_util::FutureExt;
 use http::HeaderMap;
@@ -201,25 +201,27 @@ pub async fn create_client_and_region<T: ClientBuilder>(
     };
 
     // Build the configuration first.
-    let mut config_builder = SdkConfig::builder()
-        .http_client(connector)
-        .sleep_impl(Arc::new(TokioSleep::new()))
-        .identity_cache(auth.credentials_cache().await?)
-        .credentials_provider(
+    let mut config_builder = SdkConfig::builder();
+
+    config_builder
+        .set_http_client(Some(SharedHttpClient::new(connector)))
+        .set_sleep_impl(Some(SharedAsyncSleep::new(Arc::new(TokioSleep::new()))))
+        .set_identity_cache(Some(auth.credentials_cache().await?))
+        .set_region(Some(region.clone()))
+        .set_retry_config(Some(retry_config.clone()))
+        .set_credentials_provider(
             auth.credentials_provider(region.clone(), proxy, tls_options)
                 .await?,
-        )
-        .region(region.clone())
-        .retry_config(retry_config.clone());
+        );
 
     if let Some(endpoint_override) = endpoint {
-        config_builder = config_builder.endpoint_url(endpoint_override);
+        config_builder.set_endpoint_url(Some(endpoint_override));
     }
 
     if let Some(use_fips) =
         aws_config::default_provider::use_fips::use_fips_provider(&provider_config).await
     {
-        config_builder = config_builder.use_fips(use_fips);
+        config_builder.set_use_fips(Some(use_fips));
     }
 
     if let Some(timeout) = timeout {
@@ -234,7 +236,7 @@ pub async fn create_client_and_region<T: ClientBuilder>(
             .set_connect_timeout(connect_timeout.map(Duration::from_secs))
             .set_read_timeout(read_timeout.map(Duration::from_secs));
 
-        config_builder = config_builder.timeout_config(timeout_config_builder.build());
+        config_builder.set_timeout_config(Some(timeout_config_builder.build()));
     }
 
     let config = config_builder.build();

src/sinks/aws_kinesis/firehose/integration_tests.rs

@@ -83,6 +83,7 @@ async fn firehose_put_records_without_partition_key() {
             load_timeout_secs: Some(5),
             imds: ImdsAuthentication::default(),
             region: None,
+            sign: true,
         })),
         endpoints: vec![elasticsearch_address()],
         bulk: BulkConfig {
@@ -195,6 +196,7 @@ async fn firehose_put_records_with_partition_key() {
             load_timeout_secs: Some(5),
             imds: ImdsAuthentication::default(),
             region: None,
+            sign: true,
         })),
         endpoints: vec![elasticsearch_address()],
         bulk: BulkConfig {
@@ -278,6 +280,7 @@ async fn ensure_elasticsearch_domain(domain_name: String) -> String {
                         &None,
                     )
                     .await
+                    .unwrap()
                     .unwrap(),
             )
             .endpoint_url(test_region_endpoint().endpoint().unwrap())

src/sinks/elasticsearch/common.rs

@@ -269,10 +269,14 @@ impl ElasticsearchCommon {
 #[cfg(feature = "aws-core")]
 pub async fn sign_request(
     request: &mut http::Request<Bytes>,
-    credentials_provider: &aws_credential_types::provider::SharedCredentialsProvider,
+    credentials_provider: &Option<aws_credential_types::provider::SharedCredentialsProvider>,
     region: &Option<aws_types::region::Region>,
 ) -> crate::Result<()> {
-    crate::aws::sign_request("es", request, credentials_provider, region).await
+    if let Some(credentials_provider) = credentials_provider {
+        crate::aws::sign_request("es", request, credentials_provider, region).await
+    } else {
+        Ok(())
+    }
 }
 
 async fn get_version(

src/sinks/elasticsearch/integration_tests.rs

@@ -290,6 +290,7 @@ async fn auto_version_aws() {
                 load_timeout_secs: Some(5),
                 imds: ImdsAuthentication::default(),
                 region: None,
+                sign: true,
             },
         )),
         endpoints: vec![aws_server()],
@@ -396,6 +397,7 @@ async fn insert_events_on_aws() {
                     load_timeout_secs: Some(5),
                     imds: ImdsAuthentication::default(),
                     region: None,
+                    sign: true,
                 },
             )),
             endpoints: vec![aws_server()],
@@ -422,6 +424,7 @@ async fn insert_events_on_aws_with_compression() {
                     load_timeout_secs: Some(5),
                     imds: ImdsAuthentication::default(),
                     region: None,
+                    sign: true,
                 },
             )),
             endpoints: vec![aws_server()],

src/sinks/prometheus/remote_write/service.rs

@@ -94,10 +94,14 @@ impl Service<RemoteWriteRequest> for RemoteWriteService {
 #[cfg(feature = "aws-core")]
 async fn sign_request(
     request: &mut http::Request<Bytes>,
-    credentials_provider: &SharedCredentialsProvider,
+    credentials_provider: &Option<SharedCredentialsProvider>,
     region: &Option<Region>,
 ) -> crate::Result<()> {
-    crate::aws::sign_request("aps", request, credentials_provider, region).await
+    if let Some(credentials_provider) = credentials_provider {
+        crate::aws::sign_request("aps", request, credentials_provider, region).await
+    } else {
+        Ok(())
+    }
 }
 
 pub(super) async fn build_request(

src/sinks/util/auth.rs

@@ -3,7 +3,7 @@ pub enum Auth {
     Basic(crate::http::Auth),
     #[cfg(feature = "aws-core")]
     Aws {
-        credentials_provider: aws_credential_types::provider::SharedCredentialsProvider,
+        credentials_provider: Option<aws_credential_types::provider::SharedCredentialsProvider>,
         region: aws_types::region::Region,
     },
 }

src/sources/aws_sqs/integration_tests.rs

@@ -55,6 +55,7 @@ async fn get_sqs_client() -> aws_sdk_sqs::Client {
             AwsAuthentication::test_auth()
                 .credentials_provider(Region::new("custom"), &Default::default(), &None)
                 .await
+                .unwrap()
                 .unwrap(),
         )
         .endpoint_url(sqs_address())

website/cue/reference/components/sinks/base/aws_cloudwatch_logs.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_cloudwatch_logs: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	batch: {

website/cue/reference/components/sinks/base/aws_cloudwatch_metrics.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_cloudwatch_metrics: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	batch: {

website/cue/reference/components/sinks/base/aws_kinesis_firehose.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_kinesis_firehose: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	batch: {

website/cue/reference/components/sinks/base/aws_kinesis_streams.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_kinesis_streams: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	batch: {

website/cue/reference/components/sinks/base/aws_s3.cue

@@ -202,6 +202,16 @@ base: components: sinks: aws_s3: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	batch: {

website/cue/reference/components/sinks/base/aws_sns.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_sns: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	encoding: {

website/cue/reference/components/sinks/base/aws_sqs.cue

@@ -127,6 +127,16 @@ base: components: sinks: aws_sqs: configuration: {
 				required:    true
 				type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"]
 			}
+			sign: {
+				description: """
+					Whether to sign requests.
+
+					Setting this to false can be useful when writing to an AWS S3 bucket that allows
+					anonymous requests.
+					"""
+				required: false
+				type: bool: default: true
+			}
 		}
 	}
 	encoding: {