diff --git a/samples/LargePayloadConsoleApp/Program.cs b/samples/LargePayloadConsoleApp/Program.cs index 97e53112..da293b01 100644 --- a/samples/LargePayloadConsoleApp/Program.cs +++ b/samples/LargePayloadConsoleApp/Program.cs @@ -129,7 +129,8 @@ return string.Empty; } - if (value.StartsWith("blob:v1:", StringComparison.Ordinal)) + if (value.StartsWith("blob:v1:", StringComparison.Ordinal) + || value.StartsWith("blob:v2:", StringComparison.Ordinal)) { throw new InvalidOperationException("Activity received a payload token instead of raw input."); } diff --git a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj index 0a524ed6..78d80fd7 100644 --- a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj +++ b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj @@ -26,4 +26,8 @@ + + + + \ No newline at end of file diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs index e01d2e74..ee78f814 100644 --- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs +++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs @@ -13,11 +13,14 @@ namespace Microsoft.DurableTask; /// /// Azure Blob Storage implementation of . -/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:<container>:<blobName>". +/// Stores payloads as blobs and returns self-describing opaque tokens in the form +/// "blob:v2:<fullBlobUrl>", where the URL is the blob's absolute URI including the storage account. +/// Legacy "blob:v1:<container>:<blobName>" tokens are still recognized for read back-compatibility. /// public sealed class BlobPayloadStore : PayloadStore { - const string TokenPrefix = "blob:v1:"; + const string TokenPrefixV1 = "blob:v1:"; + const string TokenPrefixV2 = "blob:v2:"; const string ContentEncodingGzip = "gzip"; const int MaxRetryAttempts = 8; const int BaseDelayMs = 250; @@ -25,6 +28,7 @@ public sealed class BlobPayloadStore : PayloadStore const int NetworkTimeoutMinutes = 2; readonly BlobContainerClient containerClient; readonly LargePayloadStorageOptions options; + readonly BlobClientOptions clientOptions; /// /// Initializes a new instance of the class. @@ -48,7 +52,7 @@ public BlobPayloadStore(LargePayloadStorageOptions options) nameof(options)); } - BlobClientOptions clientOptions = new() + this.clientOptions = new BlobClientOptions { Retry = { @@ -61,8 +65,8 @@ public BlobPayloadStore(LargePayloadStorageOptions options) }; BlobServiceClient serviceClient = hasIdentityAuth - ? new BlobServiceClient(options.AccountUri, options.Credential, clientOptions) - : new BlobServiceClient(options.ConnectionString, clientOptions); + ? new BlobServiceClient(options.AccountUri, options.Credential, this.clientOptions) + : new BlobServiceClient(options.ConnectionString, this.clientOptions); this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName); } @@ -105,44 +109,48 @@ public override async Task UploadAsync(string payLoad, CancellationToken await blobStream.FlushAsync(cancellationToken); } - return EncodeToken(this.containerClient.Name, blobName); + return EncodeToken(blob.Uri); } /// public override async Task DownloadAsync(string token, CancellationToken cancellationToken) { - (string container, string name) = DecodeToken(token); - if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal)) - { - throw new ArgumentException("Token container does not match configured container.", nameof(token)); - } - - BlobClient blob = this.containerClient.GetBlobClient(name); + (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) = DecodeToken(token); - try + if (!isV2) { - using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken); - Stream contentStream = result.Content; - bool isGzip = string.Equals( - result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase); - - if (isGzip) + // v1 tokens do not carry the account, so the payload is assumed to live in the configured container. + if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal)) { - using GZipStream decompressed = new(contentStream, CompressionMode.Decompress); - using StreamReader reader = new(decompressed, Encoding.UTF8); - return await ReadToEndAsync(reader, cancellationToken); + throw new ArgumentException("Token container does not match configured container.", nameof(token)); } - using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8); - return await ReadToEndAsync(uncompressedReader, cancellationToken); + return await DownloadFromBlobAsync(this.containerClient.GetBlobClient(name), cancellationToken); } - catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound) + + // v2 tokens are self-describing: honor the account and container encoded in the token. + BlobClient blob; + if (this.IsConfiguredContainer(containerUri!)) + { + // Same account and container as the configured store: reuse it (works with any auth mode). + blob = this.containerClient.GetBlobClient(name); + } + else if (this.options.Credential != null) + { + // The payload lives in a different account (e.g. the store was repointed). Identity auth can still + // read it as long as the credential has RBAC access to that account. + blob = new BlobClient(blobUri, this.options.Credential, this.clientOptions); + } + else { throw new PayloadStorageException( - $"The blob '{name}' was not found in container '{container}'. " + - "The payload may have been deleted or the container was never created.", - ex); + $"The externalized payload lives in a different storage account ('{containerUri}') than the " + + $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload reads " + + "require identity (AAD) authentication with access to both accounts; connection-string / " + + "account-key credentials are account-specific and cannot read another account."); } + + return await DownloadFromBlobAsync(blob, cancellationToken); } /// @@ -153,7 +161,60 @@ public override bool IsKnownPayloadToken(string value) return false; } - return value.StartsWith(TokenPrefix, StringComparison.Ordinal); + return value.StartsWith(TokenPrefixV1, StringComparison.Ordinal) + || value.StartsWith(TokenPrefixV2, StringComparison.Ordinal); + } + + /// + /// Encodes a self-describing v2 payload token from the blob's absolute URI. The token carries the full blob + /// URL (including the storage account) so readers can locate the payload without relying on the currently + /// configured store. BlobClient.Uri contains no SAS or account key and is safe to persist. + /// + /// The absolute URI of the blob holding the payload. + /// An opaque payload token in the form "blob:v2:<fullBlobUrl>". + internal static string EncodeToken(Uri blobUri) => $"{TokenPrefixV2}{blobUri}"; + + /// + /// Decodes a payload token. Supports self-describing v2 tokens ("blob:v2:<fullBlobUrl>") and legacy v1 + /// tokens ("blob:v1:<container>:<blobName>"), the latter for read back-compatibility. + /// + /// The payload token to decode. + /// + /// A tuple describing the token: whether it is v2, the container and blob names, and (for v2 only) the + /// absolute blob URI and its container-level URI. + /// + internal static (bool IsV2, string Container, string Name, Uri? BlobUri, Uri? ContainerUri) DecodeToken( + string token) + { + if (token.StartsWith(TokenPrefixV2, StringComparison.Ordinal)) + { + string rest = token.Substring(TokenPrefixV2.Length); + if (!Uri.TryCreate(rest, UriKind.Absolute, out Uri? blobUri)) + { + throw new ArgumentException("Invalid external payload token format.", nameof(token)); + } + + BlobUriBuilder builder = new(blobUri); + string container = builder.BlobContainerName; + string name = builder.BlobName; + builder.BlobName = string.Empty; + Uri containerUri = builder.ToUri(); + return (true, container, name, blobUri, containerUri); + } + + if (token.StartsWith(TokenPrefixV1, StringComparison.Ordinal)) + { + string rest = token.Substring(TokenPrefixV1.Length); + int sep = rest.IndexOf(':'); + if (sep <= 0 || sep >= rest.Length - 1) + { + throw new ArgumentException("Invalid external payload token format.", nameof(token)); + } + + return (false, rest.Substring(0, sep), rest.Substring(sep + 1), null, null); + } + + throw new ArgumentException("Invalid external payload token.", nameof(token)); } static async Task WritePayloadAsync(byte[] payloadBuffer, Stream target, CancellationToken cancellationToken) @@ -177,22 +238,43 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken #endif } - static string EncodeToken(string container, string name) => $"blob:v1:{container}:{name}"; - - static (string Container, string Name) DecodeToken(string token) + static async Task DownloadFromBlobAsync(BlobClient blob, CancellationToken cancellationToken) { - if (!token.StartsWith(TokenPrefix, StringComparison.Ordinal)) + try { - throw new ArgumentException("Invalid external payload token.", nameof(token)); - } + using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken); + Stream contentStream = result.Content; + bool isGzip = string.Equals( + result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase); - string rest = token.Substring(TokenPrefix.Length); - int sep = rest.IndexOf(':'); - if (sep <= 0 || sep >= rest.Length - 1) + if (isGzip) + { + using GZipStream decompressed = new(contentStream, CompressionMode.Decompress); + using StreamReader reader = new(decompressed, Encoding.UTF8); + return await ReadToEndAsync(reader, cancellationToken); + } + + using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8); + return await ReadToEndAsync(uncompressedReader, cancellationToken); + } + catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound) { - throw new ArgumentException("Invalid external payload token format.", nameof(token)); + throw new PayloadStorageException( + $"The blob '{blob.Name}' was not found in container '{blob.BlobContainerName}'. " + + "The payload may have been deleted or the container was never created.", + ex); } + } - return (rest.Substring(0, sep), rest.Substring(sep + 1)); + bool IsConfiguredContainer(Uri tokenContainerUri) + { + Uri configured = this.containerClient.Uri; + return string.Equals(tokenContainerUri.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase) + && string.Equals(tokenContainerUri.Host, configured.Host, StringComparison.OrdinalIgnoreCase) + && tokenContainerUri.Port == configured.Port + && string.Equals( + tokenContainerUri.AbsolutePath.TrimEnd('/'), + configured.AbsolutePath.TrimEnd('/'), + StringComparison.Ordinal); } } diff --git a/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs b/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs new file mode 100644 index 00000000..e5ba58d7 --- /dev/null +++ b/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs @@ -0,0 +1,101 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Grpc.Tests; + +public class BlobPayloadStoreTests +{ + [Fact] + public void EncodeToken_FromBlobUri_ProducesSelfDescribingV2Token() + { + // Arrange + Uri blobUri = new("https://myaccount.blob.core.windows.net/mycontainer/abc123def456"); + + // Act + string token = BlobPayloadStore.EncodeToken(blobUri); + + // Assert + Assert.StartsWith("blob:v2:", token); + Assert.Contains("mycontainer", token); + Assert.Contains("abc123def456", token); + + // Round-trip: decoding the encoded token yields the original container and blob name. + (bool isV2, string container, string name, _, _) = BlobPayloadStore.DecodeToken(token); + Assert.True(isV2); + Assert.Equal("mycontainer", container); + Assert.Equal("abc123def456", name); + } + + [Theory] + [InlineData("https://myaccount.blob.core.windows.net/mycontainer/abc123def456", "mycontainer", "abc123def456")] + [InlineData("http://127.0.0.1:10000/devstoreaccount1/mycontainer/abc123def456", "mycontainer", "abc123def456")] + public void DecodeToken_V2Url_ParsesContainerAndBlob(string blobUrl, string expectedContainer, string expectedBlob) + { + // Arrange + string token = "blob:v2:" + blobUrl; + + // Act + (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) = + BlobPayloadStore.DecodeToken(token); + + // Assert + Assert.True(isV2); + Assert.Equal(expectedContainer, container); + Assert.Equal(expectedBlob, name); + Assert.Equal(new Uri(blobUrl), blobUri); + Assert.NotNull(containerUri); + Assert.EndsWith(expectedContainer, containerUri!.AbsolutePath); + } + + [Fact] + public void DecodeToken_LegacyV1Token_ParsesContainerAndBlob() + { + // Arrange + string token = "blob:v1:mycontainer:abc123def456"; + + // Act + (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) = + BlobPayloadStore.DecodeToken(token); + + // Assert + Assert.False(isV2); + Assert.Equal("mycontainer", container); + Assert.Equal("abc123def456", name); + Assert.Null(blobUri); + Assert.Null(containerUri); + } + + [Fact] + public void IsKnownPayloadToken_RecognizesV1AndV2_RejectsOtherValues() + { + // Arrange + BlobPayloadStore store = CreateStore(); + + // Act & Assert + Assert.True(store.IsKnownPayloadToken("blob:v1:mycontainer:abc123")); + Assert.True(store.IsKnownPayloadToken("blob:v2:https://acct.blob.core.windows.net/mycontainer/abc123")); + Assert.False(store.IsKnownPayloadToken("just a normal payload")); + Assert.False(store.IsKnownPayloadToken(string.Empty)); + } + + [Fact] + public async Task DownloadAsync_V2TokenForDifferentAccountWithoutIdentity_ThrowsBeforeNetworkCall() + { + // Arrange: the configured store uses a connection string (account-key auth, no TokenCredential), while + // the token points at a different storage account. Account keys are account-specific, so cross-account + // reads are impossible and must fail fast (before any network call) with a clear error. + BlobPayloadStore store = CreateStore(); + string token = + "blob:v2:https://otheraccount.blob.core.windows.net/othercontainer/" + Guid.NewGuid().ToString("N"); + + // Act + PayloadStorageException ex = await Assert.ThrowsAsync( + () => store.DownloadAsync(token, CancellationToken.None)); + + // Assert + Assert.Contains("different storage account", ex.Message); + } + + static BlobPayloadStore CreateStore() => + new(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { ContainerName = "test" }); +}