feat(aws_s3 source): add custom SQS authentication option

Add support for configuring separate authentication credentials for SQS
in the AWS S3 source. This enables cross-account scenarios where S3
buckets and SQS queues are in different AWS accounts or require
different permission models.

Changes:
- Add optional `sqs.auth` field to SQS configuration
- Add optional `sqs.deferred.auth` field for deferred queue
- Modify client creation logic to use SQS-specific auth when provided
- Fallback to main `auth` when `sqs.auth` is not specified
- Full backwards compatibility maintained

Configuration example:
```yaml
[sources.my_s3_source]
type = "aws_s3"
auth.access_key_id = "S3_KEY"
auth.secret_access_key = "S3_SECRET"
sqs.queue_url = "https://sqs.us-east-1.amazonaws.com/123/queue"
sqs.auth.access_key_id = "SQS_KEY"
sqs.auth.secret_access_key = "SQS_SECRET"
```

Testing:
- Unit tests for config parsing (all auth variants)
- Integration tests with LocalStack (multi-auth scenarios)
- Backwards compatibility tests (existing configs work unchanged)

Files modified:
- src/sources/aws_s3/mod.rs (client creation logic)
- src/sources/aws_s3/sqs.rs (config structures and tests)
- changelog.d/aws_s3_custom_sqs_auth.enhancement.md (changelog)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: kfirg-cetu

changelog.d/aws_s3_custom_sqs_auth.enhancement.md

@@ -0,0 +1,7 @@
+The `aws_s3` source now supports configuring separate authentication credentials for SQS via the new `sqs.auth` configuration option. This enables cross-account scenarios where S3 buckets and SQS queues are in different AWS accounts or require different permission models.
+
+When `sqs.auth` is not specified, the source falls back to using the main `auth` configuration, maintaining full backwards compatibility with existing deployments.
+
+The `sqs.deferred.auth` option is also available for configuring separate authentication for the deferred message queue.
+
+authors: kfir

src/sources/aws_s3/mod.rs

@@ -263,20 +263,45 @@ impl AwsS3Config {
 
         match self.sqs {
             Some(ref sqs) => {
+                // Use SQS-specific auth if provided, otherwise fall back to main auth
+                let sqs_auth = sqs.auth.as_ref().unwrap_or(&self.auth);
+
                 let (sqs_client, region) = create_client_and_region::<SqsClientBuilder>(
                     &SqsClientBuilder {},
-                    &self.auth,
+                    sqs_auth,
                     region.clone(),
-                    endpoint,
+                    endpoint.clone(),
                     proxy,
                     sqs.tls_options.as_ref(),
                     sqs.timeout.as_ref(),
                 )
                 .await?;
 
+                // Create deferred SQS client if deferred queue has separate auth
+                let deferred_sqs_client = if let Some(ref deferred) = sqs.deferred {
+                    if let Some(ref deferred_auth) = deferred.auth {
+                        let (client, _) = create_client_and_region::<SqsClientBuilder>(
+                            &SqsClientBuilder {},
+                            deferred_auth,
+                            Some(region.clone()),
+                            endpoint.clone(),
+                            proxy,
+                            sqs.tls_options.as_ref(),
+                            sqs.timeout.as_ref(),
+                        )
+                        .await?;
+                        Some(client)
+                    } else {
+                        None
+                    }
+                } else {
+                    None
+                };
+
                 let ingestor = sqs::Ingestor::new(
                     region,
                     sqs_client,
+                    deferred_sqs_client,
                     s3_client,
                     sqs.clone(),
                     self.compression,
@@ -1068,4 +1093,130 @@ mod integration_tests {
         .await
         .unwrap()
     }
+
+    #[tokio::test]
+    async fn s3_process_message_with_different_sqs_auth() {
+        trace_init();
+
+        let logs: Vec<String> = random_lines(100).take(10).collect();
+
+        // Create config with separate SQS auth
+        let s3 = s3_client().await;
+        let sqs = sqs_client().await;
+        let queue = create_queue(&sqs).await;
+        let bucket = create_bucket(&s3).await;
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
+
+        let mut config = config(&queue, None, false, DeserializerConfig::Bytes);
+
+        // Add custom SQS auth (LocalStack accepts any credentials)
+        config.sqs.as_mut().unwrap().auth = Some(AwsAuthentication::AccessKey {
+            access_key_id: "DIFFERENT_KEY_ID".to_string().into(),
+            secret_access_key: "DIFFERENT_SECRET_KEY".to_string().into(),
+            session_token: None,
+            assume_role: None,
+            external_id: None,
+            region: None,
+            session_name: None,
+        });
+
+        // Verify the source builds and runs with different SQS auth
+        let key = uuid::Uuid::new_v4().to_string();
+        let payload = logs.join("\n").into_bytes();
+
+        s3.put_object()
+            .bucket(bucket.clone())
+            .key(key.clone())
+            .body(ByteStream::from(payload))
+            .send()
+            .await
+            .expect("Could not put object");
+
+        let sqs_client = sqs_client().await;
+        let mut s3_event: S3Event = serde_json::from_str(
+            r#"{"Records":[{"eventVersion":"2.1","eventSource":"aws:s3","awsRegion":"us-east-1","eventTime":"2022-03-24T19:43:00.548Z","eventName":"ObjectCreated:Put","userIdentity":{"principalId":"AWS:ARNOTAREALIDD4:user.name"},"requestParameters":{"sourceIPAddress":"136.56.73.213"},"responseElements":{"x-amz-request-id":"ZX6X98Q6NM9NQTP3","x-amz-id-2":"ESLLtyT4N5cAPW+C9EXwtaeEWz6nq7eCA6txjZKlG2Q7xp2nHXQI69Od2B0PiYIbhUiX26NrpIQPV0lLI6js3nVNmYo2SWBs"},"s3":{"s3SchemaVersion":"1.0","configurationId":"asdfasdf","bucket":{"name":"bucket-name","ownerIdentity":{"principalId":"A3PEG170DF9VNQ"},"arn":"arn:aws:s3:::nfox-testing-vector"},"object":{"key":"test-log.txt","size":33,"eTag":"c981ce6672c4251048b0b834e334007f","sequencer":"00623CC9C47AB5634C"}}}]}"#,
+        )
+        .unwrap();
+
+        s3_event.records[0].s3.bucket.name.clone_from(&bucket);
+        s3_event.records[0].s3.object.key.clone_from(&key);
+
+        sqs_client
+            .send_message()
+            .queue_url(queue.clone())
+            .message_body(serde_json::to_string(&s3_event).unwrap())
+            .send()
+            .await
+            .unwrap();
+
+        let (tx, rx) = SourceSender::new_test_finalize(Delivered);
+        let cx = SourceContext::new_test(tx, None);
+        let source = config.build(cx).await.unwrap();
+        tokio::spawn(async move { source.await.unwrap() });
+
+        let events = collect_n(rx, logs.len()).await;
+        assert_eq!(logs.len(), events.len());
+
+        tokio::time::sleep(Duration::from_secs(10)).await;
+        assert_eq!(count_messages(&sqs, &queue, 0).await, 0);
+    }
+
+    #[tokio::test]
+    async fn s3_process_message_without_sqs_auth_uses_default() {
+        trace_init();
+
+        let logs: Vec<String> = random_lines(100).take(10).collect();
+
+        // Test backwards compatibility - config without sqs.auth should work
+        test_event(
+            None,
+            None,
+            None,
+            None,
+            logs.join("\n").into_bytes(),
+            logs,
+            Delivered,
+            false,
+            DeserializerConfig::Bytes,
+            None,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn s3_process_message_with_deferred_auth() {
+        trace_init();
+
+        let s3 = s3_client().await;
+        let sqs = sqs_client().await;
+        let queue = create_queue(&sqs).await;
+        let deferred_queue = create_queue(&sqs).await;
+        let _bucket = create_bucket(&s3).await;
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
+
+        let mut config = config(&queue, None, false, DeserializerConfig::Bytes);
+
+        // Add deferred queue with custom auth
+        config.sqs.as_mut().unwrap().deferred = Some(sqs::DeferredConfig {
+            queue_url: deferred_queue.clone(),
+            max_age_secs: 0, // Set to 0 so all messages are deferred
+            auth: Some(AwsAuthentication::AccessKey {
+                access_key_id: "DEFERRED_KEY_ID".to_string().into(),
+                secret_access_key: "DEFERRED_SECRET_KEY".to_string().into(),
+                session_token: None,
+                assume_role: None,
+                external_id: None,
+                region: None,
+                session_name: None,
+            }),
+        });
+
+        // Verify the source builds correctly with deferred auth
+        let (tx, _rx) = SourceSender::new_test();
+        let cx = SourceContext::new_test(tx, None);
+        let source = config.build(cx).await;
+        assert!(source.is_ok(), "Source should build successfully with deferred auth");
+    }
 }

src/sources/aws_s3/sqs.rs

@@ -43,7 +43,7 @@ use vector_lib::{
 
 use crate::{
     SourceSender,
-    aws::AwsTimeout,
+    aws::{AwsAuthentication, AwsTimeout},
     codecs::Decoder,
     common::backoff::ExponentialBackoff,
     config::{SourceAcknowledgementsConfig, SourceContext},
@@ -83,6 +83,15 @@ pub(super) struct DeferredConfig {
     #[configurable(metadata(docs::type_unit = "seconds"))]
     #[configurable(metadata(docs::examples = 3600))]
     pub(super) max_age_secs: u64,
+
+    /// Authentication configuration for the deferred queue.
+    ///
+    /// If not specified, the main `auth` configuration will be used.
+    ///
+    /// This allows the deferred queue to be in a different AWS account or use different credentials.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub(super) auth: Option<AwsAuthentication>,
 }
 
 /// SQS configuration options.
@@ -180,6 +189,16 @@ pub(super) struct Config {
     #[serde(flatten)]
     pub(super) timeout: Option<AwsTimeout>,
 
+    /// Authentication configuration for SQS.
+    ///
+    /// If not specified, the main `auth` configuration will be used.
+    ///
+    /// This allows SQS to use different credentials than S3, enabling cross-account scenarios
+    /// where the S3 bucket and SQS queue are in different AWS accounts.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub(super) auth: Option<AwsAuthentication>,
+
     /// Configuration for deferring events to another queue based on their age.
     #[configurable(derived)]
     pub(super) deferred: Option<DeferredConfig>,
@@ -279,6 +298,7 @@ pub struct State {
 
     s3_client: S3Client,
     sqs_client: SqsClient,
+    deferred_sqs_client: Option<SqsClient>,
 
     multiline: Option<line_agg::Config>,
     compression: super::Compression,
@@ -303,6 +323,7 @@ impl Ingestor {
     pub(super) async fn new(
         region: Region,
         sqs_client: SqsClient,
+        deferred_sqs_client: Option<SqsClient>,
         s3_client: S3Client,
         config: Config,
         compression: super::Compression,
@@ -319,6 +340,7 @@ impl Ingestor {
 
             s3_client,
             sqs_client,
+            deferred_sqs_client,
 
             compression,
             multiline,
@@ -896,8 +918,14 @@ impl IngestorProcess {
         entries: Vec<SendMessageBatchRequestEntry>,
         queue_url: String,
     ) -> Result<SendMessageBatchOutput, SdkError<SendMessageBatchError, HttpResponse>> {
-        self.state
-            .sqs_client
+        // Use deferred SQS client if available, otherwise use main client
+        let client = self
+            .state
+            .deferred_sqs_client
+            .as_ref()
+            .unwrap_or(&self.state.sqs_client);
+
+        client
             .send_message_batch()
             .queue_url(queue_url.clone())
             .set_entries(Some(entries))
@@ -1299,3 +1327,121 @@ fn parse_sqs_config() {
     );
     assert!(test.is_err());
 }
+
+#[test]
+fn parse_sqs_config_with_custom_auth() {
+    let config: Config = toml::from_str(
+        r#"
+            queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+            [auth]
+            access_key_id = "AKIAIOSFODNN7EXAMPLE"
+            secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+        "#,
+    )
+    .unwrap();
+    assert_eq!(
+        config.queue_url,
+        "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+    );
+    assert!(config.auth.is_some());
+
+    // Verify the auth was parsed correctly
+    match config.auth.unwrap() {
+        AwsAuthentication::AccessKey {
+            access_key_id,
+            secret_access_key,
+            ..
+        } => {
+            assert_eq!(access_key_id.inner(), "AKIAIOSFODNN7EXAMPLE");
+            assert_eq!(
+                secret_access_key.inner(),
+                "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+            );
+        }
+        _ => panic!("Expected AccessKey auth variant"),
+    }
+}
+
+#[test]
+fn parse_sqs_config_with_assume_role() {
+    let config: Config = toml::from_str(
+        r#"
+            queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+            [auth]
+            assume_role = "arn:aws:iam::123456789012:role/SQSRole"
+            external_id = "external123"
+        "#,
+    )
+    .unwrap();
+    assert_eq!(
+        config.queue_url,
+        "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+    );
+    assert!(config.auth.is_some());
+
+    // Verify role assumption was parsed correctly
+    match config.auth.unwrap() {
+        AwsAuthentication::Role {
+            assume_role,
+            external_id,
+            ..
+        } => {
+            assert_eq!(assume_role, "arn:aws:iam::123456789012:role/SQSRole");
+            assert_eq!(external_id, Some("external123".to_string()));
+        }
+        _ => panic!("Expected Role auth variant"),
+    }
+}
+
+#[test]
+fn parse_sqs_config_without_auth_uses_default() {
+    let config: Config = toml::from_str(
+        r#"
+            queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+        "#,
+    )
+    .unwrap();
+    assert_eq!(
+        config.queue_url,
+        "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+    );
+    // Auth should be None, allowing fallback to main auth
+    assert!(config.auth.is_none());
+}
+
+#[test]
+fn parse_deferred_config_with_custom_auth() {
+    let config: Config = toml::from_str(
+        r#"
+            queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
+            [deferred]
+            queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/MyDeferredQueue"
+            max_age_secs = 3600
+            [deferred.auth]
+            access_key_id = "AKIADEFERREDEXAMPLE"
+            secret_access_key = "deferredSecretKeyExample"
+        "#,
+    )
+    .unwrap();
+
+    let deferred = config.deferred.expect("Expected deferred config");
+    assert_eq!(
+        deferred.queue_url,
+        "https://sqs.us-east-1.amazonaws.com/123456789012/MyDeferredQueue"
+    );
+    assert_eq!(deferred.max_age_secs, 3600);
+    assert!(deferred.auth.is_some());
+
+    // Verify deferred auth was parsed correctly
+    match deferred.auth.unwrap() {
+        AwsAuthentication::AccessKey {
+            access_key_id,
+            secret_access_key,
+            ..
+        } => {
+            assert_eq!(access_key_id.inner(), "AKIADEFERREDEXAMPLE");
+            assert_eq!(secret_access_key.inner(), "deferredSecretKeyExample");
+        }
+        _ => panic!("Expected AccessKey auth variant"),
+    }
+}