[Filebeat][aws-s3] Need to add support for new schema for Security Lake subscriber queue format

[janvi-elastic]

We want to collect data from SQS Queue which is automatically created when we add data subscriber in Amazon Security Lake. In that we are getting s3 path details instead of s3 object. Here is the example.

{
    "source": "aws.s3",
    "time": "2023-09-16T05:27:07Z",
    "account": "123456789012",
    "region": "us-west-2",
    "resources": ["resource_arn"],
    "detail":{
        "bucket": {
            "name": "aws-security-data-lake-us-west-2"
        },
        "object": {
            "key": "aws/ROUTE53/1.0/region=us-west-2/accountId=12345678/eventDay=20230916/test.gz.parquet",
            "size": 13176,
            "etag": "asdfghj567ghwjhhd"
        },
        "request-id": "request-id",
        "requester": "securitylake.amazonaws.com"
    }
}

While running filebeat we are facing below error.

{"log.level":"warn","@timestamp":"2023-09-20T10:46:40.686+0530","log.logger":"input.aws-s3.sqs","log.origin":{"file.name":"awss3/sqs.go","file.line":94},"message":"Failed processing SQS message.","service.name":"filebeat","id":"id","queue_url":"https://sqs.us-west-2.amazonaws.com/account_id/AmazonSecurityLake-Main-Queue","error":{"message":"failed processing SQS message (message was deleted): non-retryable error: the message is an invalid S3 notification: missing Records field"},"message_id":"1e49a07d-73c2-4d4f-9587-6b11daf245e6","elapsed_time_ns":720439234,"ecs.version":"1.6.0"}

Describe a specific use case for the enhancement or feature:

While creating Amazon Security Lake integration.

Related info: https://docs.aws.amazon.com/security-lake/latest/userguide/subscriber-data-access.html#sample-notification

[elasticmachine]

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

[andrewkroh]

This can be handled by https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-aws-s3.html#_sqs_notification_parsing_script_source. The crowdstrike.fdr integration has a good example to follow.

[elasticmachine]

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)