diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs index 332d020933..8195178bcb 100644 --- a/src/ModularPipelines/Context/Downloader.cs +++ b/src/ModularPipelines/Context/Downloader.cs @@ -25,14 +25,14 @@ public Downloader(IModuleLoggerProvider moduleLoggerProvider, IHttpContext http, public async Task DownloadStringAsync(DownloadOptions options, CancellationToken cancellationToken = default) { - var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); + using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); return await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); } public async Task DownloadFileAsync(DownloadFileOptions options, CancellationToken cancellationToken = default) { - var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); + using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); await using (stream.ConfigureAwait(false)) @@ -44,10 +44,26 @@ public async Task DownloadFileAsync(DownloadFileOptions options, Cancellat throw new IOException($"{filePathToSave} already exists and overwrite is false"); } - var newFile = _fileSystemProvider.Create(filePathToSave); - await using (newFile.ConfigureAwait(false)) + var destinationDirectory = Path.GetDirectoryName(filePathToSave); + var temporaryPath = Path.Combine( + destinationDirectory ?? string.Empty, + _fileSystemProvider.GetRandomFileName()); + try { - await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false); + var newFile = _fileSystemProvider.Create(temporaryPath); + await using (newFile.ConfigureAwait(false)) + { + await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false); + } + + _fileSystemProvider.MoveFile(temporaryPath, filePathToSave, options.Overwrite); + } + finally + { + if (_fileSystemProvider.FileExists(temporaryPath)) + { + _fileSystemProvider.DeleteFile(temporaryPath); + } } _moduleLoggerProvider.GetLogger().LogInformation("File {Uri} downloaded to {SaveLocation}", options.DownloadUri, filePathToSave); @@ -68,7 +84,15 @@ public async Task DownloadResponseAsync(DownloadOptions opt HttpClient = options.HttpClient, }, cancellationToken).ConfigureAwait(false); - return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + try + { + return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + } + catch + { + response.Dispose(); + throw; + } } /// diff --git a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs index 9c544b37df..7313398da8 100644 --- a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs @@ -8,38 +8,68 @@ public interface IFileSystemProvider { // File read operations Task ReadAllTextAsync(string path, CancellationToken cancellationToken = default); + IAsyncEnumerable ReadLinesAsync(string path, CancellationToken cancellationToken = default); + Task ReadAllBytesAsync(string path, CancellationToken cancellationToken = default); // File write operations Task WriteAllTextAsync(string path, string contents, CancellationToken cancellationToken = default); + Task WriteAllBytesAsync(string path, byte[] contents, CancellationToken cancellationToken = default); + Task WriteAllLinesAsync(string path, IEnumerable contents, CancellationToken cancellationToken = default); + Task AppendAllTextAsync(string path, string contents, CancellationToken cancellationToken = default); + Task AppendAllLinesAsync(string path, IEnumerable contents, CancellationToken cancellationToken = default); // File stream operations Stream OpenRead(string path); + Stream Create(string path); + Stream Open(string path, FileMode mode, FileAccess access); // File management operations void DeleteFile(string path); + void CopyFile(string sourcePath, string destinationPath, bool overwrite); + void MoveFile(string sourcePath, string destinationPath); + + void MoveFile(string sourcePath, string destinationPath, bool overwrite) + { + if (overwrite && FileExists(destinationPath)) + { + throw new NotSupportedException( + "This file system provider does not support atomic overwrite moves."); + } + + MoveFile(sourcePath, destinationPath); + } + bool FileExists(string path); // Directory operations void CreateDirectory(string path); + void DeleteDirectory(string path, bool recursive); + void MoveDirectory(string sourcePath, string destinationPath); + bool DirectoryExists(string path); + IEnumerable EnumerateFiles(string path, string searchPattern, SearchOption searchOption); + IEnumerable EnumerateDirectories(string path, string searchPattern, SearchOption searchOption); // Path operations string GetTempPath(); + string GetRandomFileName(); + string Combine(params string[] paths); + string GetRelativePath(string relativeTo, string path); } diff --git a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs index a245065efe..c7bd3a9ec7 100644 --- a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs @@ -9,11 +9,13 @@ namespace ModularPipelines.FileSystem; public sealed class SystemFileSystemProvider : IFileSystemProvider { /// - /// Singleton instance for use when no DI is available. + /// Gets singleton instance for use when no DI is available. /// public static SystemFileSystemProvider Instance { get; } = new(); - private SystemFileSystemProvider() { } + private SystemFileSystemProvider() + { + } // File read operations public Task ReadAllTextAsync(string path, CancellationToken cancellationToken = default) @@ -61,6 +63,9 @@ public void CopyFile(string sourcePath, string destinationPath, bool overwrite) public void MoveFile(string sourcePath, string destinationPath) => System.IO.File.Move(sourcePath, destinationPath); + public void MoveFile(string sourcePath, string destinationPath, bool overwrite) + => System.IO.File.Move(sourcePath, destinationPath, overwrite); + public bool FileExists(string path) => System.IO.File.Exists(path); diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index 170d14a90d..7517ce645b 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -28,36 +28,82 @@ public Http( public async Task SendAsync(HttpOptions httpOptions, CancellationToken cancellationToken = default) { - // Priority: options property > pipeline default > no timeout - var effectiveTimeout = httpOptions.Timeout ?? _pipelineOptions.Value.DefaultHttpTimeout; + var httpClient = httpOptions.HttpClient ?? GetHttpClient(httpOptions.LoggingType); + + // Priority: options property > pipeline default > HttpClient timeout + var effectiveTimeout = httpOptions.Timeout + ?? _pipelineOptions.Value.DefaultHttpTimeout + ?? GetFiniteTimeout(httpClient); // Create timeout token if timeout is specified - using var timeoutCts = effectiveTimeout.HasValue + var timeoutCts = effectiveTimeout.HasValue ? new CancellationTokenSource(effectiveTimeout.Value) : null; - // Link the timeout token with the passed cancellation token, or just use the original if no timeout - using var linkedCts = timeoutCts != null + // Keep caller cancellation alive through streamed body consumption even when no + // finite timeout applies. + var linkedCts = timeoutCts != null ? CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken) - : null; + : cancellationToken.CanBeCanceled + ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken) + : null; var effectiveCancellationToken = linkedCts?.Token ?? cancellationToken; - if (httpOptions.HttpClient != null) + try { - return await SendAndWrapLogging(httpOptions, effectiveCancellationToken).ConfigureAwait(false); + HttpResponseMessage WrapResponse(HttpResponseMessage response) + { + if (linkedCts is null) + { + return response; + } + + response.Content = new TimeoutHttpContent(response.Content, timeoutCts, linkedCts); + timeoutCts = null; + linkedCts = null; + return response; + } + + if (httpOptions.HttpClient != null) + { + return await SendAndWrapLogging( + httpClient, + httpOptions, + WrapResponse, + effectiveCancellationToken) + .ConfigureAwait(false); + } + + var response = await httpClient.SendAsync( + httpOptions.HttpRequestMessage, + HttpCompletionOption.ResponseHeadersRead, + effectiveCancellationToken) + .ConfigureAwait(false); + + response = WrapResponse(response); + try + { + if (!httpOptions.ThrowOnNonSuccessStatusCode) + { + return response; + } + + return await response + .EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken) + .ConfigureAwait(false); + } + catch + { + response.Dispose(); + throw; + } } - - var httpClient = GetHttpClient(httpOptions.LoggingType); - - var response = await httpClient.SendAsync(httpOptions.HttpRequestMessage, effectiveCancellationToken).ConfigureAwait(false); - - if (!httpOptions.ThrowOnNonSuccessStatusCode) + finally { - return response; + linkedCts?.Dispose(); + timeoutCts?.Dispose(); } - - return await response.EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken).ConfigureAwait(false); } private HttpClient GetHttpClient(HttpLoggingType loggingType) @@ -66,34 +112,60 @@ private HttpClient GetHttpClient(HttpLoggingType loggingType) return _httpClientFactory.CreateClient(clientName); } - private async Task SendAndWrapLogging(HttpOptions httpOptions, CancellationToken cancellationToken) + private async Task SendAndWrapLogging( + HttpClient httpClient, + HttpOptions httpOptions, + Func wrapResponse, + CancellationToken cancellationToken) { var logger = _moduleLoggerProvider.GetLogger(); var loggingOptions = GetEffectiveLoggingOptions(httpOptions); if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Request)) { - await _httpLogger.PrintRequest(httpOptions.HttpRequestMessage, logger, loggingOptions).ConfigureAwait(false); + await _httpLogger + .PrintRequest( + httpOptions.HttpRequestMessage, + logger, + loggingOptions, + cancellationToken) + .ConfigureAwait(false); } var stopWatch = Stopwatch.StartNew(); - var response = await httpOptions.HttpClient!.SendAsync(httpOptions.HttpRequestMessage, cancellationToken).ConfigureAwait(false); + var response = await httpClient.SendAsync( + httpOptions.HttpRequestMessage, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken) + .ConfigureAwait(false); - LogStatusCode(response.StatusCode, httpOptions, loggingOptions); - LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions); - - if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response)) + try { - await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false); - } + LogStatusCode(response.StatusCode, httpOptions, loggingOptions); + LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions); - if (!httpOptions.ThrowOnNonSuccessStatusCode) + if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response)) + { + await _httpLogger + .PrintResponse(response, logger, loggingOptions, cancellationToken) + .ConfigureAwait(false); + } + + response = wrapResponse(response); + + if (!httpOptions.ThrowOnNonSuccessStatusCode) + { + return response; + } + + return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + } + catch { - return response; + response.Dispose(); + throw; } - - return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); } private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions) @@ -112,6 +184,13 @@ private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions) return HttpLoggingOptions.Default; } + private static TimeSpan? GetFiniteTimeout(HttpClient httpClient) + { + return httpClient.Timeout == System.Threading.Timeout.InfiniteTimeSpan + ? null + : httpClient.Timeout; + } + private void LogDuration(TimeSpan duration, HttpOptions httpOptions, HttpLoggingOptions loggingOptions) { if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Duration) && loggingOptions.LogDuration) diff --git a/src/ModularPipelines/Http/HttpBodySecretRedactor.cs b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs new file mode 100644 index 0000000000..403bdc7776 --- /dev/null +++ b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs @@ -0,0 +1,62 @@ +using Microsoft.Extensions.Options; +using ModularPipelines.Constants; +using ModularPipelines.Engine; +using ModularPipelines.Options; + +namespace ModularPipelines.Http; + +internal static class HttpBodySecretRedactor +{ + public static string Redact( + string body, + bool isTruncated, + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions maskingOptions) + { + var boundarySafeBody = isTruncated + ? RedactPartialSecretAtBoundary(body, secretProvider.GetSnapshot().Secrets, maskingOptions.Value) + : body; + return secretObfuscator.Obfuscate(boundarySafeBody, null); + } + + private static string RedactPartialSecretAtBoundary( + string body, + IReadOnlyList secrets, + SecretMaskingOptions options) + { + if (body.Length == 0) + { + return body; + } + + var comparison = options.CaseInsensitive + ? StringComparison.OrdinalIgnoreCase + : StringComparison.Ordinal; + var longestPartialMatch = 0; + + foreach (var secret in secrets) + { + var maximumLength = Math.Min(body.Length, secret.Length - 1); + for (var length = maximumLength; length > longestPartialMatch; length--) + { + if (body.AsSpan(body.Length - length, length) + .Equals(secret.AsSpan(0, length), comparison)) + { + longestPartialMatch = length; + break; + } + } + } + + if (longestPartialMatch == 0) + { + return body; + } + + var mask = string.IsNullOrWhiteSpace(options.MaskValue) + ? LoggingConstants.SecretMask + : options.MaskValue; + return string.Concat(body.AsSpan(0, body.Length - longestPartialMatch), mask); + } +} diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs new file mode 100644 index 0000000000..2ab2ef7a1c --- /dev/null +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -0,0 +1,293 @@ +using System.Buffers; +using System.Text; + +namespace ModularPipelines.Http; + +/// +/// Reads bounded text previews while preserving the complete content stream for its consumer. +/// +internal static class HttpContentPreviewReader +{ + private const int MaximumPreviewBufferSize = 81920; + + /// + /// Reads at most one byte beyond the configured preview limit. + /// + /// The HTTP content to preview. + /// The maximum number of bytes to include in the preview. + /// The cancellation token. + /// The preview, truncation state, replayable content, and declared total length. + public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)> + ReadAsync( + HttpContent content, + int maxBytes, + CancellationToken cancellationToken) + { + if (maxBytes <= 0 || maxBytes == int.MaxValue) + { + var totalLength = content.Headers.ContentLength; + var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false); + var unboundedReplayContent = new ByteArrayContent(bytes); + CopyHeaders(content, unboundedReplayContent); + content.Dispose(); + return (DecodeCompletePrefix(GetEncoding(unboundedReplayContent), bytes, bytes.Length), + false, + unboundedReplayContent, + totalLength ?? bytes.LongLength); + } + + var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + using var cancellationRegistration = cancellationToken.Register( + static state => ((Stream) state!).Dispose(), + stream); + var probeLength = maxBytes + 1; + var readBuffer = ArrayPool.Shared.Rent( + Math.Min(probeLength, MaximumPreviewBufferSize)); + using var buffer = new MemoryStream(GetInitialCapacity(content, probeLength)); + + try + { + while (buffer.Length < probeLength) + { + var bytesRemaining = probeLength - (int) buffer.Length; + var read = await stream.ReadAsync( + readBuffer.AsMemory( + 0, + Math.Min(MaximumPreviewBufferSize, bytesRemaining)), + cancellationToken) + .ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + if (read == 0) + { + break; + } + + buffer.Write(readBuffer, 0, read); + } + } + catch (Exception exception) + when (cancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP content preview read was cancelled.", + exception, + cancellationToken); + } + finally + { + ArrayPool.Shared.Return(readBuffer); + } + + var bufferedBytes = buffer.ToArray(); + var bytesRead = bufferedBytes.Length; + var isTruncated = bytesRead > maxBytes; + var previewLength = Math.Min(bytesRead, maxBytes); + var encoding = GetEncoding(content); + var preview = DecodeCompletePrefix(encoding, bufferedBytes, previewLength); + var replayContent = CreateReplayContent(content, stream, bufferedBytes); + + return (preview, isTruncated, replayContent, content.Headers.ContentLength); + } + + /// + /// Reads request content into replayable storage so redirects and authentication retries can resend it. + /// + /// The HTTP request content to preview. + /// The maximum number of bytes to include in the preview. + /// The cancellation token. + /// The preview, truncation state, replayable content, and total length. + /// This method takes ownership of and disposes . + public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)> + ReadReplayableAsync( + HttpContent content, + int maxBytes, + CancellationToken cancellationToken) + { + var totalLength = content.Headers.ContentLength; + var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false); + var previewLength = maxBytes <= 0 || maxBytes == int.MaxValue + ? bytes.Length + : Math.Min(bytes.Length, maxBytes); + var isTruncated = previewLength < bytes.Length; + var preview = DecodeCompletePrefix(GetEncoding(content), bytes, previewLength); + var replayContent = new ByteArrayContent(bytes); + CopyHeaders(content, replayContent); + content.Dispose(); + + return ( + preview, + isTruncated, + replayContent, + totalLength ?? bytes.LongLength); + } + + private static async Task ReadAllBytesAsync( + HttpContent content, + CancellationToken cancellationToken) + { + var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + using var cancellationRegistration = cancellationToken.Register( + static state => ((Stream) state!).Dispose(), + stream); + using var buffer = new MemoryStream(); + + try + { + await stream.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + return buffer.ToArray(); + } + catch (Exception exception) + when (cancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP content read was cancelled.", + exception, + cancellationToken); + } + } + + private static int GetInitialCapacity(HttpContent content, int probeLength) + { + var declaredLength = content.Headers.ContentLength; + return declaredLength is >= 0 + ? (int) Math.Min(Math.Min(declaredLength.Value, probeLength), MaximumPreviewBufferSize) + : Math.Min(probeLength, MaximumPreviewBufferSize); + } + + private static Encoding GetEncoding(HttpContent content) + { + var charset = content.Headers.ContentType?.CharSet?.Trim('"'); + if (string.IsNullOrWhiteSpace(charset)) + { + return Encoding.UTF8; + } + + try + { + return Encoding.GetEncoding(charset); + } + catch (ArgumentException) + { + return Encoding.UTF8; + } + } + + private static string DecodeCompletePrefix(Encoding encoding, byte[] buffer, int length) + { + var decoder = encoding.GetDecoder(); + var characters = new char[encoding.GetMaxCharCount(length)]; + decoder.Convert( + buffer.AsSpan(0, length), + characters, + flush: false, + out _, + out var charactersUsed, + out _); + return new string(characters, 0, charactersUsed); + } + + private static StreamContent CreateReplayContent( + HttpContent originalContent, + Stream remainingStream, + byte[] prefix) + { + var replayContent = new StreamContent( + new PrefixReplayStream(prefix, remainingStream, originalContent)); + CopyHeaders(originalContent, replayContent); + + return replayContent; + } + + private static void CopyHeaders(HttpContent source, HttpContent destination) + { + foreach (var header in source.Headers) + { + destination.Headers.TryAddWithoutValidation(header.Key, header.Value); + } + } + + private sealed class PrefixReplayStream( + byte[] prefix, + Stream remainingStream, + HttpContent ownedContent) : Stream + { + private int _prefixPosition; + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + var prefixBytesRead = ReadPrefix(buffer.AsSpan(offset, count)); + return prefixBytesRead > 0 + ? prefixBytesRead + : remainingStream.Read(buffer, offset, count); + } + + public override Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + var prefixBytesRead = ReadPrefix(buffer.Span); + return prefixBytesRead > 0 + ? prefixBytesRead + : await remainingStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + protected override void Dispose(bool disposing) + { + if (disposing) + { + ownedContent.Dispose(); + } + + base.Dispose(disposing); + } + + private int ReadPrefix(Span destination) + { + var bytesToCopy = Math.Min(destination.Length, prefix.Length - _prefixPosition); + if (bytesToCopy <= 0) + { + return 0; + } + + prefix.AsSpan(_prefixPosition, bytesToCopy).CopyTo(destination); + _prefixPosition += bytesToCopy; + return bytesToCopy; + } + } +} diff --git a/src/ModularPipelines/Http/HttpLogger.cs b/src/ModularPipelines/Http/HttpLogger.cs index 4f08d1c917..5ffe4d0467 100644 --- a/src/ModularPipelines/Http/HttpLogger.cs +++ b/src/ModularPipelines/Http/HttpLogger.cs @@ -28,7 +28,20 @@ public HttpLogger(IHttpRequestFormatter requestFormatter, IHttpResponseFormatter /// A representing the asynchronous operation. public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) { - return PrintRequest(request, logger, HttpLoggingOptions.Default); + return PrintRequest( + request, + logger, + HttpLoggingOptions.Default, + CancellationToken.None); + } + + /// + public Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger, HttpLoggingOptions.Default, cancellationToken); } /// @@ -38,14 +51,26 @@ public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) /// The current module logger. /// Options controlling what parts of the request to log. /// A representing the asynchronous operation. - public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options) + public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options) + { + return PrintRequest(request, logger, options, CancellationToken.None); + } + + /// + public async Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) { if (!options.LogRequest) { return; } - var formattedRequest = await _requestFormatter.FormatAsync(request, options).ConfigureAwait(false); + var formattedRequest = await _requestFormatter + .FormatAsync(request, options, cancellationToken) + .ConfigureAwait(false); logger.LogInformation("HTTP Request:\n{Request}", formattedRequest); } @@ -57,7 +82,20 @@ public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, /// A representing the asynchronous operation. public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) { - return PrintResponse(response, logger, HttpLoggingOptions.Default); + return PrintResponse( + response, + logger, + HttpLoggingOptions.Default, + CancellationToken.None); + } + + /// + public Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger, HttpLoggingOptions.Default, cancellationToken); } /// @@ -67,20 +105,32 @@ public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) /// The current module logger. /// Options controlling what parts of the response to log. /// A representing the asynchronous operation. - public async Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options) + public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options) + { + return PrintResponse(response, logger, options, CancellationToken.None); + } + + /// + public async Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) { if (!options.LogResponse) { return; } - var formattedResponse = await _responseFormatter.FormatAsync(response, options).ConfigureAwait(false); + var formattedResponse = await _responseFormatter + .FormatAsync(response, options, cancellationToken) + .ConfigureAwait(false); logger.LogInformation("HTTP Response:\n{Response}", formattedResponse); } public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger) { - var statusCode = httpStatusCode == null ? null as int? : (int)httpStatusCode; + var statusCode = httpStatusCode == null ? null as int? : (int) httpStatusCode; var icon = statusCode is >= 200 and < 300 ? "+" : "x"; logger.LogInformation("{Icon} HTTP Status: {StatusCode} {HttpStatusCode}", icon, statusCode, httpStatusCode); diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs index 6907bb2d97..7512223da8 100644 --- a/src/ModularPipelines/Http/HttpRequestFormatter.cs +++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs @@ -1,5 +1,6 @@ using System.Net.Http.Headers; using System.Text; +using Microsoft.Extensions.Options; using ModularPipelines.Constants; using ModularPipelines.Engine; using ModularPipelines.Options; @@ -31,10 +32,17 @@ namespace ModularPipelines.Http; internal class HttpRequestFormatter : IHttpRequestFormatter { private readonly ISecretObfuscator _secretObfuscator; + private readonly ISecretProvider _secretProvider; + private readonly IOptions _secretMaskingOptions; - public HttpRequestFormatter(ISecretObfuscator secretObfuscator) + public HttpRequestFormatter( + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions secretMaskingOptions) { _secretObfuscator = secretObfuscator; + _secretProvider = secretProvider; + _secretMaskingOptions = secretMaskingOptions; } private static readonly HashSet TextMediaTypes = new(StringComparer.OrdinalIgnoreCase) @@ -70,7 +78,7 @@ public async Task FormatAsync(HttpRequestMessage request, HttpLoggingOpt if (options.LogRequestBody) { - await AppendBodyAsync(sb, request.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); + await AppendBodyAsync(sb, request, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); } return sb.ToString(); @@ -127,10 +135,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList s return false; } - private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken) + private async Task AppendBodyAsync( + StringBuilder sb, + HttpRequestMessage request, + int maxBodySize, + CancellationToken cancellationToken) { sb.AppendLine("Body"); + var content = request.Content; if (content == null) { sb.AppendLine("\t(null)"); @@ -146,29 +159,43 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m return; } - var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader + .ReadReplayableAsync(content, maxBodySize, cancellationToken) + .ConfigureAwait(false); + request.Content = replayContent; - if (string.IsNullOrWhiteSpace(body)) + if (string.IsNullOrWhiteSpace(body) && !isTruncated) { sb.AppendLine("\t(empty)"); return; } // Obfuscate sensitive values in the body - var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); + var obfuscatedBody = HttpBodySecretRedactor.Redact( + body, + isTruncated, + _secretObfuscator, + _secretProvider, + _secretMaskingOptions); - // Truncate if body is too large - if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize) + sb.AppendLine($"\t{obfuscatedBody}"); + if (isTruncated) { - sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}"); - sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]"); - } - else - { - sb.AppendLine($"\t{obfuscatedBody}"); + AppendTruncationMessage(sb, maxBodySize, totalLength); } } + private static void AppendTruncationMessage( + StringBuilder sb, + int maxBodySize, + long? totalLength) + { + var size = totalLength.HasValue + ? $"total size: {totalLength.Value:N0} bytes" + : $"after {maxBodySize:N0} bytes"; + sb.AppendLine($"\t... [truncated, {size}]"); + } + private static bool IsBinaryContent(HttpContent content) { var contentType = content.Headers.ContentType?.MediaType; diff --git a/src/ModularPipelines/Http/HttpResponseExtensions.cs b/src/ModularPipelines/Http/HttpResponseExtensions.cs index d4cfff64de..c4123c4543 100644 --- a/src/ModularPipelines/Http/HttpResponseExtensions.cs +++ b/src/ModularPipelines/Http/HttpResponseExtensions.cs @@ -30,28 +30,20 @@ public static async Task EnsureSuccessStatusCodeWithContent { responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); } - catch (ObjectDisposedException) - { - // Response content stream was already disposed - continue with null content - } - catch (InvalidOperationException) - { - // Content stream is in an invalid state - continue with null content - } - catch (HttpRequestException) - { - // Network error while reading content - continue with null content - } catch (OperationCanceledException) { - // Reading was cancelled - continue with null content + throw; } catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException)) { - // Other unexpected errors reading content - continue with null content + cancellationToken.ThrowIfCancellationRequested(); + + // Errors reading content do not hide the primary HTTP status failure. // The primary error information is in the status code and reason phrase } + cancellationToken.ThrowIfCancellationRequested(); + throw new HttpResponseException( response.StatusCode, response.ReasonPhrase, diff --git a/src/ModularPipelines/Http/HttpResponseFormatter.cs b/src/ModularPipelines/Http/HttpResponseFormatter.cs index 886a17d963..3c49af94aa 100644 --- a/src/ModularPipelines/Http/HttpResponseFormatter.cs +++ b/src/ModularPipelines/Http/HttpResponseFormatter.cs @@ -1,5 +1,6 @@ using System.Net.Http.Headers; using System.Text; +using Microsoft.Extensions.Options; using ModularPipelines.Constants; using ModularPipelines.Engine; using ModularPipelines.Options; @@ -31,10 +32,17 @@ namespace ModularPipelines.Http; internal class HttpResponseFormatter : IHttpResponseFormatter { private readonly ISecretObfuscator _secretObfuscator; + private readonly ISecretProvider _secretProvider; + private readonly IOptions _secretMaskingOptions; - public HttpResponseFormatter(ISecretObfuscator secretObfuscator) + public HttpResponseFormatter( + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions secretMaskingOptions) { _secretObfuscator = secretObfuscator; + _secretProvider = secretProvider; + _secretMaskingOptions = secretMaskingOptions; } private static readonly HashSet TextMediaTypes = new(StringComparer.OrdinalIgnoreCase) @@ -70,7 +78,7 @@ public async Task FormatAsync(HttpResponseMessage response, HttpLoggingO if (options.LogResponseBody) { - await AppendBodyAsync(sb, response.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); + await AppendBodyAsync(sb, response, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); } return sb.ToString(); @@ -127,10 +135,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList s return false; } - private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken) + private async Task AppendBodyAsync( + StringBuilder sb, + HttpResponseMessage response, + int maxBodySize, + CancellationToken cancellationToken) { sb.AppendLine("Body"); + var content = response.Content; if (content == null) { sb.AppendLine("\t(null)"); @@ -146,29 +159,43 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m return; } - var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader + .ReadAsync(content, maxBodySize, cancellationToken) + .ConfigureAwait(false); + response.Content = replayContent; - if (string.IsNullOrWhiteSpace(body)) + if (string.IsNullOrWhiteSpace(body) && !isTruncated) { sb.AppendLine("\t(empty)"); return; } // Obfuscate sensitive values in the body - var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); + var obfuscatedBody = HttpBodySecretRedactor.Redact( + body, + isTruncated, + _secretObfuscator, + _secretProvider, + _secretMaskingOptions); - // Truncate if body is too large - if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize) + sb.AppendLine($"\t{obfuscatedBody}"); + if (isTruncated) { - sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}"); - sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]"); - } - else - { - sb.AppendLine($"\t{obfuscatedBody}"); + AppendTruncationMessage(sb, maxBodySize, totalLength); } } + private static void AppendTruncationMessage( + StringBuilder sb, + int maxBodySize, + long? totalLength) + { + var size = totalLength.HasValue + ? $"total size: {totalLength.Value:N0} bytes" + : $"after {maxBodySize:N0} bytes"; + sb.AppendLine($"\t... [truncated, {size}]"); + } + private static bool IsBinaryContent(HttpContent content) { var contentType = content.Headers.ContentType?.MediaType; diff --git a/src/ModularPipelines/Http/IHttpLogger.cs b/src/ModularPipelines/Http/IHttpLogger.cs index 25699c97a1..bd075570f9 100644 --- a/src/ModularPipelines/Http/IHttpLogger.cs +++ b/src/ModularPipelines/Http/IHttpLogger.cs @@ -17,6 +17,21 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintRequest(HttpRequestMessage request, IModuleLogger logger); + /// + /// Prints the HTTP request. + /// + /// The HTTP request to print. + /// The current module logger. + /// A token to cancel request formatting. + /// A representing the asynchronous operation. + Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger).WaitAsync(cancellationToken); + } + /// /// Prints the HTTP request with logging options. /// @@ -26,6 +41,23 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options); + /// + /// Prints the HTTP request with logging options. + /// + /// The HTTP request to print. + /// The current module logger. + /// Options controlling what parts of the request to log. + /// A token to cancel request formatting. + /// A representing the asynchronous operation. + Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger, options).WaitAsync(cancellationToken); + } + /// /// Prints the HTTP response. /// @@ -34,6 +66,21 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintResponse(HttpResponseMessage response, IModuleLogger logger); + /// + /// Prints the HTTP response. + /// + /// The HTTP response to print. + /// The current module logger. + /// A token to cancel response formatting. + /// A representing the asynchronous operation. + Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger).WaitAsync(cancellationToken); + } + /// /// Prints the HTTP response with logging options. /// @@ -43,6 +90,23 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options); + /// + /// Prints the HTTP response with logging options. + /// + /// The HTTP response to print. + /// The current module logger. + /// Options controlling what parts of the response to log. + /// A token to cancel response formatting. + /// A representing the asynchronous operation. + Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger, options).WaitAsync(cancellationToken); + } + /// /// Prints the HTTP status code. /// diff --git a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs index 21aa4271df..8c3603e90f 100644 --- a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs +++ b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs @@ -17,10 +17,12 @@ public RequestLoggingHttpHandler(IModuleLoggerProvider loggerProvider, IHttpLogg protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { var logger = _loggerProvider.GetLogger(); - await _httpLogger.PrintRequest(request, logger).ConfigureAwait(false); + await _httpLogger + .PrintRequest(request, logger, cancellationToken) + .ConfigureAwait(false); var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false); return response; } -} \ No newline at end of file +} diff --git a/src/ModularPipelines/Http/ResilienceHttpHandler.cs b/src/ModularPipelines/Http/ResilienceHttpHandler.cs index 9774c8c6c4..49561e2c2f 100644 --- a/src/ModularPipelines/Http/ResilienceHttpHandler.cs +++ b/src/ModularPipelines/Http/ResilienceHttpHandler.cs @@ -105,14 +105,21 @@ private Task OnRetry(DelegateResult outcome, TimeSpan delay outcome.Exception.GetType().Name, outcome.Exception.Message, retryAttempt, - (int)delay.TotalMilliseconds); + (int) delay.TotalMilliseconds); } else if (outcome.Result != null) { - logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms", - (int)outcome.Result.StatusCode, - retryAttempt, - (int)delay.TotalMilliseconds); + try + { + logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms", + (int) outcome.Result.StatusCode, + retryAttempt, + (int) delay.TotalMilliseconds); + } + finally + { + outcome.Result.Dispose(); + } } return Task.CompletedTask; diff --git a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs index d1d9fce4eb..df3a9c3479 100644 --- a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs +++ b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs @@ -18,14 +18,18 @@ protected override async Task SendAsync(HttpRequestMessage { var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false); - // Buffer the response content so it can be read multiple times. - // Without this, reading the body for logging consumes the stream, - // making it unreadable by subsequent code. See issue #1610. - await response.Content.LoadIntoBufferAsync().ConfigureAwait(false); - - var logger = _loggerProvider.GetLogger(); - await _httpLogger.PrintResponse(response, logger).ConfigureAwait(false); - - return response; + try + { + var logger = _loggerProvider.GetLogger(); + await _httpLogger + .PrintResponse(response, logger, cancellationToken) + .ConfigureAwait(false); + return response; + } + catch + { + response.Dispose(); + throw; + } } } diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs new file mode 100644 index 0000000000..1c7dbcf5ed --- /dev/null +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -0,0 +1,300 @@ +using System.Net; + +namespace ModularPipelines.Http; + +internal sealed class TimeoutHttpContent : HttpContent +{ + private readonly HttpContent _innerContent; + private readonly CancellationTokenSource? _timeoutCancellationTokenSource; + private readonly CancellationTokenSource _linkedCancellationTokenSource; + + public TimeoutHttpContent( + HttpContent innerContent, + CancellationTokenSource? timeoutCancellationTokenSource, + CancellationTokenSource linkedCancellationTokenSource) + { + _innerContent = innerContent; + _timeoutCancellationTokenSource = timeoutCancellationTokenSource; + _linkedCancellationTokenSource = linkedCancellationTokenSource; + + foreach (var header in innerContent.Headers) + { + Headers.TryAddWithoutValidation(header.Key, header.Value); + } + } + + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) + { + return CopyInnerContentToAsync(stream, CancellationToken.None); + } + + protected override async Task SerializeToStreamAsync( + Stream stream, + TransportContext? context, + CancellationToken cancellationToken) + { + await CopyInnerContentToAsync(stream, cancellationToken).ConfigureAwait(false); + } + + protected override async Task CreateContentReadStreamAsync() + { + _linkedCancellationTokenSource.Token.ThrowIfCancellationRequested(); + var stream = await _innerContent + .ReadAsStreamAsync(_linkedCancellationTokenSource.Token) + .ConfigureAwait(false); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + + protected override Stream CreateContentReadStream(CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); + var stream = _innerContent.ReadAsStream(effectiveCancellationToken); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + + protected override async Task CreateContentReadStreamAsync( + CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); + var stream = await _innerContent + .ReadAsStreamAsync(effectiveCancellationToken) + .ConfigureAwait(false); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + + protected override bool TryComputeLength(out long length) + { + if (_innerContent.Headers.ContentLength is { } contentLength) + { + length = contentLength; + return true; + } + + length = 0; + return false; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _innerContent.Dispose(); + _linkedCancellationTokenSource.Dispose(); + _timeoutCancellationTokenSource?.Dispose(); + } + + base.Dispose(disposing); + } + + private async Task CopyInnerContentToAsync( + Stream destination, + CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); + var source = await _innerContent + .ReadAsStreamAsync(effectiveCancellationToken) + .ConfigureAwait(false); + using var cancellationRegistration = effectiveCancellationToken.Register( + static state => ((Stream) state!).Dispose(), + source); + + try + { + await source.CopyToAsync(destination, effectiveCancellationToken).ConfigureAwait(false); + effectiveCancellationToken.ThrowIfCancellationRequested(); + } + catch (Exception exception) + when (effectiveCancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP response body copy was cancelled.", + exception, + effectiveCancellationToken); + } + } + + private CancellationTokenSource? CreateReadCancellationTokenSource( + CancellationToken cancellationToken) + { + return cancellationToken.CanBeCanceled && + cancellationToken != _linkedCancellationTokenSource.Token + ? CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _linkedCancellationTokenSource.Token) + : null; + } + + private sealed class TimeoutReadStream( + Stream innerStream, + CancellationToken timeoutCancellationToken) : Stream + { + private readonly CancellationTokenRegistration _timeoutRegistration = + timeoutCancellationToken.Register( + static state => ((Stream) state!).Dispose(), + innerStream); + + public override bool CanRead => innerStream.CanRead; + + public override bool CanSeek => innerStream.CanSeek; + + public override bool CanWrite => innerStream.CanWrite; + + public override long Length => innerStream.Length; + + public override long Position + { + get => innerStream.Position; + set => innerStream.Position = value; + } + + public override void Flush() => innerStream.Flush(); + + public override Task FlushAsync(CancellationToken cancellationToken) + { + return innerStream.FlushAsync(cancellationToken); + } + + public override int Read(byte[] buffer, int offset, int count) + { + timeoutCancellationToken.ThrowIfCancellationRequested(); + try + { + var bytesRead = innerStream.Read(buffer, offset, count); + timeoutCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; + } + catch (Exception exception) + when (timeoutCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) + { + throw CreateTimeoutException(exception); + } + } + + public override int Read(Span buffer) + { + timeoutCancellationToken.ThrowIfCancellationRequested(); + try + { + var bytesRead = innerStream.Read(buffer); + timeoutCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; + } + catch (Exception exception) + when (timeoutCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) + { + throw CreateTimeoutException(exception); + } + } + + public override Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + using var readCancellationTokenSource = + CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? timeoutCancellationToken; + effectiveCancellationToken.ThrowIfCancellationRequested(); + using var cancellationRegistration = readCancellationTokenSource?.Token.Register( + static state => ((Stream) state!).Dispose(), + innerStream); + try + { + var bytesRead = await innerStream.ReadAsync( + buffer, + effectiveCancellationToken) + .ConfigureAwait(false); + effectiveCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; + } + catch (Exception exception) + when (effectiveCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) + { + throw CreateReadCancellationException( + exception, + effectiveCancellationToken); + } + } + + public override long Seek(long offset, SeekOrigin origin) + { + return innerStream.Seek(offset, origin); + } + + public override void SetLength(long value) => innerStream.SetLength(value); + + public override void Write(byte[] buffer, int offset, int count) + { + innerStream.Write(buffer, offset, count); + } + + public override async ValueTask DisposeAsync() + { + _timeoutRegistration.Dispose(); + await innerStream.DisposeAsync().ConfigureAwait(false); + GC.SuppressFinalize(this); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _timeoutRegistration.Dispose(); + innerStream.Dispose(); + } + + base.Dispose(disposing); + } + + private CancellationTokenSource? CreateReadCancellationTokenSource( + CancellationToken cancellationToken) + { + return cancellationToken.CanBeCanceled && + cancellationToken != timeoutCancellationToken + ? CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + timeoutCancellationToken) + : null; + } + + private OperationCanceledException CreateTimeoutException(Exception exception) + { + return new OperationCanceledException( + "The HTTP response body read timed out.", + exception, + timeoutCancellationToken); + } + + private static OperationCanceledException CreateReadCancellationException( + Exception exception, + CancellationToken cancellationToken) + { + return new OperationCanceledException( + "The HTTP response body read was cancelled.", + exception, + cancellationToken); + } + } +} diff --git a/src/ModularPipelines/Options/DownloadOptions.cs b/src/ModularPipelines/Options/DownloadOptions.cs index 2b16950971..601eda114d 100644 --- a/src/ModularPipelines/Options/DownloadOptions.cs +++ b/src/ModularPipelines/Options/DownloadOptions.cs @@ -21,5 +21,6 @@ public record DownloadOptions(Uri DownloadUri) /// /// Gets the type of HTTP logging to perform during the download. /// - public HttpLoggingType LoggingType { get; init; } = HttpLoggingType.Request | HttpLoggingType.Response | HttpLoggingType.StatusCode | HttpLoggingType.Duration; -} \ No newline at end of file + public HttpLoggingType LoggingType { get; init; } = + HttpLoggingType.Request | HttpLoggingType.StatusCode | HttpLoggingType.Duration; +} diff --git a/src/ModularPipelines/Options/HttpLoggingOptions.cs b/src/ModularPipelines/Options/HttpLoggingOptions.cs index 7f364cef51..916a4e55cc 100644 --- a/src/ModularPipelines/Options/HttpLoggingOptions.cs +++ b/src/ModularPipelines/Options/HttpLoggingOptions.cs @@ -67,7 +67,7 @@ public record HttpLoggingOptions public bool LogDuration { get; init; } = true; /// - /// Gets the maximum body size in characters to log. Default is 4096. + /// Gets the maximum body size in bytes to read and log. Default is 4096. /// Bodies larger than this will be truncated with a message indicating the full size. /// Set to 0 or negative to disable truncation. /// diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs new file mode 100644 index 0000000000..3e0aed2146 --- /dev/null +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -0,0 +1,286 @@ +using System.Net.Http.Headers; +using System.Text; +using ModularPipelines.Constants; +using ModularPipelines.Engine; +using ModularPipelines.Http; +using ModularPipelines.Options; +using Moq; + +namespace ModularPipelines.UnitTests.Context; + +public class HttpFormatterTests +{ + [Test] + public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); + using var response = new HttpResponseMessage + { + Content = CreateTextContent(stream), + }; + var formatter = CreateResponseFormatter(); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 5, + }); + var bytesReadForPreview = stream.BytesRead; + var replayedBody = await response.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(bytesReadForPreview).IsLessThanOrEqualTo(6); + await Assert.That(formatted).Contains("\tabcde"); + await Assert.That(formatted).DoesNotContain("abcdef"); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + [Test] + public async Task ResponseFormatter_DoesNotAllocateEntireLargePreviewLimit() + { + var stream = new CountingReadStream([]); + using var response = new HttpResponseMessage + { + Content = CreateTextContent(stream), + }; + var formatter = CreateResponseFormatter(); + + await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 64 * 1024 * 1024, + }); + + await Assert.That(stream.MaximumRequestedReadSize).IsLessThanOrEqualTo(81920); + } + + [Test] + public async Task RequestFormatter_PreservesBodyForRepeatedSends() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); + var content = CreateTextContent(stream); + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = content, + }; + var formatter = CreateRequestFormatter(); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 5, + }); + using var firstSend = new MemoryStream(); + using var secondSend = new MemoryStream(); + await request.Content.CopyToAsync(firstSend); + await request.Content.CopyToAsync(secondSend); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains("\tabcde"); + await Assert.That(formatted).DoesNotContain("abcdef"); + await Assert.That(Encoding.UTF8.GetString(firstSend.ToArray())).IsEqualTo(body); + await Assert.That(Encoding.UTF8.GetString(secondSend.ToArray())).IsEqualTo(body); + await Assert.That(stream.IsDisposed).IsTrue(); + await Assert.That(content.DisposeCount).IsEqualTo(1); + } + } + + [Test] + public async Task ResponseFormatter_UnboundedPreviewPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))), + }; + var formatter = CreateResponseFormatter(); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 0, + }); + var replayedBody = await response.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains(body); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + [Test] + public async Task RequestFormatter_UnboundedPreviewPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))), + }; + var formatter = CreateRequestFormatter(); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 0, + }); + var replayedBody = await request.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains(body); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + [Test] + public async Task ResponseFormatter_RedactsSecretCrossingPreviewBoundary() + { + const string secret = "secret-value"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateResponseFormatter(secret); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 13, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-secret"); + } + } + + [Test] + public async Task RequestFormatter_RedactsSecretCrossingPreviewBoundary() + { + const string secret = "secret-value"; + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateRequestFormatter(secret); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 13, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-secret"); + } + } + + [Test] + public async Task ResponseFormatter_RedactsSecretWhenPreviewSplitsMultibyteCharacter() + { + const string secret = "abcédef"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateResponseFormatter(secret); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 11, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-abc"); + } + } + + private static CountingStreamContent CreateTextContent(Stream stream) + { + var content = new CountingStreamContent(stream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain") + { + CharSet = "utf-8", + }; + return content; + } + + private sealed class CountingStreamContent(Stream stream) : StreamContent(stream) + { + public int DisposeCount { get; private set; } + + protected override void Dispose(bool disposing) + { + DisposeCount++; + base.Dispose(disposing); + } + } + + private static ISecretObfuscator CreateObfuscator(params string[] secrets) + { + var obfuscator = new Mock(); + obfuscator + .Setup(x => x.Obfuscate(It.IsAny(), It.IsAny())) + .Returns((string? input, object? _) => secrets.Aggregate( + input ?? string.Empty, + (value, secret) => value.Replace( + secret, + LoggingConstants.SecretMask, + StringComparison.Ordinal))); + return obfuscator.Object; + } + + private static HttpRequestFormatter CreateRequestFormatter(params string[] secrets) + { + return new HttpRequestFormatter( + CreateObfuscator(secrets), + CreateSecretProvider(secrets), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())); + } + + private static HttpResponseFormatter CreateResponseFormatter(params string[] secrets) + { + return new HttpResponseFormatter( + CreateObfuscator(secrets), + CreateSecretProvider(secrets), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())); + } + + private static ISecretProvider CreateSecretProvider(string[] secrets) + { + var secretProvider = new Mock(); + secretProvider + .Setup(x => x.GetSnapshot()) + .Returns(new SecretSnapshot(0, secrets)); + return secretProvider.Object; + } + + private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents) + { + public int BytesRead { get; private set; } + + public int MaximumRequestedReadSize { get; private set; } + + public bool IsDisposed { get; private set; } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + MaximumRequestedReadSize = Math.Max(MaximumRequestedReadSize, buffer.Length); + var read = await base.ReadAsync(buffer, cancellationToken); + BytesRead += read; + return read; + } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } +} diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 2873eeb409..8bbbfeafb6 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -1,11 +1,17 @@ +using System.Net; +using System.Net.Http.Headers; using System.Reflection; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.Engine; +using ModularPipelines.Extensions; using ModularPipelines.Http; +using ModularPipelines.Logging; using ModularPipelines.Options; using ModularPipelines.TestHelpers; using ModularPipelines.UnitTests.Helpers; +using Moq; using NReco.Logging.File; using File = System.IO.File; using LogLevel = Microsoft.Extensions.Logging.LogLevel; @@ -14,6 +20,665 @@ namespace ModularPipelines.UnitTests.Context; public class HttpTests : TestBase { + [Test] + public async Task SendAsync_ReturnsAfterHeadersWithoutBufferingResponseBody() + { + var content = new BlockingHttpContent(); + var handler = new ImmediateResponseHandler(content); + using var httpClient = new HttpClient(handler); + var result = await GetService((_, _) => { }); + var sendTask = result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/large-file")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + }); + + await handler.RequestReceived.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + HttpResponseMessage? response = null; + try + { + response = await sendTask.WaitAsync(TimeSpan.FromSeconds(1)); + } + finally + { + content.Release(); + response ??= await sendTask; + response.Dispose(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_ConfiguredTimeoutCancelsStreamedBody(bool useRequestTimeout) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient(new ImmediateResponseHandler(new StreamContent(contentStream))); + var result = await GetService((builder, _) => + builder.ConfigurePipelineOptions(options => options with + { + DefaultHttpTimeout = useRequestTimeout ? null : timeout, + })); + using var response = await result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-body")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = useRequestTimeout ? timeout : null, + }); + var stream = await response.Content.ReadAsStreamAsync(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_HttpClientTimeoutCancelsStreamedBody(bool useCustomClient) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = timeout, + }; + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/client-timeout")) + { + HttpClient = useCustomClient ? httpClient : null, + LoggingType = HttpLoggingType.None, + }); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + public async Task SendAsync_InfiniteTimeoutRetainsCallerCancellationForStreamedBody() + { + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var cancellationTokenSource = new CancellationTokenSource(); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/caller-cancellation")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + }, cancellationTokenSource.Token); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + cancellationTokenSource.Cancel(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_NormalizesAsyncStreamFailureAfterTimeout( + bool throwIOExceptionWhenDisposed) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/async-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_NormalizesAsyncStreamFailureAfterPerReadCancellation( + bool throwIOExceptionWhenDisposed) + { + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/read-cancellation")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = TimeSpan.FromMinutes(1), + }); + var stream = response.Content.ReadAsStream(); + using var cancellationTokenSource = new CancellationTokenSource(); + var readTask = stream + .ReadAsync(new byte[1], cancellationTokenSource.Token) + .AsTask(); + + await contentStream.ReadStarted.WaitAsync(TimeSpan.FromSeconds(1)); + cancellationTokenSource.Cancel(); + + try + { + var exception = await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + await Assert.That(exception!.CancellationToken.IsCancellationRequested).IsTrue(); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task SendAsync_DoesNotTreatCancellationAsSuccessfulEndOfStream( + bool useBufferedCopy) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + ignoreAsyncCancellation: true, + returnEofWhenDisposed: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/cancelled-eof")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + + Task readTask; + if (useBufferedCopy) + { + readTask = response.Content.ReadAsStringAsync(); + } + else + { + var stream = await response.Content.ReadAsStreamAsync(); + readTask = stream.ReadAsync(new byte[1]).AsTask(); + } + + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + public async Task SendAsync_CancelsLegacyResponseLogger() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + new LegacyBodyLogger(logRequest: false), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/legacy-logger")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + public async Task SendAsync_CancelsLegacyRequestLogger() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StringContent("response"))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + new LegacyBodyLogger(logRequest: true), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var request = new HttpRequestMessage( + HttpMethod.Post, + "https://example.test/legacy-request-logger") + { + Content = new StreamContent(contentStream), + }; + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions(request) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Request, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + [Arguments(true, true, false)] + [Arguments(true, false, false)] + [Arguments(false, true, false)] + [Arguments(false, false, false)] + [Arguments(true, false, true)] + [Arguments(false, false, true)] + public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead( + bool useCopyTo, + bool throwIOExceptionWhenDisposed, + bool returnEofWhenDisposed) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + returnEofWhenDisposed: returnEofWhenDisposed); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var result = await GetService((_, _) => { }); + using var response = await result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/synchronous-read")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + var stream = await response.Content.ReadAsStreamAsync(); + var readTask = Task.Run(() => + { + if (useCopyTo) + { + stream.CopyTo(Stream.Null); + } + else + { + stream.ReadExactly(new byte[1]); + } + }); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout() + { + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(new MemoryStream([1, 2, 3])))) + { + Timeout = TimeSpan.FromSeconds(5), + }; + Type? loggedContentType = null; + var httpLogger = new Mock(); + httpLogger + .Setup(x => x.PrintResponse( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (response, _, _, _) => loggedContentType = response.Content.GetType()) + .Returns(Task.CompletedTask); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + httpLogger.Object, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/binary")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + }); + + using (Assert.Multiple()) + { + await Assert.That(loggedContentType).IsEqualTo(typeof(StreamContent)); + await Assert.That(response.Content).IsTypeOf(); + } + } + + [Test] + [Arguments(4096, false)] + [Arguments(0, false)] + [Arguments(0, true)] + public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging( + int maxBodySizeToLog, + bool returnEofWhenDisposed) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + ignoreAsyncCancellation: true, + returnEofWhenDisposed: returnEofWhenDisposed); + var content = new StreamContent(contentStream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + using var httpClient = new HttpClient(new ImmediateResponseHandler(content)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-custom-log-body")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + LogSettings = new HttpLoggingOptions + { + MaxBodySizeToLog = maxBodySizeToLog, + }, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + public async Task SendAsync_ConfiguredTimeoutInterruptsBufferedBodyCopy() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/buffered-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsStringAsync().WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + [Arguments(true, true)] + [Arguments(true, false)] + [Arguments(false, true)] + [Arguments(false, false)] + public async Task SendAsync_PreservesCancellationWhileReadingErrorContent( + bool useCustomClient, + bool useConfiguredTimeout) + { + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler( + new StreamContent(contentStream), + HttpStatusCode.InternalServerError)) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.CancelAfter(TimeSpan.FromMilliseconds(100)); + + try + { + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-error-body")) + { + HttpClient = useCustomClient ? httpClient : null, + LoggingType = HttpLoggingType.None, + ThrowOnNonSuccessStatusCode = true, + Timeout = useConfiguredTimeout ? TimeSpan.FromMilliseconds(100) : null, + }, + useConfiguredTimeout ? CancellationToken.None : cancellationTokenSource.Token) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent() + { + var timeout = TimeSpan.FromMilliseconds(100); + var content = new StreamContent( + new MemoryStream("response body"u8.ToArray())); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + using var httpClient = new HttpClient(new ImmediateResponseHandler(content)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/logged-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }); + + await Task.Delay(timeout + TimeSpan.FromMilliseconds(50)); + + await Assert.ThrowsAsync(async () => + { + var stream = await response.Content.ReadAsStreamAsync(); + await stream.ReadExactlyAsync(new byte[1]); + }); + } + + [Test] + public async Task SendAsync_ConfiguredTimeoutCancelsFactoryResponseLogging() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + var content = new StreamContent(contentStream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var responseLoggingHandler = new ResponseLoggingHttpHandler( + moduleLoggerProvider.Object, + httpLogger) + { + InnerHandler = new ImmediateResponseHandler(content), + }; + using var httpClient = new HttpClient(responseLoggingHandler); + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + try + { + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-log-body")) + { + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + public async Task ResilienceHandler_DisposesRetryableResponseBeforeNextAttempt() + { + var retryContent = new TrackingStringContent("retry"); + var innerHandler = new SequenceResponseHandler( + new HttpResponseMessage(HttpStatusCode.ServiceUnavailable) + { + Content = retryContent, + }, + new HttpResponseMessage(HttpStatusCode.OK)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + using var handler = new ResilienceHttpHandler( + moduleLoggerProvider.Object, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions + { + DefaultHttpResilienceOptions = new HttpResilienceOptions + { + MaxRetryAttempts = 1, + InitialDelay = TimeSpan.Zero, + JitterFactor = 0, + }, + })) + { + InnerHandler = innerHandler, + }; + using var invoker = new HttpMessageInvoker(handler); + using var request = new HttpRequestMessage(HttpMethod.Get, "https://example.test/retry"); + using var response = await invoker.SendAsync(request, CancellationToken.None); + + using (Assert.Multiple()) + { + await Assert.That(response.StatusCode).IsEqualTo(HttpStatusCode.OK); + await Assert.That(innerHandler.CallCount).IsEqualTo(2); + await Assert.That(retryContent.IsDisposed).IsTrue(); + } + } + [Test] public async Task PublicApi_DoesNotExposeRawHttpClients() { @@ -155,4 +820,227 @@ await http.SendAsync(new HttpOptions(new HttpRequestMessage(HttpMethod.Get, serv await Assert.That(indexOfDuration).IsLessThan(indexOfResponse); } } + + private sealed class ImmediateResponseHandler( + HttpContent content, + HttpStatusCode statusCode = HttpStatusCode.OK) : HttpMessageHandler + { + public TaskCompletionSource RequestReceived { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + RequestReceived.TrySetResult(); + return Task.FromResult(new HttpResponseMessage(statusCode) + { + Content = content, + RequestMessage = request, + }); + } + } + + private sealed class BlockingHttpContent : HttpContent + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Release() => _release.TrySetResult(); + + protected override async Task SerializeToStreamAsync( + Stream stream, + TransportContext? context) + { + await _release.Task; + await stream.WriteAsync("response body"u8.ToArray()); + } + + protected override bool TryComputeLength(out long length) + { + length = 0; + return false; + } + } + + private sealed class BlockingReadStream( + bool throwIOExceptionWhenDisposed = false, + bool ignoreAsyncCancellation = false, + bool returnEofWhenDisposed = false) : Stream + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _readStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ManualResetEventSlim _synchronousRelease = new(); + private bool _isDisposed; + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public Task ReadStarted => _readStarted.Task; + + public void Release() + { + _release.TrySetResult(); + _synchronousRelease.Set(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + return WaitForSynchronousRead(); + } + + public override int Read(Span buffer) + { + return WaitForSynchronousRead(); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + _readStarted.TrySetResult(); + if (ignoreAsyncCancellation) + { + await _release.Task; + if (!returnEofWhenDisposed) + { + ThrowIfDisposed(); + } + } + else + { + await _release.Task.WaitAsync(cancellationToken); + } + + return 0; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _isDisposed = true; + Release(); + } + + base.Dispose(disposing); + } + + private int WaitForSynchronousRead() + { + _synchronousRelease.Wait(); + ThrowIfDisposed(); + return 0; + } + + public override long Seek(long offset, SeekOrigin origin) + { + throw new NotSupportedException(); + } + + public override void SetLength(long value) + { + throw new NotSupportedException(); + } + + public override void Write(byte[] buffer, int offset, int count) + { + throw new NotSupportedException(); + } + + private void ThrowIfDisposed() + { + if (_isDisposed) + { + throw throwIOExceptionWhenDisposed + ? new IOException("The stream was interrupted.") + : new ObjectDisposedException(nameof(BlockingReadStream)); + } + } + } + + private sealed class LegacyBodyLogger(bool logRequest) : IHttpLogger + { + public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) + { + return logRequest + ? request.Content!.ReadAsStringAsync() + : Task.CompletedTask; + } + + public Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options) + { + return logRequest + ? request.Content!.ReadAsStringAsync() + : Task.CompletedTask; + } + + public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) + { + return logRequest + ? Task.CompletedTask + : response.Content.ReadAsStringAsync(); + } + + public Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options) + { + return logRequest + ? Task.CompletedTask + : response.Content.ReadAsStringAsync(); + } + + public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger) + { + } + + public void PrintDuration(TimeSpan duration, IModuleLogger logger) + { + } + } + + private sealed class SequenceResponseHandler(params HttpResponseMessage[] responses) : HttpMessageHandler + { + public int CallCount { get; private set; } + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + return Task.FromResult(responses[CallCount++]); + } + } + + private sealed class TrackingStringContent(string content) : StringContent(content) + { + public bool IsDisposed { get; private set; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } } diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index 8cf8da07ac..a2ef3832e4 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -1,13 +1,26 @@ +using System.Net; using ModularPipelines.Context; using ModularPipelines.Context.Domains.Files; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.Exceptions; +using ModularPipelines.FileSystem; +using ModularPipelines.Logging; using ModularPipelines.Options; using ModularPipelines.TestHelpers; +using Moq; namespace ModularPipelines.UnitTests.Helpers; public class DownloaderTests : TestBase { + [Test] + public async Task DownloadOptions_DisableResponseBodyLoggingByDefault() + { + var options = new DownloadOptions(new Uri("https://example.test/file")); + + await Assert.That(options.LoggingType.HasFlag(ModularPipelines.Http.HttpLoggingType.Response)).IsFalse(); + } + [Test] public async Task Can_Download() { @@ -19,4 +32,219 @@ public async Task Can_Download() await Assert.That(checksum.Md5(file)).IsEqualTo("AEDF5D7C23744269F358814E602AFE89"); } + + [Test] + public async Task DownloadStringAsync_DisposesResponse() + { + var content = new TrackingStringContent("download"); + var downloader = CreateDownloader(content); + + var result = await downloader.DownloadStringAsync( + new DownloadOptions(new Uri("https://example.test/download"))); + + using (Assert.Multiple()) + { + await Assert.That(result).IsEqualTo("download"); + await Assert.That(content.IsDisposed).IsTrue(); + } + } + + [Test] + public async Task DownloadFileAsync_DisposesResponse() + { + var content = new TrackingStringContent("download"); + var fileSystemProvider = new Mock(); + fileSystemProvider + .Setup(x => x.Create(It.IsAny())) + .Returns(() => new MemoryStream()); + fileSystemProvider + .Setup(x => x.GetRandomFileName()) + .Returns("random.tmp"); + var downloader = CreateDownloader(content, fileSystemProvider.Object); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = "download.bin", + }); + + await Assert.That(content.IsDisposed).IsTrue(); + } + + [Test] + public async Task DownloadFileAsync_UsesBoundedSiblingTemporaryName() + { + var destination = Path.Combine( + "downloads", + new string('x', 240) + ".bin"); + var expectedTemporaryPath = Path.Combine("downloads", "random.tmp"); + var fileSystemProvider = new Mock(); + fileSystemProvider + .Setup(x => x.GetRandomFileName()) + .Returns("random.tmp"); + fileSystemProvider + .Setup(x => x.Create(expectedTemporaryPath)) + .Returns(() => new MemoryStream()); + var downloader = CreateDownloader( + new StringContent("download"), + fileSystemProvider.Object); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + }); + + fileSystemProvider.Verify(x => x.Create(expectedTemporaryPath), Times.Once()); + fileSystemProvider.Verify( + x => x.MoveFile(expectedTemporaryPath, destination, true), + Times.Once()); + } + + [Test] + public async Task LegacyProviderFallbackDoesNotDeleteExistingDestination() + { + const string destination = "download.bin"; + var fileSystemProvider = new Mock + { + CallBase = true, + }; + fileSystemProvider + .Setup(x => x.FileExists(destination)) + .Returns(true); + + var exception = Assert.Throws(() => + fileSystemProvider.Object.MoveFile("temporary.download", destination, overwrite: true)); + + await Assert.That(exception.Message).Contains("does not support atomic overwrite moves"); + fileSystemProvider.Verify(x => x.DeleteFile(destination), Times.Never()); + fileSystemProvider.Verify( + x => x.MoveFile(It.IsAny(), It.IsAny()), + Times.Never()); + } + + [Test] + public async Task DownloadFileAsync_PreservesExistingFileWhenStreamingFails() + { + var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}"); + var destination = Path.Combine(directory, "download.bin"); + Directory.CreateDirectory(directory); + await System.IO.File.WriteAllTextAsync(destination, "original"); + + try + { + var content = new StreamContent(new FailingReadStream("partial download"u8.ToArray())); + var downloader = CreateDownloader(content); + + await Assert.ThrowsAsync(() => downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + })); + + using (Assert.Multiple()) + { + await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("original"); + await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1); + } + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + + [Test] + public async Task DownloadFileAsync_ReplacesExistingFileAfterStreamingSucceeds() + { + var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}"); + var destination = Path.Combine(directory, "download.bin"); + Directory.CreateDirectory(directory); + await System.IO.File.WriteAllTextAsync(destination, "original"); + + try + { + var downloader = CreateDownloader(new StringContent("replacement")); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + }); + + using (Assert.Multiple()) + { + await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("replacement"); + await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1); + } + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + + [Test] + public async Task DownloadResponseAsync_DisposesResponseWhenStatusValidationFails() + { + var content = new TrackingStringContent("failure"); + var downloader = CreateDownloader(content, statusCode: HttpStatusCode.BadRequest); + + await Assert.ThrowsAsync(() => + downloader.DownloadResponseAsync( + new DownloadOptions(new Uri("https://example.test/download")))); + + await Assert.That(content.IsDisposed).IsTrue(); + } + + private static Downloader CreateDownloader( + HttpContent content, + IFileSystemProvider? fileSystemProvider = null, + HttpStatusCode statusCode = HttpStatusCode.OK) + { + var http = new Mock(); + http.Setup(x => x.SendAsync( + It.IsAny(), + It.IsAny())) + .ReturnsAsync(new HttpResponseMessage(statusCode) + { + Content = content, + }); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + + return new Downloader( + moduleLoggerProvider.Object, + http.Object, + fileSystemProvider ?? SystemFileSystemProvider.Instance); + } + + private sealed class TrackingStringContent(string content) : StringContent(content) + { + public bool IsDisposed { get; private set; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } + + private sealed class FailingReadStream(byte[] content) : MemoryStream(content) + { + private bool _hasRead; + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + if (_hasRead) + { + throw new IOException("Simulated streaming failure"); + } + + _hasRead = true; + return base.ReadAsync(buffer, cancellationToken); + } + } }