Enforce payload checksums

CRC32, CRC32C, SHA1, and SHA256 are support but CRC64NVME is not.
Fixes #806.  References #830.

src/main/java/org/gaul/s3proxy/ChunkedInputStream.java

@@ -19,10 +19,15 @@ package org.gaul.s3proxy;
 import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Base64;
 
+import javax.annotation.Nullable;
+
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
 import com.google.common.io.ByteStreams;
 
-
 /**
  * Parse an AWS v4 signature chunked stream.  Reference:
  * https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
@@ -36,9 +41,27 @@ final class ChunkedInputStream extends FilterInputStream {
             justification = "https://github.com/gaul/s3proxy/issues/205")
     @SuppressWarnings("UnusedVariable")
     private String currentSignature;
+    private final Hasher hasher;
 
     ChunkedInputStream(InputStream is) {
         super(is);
+        hasher = null;
+    }
+
+    ChunkedInputStream(InputStream is, @Nullable String trailer) {
+        super(is);
+        if ("x-amz-checksum-crc32".equals(trailer)) {
+            hasher = Hashing.crc32().newHasher();
+        } else if ("x-amz-checksum-crc32c".equals(trailer)) {
+            hasher = Hashing.crc32c().newHasher();
+        } else if ("x-amz-checksum-sha1".equals(trailer)) {
+            hasher = Hashing.sha1().newHasher();
+        } else if ("x-amz-checksum-sha256".equals(trailer)) {
+            hasher = Hashing.sha256().newHasher();
+        } else {
+            // TODO: Guava does not support x-amz-checksum-crc64nvme
+            hasher = null;
+        }
     }
 
     @Override
@@ -50,6 +73,25 @@ final class ChunkedInputStream extends FilterInputStream {
             }
             String[] parts = line.split(";", 2);
             if (parts[0].startsWith("x-amz-checksum-")) {
+                String[] checksumParts = parts[0].split(":", 2);
+                var expectedHash = checksumParts[1];
+                String actualHash;
+                switch (checksumParts[0]) {
+                case "x-amz-checksum-crc32":
+                case "x-amz-checksum-crc32c":
+                    // Use big-endian to match AWS
+                    actualHash = Base64.getEncoder().encodeToString(ByteBuffer.allocate(4).putInt(hasher.hash().asInt()).array());
+                    break;
+                case "x-amz-checksum-sha1":
+                case "x-amz-checksum-sha256":
+                    actualHash = Base64.getEncoder().encodeToString(hasher.hash().asBytes());
+                    break;
+                default:
+                    throw new IllegalArgumentException("Unknown value: " + checksumParts[0]);
+                }
+                if (!expectedHash.equals(actualHash)) {
+                    throw new IOException(new S3Exception(S3ErrorCode.BAD_DIGEST));
+                }
                 currentLength = 0;
             } else {
                 currentLength = Integer.parseInt(parts[0], 16);
@@ -60,10 +102,14 @@ final class ChunkedInputStream extends FilterInputStream {
             chunk = new byte[currentLength];
             currentIndex = 0;
             ByteStreams.readFully(in, chunk);
+            if (hasher != null) {
+                hasher.putBytes(chunk);
+            }
             // TODO: check currentSignature
             if (currentLength == 0) {
                 return -1;
             }
+            // consume trailing \r\n
             readLine(in);
         }
         return chunk[currentIndex++] & 0xFF;

src/main/java/org/gaul/s3proxy/S3ProxyHandler.java

@@ -516,7 +516,7 @@ public class S3ProxyHandler {
             if ("STREAMING-AWS4-HMAC-SHA256-PAYLOAD".equals(contentSha256)) {
                 is = new ChunkedInputStream(is);
             } else if ("STREAMING-UNSIGNED-PAYLOAD-TRAILER".equals(contentSha256)) {
-                is = new ChunkedInputStream(is);
+                is = new ChunkedInputStream(is, request.getHeader(AwsHttpHeaders.TRAILER));
             }
         } else if (requestIdentity == null) {
             throw new S3Exception(S3ErrorCode.ACCESS_DENIED);
@@ -608,7 +608,7 @@ public class S3ProxyHandler {
                         is = new ChunkedInputStream(is);
                     } else if ("STREAMING-UNSIGNED-PAYLOAD-TRAILER".equals(contentSha256)) {
                         payload = new byte[0];
-                        is = new ChunkedInputStream(is);
+                        is = new ChunkedInputStream(is, request.getHeader(AwsHttpHeaders.TRAILER));
                     } else if ("UNSIGNED-PAYLOAD".equals(contentSha256)) {
                         payload = new byte[0];
                     } else {

src/main/java/org/gaul/s3proxy/S3ProxyHandlerJetty.java

@@ -141,6 +141,14 @@ final class S3ProxyHandlerJetty extends AbstractHandler {
                     ise.getMessage());
             baseRequest.setHandled(true);
             return;
+        } catch (IOException ioe) {
+            var cause = Throwables2.getFirstThrowableOfType(ioe, S3Exception.class);
+            if (cause != null) {
+                sendS3Exception(request, response, cause);
+                baseRequest.setHandled(true);
+                return;
+            }
+            throw ioe;
         } catch (KeyNotFoundException knfe) {
             S3ErrorCode code = S3ErrorCode.NO_SUCH_KEY;
             handler.sendSimpleErrorResponse(request, response, code,

src/test/java/org/gaul/s3proxy/AwsSdk2Test.java

@@ -30,6 +30,7 @@ import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.utils.AttributeMap;
 
@@ -84,6 +85,11 @@ public final class AwsSdk2Test {
         var putRequest = PutObjectRequest.builder()
                 .bucket(containerName)
                 .key(key)
+                // TODO: parameterize test with JUnit 5
+                //.checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                .checksumAlgorithm(ChecksumAlgorithm.CRC32_C)
+                //.checksumAlgorithm(ChecksumAlgorithm.SHA1)
+                //.checksumAlgorithm(ChecksumAlgorithm.SHA256)
                 .build();
 
         s3Client.putObject(putRequest, RequestBody.fromBytes(byteSource.read()));