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

Author: v-pratap

google/cloud/storage/async/writer.cc

@@ -70,21 +70,22 @@ future<StatusOr<AsyncToken>> AsyncWriter::Write(AsyncToken token,
 }
 
 future<StatusOr<google::storage::v2::Object>> AsyncWriter::Finalize(
-    AsyncToken token, WritePayload payload) {
+    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{});
+    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"
@@ -115,13 +116,18 @@ class AsyncWriter {
   future<StatusOr<AsyncToken>> Write(AsyncToken token, WritePayload payload);
 
   /// Finalize the upload with the existing data.
-  future<StatusOr<google::storage::v2::Object>> Finalize(AsyncToken token);
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      AsyncToken token,
+      absl::optional<Crc32cChecksumValue> const& expected_checksum =
+          absl::nullopt);
 
   /**
    * Upload @p payload and then finalize the upload.
    */
-  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 =
+          absl::nullopt);
 
   /**
    * 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

@@ -137,7 +137,16 @@ StatusOr<std::unique_ptr<storage::AsyncWriterConnection>> MakeAppendableWriter(
         current, request, std::move(rpc->stream), hash, resource, false);
   } else {
     persisted_size = rpc->first_response.persisted_size();
-    hash = CreateHashFunction(*current);
+    if (current->get<storage::EnableCrc32cValidationOption>() &&
+        rpc->first_response.has_persisted_data_checksums() &&
+        rpc->first_response.persisted_data_checksums().has_crc32c()) {
+      hash = std::make_shared<
+          ::google::cloud::storage::internal::Crc32cHashFunction>(
+          rpc->first_response.persisted_data_checksums().crc32c(),
+          persisted_size);
+    } else {
+      hash = CreateHashFunction(*current);
+    }
     auto checksums = rpc->first_response.has_persisted_data_checksums()
                          ? absl::make_optional(
                                rpc->first_response.persisted_data_checksums())

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

@@ -106,6 +106,11 @@ class AsyncWriterConnectionBufferedState
     return Impl(std::unique_lock<std::mutex>(mu_))->PersistedState();
   }
 
+  absl::optional<google::storage::v2::ObjectChecksums> PersistedChecksums()
+      const {
+    return Impl(std::unique_lock<std::mutex>(mu_))->PersistedChecksums();
+  }
+
   future<Status> Write(storage::WritePayload const& p) {
     std::unique_lock<std::mutex> lk(mu_);
     resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
@@ -114,7 +119,15 @@ class AsyncWriterConnectionBufferedState
 
   future<StatusOr<google::storage::v2::Object>> Finalize(
       storage::WritePayload const& p) {
+    return Finalize(p, absl::nullopt);
+  }
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      storage::WritePayload const& p,
+      absl::optional<storage::Crc32cChecksumValue> const& expected_checksum) {
     std::unique_lock<std::mutex> lk(mu_);
+    if (expected_checksum.has_value()) {
+      expected_checksum_ = expected_checksum;
+    }
     resend_buffer_.Append(WritePayloadImpl::GetImpl(p));
     finalize_ = true;
     HandleNewData(std::move(lk), true);
@@ -246,7 +259,7 @@ class AsyncWriterConnectionBufferedState
     auto impl = Impl(lk);
     lk.unlock();
     // Finalize with an empty payload.
-    (void)impl->Finalize(storage::WritePayload{})
+    (void)impl->Finalize(storage::WritePayload{}, expected_checksum_)
         .then([w = WeakFromThis()](auto f) {
           if (auto self = w.lock()) return self->OnFinalize(f.get());
         });
@@ -636,6 +649,7 @@ class AsyncWriterConnectionBufferedState
   // Retrieve the future in the constructor, as some operations reset
   // finalized_.
   future<StatusOr<google::storage::v2::Object>> finalized_future_;
+  absl::optional<storage::Crc32cChecksumValue> expected_checksum_;
 
   // The result of calling `Close()`. Note that only one such call is ever
   // made.
@@ -763,13 +777,24 @@ class AsyncWriterConnectionBuffered : public storage::AsyncWriterConnection {
     return state_->PersistedState();
   }
 
+  absl::optional<google::storage::v2::ObjectChecksums> PersistedChecksums()
+      const override {
+    return state_->PersistedChecksums();
+  }
+
   future<Status> Write(storage::WritePayload p) override {
     return state_->Write(std::move(p));
   }
 
   future<StatusOr<google::storage::v2::Object>> Finalize(
       storage::WritePayload p) override {
-    return state_->Finalize(std::move(p));
+    return Finalize(std::move(p), absl::nullopt);
+  }
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      storage::WritePayload p,
+      absl::optional<storage::Crc32cChecksumValue> const& expected_checksum)
+      override {
+    return state_->Finalize(std::move(p), expected_checksum);
   }
 
   future<Status> Flush(storage::WritePayload p) override {

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

@@ -140,22 +140,48 @@ future<Status> AsyncWriterConnectionImpl::Write(storage::WritePayload payload) {
 }
 
 future<StatusOr<google::storage::v2::Object>>
-AsyncWriterConnectionImpl::Finalize(storage::WritePayload payload) {
+AsyncWriterConnectionImpl::Finalize(
+    storage::WritePayload payload,
+    absl::optional<storage::Crc32cChecksumValue> const& expected_checksum) {
   auto write = MakeRequest();
   write.set_finish_write(true);
 
   auto p = WritePayloadImpl::GetImpl(payload);
   auto size = p.size();
+
+  if (p.empty() && expected_checksum.has_value()) {
+    auto const actual = hash_function_->Finish().crc32c;
+    if (!actual.empty() && expected_checksum->value() != actual) {
+      return make_ready_future(StatusOr<google::storage::v2::Object>(
+          google::cloud::internal::DataLossError(
+              "client checksum mismatch: expected " +
+                  expected_checksum->value() + " got " + actual,
+              GCP_ERROR_INFO())));
+    }
+  }
+
   auto action = request_.has_append_object_spec() ||
                         request_.write_object_spec().appendable()
                     ? PartialUpload::kFinalize
                     : PartialUpload::kFinalizeWithChecksum;
   auto coro = PartialUpload::Call(impl_, hash_function_, std::move(write),
                                   std::move(p), std::move(action));
-  return coro->Start().then([coro, size, this](auto f) mutable {
-    coro.reset();  // breaks the cycle between the completion queue and coro
-    return OnFinalUpload(size, f.get());
-  });
+  return coro->Start().then(
+      [coro, size, expected_checksum, this](auto f) mutable {
+        coro.reset();  // breaks the cycle between the completion queue and coro
+        auto res = f.get();
+        if (res.ok() && *res && expected_checksum.has_value()) {
+          auto const actual = hash_function_->Finish().crc32c;
+          if (!actual.empty() && expected_checksum->value() != actual) {
+            return make_ready_future(StatusOr<google::storage::v2::Object>(
+                google::cloud::internal::DataLossError(
+                    "client checksum mismatch: expected " +
+                        expected_checksum->value() + " got " + actual,
+                    GCP_ERROR_INFO())));
+          }
+        }
+        return OnFinalUpload(size, std::move(res));
+      });
 }
 
 future<Status> AsyncWriterConnectionImpl::Flush(storage::WritePayload payload) {

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

@@ -69,7 +69,13 @@ class AsyncWriterConnectionImpl : public storage::AsyncWriterConnection {
 
   future<Status> Write(storage::WritePayload payload) override;
   future<StatusOr<google::storage::v2::Object>> Finalize(
-      storage::WritePayload) override;
+      storage::WritePayload payload) override {
+    return Finalize(std::move(payload), absl::nullopt);
+  }
+  future<StatusOr<google::storage::v2::Object>> Finalize(
+      storage::WritePayload payload,
+      absl::optional<storage::Crc32cChecksumValue> const& expected_checksum)
+      override;
   future<Status> Flush(storage::WritePayload payload) override;
   future<Status> Close(storage::WritePayload payload) override;
   future<StatusOr<std::int64_t>> Query() override;

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

@@ -904,6 +904,68 @@ TEST(AsyncWriterConnectionTest, CloseError) {
   EXPECT_THAT(response.get(), StatusIs(PermanentError().code()));
 }
 
+TEST(AsyncWriterConnectionTest, FinalizeExpectedChecksumMismatchImmediate) {
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Finish).WillOnce([] {
+    return make_ready_future(Status{});
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Finish).WillOnce(Return(storage::internal::HashValues{"ImIEBA==", ""}));
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), MakeRequest(), std::move(mock), hash, 1024);
+  auto response = tested->Finalize(WritePayload{}, storage::Crc32cChecksumValue("AAAAAA=="));
+  EXPECT_THAT(response.get(), StatusIs(StatusCode::kDataLoss));
+}
+
+TEST(AsyncWriterConnectionTest, FinalizeExpectedChecksumMatchImmediate) {
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Finish).WillOnce([] {
+    return make_ready_future(Status{});
+  });
+  EXPECT_CALL(*mock, Write).WillOnce([](Request const&, grpc::WriteOptions) {
+    return make_ready_future(true);
+  });
+  EXPECT_CALL(*mock, Read).WillOnce([] {
+    return make_ready_future(absl::make_optional(MakeTestResponse()));
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Finish).WillRepeatedly(Return(storage::internal::HashValues{"ImIEBA==", ""}));
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), MakeRequest(), std::move(mock), hash, 1024);
+  auto response = tested->Finalize(WritePayload{}, storage::Crc32cChecksumValue("ImIEBA=="));
+  EXPECT_THAT(response.get(), IsOk());
+}
+
+TEST(AsyncWriterConnectionTest, FinalizeExpectedChecksumMismatchOnComplete) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Finish).WillOnce([] {
+    return make_ready_future(Status{});
+  });
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        return sequencer.PushBack("Write");
+      });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  EXPECT_CALL(*hash, Finish).WillRepeatedly(Return(storage::internal::HashValues{"ImIEBA==", ""}));
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), MakeRequest(), std::move(mock), hash, 1024);
+  auto response = tested->Finalize(WritePayload(std::string(128, 'A')), storage::Crc32cChecksumValue("AAAAAA=="));
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  EXPECT_THAT(response.get(), StatusIs(StatusCode::kDataLoss));
+}
+
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal
 }  // namespace cloud