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

Author: v-pratap

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

@@ -26,9 +26,12 @@ 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_(std::move(hash_function)) {}
 
 StatusOr<storage::internal::HttpResponse> GrpcObjectReadSource::Close() {
   if (stream_) stream_ = nullptr;
@@ -74,18 +77,30 @@ 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()) return handle_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 +130,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

@@ -21,7 +21,9 @@
 #include "google/cloud/future.h"
 #include "google/cloud/internal/streaming_read_rpc.h"
 #include "absl/functional/function_ref.h"
+#include "google/cloud/storage/internal/hash_function.h"
 #include "google/storage/v2/storage.pb.h"
+#include "absl/types/optional.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

@@ -18,6 +18,7 @@
 #include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
 #include "google/cloud/storage/testing/mock_storage_stub.h"
 #include "google/cloud/testing_util/status_matchers.h"
+#include "google/cloud/storage/internal/hash_function_impl.h"
 #include <google/protobuf/text_format.h>
 #include <google/protobuf/util/message_differencer.h>
 #include <gmock/gmock.h>
@@ -350,6 +351,27 @@ 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")));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

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

@@ -31,6 +31,7 @@
 #include "google/cloud/storage/internal/grpc/synthetic_self_link.h"
 #include "google/cloud/storage/internal/storage_stub_factory.h"
 #include "google/cloud/storage/options.h"
+#include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/internal/big_endian.h"
 #include "google/cloud/internal/invoke_result.h"
 #include "google/cloud/internal/make_status.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(