Prevent validating crc32 checksum on incomplete response stream (#5224)

* prevent validating crc32 checksum on incomplete response stream

* changelog

* added log when skipping crc32 validation

.changes/next-release/bugfix-AWSSDKforJavav2-f86abd6.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Prevent calculating crc32 checksum on an incomplete response, due to error"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/util/Crc32ChecksumValidatingInputStream.java

@@ -20,16 +20,20 @@
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.core.exception.Crc32MismatchException;
 import software.amazon.awssdk.core.io.SdkFilterInputStream;
+import software.amazon.awssdk.utils.Logger;
 
 /**
  * Wraps the provided input stream with a {@link Crc32ChecksumCalculatingInputStream} and after the stream is closed
  * will validate the calculated checksum against the actual checksum.
  */
 @SdkInternalApi
 public class Crc32ChecksumValidatingInputStream extends SdkFilterInputStream {
+    private static final Logger log = Logger.loggerFor(Crc32ChecksumValidatingInputStream.class);
 
     private final long expectedChecksum;
 
+    private boolean shouldValidateChecksum = true;
+
     /**
      * @param in               Input stream to content.
      * @param expectedChecksum Expected CRC32 checksum returned by the service.
@@ -53,7 +57,31 @@ public void close() throws IOException {
         }
     }
 
+    @Override
+    public int read() throws IOException {
+        try {
+            return super.read();
+        } catch (Throwable e) {
+            shouldValidateChecksum = false;
+            throw e;
+        }
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        try {
+            return super.read(b, off, len);
+        } catch (Throwable e) {
+            shouldValidateChecksum = false;
+            throw e;
+        }
+    }
+
     private void validateChecksum() throws Crc32MismatchException {
+        if (!shouldValidateChecksum) {
+            log.debug(() -> "Skipping CRC32 validation due to error encountered while reading from input stream");
+            return;
+        }
         long actualChecksum = ((Crc32ChecksumCalculatingInputStream) in).getCrc32Checksum();
         if (expectedChecksum != actualChecksum) {
             throw Crc32MismatchException.builder()

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/util/Crc32ChecksumValidatingInputStreamTest.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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.awssdk.core.internal.util;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import java.util.zip.CRC32;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.exception.Crc32MismatchException;
+import software.amazon.awssdk.testutils.InputStreamUtils;
+import software.amazon.awssdk.utils.StringInputStream;
+
+class Crc32ChecksumValidatingInputStreamTest {
+
+    @Test
+    void noException_correctChecksum_shouldNotThrow() {
+        String data = RandomStringUtils.random(128);
+        byte[] dataBytes = data.getBytes(StandardCharsets.UTF_8);
+        CRC32 crc32 = new CRC32();
+        crc32.update(dataBytes, 0, dataBytes.length);
+        Crc32ChecksumValidatingInputStream is = new Crc32ChecksumValidatingInputStream(
+            new StringInputStream(data), crc32.getValue());
+        assertThatCode(() -> InputStreamUtils.drainInputStream(is)).doesNotThrowAnyException();
+        assertThatCode(is::close).doesNotThrowAnyException();
+    }
+
+    @Test
+    void noException_incorrectChecksum_shouldThrowCRC32Exception() {
+        String data = RandomStringUtils.random(128);
+        byte[] dataBytes = data.getBytes(StandardCharsets.UTF_8);
+        CRC32 crc32 = new CRC32();
+        crc32.update(dataBytes, 0, dataBytes.length);
+        Crc32ChecksumValidatingInputStream is = new Crc32ChecksumValidatingInputStream(
+            new StringInputStream(data), crc32.getValue() == 0 ? 1 : crc32.getValue()/2);
+        InputStreamUtils.drainInputStream(is);
+
+        assertThatThrownBy(is::close).isInstanceOf(Crc32MismatchException.class);
+    }
+
+    @Test
+    void exceptionWhileReading_shouldNotValidateChecksum() {
+        Crc32ChecksumValidatingInputStream is = new Crc32ChecksumValidatingInputStream(
+            new FailAfterNInputStream(128, new IOException("test io exception")), 123);
+        assertThatThrownBy(() -> InputStreamUtils.drainInputStream(is))
+            .hasCauseInstanceOf(IOException.class)
+            .hasMessageContaining("test io exception");
+
+        assertThatCode(is::close).doesNotThrowAnyException();
+    }
+
+    @Test
+    void exception_readMethodCall_shouldNotValidateChecksum() {
+        Crc32ChecksumValidatingInputStream is = new Crc32ChecksumValidatingInputStream(
+            new FailAfterNInputStream(2, new IOException("test io exception")), 123);
+        assertThatThrownBy(() -> {
+            is.read();
+            is.read();
+        })
+            .isInstanceOf(IOException.class)
+            .hasMessageContaining("test io exception");
+
+        assertThatCode(is::close).doesNotThrowAnyException();
+    }
+
+    class FailAfterNInputStream extends InputStream {
+        private final int failAfter;
+        private final Random random;
+        private final IOException exceptionToThrow;
+
+        private int totalRead = 0;
+
+        FailAfterNInputStream(int failAfter, IOException exceptionToThrow) {
+            this.failAfter = failAfter;
+            this.exceptionToThrow = exceptionToThrow;
+            this.random = new Random();
+        }
+
+        @Override
+        public int read() throws IOException {
+            totalRead++;
+            if (totalRead >= failAfter) {
+                throw exceptionToThrow;
+            }
+            return random.nextInt();
+        }
+    }
+}