diff --git a/src/internal_events/aws_sqs.rs b/src/internal_events/aws_sqs.rs index b0c14493ea57a..d2e1e491133fa 100644 --- a/src/internal_events/aws_sqs.rs +++ b/src/internal_events/aws_sqs.rs @@ -372,3 +372,18 @@ impl InternalEvent for SqsS3EventRecordInvalidEventIgnored<'_> { .increment(1); } } + +#[derive(Debug, NamedInternalEvent)] +pub struct SqsCloudTrailNotificationIgnored<'a> { + pub bucket: &'a str, + pub object_count: usize, +} + +impl InternalEvent for SqsCloudTrailNotificationIgnored<'_> { + fn emit(self) { + warn!(message = "Ignored CloudTrail log delivery notification in SQS message; only S3 event notifications are ingested.", + bucket = %self.bucket, object_count = %self.object_count); + counter!("sqs_s3_event_record_ignored_total", "ignore_type" => "cloudtrail_notification") + .increment(1); + } +} diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs index c8a922f7b69e0..c96ee99f00207 100644 --- a/src/sources/aws_s3/sqs.rs +++ b/src/sources/aws_s3/sqs.rs @@ -50,7 +50,7 @@ use crate::{ event::{BatchNotifier, BatchStatus, EstimatedJsonEncodedSizeOf, Event, LogEvent}, internal_events::{ EventsReceived, S3ObjectProcessingFailed, S3ObjectProcessingSucceeded, - SqsMessageDeleteBatchError, SqsMessageDeletePartialError, SqsMessageDeleteSucceeded, + SqsCloudTrailNotificationIgnored, SqsMessageDeleteBatchError, SqsMessageDeletePartialError, SqsMessageDeleteSucceeded, SqsMessageProcessingError, SqsMessageProcessingSucceeded, SqsMessageReceiveError, SqsMessageReceiveSucceeded, SqsMessageSendBatchError, SqsMessageSentPartialError, SqsMessageSentSucceeded, SqsS3EventRecordInvalidEventIgnored, StreamClosedError, @@ -628,6 +628,14 @@ impl IngestorProcess { Ok(()) } SqsEvent::Event(s3_event) => self.handle_s3_event(s3_event).await, + // Returning Ok deletes the message; left unparsed it is redelivered until retention expires. + SqsEvent::CloudTrailNotification(notification) => { + emit!(SqsCloudTrailNotificationIgnored { + bucket: ¬ification.s3_bucket, + object_count: notification.s3_object_key.len(), + }); + Ok(()) + } } } @@ -1003,6 +1011,17 @@ pub struct SnsNotification { enum SqsEvent { Event(S3Event), TestEvent(S3TestEvent), + // Keep last: untagged variants are tried in order, so existing matches are unchanged. + CloudTrailNotification(CloudTrailNotification), +} + +// Sent by a trail's own SNS topic, which customers often share with the bucket's S3 event topic. +// https://docs.aws.amazon.com/awscloudtrail/latest/userguide/configure-cloudtrail-to-send-notifications.html +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CloudTrailNotification { + pub s3_bucket: String, + pub s3_object_key: Vec, } #[derive(Clone, Debug, Deserialize)] @@ -1257,6 +1276,89 @@ fn test_s3_sns_testevent() { assert_eq!(value.event.name, "TestEvent".to_string()); } +#[test] +fn test_cloudtrail_notification() { + let value: SqsEvent = serde_json::from_str( + r#"{ + "s3Bucket":"bucketname", + "s3ObjectKey":["AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/123456789012_CloudTrail_us-east-1_20260919T0410Z_abcdefgh.json.gz"] + }"#, + ) + .unwrap(); + + match value { + SqsEvent::CloudTrailNotification(notification) => { + assert_eq!(notification.s3_bucket, "bucketname".to_string()); + assert_eq!(notification.s3_object_key.len(), 1); + } + other => panic!("expected CloudTrailNotification, got {other:?}"), + } +} + +#[test] +fn test_sns_cloudtrail_notification() { + let sns_value: SnsNotification = serde_json::from_str( + r#"{ + "Type" : "Notification", + "MessageId" : "63a3f6b6-d533-4a47-aef9-fcf5cf758c76", + "TopicArn" : "arn:aws:sns:us-west-2:123456789012:MyTopic", + "Message" : "{\"s3Bucket\":\"bucketname\",\"s3ObjectKey\":[\"AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/a.json.gz\",\"AWSLogs/123456789012/CloudTrail/us-east-1/2026/09/19/b.json.gz\"]}", + "Timestamp" : "2012-03-29T05:12:16.901Z", + "SignatureVersion" : "1", + "Signature" : "EXAMPLEnTrFPa3...", + "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem", + "UnsubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-west-2:123456789012:MyTopic:c7fe3a54-ab0e-4ec2-88e0-db410a0f2bee" + }"#, + ).unwrap(); + + let value: SqsEvent = serde_json::from_str(sns_value.message.as_ref()).unwrap(); + + match value { + SqsEvent::CloudTrailNotification(notification) => { + assert_eq!(notification.s3_bucket, "bucketname".to_string()); + assert_eq!(notification.s3_object_key.len(), 2); + } + other => panic!("expected CloudTrailNotification, got {other:?}"), + } +} + +// The CloudTrail variant must never shadow the shapes that were already handled. +#[test] +fn test_sqs_event_variant_order() { + let s3_event: SqsEvent = serde_json::from_str( + r#"{ + "Records":[{ + "eventVersion":"2.1", + "eventSource":"aws:s3", + "awsRegion":"us-east-1", + "eventTime":"2026-09-19T04:10:00.000Z", + "eventName":"ObjectCreated:Put", + "s3":{ + "bucket":{"name":"bucketname"}, + "object":{"key":"AWSLogs/123456789012/CloudTrail/us-east-1/a.json.gz"} + } + }] + }"#, + ) + .unwrap(); + assert!(matches!(s3_event, SqsEvent::Event(_))); + + let test_event: SqsEvent = serde_json::from_str( + r#"{ + "Service":"Amazon S3", + "Event":"s3:TestEvent", + "Time":"2014-10-13T15:57:02.089Z", + "Bucket":"bucketname" + }"#, + ) + .unwrap(); + assert!(matches!(test_event, SqsEvent::TestEvent(_))); + + // Anything else must keep failing so it stays visible as a processing error. + assert!(serde_json::from_str::(r#"{"s3Bucket":"bucketname"}"#).is_err()); + assert!(serde_json::from_str::(r#"{"detail-type":"Object Created"}"#).is_err()); +} + #[test] fn parse_sqs_config() { let config: Config = toml::from_str(