Memory leak in S3EncryptionClient.getObject()

### Security issue notifications

If you discover a potential security issue in the Amazon S3 Encryption Client we ask that you notify AWS Security via our [vulnerability reporting page](https://aws.amazon.com/security/vulnerability-reporting/). Please do **not** create a public GitHub issue.

_noted_ ✅

### Problem:

S3EncryptionClient.getObject() leaks memory on every call because the ResponseInputStream returned by the internal async pipeline is never closed.

Code in question: `src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java`

```
@Override
public T getObject(GetObjectRequest getObjectRequest,
ResponseTransformer responseTransformer)
throws AwsServiceException, SdkClientException {

// GetEncryptedObjectPipeline builder

try {
ResponseInputStream joinFutureGet = pipeline.getObject(getObjectRequest, AsyncResponseTransformer.toBlockingInputStream()).join();
return responseTransformer.transform(joinFutureGet.response(), AbortableInputStream.create(joinFutureGet));
} catch (CompletionException e) {
throw new S3EncryptionClientException(e.getCause().getMessage(), e.getCause());
} catch (Exception e) {
throw new S3EncryptionClientException("Unable to transform response.", e);
}
}
```

The `joinFutureGet` ResponseInputStream is never closed so in an internal service where we use this method to read client side encrypted files in batches we get these errors:

```
io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 2048 byte(s) of direct memory (used: 7,025,458,950, max: 7,025,459,200)
at io.netty.buffer.UnpooledByteBufAllocator.newDirectBuffer(UnpooledByteBufAllocator.java:97)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:168)
at io.netty.handler.ssl.SslHandler.allocate(SslHandler.java:2381)
at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1488)
```

### Reproduction test:

```
import java.lang.management.BufferPoolMXBean;
import java.lang.management.ManagementFactory;
import java.net.URI;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3Configuration;
import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.encryption.s3.S3EncryptionClient;

// Requires LocalStack: docker run -d -p 4566:4566 -e SERVICES=s3 localstack/localstack:3.0.2

var creds = StaticCredentialsProvider.create(AwsBasicCredentials.create("test", "test"));
var endpoint = URI.create("http://localhost:4566");

var encryptionClient = S3EncryptionClient.builder()
.credentialsProvider(creds)
.region(Region.EU_CENTRAL_1)
.endpointOverride(endpoint)
.forcePathStyle(true)
.kmsKeyId("alias/test-key")
.enableLegacyWrappingAlgorithms(true)
.enableLegacyUnauthenticatedModes(true)
.build();

// Create bucket and upload a test file
encryptionClient.createBucket(CreateBucketRequest.builder().bucket("test").build());
encryptionClient.putObject(
PutObjectRequest.builder().bucket("test").key("file.txt").build(),
RequestBody.fromString("hello"));

long before = getDirectMemoryUsed();

for (int i = 0; i < 5000; i++) {
try {
encryptionClient.getObjectAsBytes(
GetObjectRequest.builder().bucket("test").key("file.txt").build());
} catch (Exception e) {
// Decryption fails (no real KMS), but the leak happens regardless
}
}

long after = getDirectMemoryUsed();
System.out.printf("Leaked: %,d bytes in 5000 calls (~%d bytes/call)%n", after - before, (after - before) / 5000);

static long getDirectMemoryUsed() {
return ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class).stream()
.filter(pool -> pool.getName().equals("direct"))
.mapToLong(BufferPoolMXBean::getMemoryUsed)
.sum();
}
```
Results:

5000 getObjectAsBytes() calls:
Direct memory before: 45,056 bytes
Direct memory after: 4,993,032 bytes
Leaked: 4,947,976 bytes

Also got a warning:
```
LEAK: ByteBuf.release() was not called before it's garbage-collected.
Created at:
io.netty.buffer.AdaptiveByteBufAllocator.newDirectBuffer(AdaptiveByteBufAllocator.java:67)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:168)
...
```

### Solution:

Open the stream with try catch with resources block eg:

```
try (ResponseInputStream joinFutureGet =
pipeline.getObject(getObjectRequest, AsyncResponseTransformer.toBlockingInputStream()).join()) {
return responseTransformer.transform(joinFutureGet.response(), AbortableInputStream.create(joinFutureGet));
} catch (CompletionException e) {
throw new S3EncryptionClientException(e.getCause().getMessage(), e.getCause());
} catch (Exception e) {
throw new S3EncryptionClientException("Unable to transform response.", e);
}
```

### Out of scope:

N/A

☑️ (NOTE: If you believe this might be a security issue, please email aws-security@amazon.com instead of creating a GitHub issue. For more details, see the AWS Vulnerability Reporting Guide: https://aws.amazon.com/security/vulnerability-reporting/ )

Lies src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java und konzentriere dich auf getObject() und den ResponseInputStream, der von der asynchronen Pipeline zurückgegeben wird. Reproduziere das Problem mit dem bereitgestellten 5,000-call-LocalStack-Test und überprüfe anschließend, dass wiederholte Aufrufe den Direct Memory nicht mehr erhöhen und keine Netty-Leak-Warnungen erzeugen.

Vom Indexierungsmodell aus dem Issue-Text verfasst.