feat(storage): add full object checksum validation for bidi flow (#13266)

Adding full object checksum for bidi flow

Refer to: go/full_checksum_java

---------

Co-authored-by: Dhriti Chopra <dhritichopra@google.com>

Author: Dhriti07

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseObjectReadSessionStreamRead.java

@@ -150,7 +150,10 @@ private AccumulatingRead(
         IOAutoCloseable onCloseCallback) {
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = readId;
-      this.hasher = hasher;
+      this.hasher =
+          (rangeSpec.begin() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+              ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
+              : hasher;
       this.complete = SettableApiFuture.create();
       this.childRefs = Collections.synchronizedList(new ArrayList<>());
     }
@@ -280,7 +283,10 @@ static class StreamingRead extends BaseObjectReadSessionStreamRead<ScatteringByt
         IOAutoCloseable onCloseCallback) {
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = new AtomicLong(readId);
-      this.hasher = hasher;
+      this.hasher =
+          (rangeSpec.begin() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+              ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
+              : hasher;
       this.closed = false;
       this.failFuture = SettableApiFuture.create();
       this.queue = new ArrayBlockingQueue<>(2);

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java

@@ -351,6 +351,12 @@ public void onResponse(BidiReadObjectResponse response) {
             executor.execute(
                 StorageException.liftToRunnable(
                     () -> {
+                      try {
+                        validateCumulativeChecksum(read);
+                      } catch (UncheckedCumulativeChecksumMismatchException e) {
+                        state.removeOutstandingReadOnFailure(id, read::fail).onFailure(e);
+                        return;
+                      }
                       read.eof();
                       // don't remove the outstanding read until the future has been resolved
                       state.removeOutstandingRead(id);
@@ -545,6 +551,15 @@ public void onComplete() {
     }
   }
 
+  private void validateCumulativeChecksum(ObjectReadSessionStreamRead<?> read) {
+    Hasher hasher = read.hasher();
+    if (hasher instanceof CumulativeHasher) {
+      CumulativeHasher cumulativeHasher = (CumulativeHasher) hasher;
+      com.google.storage.v2.Object metadata = state.getMetadata();
+      cumulativeHasher.validateCumulativeChecksum(metadata);
+    }
+  }
+
   static ObjectReadSessionStream create(
       ScheduledExecutorService executor,
       ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedReadableByteChannelTest.java

@@ -220,7 +220,6 @@ public void call(
     }
   }
 
-  @Test
   public void validateCumulativeChecksum_skippedForRangedRead() throws IOException {
     ChecksummedTestContent testContent =
         ChecksummedTestContent.of(DataGenerator.base64Characters().genBytes(10));