Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,10 @@ private AccumulatingRead(
IOAutoCloseable onCloseCallback) {
super(rangeSpec, retryContext, onCloseCallback);
this.readId = readId;
this.hasher = hasher;
this.hasher =
(rangeSpec.begin() == 0)
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;
Comment on lines +153 to +156

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The condition rangeSpec.begin() == 0 is too broad for full object checksum validation. It will trigger for any partial read that starts at the beginning of the object (e.g., reading only the first 100 bytes). This results in unnecessary overhead and potential validation failures if the hasher attempts to compare a partial range's checksum against the full object's checksum. Consider using rangeSpec.isAll() to ensure validation is only enabled for full object reads.

Suggested change
this.hasher =
(rangeSpec.begin() == 0)
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;
this.hasher =
rangeSpec.isAll()
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;

this.complete = SettableApiFuture.create();
this.childRefs = Collections.synchronizedList(new ArrayList<>());
}
Expand Down Expand Up @@ -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)
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;
Comment on lines +286 to +289

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to the AccumulatingRead constructor, using rangeSpec.begin() == 0 here will enable CumulativeHasher for partial reads starting at offset 0. It is better to use rangeSpec.isAll() to target only full object reads.

Suggested change
this.hasher =
(rangeSpec.begin() == 0)
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;
this.hasher =
rangeSpec.isAll()
? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
: hasher;

this.closed = false;
this.failFuture = SettableApiFuture.create();
this.queue = new ArrayBlockingQueue<>(2);
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -91,7 +92,15 @@ final class GapicUnbufferedReadableByteChannel
this.result = result;
this.read = read;
this.req = req;
this.hasher = hasher;
this.hasher =
(req.getReadOffset() == 0)
? 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;
Expand Down Expand Up @@ -174,6 +183,7 @@ public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
}
if (take == EOF_MARKER) {
complete = true;
validateCumulativeChecksum();
break;
}

Expand Down Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading