feat(Storage)!: Enable full object checksum validation for resumable uploads

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1.Tests/UploadObjectOptionsTest.cs

@@ -54,7 +54,7 @@ public void InvalidChunkSize(int chunkSize)
         [Fact]
         public void ModifyMediaUpload_DefaultOptions()
         {
-            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null);
+            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null, null);
             var options = new UploadObjectOptions();
             options.ModifyMediaUpload(upload);
             Assert.Equal(ResumableUpload<InsertMediaUpload>.DefaultChunkSize, upload.ChunkSize);
@@ -71,7 +71,7 @@ public void ModifyMediaUpload_DefaultOptions()
         [Fact]
         public void ModifyMediaUpload_AllOptions_PositiveMatch()
         {
-            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null);
+            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null, null);
             var options = new UploadObjectOptions
             {
                 ChunkSize = UploadObjectOptions.MinimumChunkSize * 3,
@@ -96,7 +96,7 @@ public void ModifyMediaUpload_AllOptions_PositiveMatch()
         [Fact]
         public void ModifyMediaUpload_AllOptions_NegativeMatch()
         {
-            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null);
+            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null, null);
             var options = new UploadObjectOptions
             {
                 ChunkSize = UploadObjectOptions.MinimumChunkSize * 3,
@@ -117,7 +117,7 @@ public void ModifyMediaUpload_AllOptions_NegativeMatch()
         [Fact]
         public void ModifyMediaUpload_MatchNotMatchConflicts()
         {
-            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null);
+            var upload = new CustomMediaUpload(new DummyService(), null, "bucket", new MemoryStream(), null, null);
             Assert.Throws<ArgumentException>(() =>
             {
                 var options = new UploadObjectOptions { IfGenerationMatch = 1L, IfGenerationNotMatch = 2L };

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/CustomMediaUpload.cs

@@ -1,4 +1,4 @@
-// Copyright 2017 Google Inc. All Rights Reserved.
+// Copyright 2017 Google Inc. All Rights Reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,12 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using Google.Api.Gax;
 using Google.Apis.Services;
+using Google.Apis.Upload;
 using System;
 using System.IO;
 using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
 using static Google.Apis.Storage.v1.ObjectsResource;
-using Google.Apis.Upload;
 
 namespace Google.Cloud.Storage.V1
 {
@@ -26,12 +29,170 @@ namespace Google.Cloud.Storage.V1
     /// </summary>
     internal sealed class CustomMediaUpload : InsertMediaUpload
     {
+        private const string GoogleHashHeader = "x-goog-hash";
+
         public CustomMediaUpload(IClientService service, Apis.Storage.v1.Data.Object body, string bucket,
-            Stream stream, string contentType)
-            : base(service, body, bucket, stream, contentType)
+            Stream stream, string contentType, UploadObjectOptions options)
+            : base(service, body, bucket, (options?.UploadValidationMode ?? UploadObjectOptions.DefaultValidationMode) != UploadValidationMode.None ? new HashingStream(stream) : stream, contentType)
         {
+            var validationMode = options?.UploadValidationMode ?? UploadObjectOptions.DefaultValidationMode;
+            GaxPreconditions.CheckEnumValue(validationMode, nameof(UploadObjectOptions.UploadValidationMode));
+            if (validationMode != UploadValidationMode.None)
+            {
+                var hashingStream = ContentStream as HashingStream;
+                LastRequestExecuting += (HttpRequestMessage request) =>
+                {
+                    if (hashingStream != null)
+                    {
+                        if (hashingStream.HasGaps)
+                        {
+                            throw new ArgumentException(
+                                "Cannot perform hash validation when resuming an upload from a non-zero offset, " +
+                                "as the complete stream contents are required to compute the hash. " +
+                                "To resume this upload, disable validation by setting UploadValidationMode to None.",
+                                nameof(stream));
+                        }
+                        if (hashingStream.IsHashComplete)
+                        {
+                            var calculatedHash = hashingStream.GetBase64Hash();
+                            bool hasCrc32c = false;
+                            if (request.Headers.TryGetValues(GoogleHashHeader, out var values))
+                            {
+                                foreach (var value in values)
+                                {
+                                    if (value?.IndexOf("crc32c", StringComparison.OrdinalIgnoreCase) >= 0)
+                                    {
+                                        hasCrc32c = true;
+                                        break;
+                                    }
+                                }
+                            }
+                            if (!hasCrc32c)
+                            {
+                                request.Headers.TryAddWithoutValidation(GoogleHashHeader, $"crc32c={calculatedHash}");
+                            }
+                        }
+                    }
+                };
+            }
         }
 
         internal new ResumableUploadOptions Options => base.Options;
+
+        internal sealed class HashingStream : Stream
+        {
+            private readonly Stream _stream;
+            private readonly Crc32c _hasher;
+            private long _maxPositionHashed = 0;
+            private long _position = 0;
+            private bool _hasGaps = false;
+            private bool _reachedEof = false;
+            public bool HasGaps => _hasGaps;
+
+            public HashingStream(Stream stream)
+            {
+                _stream = stream;
+                _hasher = new Crc32c();
+            }
+
+            public bool IsHashComplete => !_hasGaps && (_stream.CanSeek ? _maxPositionHashed == _stream.Length : _reachedEof);
+
+            public override int Read(byte[] buffer, int offset, int count)
+            {
+                long startingPos = _stream.CanSeek ? _stream.Position : _position;
+                int bytesRead = _stream.Read(buffer, offset, count);
+                if (count > 0 && bytesRead == 0)
+                {
+                    _reachedEof = true;
+                }
+                ProcessBytes(buffer, offset, bytesRead, startingPos);
+                if (!_stream.CanSeek)
+                {
+                    _position += bytesRead;
+                }
+                return bytesRead;
+            }
+
+            public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+            {
+                long startingPos = _stream.CanSeek ? _stream.Position : _position;
+                int bytesRead = await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
+                if (count > 0 && bytesRead == 0)
+                {
+                    _reachedEof = true;
+                }
+                ProcessBytes(buffer, offset, bytesRead, startingPos);
+                if (!_stream.CanSeek)
+                {
+                    _position += bytesRead;
+                }
+                return bytesRead;
+            }
+
+            private void ProcessBytes(byte[] buffer, int offset, int bytesRead, long startingPos)
+            {
+                if (bytesRead <= 0) return;
+
+                if (startingPos > _maxPositionHashed)
+                {
+                    _hasGaps = true;
+                    return;
+                }
+
+                // Only hash bytes that are beyond the furthest point we've already hashed.
+                // This handles the rewind and re-read scenario during retries.
+                if (startingPos + bytesRead > _maxPositionHashed)
+                {
+                    long newBytesStart = Math.Max(startingPos, _maxPositionHashed);
+                    int actuallyNewCount = (int) ((startingPos + bytesRead) - newBytesStart);
+                    int bufferOffset = offset + (int) (newBytesStart - startingPos);
+
+                    _hasher.UpdateHash(buffer, bufferOffset, actuallyNewCount);
+                    _maxPositionHashed = startingPos + bytesRead;
+                }
+            }
+
+            public override long Position
+            {
+                get => _stream.CanSeek ? _stream.Position : _position;
+                set
+                {
+                    if (_stream.CanSeek)
+                    {
+                        _stream.Position = value;
+                    }
+                    else
+                    {
+                        throw new NotSupportedException();
+                    }
+                }
+            }
+
+            public override long Seek(long offset, SeekOrigin origin)
+            {
+                if (_stream.CanSeek)
+                {
+                    return _stream.Seek(offset, origin);
+                }
+                throw new NotSupportedException();
+            }
+
+            public string GetBase64Hash() => Convert.ToBase64String(_hasher.GetHash());
+            public override bool CanRead => _stream.CanRead;
+            public override bool CanSeek => _stream.CanSeek;
+            public override bool CanWrite => _stream.CanWrite;
+            public override long Length => _stream.Length;
+            public override void Flush() => _stream.Flush();
+            public override void SetLength(long value) => _stream.SetLength(value);
+            public override void Write(byte[] buffer, int offset, int count) => _stream.Write(buffer, offset, count);
+            protected override void Dispose(bool disposing)
+            {
+                if (disposing)
+                {
+                    _stream?.Dispose();
+                }
+                base.Dispose(disposing);
+            }
+        }
     }
 }

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1.csproj

@@ -9,6 +9,7 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="ConfigureAwaitChecker.Analyzer" PrivateAssets="All" />
+    <PackageReference Include="Google.Apis" VersionOverride="1.76"/>
     <PackageReference Include="Google.Api.Gax.Rest" />
     <PackageReference Include="Google.Apis.Storage.v1" VersionOverride="[1.74.0.4115, 2.0.0.0)" />
   </ItemGroup>

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/StorageClientImpl.UploadObject.cs

@@ -48,7 +48,7 @@ public override ObjectsResource.InsertMediaUpload CreateObjectUploader(
         {
             ValidateObject(destination, nameof(destination));
             GaxPreconditions.CheckNotNull(source, nameof(source));
-            var mediaUpload = new CustomMediaUpload(Service, destination, destination.Bucket, source, destination.ContentType);
+            var mediaUpload = new CustomMediaUpload(Service, destination, destination.Bucket, source, destination.ContentType, options);
             options?.ModifyMediaUpload(mediaUpload);
             ApplyEncryptionKey(options?.EncryptionKey, options?.KmsKeyName, mediaUpload);
             return mediaUpload;
@@ -143,9 +143,6 @@ private sealed class UploadHelper
         {
             private readonly StorageClient _client;
             private readonly ObjectsResource.InsertMediaUpload _mediaUpload;
-            private readonly Crc32c _crc;
-            private readonly Action<Object> _validationFailureAction;
-            private readonly Func<Object, CancellationToken, Task> _validationFailureAsyncAction;
 
             internal UploadHelper(
                 StorageClient client,
@@ -160,43 +157,13 @@ internal UploadHelper(
                 {
                     _mediaUpload.ProgressChanged += progress.Report;
                 }
-
-                var validationMode = options?.UploadValidationMode ?? UploadObjectOptions.DefaultValidationMode;
-                GaxPreconditions.CheckEnumValue(validationMode, nameof(UploadObjectOptions.UploadValidationMode));
-                switch (validationMode)
-                {
-                    case UploadValidationMode.DeleteAndThrow:
-                        _crc = new Crc32c();
-                        _mediaUpload.UploadStreamInterceptor += _crc.UpdateHash;
-                        _validationFailureAction = obj => client.DeleteObject(obj, new DeleteObjectOptions { Generation = obj.Generation });
-                        _validationFailureAsyncAction = (obj, token) => client.DeleteObjectAsync(obj, new DeleteObjectOptions { Generation = obj.Generation }, token);
-                        break;
-                    case UploadValidationMode.ThrowOnly:
-                        _crc = new Crc32c();
-                        _mediaUpload.UploadStreamInterceptor += _crc.UpdateHash;
-                        break;
-                }
             }
 
             internal Object Execute()
             {
                 _mediaUpload.Upload();
                 _mediaUpload.GetProgress().ThrowOnFailure();
                 var result = _mediaUpload.ResponseBody;
-                var hash = _crc == null ? result.Crc32c : Convert.ToBase64String(_crc.GetHash());
-                if (hash != result.Crc32c)
-                {
-                    AggregateException additionalFailures = null;
-                    try
-                    {
-                        _validationFailureAction?.Invoke(result);
-                    }
-                    catch (Exception e)
-                    {
-                        additionalFailures = new AggregateException(e);
-                    }
-                    throw new UploadValidationException(hash, result, additionalFailures);
-                }
                 return result;
             }
 
@@ -205,23 +172,6 @@ internal async Task<Object> ExecuteAsync(CancellationToken cancellationToken)
                 await _mediaUpload.UploadAsync(cancellationToken).ConfigureAwait(false);
                 _mediaUpload.GetProgress().ThrowOnFailure();
                 var result = _mediaUpload.ResponseBody;
-                var hash = _crc == null ? result.Crc32c : Convert.ToBase64String(_crc.GetHash());
-                if (hash != result.Crc32c)
-                {
-                    AggregateException additionalFailures = null;
-                    try
-                    {
-                        if (_validationFailureAsyncAction != null)
-                        {
-                            await _validationFailureAsyncAction(result, cancellationToken).ConfigureAwait(false);
-                        }
-                    }
-                    catch (Exception e)
-                    {
-                        additionalFailures = new AggregateException(e);
-                    }
-                    throw new UploadValidationException(hash, result, additionalFailures);
-                }
                 return result;
             }
         }

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/UploadObjectOptions.cs

@@ -32,7 +32,7 @@ public sealed class UploadObjectOptions
         /// </summary>
         public const int MinimumChunkSize = ResumableUpload<Object>.MinimumChunkSize;
 
-        internal static UploadValidationMode DefaultValidationMode { get; } = V1.UploadValidationMode.DeleteAndThrow;
+        internal static UploadValidationMode DefaultValidationMode { get; } = V1.UploadValidationMode.RejectAndThrow;
 
         /// <summary>
         /// Precondition for upload: the object is only uploaded if the existing object's