From 7ed62df0410f516079a89e4ea3fd0a0b0b37ae2a Mon Sep 17 00:00:00 2001 From: "David Muto (pseudomuto)" Date: Wed, 2 Sep 2026 15:23:54 -0400 Subject: [PATCH] [cloud]: Validate Cloud Namespaces The canonical "Cloud" config derives its endpoint from the translated `namespace: {{ .RemoteNamespace }}.tmprl.cloud:7233` with a ".$TEMPORAL_ACCOUNT" suffix. Config loading expands environment variables with os.Expand, which substitutes empty for an unset one, so a blank TEMPORAL_ACCOUNT silently leaves a bare-dot suffix and every namespace translates to ".". The proxy would then dial "..tmprl.cloud" and the operator saw nothing but a DNS failure. More generally, the proxy had no notion of a Cloud upstream, so a namespace Cloud would reject was indistinguishable from a valid one. When an upstream is Temporal Cloud, namespaces (post-translation) are now validated. For non-templated hosts, this is done at startup to prevent mistakes. For dynamic ones, a DEBUG log is printed indicating why the namespace is invalid. Requests are still forwarded upstream for now. > NOTE: Auto-detection means a config the proxy used to accept can now fail to start, but only one that could never have worked against Cloud anyways. --- internal/cloud/doc.go | 8 ++ internal/cloud/endpoint.go | 25 ++++ internal/cloud/endpoint_test.go | 37 +++++ internal/cloud/namespace.go | 101 ++++++++++++++ internal/cloud/namespace_test.go | 191 ++++++++++++++++++++++++++ internal/config/upstream.go | 60 +++++++++ internal/config/upstream_test.go | 222 ++++++++++++++++++++++++++++++ internal/dataplane/dataplane.go | 12 ++ internal/proxy/cloudns.go | 98 ++++++++++++++ internal/proxy/cloudns_test.go | 223 +++++++++++++++++++++++++++++++ 10 files changed, 977 insertions(+) create mode 100644 internal/cloud/doc.go create mode 100644 internal/cloud/endpoint.go create mode 100644 internal/cloud/endpoint_test.go create mode 100644 internal/cloud/namespace.go create mode 100644 internal/cloud/namespace_test.go create mode 100644 internal/proxy/cloudns.go create mode 100644 internal/proxy/cloudns_test.go diff --git a/internal/cloud/doc.go b/internal/cloud/doc.go new file mode 100644 index 0000000..452fa62 --- /dev/null +++ b/internal/cloud/doc.go @@ -0,0 +1,8 @@ +// Package cloud holds the rules that are specific to Temporal Cloud rather than +// to any Temporal Service. +// +// Today that means namespace validation. Cloud identifies a namespace as +// ".", a shape self-hosted deployments do not impose, so +// [ValidateNamespace] checks a string against it before the proxy uses that +// string to address a Cloud upstream. +package cloud diff --git a/internal/cloud/endpoint.go b/internal/cloud/endpoint.go new file mode 100644 index 0000000..5c5f322 --- /dev/null +++ b/internal/cloud/endpoint.go @@ -0,0 +1,25 @@ +package cloud + +import ( + "net" + "strings" +) + +// endpointSuffix is the domain Temporal Cloud serves its endpoints from. +const endpointSuffix = ".tmprl.cloud" + +// IsEndpoint reports whether hostPort addresses Temporal Cloud. The port is +// optional, and a template action in place of the host is tolerated, since a +// templated address is rendered per request but keeps its domain. +// +// This recognizes per-namespace and regional endpoints. Private-link endpoints +// use per-VPC hostnames that carry no Cloud domain, so those have to be declared +// rather than detected. +func IsEndpoint(hostPort string) bool { + host := hostPort + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + + return strings.HasSuffix(host, endpointSuffix) +} diff --git a/internal/cloud/endpoint_test.go b/internal/cloud/endpoint_test.go new file mode 100644 index 0000000..de734f2 --- /dev/null +++ b/internal/cloud/endpoint_test.go @@ -0,0 +1,37 @@ +package cloud_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/internal/cloud" +) + +func TestIsEndpoint(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + hostPort string + want bool + }{ + {name: "per-namespace endpoint", hostPort: "quickstart.a1b2c.tmprl.cloud:7233", want: true}, + {name: "regional endpoint", hostPort: "us-west-2.aws.tmprl.cloud:7233", want: true}, + {name: "templated host", hostPort: "{{ .RemoteNamespace }}.tmprl.cloud:7233", want: true}, + {name: "no port", hostPort: "quickstart.a1b2c.tmprl.cloud", want: true}, + {name: "self-hosted", hostPort: "localhost:7233"}, + {name: "private-link hostname", hostPort: "vpce-0abc123.vpce-svc-0def456.us-east-1.vpce.amazonaws.com:7233"}, + {name: "lookalike domain", hostPort: "evil-tmprl.cloud:7233"}, + {name: "bare domain", hostPort: "tmprl.cloud:7233"}, + {name: "empty"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + require.Equal(t, tt.want, cloud.IsEndpoint(tt.hostPort)) + }) + } +} diff --git a/internal/cloud/namespace.go b/internal/cloud/namespace.go new file mode 100644 index 0000000..9aed9eb --- /dev/null +++ b/internal/cloud/namespace.go @@ -0,0 +1,101 @@ +package cloud + +import ( + "errors" + "fmt" + "regexp" + "strings" + + "github.com/temporalio/temporal-proxy/pkg/validation" +) + +const ( + nameMinLen = 2 + nameMaxLen = 39 + + accountIDMinLen = 5 + accountIDMaxLen = 20 +) + +// start with letter, end with letter/number, contain only a-z0-9- +var nsNameRegex = regexp.MustCompile(`^[a-z][a-z0-9-]*[a-z0-9]$`) + +// ValidateAccountID checks that id is shaped like the account-id label of a +// Temporal Cloud namespace. +// +// See: https://docs.temporal.io/cloud/namespaces for details +func ValidateAccountID(id string) error { + return validation.Validate("", accountIDRule(id)) +} + +// ValidateNamespace checks that ns is a well-formed Temporal Cloud namespace +// identifier, meaning ".". Every broken rule is reported, not +// just the first, so a caller can show the whole story at once. +// +// See: https://docs.temporal.io/cloud/namespaces for details +func ValidateNamespace(ns string) error { + // Exactly one separator: without it there is no account id, and with more + // than one there is no telling which label is which. + name, accountID, found := strings.Cut(ns, ".") + malformed := !found || strings.Contains(accountID, ".") + + return validation.Validate( + "", + validation.WhenRules( + func() bool { return malformed }, + func() validation.Errors { + return validation.Errors{{ + Field: "id", + Message: "is malformed. Should be .", + }} + }, + ), + validation.WhenRules( + func() bool { return !malformed }, + validation.Field( + "name", + name, + validation.Required[string](), + size(nameMinLen, nameMaxLen), + func(v string) error { + if v != strings.ToLower(v) { + return errors.New("must be lowercase") + } + + return nil + }, + func(v string) error { + if !nsNameRegex.MatchString(v) { + return errors.New("must begin with a letter, end with a letter or number, and contain only letters, numbers, and the hyphen") + } + + return nil + }, + ), + accountIDRule(accountID), + ), + ) +} + +// accountIDRule builds the account-id checks, shared by [ValidateAccountID] and +// the second label of [ValidateNamespace] so the rule is defined once. +func accountIDRule(id string) validation.Rule { + return validation.Field( + "account-id", + id, + validation.Required[string](), + size(accountIDMinLen, accountIDMaxLen), + ) +} + +// size builds a check rejecting strings shorter than n or longer than m, both +// inclusive. +func size(n, m int) validation.Check[string] { + return func(s string) error { + if len(s) < n || len(s) > m { + return fmt.Errorf("must be between %d and %d characters", n, m) + } + + return nil + } +} diff --git a/internal/cloud/namespace_test.go b/internal/cloud/namespace_test.go new file mode 100644 index 0000000..0e4ee8b --- /dev/null +++ b/internal/cloud/namespace_test.go @@ -0,0 +1,191 @@ +package cloud_test + +import ( + "errors" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/internal/cloud" + "github.com/temporalio/temporal-proxy/pkg/validation" +) + +const ( + malformedID = "is malformed. Should be ." + nameShape = "must begin with a letter, end with a letter or number, and contain only letters, numbers, and the hyphen" + nameSize = "must be between 2 and 39 characters" + accountSize = "must be between 5 and 20 characters" +) + +func TestValidateAccountID(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + id string + wantErrs []validation.Error + }{ + {name: "typical account id", id: "a1b2c"}, + {name: "longest", id: strings.Repeat("a", 20)}, + { + name: "empty", + id: "", + wantErrs: []validation.Error{ + {Field: "account-id", Message: "is required"}, + {Field: "account-id", Message: accountSize}, + }, + }, + { + name: "too short", + id: "a1b2", + wantErrs: []validation.Error{{Field: "account-id", Message: accountSize}}, + }, + { + name: "too long", + id: strings.Repeat("a", 21), + wantErrs: []validation.Error{{Field: "account-id", Message: accountSize}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + err := cloud.ValidateAccountID(tt.id) + if len(tt.wantErrs) == 0 { + require.NoError(t, err) + return + } + + var errs validation.Errors + require.True(t, errors.As(err, &errs), "expected validation.Errors, got %T", err) + require.ElementsMatch(t, tt.wantErrs, []validation.Error(errs)) + }) + } +} + +func TestValidateNamespace(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + ns string + wantErrs []validation.Error + }{ + {name: "typical namespace", ns: "my-namespace.a2dd6"}, + {name: "digits in name", ns: "ns1.a2dd6"}, + {name: "shortest name", ns: "ab.a2dd6"}, + {name: "longest name", ns: strings.Repeat("a", 39) + ".a2dd6"}, + {name: "longest account id", ns: "myns." + strings.Repeat("a", 20)}, + {name: "account id is not name-shaped", ns: "myns.A_2xy"}, + { + name: "empty", + ns: "", + wantErrs: []validation.Error{{Field: "id", Message: malformedID}}, + }, + { + name: "no separator", + ns: "mynamespace", + wantErrs: []validation.Error{{Field: "id", Message: malformedID}}, + }, + { + name: "extra separator", + ns: "my.name.space", + wantErrs: []validation.Error{{Field: "id", Message: malformedID}}, + }, + { + name: "missing name", + ns: ".a2dd6", + wantErrs: []validation.Error{ + {Field: "name", Message: "is required"}, + {Field: "name", Message: nameSize}, + {Field: "name", Message: nameShape}, + }, + }, + { + name: "missing account id", + ns: "myns.", + wantErrs: []validation.Error{ + {Field: "account-id", Message: "is required"}, + {Field: "account-id", Message: accountSize}, + }, + }, + { + name: "name too short", + ns: "a.a2dd6", + wantErrs: []validation.Error{ + {Field: "name", Message: nameSize}, + {Field: "name", Message: nameShape}, + }, + }, + { + name: "name too long", + ns: strings.Repeat("a", 40) + ".a2dd6", + wantErrs: []validation.Error{{Field: "name", Message: nameSize}}, + }, + { + name: "uppercase name", + ns: "My-NS.a2dd6", + wantErrs: []validation.Error{ + {Field: "name", Message: "must be lowercase"}, + {Field: "name", Message: nameShape}, + }, + }, + { + name: "name starts with a hyphen", + ns: "-myns.a2dd6", + wantErrs: []validation.Error{{Field: "name", Message: nameShape}}, + }, + { + name: "name ends with a hyphen", + ns: "myns-.a2dd6", + wantErrs: []validation.Error{{Field: "name", Message: nameShape}}, + }, + { + name: "name starts with a digit", + ns: "1myns.a2dd6", + wantErrs: []validation.Error{{Field: "name", Message: nameShape}}, + }, + { + name: "underscore in name", + ns: "my_ns.a2dd6", + wantErrs: []validation.Error{{Field: "name", Message: nameShape}}, + }, + { + name: "account id too short", + ns: "myns.a2dd", + wantErrs: []validation.Error{{Field: "account-id", Message: accountSize}}, + }, + { + name: "account id too long", + ns: "myns." + strings.Repeat("a", 21), + wantErrs: []validation.Error{{Field: "account-id", Message: accountSize}}, + }, + { + name: "both labels bad", + ns: "-.abc", + wantErrs: []validation.Error{ + {Field: "name", Message: nameSize}, + {Field: "name", Message: nameShape}, + {Field: "account-id", Message: accountSize}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + err := cloud.ValidateNamespace(tt.ns) + if len(tt.wantErrs) == 0 { + require.NoError(t, err) + return + } + + var errs validation.Errors + require.True(t, errors.As(err, &errs), "expected validation.Errors, got %T", err) + require.ElementsMatch(t, tt.wantErrs, []validation.Error(errs)) + }) + } +} diff --git a/internal/config/upstream.go b/internal/config/upstream.go index 8d0b10d..3e4faa7 100644 --- a/internal/config/upstream.go +++ b/internal/config/upstream.go @@ -4,6 +4,7 @@ import ( "fmt" "strings" + "github.com/temporalio/temporal-proxy/internal/cloud" "github.com/temporalio/temporal-proxy/pkg/validation" ) @@ -12,8 +13,14 @@ type ( // workers to along with configuration for that remote cluster. Name // identifies the upstream so routing rules can refer to it; it must be // unique within the config. + // + // Cloud declares the upstream to be Temporal Cloud, which turns on + // Cloud-specific namespace rules. It is only needed for an address + // [cloud.IsEndpoint] does not recognize, such as a private-link hostname; a + // .tmprl.cloud address is detected without it. Upstream struct { Name string `yaml:"name"` + Cloud bool `yaml:"cloud"` Listen ListenConfig `yaml:",inline"` Namespaces NamespaceConfig `yaml:"namespaces"` Credentials *CredentialConfig `yaml:"credentials"` @@ -78,9 +85,22 @@ func (u *Upstream) Validate() error { return validation.Errors{{Field: "credentials", Message: "requires TLS to the upstream"}} }, ), + validation.WhenRules(u.IsCloud, u.cloudRules()...), ) } +// IsCloud reports whether the upstream is Temporal Cloud, either because it says +// so or because its address is a Cloud endpoint. The TLS server name counts too: +// a private-link upstream reaches Cloud through a per-VPC hostname but still +// pins Cloud's certificate. +func (u *Upstream) IsCloud() bool { + if u.Cloud || cloud.IsEndpoint(u.Listen.HostPort) { + return true + } + + return u.Listen.TLS != nil && cloud.IsEndpoint(u.Listen.TLS.ServerName) +} + // IsTemplated reports whether the upstream must be resolved per request because // its hostPort, or its TLS server name when one is configured, contains a // text/template action. @@ -92,6 +112,46 @@ func (u *Upstream) IsTemplated() bool { return u.Listen.TLS != nil && isTemplated(u.Listen.TLS.ServerName) } +// cloudRules builds the namespace rules that only hold for a Temporal Cloud +// upstream, where every remote name has to be a Cloud namespace identifier. +// They live here rather than under [NamespaceRules.Validate] because that runs a +// level down and cannot see whether the upstream is Cloud. +func (u *Upstream) cloudRules() []validation.Rule { + nsRules := &u.Namespaces.Rules + + return []validation.Rule{ + func() validation.Errors { + id, found := strings.CutPrefix(nsRules.Suffix, ".") + if nsRules.Suffix == "" || (found && cloud.ValidateAccountID(id) == nil) { + return nil + } + + return validation.Errors{{ + Subject: "namespaces.rules", + Field: "suffix", + Message: `must be "." for a Temporal Cloud upstream`, + }} + }, + func() validation.Errors { + var errs validation.Errors + for i, m := range nsRules.Overrides { + // An empty remote is already reported as required. + if m.Remote == "" || cloud.ValidateNamespace(m.Remote) == nil { + continue + } + + errs = append(errs, validation.Error{ + Subject: fmt.Sprintf("namespaces.rules.overrides[%d]", i), + Field: "remote", + Message: "must be a Temporal Cloud namespace (.)", + }) + } + + return errs + }, + } +} + func (ul UpstreamList) Validate() error { names := make([]string, len(ul)) hostPorts := make([]string, len(ul)) diff --git a/internal/config/upstream_test.go b/internal/config/upstream_test.go index 158eca6..ffafe09 100644 --- a/internal/config/upstream_test.go +++ b/internal/config/upstream_test.go @@ -616,3 +616,225 @@ func TestNamespaceRulesConfigured(t *testing.T) { }) } } + +func TestUpstream_IsCloud(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + upstream *config.Upstream + want bool + }{ + { + name: "per-namespace endpoint", + upstream: &config.Upstream{ + Name: "cloud", + Listen: config.ListenConfig{HostPort: "quickstart.a1b2c.tmprl.cloud:7233"}, + }, + want: true, + }, + { + name: "templated endpoint", + upstream: &config.Upstream{ + Name: "cloud", + Listen: config.ListenConfig{HostPort: "{{ .RemoteNamespace }}.tmprl.cloud:7233"}, + }, + want: true, + }, + { + name: "declared, address says nothing", + upstream: &config.Upstream{ + Name: "private-link", + Cloud: true, + Listen: config.ListenConfig{HostPort: "vpce-0abc123.us-east-1.vpce.amazonaws.com:7233"}, + }, + want: true, + }, + { + // A private-link upstream reaches Cloud through a per-VPC hostname but + // still pins Cloud's certificate, so the server name gives it away. + name: "cloud server name with a private-link address", + upstream: &config.Upstream{ + Name: "private-link", + Listen: config.ListenConfig{ + HostPort: "vpce-0abc123.us-east-1.vpce.amazonaws.com:7233", + TLS: &config.TLSConfig{ServerName: "quickstart.a1b2c.tmprl.cloud"}, + }, + }, + want: true, + }, + { + name: "self-hosted, no TLS", + upstream: &config.Upstream{ + Name: "local", + Listen: config.ListenConfig{HostPort: "localhost:7233"}, + }, + }, + { + name: "self-hosted with TLS", + upstream: &config.Upstream{ + Name: "local", + Listen: config.ListenConfig{ + HostPort: "temporal.internal:7233", + TLS: &config.TLSConfig{ServerName: "temporal.internal"}, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + require.Equal(t, tt.want, tt.upstream.IsCloud()) + }) + } +} + +func TestUpstream_Validate_Cloud(t *testing.T) { + t.Parallel() + + // Every case shares a Cloud address, so IsCloud is true without the flag + // except where the case says otherwise. + upstream := func(rules config.NamespaceRules) *config.Upstream { + return &config.Upstream{ + Name: "cloud", + Listen: config.ListenConfig{HostPort: "{{ .RemoteNamespace }}.tmprl.cloud:7233"}, + Namespaces: config.NamespaceConfig{Rules: rules}, + } + } + + tests := []struct { + name string + upstream *config.Upstream + wantTuples [][2]string + }{ + { + name: "account suffix", + upstream: upstream(config.NamespaceRules{Suffix: ".a1b2c"}), + }, + { + name: "no rules at all", + upstream: upstream(config.NamespaceRules{}), + }, + { + // The motivating bug: an unset TEMPORAL_ACCOUNT expands to empty, so + // the suffix is a bare dot and every namespace translates to ".". + name: "suffix is a bare dot", + upstream: upstream(config.NamespaceRules{Suffix: "."}), + wantTuples: [][2]string{{"namespaces.rules", "suffix"}}, + }, + { + name: "suffix has no leading dot", + upstream: upstream(config.NamespaceRules{Suffix: "-prod"}), + wantTuples: [][2]string{{"namespaces.rules", "suffix"}}, + }, + { + name: "suffix account id too short", + upstream: upstream(config.NamespaceRules{Suffix: ".a1b"}), + wantTuples: [][2]string{{"namespaces.rules", "suffix"}}, + }, + { + name: "override remote is a cloud namespace", + upstream: upstream(config.NamespaceRules{ + Overrides: []config.NamespaceMapping{{Local: "orders", Remote: "orders.a1b2c"}}, + }), + }, + { + name: "override remote is not a cloud namespace", + upstream: upstream(config.NamespaceRules{ + Overrides: []config.NamespaceMapping{{Local: "orders", Remote: "orders"}}, + }), + wantTuples: [][2]string{{"namespaces.rules.overrides[0]", "remote"}}, + }, + { + name: "only the offending override is reported", + upstream: upstream(config.NamespaceRules{ + Overrides: []config.NamespaceMapping{ + {Local: "orders", Remote: "orders.a1b2c"}, + {Local: "billing", Remote: "billing"}, + }, + }), + wantTuples: [][2]string{{"namespaces.rules.overrides[1]", "remote"}}, + }, + { + // An empty remote is already reported as required, so the cloud rule + // stays quiet rather than piling a second entry onto the same field. + name: "empty override remote reports required only", + upstream: upstream(config.NamespaceRules{ + Overrides: []config.NamespaceMapping{{Local: "orders", Remote: ""}}, + }), + wantTuples: [][2]string{{"namespaces.rules.overrides[0]", "remote"}}, + }, + { + name: "declared cloud with a private-link address", + upstream: &config.Upstream{ + Name: "private-link", + Cloud: true, + Listen: config.ListenConfig{HostPort: "vpce-0abc123.us-east-1.vpce.amazonaws.com:7233"}, + Namespaces: config.NamespaceConfig{Rules: config.NamespaceRules{Suffix: "."}}, + }, + wantTuples: [][2]string{{"namespaces.rules", "suffix"}}, + }, + { + // The gate is the whole point: a self-hosted upstream is free to use + // names Cloud would reject. + name: "self-hosted upstream is not held to cloud rules", + upstream: &config.Upstream{ + Name: "local", + Listen: config.ListenConfig{HostPort: "localhost:7233"}, + Namespaces: config.NamespaceConfig{Rules: config.NamespaceRules{ + Suffix: "-prod", + Overrides: []config.NamespaceMapping{{Local: "orders", Remote: "orders"}}, + }}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + err := tt.upstream.Validate() + if len(tt.wantTuples) == 0 { + require.NoError(t, err) + return + } + + var errs validation.Errors + require.True(t, errors.As(err, &errs), "expected validation.Errors, got %T", err) + + got := make([][2]string, len(errs)) + for i, e := range errs { + got[i] = [2]string{e.Subject, e.Field} + } + + require.ElementsMatch(t, tt.wantTuples, got) + }) + } +} + +func TestUpstreamCloudErrorPath(t *testing.T) { + t.Parallel() + + // The flag round-trips through YAML, and a cloud failure reports the whole + // dotted path so an operator can find the offending key. + cfg, err := config.Load(strings.NewReader(` +hostPort: 127.0.0.1:7233 +upstreams: + - name: private-link + cloud: true + hostPort: vpce-0abc123.us-east-1.vpce.amazonaws.com:7233 + namespaces: + rules: + suffix: . +`)) + require.NoError(t, err) + require.True(t, cfg.Upstreams[0].Cloud) + + require.EqualError( + t, + cfg.Validate(), + `upstreams[0].namespaces.rules: suffix: must be "." for a Temporal Cloud upstream`, + ) +} diff --git a/internal/dataplane/dataplane.go b/internal/dataplane/dataplane.go index 3725bd3..ee8da89 100644 --- a/internal/dataplane/dataplane.go +++ b/internal/dataplane/dataplane.go @@ -24,6 +24,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/transport/socket" "github.com/temporalio/temporal-proxy/pkg/crypto" "github.com/temporalio/temporal-proxy/pkg/logger" + "github.com/temporalio/temporal-proxy/pkg/logger/tag" ) type ( @@ -309,6 +310,17 @@ func newUpstreamTier( dialOpts = append(dialOpts, proxy.TranslationDialOptions(o.translator, rules.Remote, rules.Local)...) } + // Cloud derives its endpoint and authorizes requests by the translated + // namespace, so report one that cannot work there. Remote is identity when no + // rules are configured, which still catches a client sending a short name to + // an upstream that expects fully-qualified ones. + if up.IsCloud() { + dialOpts = append(dialOpts, proxy.CloudNamespaceDialOptions( + rules.Remote, + o.logger.With(tag.String("upstream", up.Name)), + )...) + } + cp, err := outbound.CredentialProviderFor(up.Credentials) if err != nil { return nil, nil, fmt.Errorf("invalid credentials for upstream %q: %w", up.Name, err) diff --git a/internal/proxy/cloudns.go b/internal/proxy/cloudns.go new file mode 100644 index 0000000..d486707 --- /dev/null +++ b/internal/proxy/cloudns.go @@ -0,0 +1,98 @@ +package proxy + +import ( + "context" + + "google.golang.org/grpc" + + "github.com/temporalio/temporal-proxy/internal/cloud" + "github.com/temporalio/temporal-proxy/internal/transport/meta" + "github.com/temporalio/temporal-proxy/pkg/logger" + "github.com/temporalio/temporal-proxy/pkg/logger/tag" +) + +// CloudNamespaceDialOptions returns the dial options that report a request whose +// translated namespace is not shaped like a Temporal Cloud namespace. out maps a +// local namespace to its remote name, and log may be nil, which reports nothing. +// Callers fold them into the dial options for the upstream connection, and only +// for an upstream that is Temporal Cloud. +// +// This is diagnostic: nothing is rejected and the request travels unchanged. +func CloudNamespaceDialOptions(out func(string) string, log logger.Logger) []grpc.DialOption { + if log != nil { + log = log.With(tag.Component("translation")) + } + + return []grpc.DialOption{ + grpc.WithChainUnaryInterceptor(cloudNamespaceUnaryInterceptor(out, log)), + grpc.WithChainStreamInterceptor(cloudNamespaceStreamInterceptor(out, log)), + } +} + +// cloudNamespaceUnaryInterceptor checks the translated namespace once per call. +func cloudNamespaceUnaryInterceptor(out func(string) string, log logger.Logger) grpc.UnaryClientInterceptor { + return func( + ctx context.Context, + method string, + req, reply any, + cc *grpc.ClientConn, + invoker grpc.UnaryInvoker, + opts ...grpc.CallOption, + ) error { + logNonCloudNamespace(ctx, log, method, out) + + return invoker(ctx, method, req, reply, cc, opts...) + } +} + +// cloudNamespaceStreamInterceptor checks the translated namespace once per stream +// open rather than once per message, since every message on a stream carries the +// same namespace. +func cloudNamespaceStreamInterceptor(out func(string) string, log logger.Logger) grpc.StreamClientInterceptor { + return func( + ctx context.Context, + desc *grpc.StreamDesc, + cc *grpc.ClientConn, + method string, + streamer grpc.Streamer, + opts ...grpc.CallOption, + ) (grpc.ClientStream, error) { + logNonCloudNamespace(ctx, log, method, out) + + return streamer(ctx, desc, cc, method, opts...) + } +} + +// logNonCloudNamespace emits a debug entry when the request's local namespace, +// translated through out, is not shaped like a Temporal Cloud namespace. A Cloud +// upstream derives its endpoint and authorizes requests by that name, so a +// malformed one otherwise fails as an opaque DNS or NotFound error. +// +// It is a no-op when log is nil, and when the request carries no local +// namespace: the router stamps the namespace header on every request, so +// translating an absent one would report a namespace no client asked for. +func logNonCloudNamespace(ctx context.Context, log logger.Logger, method string, out func(string) string) { + if log == nil { + return + } + + localNS := meta.NamespaceFrom(ctx) + if localNS == "" { + return + } + + remoteNS := out(localNS) + + err := cloud.ValidateNamespace(remoteNS) + if err == nil { + return + } + + log.Debug( + "outbound namespace is not valid", + tag.String("method", method), + tag.String("localNamespace", localNS), + tag.String("remoteNamespace", remoteNS), + tag.Error(err), + ) +} diff --git a/internal/proxy/cloudns_test.go b/internal/proxy/cloudns_test.go new file mode 100644 index 0000000..1ac156f --- /dev/null +++ b/internal/proxy/cloudns_test.go @@ -0,0 +1,223 @@ +package proxy_test + +import ( + "context" + "net" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + workflowservice "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/temporalio/temporal-proxy/internal/cloud" + "github.com/temporalio/temporal-proxy/internal/proxy" + "github.com/temporalio/temporal-proxy/internal/transport/meta" + "github.com/temporalio/temporal-proxy/pkg/logger" + "github.com/temporalio/temporal-proxy/pkg/logger/tag" +) + +// invalidNSMsg is the entry the cloud namespace check emits. +const invalidNSMsg = "outbound namespace is not valid" + +// countingLogger counts Debug entries, which TestLogger does not expose. +type countingLogger struct { + logger.Logger + + debugs *atomic.Int64 +} + +func TestCloudNamespaceDialOptionsUnary(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + ctx func(*testing.T) context.Context + out func(string) string + nilLog bool + wantTags []tag.Tag + }{ + { + name: "cloud-shaped translation is quiet", + ctx: namespacedContext("orders"), + out: cloudRemote, + }, + { + name: "not cloud-shaped is reported", + ctx: namespacedContext("orders"), + out: badRemote, + wantTags: []tag.Tag{ + tag.String("upstream", "cloud"), + tag.Component("translation"), + tag.String("method", "/svc/Method"), + tag.String("localNamespace", "orders"), + tag.String("remoteNamespace", "orders."), + tag.Error(cloud.ValidateNamespace("orders.")), + }, + }, + { + // A Cloud upstream with no translation rules expects clients to send + // fully-qualified names, so a short one is wrong on its own. + name: "identity translation still checks the local name", + ctx: namespacedContext("orders"), + out: func(s string) string { return s }, + wantTags: []tag.Tag{ + tag.String("upstream", "cloud"), + tag.Component("translation"), + tag.String("method", "/svc/Method"), + tag.String("localNamespace", "orders"), + tag.String("remoteNamespace", "orders"), + tag.Error(cloud.ValidateNamespace("orders")), + }, + }, + { + name: "no namespace on the context", + ctx: func(t *testing.T) context.Context { return t.Context() }, + out: badRemote, + }, + { + // The router stamps the header even for namespace-less calls such as + // GetSystemInfo, so an empty value must not be translated. + name: "empty namespace value", + ctx: namespacedContext(""), + out: badRemote, + }, + { + name: "nil logger", + ctx: namespacedContext("orders"), + out: badRemote, + nilLog: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + log := logger.NewTestLogger() + + var passed logger.Logger + if !tt.nilLog { + passed = log.With(tag.String("upstream", "cloud")) + } + + cc := clientWithOptions(t, "passthrough:///127.0.0.1:1", proxy.CloudNamespaceDialOptions(tt.out, passed)) + + ctx, cancel := context.WithCancel(tt.ctx(t)) + cancel() + + err := cc.Invoke( + ctx, + "/svc/Method", + &workflowservice.StartWorkflowExecutionRequest{}, + &workflowservice.StartWorkflowExecutionResponse{}, + ) + require.Error(t, err, "the canceled call still fails; the check does not swallow it") + + if len(tt.wantTags) == 0 { + require.False(t, log.Contains(invalidNSMsg)) + return + } + + require.True(t, log.ContainsEntry(logger.LevelDebug, invalidNSMsg, tt.wantTags...)) + }) + } +} + +func TestCloudNamespaceDialOptionsStreamChecksOncePerOpen(t *testing.T) { + t.Parallel() + + log := logger.NewTestLogger() + debugs := new(atomic.Int64) + counting := countingLogger{Logger: log.With(tag.String("upstream", "cloud")), debugs: debugs} + + // A live stream is the only way to send more than once, so this needs a real + // server. It only drains, which is enough to keep SendMsg deterministic. + cc := clientWithOptions(t, drainingServer(t), proxy.CloudNamespaceDialOptions(badRemote, counting)) + + cs, err := cc.NewStream( + meta.WithNamespace(t.Context(), "orders"), + &grpc.StreamDesc{ClientStreams: true, ServerStreams: true}, + "/svc/Stream", + ) + require.NoError(t, err) + + for range 3 { + require.NoError(t, cs.SendMsg(&workflowservice.StartWorkflowExecutionRequest{Namespace: "orders"})) + } + + require.Equal(t, int64(1), debugs.Load(), "opening the stream checks once; messages do not") + require.True(t, log.ContainsEntry( + logger.LevelDebug, invalidNSMsg, + tag.String("upstream", "cloud"), + tag.Component("translation"), + tag.String("method", "/svc/Stream"), + tag.String("localNamespace", "orders"), + tag.String("remoteNamespace", "orders."), + tag.Error(cloud.ValidateNamespace("orders.")), + )) +} + +func (c countingLogger) Debug(msg string, tags ...tag.Tag) { + c.debugs.Add(1) + c.Logger.Debug(msg, tags...) +} + +// With keeps the counter attached to derived loggers, which matters because +// CloudNamespaceDialOptions decorates the logger it is handed. +func (c countingLogger) With(tags ...tag.Tag) logger.Logger { + return countingLogger{Logger: c.Logger.With(tags...), debugs: c.debugs} +} + +// namespacedContext builds a context carrying ns the way the router stamps it. +func namespacedContext(ns string) func(*testing.T) context.Context { + return func(t *testing.T) context.Context { + return meta.WithNamespace(t.Context(), ns) + } +} + +// clientWithOptions returns a lazy client to target with opts installed. It +// dials nothing until a call is made. +func clientWithOptions(t *testing.T, target string, opts []grpc.DialOption) *grpc.ClientConn { + t.Helper() + + cc, err := grpc.NewClient( + target, + append([]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, opts...)..., + ) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + return cc +} + +// drainingServer starts a gRPC server that accepts any method and reads every +// message until the client stops sending, and returns its address. It exists so +// a test can send on a real stream without a service implementation. +func drainingServer(t *testing.T) string { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + srv := grpc.NewServer(grpc.UnknownServiceHandler(func(_ any, stream grpc.ServerStream) error { + for { + if err := stream.RecvMsg(new(workflowservice.StartWorkflowExecutionRequest)); err != nil { + return nil + } + } + })) + + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + return lis.Addr().String() +} + +// cloudRemote maps a local name to a well-formed Cloud namespace. +func cloudRemote(s string) string { return s + ".a1b2c" } + +// badRemote maps a local name the way an unset account variable would, leaving a +// trailing dot and no account id. +func badRemote(s string) string { return s + "." }