Add new Checksum Algorithms for data integrity validation (#161)

Author: push2stack

pom.xml

@@ -63,6 +63,13 @@
             <scope>provided</scope> <!-- lombok is not needed at runtime -->
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk.crt</groupId>
+            <artifactId>aws-crt</artifactId>
+            <version>0.39.4</version>
+        </dependency>
+
+
         <!-- Test Dependencies -->
         <dependency>
             <groupId>ch.qos.logback</groupId>

src/main/java/io/github/jpmorganchase/fusion/api/request/PartFetcher.java

@@ -2,9 +2,11 @@
 
 import static io.github.jpmorganchase.fusion.api.tools.ResponseChecker.checkResponseStatus;
 
+import io.github.jpmorganchase.fusion.FusionConfiguration;
 import io.github.jpmorganchase.fusion.api.response.GetPartResponse;
 import io.github.jpmorganchase.fusion.api.response.Head;
 import io.github.jpmorganchase.fusion.api.stream.IntegrityCheckingInputStream;
+import io.github.jpmorganchase.fusion.digest.PartChecker;
 import io.github.jpmorganchase.fusion.http.Client;
 import io.github.jpmorganchase.fusion.http.HttpResponse;
 import io.github.jpmorganchase.fusion.oauth.provider.FusionTokenProvider;
@@ -28,6 +30,7 @@ public class PartFetcher {
 
     Client client;
     FusionTokenProvider credentials;
+    FusionConfiguration configuration;
 
     /**
      * Makes a call to get the part corresponding to the part number in the {@link PartRequest}.
@@ -55,9 +58,15 @@ private IntegrityCheckingInputStream getIntegrityCheckingInputStream(
         return IntegrityCheckingInputStream.builder()
                 .part(response.getBody())
                 .checksum(head.getChecksum())
+                .partChecker(
+                        PartChecker.builder().digestAlgo(getDigestAlgo(head)).build())
                 .build();
     }
 
+    private String getDigestAlgo(Head head) {
+        return head.getChecksumAlgorithm() != null ? head.getChecksumAlgorithm() : configuration.getDigestAlgorithm();
+    }
+
     private Head getHead(HttpResponse<InputStream> response, PartRequest pr) {
         if (pr.isHeadRequired()) {
             return Head.builder().fromHeaders(response.getHeaders()).build();

src/main/java/io/github/jpmorganchase/fusion/api/response/Head.java

@@ -18,15 +18,19 @@
 @EqualsAndHashCode
 public class Head {
 
-    private static String CHECKSUM_HEADER = "x-jpmc-checksum-sha256";
+    private static String CHECKSUM_HEADER = "x-jpmc-checksum";
+    private static String CHECKSUM_SHA256_HEADER = "x-jpmc-checksum-sha256";
     private static String VERSION_HEADER = "x-jpmc-version-id";
     private static String PART_COUNT_HEADER = "x-jpmc-mp-parts-count";
     private static String CONTENT_LENGTH_HEADER = "Content-Length";
     private static String CONTENT_RANGE_HEADER = "Content-Range";
     private static String CHECKSUM_SEPARATOR = "-";
+    private static String DEFAULT_ALGORITHM = "SHA-256";
+    private static String CHECKSUM_ALGO_HEADER = "x-jpmc-checksum-algorithm";
 
     private String version;
     private String checksum;
+    private String checksumAlgorithm;
     private int partCount;
     private long contentLength;
     private ContentRange contentRange;
@@ -58,10 +62,11 @@ public Head build() {
                 handlePartCountHeader();
                 handleChecksumHeader();
                 handleVersionHeader();
+                handleChecksumAlgorithmHeader();
                 handleContentLength();
                 handleContentRangeHeader();
             }
-            return new Head(version, checksum, partCount, contentLength, contentRange, isMultipart);
+            return new Head(version, checksum, checksumAlgorithm, partCount, contentLength, contentRange, isMultipart);
         }
 
         private void handleContentLength() {
@@ -88,6 +93,22 @@ private void handleChecksumHeader() {
                 if (!values.isEmpty() && nonNull(values.get(0))) {
                     this.checksum = values.get(0).split(CHECKSUM_SEPARATOR)[0];
                 }
+            } else {
+                List<String> values = this.headers.get(CHECKSUM_SHA256_HEADER);
+                if (nonNull(values) && !values.isEmpty() && nonNull(values.get(0))) {
+                    this.checksum = values.get(0).split(CHECKSUM_SEPARATOR)[0];
+                }
+            }
+        }
+
+        private void handleChecksumAlgorithmHeader() {
+            if (this.headers.containsKey(CHECKSUM_ALGO_HEADER)) {
+                List<String> values = this.headers.get(CHECKSUM_ALGO_HEADER);
+                if (!values.isEmpty() && nonNull(values.get(0))) {
+                    this.checksumAlgorithm = values.get(0);
+                }
+            } else {
+                this.checksumAlgorithm = DEFAULT_ALGORITHM;
             }
         }
 

src/main/java/io/github/jpmorganchase/fusion/digest/PartChecker.java

@@ -1,9 +1,9 @@
 package io.github.jpmorganchase.fusion.digest;
 
+import io.github.jpmorganchase.fusion.digest.checksum.DigestProvider;
+import io.github.jpmorganchase.fusion.digest.checksum.DigestProviderService;
 import java.io.IOException;
 import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.Base64;
 import java.util.Objects;
 import lombok.Builder;
 import lombok.Getter;
@@ -20,44 +20,47 @@ public class PartChecker {
 
     MessageDigest digest;
 
-    String digestAlgo;
+    private DigestProvider digestProvider;
+    private String digestAlgo;
+    private DigestProviderService digestProviderService;
 
-    public PartChecker(String digestAlgo) {
+    public PartChecker(String digestAlgo, DigestProviderService digestProviderService) {
         this.digestAlgo = digestAlgo;
+        this.digestProviderService = digestProviderService;
     }
 
     public void update(int bytesRead) throws IOException {
-        if (Objects.isNull(digest)) {
+        if (Objects.isNull(digestProvider)) {
             init();
         }
-        digest.update(Integer.valueOf(bytesRead).byteValue());
+        digestProvider.update(Integer.valueOf(bytesRead).byteValue());
     }
 
-    public void verify(String checksum) throws IOException {
-        String encodedDigest = Base64.getEncoder().encodeToString(digest.digest());
-        if (!Objects.isNull(checksum) && !checksum.equals(encodedDigest)) {
+    public void verify(String expectedChecksum) throws IOException {
+        String calculatedChecksum = digestProvider.getDigest();
+
+        if (!Objects.equals(expectedChecksum, calculatedChecksum)) {
             log.error(
                     "Corrupted stream encountered, failed to verify checksum [{}] against calculated checksum [{}]",
-                    checksum,
-                    encodedDigest);
+                    expectedChecksum,
+                    calculatedChecksum);
             throw new IOException("Corrupted stream, verification of checksum failed");
         }
     }
 
     private void init() throws IOException {
-        try {
-            digest = MessageDigest.getInstance(digestAlgo);
-        } catch (NoSuchAlgorithmException e) {
-            throw new IOException("Invalid digest algorithm provided", e);
-        }
+        digestProvider = digestProviderService.getDigestProvider(digestAlgo);
     }
 
     public static class PartCheckerBuilder {
+
+        DigestProviderService digestProviderService = new DigestProviderService();
+
         public PartChecker build() {
             if (digestAlgo == null) {
                 digestAlgo = DEFAULT_DIGEST_ALGO;
             }
-            return new PartChecker(digestAlgo);
+            return new PartChecker(digestAlgo, digestProviderService);
         }
     }
 }

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/CRC32CProvider.java

@@ -0,0 +1,29 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.util.Base64;
+import java.util.zip.Checksum;
+import software.amazon.awssdk.crt.checksums.CRC32C;
+
+public class CRC32CProvider implements DigestProvider {
+
+    private final Checksum checksum;
+
+    public CRC32CProvider() {
+        checksum = new CRC32C();
+    }
+
+    @Override
+    public void update(byte singleByte) {
+        checksum.update(singleByte);
+    }
+
+    @Override
+    public String getDigest() {
+        long checksumValue = checksum.getValue();
+        return Base64.getEncoder().encodeToString(longToBytes(checksumValue));
+    }
+
+    private byte[] longToBytes(long value) {
+        return new byte[] {(byte) (value >>> 24), (byte) (value >>> 16), (byte) (value >>> 8), (byte) value};
+    }
+}

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/CRC32Provider.java

@@ -0,0 +1,30 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.util.Base64;
+import java.util.zip.Checksum;
+import software.amazon.awssdk.crt.checksums.CRC32;
+
+public class CRC32Provider implements DigestProvider {
+
+    private final Checksum checksum;
+
+    public CRC32Provider() {
+        checksum = new CRC32();
+    }
+
+    @Override
+    public void update(byte singleByte) {
+        checksum.update(singleByte);
+    }
+
+    @Override
+    public String getDigest() {
+        long checksumValue = checksum.getValue();
+
+        return Base64.getEncoder().encodeToString(longToBytes(checksumValue));
+    }
+
+    private byte[] longToBytes(long value) {
+        return new byte[] {(byte) (value >>> 24), (byte) (value >>> 16), (byte) (value >>> 8), (byte) value};
+    }
+}

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/CRC64NVMEProvider.java

@@ -0,0 +1,38 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.util.Base64;
+import java.util.zip.Checksum;
+import software.amazon.awssdk.crt.checksums.CRC64NVME;
+
+public class CRC64NVMEProvider implements DigestProvider {
+
+    private final Checksum checksum;
+
+    public CRC64NVMEProvider() {
+        this.checksum = new CRC64NVME();
+    }
+
+    @Override
+    public void update(byte singleByte) {
+        checksum.update(singleByte);
+    }
+
+    @Override
+    public String getDigest() {
+        long checksumValue = checksum.getValue();
+        return Base64.getEncoder().encodeToString(longToBytes(checksumValue));
+    }
+
+    private byte[] longToBytes(long value) {
+        return new byte[] {
+            (byte) (value >>> 56),
+            (byte) (value >>> 48),
+            (byte) (value >>> 40),
+            (byte) (value >>> 32),
+            (byte) (value >>> 24),
+            (byte) (value >>> 16),
+            (byte) (value >>> 8),
+            (byte) value
+        };
+    }
+}

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/DigestProvider.java

@@ -0,0 +1,10 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.io.IOException;
+
+public interface DigestProvider {
+
+    void update(byte singleByte);
+
+    String getDigest() throws IOException;
+}

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/DigestProviderService.java

@@ -0,0 +1,28 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.io.IOException;
+import java.security.NoSuchAlgorithmException;
+
+public class DigestProviderService {
+
+    public DigestProvider getDigestProvider(String digestAlgo) throws IOException {
+        try {
+            switch (digestAlgo) {
+                case "CRC32":
+                    return new CRC32Provider();
+                case "CRC32C":
+                    return new CRC32CProvider();
+                case "CRC64NVME":
+                    return new CRC64NVMEProvider();
+                case "SHA-1":
+                case "SHA-256":
+                case "MD5":
+                    return new SHAProvider(digestAlgo);
+                default:
+                    throw new IOException("Invalid digest algorithm provided");
+            }
+        } catch (NoSuchAlgorithmException e) {
+            throw new IOException("Invalid digest algorithm provided", e);
+        }
+    }
+}

src/main/java/io/github/jpmorganchase/fusion/digest/checksum/SHAProvider.java

@@ -0,0 +1,24 @@
+package io.github.jpmorganchase.fusion.digest.checksum;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+
+public class SHAProvider implements DigestProvider {
+
+    private final MessageDigest digest;
+
+    public SHAProvider(String algorithm) throws NoSuchAlgorithmException {
+        this.digest = MessageDigest.getInstance(algorithm);
+    }
+
+    @Override
+    public void update(byte singleByte) {
+        digest.update(singleByte);
+    }
+
+    @Override
+    public String getDigest() {
+        return Base64.getEncoder().encodeToString(digest.digest());
+    }
+}