feat(storage): Add full object checksum validation for appendable uploads

google/cloud/storage/async/writer.cc

@@ -71,20 +71,31 @@ future<StatusOr<AsyncToken>> AsyncWriter::Write(AsyncToken token,
 
 future<StatusOr<google::storage::v2::Object>> AsyncWriter::Finalize(
     AsyncToken token, WritePayload payload) {
+  return Finalize(std::move(token), std::move(payload), absl::nullopt);
+}
+
+future<StatusOr<google::storage::v2::Object>> AsyncWriter::Finalize(
+    AsyncToken token, WritePayload payload,
+    absl::optional<Crc32cChecksumValue> const& expected_checksum) {
   if (!impl_) return StreamError<google::storage::v2::Object>(GCP_ERROR_INFO());
   auto t = storage_internal::MakeAsyncToken(impl_.get());
   if (token != t) {
     return TokenError<google::storage::v2::Object>(GCP_ERROR_INFO());
   }
 
-  return impl_->Finalize(std::move(payload)).then([impl = impl_](auto f) {
-    return f.get();
-  });
+  return impl_->Finalize(std::move(payload), expected_checksum)
+      .then([impl = impl_](auto f) { return f.get(); });
 }
 
 future<StatusOr<google::storage::v2::Object>> AsyncWriter::Finalize(
     AsyncToken token) {
-  return Finalize(std::move(token), WritePayload{});
+  return Finalize(std::move(token), WritePayload{}, absl::nullopt);
+}
+
+future<StatusOr<google::storage::v2::Object>> AsyncWriter::Finalize(
+    AsyncToken token,
+    absl::optional<Crc32cChecksumValue> const& expected_checksum) {
+  return Finalize(std::move(token), WritePayload{}, expected_checksum);
 }
 
 future<Status> AsyncWriter::Flush() {

google/cloud/storage/async/writer.h

@@ -18,6 +18,7 @@
 #include "google/cloud/storage/async/token.h"
 #include "google/cloud/storage/async/write_payload.h"
 #include "google/cloud/storage/async/writer_connection.h"
+#include "google/cloud/storage/hashing_options.h"
 #include "google/cloud/future.h"
 #include "google/cloud/status_or.h"
 #include "google/cloud/version.h"
@@ -114,14 +115,32 @@ class AsyncWriter {
   /// Upload @p payload returning a new token to continue the upload.
   future<StatusOr<AsyncToken>> Write(AsyncToken token, WritePayload payload);
 
-  /// Finalize the upload with the existing data.
+  /**
+   * Finalize the upload with the existing data.
+   *
+   * @warning For appendable uploads, the SDK does not incrementally compute
+   * full-object hashes. You *must* calculate and inject your own expected
+   * checksum here to get end-to-end data integrity. If you omit it, the object
+   * will finalize without full-object verification.
+   */
   future<StatusOr<google::storage::v2::Object>> Finalize(AsyncToken token);
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      AsyncToken token,
+      absl::optional<Crc32cChecksumValue> const& expected_checksum);
 
   /**
    * Upload @p payload and then finalize the upload.
+   *
+   * @warning For appendable uploads, the SDK does not incrementally compute
+   * full-object hashes. You *must* calculate and inject your own expected
+   * checksum here to get end-to-end data integrity. If you omit it, the object
+   * will finalize without full-object verification.
    */
   future<StatusOr<google::storage::v2::Object>> Finalize(AsyncToken token,
                                                          WritePayload payload);
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      AsyncToken token, WritePayload payload,
+      absl::optional<Crc32cChecksumValue> const& expected_checksum);
 
   /**
    * Flush any buffered data to the service.

google/cloud/storage/async/writer_connection.h

@@ -16,6 +16,7 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_ASYNC_WRITER_CONNECTION_H
 
 #include "google/cloud/storage/async/write_payload.h"
+#include "google/cloud/storage/hashing_options.h"
 #include "google/cloud/storage/object_metadata.h"
 #include "google/cloud/future.h"
 #include "google/cloud/rpc_metadata.h"
@@ -112,6 +113,11 @@ class AsyncWriterConnection {
   /// Finalizes an upload.
   virtual future<StatusOr<google::storage::v2::Object>> Finalize(
       WritePayload) = 0;
+  virtual future<StatusOr<google::storage::v2::Object>> Finalize(
+      WritePayload p,
+      absl::optional<Crc32cChecksumValue> const& /*expected_checksum*/) {
+    return Finalize(std::move(p));
+  }
 
   /// Uploads some data to the service and flushes the value.
   virtual future<Status> Flush(WritePayload payload) = 0;

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

@@ -79,7 +79,34 @@ inline std::unique_ptr<storage::AsyncIdempotencyPolicy> idempotency_policy(
   return options.get<storage::AsyncIdempotencyPolicyOption>()();
 }
 
-std::unique_ptr<storage::internal::HashFunction> CreateHashFunction(
+std::unique_ptr<storage::internal::HashFunction> CreateWriterHashFunction(
+    Options const& options) {
+  auto crc32c = std::unique_ptr<storage::internal::HashFunction>();
+  if (options.has<storage::UseCrc32cValueOption>()) {
+    crc32c = std::make_unique<storage::internal::PrecomputedHashFunction>(
+        storage::internal::HashValues{
+            Crc32cFromProto(options.get<storage::UseCrc32cValueOption>()),
+            /*.md5=*/{}});
+  }
+
+  auto md5 = std::unique_ptr<storage::internal::HashFunction>();
+  if (options.has<storage::UseMD5ValueOption>()) {
+    md5 = std::make_unique<storage::internal::PrecomputedHashFunction>(
+        storage::internal::HashValues{
+            /*.crc32=*/{},
+            MD5FromProto(options.get<storage::UseMD5ValueOption>())});
+  }
+
+  if (crc32c && md5) {
+    return std::make_unique<storage::internal::CompositeFunction>(
+        std::move(crc32c), std::move(md5));
+  }
+  if (crc32c) return crc32c;
+  if (md5) return md5;
+  return storage::internal::CreateNullHashFunction();
+}
+
+std::unique_ptr<storage::internal::HashFunction> CreateReaderHashFunction(
     Options const& options) {
   auto crc32c = std::unique_ptr<storage::internal::HashFunction>();
   if (options.has<storage::UseCrc32cValueOption>()) {
@@ -125,19 +152,12 @@ StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> MakeAppendableWriter(
 
   if (rpc->first_response.has_resource()) {
     auto const& resource = rpc->first_response.resource();
-    if (current->get<storage::EnableCrc32cValidationOption>() &&
-        resource.has_checksums() && resource.checksums().has_crc32c()) {
-      hash = std::make_shared<
-          ::google::cloud::storage::internal::Crc32cHashFunction>(
-          resource.checksums().crc32c(), resource.size());
-    } else {
-      hash = CreateHashFunction(*current);
-    }
+    hash = CreateWriterHashFunction(*current);
     impl = std::make_unique<AsyncWriterConnectionImpl>(
         current, request, std::move(rpc->stream), hash, resource, false);
   } else {
     persisted_size = rpc->first_response.persisted_size();
-    hash = CreateHashFunction(*current);
+    hash = CreateWriterHashFunction(*current);
     auto checksums = rpc->first_response.has_persisted_data_checksums()
                          ? absl::make_optional(
                                rpc->first_response.persisted_data_checksums())
@@ -208,7 +228,7 @@ future<StatusOr<google::storage::v2::Object>> AsyncConnectionImpl::InsertObject(
         options->get<storage::TransferStallMinimumRateOption>(),
         google::storage::v2::ServiceConstants::MAX_WRITE_CHUNK_BYTES);
 
-    auto hash_function = CreateHashFunction(*options);
+    auto hash_function = CreateWriterHashFunction(*options);
     ApplyRoutingHeaders(*context, request.write_object_spec());
     AddIdempotencyToken(*context, id);
     auto rpc = stub->AsyncWriteObject(cq, std::move(context), options);
@@ -314,7 +334,7 @@ AsyncConnectionImpl::ReadObject(ReadObjectParams p) {
   // this function.
   auto hash_function =
       std::make_shared<storage::internal::Crc32cMessageHashFunction>(
-          CreateHashFunction(*current));
+          CreateReaderHashFunction(*current));
   auto hash_validator = CreateHashValidator(p.request, *current);
 
   absl::optional<std::int64_t> requested_length;
@@ -730,7 +750,7 @@ AsyncConnectionImpl::StartUnbufferedUploadImpl(
         StatusOr<std::unique_ptr<storage::AsyncWriterConnection>>(
             std::move(response).status()));
   }
-  auto hash_function = CreateHashFunction(*current);
+  auto hash_function = CreateWriterHashFunction(*current);
   auto configure =
       [current, upload = response->upload_id()](grpc::ClientContext& context) {
         ApplyResumableUploadRoutingHeader(context, upload);
@@ -768,7 +788,7 @@ AsyncConnectionImpl::ResumeUnbufferedUploadImpl(
   // the upload resumes from the beginning of the file.
   auto hash_function = storage::internal::CreateNullHashFunction();
   if (response->persisted_size() == 0) {
-    hash_function = CreateHashFunction(*current);
+    hash_function = CreateWriterHashFunction(*current);
   }
   auto configure =
       [current, upload_id = query.upload_id()](grpc::ClientContext& context) {