feat(storage): add full object checksum validation for grpc flow

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

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License 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 com.google.cloud.storage;
+
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.Object;
+import io.grpc.Status.Code;
+import java.nio.ByteBuffer;
+import java.util.Locale;
+import java.util.OptionalLong;
+import java.util.function.Supplier;
+
+/**
+ * A wrapper around hasher that accumulates checksums and validates them at the end of the read if
+ * it was a full object read.
+ */
+final class CumulativeHasher implements Hasher {
+  private final Hasher delegate;
+  private final long startOffset;
+  private final OptionalLong limit;
+  private Crc32cLengthKnown cumulativeHash;
+
+  CumulativeHasher(Hasher delegate, long startOffset, OptionalLong limit) {
+    this.delegate = delegate;
+    this.startOffset = startOffset;
+    this.limit = limit;
+    this.cumulativeHash = Crc32cValue.zero();
+  }
+
+  @Override
+  public Crc32cLengthKnown hash(ByteBuffer b) {
+    return delegate.hash(b);
+  }
+
+  @Override
+  public Crc32cLengthKnown hash(ByteString byteString) {
+    return delegate.hash(byteString);
+  }
+
+  @Override
+  public void validate(Crc32cValue<?> expected, Supplier<ByteBuffer> b)
+      throws ChecksumMismatchException {
+    ByteBuffer byteBuffer = b.get();
+    Crc32cLengthKnown actual = delegate.hash(byteBuffer);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new ChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public void validate(Crc32cValue<?> expected, ByteString byteString)
+      throws ChecksumMismatchException {
+    Crc32cLengthKnown actual = delegate.hash(byteString);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new ChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public void validateUnchecked(Crc32cValue<?> expected, ByteString byteString)
+      throws UncheckedChecksumMismatchException {
+    Crc32cLengthKnown actual = delegate.hash(byteString);
+    if (actual != null) {
+      if (expected != null && !actual.eqValue(expected)) {
+        throw new UncheckedChecksumMismatchException(expected, actual);
+      }
+      accumulate(actual);
+    }
+  }
+
+  @Override
+  public <C extends Crc32cValue<?>> C nullSafeConcat(C r1, Crc32cLengthKnown r2) {
+    return delegate.nullSafeConcat(r1, r2);
+  }
+
+  @Override
+  public Crc32cLengthKnown initialValue() {
+    return delegate.initialValue();
+  }
+
+  // Checks if it was a full object read.
+  boolean qualifiesForVerification(Object metadata) {
+    return startOffset == 0
+        && metadata != null
+        && metadata.hasChecksums()
+        && metadata.getChecksums().hasCrc32C()
+        && (!limit.isPresent() || limit.getAsLong() >= metadata.getSize());
+  }
+
+  void validateCumulativeChecksum(Object metadata)
+      throws UncheckedCumulativeChecksumMismatchException {
+    if (qualifiesForVerification(metadata)) {
+      Crc32cValue<?> expected = Crc32cValue.of(metadata.getChecksums().getCrc32C());
+      Crc32cLengthKnown actual = getCumulativeHash();
+      if (!actual.eqValue(expected)) {
+        throw new UncheckedCumulativeChecksumMismatchException(expected, actual);
+      }
+    }
+  }
+
+  private void accumulate(Crc32cLengthKnown actual) {
+    cumulativeHash = cumulativeHash.concat(actual);
+  }
+
+  Crc32cLengthKnown getCumulativeHash() {
+    return cumulativeHash;
+  }
+}
+
+final class UncheckedCumulativeChecksumMismatchException
+    extends com.google.api.gax.rpc.DataLossException {
+  private static final GrpcStatusCode STATUS_CODE = GrpcStatusCode.of(Code.DATA_LOSS);
+  private final Crc32cValue<?> expected;
+  private final Crc32cLengthKnown actual;
+
+  UncheckedCumulativeChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    super(
+        String.format(
+            Locale.US,
+            "Mismatch cumulative checksum value. Expected %s actual %s",
+            expected.debugString(),
+            actual.debugString()),
+        /* cause= */ null,
+        STATUS_CODE,
+        /* retryable= */ false);
+    this.expected = expected;
+    this.actual = actual;
+  }
+
+  Crc32cValue<?> getExpected() {
+    return expected;
+  }
+
+  Crc32cLengthKnown getActual() {
+    return actual;
+  }
+}

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

@@ -49,6 +49,7 @@
 import java.nio.channels.ScatteringByteChannel;
 import java.util.List;
 import java.util.Locale;
+import java.util.OptionalLong;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
@@ -91,7 +92,15 @@ final class GapicUnbufferedReadableByteChannel
     this.result = result;
     this.read = read;
     this.req = req;
-    this.hasher = hasher;
+    this.hasher =
+        (req.getReadOffset() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+            ? new CumulativeHasher(
+                hasher,
+                0,
+                req.getReadLimit() <= 0
+                    ? OptionalLong.empty()
+                    : OptionalLong.of(req.getReadLimit()))
+            : hasher;
     this.fetchOffset = new AtomicLong(req.getReadOffset());
     this.blobOffset = req.getReadOffset();
     this.retrier = retrier;
@@ -174,6 +183,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
       }
       if (take == EOF_MARKER) {
         complete = true;
+        validateCumulativeChecksum();
         break;
       }
 
@@ -311,6 +321,17 @@ private IOException createError(String message) throws IOException {
     return new IOException(message, cause);
   }
 
+  private void validateCumulativeChecksum() throws IOException {
+    if (hasher instanceof CumulativeHasher) {
+      CumulativeHasher cumulativeHasher = (CumulativeHasher) hasher;
+      try {
+        cumulativeHasher.validateCumulativeChecksum(metadata);
+      } catch (UncheckedCumulativeChecksumMismatchException exception) {
+        throw new IOException(StorageException.coalesce(exception));
+      }
+    }
+  }
+
   private final class ReadObjectObserver extends StateCheckingResponseObserver<ReadObjectResponse> {
 
     private final SettableApiFuture<Void> open = SettableApiFuture.create();

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

@@ -212,7 +212,7 @@ final class ChecksumMismatchException extends IOException {
     private final Crc32cValue<?> expected;
     private final Crc32cLengthKnown actual;
 
-    private ChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    ChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
       super(
           String.format(
               Locale.US,
@@ -237,7 +237,7 @@ final class UncheckedChecksumMismatchException extends DataLossException {
     private final Crc32cValue<?> expected;
     private final Crc32cLengthKnown actual;
 
-    private UncheckedChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
+    UncheckedChecksumMismatchException(Crc32cValue<?> expected, Crc32cLengthKnown actual) {
       super(
           String.format(
               "Mismatch checksum value. Expected %s actual %s",