Make externalized large-payload token self-describing (token v2)

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.");
             }

src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj

@@ -26,4 +26,8 @@
   <ItemGroup>
     <SharedSection Include="Core" />
   </ItemGroup>
+
+  <ItemGroup>
+    <InternalsVisibleTo Include="Microsoft.DurableTask.Grpc.IntegrationTests" Key="$(StrongNamePublicKey)" />
+  </ItemGroup>
 </Project>

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -13,18 +13,22 @@ namespace Microsoft.DurableTask;
 
 /// <summary>
 /// Azure Blob Storage implementation of <see cref="PayloadStore"/>.
-/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:&lt;container&gt;:&lt;blobName&gt;".
+/// Stores payloads as blobs and returns self-describing opaque tokens in the form
+/// <c>blob:v2:{fullBlobUrl}</c>, where the URL is the blob's absolute URI including the storage account.
+/// Legacy <c>blob:v1:{container}:{blobName}</c> tokens are still recognized for read back-compatibility.
 /// </summary>
 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;
     const int MaxDelayMs = 10_000;
     const int NetworkTimeoutMinutes = 2;
     readonly BlobContainerClient containerClient;
     readonly LargePayloadStorageOptions options;
+    readonly BlobClientOptions clientOptions;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="BlobPayloadStore"/> 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<string> UploadAsync(string payLoad, CancellationToken
             await blobStream.FlushAsync(cancellationToken);
         }
 
-        return EncodeToken(this.containerClient.Name, blobName);
+        return EncodeToken(blob.Uri);
     }
 
     /// <inheritdoc/>
     public override async Task<string> 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);
+        DecodeTokenResult decoded = DecodeToken(token);
 
-        try
+        if (!decoded.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(decoded.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(decoded.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(decoded.ContainerUri!))
+        {
+            // Same account and container as the configured store: reuse it (works with any auth mode).
+            blob = this.containerClient.GetBlobClient(decoded.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(decoded.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 ('{decoded.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);
     }
 
     /// <inheritdoc/>
@@ -153,7 +161,59 @@ 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);
+    }
+
+    /// <summary>
+    /// 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. <c>BlobClient.Uri</c> contains no SAS or account key and is safe to persist.
+    /// </summary>
+    /// <param name="blobUri">The absolute URI of the blob holding the payload.</param>
+    /// <returns>An opaque payload token in the form <c>blob:v2:{fullBlobUrl}</c>.</returns>
+    internal static string EncodeToken(Uri blobUri) => $"{TokenPrefixV2}{blobUri}";
+
+    /// <summary>
+    /// Decodes a payload token. Supports self-describing v2 tokens (<c>blob:v2:{fullBlobUrl}</c>) and legacy v1
+    /// tokens (<c>blob:v1:{container}:{blobName}</c>), the latter for read back-compatibility.
+    /// </summary>
+    /// <param name="token">The payload token to decode.</param>
+    /// <returns>
+    /// A <see cref="DecodeTokenResult"/> 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.
+    /// </returns>
+    internal static DecodeTokenResult 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 new(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 new(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 +237,59 @@ static async Task<string> 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<string> 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);
     }
+
+    /// <summary>
+    /// The result of decoding an externalized payload token.
+    /// </summary>
+    /// <param name="IsV2">Whether the token is a self-describing v2 token.</param>
+    /// <param name="Container">The name of the container holding the payload.</param>
+    /// <param name="Name">The name of the blob holding the payload.</param>
+    /// <param name="BlobUri">
+    /// The absolute URI of the blob holding the payload, or <see langword="null"/> for v1 tokens, which do not
+    /// carry the storage account.
+    /// </param>
+    /// <param name="ContainerUri">
+    /// The container-level URI of <paramref name="BlobUri"/>, or <see langword="null"/> for v1 tokens.
+    /// </param>
+    internal readonly record struct DecodeTokenResult(
+        bool IsV2, string Container, string Name, Uri? BlobUri, Uri? ContainerUri);
 }

test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs

@@ -0,0 +1,99 @@
+// 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.
+        BlobPayloadStore.DecodeTokenResult decoded = BlobPayloadStore.DecodeToken(token);
+        Assert.True(decoded.IsV2);
+        Assert.Equal("mycontainer", decoded.Container);
+        Assert.Equal("abc123def456", decoded.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
+        BlobPayloadStore.DecodeTokenResult decoded = BlobPayloadStore.DecodeToken(token);
+
+        // Assert
+        Assert.True(decoded.IsV2);
+        Assert.Equal(expectedContainer, decoded.Container);
+        Assert.Equal(expectedBlob, decoded.Name);
+        Assert.Equal(new Uri(blobUrl), decoded.BlobUri);
+        Assert.NotNull(decoded.ContainerUri);
+        Assert.EndsWith(expectedContainer, decoded.ContainerUri!.AbsolutePath);
+    }
+
+    [Fact]
+    public void DecodeToken_LegacyV1Token_ParsesContainerAndBlob()
+    {
+        // Arrange
+        string token = "blob:v1:mycontainer:abc123def456";
+
+        // Act
+        BlobPayloadStore.DecodeTokenResult decoded = BlobPayloadStore.DecodeToken(token);
+
+        // Assert
+        Assert.False(decoded.IsV2);
+        Assert.Equal("mycontainer", decoded.Container);
+        Assert.Equal("abc123def456", decoded.Name);
+        Assert.Null(decoded.BlobUri);
+        Assert.Null(decoded.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<PayloadStorageException>(
+            () => store.DownloadAsync(token, CancellationToken.None));
+
+        // Assert
+        Assert.Contains("different storage account", ex.Message);
+    }
+
+    static BlobPayloadStore CreateStore() =>
+        new(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { ContainerName = "test" });
+}