diff --git a/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs b/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs index d77909b41..72b1aaa9e 100644 --- a/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs +++ b/Duplicati/Library/Backend/AliyunOSS/AliyunOSSBackend.cs @@ -198,9 +198,14 @@ namespace Duplicati.Library.Backend.AliyunOSS var client = GetClient(); try { + var metadata = new ObjectMetadata + { + ContentLength = stream.Length + }; + using var timeoutStream = stream.ObserveReadTimeout(_timeouts.ReadWriteTimeout, false); var objectResult = await Task.Factory.FromAsync( - (cb, state) => client.BeginPutObject(bucketName, objectName, timeoutStream, cb, state), + (cb, state) => client.BeginPutObject(bucketName, objectName, timeoutStream, metadata, cb, state), client.EndPutObject, null).ConfigureAwait(false); diff --git a/Duplicati/Library/Backend/Backblaze/B2.cs b/Duplicati/Library/Backend/Backblaze/B2.cs index 63c7a04c0..967c41a68 100644 --- a/Duplicati/Library/Backend/Backblaze/B2.cs +++ b/Duplicati/Library/Backend/Backblaze/B2.cs @@ -310,43 +310,8 @@ public class B2 : IStreamingBackend public async Task PutAsync(string remotename, Stream stream, CancellationToken cancelToken) { - TempFile? tmp = null; - - var measure = stream; - while (measure is OverrideableStream os && - os.GetType().GetField("m_basestream", BindingFlags.NonPublic | BindingFlags.Instance)?.GetValue(os) is Stream baseStream) - { - measure = baseStream; - } - - if (measure == null) - throw new Exception($"Unable to unwrap stream from: {stream.GetType()}"); - - string sha1; - if (measure.CanSeek) - { - var p = measure.Position; - - // Compute the hash - using (var hashalg = HashFactory.CreateHasher("SHA1")) - sha1 = Utility.Utility.ByteArrayAsHexString(hashalg.ComputeHash(measure)); - - measure.Position = p; - } - else - { - // No seeking possible, use a temp file - tmp = new TempFile(); - await using (var sr = File.OpenWrite(tmp)) - using (var hasher = HashFactory.CreateHasher("SHA1")) - await using (var hc = new HashCalculatingStream(measure, hasher)) - { - await Utility.Utility.CopyStreamAsync(hc, sr, cancelToken).ConfigureAwait(false); - sha1 = hc.GetFinalHashString(); - } - - stream = File.OpenRead(tmp); - } + (stream, var sha1, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(stream, "SHA1", cancelToken).ConfigureAwait(false); + using var _ = tmp; if (_filecache == null) await RebuildFileCache(cancelToken).ConfigureAwait(false); @@ -400,10 +365,6 @@ public class B2 : IStreamingBackend throw; } - finally - { - tmp?.Dispose(); - } } /// diff --git a/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs b/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs index d4f5d7809..c8c3f5432 100644 --- a/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs +++ b/Duplicati/Library/Backend/Idrivee2/Idrivee2Backend.cs @@ -190,9 +190,8 @@ namespace Duplicati.Library.Backend if (m_s3Client == null) { (var accessKeyId, var accessKeySecret) = m_auth.GetCredentials(); - // TODO: Do not make blocking calls in the constructor var host = await GetRegionEndpointAsync("https://api.idrivee2.com/api/service/get_region_end_point/" + accessKeyId, cancellationToken).ConfigureAwait(false); - m_s3Client = new S3AwsClient(accessKeyId, accessKeySecret, null, host, null, true, false, m_timeouts, m_options); + m_s3Client = new S3AwsClient(accessKeyId, accessKeySecret, null, host, null, true, false, false, m_timeouts, m_options); } return m_s3Client; diff --git a/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs b/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs index ec4e5f30f..805544398 100644 --- a/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs +++ b/Duplicati/Library/Backend/Jottacloud/Jottacloud.cs @@ -489,108 +489,71 @@ namespace Duplicati.Library.Backend // we specify the file size in a custom header. And if the stream is not seek-able we are not able // to use stream.Length, so we are back at square one. - // TODO: This unwrapping code is shared with CloudFiles - TempFile? tmpFile = null; - Stream? baseStream = stream; - while (baseStream is OverrideableStream) - baseStream = typeof(OverrideableStream).GetField("m_basestream", System.Reflection.BindingFlags.DeclaredOnly | System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)?.GetValue(baseStream) as System.IO.Stream; - if (baseStream == null) - throw new Exception(string.Format("Unable to unwrap stream from: {0}", stream.GetType())); - string md5Hash; - if (baseStream.CanSeek) - { - var originalPosition = baseStream.Position; - using (var md5 = MD5.Create()) - md5Hash = Utility.Utility.ByteArrayAsHexString(md5.ComputeHash(baseStream)); - baseStream.Position = originalPosition; - } - else - { - // No seeking possible, use a temp file - tmpFile = new TempFile(); - using (var os = File.OpenWrite(tmpFile)) - using (var hasher = MD5.Create()) - using (var md5 = new HashCalculatingStream(baseStream, hasher)) - { - await Utility.Utility.CopyStreamAsync(md5, os, true, cancelToken).ConfigureAwait(false); - md5Hash = md5.GetFinalHashString(); - } - stream = File.OpenRead(tmpFile); - } - try - { - // Create request, with query parameter, and a few custom headers. - // NB: If we wanted to we could send the same POST request as below but without the file contents - // and with "cphash=[md5Hash]" as the only query parameter. Then we will get an HTTP 200 (OK) response - // if an identical file already exists, and we can skip uploading the new file. We will get - // HTTP 404 (Not Found) if file does not exists or it exists with a different hash, in which - // case we must send a new request to upload the new content. - var fileSize = stream.Length; - var req = CreateRequest(WebRequestMethods.Http.Post, remotename, "umode=nomultipart", true); - req.Headers.Add("JMd5", md5Hash); // Not required, but it will make the server verify the content and mark the file as corrupt if there is a mismatch. - req.Headers.Add("JSize", fileSize.ToString()); // Required, and used to mark file as incomplete if we upload something be the total size of the original file! - // File time stamp headers: Since we are working with a stream here we do not know the local file's timestamps, - // and then we can just omit the JCreated and JModified and let the server automatically set the current time. - //req.Headers.Add("JCreated", timeCreated); - //req.Headers.Add("JModified", timeModified); - req.ContentType = "application/octet-stream"; - req.ContentLength = fileSize; + (stream, var md5Hash, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(stream, "MD5", cancelToken).ConfigureAwait(false); + using var _ = tmp; - // Write post data request - var areq = new AsyncHttpRequest(req); - using (var rs = areq.GetRequestStream()) - using (var ts = rs.ObserveWriteTimeout(m_timeouts.ReadWriteTimeout)) - await Utility.Utility.CopyStreamAsync(stream, rs, true, cancelToken).ConfigureAwait(false); - // Send request, and check response - using (var resp = (HttpWebResponse)areq.GetResponse()) - { - if (resp.StatusCode != HttpStatusCode.Created) - throw new WebException(Strings.Jottacloud.FileUploadError, null, WebExceptionStatus.ProtocolError, resp); + // Create request, with query parameter, and a few custom headers. + // NB: If we wanted to we could send the same POST request as below but without the file contents + // and with "cphash=[md5Hash]" as the only query parameter. Then we will get an HTTP 200 (OK) response + // if an identical file already exists, and we can skip uploading the new file. We will get + // HTTP 404 (Not Found) if file does not exists or it exists with a different hash, in which + // case we must send a new request to upload the new content. + var fileSize = stream.Length; + var req = CreateRequest(WebRequestMethods.Http.Post, remotename, "umode=nomultipart", true); + req.Headers.Add("JMd5", md5Hash); // Not required, but it will make the server verify the content and mark the file as corrupt if there is a mismatch. + req.Headers.Add("JSize", fileSize.ToString()); // Required, and used to mark file as incomplete if we upload something be the total size of the original file! + // File time stamp headers: Since we are working with a stream here we do not know the local file's timestamps, + // and then we can just omit the JCreated and JModified and let the server automatically set the current time. + //req.Headers.Add("JCreated", timeCreated); + //req.Headers.Add("JModified", timeModified); + req.ContentType = "application/octet-stream"; + req.ContentLength = fileSize; - // Request seems to be successful, but we must verify the response XML content to be sure that the file - // was correctly uploaded: The server will verify the JSize header and mark the file as incomplete if - // there was mismatch, and it will verify the JMd5 header and mark the file as corrupt if there was a hash - // mismatch. The returned XML contains a file element, and if upload was error free it contains a single - // child element "currentRevision", which has a "state" child element with the string "COMPLETED". - // If there was a problem we should have a "latestRevision" child element, and this will have state with - // value "INCOMPLETE" or "CORRUPT". If the file was new or had no previous complete versions the latestRevision - // will be the only child, but if not there may also be a "currentRevision" representing the previous - // complete version - and then we need to detect the case where our upload failed but there was an existing - // complete version! - using (var rs = areq.GetResponseStream()) + // Write post data request + var areq = new AsyncHttpRequest(req); + using (var rs = areq.GetRequestStream()) + using (var ts = rs.ObserveWriteTimeout(m_timeouts.ReadWriteTimeout)) + await Utility.Utility.CopyStreamAsync(stream, rs, true, cancelToken).ConfigureAwait(false); + // Send request, and check response + using (var resp = (HttpWebResponse)areq.GetResponse()) + { + if (resp.StatusCode != HttpStatusCode.Created) + throw new WebException(Strings.Jottacloud.FileUploadError, null, WebExceptionStatus.ProtocolError, resp); + + // Request seems to be successful, but we must verify the response XML content to be sure that the file + // was correctly uploaded: The server will verify the JSize header and mark the file as incomplete if + // there was mismatch, and it will verify the JMd5 header and mark the file as corrupt if there was a hash + // mismatch. The returned XML contains a file element, and if upload was error free it contains a single + // child element "currentRevision", which has a "state" child element with the string "COMPLETED". + // If there was a problem we should have a "latestRevision" child element, and this will have state with + // value "INCOMPLETE" or "CORRUPT". If the file was new or had no previous complete versions the latestRevision + // will be the only child, but if not there may also be a "currentRevision" representing the previous + // complete version - and then we need to detect the case where our upload failed but there was an existing + // complete version! + using (var rs = areq.GetResponseStream()) + { + var doc = new System.Xml.XmlDocument(); + try { doc.Load(rs); } + catch (System.Xml.XmlException) { - var doc = new System.Xml.XmlDocument(); - try { doc.Load(rs); } - catch (System.Xml.XmlException) - { - throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); - } - bool uploadCompletedSuccessfully = false; - var xFile = doc["file"]; - if (xFile != null) - { - var xRevState = xFile.SelectSingleNode("latestRevision"); - if (xRevState == null) - { - xRevState = xFile.SelectSingleNode("currentRevision/state"); - if (xRevState != null) - uploadCompletedSuccessfully = xRevState.InnerText == "COMPLETED"; // Success: There is no "latestRevision", only a "currentRevision" (and it specifies the file is complete, but I think it always will). - } - } - if (!uploadCompletedSuccessfully) // Report error (and we just let the incomplete/corrupt file revision stay on the server..) - throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); + throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); } + bool uploadCompletedSuccessfully = false; + var xFile = doc["file"]; + if (xFile != null) + { + var xRevState = xFile.SelectSingleNode("latestRevision"); + if (xRevState == null) + { + xRevState = xFile.SelectSingleNode("currentRevision/state"); + if (xRevState != null) + uploadCompletedSuccessfully = xRevState.InnerText == "COMPLETED"; // Success: There is no "latestRevision", only a "currentRevision" (and it specifies the file is complete, but I think it always will). + } + } + if (!uploadCompletedSuccessfully) // Report error (and we just let the incomplete/corrupt file revision stay on the server..) + throw new WebException(Strings.Jottacloud.FileUploadError, WebExceptionStatus.ProtocolError); } } - finally - { - try - { - if (tmpFile != null) - tmpFile.Dispose(); - } - catch { } - } } } } diff --git a/Duplicati/Library/Backend/S3/S3AwsClient.cs b/Duplicati/Library/Backend/S3/S3AwsClient.cs index d21ea417e..ae2c5957a 100644 --- a/Duplicati/Library/Backend/S3/S3AwsClient.cs +++ b/Duplicati/Library/Backend/S3/S3AwsClient.cs @@ -61,6 +61,10 @@ namespace Duplicati.Library.Backend /// The option to specify if chunk encoding should be used /// private readonly bool m_useChunkEncoding; + /// + /// The option to specify if payload signing should be disabled + /// + private readonly bool m_disablePayloadSigning; /// /// The DNS host of the S3 server @@ -93,7 +97,7 @@ namespace Duplicati.Library.Backend ]); public S3AwsClient(string awsID, string awsKey, string? locationConstraint, string servername, - string? storageClass, bool useSSL, bool disableChunkEncoding, TimeoutOptionsHelper.Timeouts timeouts, Dictionary options) + string? storageClass, bool useSSL, bool disableChunkEncoding, bool disablePayloadSigning, TimeoutOptionsHelper.Timeouts timeouts, Dictionary options) { var cfg = GetDefaultAmazonS3Config(); cfg.UseHttp = !useSSL; @@ -109,6 +113,7 @@ namespace Duplicati.Library.Backend m_storageClass = storageClass; m_dnsHost = string.IsNullOrWhiteSpace(cfg.ServiceURL) ? null : new System.Uri(cfg.ServiceURL).Host; m_useChunkEncoding = !disableChunkEncoding; + m_disablePayloadSigning = disablePayloadSigning; m_archiveClasses = ParseStorageClasses(options.GetValueOrDefault(S3_ARCHIVE_CLASSES_OPTION)); } @@ -217,13 +222,20 @@ namespace Duplicati.Library.Backend public virtual async Task AddFileStreamAsync(string bucketName, string keyName, Stream source, CancellationToken cancelToken) { + (source, var md5, var tmp) = await Utility.Utility.CalculateThrottledStreamHash(source, "MD5", cancelToken).ConfigureAwait(false); + using var _ = tmp; + + md5 = Convert.ToBase64String(Utility.Utility.HexStringAsByteArray(md5)); + using var ts = source.ObserveReadTimeout(m_timeouts.ReadWriteTimeout, false); var objectAddRequest = new PutObjectRequest { BucketName = bucketName, Key = keyName, InputStream = ts, - UseChunkEncoding = m_useChunkEncoding + UseChunkEncoding = m_useChunkEncoding, + MD5Digest = md5, + DisablePayloadSigning = m_disablePayloadSigning }; if (!string.IsNullOrWhiteSpace(m_storageClass)) objectAddRequest.StorageClass = new S3StorageClass(m_storageClass); diff --git a/Duplicati/Library/Backend/S3/S3Backend.cs b/Duplicati/Library/Backend/S3/S3Backend.cs index 8d4a2be0e..b6b21c166 100644 --- a/Duplicati/Library/Backend/S3/S3Backend.cs +++ b/Duplicati/Library/Backend/S3/S3Backend.cs @@ -41,6 +41,7 @@ namespace Duplicati.Library.Backend private const string SSL_OPTION = "use-ssl"; private const string S3_CLIENT_OPTION = "s3-client"; private const string S3_DISABLE_CHUNK_ENCODING_OPTION = "s3-disable-chunk-encoding"; + private const string S3_DISABLE_PAYLOAD_SIGNING_OPTION = "s3-disable-payload-signing"; private const string S3_LIST_API_VERSION_OPTION = "s3-list-api-version"; private const string S3_RECURSIVE_LIST = "s3-recursive-list"; @@ -242,14 +243,15 @@ namespace Duplicati.Library.Backend if (!options.ContainsKey("s3-ext-forcepathstyle") && !hostname.EndsWith(".amazonaws.com", StringComparison.OrdinalIgnoreCase)) options["s3-ext-forcepathstyle"] = "true"; - var disableChunkEncoding = Utility.Utility.ParseBoolOption(options, S3_DISABLE_CHUNK_ENCODING_OPTION); var s3ClientOptionValue = options.GetValueOrDefault(S3_CLIENT_OPTION); (var awsID, var awsKey) = auth.GetCredentials(); if (string.IsNullOrWhiteSpace(s3ClientOptionValue) || string.Equals(s3ClientOptionValue, "aws", StringComparison.OrdinalIgnoreCase)) { - m_s3Client = new S3AwsClient(awsID, awsKey, locationConstraint, hostname, storageClass, useSSL, disableChunkEncoding, timeout, options); + var disableChunkEncoding = Utility.Utility.ParseBoolOption(options, S3_DISABLE_CHUNK_ENCODING_OPTION); + var disablePayloadSigning = Utility.Utility.ParseBoolOption(options, S3_DISABLE_PAYLOAD_SIGNING_OPTION); + m_s3Client = new S3AwsClient(awsID, awsKey, locationConstraint, hostname, storageClass, useSSL, disableChunkEncoding, disablePayloadSigning, timeout, options); } else if (string.Equals(s3ClientOptionValue, "minio", StringComparison.OrdinalIgnoreCase)) { @@ -336,6 +338,7 @@ namespace Duplicati.Library.Backend new CommandLineArgument(SSL_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionUseSSLShort, Strings.S3Backend.DescriptionUseSSLLong), new CommandLineArgument(S3_CLIENT_OPTION, CommandLineArgument.ArgumentType.Enumeration, Strings.S3Backend.S3ClientDescriptionShort, Strings.S3Backend.S3ClientDescriptionLong, "aws", null, new string[] { "aws", "minio" }), new CommandLineArgument(S3_DISABLE_CHUNK_ENCODING_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionDisableChunkEncodingShort, Strings.S3Backend.DescriptionDisableChunkEncodingLong, "false"), + new CommandLineArgument(S3_DISABLE_PAYLOAD_SIGNING_OPTION, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionDisablePayloadSigningShort, Strings.S3Backend.DescriptionDisablePayloadSigningLong, "false"), new CommandLineArgument(S3AwsClient.S3_ARCHIVE_CLASSES_OPTION, CommandLineArgument.ArgumentType.Flags, Strings.S3Backend.S3ArchiveClassesDescriptionShort, Strings.S3Backend.S3ArchiveClassesDescriptionLong, string.Join(",", S3AwsClient.DEFAULT_ARCHIVE_CLASSES.Select(x => x.Value)), null, KNOWN_S3_STORAGE_CLASSES.Select(x => x.Value).WhereNotNullOrWhiteSpace().ToArray()), new CommandLineArgument(S3_LIST_API_VERSION_OPTION, CommandLineArgument.ArgumentType.Enumeration, Strings.S3Backend.DescriptionListApiVersionShort, Strings.S3Backend.DescriptionListApiVersionLong, "v1", null, ["v1", "v2"]), new CommandLineArgument(S3_RECURSIVE_LIST, CommandLineArgument.ArgumentType.Boolean, Strings.S3Backend.DescriptionRecursiveListShort, Strings.S3Backend.DescriptionRecursiveListLong, "false"), diff --git a/Duplicati/Library/Utility/OverrideableStream.cs b/Duplicati/Library/Utility/OverrideableStream.cs index 35e8a5273..b331c0b2f 100644 --- a/Duplicati/Library/Utility/OverrideableStream.cs +++ b/Duplicati/Library/Utility/OverrideableStream.cs @@ -37,6 +37,11 @@ namespace Duplicati.Library.Utility /// protected Stream m_basestream; + /// + /// The base stream that is wrapped + /// + public Stream BaseStream => m_basestream; + /// /// Creates a new instance /// diff --git a/Duplicati/Library/Utility/Utility.cs b/Duplicati/Library/Utility/Utility.cs index 7f54fb5a4..d29f548ab 100644 --- a/Duplicati/Library/Utility/Utility.cs +++ b/Duplicati/Library/Utility/Utility.cs @@ -23,6 +23,7 @@ using System; using System.Collections.Generic; +using System.Diagnostics.CodeAnalysis; using System.Globalization; using System.IO; using System.Linq; @@ -1397,7 +1398,8 @@ namespace Duplicati.Library.Utility /// The wrapped commandline element. /// The argument to wrap. /// A flag indicating if environment variables are allowed to be expanded - public static string WrapCommandLineElement(string? arg, bool allowEnvExpansion) + [return: NotNullIfNotNull("arg")] + public static string? WrapCommandLineElement(string? arg, bool allowEnvExpansion) { if (string.IsNullOrWhiteSpace(arg)) return arg; @@ -1847,5 +1849,90 @@ namespace Duplicati.Library.Utility : Timeout.Infinite }; } + + /// + /// The types of streams that are considered basic (i.e. not wrapped) + /// + private static readonly IReadOnlySet _basicStreamTypes = new HashSet + { + typeof(FileStream), + typeof(MemoryStream), + typeof(NetworkStream), + typeof(BufferedStream), + typeof(System.IO.Compression.DeflateStream), + typeof(System.IO.Compression.GZipStream), + typeof(System.IO.Compression.ZLibStream) + }; + + /// + /// Unwraps a stream from layers of wrapping streams + /// + /// The stream to unwrap + /// The unwrapped stream + public static Stream UnwrapThrottledStream(this Stream stream) + { + if (stream == null) + throw new ArgumentNullException(nameof(stream)); + + var previousStream = stream; + + do + { + previousStream = stream; + + while (stream is WrappingStream wrappingStream) + stream = wrappingStream.BaseStream; + while (stream is OverrideableStream overrideableStream) + stream = overrideableStream.BaseStream; + + } while (stream != previousStream); + +#if DEBUG + if (!_basicStreamTypes.Contains(stream.GetType())) + throw new InvalidOperationException($"The unwrapped stream is not a basic stream, but a {stream.GetType()}"); +#endif + + return stream; + } + + /// + /// Calculates the hash of a throttled stream and returns the stream to read + /// + /// The source stream + /// The hash algorithm to use + /// The cancellation token to observe + /// A tuple with the stream, the hash and a temporary file if used + public static async Task<(Stream content, string hash, TempFile? tmpfile)> CalculateThrottledStreamHash(Stream stream, string hashalgorithm, CancellationToken cancelToken) + { + TempFile? tmp = null; + string contentHash; + var measure = stream.UnwrapThrottledStream(); + if (measure.CanSeek) + { + var p = measure.Position; + + // Compute the hash + using (var hashalg = HashFactory.CreateHasher(hashalgorithm)) + contentHash = ByteArrayAsHexString(hashalg.ComputeHash(measure)); + + measure.Position = p; + } + else + { + // No seeking possible, use a temp file + tmp = new TempFile(); + await using (var sr = File.OpenWrite(tmp)) + using (var hasher = HashFactory.CreateHasher(hashalgorithm)) + await using (var hc = new HashCalculatingStream(measure, hasher)) + { + await CopyStreamAsync(hc, sr, cancelToken).ConfigureAwait(false); + contentHash = hc.GetFinalHashString(); + } + + stream = File.OpenRead(tmp); + } + + return (stream, contentHash, tmp); + } } }