fix: rewrite object in CMEK enabled bucket

google/cloud/storage/blob.py

@@ -3576,7 +3576,15 @@ def rewrite(
  if source.generation:
  query_params["sourceGeneration"] = source.generation
 
- if self.kms_key_name is not None:
+ # When a Customer Managed Encryption Key is used to encrypt Cloud Storage object
+ # at rest, object resource metadata will store the version of the Key Management
+ # Service cryptographic material. If a Blob instance with KMS Key metadata set is
+ # used to rewrite the object, then the existing kmsKeyName version
+ # value can't be used in the rewrite request and the client instead ignores it.
+ if (
+ self.kms_key_name is not None
+ and "cryptoKeyVersions" not in self.kms_key_name
+ ):
  query_params["destinationKmsKeyName"] = self.kms_key_name
 
  _add_generation_match_parameters(

tests/system/test_kms_integration.py

@@ -224,6 +224,17 @@ def test_blob_rewrite_rotate_csek_to_cmek(
 
  assert dest.download_as_bytes() == source_data
 
+ # Test existing kmsKeyName version is ignored in the rewrite request
+ dest = kms_bucket.get_blob(blob_name)
+ source = kms_bucket.get_blob(blob_name)
+ token, rewritten, total = dest.rewrite(source)
+
+ while token is not None:
+ token, rewritten, total = dest.rewrite(source, token=token)
+
+ assert rewritten == len(source_data)
+ assert dest.download_as_bytes() == source_data
+
 
 def test_blob_upload_w_bucket_cmek_enabled(
  kms_bucket,

tests/unit/test_blob.py

@@ -4942,6 +4942,58 @@ def test_rewrite_same_name_w_old_key_new_kms_key(self):
  _target_object=dest,
  )
 
+ def test_rewrite_same_name_w_kms_key_w_version(self):
+ blob_name = "blob"
+ source_key = b"01234567890123456789012345678901" # 32 bytes
+ source_key_b64 = base64.b64encode(source_key).rstrip().decode("ascii")
+ source_key_hash = hashlib.sha256(source_key).digest()
+ source_key_hash_b64 = base64.b64encode(source_key_hash).rstrip().decode("ascii")
+ dest_kms_resource = (
+ "projects/test-project-123/"
+ "locations/us/"
+ "keyRings/test-ring/"
+ "cryptoKeys/test-key"
+ "cryptoKeyVersions/1"
+ )
+ bytes_rewritten = object_size = 42
+ api_response = {
+ "totalBytesRewritten": bytes_rewritten,
+ "objectSize": object_size,
+ "done": True,
+ "resource": {"etag": "DEADBEEF"},
+ }
+ client = mock.Mock(spec=["_post_resource"])
+ client._post_resource.return_value = api_response
+ bucket = _Bucket(client=client)
+ source = self._make_one(blob_name, bucket=bucket, encryption_key=source_key)
+ dest = self._make_one(blob_name, bucket=bucket, kms_key_name=dest_kms_resource)
+
+ token, rewritten, size = dest.rewrite(source)
+
+ self.assertIsNone(token)
+ self.assertEqual(rewritten, bytes_rewritten)
+ self.assertEqual(size, object_size)
+
+ expected_path = f"/b/name/o/{blob_name}/rewriteTo/b/name/o/{blob_name}"
+ expected_data = {"kmsKeyName": dest_kms_resource}
+ # The kmsKeyName version value can't be used in the rewrite request,
+ # so the client instead ignores it.
+ expected_query_params = {}
+ expected_headers = {
+ "X-Goog-Copy-Source-Encryption-Algorithm": "AES256",
+ "X-Goog-Copy-Source-Encryption-Key": source_key_b64,
+ "X-Goog-Copy-Source-Encryption-Key-Sha256": source_key_hash_b64,
+ }
+ client._post_resource.assert_called_once_with(
+ expected_path,
+ expected_data,
+ query_params=expected_query_params,
+ headers=expected_headers,
+ timeout=self._get_default_timeout(),
+ retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+ _target_object=dest,
+ )
+
  def test_update_storage_class_invalid(self):
  blob_name = "blob-name"
  bucket = _Bucket()