durable task scheduler auth extension support

Microsoft.DurableTask.sln

@@ -71,6 +71,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Analyzers.Tests", "test\Ana
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "AzureFunctionsApp.Tests", "samples\AzureFunctionsUnitTests\AzureFunctionsApp.Tests.csproj", "{FC2692E7-79AE-400E-A50F-8E0BCC8C9BD9}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Extensions", "Extensions", "{5227C712-2355-403F-90D6-51D0BCAE4D38}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Azure", "src\Extensions\Azure\Azure.csproj", "{662BF73D-A4DD-4910-8625-7C12F1ACDBEC}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -185,6 +189,10 @@ Global
 		{FC2692E7-79AE-400E-A50F-8E0BCC8C9BD9}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{FC2692E7-79AE-400E-A50F-8E0BCC8C9BD9}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{FC2692E7-79AE-400E-A50F-8E0BCC8C9BD9}.Release|Any CPU.Build.0 = Release|Any CPU
+		{662BF73D-A4DD-4910-8625-7C12F1ACDBEC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{662BF73D-A4DD-4910-8625-7C12F1ACDBEC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{662BF73D-A4DD-4910-8625-7C12F1ACDBEC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{662BF73D-A4DD-4910-8625-7C12F1ACDBEC}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -220,6 +228,8 @@ Global
 		{998E9D97-BD36-4A9D-81FC-5DAC1CE40083} = {8AFC9781-F6F1-4696-BB4A-9ED7CA9D612B}
 		{541FCCCE-1059-4691-B027-F761CD80DE92} = {E5637F81-2FB9-4CD7-900D-455363B142A7}
 		{FC2692E7-79AE-400E-A50F-8E0BCC8C9BD9} = {EFF7632B-821E-4CFC-B4A0-ED4B24296B17}
+		{5227C712-2355-403F-90D6-51D0BCAE4D38} = {8AFC9781-F6F1-4696-BB4A-9ED7CA9D612B}
+		{662BF73D-A4DD-4910-8625-7C12F1ACDBEC} = {5227C712-2355-403F-90D6-51D0BCAE4D38}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {AB41CB55-35EA-4986-A522-387AB3402E71}

src/Extensions/Azure/Azure.csproj

@@ -0,0 +1,27 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>netstandard2.0;net6.0</TargetFrameworks>
+    <PackageDescription>Azure extensions for the Durable Task Framework.</PackageDescription>
+    <EnableStyleCop>true</EnableStyleCop>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.Identity" Version="1.13.1" />
+    <PackageReference Include="Grpc.Net.Client" Version="2.67.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="../../Client/Core/Client.csproj" />
+    <ProjectReference Include="../../Worker/Core/Worker.csproj" />
+    <ProjectReference Include="../../Worker/Grpc/Worker.Grpc.csproj" />
+    <ProjectReference Include="../../Client/Grpc/Client.Grpc.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <SharedSection Include="Core" />
+    <SharedSection Include="DependencyInjection" />
+    <SharedSection Include="Grpc" />
+  </ItemGroup>
+
+</Project>

src/Extensions/Azure/DurableTaskSchedulerConnectionString.cs

@@ -0,0 +1,82 @@
+// ------------------------------------------------------------
+//  Copyright (c) Microsoft Corporation.All rights reserved.
+// ------------------------------------------------------------
+
+using System.Data.Common;
+
+namespace DurableTask.Extensions.Azure;
+
+/// <summary>
+/// Represents the constituent parts of a connection string for a Durable Task Scheduler service.
+/// </summary>
+public sealed class DurableTaskSchedulerConnectionString
+{
+    readonly DbConnectionStringBuilder builder;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="DurableTaskSchedulerConnectionString"/> class.
+    /// </summary>
+    /// <param name="connectionString">A connection string for a Durable Task Scheduler service.</param>
+    public DurableTaskSchedulerConnectionString(string connectionString)
+    {
+        this.builder = new() { ConnectionString = connectionString };
+    }
+
+    /// <summary>
+    /// Gets the authentication method specified in the connection string (if any).
+    /// </summary>
+    public string Authentication => this.GetRequiredValue("Authentication");
+
+    /// <summary>
+    /// Gets the managed identity or workload identity client ID specified in the connection string (if any).
+    /// </summary>
+    public string? ClientId => this.GetValue("ClientID");
+
+    /// <summary>
+    /// Gets the "AdditionallyAllowedTenants" property, optionally used by Workload Identity.
+    /// Multiple values can be separated by a comma.
+    /// </summary>
+    public IList<string>? AdditionallyAllowedTenants =>
+        string.IsNullOrEmpty(this.AdditionallyAllowedTenantsStr)
+            ? null
+            : this.AdditionallyAllowedTenantsStr!.Split(',');
+
+    /// <summary>
+    /// Gets the "TenantId" property, optionally used by Workload Identity.
+    /// </summary>
+    public string? TenantId => this.GetValue("TenantId");
+
+    /// <summary>
+    /// Gets the "TokenFilePath" property, optionally used by Workload Identity.
+    /// </summary>
+    public string? TokenFilePath => this.GetValue("TokenFilePath");
+
+    /// <summary>
+    /// Gets the endpoint specified in the connection string (if any).
+    /// </summary>
+    public string Endpoint => this.GetRequiredValue("Endpoint");
+
+    /// <summary>
+    /// Gets the task hub name specified in the connection string.
+    /// </summary>
+    public string TaskHubName => this.GetRequiredValue("TaskHub");
+
+    string? AdditionallyAllowedTenantsStr => this.GetValue("AdditionallyAllowedTenants");
+
+    string? GetValue(string name) =>
+        this.builder.TryGetValue(name, out object? value)
+            ? value as string
+            : null;
+
+    string GetRequiredValue(string name)
+    {
+        string? value = this.GetValue(name);
+        if (string.IsNullOrEmpty(value))
+        {
+            throw new ArgumentNullException(
+                $"The connection string is missing the required '{name}' property.");
+        }
+
+        return value!;
+    }
+}

src/Extensions/Azure/DurableTaskSchedulerExtensions.cs

@@ -0,0 +1,156 @@
+using Azure.Core;
+using Microsoft.DurableTask.Client;
+using Microsoft.DurableTask.Worker;
+using System.Diagnostics;
+
+namespace DurableTask.Extensions.Azure;
+
+// NOTE: These extension methods will eventually be provided by the Durable Task SDK itself.
+public static class DurableTaskSchedulerExtensions
+{
+    // Configure the Durable Task *Worker* to use the Durable Task Scheduler service with the specified options.
+    public static void UseDurableTaskScheduler(
+        this IDurableTaskWorkerBuilder builder,
+        string endpointAddress,
+        string taskHubName,
+        TokenCredential credential,
+        Action<DurableTaskSchedulerOptions>? configure = null)
+    {
+        DurableTaskSchedulerOptions options = new(endpointAddress, taskHubName, credential);
+
+        configure?.Invoke(options);
+
+        builder.UseGrpc(GetGrpcChannelForOptions(options));
+    }
+
+    public static void UseDurableTaskScheduler(
+        this IDurableTaskWorkerBuilder builder,
+        string connectionString,
+        Action<DurableTaskSchedulerOptions>? configure = null)
+    {
+        var options = DurableTaskSchedulerOptions.FromConnectionString(connectionString);
+        configure?.Invoke(options);
+        builder.UseGrpc(GetGrpcChannelForOptions(options));
+    }
+
+    // Configure the Durable Task *Client* to use the Durable Task Scheduler service with the specified options.
+    public static void UseDurableTaskScheduler(
+        this IDurableTaskClientBuilder builder,
+        string endpointAddress,
+        string taskHubName,
+        TokenCredential credential,
+        Action<DurableTaskSchedulerOptions>? configure = null)
+    {
+        DurableTaskSchedulerOptions options = new(endpointAddress, taskHubName, credential);
+
+        configure?.Invoke(options);
+
+        builder.UseGrpc(GetGrpcChannelForOptions(options));
+    }
+
+    public static void UseDurableTaskScheduler(
+        this IDurableTaskClientBuilder builder,
+        string connectionString,
+        Action<DurableTaskSchedulerOptions>? configure = null)
+    {
+        var options = DurableTaskSchedulerOptions.FromConnectionString(connectionString);
+        configure?.Invoke(options);
+        builder.UseGrpc(GetGrpcChannelForOptions(options));
+    }
+
+    static GrpcChannel GetGrpcChannelForOptions(DurableTaskSchedulerOptions options)
+    {
+        if (string.IsNullOrEmpty(options.EndpointAddress))
+        {
+            throw RequiredOptionMissing(nameof(options.TaskHubName));
+        }
+
+        if (string.IsNullOrEmpty(options.TaskHubName))
+        {
+            throw RequiredOptionMissing(nameof(options.TaskHubName));
+        }
+
+        TokenCredential credential = options.Credential ?? throw RequiredOptionMissing(nameof(options.Credential));
+
+        string taskHubName = options.TaskHubName;
+        string endpoint = options.EndpointAddress;
+
+        if (!endpoint.Contains("://"))
+        {
+            endpoint = $"https://{endpoint}";
+        }
+
+        string resourceId = options.ResourceId ?? "https://durabletask.io";
+#if NET6_0
+        int processId = Environment.ProcessId;
+#else
+        int processId = Process.GetCurrentProcess().Id;
+#endif
+        string workerId = options.WorkerId ?? $"{Environment.MachineName},{processId},{Guid.NewGuid()}";
+
+        TokenCache? cache =
+            options.Credential is not null
+                ? new(
+                    options.Credential,
+                    new(new[] { $"{options.ResourceId}/.default" }),
+                    TimeSpan.FromMinutes(5))
+                : null;
+
+        CallCredentials managedBackendCreds = CallCredentials.FromInterceptor(
+            async (context, metadata) =>
+            {
+                metadata.Add("taskhub", taskHubName);
+                metadata.Add("workerid", workerId);
+
+                if (cache is null)
+                {
+                    return;
+                }
+
+                AccessToken token = await cache.GetTokenAsync(context.CancellationToken);
+
+                metadata.Add("Authorization", $"Bearer {token.Token}");
+            });
+
+    #if NET6_0
+        return GrpcChannel.ForAddress(
+            endpoint,
+            new GrpcChannelOptions
+            {
+                Credentials = ChannelCredentials.Create(ChannelCredentials.SecureSsl, managedBackendCreds),
+            });
+    #else
+        return new GrpcChannel(
+            endpoint,
+            ChannelCredentials.Create(ChannelCredentials.SecureSsl, managedBackendCreds));
+    #endif
+    }
+
+    static Exception RequiredOptionMissing(string optionName)
+    {
+        return new ArgumentException(message: $"Required option '{optionName}' was not provided.");
+    }
+
+    sealed class TokenCache(TokenCredential credential, TokenRequestContext context, TimeSpan margin)
+    {
+        readonly TokenCredential credential = credential;
+        readonly TokenRequestContext context = context;
+        readonly TimeSpan margin = margin;
+
+        AccessToken? token;
+
+        public async Task<AccessToken> GetTokenAsync(CancellationToken cancellationToken)
+        {
+            DateTimeOffset nowWithMargin = DateTimeOffset.UtcNow.Add(this.margin);
+
+            if (this.token is null
+                || this.token.Value.RefreshOn < nowWithMargin
+                || this.token.Value.ExpiresOn < nowWithMargin)
+            {
+                this.token = await this.credential.GetTokenAsync(this.context, cancellationToken);
+            }
+
+            return this.token.Value;
+        }
+    }
+}