From 9f7a809fc75f0d41fa6668614f73790f788a69f0 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Sat, 26 Sep 2026 00:06:55 -0700 Subject: [PATCH] Support configurable DTS token audiences and government defaults Add ResourceId connection strings, per-instance government defaults, scope normalization, and separate Azure Identity authority configuration with recording-credential and TLS reconnect coverage. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 + durabletaskscheduler/README.md | 87 +++- durabletaskscheduler/authentication_test.go | 3 +- durabletaskscheduler/authority_host_test.go | 113 +++++ durabletaskscheduler/connection.go | 34 +- durabletaskscheduler/connection_test.go | 8 +- durabletaskscheduler/options.go | 77 +++- durabletaskscheduler/options_test.go | 10 +- durabletaskscheduler/resource_id_test.go | 442 ++++++++++++++++++++ samples/authentication/README.md | 35 ++ samples/authentication/authentication.go | 5 + 11 files changed, 783 insertions(+), 33 deletions(-) create mode 100644 durabletaskscheduler/authority_host_test.go create mode 100644 durabletaskscheduler/resource_id_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 31bf4581..cac94858 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/durabletaskscheduler/README.md b/durabletaskscheduler/README.md index 15002a3d..36716fd3 100644 --- a/durabletaskscheduler/README.md +++ b/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://;TaskHub=;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 diff --git a/durabletaskscheduler/authentication_test.go b/durabletaskscheduler/authentication_test.go index b562d5b7..47885260 100644 --- a/durabletaskscheduler/authentication_test.go +++ b/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) diff --git a/durabletaskscheduler/authority_host_test.go b/durabletaskscheduler/authority_host_test.go new file mode 100644 index 00000000..acb1bec5 --- /dev/null +++ b/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") + }) + } +} diff --git a/durabletaskscheduler/connection.go b/durabletaskscheduler/connection.go index bf8fbb51..96039be6 100644 --- a/durabletaskscheduler/connection.go +++ b/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 diff --git a/durabletaskscheduler/connection_test.go b/durabletaskscheduler/connection_test.go index 2bb74e0e..56d58e75 100644 --- a/durabletaskscheduler/connection_test.go +++ b/durabletaskscheduler/connection_test.go @@ -30,8 +30,9 @@ import ( ) type recordingCredential struct { - mu sync.Mutex - options []policy.TokenRequestOptions + mu sync.Mutex + options []policy.TokenRequestOptions + refreshOn time.Time } type failingCredential struct{} @@ -118,7 +119,7 @@ func (c *recordingCredential) GetToken( c.mu.Lock() c.options = append(c.options, options) c.mu.Unlock() - return azcore.AccessToken{Token: "token", ExpiresOn: time.Now().Add(time.Hour)}, nil + return azcore.AccessToken{Token: "token", ExpiresOn: time.Now().Add(time.Hour), RefreshOn: c.refreshOn}, nil } type metadataServer struct { @@ -826,6 +827,7 @@ func insecureBufconnOptions(t *testing.T, listener *bufconn.Listener) *Options { // TestNewPerRPCCredentialsPreservesRoleMetadata pins the metadata each role // sends, including the user-agent override and worker identity. func TestNewPerRPCCredentialsPreservesRoleMetadata(t *testing.T) { + t.Setenv("REGION_NAME", "") tests := []struct { name string role connectionRole diff --git a/durabletaskscheduler/options.go b/durabletaskscheduler/options.go index 79aa388c..9ccc9ff5 100644 --- a/durabletaskscheduler/options.go +++ b/durabletaskscheduler/options.go @@ -7,6 +7,7 @@ import ( "fmt" "net" "net/url" + "os" "strings" "time" @@ -17,9 +18,12 @@ import ( ) const ( - // DefaultResourceID is the Azure resource used to request DTS access tokens. + // DefaultResourceID is the public-cloud token audience for DTS. DefaultResourceID = "https://durabletask.io" + // GovernmentResourceID is the token audience for DTS in US Government and DoD regions. + GovernmentResourceID = "https://durabletask.azure.us" + // DefaultMaxReceiveMessageSize and DefaultMaxSendMessageSize are the default // per-message gRPC bounds. DefaultMaxReceiveMessageSize = 64 * 1024 * 1024 @@ -55,12 +59,24 @@ type Options struct { // Credential is required by, and valid only for, AuthenticationTokenCredential. Credential azcore.TokenCredential - // ResourceID is the DTS resource the access token is requested for. The - // requested scope is the resource ID without surrounding whitespace or - // trailing slashes plus "/.default". Exactly empty uses DefaultResourceID; - // a whitespace-only value is rejected. + // ResourceID is a token audience URI, not an Azure Resource Manager path. + // Surrounding whitespace, trailing slashes, and one existing "/.default" + // suffix (case-insensitive) are removed before appending "/.default". + // Exactly empty selects GovernmentResourceID when REGION_NAME starts with + // usgov or usdod (case-insensitive), otherwise DefaultResourceID. Constructors + // resolve this default per options instance; hand-built or reset-empty + // options resolve it when creating a client or worker. Values that normalize + // to empty are rejected. This does not configure the endpoint or authority. ResourceID string + // AuthorityHost optionally configures the Entra authority for SDK-created + // DefaultAzure, WorkloadIdentity, Environment, and InteractiveBrowser + // credentials. Empty preserves Azure Identity defaults, including + // AZURE_AUTHORITY_HOST. Other modes reject a nonempty value: configure an + // explicit Credential or developer tool directly; managed identity uses + // the hosting environment's identity endpoint. + AuthorityHost string + // ClientID is used by ManagedIdentity, WorkloadIdentity, and // InteractiveBrowser. Other modes ignore it, except None and TokenCredential, // which reject it. @@ -141,7 +157,7 @@ func NewOptions(endpointAddress, taskHubName string) *Options { EndpointAddress: endpointAddress, TaskHubName: taskHubName, Authentication: AuthenticationDefaultAzure, - ResourceID: DefaultResourceID, + ResourceID: defaultResourceID(), HelloTimeout: 30 * time.Second, MaxReceiveMessageSize: DefaultMaxReceiveMessageSize, MaxSendMessageSize: DefaultMaxSendMessageSize, @@ -153,6 +169,14 @@ func NewOptions(endpointAddress, taskHubName string) *Options { } } +func defaultResourceID() string { + region := strings.ToLower(os.Getenv("REGION_NAME")) + if strings.HasPrefix(region, "usgov") || strings.HasPrefix(region, "usdod") { + return GovernmentResourceID + } + return DefaultResourceID +} + func NewOptionsWithCredential(endpointAddress, taskHubName string, credential azcore.TokenCredential) *Options { options := NewOptions(endpointAddress, taskHubName) options.Authentication = AuthenticationTokenCredential @@ -166,13 +190,13 @@ func NewOptionsWithCredential(endpointAddress, taskHubName string, credential az // // Keys and Authentication values are case-insensitive, surrounding whitespace is // trimmed, empty segments are skipped, and a repeated key uses its last value. -// Supported optional keys are ClientID, TenantID, TokenFilePath, and the -// comma-separated AdditionallyAllowedTenants. +// Supported optional keys are ResourceId, AuthorityHost, ClientID, TenantID, +// TokenFilePath, and the comma-separated AdditionallyAllowedTenants. ResourceId +// follows Options.ResourceID, including rejection of whitespace-only values. func NewOptionsFromConnectionString(connectionString string) (*Options, error) { values := make(map[string]string) for _, segment := range strings.Split(connectionString, ";") { - segment = strings.TrimSpace(segment) - if segment == "" { + if strings.TrimSpace(segment) == "" { continue } keyValue := strings.SplitN(segment, "=", 2) @@ -182,7 +206,11 @@ func NewOptionsFromConnectionString(connectionString string) (*Options, error) { key := strings.ToLower(strings.TrimSpace(keyValue[0])) value := strings.TrimSpace(keyValue[1]) switch key { - case "endpoint", "taskhub", "authentication", "clientid", "tenantid", "tokenfilepath", "additionallyallowedtenants": + case "resourceid": + // Preserve the raw value so whitespace-only input is not mistaken + // for omission and a meaningful /.default segment is not stripped twice. + values[key] = keyValue[1] + case "endpoint", "taskhub", "authentication", "authorityhost", "clientid", "tenantid", "tokenfilepath", "additionallyallowedtenants": values[key] = value default: return nil, fmt.Errorf("unsupported connection string key %q", keyValue[0]) @@ -208,6 +236,10 @@ func NewOptionsFromConnectionString(connectionString string) (*Options, error) { return nil, err } options.Authentication = parsedAuthentication + if resourceID := values["resourceid"]; resourceID != "" { + options.ResourceID = resourceID + } + options.AuthorityHost = values["authorityhost"] options.ClientID = values["clientid"] options.TenantID = values["tenantid"] options.TokenFilePath = values["tokenfilepath"] @@ -306,11 +338,11 @@ func (o *Options) Validate() error { if o.ChannelRecreateMinInterval < 0 { return fmt.Errorf("DTS channel recreate minimum interval cannot be negative") } - if strings.ContainsAny(o.ResourceID, "\r\n") { + if strings.ContainsAny(strings.TrimSpace(o.ResourceID), "\r\n") { return fmt.Errorf("DTS resource ID cannot contain newlines") } - if o.ResourceID != "" && strings.TrimSpace(o.ResourceID) == "" { - return fmt.Errorf("DTS resource ID cannot be blank") + if o.ResourceID != "" && tokenScope(o.ResourceID) == "/.default" { + return fmt.Errorf("DTS resource ID cannot be empty after normalization; set ResourceID to a token audience URI or leave it exactly empty to use the default") } if _, err := api.NormalizeLargePayloadOptions(o.LargePayloads); err != nil { return fmt.Errorf("invalid DTS large payload options: %w", err) @@ -339,6 +371,20 @@ func (o *Options) Validate() error { if authentication == "" { authentication = AuthenticationDefaultAzure } + if o.AuthorityHost != "" { + authority, err := url.Parse(strings.TrimSpace(o.AuthorityHost)) + if err != nil || authority.Scheme != "https" || authority.Hostname() == "" || + authority.User != nil || authority.RawQuery != "" || authority.Fragment != "" || + (authority.Path != "" && authority.Path != "/") { + return fmt.Errorf("DTS AuthorityHost must be an HTTPS authority URL without a path, credentials, query, or fragment") + } + switch authentication { + case AuthenticationManagedIdentity: + return fmt.Errorf("DTS Authentication ManagedIdentity does not use AuthorityHost; managed identity uses the hosting environment's identity endpoint") + case AuthenticationAzureCLI, AuthenticationAzurePowerShell: + return fmt.Errorf("DTS Authentication %s does not use AuthorityHost; configure the developer tool's cloud instead", authentication) + } + } switch authentication { case AuthenticationNone: if o.Credential != nil { @@ -392,6 +438,9 @@ func (o *Options) Validate() error { // mode can never consume. func (o *Options) identityFieldsInUse() []string { var fields []string + if strings.TrimSpace(o.AuthorityHost) != "" { + fields = append(fields, "AuthorityHost") + } if strings.TrimSpace(o.ClientID) != "" { fields = append(fields, "ClientID") } diff --git a/durabletaskscheduler/options_test.go b/durabletaskscheduler/options_test.go index d27bd465..93cf9a4e 100644 --- a/durabletaskscheduler/options_test.go +++ b/durabletaskscheduler/options_test.go @@ -10,6 +10,7 @@ import ( ) func TestNewOptionsFromConnectionString(t *testing.T) { + t.Setenv("REGION_NAME", "") tests := []struct { name string value string @@ -392,6 +393,7 @@ func TestNewOptionsFromConnectionStringDoesNotForceInsecureForHTTPS(t *testing.T } func TestNewOptionsDefaults(t *testing.T) { + t.Setenv("REGION_NAME", "") options := NewOptions("scheduler.example.com", "hub") require.Equal(t, "scheduler.example.com", options.EndpointAddress) require.Equal(t, "hub", options.TaskHubName) @@ -506,6 +508,7 @@ func TestOptionsValidateNilReceiver(t *testing.T) { } func TestPrepareOptionsAppliesDefaultsAndCopiesTenants(t *testing.T) { + t.Setenv("REGION_NAME", "") options := &Options{ EndpointAddress: "scheduler.example.com", TaskHubName: "hub", @@ -527,6 +530,7 @@ func TestPrepareOptionsAppliesDefaultsAndCopiesTenants(t *testing.T) { // exactly empty defaults, surrounding whitespace on a real value is trimmed, // and a whitespace-only value is rejected instead of collapsing to the default. func TestPrepareOptionsResourceIDNormalization(t *testing.T) { + t.Setenv("REGION_NAME", "") for _, test := range []struct { name string resourceID string @@ -551,9 +555,9 @@ func TestPrepareOptionsResourceIDNormalization(t *testing.T) { want: "https://custom.example.com", wantScope: "https://custom.example.com/.default", }, - {name: "spaces only", resourceID: " ", wantErr: "resource ID cannot be blank"}, - {name: "tab only", resourceID: "\t", wantErr: "resource ID cannot be blank"}, - {name: "newline only", resourceID: "\n", wantErr: "resource ID cannot contain newlines"}, + {name: "spaces only", resourceID: " ", wantErr: "resource ID cannot be empty after normalization"}, + {name: "tab only", resourceID: "\t", wantErr: "resource ID cannot be empty after normalization"}, + {name: "newline only", resourceID: "\n", wantErr: "resource ID cannot be empty after normalization"}, } { t.Run(test.name, func(t *testing.T) { options := NewOptions("scheduler.example.com", "hub") diff --git a/durabletaskscheduler/resource_id_test.go b/durabletaskscheduler/resource_id_test.go new file mode 100644 index 00000000..d8f76133 --- /dev/null +++ b/durabletaskscheduler/resource_id_test.go @@ -0,0 +1,442 @@ +package durabletaskscheduler + +import ( + "context" + "crypto/tls" + "crypto/x509" + "net/http/httptest" + "os" + "os/exec" + "testing" + "time" + + "github.com/Azure/azure-sdk-for-go/sdk/azcore" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/microsoft/durabletask-go/api" + durabletaskclient "github.com/microsoft/durabletask-go/client" + "github.com/microsoft/durabletask-go/internal/protos" + "github.com/microsoft/durabletask-go/task" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" + "google.golang.org/protobuf/types/known/emptypb" +) + +var resourceIDCases = []struct { + name string + region string + resourceID string + want string +}{ + {name: "unset region", want: DefaultResourceID}, + {name: "empty region", want: DefaultResourceID}, + {name: "public", region: "westus2", want: DefaultResourceID}, + {name: "china is not inferred", region: "chinaeast2", want: DefaultResourceID}, + {name: "government substring", region: "notusgov", want: DefaultResourceID}, + {name: "DoD substring", region: "notusdod", want: DefaultResourceID}, + {name: "government", region: "usgovvirginia", want: GovernmentResourceID}, + {name: "government uppercase", region: "USGOVARIZONA", want: GovernmentResourceID}, + {name: "government mixed case", region: "UsGovTexas", want: GovernmentResourceID}, + {name: "DoD", region: "usdodcentral", want: GovernmentResourceID}, + {name: "DoD uppercase", region: "USDODEAST", want: GovernmentResourceID}, + {name: "DoD mixed case", region: "UsDodCentral", want: GovernmentResourceID}, + {name: "exact government prefix", region: "usgov", want: GovernmentResourceID}, + {name: "exact DoD prefix", region: "usdod", want: GovernmentResourceID}, + {name: "leading space is not a prefix", region: " usgovvirginia", want: DefaultResourceID}, + {name: "public override government", region: "usgovvirginia", resourceID: DefaultResourceID, want: DefaultResourceID}, + {name: "public override DoD", region: "usdodcentral", resourceID: DefaultResourceID, want: DefaultResourceID}, + {name: "government override public", region: "westus2", resourceID: GovernmentResourceID, want: GovernmentResourceID}, + {name: "custom", region: "chinaeast2", resourceID: "https://Custom.Example/Resource", want: "https://Custom.Example/Resource"}, + {name: "trailing slash", resourceID: GovernmentResourceID + "/", want: GovernmentResourceID}, + {name: "existing suffix", resourceID: GovernmentResourceID + "/.default", want: GovernmentResourceID}, + {name: "uppercase suffix and slashes", resourceID: GovernmentResourceID + "//.DEFAULT//", want: GovernmentResourceID}, + {name: "surrounding whitespace", resourceID: " \t\n" + GovernmentResourceID + "/.default/ \r\n", want: GovernmentResourceID}, + {name: "custom URI casing", region: "usgovvirginia", resourceID: "api://CustomAudience/resource/.DEFAULT/", want: "api://CustomAudience/resource"}, + {name: "strip only one suffix", resourceID: "api://custom/.default/.default", want: "api://custom/.default"}, + {name: "preserve meaningful suffix casing", resourceID: "api://custom/.DEFAULT//.default/", want: "api://custom/.DEFAULT"}, + {name: "preserve equals", resourceID: "api://Custom/resource=value", want: "api://Custom/resource=value"}, +} + +func setResourceRegion(t *testing.T, name, region string) { + t.Helper() + t.Setenv("REGION_NAME", region) + if name == "unset region" { + require.NoError(t, os.Unsetenv("REGION_NAME")) + } +} + +func recordedTokenOptions(c *recordingCredential) []policy.TokenRequestOptions { + c.mu.Lock() + defer c.mu.Unlock() + return append([]policy.TokenRequestOptions(nil), c.options...) +} + +func assertResourceScopeAndRefresh(t *testing.T, options *Options, want string) { + t.Helper() + for _, role := range []connectionRole{clientRole, workerRole} { + credential := &recordingCredential{} + prepared, err := prepareOptionsWith(options, func(credentialSpec) (azcore.TokenCredential, error) { + return credential, nil + }) + require.NoError(t, err) + if options.Authentication == AuthenticationTokenCredential { + var ok bool + credential, ok = prepared.Credential.(*recordingCredential) + require.True(t, ok) + } + before := len(recordedTokenOptions(credential)) + perRPC := newPerRPCCredentials(&prepared, role, "worker") + require.Len(t, recordedTokenOptions(credential), before, "construction must not acquire a token") + for range 2 { + values, err := perRPC.GetRequestMetadata(context.Background()) + require.NoError(t, err) + if options.Authentication == AuthenticationNone { + require.NotContains(t, values, "authorization") + } else { + require.Equal(t, "Bearer token", values["authorization"]) + } + } + if options.Authentication == AuthenticationNone { + require.Empty(t, recordedTokenOptions(credential)) + continue + } + require.Len(t, recordedTokenOptions(credential), before+1, "fresh tokens must be cached") + perRPC.token.Store(&schedulerCredentialState{ + refreshAfter: time.Now().Add(-time.Hour), + validUntil: time.Now().Add(-time.Minute), + }) + errs := callSchedulerCredentialsConcurrently(perRPC, 16) + for range 16 { + require.NoError(t, <-errs) + } + calls := recordedTokenOptions(credential)[before:] + require.Len(t, calls, 2) + for _, call := range calls { + require.Equal(t, []string{want + "/.default"}, call.Scopes) + require.Empty(t, call.TenantID) + } + } +} + +func TestResourceIDScopesForEveryAuthenticationMode(t *testing.T) { + for _, tt := range resourceIDCases { + t.Run(tt.name, func(t *testing.T) { + setResourceRegion(t, tt.name, tt.region) + for _, authentication := range allAuthenticationTypes { + t.Run(string(authentication), func(t *testing.T) { + for _, source := range []string{"options", "connection string"} { + if source == "connection string" && authentication == AuthenticationTokenCredential { + continue + } + t.Run(source, func(t *testing.T) { + options := NewOptions("https://scheduler.example.com", "hub") + options.Authentication = authentication + options.ResourceID = tt.resourceID + if source == "connection string" { + var err error + options, err = NewOptionsFromConnectionString( + "Endpoint=https://scheduler.example.com;TaskHub=hub;Authentication=" + + string(authentication) + ";ResourceId=" + tt.resourceID, + ) + require.NoError(t, err) + } + if authentication == AuthenticationTokenCredential { + options.Credential = &recordingCredential{} + } + original := options.ResourceID + assertResourceScopeAndRefresh(t, options, tt.want) + require.Equal(t, original, options.ResourceID, "preparation must not mutate caller options") + require.Equal(t, "https://scheduler.example.com", options.EndpointAddress) + require.Empty(t, options.AuthorityHost) + }) + } + }) + } + }) + } +} + +func TestResourceIDDefaultsArePerInstance(t *testing.T) { + for _, source := range []string{"NewOptions", "NewOptionsWithCredential", "connection string", "connection string empty", "literal", "reset empty"} { + t.Run(source, func(t *testing.T) { + newOptions := func() *Options { + credential := &recordingCredential{} + switch source { + case "NewOptions": + return NewOptions("scheduler.example.com", "hub") + case "NewOptionsWithCredential": + return NewOptionsWithCredential("scheduler.example.com", "hub", credential) + case "connection string", "connection string empty": + connectionString := "Endpoint=scheduler.example.com;TaskHub=hub;Authentication=DefaultAzure" + if source == "connection string empty" { + connectionString += ";ResourceId=" + } + options, err := NewOptionsFromConnectionString(connectionString) + require.NoError(t, err) + return options + case "literal": + return &Options{EndpointAddress: "scheduler.example.com", TaskHubName: "hub"} + default: + options := NewOptions("scheduler.example.com", "hub") + options.ResourceID = "" + return options + } + } + var instances []Options + for _, region := range []string{"UsGovVirginia", "westus2", "USDODEAST"} { + t.Setenv("REGION_NAME", region) + options := newOptions() + if source != "literal" && source != "reset empty" { + t.Setenv("REGION_NAME", "changed-after-options") + } + prepared, err := prepareOptionsWith(options, func(credentialSpec) (azcore.TokenCredential, error) { + return &recordingCredential{}, nil + }) + require.NoError(t, err) + instances = append(instances, prepared) + } + t.Setenv("REGION_NAME", "chinaeast2") + for i := range instances { + want := GovernmentResourceID + if i == 1 { + want = DefaultResourceID + } + assertResourceScopeAndRefresh(t, &instances[i], want) + } + }) + } +} + +func TestInvalidResourceIDsFailBeforeAuthentication(t *testing.T) { + for _, resourceID := range []string{" \t ", "\r\n", "///", "/.default", "/.DEFAULT///", " /.DEFAULT/// "} { + t.Run(resourceID, func(t *testing.T) { + for _, authentication := range allAuthenticationTypes { + options := NewOptions("scheduler.example.com", "hub") + options.Authentication = authentication + options.ResourceID = resourceID + require.ErrorContains(t, options.Validate(), "resource ID cannot be empty after normalization") + _, err := prepareOptionsWith(options, func(credentialSpec) (azcore.TokenCredential, error) { + t.Fatal("invalid ResourceID must be rejected before credential construction") + return nil, nil + }) + require.ErrorContains(t, err, "resource ID cannot be empty after normalization") + client, err := NewClient(context.Background(), options, nil) + require.Nil(t, client) + require.ErrorContains(t, err, "resource ID cannot be empty after normalization") + worker, err := NewWorker(options, task.NewTaskRegistry(), nil) + require.Nil(t, worker) + require.ErrorContains(t, err, "resource ID cannot be empty after normalization") + if authentication != AuthenticationTokenCredential { + _, err = NewOptionsFromConnectionString("Endpoint=scheduler.example.com;TaskHub=hub;Authentication=" + + string(authentication) + ";ResourceId=" + resourceID) + require.ErrorContains(t, err, "resource ID cannot be empty after normalization") + } + } + }) + } +} + +func TestResourceIDConnectionStringKeysAndDuplicates(t *testing.T) { + t.Setenv("REGION_NAME", "usgovvirginia") + for _, key := range []string{"ResourceId", "resourceid", "RESOURCEID", "ReSoUrCeId"} { + for _, value := range []string{"", "api://Custom/.default/.DEFAULT/"} { + options, err := NewOptionsFromConnectionString( + "Endpoint=scheduler.example.com;TaskHub=hub;Authentication=DefaultAzure;ResourceId=///;" + key + "=" + value, + ) + require.NoError(t, err) + want := "api://Custom/.default" + if value == "" { + want = GovernmentResourceID + } + assertResourceScopeAndRefresh(t, options, want) + } + _, err := NewOptionsFromConnectionString( + "Endpoint=scheduler.example.com;TaskHub=hub;Authentication=None;ResourceId=api://valid;" + key + "= ", + ) + require.ErrorContains(t, err, "resource ID cannot be empty after normalization") + } +} + +// Called only in an isolated test subprocess because fallback roots are +// process-global and cannot be restored. No system certificate is installed. +func audienceTestTLS() *tls.Config { + server := httptest.NewTLSServer(nil) + defer server.Close() + roots := x509.NewCertPool() + roots.AddCert(server.Certificate()) + x509.SetFallbackRoots(roots) + return server.TLS.Clone() +} + +type audienceServer struct { + metadataServer + disconnect chan struct{} +} + +func (s *audienceServer) RaiseEvent(context.Context, *protos.RaiseEventRequest) (*protos.RaiseEventResponse, error) { + return nil, status.Error(codes.Unavailable, "replace the test channel") +} + +func (s *audienceServer) GetWorkItems(_ *protos.GetWorkItemsRequest, stream protos.TaskHubSidecarService_GetWorkItemsServer) error { + incoming, _ := metadata.FromIncomingContext(stream.Context()) + s.metadata <- incoming + select { + case <-s.disconnect: + return status.Error(codes.Unavailable, "reconnect the test worker") + case <-stream.Context().Done(): + return nil + } +} + +func awaitAudienceMetadata(t *testing.T, server *audienceServer) metadata.MD { + t.Helper() + select { + case incoming := <-server.metadata: + require.Equal(t, []string{"Bearer token"}, incoming.Get("authorization")) + return incoming + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for authenticated RPC") + return nil + } +} + +func TestPublicAuthenticationPathsPreserveResourceIDAcrossReconnects(t *testing.T) { + if os.Getenv("DTS_AUDIENCE_TLS_TEST") != "1" { + command := exec.Command(os.Args[0], "-test.run=^TestPublicAuthenticationPathsPreserveResourceIDAcrossReconnects$", "-test.timeout=2m") + command.Env = append(os.Environ(), "DTS_AUDIENCE_TLS_TEST=1") + output, err := command.CombinedOutput() + require.NoError(t, err, "%s", output) + return + } + t.Setenv("GODEBUG", os.Getenv("GODEBUG")+",x509usefallbackroots=1") + tlsConfig := audienceTestTLS() + for _, tt := range resourceIDCases { + t.Run(tt.name, func(t *testing.T) { + for _, path := range []string{"client", "worker", "compatibility listener"} { + t.Run(path, func(t *testing.T) { + setResourceRegion(t, tt.name, tt.region) + server := &audienceServer{ + metadataServer: metadataServer{metadata: make(chan metadata.MD, 16)}, + disconnect: make(chan struct{}, 1), + } + listener := bufconn.Listen(1024 * 1024) + grpcServer := grpc.NewServer(grpc.Creds(credentials.NewTLS(tlsConfig))) + protos.RegisterTaskHubSidecarServiceServer(grpcServer, server) + go func() { _ = grpcServer.Serve(listener) }() + defer grpcServer.Stop() + defer func() { require.NoError(t, listener.Close()) }() + + credential := &recordingCredential{} + if tt.name == "government" { + credential.refreshOn = time.Now().Add(-time.Hour) + } + options := NewOptionsWithCredential("https://127.0.0.1", "hub", credential) + options.ResourceID = tt.resourceID + options.dialer = bufconnDialer(listener) + options.ChannelRecreateFailureThreshold = 1 + options.ChannelRecreateMinInterval = 0 + connections := make(chan *grpc.ClientConn, 1) + options.UnaryInterceptors = []grpc.UnaryClientInterceptor{ + func(ctx context.Context, method string, request, reply any, connection *grpc.ClientConn, invoker grpc.UnaryInvoker, callOptions ...grpc.CallOption) error { + select { + case connections <- connection: + default: + } + return invoker(ctx, method, request, reply, connection, callOptions...) + }, + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + wantTokenCalls := 2 + refreshTransport := func() { + if tt.name != "government" { + return + } + connection := <-connections + // Exercise refresh on the same public transport after its + // minimum one-second cache interval, before recreating it. + require.Eventually(t, func() bool { + _, err := protos.NewTaskHubSidecarServiceClient(connection).Hello(ctx, &emptypb.Empty{}) + require.NoError(t, err) + awaitAudienceMetadata(t, server) + return len(recordedTokenOptions(credential)) == 2 + }, 3*time.Second, 20*time.Millisecond) + wantTokenCalls++ + } + + if path == "worker" { + worker, err := NewWorker(options, task.NewTaskRegistry(), nil, + durabletaskclient.WithWorkerReconnectBackoff(time.Millisecond, time.Millisecond)) + require.NoError(t, err) + require.Empty(t, recordedTokenOptions(credential), "worker token acquisition stays lazy") + t.Setenv("REGION_NAME", "changed-after-worker-construction") + options.ResourceID = "api://changed-after-worker-construction" + require.NoError(t, worker.Start(ctx)) + defer func() { require.NoError(t, worker.Shutdown(ctx)) }() + hello := awaitAudienceMetadata(t, server) + stream := awaitAudienceMetadata(t, server) + require.Equal(t, hello.Get("workerid"), stream.Get("workerid")) + require.Len(t, recordedTokenOptions(credential), 1) + refreshTransport() + server.disconnect <- struct{}{} + reconnectedHello := awaitAudienceMetadata(t, server) + reconnectedStream := awaitAudienceMetadata(t, server) + require.Equal(t, hello.Get("workerid"), reconnectedHello.Get("workerid")) + require.Equal(t, hello.Get("workerid"), reconnectedStream.Get("workerid")) + } else { + client, err := NewClient(ctx, options, nil) + require.NoError(t, err) + defer func() { require.NoError(t, client.Close()) }() + awaitAudienceMetadata(t, server) + require.Len(t, recordedTokenOptions(credential), 1, "client acquires a token for its eager Hello") + if path == "compatibility listener" { + require.NoError(t, client.StartWorkItemListener(ctx, task.NewTaskRegistry(), + durabletaskclient.WithWorkerReconnectBackoff(time.Millisecond, time.Millisecond))) + awaitAudienceMetadata(t, server) + awaitAudienceMetadata(t, server) + require.Len(t, recordedTokenOptions(credential), 1, "listener shares the client's cached token") + } + t.Setenv("REGION_NAME", "changed-after-client-construction") + options.ResourceID = "api://changed-after-client-construction" + refreshTransport() + client.connection.mu.Lock() + old := client.connection.current + client.connection.mu.Unlock() + require.Error(t, client.RaiseEvent(ctx, api.InstanceID("test"), "recreate")) + awaitAudienceMetadata(t, server) + // The replacement Hello can arrive before the swap completes. + require.Eventually(t, func() bool { + client.connection.mu.Lock() + defer client.connection.mu.Unlock() + return !client.connection.recreateInFlight && client.connection.current != old + }, 5*time.Second, time.Millisecond) + if path == "compatibility listener" { + server.disconnect <- struct{}{} + awaitAudienceMetadata(t, server) + awaitAudienceMetadata(t, server) + } + } + calls := recordedTokenOptions(credential) + require.Len(t, calls, wantTokenCalls) + for _, call := range calls { + require.Equal(t, []string{tt.want + "/.default"}, call.Scopes) + } + require.Empty(t, options.AuthorityHost) + require.Equal(t, "https://127.0.0.1", options.EndpointAddress) + }) + } + }) + } +} + +func TestResourceIDDoesNotInferAudienceFromEndpoint(t *testing.T) { + t.Setenv("REGION_NAME", "") + for _, endpoint := range []string{"https://scheduler.durabletask.azure.us", "https://scheduler.durabletask.io"} { + options := NewOptions(endpoint, "hub") + assertResourceScopeAndRefresh(t, options, DefaultResourceID) + } +} diff --git a/samples/authentication/README.md b/samples/authentication/README.md index 38817e11..2f63ac76 100644 --- a/samples/authentication/README.md +++ b/samples/authentication/README.md @@ -25,6 +25,37 @@ for the supported credential modes and their settings. No account keys or access tokens belong in source code or logs. +## Azure Government + +Configure all three independently: the scheduler endpoint, the token audience, +and the credential authority. This connection string runs both halves of the +sample against your government scheduler: + +```bash +export DTS_CONNECTION_STRING='Endpoint=https://;TaskHub=;Authentication=DefaultAzure;ResourceId=https://durabletask.azure.us;AuthorityHost=https://login.microsoftonline.us/' +go run ./samples/authentication +``` + +The programmatic half copies `ResourceID` onto the scheduler options and forwards +`AuthorityHost` to `DefaultAzureCredential` through +`azcore.ClientOptions.Cloud.ActiveDirectoryAuthorityHost`. Omitting +`AuthorityHost` preserves Azure Identity defaults, including `AZURE_AUTHORITY_HOST`. +For CLI-backed local development, first configure +`az cloud set --name AzureUSGovernment` and then `az login`; setting an authority on +`DefaultAzureCredential` does not configure developer tools. + +Omitting `ResourceId`, or using `ResourceId=`, selects the government audience +when `REGION_NAME` starts with `usgov` or `usdod` (case-insensitively). This is an +intentional default change; all other regions retain `https://durabletask.io`. +An explicit audience wins in either direction. It is a token audience URI, +not an ARM resource path, and never changes the authority or endpoint. + +For `Authentication=ManagedIdentity`, omit `AuthorityHost`: managed identities +use their hosting environment's identity endpoint. For `AzureCLI` or +`AzurePowerShell`, configure the tool's cloud separately rather than supplying +`AuthorityHost`. See the [SDK reference](../../durabletaskscheduler/README.md#azure-government-example) +for caller-supplied credential configuration. + ## Expected result and cleanup Both modes report their completed, uniquely named instance and the process ends @@ -35,3 +66,7 @@ The sample terminates/purges only its own instance IDs and descendants, then stops its workers and closes its clients. It never resets or deletes the task hub. Missing credentials, missing data-plane permissions, incorrect results, and cleanup failures produce a nonzero exit. + +The SDK's automated audience tests use recording credentials and local TLS gRPC +servers; they do not establish live Azure Government authentication. Running this +sample with a real, authorized government identity is a separate live check. diff --git a/samples/authentication/authentication.go b/samples/authentication/authentication.go index 6631a212..6e91dbcf 100644 --- a/samples/authentication/authentication.go +++ b/samples/authentication/authentication.go @@ -10,6 +10,8 @@ import ( "net/url" "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/azidentity" "github.com/microsoft/durabletask-go/api" "github.com/microsoft/durabletask-go/durabletaskscheduler" @@ -43,6 +45,9 @@ func run() error { } credential, err := azidentity.NewDefaultAzureCredential(&azidentity.DefaultAzureCredentialOptions{ + ClientOptions: azcore.ClientOptions{ + Cloud: cloud.Configuration{ActiveDirectoryAuthorityHost: options.AuthorityHost}, + }, TenantID: options.TenantID, }) if err != nil {