feat(storage): Add read chunkwise checksum validation for sync grpc read (#16107)

* feat(storage): Add read chunkwise checksum validation for sync grpc read

* resolve the ai comment

* add unit tests

Author: v-pratap

google/cloud/storage/internal/grpc/object_read_source.cc

@@ -26,9 +26,14 @@ namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
-GrpcObjectReadSource::GrpcObjectReadSource(TimerSource timer_source,
-                                           std::unique_ptr<StreamingRpc> stream)
-    : timer_source_(std::move(timer_source)), stream_(std::move(stream)) {}
+GrpcObjectReadSource::GrpcObjectReadSource(
+    TimerSource timer_source, std::unique_ptr<StreamingRpc> stream,
+    std::shared_ptr<storage::internal::HashFunction> hash_function)
+    : timer_source_(std::move(timer_source)),
+      stream_(std::move(stream)),
+      hash_function_(hash_function
+                         ? std::move(hash_function)
+                         : storage::internal::CreateNullHashFunction()) {}
 
 StatusOr<storage::internal::HttpResponse> GrpcObjectReadSource::Close() {
   if (stream_) stream_ = nullptr;
@@ -74,18 +79,33 @@ StatusOr<storage::internal::ReadSourceResult> GrpcObjectReadSource::Read(
       if (!status_.ok()) return status_;
       return result;
     }
-    HandleResponse(result, buf, n, std::move(response));
+    auto handle_status = HandleResponse(result, buf, n, std::move(response));
+    if (!handle_status.ok()) {
+      status_ = handle_status;
+      stream_.reset();
+      return status_;
+    }
   }
 
   return result;
 }
 
-void GrpcObjectReadSource::HandleResponse(
+Status GrpcObjectReadSource::HandleResponse(
     storage::internal::ReadSourceResult& result, char* buf, std::size_t n,
     google::storage::v2::ReadObjectResponse response) {
+  if (!offset_ && response.has_content_range()) {
+    offset_ = response.content_range().start();
+  }
   // The google.storage.v1.Storage documentation says this field can be
   // empty.
   if (response.has_checksummed_data()) {
+    auto const& data = response.checksummed_data();
+    auto status = hash_function_->Update(offset_.value_or(0), GetContent(data),
+                                         data.crc32c());
+    if (!status.ok()) return status;
+
+    offset_ = offset_.value_or(0) + GetContent(data).size();
+
     auto const offset = result.bytes_received;
     result.bytes_received += buffer_.HandleResponse(
         buf + offset, n - offset,
@@ -115,6 +135,7 @@ void GrpcObjectReadSource::HandleResponse(
         result.storage_class.value_or(metadata.storage_class());
     result.size = result.size.value_or(metadata.size());
   }
+  return {};
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/grpc/object_read_source.h

@@ -16,11 +16,13 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_GRPC_OBJECT_READ_SOURCE_H
 
 #include "google/cloud/storage/internal/grpc/buffer_read_object_data.h"
+#include "google/cloud/storage/internal/hash_function.h"
 #include "google/cloud/storage/internal/object_read_source.h"
 #include "google/cloud/storage/version.h"
 #include "google/cloud/future.h"
 #include "google/cloud/internal/streaming_read_rpc.h"
 #include "absl/functional/function_ref.h"
+#include "absl/types/optional.h"
 #include "google/storage/v2/storage.pb.h"
 #include <functional>
 #include <string>
@@ -49,8 +51,10 @@ class GrpcObjectReadSource : public storage::internal::ObjectReadSource {
   // `false` if the timer was canceled, and with `true` if the timer triggered.
   using TimerSource = std::function<future<bool>()>;
 
-  explicit GrpcObjectReadSource(TimerSource timer_source,
-                                std::unique_ptr<StreamingRpc> stream);
+  explicit GrpcObjectReadSource(
+      TimerSource timer_source, std::unique_ptr<StreamingRpc> stream,
+      std::shared_ptr<storage::internal::HashFunction> hash_function =
+          storage::internal::CreateNullHashFunction());
 
   ~GrpcObjectReadSource() override = default;
 
@@ -65,12 +69,14 @@ class GrpcObjectReadSource : public storage::internal::ObjectReadSource {
                                                      std::size_t n) override;
 
  private:
-  void HandleResponse(storage::internal::ReadSourceResult& result, char* buf,
-                      std::size_t n,
-                      google::storage::v2::ReadObjectResponse response);
+  Status HandleResponse(storage::internal::ReadSourceResult& result, char* buf,
+                        std::size_t n,
+                        google::storage::v2::ReadObjectResponse response);
 
   TimerSource timer_source_;
   std::unique_ptr<StreamingRpc> stream_;
+  std::shared_ptr<storage::internal::HashFunction> hash_function_;
+  absl::optional<std::int64_t> offset_;
 
   // In some cases the gRPC response may contain more data than the buffer
   // provided by the application. This buffer stores any excess results.

google/cloud/storage/internal/grpc/object_read_source_test.cc

@@ -16,6 +16,7 @@
 #include "google/cloud/storage/hashing_options.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
 #include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
+#include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/storage/testing/mock_storage_stub.h"
 #include "google/cloud/testing_util/status_matchers.h"
 #include <google/protobuf/text_format.h>
@@ -350,6 +351,72 @@ TEST(GrpcObjectReadSource, StallTimeout) {
   EXPECT_THAT(response, StatusIs(StatusCode::kDeadlineExceeded));
 }
 
+TEST(GrpcObjectReadSource, ChecksumMismatch) {
+  auto mock = std::make_unique<MockObjectMediaStream>();
+  std::string const contents = "0123456789";
+
+  ::testing::InSequence sequence;
+  EXPECT_CALL(*mock, Read)
+      .WillOnce([&contents](storage_proto::ReadObjectResponse* r) {
+        SetContent(*r, contents);
+        r->mutable_checksummed_data()->set_crc32c(123);  // Wrong CRC
+        return absl::nullopt;
+      });
+
+  auto hash_function =
+      std::make_shared<storage::internal::Crc32cMessageHashFunction>(
+          storage::internal::CreateNullHashFunction());
+
+  GrpcObjectReadSource tested(MakeSimpleTimerSource(), std::move(mock),
+                              std::move(hash_function));
+  std::vector<char> buffer(1024);
+  auto response = tested.Read(buffer.data(), buffer.size());
+  EXPECT_THAT(response, StatusIs(StatusCode::kInvalidArgument,
+                                 HasSubstr("mismatched crc32c checksum")));
+  EXPECT_THAT(tested.Close(),
+              StatusIs(StatusCode::kInvalidArgument,
+                       HasSubstr("mismatched crc32c checksum")));
+}
+
+TEST(GrpcObjectReadSource, ChecksumWithNonZeroOffset) {
+  auto mock = std::make_unique<MockObjectMediaStream>();
+  std::string const contents1 = "0123456789";
+  std::string const contents2 = "abcdefghij";
+  auto const expected_crc32c_1 = storage::ComputeCrc32cChecksum(contents1);
+  auto const expected_crc32c_2 = storage::ComputeCrc32cChecksum(contents2);
+
+  ::testing::InSequence sequence;
+  EXPECT_CALL(*mock, Read)
+      .WillOnce([&contents1,
+                 expected_crc32c_1](storage_proto::ReadObjectResponse* r) {
+        SetContent(*r, contents1);
+        r->mutable_content_range()->set_start(100);
+        r->mutable_checksummed_data()->set_crc32c(
+            storage_internal::Crc32cToProto(expected_crc32c_1).value());
+        return absl::nullopt;
+      })
+      .WillOnce([&contents2,
+                 expected_crc32c_2](storage_proto::ReadObjectResponse* r) {
+        SetContent(*r, contents2);
+        r->mutable_checksummed_data()->set_crc32c(
+            storage_internal::Crc32cToProto(expected_crc32c_2).value());
+        return absl::nullopt;
+      })
+      .WillOnce(Return(Status{}));
+  EXPECT_CALL(*mock, GetRequestMetadata).WillOnce(Return(RpcMetadata{}));
+
+  auto hash_function =
+      std::make_shared<storage::internal::Crc32cMessageHashFunction>(
+          storage::internal::CreateNullHashFunction());
+
+  GrpcObjectReadSource tested(MakeSimpleTimerSource(), std::move(mock),
+                              std::move(hash_function));
+  std::vector<char> buffer(1024);
+  auto response = tested.Read(buffer.data(), buffer.size());
+  ASSERT_STATUS_OK(response);
+  EXPECT_EQ(contents1.size() + contents2.size(), response->bytes_received);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

google/cloud/storage/internal/grpc/stub.cc

@@ -29,6 +29,7 @@
 #include "google/cloud/storage/internal/grpc/sign_blob_request_parser.h"
 #include "google/cloud/storage/internal/grpc/split_write_object_data.h"
 #include "google/cloud/storage/internal/grpc/synthetic_self_link.h"
+#include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/storage/internal/storage_stub_factory.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/internal/big_endian.h"
@@ -445,9 +446,14 @@ GrpcStub::ReadObject(rest_internal::RestContext& context,
     };
   }
 
+  auto hash_function =
+      std::make_shared<storage::internal::Crc32cMessageHashFunction>(
+          storage::internal::CreateHashFunction(request));
+
   return std::unique_ptr<storage::internal::ObjectReadSource>(
       std::make_unique<GrpcObjectReadSource>(std::move(timer_source),
-                                             std::move(stream)));
+                                             std::move(stream),
+                                             std::move(hash_function)));
 }
 
 StatusOr<storage::internal::ListObjectsResponse> GrpcStub::ListObjects(