Support configurable DTS token audiences and government defaults

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `ResourceId` connection-string support for every DTS authentication mode and optional `AuthorityHost` configuration for SDK-created DefaultAzure, WorkloadIdentity, Environment, and InteractiveBrowser credentials. Caller-supplied credentials own their authority; managed identity and developer-tool cloud configuration remain separate.
 - Added a flat catalogue of directly runnable DTS samples with outcome assertions, owned-resource cleanup, and process-level E2E validation selected by sample name using Go's `-run` flag.
 - Added the exported `task.CallActivityOption` type, per-activity tag options, completion-action tag propagation, and distinct activity and sub-orchestration action trace contexts.
 - Added the top-level `durabletaskscheduler` transport package, a dedicated resilient gRPC worker, DTS emulator tests, and an environment-driven sample.
@@ -56,6 +57,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- DTS token audiences now default per options/client/worker instance to `https://durabletask.azure.us` when `REGION_NAME` starts with `usgov` or `usdod` (case-insensitively), otherwise `https://durabletask.io`. Explicit `Options.ResourceID` or connection-string `ResourceId` overrides the default without changing the endpoint or credential authority. Resource audiences normalize surrounding whitespace, trailing slashes, and one existing `/.default` suffix before token requests; values that normalize to empty are rejected. Audiences remain stable across token refreshes and reconnects. Government-region applications requiring the old audience must explicitly set `https://durabletask.io`.
 - Restore DTS-provided activity trace parents on `ActivityContext.Context()` without emitting duplicate SDK durable spans, preserving trace continuity for application instrumentation.
 - Missing-instance orchestration waits now return `api.ErrInstanceNotFound` immediately instead of retrying `NotFound` until the caller deadline.
 - After a worker has started, it reconnects with bounded backoff after `Unauthenticated` or `PermissionDenied` stream and reconnect-handshake responses so refreshed credentials and propagated RBAC can recover without a process restart. The initial `Hello` remains fail-fast.

durabletaskscheduler/README.md

@@ -48,16 +48,95 @@ fields Azure Identity for Go supports for it:
 | `None` | rejected | rejected | rejected | rejected |
 | `TokenCredential` | rejected | rejected | rejected | rejected |
 
-`Environment` is configured entirely by the `AZURE_*` environment variables, and
+`Environment` gets its identity settings from the `AZURE_*` environment variables, and
 `WorkloadIdentity` falls back to them for any field left unset. Fields are
 trimmed before use and blank `AdditionallyAllowedTenants` entries are dropped.
 `None` and `TokenCredential` construct no Azure Identity credential, so
 supplying identity fields for them fails validation instead of being silently
 ignored.
 
-Access tokens are requested for `Options.ResourceID` (default
-`https://durabletask.io`) with trailing slashes removed and `/.default`
-appended. Token acquisition failures are surfaced as retriable `Unavailable`
+Access tokens are requested for `Options.ResourceID`, also available as the
+`ResourceId` connection-string key. This is a **token audience URI**, not an Azure
+Resource Manager resource path. An explicit nonempty value overrides the default.
+An omitted or exactly empty value selects `https://durabletask.azure.us` when
+`REGION_NAME` starts with `usgov` or `usdod`, case-insensitively; otherwise it
+selects `https://durabletask.io`. This intentionally changes the default for
+government-region hosts. Set `ResourceId=https://durabletask.io` explicitly to
+retain the public audience there. Substring matches such as `notusgov` and
+`notusdod`, and other regions such as `chinaeast2`, retain the public default.
+The service endpoint is never used to infer the audience.
+
+`NewOptions`, `NewOptionsWithCredential`, and `NewOptionsFromConnectionString`
+resolve the default when each options instance is created. A hand-built `Options`
+value, or one whose `ResourceID` is reset to `""`, resolves it when `NewClient` or
+`NewWorker` is called. The selected audience is retained across token refreshes,
+management-channel recreation, worker reconnects, and compatibility listeners.
+Later changes to the environment or caller's options do not change a running
+client or worker.
+
+Normalization trims surrounding whitespace and trailing `/` characters, removes
+**one** existing `/.default` suffix (case-insensitively), then trims trailing `/`
+characters again. The token request appends `/.default`; custom URI casing is
+preserved. For example, `https://durabletask.azure.us//.DEFAULT//` requests
+`https://durabletask.azure.us/.default`, and
+`api://CustomAudience/resource/.DEFAULT/` requests
+`api://CustomAudience/resource/.default`. Whitespace-only values, `///`,
+`/.default`, and `/.DEFAULT///` are rejected, including in connection strings.
+The options retain the input's suffix so validation and reconnects cannot
+accidentally strip meaningful URI segments by normalizing an already-normalized
+value again.
+
+Resource audience, credential authority/cloud, and service endpoint are separate.
+Neither `ResourceID` nor `REGION_NAME` changes the endpoint or authority.
+For SDK-created credentials, optional `Options.AuthorityHost` (or connection-string
+`AuthorityHost`) forwards an HTTPS authority URL through Azure Identity's
+`ClientOptions.Cloud.ActiveDirectoryAuthorityHost`. It is supported by
+`DefaultAzure`, `WorkloadIdentity`, `Environment`, and `InteractiveBrowser`.
+Omission preserves Azure Identity's defaults, including `AZURE_AUTHORITY_HOST`.
+An explicit value overrides that environment setting.
+
+For `Authentication=TokenCredential`, configure authority on the supplied
+credential itself; `GetToken` has no per-request authority override. `None`,
+`TokenCredential`, `ManagedIdentity`, `AzureCLI`, and `AzurePowerShell` reject a
+nonempty `AuthorityHost` instead of silently ignoring it. Managed identity uses
+the hosting environment's identity endpoint, not an Entra authority override.
+Developer-tool credentials require the tool's own cloud configuration (including
+when used within `DefaultAzure`); `AuthorityHost` does not configure those tools.
+
+### Azure Government example
+
+Configure the actual scheduler endpoint separately from the token audience and
+credential authority:
+
+```go
+credential, err := azidentity.NewDefaultAzureCredential(&azidentity.DefaultAzureCredentialOptions{
+    ClientOptions: azcore.ClientOptions{Cloud: cloud.AzureGovernment},
+})
+if err != nil {
+    return err
+}
+options := durabletaskscheduler.NewOptionsWithCredential(endpoint, taskHub, credential)
+options.ResourceID = durabletaskscheduler.GovernmentResourceID
+// Pass these options to NewClient and NewWorker. Do not set AuthorityHost:
+// the supplied credential already owns its authority configuration.
+```
+
+The Azure imports are `github.com/Azure/azure-sdk-for-go/sdk/azcore`,
+`github.com/Azure/azure-sdk-for-go/sdk/azcore/cloud`, and
+`github.com/Azure/azure-sdk-for-go/sdk/azidentity`. For SDK-created credentials,
+the equivalent connection string is:
+
+```text
+Endpoint=https://<government-scheduler-host>;TaskHub=<hub>;Authentication=DefaultAzure;ResourceId=https://durabletask.azure.us;AuthorityHost=https://login.microsoftonline.us/
+```
+
+The [authentication sample](../samples/authentication) exercises both
+connection-string and caller-created credentials, including government-cloud
+configuration.
+
+### Token and connection lifecycle
+
+Token acquisition failures are surfaced as retriable `Unavailable`
 gRPC errors. Tokens and immutable authorization metadata are cached until the
 credential's `RefreshOn` time (or five minutes before expiry), and concurrent
 refreshes are coalesced. Every RPC carries `taskhub` and `x-user-agent` metadata; worker

durabletaskscheduler/authentication_test.go

@@ -345,7 +345,7 @@ func TestTokenScope(t *testing.T) {
 func TestOptionsValidateRejectsBlankOrInjectedResourceID(t *testing.T) {
 	options := NewOptions("scheduler.example.com", "hub")
 	options.ResourceID = "   "
-	require.ErrorContains(t, options.Validate(), "resource ID cannot be blank")
+	require.ErrorContains(t, options.Validate(), "resource ID cannot be empty after normalization")
 
 	options.ResourceID = "https://durabletask.io\r\nx"
 	require.ErrorContains(t, options.Validate(), "resource ID cannot contain newlines")
@@ -658,6 +658,7 @@ func TestConnectRejectsCredentialsOnPlaintextTransport(t *testing.T) {
 }
 
 func TestNewOptionsWithCredentialUsesTokenCredentialMode(t *testing.T) {
+	t.Setenv("REGION_NAME", "")
 	credential := stubCredential{name: "explicit"}
 	options := NewOptionsWithCredential("scheduler.example.com", "hub", credential)
 	require.Equal(t, AuthenticationTokenCredential, options.Authentication)

durabletaskscheduler/authority_host_test.go

@@ -0,0 +1,113 @@
+package durabletaskscheduler
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/cloud"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAuthorityHostConfigurationPerAuthenticationMode(t *testing.T) {
+	for _, authentication := range allAuthenticationTypes {
+		t.Run(string(authentication), func(t *testing.T) {
+			options := NewOptions("scheduler.example.com", "hub")
+			options.Authentication = authentication
+			options.AuthorityHost = "  " + cloud.AzureGovernment.ActiveDirectoryAuthorityHost + "  "
+			if authentication == AuthenticationTokenCredential {
+				options.Credential = stubCredential{}
+			}
+			switch authentication {
+			case AuthenticationNone, AuthenticationTokenCredential, AuthenticationManagedIdentity,
+				AuthenticationAzureCLI, AuthenticationAzurePowerShell:
+				require.ErrorContains(t, options.Validate(), "does not use AuthorityHost")
+				return
+			}
+			require.NoError(t, options.Validate())
+			spec, err := newCredentialSpec(options)
+			require.NoError(t, err)
+			require.Equal(t, cloud.AzureGovernment.ActiveDirectoryAuthorityHost, spec.authorityHost)
+			prepared, err := prepareOptionsWith(options, func(observed credentialSpec) (azcore.TokenCredential, error) {
+				require.Equal(t, spec, observed)
+				return stubCredential{}, nil
+			})
+			require.NoError(t, err)
+			require.Equal(t, options.EndpointAddress, prepared.EndpointAddress)
+			require.Equal(t, options.ResourceID, prepared.ResourceID)
+			require.Equal(t, options.AuthorityHost, prepared.AuthorityHost)
+		})
+	}
+}
+
+func TestAuthorityHostConnectionString(t *testing.T) {
+	for _, key := range []string{"AuthorityHost", "authorityhost", "AUTHORITYHOST", "AuThOrItYhOsT"} {
+		options, err := NewOptionsFromConnectionString(
+			"Endpoint=scheduler.example.com;TaskHub=hub;Authentication=DefaultAzure;ResourceId=api://Custom;" +
+				"AuthorityHost=https://previous.example;" + key + "= https://login.microsoftonline.us/ ",
+		)
+		require.NoError(t, err)
+		require.Equal(t, "https://login.microsoftonline.us/", options.AuthorityHost)
+		require.Equal(t, "api://Custom", options.ResourceID)
+		require.Equal(t, "scheduler.example.com", options.EndpointAddress)
+	}
+	for _, authority := range []string{" ", "http://login.example", "login.example", "https://", "https://user@login.example", "https://login.example/tenant", "https://login.example?query=1", "https://login.example#fragment"} {
+		options := NewOptions("scheduler.example.com", "hub")
+		options.AuthorityHost = authority
+		require.ErrorContains(t, options.Validate(), "AuthorityHost must be an HTTPS authority URL")
+	}
+}
+
+// A deliberately invalid environment authority makes dropped options observable
+// in the real Azure Identity constructors without network I/O or reflection.
+func TestAzureIdentityAuthorityForwardingAndEnvironmentDefaults(t *testing.T) {
+	for _, authentication := range []AuthenticationType{
+		AuthenticationDefaultAzure, AuthenticationWorkloadIdentity,
+		AuthenticationEnvironment, AuthenticationInteractiveBrowser,
+	} {
+		t.Run(string(authentication), func(t *testing.T) {
+			clearAzureIdentityEnvironment(t)
+			t.Setenv("AZURE_TENANT_ID", "00000000-0000-0000-0000-000000000002")
+			t.Setenv("AZURE_CLIENT_ID", "00000000-0000-0000-0000-000000000001")
+			t.Setenv("AZURE_CLIENT_SECRET", "not-a-real-secret")
+			t.Setenv("AZURE_FEDERATED_TOKEN_FILE", "unused-token-file")
+			t.Setenv("AZURE_TOKEN_CREDENTIALS", "EnvironmentCredential")
+			t.Setenv("AZURE_AUTHORITY_HOST", "http://invalid-environment-authority.example")
+			t.Setenv("REGION_NAME", "usgovvirginia")
+			options := NewOptions("scheduler.example.com", "hub")
+			options.Authentication = authentication
+
+			prepared, err := prepareOptions(options)
+			canceled, cancel := context.WithCancel(context.Background())
+			cancel()
+			if authentication == AuthenticationDefaultAzure {
+				// DefaultAzure defers inner constructor errors until GetToken.
+				require.NoError(t, err)
+				_, err = prepared.Credential.GetToken(canceled, policy.TokenRequestOptions{Scopes: []string{GovernmentResourceID + "/.default"}})
+			}
+			require.ErrorContains(t, err, "cannot use an authority host without https",
+				"omitted AuthorityHost must preserve AZURE_AUTHORITY_HOST")
+
+			options.AuthorityHost = cloud.AzureGovernment.ActiveDirectoryAuthorityHost
+			prepared, err = prepareOptions(options)
+			require.NoError(t, err, "explicit AuthorityHost must override the invalid environment authority")
+			require.NotNil(t, prepared.Credential)
+			require.Equal(t, GovernmentResourceID, prepared.ResourceID)
+			require.Equal(t, options.EndpointAddress, prepared.EndpointAddress)
+			if authentication == AuthenticationDefaultAzure {
+				_, err = prepared.Credential.GetToken(canceled, policy.TokenRequestOptions{Scopes: []string{GovernmentResourceID + "/.default"}})
+				require.Error(t, err)
+				require.NotContains(t, err.Error(), "cannot use an authority host without https",
+					"DefaultAzure must forward the explicit authority to its inner credential")
+			}
+
+			t.Setenv("AZURE_AUTHORITY_HOST", "")
+			options.AuthorityHost = ""
+			prepared, err = prepareOptions(options)
+			require.NoError(t, err, "omission must preserve Azure Identity's public default")
+			require.NotNil(t, prepared.Credential)
+			require.Empty(t, prepared.AuthorityHost, "government audience must not choose an authority")
+		})
+	}
+}

durabletaskscheduler/connection.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/cloud"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/google/uuid"
@@ -262,6 +263,7 @@ func (c *schedulerPerRPCCredentials) RequireTransportSecurity() bool {
 // left empty so credential construction is deterministic and reviewable.
 type credentialSpec struct {
 	authentication             AuthenticationType
+	authorityHost              string
 	clientID                   string
 	tenantID                   string
 	tokenFilePath              string
@@ -298,7 +300,7 @@ func newCredentialSpec(options *Options) (credentialSpec, error) {
 		spec.tokenFilePath = tokenFilePath
 		spec.additionallyAllowedTenants = additionalTenants
 	case AuthenticationEnvironment:
-		// EnvironmentCredential is configured entirely by environment variables.
+		// Identity fields come from the environment; AuthorityHost is set below.
 	case AuthenticationAzureCLI, AuthenticationAzurePowerShell:
 		spec.tenantID = tenantID
 		spec.additionallyAllowedTenants = additionalTenants
@@ -309,6 +311,10 @@ func newCredentialSpec(options *Options) (credentialSpec, error) {
 	default:
 		return credentialSpec{}, fmt.Errorf("unsupported DTS authentication type %q", options.Authentication)
 	}
+	switch options.Authentication {
+	case AuthenticationDefaultAzure, AuthenticationWorkloadIdentity, AuthenticationEnvironment, AuthenticationInteractiveBrowser:
+		spec.authorityHost = strings.TrimSpace(options.AuthorityHost)
+	}
 	return spec, nil
 }
 
@@ -325,9 +331,14 @@ func normalizeAdditionallyAllowedTenants(tenants []string) []string {
 func newAzureIdentityCredential(spec credentialSpec) (azcore.TokenCredential, error) {
 	var credential azcore.TokenCredential
 	var err error
+	clientOptions := azcore.ClientOptions{}
+	if spec.authorityHost != "" {
+		clientOptions.Cloud = cloud.Configuration{ActiveDirectoryAuthorityHost: spec.authorityHost}
+	}
 	switch spec.authentication {
 	case AuthenticationDefaultAzure:
 		credential, err = azidentity.NewDefaultAzureCredential(&azidentity.DefaultAzureCredentialOptions{
+			ClientOptions:              clientOptions,
 			TenantID:                   spec.tenantID,
 			AdditionallyAllowedTenants: spec.additionallyAllowedTenants,
 		})
@@ -339,13 +350,16 @@ func newAzureIdentityCredential(spec credentialSpec) (azcore.TokenCredential, er
 		credential, err = azidentity.NewManagedIdentityCredential(credentialOptions)
 	case AuthenticationWorkloadIdentity:
 		credential, err = azidentity.NewWorkloadIdentityCredential(&azidentity.WorkloadIdentityCredentialOptions{
+			ClientOptions:              clientOptions,
 			ClientID:                   spec.clientID,
 			TenantID:                   spec.tenantID,
 			TokenFilePath:              spec.tokenFilePath,
 			AdditionallyAllowedTenants: spec.additionallyAllowedTenants,
 		})
 	case AuthenticationEnvironment:
-		credential, err = azidentity.NewEnvironmentCredential(nil)
+		credential, err = azidentity.NewEnvironmentCredential(&azidentity.EnvironmentCredentialOptions{
+			ClientOptions: clientOptions,
+		})
 	case AuthenticationAzureCLI:
 		credential, err = azidentity.NewAzureCLICredential(&azidentity.AzureCLICredentialOptions{
 			TenantID:                   spec.tenantID,
@@ -358,6 +372,7 @@ func newAzureIdentityCredential(spec credentialSpec) (azcore.TokenCredential, er
 		})
 	case AuthenticationInteractiveBrowser:
 		credential, err = azidentity.NewInteractiveBrowserCredential(&azidentity.InteractiveBrowserCredentialOptions{
+			ClientOptions:              clientOptions,
 			ClientID:                   spec.clientID,
 			TenantID:                   spec.tenantID,
 			AdditionallyAllowedTenants: spec.additionallyAllowedTenants,
@@ -385,9 +400,15 @@ func resolveCredential(options *Options, factory credentialFactory) (azcore.Toke
 	return factory(spec)
 }
 
-// tokenScope builds the OAuth scope requested for a DTS resource ID.
+// tokenScope normalizes the raw resource ID without modifying Options, so
+// validation and channel recreation cannot strip successive /.default segments.
 func tokenScope(resourceID string) string {
-	return strings.TrimRight(strings.TrimSpace(resourceID), "/") + "/.default"
+	resourceID = strings.TrimRight(strings.TrimSpace(resourceID), "/")
+	const suffix = "/.default"
+	if len(resourceID) >= len(suffix) && strings.EqualFold(resourceID[len(resourceID)-len(suffix):], suffix) {
+		resourceID = strings.TrimRight(resourceID[:len(resourceID)-len(suffix)], "/")
+	}
+	return resourceID + suffix
 }
 
 func prepareOptions(options *Options) (Options, error) {
@@ -412,13 +433,10 @@ func prepareOptionsWith(options *Options, factory credentialFactory) (Options, e
 	if prepared.Authentication == "" {
 		prepared.Authentication = AuthenticationDefaultAzure
 	}
-	// Only an exactly empty resource ID defaults. A whitespace-only value is
-	// left intact so Validate below rejects it as documented instead of
-	// silently collapsing to the default resource.
 	if trimmed := strings.TrimSpace(prepared.ResourceID); trimmed != "" {
 		prepared.ResourceID = trimmed
 	} else if prepared.ResourceID == "" {
-		prepared.ResourceID = DefaultResourceID
+		prepared.ResourceID = defaultResourceID()
 	}
 	if prepared.HelloTimeout == 0 {
 		prepared.HelloTimeout = 30 * time.Second