feat(large-messages): add optional allowedBuckets allowlist (#2555)

* feat(large-messages): add optional allowedBuckets allowlist

Add an opt-in bucket allowlist to the v2 large messages utility. The
S3 bucket in the message pointer is controlled by the sender, so this
gives consumers fine-grained, in-application control over which buckets
the utility may read from and delete, complementing IAM permissions.

Configure via LargeMessageConfig.withAllowedBuckets(Set). When a
non-empty allowlist is set, a message whose pointer names a bucket
outside the allowlist is rejected with LargeMessageProcessingException
before any S3 interaction. An empty allowlist (default) preserves the
existing behavior, so the change is backward compatible and applies to
both the @LargeMessage annotation and the functional API.

Includes unit tests, an e2e handler and negative e2e test, and docs
with a security note, resource-scoped IAM example, and usage examples.

* fix(large-messages): wrap getAllowedBuckets() return in unmodifiableSet

Avoid exposing the internal set reference (SpotBugs EI_EXPOSE_REP).

Author: phipag

docs/utilities/large_messages.md

@@ -174,6 +174,40 @@ on the S3 bucket used for the large messages offloading:
 - `s3:GetObject`
 - `s3:DeleteObject`
 
+### Security: Restrict which buckets the utility accesses
+
+???+ warning "The message sender controls the `s3BucketName` in the payload pointer"
+    The large message body contains a pointer of the form
+    `["software.amazon.payloadoffloading.PayloadS3Pointer",{"s3BucketName":"...","s3Key":"..."}]`.
+    The utility reads `s3BucketName` from this pointer and fetches the object using your Lambda function's
+    own IAM credentials. Any sender that can write to your queue, or publish to a topic that fans out to it,
+    can name any bucket your execution role can reach. Your function then reads from that bucket. Delete-after-read
+    is enabled by default (`deleteS3Object=true`), so your function also deletes the object it read.
+
+Apply both of the following controls:
+
+1. **Configure an allowlist** with `LargeMessageConfig.init().withAllowedBuckets(...)`. The utility rejects any
+   message whose pointer names a bucket outside the allowlist before it reads or deletes from S3. See
+   [Restricting allowed buckets](#restricting-allowed-buckets).
+2. **Scope your IAM policy** to the offload bucket. Grant `s3:GetObject` and `s3:DeleteObject` on the bucket
+   ARN instead of `*`:
+
+    ```json
+    {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": [
+                    "s3:GetObject",
+                    "s3:DeleteObject"
+                ],
+                "Resource": "arn:aws:s3:::<your-offload-bucket>/*"
+            }
+        ]
+    }
+    ```
+
 ## Usage
 
 You can use the Large Messages utility with either the `@LargeMessage` annotation or the functional API.
@@ -481,6 +515,66 @@ If you need to customize this `S3Client`, you can leverage the `LargeMessageConf
     }
     ```
 
+## Restricting allowed buckets
+
+The message sender controls the bucket named in the message pointer (see
+[the security note in Permissions](#security-restrict-which-buckets-the-utility-accesses)). Use the
+`LargeMessageConfig` singleton to restrict which S3 buckets the utility reads from and deletes. This works with
+both the annotation and the functional API. When you configure a non-empty allowlist, the utility rejects any
+message whose pointer names a bucket outside the allowlist. It throws a `LargeMessageProcessingException` before
+it reads or deletes from S3. An empty allowlist, the default, applies no restriction.
+
+=== "@LargeMessage annotation"
+    ```java hl_lines="6"
+    import software.amazon.lambda.powertools.largemessages.LargeMessage;
+
+    public class SqsMessageHandler implements RequestHandler<SQSEvent, SQSBatchResponse> {
+
+        public SqsMessageHandler() {
+            LargeMessageConfig.init().withAllowedBuckets(Collections.singleton("my-offload-bucket"));
+        }
+
+        @Override
+        public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
+            for (SQSMessage message: event.getRecords()) {
+                // throws LargeMessageProcessingException if the pointer's bucket is not in the allowlist
+                processRawMessage(message, context);
+            }
+            return SQSBatchResponse.builder().build();
+        }
+
+        @LargeMessage
+        private void processRawMessage(SQSEvent.SQSMessage sqsMessage, Context context) {
+            // sqsMessage.getBody() will contain the content of the S3 object
+        }
+    }
+    ```
+
+=== "Functional API"
+    ```java hl_lines="6"
+    import software.amazon.lambda.powertools.largemessages.LargeMessages;
+
+    public class SqsMessageHandler implements RequestHandler<SQSEvent, SQSBatchResponse> {
+
+        public SqsMessageHandler() {
+            LargeMessageConfig.init().withAllowedBuckets(Collections.singleton("my-offload-bucket"));
+        }
+
+        @Override
+        public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
+            for (SQSMessage message: event.getRecords()) {
+                // throws LargeMessageProcessingException if the pointer's bucket is not in the allowlist
+                LargeMessages.processLargeMessage(message, this::processRawMessage);
+            }
+            return SQSBatchResponse.builder().build();
+        }
+
+        private void processRawMessage(SQSEvent.SQSMessage sqsMessage) {
+            // sqsMessage.getBody() will contain the content of the S3 object
+        }
+    }
+    ```
+
 ## Migration from the SQS Large Message utility
 
 - Replace the dependency in maven / gradle: `powertools-sqs` ==> `powertools-large-messages`

examples/powertools-examples-kafka/src/main/java/org/demo/kafka/protobuf/ProtobufProduct.java

@@ -0,0 +1,72 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>software.amazon.lambda</groupId>
+        <artifactId>e2e-test-handlers-parent</artifactId>
+        <version>2.10.0</version>
+    </parent>
+
+    <artifactId>e2e-test-handler-largemessage-restricted</artifactId>
+    <packaging>jar</packaging>
+    <name>E2E test handler – Large message (allowedBuckets restricted)</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>dynamodb</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.lambda</groupId>
+            <artifactId>powertools-large-messages</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.lambda</groupId>
+            <artifactId>powertools-logging-log4j</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-lambda-java-events</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.aspectj</groupId>
+            <artifactId>aspectjrt</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>dev.aspectj</groupId>
+                <artifactId>aspectj-maven-plugin</artifactId>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                    <complianceLevel>${maven.compiler.target}</complianceLevel>
+                    <aspectLibraries>
+                        <aspectLibrary>
+                            <groupId>software.amazon.lambda</groupId>
+                            <artifactId>powertools-large-messages</artifactId>
+                        </aspectLibrary>
+                        <aspectLibrary>
+                            <groupId>software.amazon.lambda</groupId>
+                            <artifactId>powertools-logging</artifactId>
+                        </aspectLibrary>
+                    </aspectLibraries>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

examples/powertools-examples-kafka/src/main/java/org/demo/kafka/protobuf/ProtobufProductOrBuilder.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2023 Amazon.com, Inc. or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package software.amazon.lambda.powertools.e2e;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
+import software.amazon.awssdk.utils.BinaryUtils;
+import software.amazon.awssdk.utils.Md5Utils;
+import software.amazon.lambda.powertools.largemessages.LargeMessage;
+import software.amazon.lambda.powertools.largemessages.LargeMessageConfig;
+import software.amazon.lambda.powertools.logging.Logging;
+
+public class Function implements RequestHandler<SQSEvent, SQSBatchResponse> {
+
+    // The real offload bucket is created per test run with a random name (largemessagebucket<random>).
+    // By pinning the allowlist to a fixed, non-matching bucket name, every offloaded message must be
+    // rejected before any S3 read or delete, exercising the bucket-allowlist security check.
+    private static final String DISALLOWED_BUCKET = "powertools-e2e-disallowed-bucket";
+
+    private static final String TABLE_FOR_ASYNC_TESTS = System.getenv("TABLE_FOR_ASYNC_TESTS");
+    private DynamoDbClient client;
+
+    public Function() {
+        LargeMessageConfig.init().withAllowedBuckets(Collections.singleton(DISALLOWED_BUCKET));
+        if (client == null) {
+            client = DynamoDbClient.builder()
+                    .httpClient(UrlConnectionHttpClient.builder().build())
+                    .region(Region.of(System.getenv("AWS_REGION")))
+                    .build();
+        }
+    }
+
+    @Logging(logEvent = true)
+    public SQSBatchResponse handleRequest(SQSEvent event, Context context) {
+        for (SQSMessage message : event.getRecords()) {
+            processRawMessage(message, context);
+        }
+        return SQSBatchResponse.builder().build();
+    }
+
+    @LargeMessage
+    private void processRawMessage(SQSMessage sqsMessage, Context context) {
+        String bodyMD5 = md5(sqsMessage.getBody());
+        if (!sqsMessage.getMd5OfBody().equals(bodyMD5)) {
+            throw new SecurityException(
+                    String.format("message digest does not match, expected %s, got %s", sqsMessage.getMd5OfBody(),
+                            bodyMD5));
+        }
+
+        Map<String, AttributeValue> item = new HashMap<>();
+        item.put("functionName", AttributeValue.builder().s(context.getFunctionName()).build());
+        item.put("id", AttributeValue.builder().s(sqsMessage.getMessageId()).build());
+        item.put("bodyMD5", AttributeValue.builder().s(bodyMD5).build());
+        item.put("bodySize",
+                AttributeValue.builder().n(String.valueOf(sqsMessage.getBody().getBytes(StandardCharsets.UTF_8).length))
+                        .build());
+
+        client.putItem(PutItemRequest.builder().tableName(TABLE_FOR_ASYNC_TESTS).item(item).build());
+    }
+
+    private String md5(String message) {
+        return BinaryUtils.toHex(Md5Utils.computeMD5Hash(message.getBytes(StandardCharsets.UTF_8)));
+    }
+}

examples/powertools-examples-kafka/src/main/java/org/demo/kafka/protobuf/ProtobufProductOuterClass.java

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration>
+    <Appenders>
+        <Console name="JsonAppender" target="SYSTEM_OUT">
+            <JsonTemplateLayout eventTemplateUri="classpath:LambdaJsonLayout.json" />
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="JsonAppender"/>
+        </Root>
+        <Logger name="JsonLogger" level="INFO" additivity="false">
+            <AppenderRef ref="JsonAppender"/>
+        </Logger>
+    </Loggers>
+</Configuration>

powertools-e2e-tests/handlers/largemessage-restricted/pom.xml

@@ -28,6 +28,7 @@
         <module>batch</module>
         <module>largemessage</module>
         <module>largemessage-functional</module>
+        <module>largemessage-restricted</module>
         <module>largemessage_idempotent</module>
         <module>logging-log4j</module>
         <module>logging-log4j-fluent-api</module>