feat(storage): Add full object read checksum validation for Open (googleapis#16120)

* feat(storage): Add full object read checksum validation for Open

* remove unnecesary changes

* remove unwanted changes

* fix the format

* Add md5 validation as well for full object read checksum

* correct the comments

* Refactor ObjectDescriptorImpl::Read to reduce cognitive complexity and resolve clang-tidy issues

Author: v-pratap

google/cloud/storage/internal/async/object_descriptor_impl.cc

@@ -17,8 +17,12 @@
 #include "google/cloud/storage/internal/async/handle_redirect_error.h"
 #include "google/cloud/storage/internal/async/multi_stream_manager.h"
 #include "google/cloud/storage/internal/async/object_descriptor_reader_tracing.h"
+#include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
 #include "google/cloud/storage/internal/hash_function.h"
 #include "google/cloud/storage/internal/hash_function_impl.h"
+#include "google/cloud/storage/internal/hash_validator.h"
+#include "google/cloud/storage/internal/hash_validator_impl.h"
+#include "google/cloud/storage/internal/hash_values.h"
 #include "google/cloud/grpc_error_delegate.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/rpc/status.pb.h"
@@ -149,15 +153,20 @@ void ObjectDescriptorImpl::MakeSubsequentStream() {
 
 std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
     ReadParams p) {
-  std::shared_ptr<storage::internal::HashFunction> hash_function =
-      std::shared_ptr<storage::internal::HashFunction>(
-          storage::internal::CreateNullHashFunction());
-  if (options_.has<storage::EnableCrc32cValidationOption>()) {
-    hash_function =
-        std::make_shared<storage::internal::Crc32cMessageHashFunction>(
-            storage::internal::CreateNullHashFunction());
-  }
-  auto range = std::make_shared<ReadRange>(p.start, p.length, hash_function);
+  // Full-object checksum validation (both CRC32C and MD5) is only supported for
+  // full-object reads (starting at offset 0 and reading the entire object).
+  //
+  // Note that MD5 validation is not supported for partial/ranged reads because
+  // GCS does not compute or send chunk-level MD5 checksums (unlike CRC32C,
+  // which is validated per-chunk on the gRPC layer).
+  bool is_full_read = (p.start == 0 && metadata_.has_value() &&
+                       (p.length == 0 || p.length >= metadata_->size()));
+
+  auto hash_function = CreateHashFunction(is_full_read);
+  auto hash_validator = CreateHashValidator(is_full_read);
+
+  auto range = std::make_shared<ReadRange>(p.start, p.length, hash_function,
+                                           std::move(hash_validator));
 
   std::unique_lock<std::mutex> lk(mu_);
   if (stream_manager_->Empty()) {
@@ -188,6 +197,80 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
   return MakeTracingObjectDescriptorReader(std::move(range));
 }
 
+std::shared_ptr<storage::internal::HashFunction>
+ObjectDescriptorImpl::CreateHashFunction(bool is_full_read) const {
+  auto const enable_crc32c =
+      options_.get<storage::EnableCrc32cValidationOption>();
+  auto const enable_md5 = options_.get<storage::EnableMD5ValidationOption>();
+
+  if (enable_crc32c) {
+    std::unique_ptr<storage::internal::HashFunction> child;
+    if (is_full_read) {
+      if (enable_md5) {
+        child = std::make_unique<storage::internal::CompositeFunction>(
+            std::make_unique<storage::internal::Crc32cHashFunction>(),
+            storage::internal::MD5HashFunction::Create());
+      } else {
+        child = std::make_unique<storage::internal::Crc32cHashFunction>();
+      }
+    } else {
+      child = storage::internal::CreateNullHashFunction();
+    }
+    return std::make_shared<storage::internal::Crc32cMessageHashFunction>(
+        std::move(child));
+  }
+  if (enable_md5 && is_full_read) {
+    return std::shared_ptr<storage::internal::HashFunction>(
+        storage::internal::MD5HashFunction::Create());
+  }
+  return std::shared_ptr<storage::internal::HashFunction>(
+      storage::internal::CreateNullHashFunction());
+}
+
+std::unique_ptr<storage::internal::HashValidator>
+ObjectDescriptorImpl::CreateHashValidator(bool is_full_read) const {
+  if (!is_full_read) {
+    return storage::internal::CreateNullHashValidator();
+  }
+
+  auto const enable_crc32c =
+      options_.get<storage::EnableCrc32cValidationOption>();
+  auto const enable_md5 = options_.get<storage::EnableMD5ValidationOption>();
+
+  std::unique_ptr<storage::internal::HashValidator> hash_validator;
+  if (enable_crc32c && enable_md5) {
+    hash_validator = std::make_unique<storage::internal::CompositeValidator>(
+        std::make_unique<storage::internal::Crc32cHashValidator>(),
+        std::make_unique<storage::internal::MD5HashValidator>());
+  } else if (enable_crc32c) {
+    hash_validator = std::make_unique<storage::internal::Crc32cHashValidator>();
+  } else if (enable_md5) {
+    hash_validator = std::make_unique<storage::internal::MD5HashValidator>();
+  } else {
+    return storage::internal::CreateNullHashValidator();
+  }
+
+  // Process the expected hashes from metadata
+  storage::internal::HashValues hashes;
+  if (metadata_->has_checksums()) {
+    auto const& checksums = metadata_->checksums();
+    if (checksums.has_crc32c()) {
+      hashes =
+          Merge(std::move(hashes),
+                storage::internal::HashValues{
+                    storage_internal::Crc32cFromProto(checksums.crc32c()), {}});
+    }
+    if (!checksums.md5_hash().empty()) {
+      hashes =
+          Merge(std::move(hashes),
+                storage::internal::HashValues{
+                    {}, storage_internal::MD5FromProto(checksums.md5_hash())});
+    }
+  }
+  hash_validator->ProcessHashValues(hashes);
+  return hash_validator;
+}
+
 void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
                                  StreamIterator it) {
   if (it->stream->write_pending ||

google/cloud/storage/internal/async/object_descriptor_impl.h

@@ -111,6 +111,11 @@ class ObjectDescriptorImpl
   bool IsResumable(StreamIterator it, Status const& status,
                    google::rpc::Status const& proto_status);
 
+  std::shared_ptr<storage::internal::HashFunction> CreateHashFunction(
+      bool is_full_read) const;
+  std::unique_ptr<storage::internal::HashValidator> CreateHashValidator(
+      bool is_full_read) const;
+
   std::unique_ptr<storage::ResumePolicy> resume_policy_prototype_;
   OpenStreamFactory make_stream_;