fix: send entire object checksum at "initiate a resumable upload session" (#1525)

* [Initiate a resumable upload session](https://cloud.google.com/storage/docs/performing-resumable-uploads#initiate-session) provides a way to specify entire object checksum as metadata. * This will be used by the server before finalization, if there's a mismatch between the checksum provided by client (ie this app) and the one calculated by server. The server will return `400: BadRequest`

Author: chandra-siri

google/cloud/storage/_media/_upload.py

@@ -411,7 +411,12 @@ class ResumableUpload(UploadBase):
  """
 
  def __init__(
- self, upload_url, chunk_size, checksum="auto", headers=None, retry=DEFAULT_RETRY
+ self,
+ upload_url,
+ chunk_size,
+ checksum="auto",
+ headers=None,
+ retry=DEFAULT_RETRY,
  ):
  super(ResumableUpload, self).__init__(upload_url, headers=headers, retry=retry)
  if chunk_size % UPLOAD_CHUNK_SIZE != 0:
@@ -472,7 +477,12 @@ def total_bytes(self):
  return self._total_bytes
 
  def _prepare_initiate_request(
- self, stream, metadata, content_type, total_bytes=None, stream_final=True
+ self,
+ stream,
+ metadata,
+ content_type,
+ total_bytes=None,
+ stream_final=True,
  ):
  """Prepare the contents of HTTP request to initiate upload.
 
@@ -955,7 +965,12 @@ class XMLMPUContainer(UploadBase):
  """
 
  def __init__(
- self, upload_url, filename, headers=None, upload_id=None, retry=DEFAULT_RETRY
+ self,
+ upload_url,
+ filename,
+ headers=None,
+ upload_id=None,
+ retry=DEFAULT_RETRY,
  ):
  super().__init__(upload_url, headers=headers, retry=retry)
  self._filename = filename

google/cloud/storage/blob.py

@@ -14,8 +14,7 @@
 
 # pylint: disable=too-many-lines
 
-"""Create / interact with Google Cloud Storage blobs.
-"""
+"""Create / interact with Google Cloud Storage blobs."""
 
 import base64
 import copy
@@ -142,8 +141,8 @@
  r"(?P<scheme>gs)://(?P<bucket_name>[a-z0-9_.-]+)/(?P<object_name>.+)"
 )
 
-_DEFAULT_CHUNKSIZE = 104857600 # 1024 * 1024 B * 100 = 100 MB
-_MAX_MULTIPART_SIZE = 8388608 # 8 MB
+_DEFAULT_CHUNKSIZE = 104857600 # 1024 * 1024 B * 100 = 100 MiB
+_MAX_MULTIPART_SIZE = 8388608 # 8 MiB
 
 _logger = logging.getLogger(__name__)
 
@@ -181,6 +180,14 @@ class Blob(_PropertyMixin):
  :type generation: long
  :param generation:
  (Optional) If present, selects a specific revision of this object.
+
+ :type crc32c_checksum: str
+ :param crc32c_checksum:
+ (Optional) If set, the CRC32C checksum of the blob's content.
+ CRC32c checksum, as described in RFC 4960, Appendix B; encoded using
+ base64 in big-endian byte order. See
+ Apenndix B: https://datatracker.ietf.org/doc/html/rfc4960#appendix-B
+ base64: https://datatracker.ietf.org/doc/html/rfc4648#section-4
  """
 
  _chunk_size = None # Default value for each instance.
@@ -214,6 +221,7 @@ def __init__(
  encryption_key=None,
  kms_key_name=None,
  generation=None,
+ crc32c_checksum=None,
  ):
  """
  property :attr:`name`
@@ -237,6 +245,9 @@ def __init__(
  if generation is not None:
  self._properties["generation"] = generation
 
+ if crc32c_checksum is not None:
+ self._properties["crc32c"] = crc32c_checksum
+
  @property
  def bucket(self):
  """Bucket which contains the object.
@@ -1643,7 +1654,9 @@ def download_as_string(
  :raises: :class:`google.cloud.exceptions.NotFound`
  """
  warnings.warn(
- _DOWNLOAD_AS_STRING_DEPRECATED, PendingDeprecationWarning, stacklevel=2
+ _DOWNLOAD_AS_STRING_DEPRECATED,
+ PendingDeprecationWarning,
+ stacklevel=2,
  )
  with create_trace_span(name="Storage.Blob.downloadAsString"):
  return self.download_as_bytes(
@@ -1999,12 +2012,18 @@ def _do_multipart_upload(
  transport = self._get_transport(client)
  if "metadata" in self._properties and "metadata" not in self._changes:
  self._changes.add("metadata")
+
  info = self._get_upload_arguments(client, content_type, command=command)
  headers, object_metadata, content_type = info
 
+ if "crc32c" in self._properties:
+ object_metadata["crc32c"] = self._properties["crc32c"]
+
  hostname = _get_host_name(client._connection)
  base_url = _MULTIPART_URL_TEMPLATE.format(
- hostname=hostname, bucket_path=self.bucket.path, api_version=_API_VERSION
+ hostname=hostname,
+ bucket_path=self.bucket.path,
+ api_version=_API_VERSION,
  )
  name_value_pairs = []
 
@@ -2195,9 +2214,14 @@ def _initiate_resumable_upload(
  if extra_headers is not None:
  headers.update(extra_headers)
 
+ if "crc32c" in self._properties:
+ object_metadata["crc32c"] = self._properties["crc32c"]
+
  hostname = _get_host_name(client._connection)
  base_url = _RESUMABLE_URL_TEMPLATE.format(
- hostname=hostname, bucket_path=self.bucket.path, api_version=_API_VERSION
+ hostname=hostname,
+ bucket_path=self.bucket.path,
+ api_version=_API_VERSION,
  )
  name_value_pairs = []
 
@@ -2234,7 +2258,11 @@ def _initiate_resumable_upload(
 
  upload_url = _add_query_parameters(base_url, name_value_pairs)
  upload = ResumableUpload(
- upload_url, chunk_size, headers=headers, checksum=checksum, retry=retry
+ upload_url,
+ chunk_size,
+ headers=headers,
+ checksum=checksum,
+ retry=retry,
  )
 
  upload.initiate(
@@ -3426,7 +3454,11 @@ def set_iam_policy(
  return Policy.from_api_repr(info)
 
  def test_iam_permissions(
- self, permissions, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+ self,
+ permissions,
+ client=None,
+ timeout=_DEFAULT_TIMEOUT,
+ retry=DEFAULT_RETRY,
  ):
  """API call: test permissions
 
@@ -3693,7 +3725,10 @@ def compose(
 
  source_objects = []
  for source, source_generation in zip(sources, if_source_generation_match):
- source_object = {"name": source.name, "generation": source.generation}
+ source_object = {
+ "name": source.name,
+ "generation": source.generation,
+ }
 
  preconditions = {}
  if source_generation is not None:
@@ -4154,7 +4189,10 @@ def open(
  "encoding, errors and newline arguments are for text mode only"
  )
  return BlobWriter(
- self, chunk_size=chunk_size, ignore_flush=ignore_flush, **kwargs
+ self,
+ chunk_size=chunk_size,
+ ignore_flush=ignore_flush,
+ **kwargs,
  )
  elif mode in ("r", "rt"):
  if ignore_flush:

google/cloud/storage/bucket.py

@@ -386,7 +386,10 @@ class LifecycleRuleSetStorageClass(dict):
  def __init__(self, storage_class, **kw):
  conditions = LifecycleRuleConditions(**kw)
  rule = {
- "action": {"type": "SetStorageClass", "storageClass": storage_class},
+ "action": {
+ "type": "SetStorageClass",
+ "storageClass": storage_class,
+ },
  "condition": dict(conditions),
  }
  super().__init__(rule)
@@ -846,6 +849,7 @@ def blob(
  encryption_key=None,
  kms_key_name=None,
  generation=None,
+ crc32c_checksum=None,
  ):
  """Factory constructor for blob object.
 
@@ -873,6 +877,14 @@ def blob(
  :param generation: (Optional) If present, selects a specific revision of
  this object.
 
+ :type crc32c_checksum: str
+ :param crc32c_checksum:
+ (Optional) If set, the CRC32C checksum of the blob's content.
+ CRC32c checksum, as described in RFC 4960, Appendix B; encoded using
+ base64 in big-endian byte order. See
+ Apenndix B: https://datatracker.ietf.org/doc/html/rfc4960#appendix-B
+ base64: https://datatracker.ietf.org/doc/html/rfc4648#section-4
+
  :rtype: :class:`google.cloud.storage.blob.Blob`
  :returns: The blob object created.
  """
@@ -883,6 +895,7 @@ def blob(
  encryption_key=encryption_key,
  kms_key_name=kms_key_name,
  generation=generation,
+ crc32c_checksum=crc32c_checksum,
  )
 
  def notification(
@@ -3253,7 +3266,10 @@ def configure_website(self, main_page_suffix=None, not_found_page=None):
  :type not_found_page: str
  :param not_found_page: The file to use when a page isn't found.
  """
- data = {"mainPageSuffix": main_page_suffix, "notFoundPage": not_found_page}
+ data = {
+ "mainPageSuffix": main_page_suffix,
+ "notFoundPage": not_found_page,
+ }
  self._patch_property("website", data)
 
  def disable_website(self):
@@ -3385,7 +3401,11 @@ def set_iam_policy(
  return Policy.from_api_repr(info)
 
  def test_iam_permissions(
- self, permissions, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+ self,
+ permissions,
+ client=None,
+ timeout=_DEFAULT_TIMEOUT,
+ retry=DEFAULT_RETRY,
  ):
  """API call: test permissions
 

tests/data/random_9_MiB_file

@@ -30,6 +30,7 @@
  ("logo", "CloudPlatform_128px_Retina.png"),
  ("big", "five-point-one-mb-file.zip"),
  ("simple", "simple.txt"),
+ ("big_9MiB", "random_9_MiB_file"),
 ]
 _file_data = {
  key: {"path": os.path.join(data_dirname, file_name)}

tests/system/conftest.py

@@ -39,6 +39,40 @@ def _check_blob_hash(blob, info):
  assert md5_hash == info["hash"]
 
 
+def test_large_file_write_from_stream_w_user_provided_checksum(
+ shared_bucket,
+ blobs_to_delete,
+ file_data,
+ service_account,
+):
+ blob = shared_bucket.blob(
+ f"LargeFile{uuid.uuid4().hex}", crc32c_checksum="20tD7w=="
+ )
+
+ info = file_data["big_9MiB"]
+ with open(info["path"], "rb") as file_obj:
+ blob.upload_from_file(file_obj)
+ blobs_to_delete.append(blob)
+
+
+def test_large_file_write_from_stream_w_user_provided_wrong_checksum(
+ shared_bucket,
+ blobs_to_delete,
+ file_data,
+ service_account,
+):
+ blob = shared_bucket.blob(
+ f"LargeFile{uuid.uuid4().hex}", crc32c_checksum="A0tD7w=="
+ )
+
+ info = file_data["big_9MiB"]
+ with pytest.raises(exceptions.BadRequest) as excep_info:
+ with open(info["path"], "rb") as file_obj:
+ blob.upload_from_file(file_obj)
+ blobs_to_delete.append(blob)
+ assert excep_info.value.code == 400
+
+
 def test_large_file_write_from_stream(
  shared_bucket,
  blobs_to_delete,

tests/system/test_blob.py

@@ -2762,12 +2762,21 @@ def _initiate_resumable_helper(
  metadata=None,
  mtls=False,
  retry=None,
+ crc32c_checksum=None,
  ):
  from google.cloud.storage._media.requests import ResumableUpload
  from google.cloud.storage.blob import _DEFAULT_CHUNKSIZE
 
  bucket = _Bucket(name="whammy", user_project=user_project)
- blob = self._make_one("blob-name", bucket=bucket, kms_key_name=kms_key_name)
+ if crc32c_checksum is None:
+ blob = self._make_one("blob-name", bucket=bucket, kms_key_name=kms_key_name)
+ else:
+ blob = self._make_one(
+ "blob-name",
+ bucket=bucket,
+ kms_key_name=kms_key_name,
+ crc32c_checksum=crc32c_checksum,
+ )
  if metadata:
  self.assertIsNone(blob.metadata)
  blob._properties["metadata"] = metadata
@@ -2919,6 +2928,10 @@ def _initiate_resumable_helper(
  else:
  # Check the mocks.
  blob._get_writable_metadata.assert_called_once_with()
+
+ if "crc32c" in blob._properties:
+ object_metadata["crc32c"] = blob._properties["crc32c"]
+
  payload = json.dumps(object_metadata).encode("utf-8")
 
  with patch.object(
@@ -2945,6 +2958,17 @@ def _initiate_resumable_helper(
  def test__initiate_resumable_upload_with_metadata(self):
  self._initiate_resumable_helper(metadata={"test": "test"})
 
+ def test__initiate_resumable_upload_with_user_provided_checksum(self):
+ self._initiate_resumable_helper(
+ crc32c_checksum="this-is-a-fake-checksum-for-unit-tests",
+ )
+
+ def test__initiate_resumable_upload_w_metadata_and_user_provided_checksum(self):
+ self._initiate_resumable_helper(
+ crc32c_checksum="test-checksum",
+ metadata={"my-fav-key": "my-fav-value"},
+ )
+
  def test__initiate_resumable_upload_with_custom_timeout(self):
  self._initiate_resumable_helper(timeout=9.58)