Merge branch 'main' of github.com:awspring/spring-cloud-aws into awsp…

…ringgh-1246

spring-cloud-aws-s3/src/main/java/io/awspring/cloud/s3/ObjectMetadata.java

@@ -25,6 +25,7 @@
  * Container for S3 Object Metadata. For information about each field look at {@link PutObjectRequest} Javadocs.
  *
  * @author Maciej Walkowiak
+ * @author Hardik Singh Behl
  * @since 3.0
  */
 public class ObjectMetadata {
@@ -116,6 +117,9 @@ public class ObjectMetadata {
 	@Nullable
 	private final String checksumAlgorithm;
 
+	@Nullable
+	private final String contentMD5;
+
 	public static Builder builder() {
 		return new Builder();
 	}
@@ -130,7 +134,7 @@ public static Builder builder() {
 			@Nullable String ssekmsKeyId, @Nullable String ssekmsEncryptionContext, @Nullable Boolean bucketKeyEnabled,
 			@Nullable String requestPayer, @Nullable String tagging, @Nullable String objectLockMode,
 			@Nullable Instant objectLockRetainUntilDate, @Nullable String objectLockLegalHoldStatus,
-			@Nullable String expectedBucketOwner, @Nullable String checksumAlgorithm) {
+			@Nullable String expectedBucketOwner, @Nullable String checksumAlgorithm, @Nullable String contentMD5) {
 		this.acl = acl;
 		this.cacheControl = cacheControl;
 		this.contentDisposition = contentDisposition;
@@ -160,6 +164,7 @@ public static Builder builder() {
 		this.objectLockLegalHoldStatus = objectLockLegalHoldStatus;
 		this.expectedBucketOwner = expectedBucketOwner;
 		this.checksumAlgorithm = checksumAlgorithm;
+		this.contentMD5 = contentMD5;
 	}
 
 	void apply(PutObjectRequest.Builder builder) {
@@ -250,6 +255,9 @@ void apply(PutObjectRequest.Builder builder) {
 		if (checksumAlgorithm != null) {
 			builder.checksumAlgorithm(checksumAlgorithm);
 		}
+		if (contentMD5 != null) {
+			builder.contentMD5(contentMD5);
+		}
 	}
 
 	void apply(CreateMultipartUploadRequest.Builder builder) {
@@ -523,6 +531,11 @@ public String getChecksumAlgorithm() {
 		return checksumAlgorithm;
 	}
 
+	@Nullable
+	public String getContentMD5() {
+		return contentMD5;
+	}
+
 	public static class Builder {
 
 		private final Map<String, String> metadata = new HashMap<>();
@@ -611,6 +624,9 @@ public static class Builder {
 		@Nullable
 		private String checksumAlgorithm;
 
+		@Nullable
+		private String contentMD5;
+
 		public Builder acl(@Nullable String acl) {
 			this.acl = acl;
 			return this;
@@ -785,13 +801,18 @@ public Builder checksumAlgorithm(@Nullable ChecksumAlgorithm checksumAlgorithm)
 			return checksumAlgorithm(checksumAlgorithm != null ? checksumAlgorithm.toString() : null);
 		}
 
+		public Builder contentMD5(@Nullable String contentMD5) {
+			this.contentMD5 = contentMD5;
+			return this;
+		}
+
 		public ObjectMetadata build() {
 			return new ObjectMetadata(acl, cacheControl, contentDisposition, contentEncoding, contentLanguage,
 					contentType, contentLength, expires, grantFullControl, grantRead, grantReadACP, grantWriteACP,
 					metadata, serverSideEncryption, storageClass, websiteRedirectLocation, sseCustomerAlgorithm,
 					sseCustomerKey, sseCustomerKeyMD5, ssekmsKeyId, ssekmsEncryptionContext, bucketKeyEnabled,
 					requestPayer, tagging, objectLockMode, objectLockRetainUntilDate, objectLockLegalHoldStatus,
-					expectedBucketOwner, checksumAlgorithm);
+					expectedBucketOwner, checksumAlgorithm, contentMD5);
 		}
 
 	}

spring-cloud-aws-s3/src/test/java/io/awspring/cloud/s3/S3TemplateIntegrationTests.java

@@ -20,12 +20,17 @@
 import static org.assertj.core.api.Assertions.assertThatNoException;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+
+import net.bytebuddy.utility.RandomString;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
 import java.time.Duration;
+import java.util.Base64;
 import java.util.List;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
@@ -47,6 +52,7 @@
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.HttpStatusCode;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
@@ -62,6 +68,7 @@
  * @author Maciej Walkowiak
  * @author Yuki Yoshida
  * @author Ziemowit Stolarczyk
+ * @author Hardik Singh Behl
  */
 @Testcontainers
 class S3TemplateIntegrationTests {
@@ -70,7 +77,7 @@ class S3TemplateIntegrationTests {
 
 	@Container
 	static LocalStackContainer localstack = new LocalStackContainer(
-			DockerImageName.parse("localstack/localstack:3.8.1"));
+			DockerImageName.parse("localstack/localstack:3.8.1")).withEnv("S3_SKIP_SIGNATURE_VALIDATION", "0");
 
 	private static S3Client client;
 
@@ -268,15 +275,21 @@ void createsWorkingSignedGetURL() throws IOException {
 
 	@Test
 	void createsWorkingSignedPutURL() throws IOException {
-		ObjectMetadata metadata = ObjectMetadata.builder().metadata("testkey", "testvalue").build();
+		String fileContent = RandomString.make();
+		long contentLength = fileContent.length();
+		String contentMD5 = calculateContentMD5(fileContent);
+
+		ObjectMetadata metadata = ObjectMetadata.builder().metadata("testkey", "testvalue").contentLength(contentLength)
+				.contentMD5(contentMD5).build();
 		URL signedPutUrl = s3Template.createSignedPutURL(BUCKET_NAME, "file.txt", Duration.ofMinutes(1), metadata,
 				"text/plain");
 
 		CloseableHttpClient httpClient = HttpClients.createDefault();
 		HttpPut httpPut = new HttpPut(signedPutUrl.toString());
 		httpPut.setHeader("x-amz-meta-testkey", "testvalue");
 		httpPut.setHeader("Content-Type", "text/plain");
-		HttpEntity body = new StringEntity("hello");
+		httpPut.setHeader("Content-MD5", contentMD5);
+		HttpEntity body = new StringEntity(fileContent);
 		httpPut.setEntity(body);
 
 		HttpResponse response = httpClient.execute(httpPut);
@@ -285,11 +298,36 @@ void createsWorkingSignedPutURL() throws IOException {
 		HeadObjectResponse headObjectResponse = client
 				.headObject(HeadObjectRequest.builder().bucket(BUCKET_NAME).key("file.txt").build());
 
-		assertThat(headObjectResponse.contentLength()).isEqualTo(5);
+		assertThat(response.getStatusLine().getStatusCode()).isEqualTo(HttpStatusCode.OK);
+		assertThat(headObjectResponse.contentLength()).isEqualTo(contentLength);
 		assertThat(headObjectResponse.metadata().containsKey("testkey")).isTrue();
 		assertThat(headObjectResponse.metadata().get("testkey")).isEqualTo("testvalue");
 	}
 
+	@Test
+	void signedPutURLFailsForNonMatchingSignature() throws IOException {
+		String fileContent = RandomString.make();
+		long contentLength = fileContent.length();
+		String contentMD5 = calculateContentMD5(fileContent);
+		String maliciousContent = RandomString.make();
+
+		ObjectMetadata metadata = ObjectMetadata.builder().contentLength(contentLength).contentMD5(contentMD5).build();
+		URL signedPutUrl = s3Template.createSignedPutURL(BUCKET_NAME, "file.txt", Duration.ofMinutes(1), metadata,
+				"text/plain");
+
+		CloseableHttpClient httpClient = HttpClients.createDefault();
+		HttpPut httpPut = new HttpPut(signedPutUrl.toString());
+		httpPut.setHeader("Content-Type", "text/plain");
+		httpPut.setHeader("Content-MD5", contentMD5);
+		HttpEntity body = new StringEntity(fileContent + maliciousContent);
+		httpPut.setEntity(body);
+
+		HttpResponse response = httpClient.execute(httpPut);
+		httpClient.close();
+
+		assertThat(response.getStatusLine().getStatusCode()).isEqualTo(HttpStatusCode.FORBIDDEN);
+	}
+
 	private void bucketDoesNotExist(ListBucketsResponse r, String bucketName) {
 		assertThat(r.buckets().stream().filter(b -> b.name().equals(bucketName)).findAny()).isEmpty();
 	}
@@ -298,6 +336,17 @@ private void bucketExists(ListBucketsResponse r, String bucketName) {
 		assertThat(r.buckets().stream().filter(b -> b.name().equals(bucketName)).findAny()).isPresent();
 	}
 
+	private String calculateContentMD5(String content) {
+		try {
+			MessageDigest md = MessageDigest.getInstance("MD5");
+			byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
+			byte[] mdBytes = md.digest(contentBytes);
+			return Base64.getEncoder().encodeToString(mdBytes);
+		} catch (Exception exception) {
+			throw new RuntimeException("Failed to calculate Content-MD5", exception);
+		}
+	}
+
 	static class Person {
 		private String firstName;
 		private String lastName;

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/QueueAttributesResolver.java

@@ -38,6 +38,7 @@
 import software.amazon.awssdk.services.sqs.model.GetQueueUrlResponse;
 import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
 import software.amazon.awssdk.services.sqs.model.QueueDoesNotExistException;
+import software.amazon.awssdk.services.sqs.model.SqsException;
 
 /**
  * Resolves {@link QueueAttributes} for the specified queue. Fetchs the queue url for a queue name, unless a url is
@@ -85,8 +86,15 @@ public CompletableFuture<QueueAttributes> resolveQueueAttributes() {
 	}
 
 	private CompletableFuture<QueueAttributes> wrapException(Throwable t) {
-		return CompletableFutures.failedFuture(new QueueAttributesResolvingException("Error resolving attributes for queue "
-			+ this.queueName + " with strategy " + this.queueNotFoundStrategy + " and queueAttributesNames " + this.queueAttributeNames,
+		String message = "Error resolving attributes for queue "
+			+ this.queueName + " with strategy " + this.queueNotFoundStrategy + " and queueAttributesNames " + this.queueAttributeNames;
+
+		if (t.getCause() instanceof SqsException) {
+			message += "\n This might be due to connectivity issues or incorrect configuration. " +
+				"Please verify your AWS credentials, network settings, and queue configuration.";
+		}
+
+		return CompletableFutures.failedFuture(new QueueAttributesResolvingException(message,
 			t instanceof CompletionException ? t.getCause() : t));
 	}
 

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/SqsTemplate.java

@@ -454,7 +454,7 @@ private Integer getDelaySeconds(Message message) {
 	private Map<MessageSystemAttributeNameForSends, MessageSystemAttributeValue> mapMessageSystemAttributes(
 			Message message) {
 		return message.attributes().entrySet().stream().filter(Predicate.not(entry -> isSkipAttribute(entry.getKey())))
-				.collect(Collectors.toMap(entry -> MessageSystemAttributeNameForSends.fromValue(entry.getKey().name()),
+				.collect(Collectors.toMap(entry -> MessageSystemAttributeNameForSends.fromValue(entry.getKey().toString()),
 						entry -> MessageSystemAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING)
 								.stringValue(entry.getValue()).build()));
 	}
@@ -602,11 +602,11 @@ private ReceiveMessageRequest doCreateReceiveMessageRequest(Duration pollTimeout
 		ReceiveMessageRequest.Builder builder = ReceiveMessageRequest.builder().queueUrl(attributes.getQueueUrl())
 				.maxNumberOfMessages(maxNumberOfMessages).messageAttributeNames(this.messageAttributeNames)
 				.attributeNamesWithStrings(this.messageSystemAttributeNames)
-				.waitTimeSeconds(pollTimeout.toSecondsPart());
+				.waitTimeSeconds(toInt(pollTimeout.toSeconds()));
 		if (additionalHeaders.containsKey(SqsHeaders.SQS_VISIBILITY_TIMEOUT_HEADER)) {
 			builder.visibilityTimeout(
-					getValueAs(additionalHeaders, SqsHeaders.SQS_VISIBILITY_TIMEOUT_HEADER, Duration.class)
-							.toSecondsPart());
+				toInt(getValueAs(additionalHeaders, SqsHeaders.SQS_VISIBILITY_TIMEOUT_HEADER, Duration.class)
+							.toSeconds()));
 		}
 		if (additionalHeaders.containsKey(SqsHeaders.SQS_RECEIVE_REQUEST_ATTEMPT_ID_HEADER)) {
 			builder.receiveRequestAttemptId(
@@ -616,6 +616,15 @@ private ReceiveMessageRequest doCreateReceiveMessageRequest(Duration pollTimeout
 		return builder.build();
 	}
 
+	// Convert a long value to an int. Values larger than Integer.MAX_VALUE are set to Integer.MAX_VALUE
+	private int toInt(long longValue) {
+		if (longValue > Integer.MAX_VALUE) {
+			return Integer.MAX_VALUE;
+		}
+
+		return (int) longValue;
+	}
+
 	private <V> V getValueAs(Map<String, Object> headers, String headerName, Class<V> valueClass) {
 		return valueClass.cast(headers.get(headerName));
 	}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SnsMessageConverter.java

@@ -72,7 +72,10 @@ private Object fromGenericMessages(List<GenericMessage<?>> messages, Class<?> ta
 		Type resolvedType = getResolvedType(targetClass, conversionHint);
 		Class<?> resolvedClazz = ResolvableType.forType(resolvedType).resolve();
 
-		return messages.stream().map(message -> fromGenericMessage(message, resolvedClazz, resolvedType)).toList();
+		Object hint = targetClass.isAssignableFrom(List.class) &&
+			conversionHint instanceof MethodParameter mp ? mp.nested() : conversionHint;
+
+		return messages.stream().map(message -> fromGenericMessage(message, resolvedClazz, hint)).toList();
 	}
 
 	private Object fromGenericMessage(GenericMessage<?> message, Class<?> targetClass,

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapper.java

@@ -79,6 +79,10 @@ public Message fromHeaders(MessageHeaders headers) {
 			attributes.put(MessageSystemAttributeName.MESSAGE_DEDUPLICATION_ID,
 					headers.get(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER, String.class));
 		}
+		if (headers.containsKey(SqsHeaders.MessageSystemAttributes.SQS_AWS_TRACE_HEADER)) {
+			attributes.put(MessageSystemAttributeName.AWS_TRACE_HEADER,
+				headers.get(SqsHeaders.MessageSystemAttributes.SQS_AWS_TRACE_HEADER, String.class));
+		}
 		Map<String, MessageAttributeValue> messageAttributes = headers.entrySet().stream()
 				.filter(entry -> !isSkipHeader(entry.getKey())).collect(Collectors.toMap(Map.Entry::getKey,
 						entry -> getMessageAttributeValue(entry.getKey(), entry.getValue())));
@@ -110,6 +114,7 @@ else if (messageHeaderValue instanceof ByteBuffer) {
 	private boolean isSkipHeader(String headerName) {
 		return SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER.equals(headerName)
 				|| SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER.equals(headerName)
+				|| SqsHeaders.MessageSystemAttributes.SQS_AWS_TRACE_HEADER.equals(headerName)
 				|| SqsHeaders.SQS_DELAY_HEADER.equals(headerName) || MessageHeaders.ID.equals(headerName)
 				|| MessageHeaders.TIMESTAMP.equals(headerName);
 	}

spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java

@@ -275,7 +275,7 @@ static class ResolvesPojoWithNotificationAnnotationListener {
 
 		@SqsListener(queueNames = RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME, id = "resolves-pojo-with-notification-message", factory = "defaultSqsListenerContainerFactory")
 		void listen(@SnsNotificationMessage MyEnvelope<MyPojo> myPojo) {
-			Assert.notNull((myPojo).data.firstField, "Received null message");
+			assertThat(myPojo.getData().getFirstField()).isEqualTo("pojoNotificationMessage");
 			logger.debug("Received message {} from queue {}", myPojo,
 					RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_QUEUE_NAME);
 			latchContainer.resolvesPojoNotificationMessageLatch.countDown();
@@ -286,6 +286,10 @@ void listen(@SnsNotificationMessage List<MyEnvelope<MyPojo>> myPojos) {
 			Assert.notEmpty(myPojos, "Received empty messages");
 			logger.debug("Received messages {} from queue {}", myPojos,
 					RESOLVES_POJO_FROM_NOTIFICATION_MESSAGE_LIST_QUEUE_NAME);
+
+			for (MyEnvelope<MyPojo> myPojo : myPojos) {
+				assertThat(myPojo.getData().getFirstField()).isEqualTo("pojoNotificationMessage");
+			}
 			latchContainer.resolvesPojoNotificationMessageListLatch.countDown();
 		}
 	}