diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs index f405ca70..54228ff3 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs @@ -2176,6 +2176,7 @@ public Task PurgeOrchestrationHistoryAsync(DateTime thresholdDateTimeUtc, Orches /// such as the input or output status fields, and for which the blob URI was stored instead. /// /// The URI of the blob. + /// The URI is not in this task hub's large-message container. public Task DownloadBlobAsync(string blobUri) { return this.messageManager.DownloadAndDecompressAsBytesAsync(new Uri(blobUri)); diff --git a/src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs b/src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs index 1bf9bfad..bc54b1e7 100644 --- a/src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs +++ b/src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs @@ -242,7 +242,7 @@ bool OrchestrationIsRunning(OrchestrationStatus? status) { // first, retrieve the entity scheduler state (= input of the orchestration state), possibly from blob storage. string serializedSchedulerState; - if (MessageManager.TryGetLargeMessageReference(state.Input, out Uri blobUrl)) + if (this.messageManager.TryGetLargeMessageReference(state.Input, out Uri blobUrl)) { serializedSchedulerState = await this.messageManager.DownloadAndDecompressAsBytesAsync(blobUrl); } diff --git a/src/DurableTask.AzureStorage/MessageManager.cs b/src/DurableTask.AzureStorage/MessageManager.cs index e5809f61..723e5e69 100644 --- a/src/DurableTask.AzureStorage/MessageManager.cs +++ b/src/DurableTask.AzureStorage/MessageManager.cs @@ -25,6 +25,7 @@ namespace DurableTask.AzureStorage using System.Threading; using System.Threading.Tasks; using Azure; + using Azure.Storage.Blobs; using Azure.Storage.Blobs.Models; using Azure.Storage.Queues.Models; using DurableTask.AzureStorage.Storage; @@ -123,17 +124,17 @@ public async Task SerializeMessageDataAsync(MessageData messageData, Can } /// - /// If the "message" of an orchestration state is actually a URI retrieves actual message from blob storage. + /// If the message is a reference in this task hub's large-message container, retrieve it from blob storage. /// Otherwise returns the message as is. /// - /// The message to be fetched if it is a url. + /// The message to be fetched if it is a large-message reference. /// A token used for canceling the operation. /// Actual string representation of message. public async Task FetchLargeMessageIfNecessary(string message, CancellationToken cancellationToken = default) { - if (TryGetLargeMessageReference(message, out Uri blobUrl)) + if (this.TryGetLargeMessageReference(message, out _, out Blob blob)) { - return await this.DownloadAndDecompressAsBytesAsync(blobUrl, cancellationToken); + return await DownloadAndDecompressAsBytesAsync(blob, cancellationToken); } else { @@ -141,14 +142,53 @@ public async Task FetchLargeMessageIfNecessary(string message, Cancellat } } - internal static bool TryGetLargeMessageReference(string messagePayload, out Uri blobUrl) + internal bool TryGetLargeMessageReference(string messagePayload, out Uri blobUrl) { - if (Uri.IsWellFormedUriString(messagePayload, UriKind.Absolute)) + return this.TryGetLargeMessageReference(messagePayload, out blobUrl, out _); + } + + private bool TryGetLargeMessageReference(string messagePayload, out Uri blobUrl, out Blob blob) + { + if (Uri.IsWellFormedUriString(messagePayload, UriKind.Absolute) && + Uri.TryCreate(messagePayload, UriKind.Absolute, out Uri candidate)) { - return Uri.TryCreate(messagePayload, UriKind.Absolute, out blobUrl); + Uri containerUri = this.blobContainer.GetBlobContainerUri(); + if (IsInLargeMessageContainer(candidate, containerUri)) + { + Blob candidateBlob = this.blobContainer.GetBlobReference(new BlobUriBuilder(candidate).BlobName); + + // SDK decoding and URI reconstruction must not change the allowed storage scope. + if (IsInLargeMessageContainer(candidateBlob.Uri, containerUri)) + { + blobUrl = candidate; + blob = candidateBlob; + return true; + } + } } blobUrl = null; + blob = null; + return false; + } + + private static bool IsInLargeMessageContainer(Uri candidate, Uri containerUri) + { + string containerPath = containerUri.AbsolutePath.TrimEnd('/') + "/"; + if (candidate.Scheme == containerUri.Scheme && + string.Equals(candidate.Host, containerUri.Host, StringComparison.OrdinalIgnoreCase) && + candidate.Port == containerUri.Port && + candidate.UserInfo.Length == 0 && + candidate.Query.Length == 0 && + candidate.Fragment.Length == 0 && + candidate.AbsolutePath.StartsWith(containerPath, StringComparison.Ordinal) && + candidate.AbsolutePath.Length > containerPath.Length) + { + var blobUri = new BlobUriBuilder(candidate); + var container = new BlobUriBuilder(containerUri); + return blobUri.BlobContainerName == container.BlobContainerName && !string.IsNullOrEmpty(blobUri.BlobName); + } + return false; } @@ -249,7 +289,11 @@ public async Task DownloadAndDecompressAsBytesAsync(string blobName, Can public Task DownloadAndDecompressAsBytesAsync(Uri blobUri, CancellationToken cancellationToken = default) { - Blob blob = this.azureStorageClient.GetBlobReference(blobUri); + if (blobUri == null || !this.TryGetLargeMessageReference(blobUri.OriginalString, out _, out Blob blob)) + { + throw new ArgumentException("Blob must belong to this task hub's large-message container.", nameof(blobUri)); + } + return DownloadAndDecompressAsBytesAsync(blob, cancellationToken); } diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 167a3311..347b9b55 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -489,7 +489,7 @@ async Task ConvertFromAsync(OrchestrationInstanceStatus orch if (this.settings.FetchLargeMessageDataEnabled) { - if (MessageManager.TryGetLargeMessageReference(orchestrationState.Input, out Uri blobUrl)) + if (this.messageManager.TryGetLargeMessageReference(orchestrationState.Input, out Uri blobUrl)) { string json = await this.messageManager.DownloadAndDecompressAsBytesAsync(blobUrl, cancellationToken); diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index 3f3be98c..90acbe99 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -31,6 +31,7 @@ namespace DurableTask.AzureStorage.Tests using System.Collections.Generic; using System.Diagnostics; using System.IO; + using System.IO.Compression; using System.Linq; using System.Net; using System.Reflection; @@ -1295,6 +1296,104 @@ public async Task TerminateOrchestration(bool enableExtendedSessions) } } + [DataTestMethod] + [DataRow("terminate")] + [DataRow("suspend")] + [DataRow("resume")] + public async Task ManagementReason_PrivateBlobUrl_RemainsLiteral(string operation) + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var container = new BlobServiceClient(connectionString) + .GetBlobContainerClient("private-reason-" + Guid.NewGuid().ToString("N")); + await container.CreateAsync(PublicAccessType.None); + + try + { + var blob = container.GetBlobClient("private-backup.json.gz"); + string canary = "synthetic-private-value-" + Guid.NewGuid().ToString("N"); + using (var buffer = new MemoryStream()) + { + using (var gzip = new GZipStream(buffer, CompressionLevel.Optimal, leaveOpen: true)) + { + byte[] payload = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(canary)); + await gzip.WriteAsync(payload, 0, payload.Length); + } + + buffer.Position = 0; + await blob.UploadAsync(buffer); + } + + var anonymousBlob = new BlobClient(blob.Uri); + var denied = await Assert.ThrowsExceptionAsync(async () => + { + using BlobDownloadStreamingResult response = (await anonymousBlob.DownloadStreamingAsync()).Value; + }); + Assert.IsTrue( + denied.Status == 401 || denied.Status == 403 || denied.Status == 404, + $"Unexpected anonymous read failure status: {denied.Status}."); + + using TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: settings => + { + settings.StorageAccountClientProvider = new StorageAccountClientProvider(connectionString); + settings.TaskHubName = "ReasonTest" + Guid.NewGuid().ToString("N"); + }); + await host.StartAsync(); + try + { + var client = await host.StartOrchestrationAsync(typeof(Orchestrations.Counter), 0); + await client.WaitForStartupAsync(StandardTimeout); + + await client.SuspendAsync("literal-control"); + var control = await client.WaitForStatusChange(StandardTimeout, OrchestrationStatus.Suspended); + Assert.AreEqual("literal-control", control.Output); + + if (operation != "resume") + { + await client.ResumeAsync("literal-control"); + await client.WaitForStatusChange(StandardTimeout, OrchestrationStatus.Running); + } + + OrchestrationStatus expectedStatus; + if (operation == "terminate") + { + await client.TerminateAsync(blob.Uri.AbsoluteUri); + expectedStatus = OrchestrationStatus.Terminated; + } + else if (operation == "suspend") + { + await client.SuspendAsync(blob.Uri.AbsoluteUri); + expectedStatus = OrchestrationStatus.Suspended; + } + else + { + await client.ResumeAsync(blob.Uri.AbsoluteUri); + expectedStatus = OrchestrationStatus.Running; + } + + var status = await client.WaitForStatusChange(StandardTimeout, expectedStatus); + Assert.AreEqual(blob.Uri.AbsoluteUri, status.Output, + "Management reasons must not be replaced with private blob contents."); + + if (expectedStatus != OrchestrationStatus.Terminated) + { + await client.TerminateAsync("cleanup"); + await client.WaitForCompletionAsync(StandardTimeout); + } + } + finally + { + await host.StopAsync(); + await host.service.DeleteAsync(); + } + } + finally + { + await container.DeleteAsync(); + } + } + /// /// End-to-end test which validates the Suspend-Resume functionality. /// diff --git a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs index de86ca3d..855f68d6 100644 --- a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs +++ b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs @@ -13,13 +13,18 @@ #nullable enable namespace DurableTask.AzureStorage.Tests { + using Azure.Storage.Blobs; + using Azure.Storage.Blobs.Specialized; using DurableTask.AzureStorage.Storage; using DurableTask.Core; using DurableTask.Core.History; using Microsoft.VisualStudio.TestTools.UnitTesting; + using Moq; + using Moq.Protected; using Newtonsoft.Json; using System; using System.Collections.Generic; + using System.Linq; using System.Text; using System.Threading.Tasks; @@ -27,6 +32,7 @@ namespace DurableTask.AzureStorage.Tests public class MessageManagerTests { const int MaxStorageQueuePayloadSizeInBytes = 45 * 1024; + const string ScopedReference = "https://referencevalidation.blob.core.windows.net/test-largemessages/message.json.gz"; [DataTestMethod] [DataRow("System.Collections.Generic.Dictionary`2[[System.String, System.Private.CoreLib],[System.String, System.Private.CoreLib]]")] @@ -188,6 +194,159 @@ public void GetBlobUrlEscaped(string blob, string blobUrl) Assert.AreEqual(expected, manager.GetBlobUrl(blob)); } + [DataTestMethod] + [DataRow(true, "instance/message-0123456789abcdef0123456789abcdef-ExecutionStarted.json.gz")] + [DataRow(false, "instance/message-0123456789abcdef0123456789abcdef-ExecutionStarted.json.gz")] + [DataRow(true, "instance/history-0000000000000001-ExecutionCompleted-12345678-Result.json.gz")] + [DataRow(false, "instance/history-0000000000000001-ExecutionCompleted-12345678-Result.json.gz")] + [DataRow(true, "instance|with spaces/history-0000000000000001-ExecutionStarted-12345678-Input.json.gz")] + [DataRow(false, "instance|with spaces/history-0000000000000001-ExecutionStarted-12345678-Input.json.gz")] + [DataRow(true, "instance/encoded%2Fname.json.gz")] + [DataRow(false, "instance/encoded%2Fname.json.gz")] + public void TryGetLargeMessageReference_AcceptsTaskHubReferences(bool useEmulator, string blobName) + { + var manager = SetupLargeMessageManager(useEmulator); + string reference = manager.GetBlobUrl(blobName); + + Assert.IsTrue(manager.TryGetLargeMessageReference(reference, out Uri blobUri)); + Assert.AreEqual(reference, blobUri.AbsoluteUri); + } + + public static IEnumerable OutOfScopePayloads => new[] + { + new object?[] { null }, + new object?[] { "" }, + new object?[] { "ordinary termination reason" }, + new object?[] { "https://example.com/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/ownerprivate/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/other-largemessages/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages-other/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/otheraccount/test-largemessages/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10001/devstoreaccount1/test-largemessages/backup.json.gz" }, + new object?[] { "https://127.0.0.1:10000/devstoreaccount1/test-largemessages/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/Test-largemessages/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages/" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages/blob.json.gz?comp=metadata" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages/blob.json.gz#fragment" }, + new object?[] { "http://user@127.0.0.1:10000/devstoreaccount1/test-largemessages/blob.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages/../ownerprivate/backup.json.gz" }, + new object?[] { "http://127.0.0.1:10000/devstoreaccount1/test-largemessages/%2e%2e/ownerprivate/backup.json.gz" }, + }; + + public static IEnumerable OutOfScopeBlobUris => OutOfScopePayloads + .Select(row => row[0] as string) + .Where(payload => payload != null) + .Select(payload => Uri.TryCreate(payload, UriKind.Absolute, out Uri? candidate) && candidate != null + ? new object[] { candidate } + : null) + .OfType(); + + public static IEnumerable ReconstructedTargetsOutsideScope => new[] + { + new object[] { "https://referencevalidation.blob.core.windows.net/other-largemessages/message.json.gz" }, + new object[] { "https://referencevalidation.blob.core.windows.net/test-largemessages-other/message.json.gz" }, + new object[] { "https://otheraccount.blob.core.windows.net/test-largemessages/message.json.gz" }, + new object[] { "https://referencevalidation.blob.core.windows.net/test-largemessages/" }, + new object[] { "https://referencevalidation.blob.core.windows.net/test-largemessages/message.json.gz?comp=metadata" }, + }; + + [DataTestMethod] + [DynamicData(nameof(OutOfScopePayloads), DynamicDataSourceType.Property)] + public void TryGetLargeMessageReference_OutOfScopePayload_IsNotRecognized(string? payload) + { + var manager = SetupLargeMessageManager(useEmulator: true); + + Assert.IsFalse(manager.TryGetLargeMessageReference(payload, out Uri blobUri)); + Assert.IsNull(blobUri); + } + + [DataTestMethod] + [DynamicData(nameof(OutOfScopePayloads), DynamicDataSourceType.Property)] + public async Task FetchLargeMessageIfNecessary_OutOfScopePayload_RemainsLiteral(string? payload) + { + var manager = SetupLargeMessageManager(useEmulator: true); + + Assert.AreEqual(payload, await manager.FetchLargeMessageIfNecessary(payload)); + } + + [DataTestMethod] + [DynamicData(nameof(OutOfScopeBlobUris), DynamicDataSourceType.Property)] + public async Task DownloadAndDecompressAsBytesAsync_OutOfScopeUri_Throws(Uri blobUri) + { + var manager = SetupLargeMessageManager(useEmulator: true); + + await Assert.ThrowsExceptionAsync(() => manager.DownloadAndDecompressAsBytesAsync(blobUri)); + } + + [DataTestMethod] + [DynamicData(nameof(ReconstructedTargetsOutsideScope), DynamicDataSourceType.Property)] + public void TryGetLargeMessageReference_ReconstructedTargetOutsideScope_IsNotRecognized(string targetUri) + { + var manager = SetupLargeMessageManagerWithReconstructedTarget(targetUri); + + Assert.IsFalse(manager.TryGetLargeMessageReference(ScopedReference, out Uri blobUri)); + Assert.IsNull(blobUri); + } + + [DataTestMethod] + [DynamicData(nameof(ReconstructedTargetsOutsideScope), DynamicDataSourceType.Property)] + public async Task FetchLargeMessageIfNecessary_ReconstructedTargetOutsideScope_RemainsLiteral(string targetUri) + { + var manager = SetupLargeMessageManagerWithReconstructedTarget(targetUri); + + Assert.AreEqual(ScopedReference, await manager.FetchLargeMessageIfNecessary(ScopedReference)); + } + + [DataTestMethod] + [DynamicData(nameof(ReconstructedTargetsOutsideScope), DynamicDataSourceType.Property)] + public async Task DownloadAndDecompressAsBytesAsync_ReconstructedTargetOutsideScope_Throws(string targetUri) + { + var manager = SetupLargeMessageManagerWithReconstructedTarget(targetUri); + + await Assert.ThrowsExceptionAsync(() => manager.DownloadAndDecompressAsBytesAsync(new Uri(ScopedReference))); + } + + private static MessageManager SetupLargeMessageManagerWithReconstructedTarget(string targetUri) + { + var blob = new Mock(MockBehavior.Strict); + blob.SetupGet(client => client.Uri).Returns(new Uri(targetUri)); + + var container = new Mock(MockBehavior.Strict); + container.SetupGet(client => client.Uri) + .Returns(new Uri("https://referencevalidation.blob.core.windows.net/test-largemessages")); + container.Protected() + .Setup("GetBlockBlobClientCore", ItExpr.IsAny()) + .Returns(blob.Object); + + var service = new Mock(MockBehavior.Strict); + service.Setup(client => client.GetBlobContainerClient("test-largemessages")).Returns(container.Object); + + var blobProvider = new Mock>(MockBehavior.Strict); + blobProvider.Setup(provider => provider.CreateOptions()).Returns(new BlobClientOptions()); + blobProvider.Setup(provider => provider.CreateClient(It.IsAny())).Returns(service.Object); + + var provider = new StorageAccountClientProvider("UseDevelopmentStorage=true"); + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(blobProvider.Object, provider.Queue, provider.Table), + }; + return new MessageManager(settings, new AzureStorageClient(settings), "test-largemessages"); + } + + private static MessageManager SetupLargeMessageManager(bool useEmulator) + { + string connectionString = useEmulator + ? "UseDevelopmentStorage=true" + : $"DefaultEndpointsProtocol=https;AccountName=referencevalidation;AccountKey={Convert.ToBase64String(new byte[32])};EndpointSuffix=core.windows.net"; + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(connectionString), + }; + + return new MessageManager(settings, new AzureStorageClient(settings), "test-largemessages"); + } + private string GetMessage(string dictionaryType) => "{\"$type\":\"DurableTask.AzureStorage.MessageData\",\"ActivityId\":\"5406d369-4369-4673-afae-6671a2fa1e57\",\"TaskMessage\":{\"$type\":\"DurableTask.Core.TaskMessage\",\"Event\":{\"$type\":\"DurableTask.Core.History.ExecutionStartedEvent\",\"OrchestrationInstance\":{\"$type\":\"DurableTask.Core.OrchestrationInstance\",\"InstanceId\":\"2.2-34a2c9d4-306e-4467-8470-a8018b2e4f11\",\"ExecutionId\":\"aae324dcc8f943e490b37ec5e5bbf9da\"},\"EventType\":0,\"ParentInstance\":null,\"Name\":\"OrchestrationName\",\"Version\":\"2.0\",\"Input\":\"input\",\"Tags\":{\"$type\":\"" + dictionaryType