diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index b690d288..f4da0ac9 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -80,6 +80,7 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB } opt.Capabilities.Add(P.WorkerCapability.LargePayloads); + opt.ExternalPayloadStoreUri = store.ExternalPayloadStoreBaseUri?.ToString(); }); return builder; diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index e01d2e74..3388d509 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -67,6 +67,15 @@ public BlobPayloadStore(LargePayloadStorageOptions options) this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); } + /// + public override Uri? ExternalPayloadStoreBaseUri => new BlobUriBuilder(this.containerClient.Uri) + { + BlobContainerName = string.Empty, + BlobName = string.Empty, + Sas = null, + Query = string.Empty, + }.ToUri(); + /// public override async Task UploadAsync(string payLoad, CancellationToken cancellationToken) { diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs index b0fe6f80..86f5b028 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/PayloadStore.cs @@ -8,6 +8,12 @@ namespace Microsoft.DurableTask; /// public abstract class PayloadStore { + /// + /// Gets the non-secret base URI (scheme+host[+account path for emulator]) of the external payload store, or + /// null if not applicable. Never contains a secret/SAS/key. + /// + public virtual Uri? ExternalPayloadStoreBaseUri => null; + /// /// Uploads a payload and returns an opaque reference token that can be embedded in orchestration messages. /// diff --git a/src/Grpc/orchestrator_service.proto b/src/Grpc/orchestrator_service.proto index 3d9194ac..18806215 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -832,6 +832,10 @@ message GetWorkItemsRequest { repeated WorkerCapability capabilities = 10; WorkItemFilters workItemFilters = 11; + + // Non-secret base URI of the external payload store (e.g. "https://.blob.core.windows.net/") where this + // worker externalizes large payloads. Empty when no external payload store is configured. Never contains a secret. + string external_payload_store_uri = 12 [json_name = "externalPayloadStoreUri"]; } enum WorkerCapability { diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..fe1a8c61 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -333,6 +333,7 @@ async ValueTask BuildRuntimeStateAsync( workerOptions.Concurrency.MaximumConcurrentEntityWorkItems, Capabilities = { this.worker.grpcOptions.Capabilities }, WorkItemFilters = this.worker.workItemFilters?.ToGrpcWorkItemFilters(), + ExternalPayloadStoreUri = this.worker.grpcOptions.ExternalPayloadStoreUri ?? string.Empty, }, cancellationToken: cancellation); } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index 59c21a00..3e893aac 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -44,6 +44,12 @@ public sealed class GrpcDurableTaskWorkerOptions : DurableTaskWorkerOptions /// public HashSet Capabilities { get; } = new() { P.WorkerCapability.HistoryStreaming }; + /// + /// Gets or sets the non-secret external payload store base URI reported to the backend on the work-item + /// handshake. Empty/null when not using externalized payloads. + /// + public string? ExternalPayloadStoreUri { get; set; } + /// /// Gets or sets the maximum size of all actions in a complete orchestration work item chunk. /// The default value is 3.9MB. We leave some headroom to account for request size overhead. diff --git a/test/Worker/Grpc.Tests/ExternalPayloadStoreUriTests.cs b/test/Worker/Grpc.Tests/ExternalPayloadStoreUriTests.cs new file mode 100644 index 00000000..84b70974 --- /dev/null +++ b/test/Worker/Grpc.Tests/ExternalPayloadStoreUriTests.cs @@ -0,0 +1,111 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Azure.Core; +using Microsoft.Extensions.DependencyInjection; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +/// +/// Tests for the non-secret external payload store base URI that the worker reports on the work-item handshake. +/// +public class ExternalPayloadStoreUriTests +{ + // Well-known Azurite development storage account key. Used here only to prove it is never leaked into the base URI. + const string DevelopmentAccountKey = + "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + + [Fact] + public void ExternalPayloadStoreBaseUri_EmulatorConnectionString_ReturnsAccountBaseUriWithoutContainerOrSas() + { + // Arrange + LargePayloadStorageOptions options = new("UseDevelopmentStorage=true") + { + ContainerName = "my-payloads", + }; + BlobPayloadStore store = new(options); + + // Act + Uri? baseUri = store.ExternalPayloadStoreBaseUri; + + // Assert + baseUri.Should().NotBeNull(); + baseUri!.AbsoluteUri.TrimEnd('/').Should().Be("http://127.0.0.1:10000/devstoreaccount1"); + baseUri.Query.Should().BeEmpty(); + baseUri.AbsoluteUri.Should().NotContain("my-payloads"); + } + + [Fact] + public void ExternalPayloadStoreBaseUri_ConnectionStringWithAccountKey_ReturnsAccountBaseUriAndNeverLeaksKey() + { + // Arrange + LargePayloadStorageOptions options = new( + $"DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey={DevelopmentAccountKey};EndpointSuffix=core.windows.net") + { + ContainerName = "my-payloads", + }; + BlobPayloadStore store = new(options); + + // Act + Uri? baseUri = store.ExternalPayloadStoreBaseUri; + + // Assert + baseUri.Should().Be(new Uri("https://myaccount.blob.core.windows.net/")); + baseUri!.Query.Should().BeEmpty(); + baseUri.AbsoluteUri.Should().NotContain(DevelopmentAccountKey); + baseUri.AbsoluteUri.Should().NotContain("my-payloads"); + } + + [Fact] + public void ExternalPayloadStoreBaseUri_AccountUriAndCredential_ReturnsAccountBaseUriWithoutContainerOrSas() + { + // Arrange + LargePayloadStorageOptions options = new( + new Uri("https://myaccount.blob.core.windows.net"), + new FakeTokenCredential()) + { + ContainerName = "my-payloads", + }; + BlobPayloadStore store = new(options); + + // Act + Uri? baseUri = store.ExternalPayloadStoreBaseUri; + + // Assert + baseUri.Should().Be(new Uri("https://myaccount.blob.core.windows.net/")); + baseUri!.Query.Should().BeEmpty(); + baseUri.AbsoluteUri.Should().NotContain("my-payloads"); + } + + [Fact] + public void UseExternalizedPayloads_SetsExternalPayloadStoreUriFromStore() + { + // Arrange + ServiceCollection services = new(); + DefaultDurableTaskWorkerBuilder builder = new(null, services); + builder.UseGrpc(GrpcChannel.ForAddress("http://localhost:9001")); + builder.UseExternalizedPayloads(opt => + { + opt.ContainerName = "my-payloads"; + opt.ConnectionString = "UseDevelopmentStorage=true"; + }); + + // Act + IServiceProvider provider = services.BuildServiceProvider(); + GrpcDurableTaskWorkerOptions options = provider.GetOptions(); + + // Assert + options.ExternalPayloadStoreUri.Should().NotBeNullOrEmpty(); + options.ExternalPayloadStoreUri!.TrimEnd('/').Should().Be("http://127.0.0.1:10000/devstoreaccount1"); + } + + sealed class FakeTokenCredential : TokenCredential + { + public override AccessToken GetToken(TokenRequestContext requestContext, CancellationToken cancellationToken) + => new("fake-token", DateTimeOffset.MaxValue); + + public override ValueTask GetTokenAsync( + TokenRequestContext requestContext, CancellationToken cancellationToken) + => new(this.GetToken(requestContext, cancellationToken)); + } +}