feat(large-messages): add optional allowedBuckets allowlist

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`

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

@@ -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>

powertools-e2e-tests/handlers/largemessage-restricted/src/main/java/software/amazon/lambda/powertools/e2e/Function.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)));
+    }
+}

powertools-e2e-tests/handlers/largemessage-restricted/src/main/resources/log4j2.xml

@@ -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/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>