diff --git a/e2e/harness_test.go b/e2e/harness_test.go index a2875e5..6ee3b35 100644 --- a/e2e/harness_test.go +++ b/e2e/harness_test.go @@ -115,17 +115,19 @@ func newFullApp(t *testing.T, cfg *config.Config) *fx.App { } // newProxyApp is a minimal fx app for the socket-level test in this package: it -// wires proxy.Module plus protoutil.Module (which provides the Translator -// proxy.Module requires). internal/proxy/fx_test.go has its own copy (with -// support for extra fx.Options) that its unit tests still depend on; this is a -// deliberate small duplication rather than an exported helper, so the two -// packages stay decoupled. +// wires proxy.Module plus protoutil.Module (which provides the Translator) and +// connect.Module (which provides the Pool), both required by proxy.Module. +// internal/proxy/fx_test.go has its own copy (with support for extra +// fx.Options) that its unit tests still depend on; this is a deliberate small +// duplication rather than an exported helper, so the two packages stay +// decoupled. func newProxyApp(t *testing.T, cfg *config.Config) *fx.App { t.Helper() return fx.New( fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), + connect.Module, protoutil.Module, proxy.Module, fx.NopLogger, @@ -191,7 +193,7 @@ func dialInbound(t *testing.T, addr string) *grpc.ClientConn { } // dialUnix returns a client connection to the proxy's unix socket for the -// given upstream host. The socket path matches what proxy.Start binds. +// given upstream host. The socket path matches what proxy.Listen binds. // internal/proxy/server_test.go has its own copy that its unit tests still // depend on; this is a deliberate small duplication for the socket-level test // in this package. diff --git a/e2e/templated_upstream_test.go b/e2e/templated_upstream_test.go new file mode 100644 index 0000000..7ae2c74 --- /dev/null +++ b/e2e/templated_upstream_test.go @@ -0,0 +1,83 @@ +package e2e + +import ( + "context" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" + + "github.com/temporalio/temporal-proxy/internal/config" +) + +// TestEndToEndTemplatedUpstreamRoutesByRenderedAddress drives the full stack +// with a single templated upstream whose hostPort renders from request +// metadata, and proves each request reaches the upstream its metadata names. +// This exercises per-request address rendering end to end, and because two +// distinct rendered targets are both reached, it also proves the connection +// pool keys by rendered target rather than collapsing onto the first dial. +func TestEndToEndTemplatedUpstreamRoutesByRenderedAddress(t *testing.T) { + t.Parallel() + + svcA, addrA := newFakeUpstream(t) + svcB, addrB := newFakeUpstream(t) + + inboundAddr := freeTCPAddr(t) + app := newFullApp(t, &config.Config{ + Listen: config.ListenConfig{HostPort: inboundAddr}, + Routing: config.Routing{DefaultUpstream: "dynamic"}, + Upstreams: []config.Upstream{{ + Name: "dynamic", + Listen: config.ListenConfig{HostPort: `{{ index .Metadata "x-upstream" }}`}, + }}, + }) + require.NoError(t, app.Err()) + + startCtx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, app.Start(startCtx)) + t.Cleanup(func() { + stopCtx, stopCancel := context.WithTimeout(t.Context(), 5*time.Second) + defer stopCancel() + _ = app.Stop(stopCtx) + }) + + conn := dialInbound(t, inboundAddr) + defer func() { _ = conn.Close() }() + client := workflowservice.NewWorkflowServiceClient(conn) + + call := func(target string) { + ctx := metadata.AppendToOutgoingContext(startCtx, "x-upstream", target) + _, err := client.GetSystemInfo(ctx, &workflowservice.GetSystemInfoRequest{}, grpc.WaitForReady(true)) + require.NoError(t, err) + } + + call(addrA) + call(addrB) + + require.NotNil(t, svcA.received(), "the request naming upstream A must reach A") + require.NotNil(t, svcB.received(), "the request naming upstream B must reach B") + require.Equal(t, []string{addrA}, svcA.received().Get("x-upstream")) + require.Equal(t, []string{addrB}, svcB.received().Get("x-upstream")) +} + +// newFakeUpstream stands up a plaintext fake WorkflowService frontend and +// returns it with its dial address. +func newFakeUpstream(t *testing.T) (*capturingWorkflowService, string) { + t.Helper() + + svc := &capturingWorkflowService{} + srv := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(srv, svc) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + return svc, lis.Addr().String() +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 59527a9..4f25529 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -359,6 +359,18 @@ func TestUpstream_IsTemplated(t *testing.T) { require.True(t, (&config.Upstream{Listen: config.ListenConfig{HostPort: "{{ .LocalNamespace }}.acme.cloud:7233"}}).IsTemplated()) require.False(t, (&config.Upstream{Listen: config.ListenConfig{HostPort: "127.0.0.1:7233"}}).IsTemplated()) + + // A templated TLS server name makes the upstream templated even when the + // hostPort is static. + require.True(t, (&config.Upstream{Listen: config.ListenConfig{ + HostPort: "127.0.0.1:7233", + TLS: &config.TLSConfig{ServerName: "{{ .RemoteNamespace }}.acme.cloud"}, + }}).IsTemplated()) + + require.False(t, (&config.Upstream{Listen: config.ListenConfig{ + HostPort: "127.0.0.1:7233", + TLS: &config.TLSConfig{ServerName: "static.acme.cloud"}, + }}).IsTemplated()) } func (e *errReader) Read(_ []byte) (int, error) { return 0, e.err } diff --git a/internal/config/upstream.go b/internal/config/upstream.go index abe1233..4c395a0 100644 --- a/internal/config/upstream.go +++ b/internal/config/upstream.go @@ -79,10 +79,15 @@ func (u *Upstream) Validate() error { ) } -// IsTemplated reports whether the upstream's hostPort contains a text/template -// action and so must be resolved per request. +// 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. func (u *Upstream) IsTemplated() bool { - return isTemplated(u.Listen.HostPort) + if isTemplated(u.Listen.HostPort) { + return true + } + + return u.Listen.TLS != nil && isTemplated(u.Listen.TLS.ServerName) } // Validate checks the namespace translation rules. diff --git a/internal/proxy/fx.go b/internal/proxy/fx.go index 02a588f..f8b9669 100644 --- a/internal/proxy/fx.go +++ b/internal/proxy/fx.go @@ -3,12 +3,15 @@ package proxy import ( "context" "fmt" + "slices" "go.uber.org/fx" + "google.golang.org/grpc" "github.com/temporalio/temporal-proxy/internal/auth" "github.com/temporalio/temporal-proxy/internal/config" "github.com/temporalio/temporal-proxy/internal/protoutil" + "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/internal/transport/creds" "github.com/temporalio/temporal-proxy/pkg/logger" ) @@ -17,48 +20,61 @@ import ( // and binds its lifecycle to the application. var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { for i := range p.Config.Upstreams { - upstream := &p.Config.Upstreams[i] - - if err := upstream.Validate(); err != nil { + up := &p.Config.Upstreams[i] + if err := up.Validate(); err != nil { return fmt.Errorf("invalid upstream configuration: %w", err) } - if upstream.IsTemplated() { - return fmt.Errorf( - "upstream %q has a templated hostPort %q: templated upstreams are not yet supported", - upstream.Name, - upstream.Listen.HostPort, - ) - } - - opts := []Option{WithCredentials(upstreamCreds(upstream))} - - rules := &upstream.Namespaces.Rules + // Request-independent dial options: namespace translation and outbound + // credentials. Per-request credentials are added by the resolver. + var dialOpts []grpc.DialOption + rules := &up.Namespaces.Rules if rules.Configured() { - opts = append(opts, WithDialOptions(translationDialOptions(p.Translator, rules.Remote, rules.Local)...)) + dialOpts = append(dialOpts, translationDialOptions(p.Translator, rules.Remote, rules.Local)...) } - cp, err := auth.CredentialProviderFor(upstream.Credentials) + cp, err := auth.CredentialProviderFor(up.Credentials) if err != nil { - return fmt.Errorf("invalid credentials for upstream %q: %w", upstream.Name, err) + return fmt.Errorf("invalid credentials for upstream %q: %w", up.Name, err) } if cp != nil { - opts = append(opts, WithDialOptions(auth.DialOptions(cp)...)) + dialOpts = append(dialOpts, auth.DialOptions(cp)...) + } + + res, err := upstreamResolver(up, dialOpts) + if err != nil { + return err } + conn, err := connect.NewConn(p.Pool.ConnOrCreate, res) + if err != nil { + return err + } + + var opts []Option if p.Logger != nil { opts = append(opts, WithLogger(p.Logger)) } - svr, err := New(upstream.Listen.HostPort, opts...) + svr, err := New(up.Listen.HostPort, conn, opts...) if err != nil { - return fmt.Errorf("failed to create proxy for upstream %q: %w", upstream.Name, err) + return fmt.Errorf("failed to create proxy for upstream %q: %w", up.Name, err) } p.Lifecycle.Append(fx.Hook{ OnStart: func(context.Context) error { + // Bind synchronously so the socket is listening before the + // inbound server (whose OnStart runs after this one) starts + // routing requests to it; then serve in the background. + lis, err := svr.Listen(p.Context) + if err != nil { + return fmt.Errorf("failed to start proxy for upstream %q: %w", up.Name, err) + } + go func() { - if err := svr.Start(p.Context); err != nil { + defer func() { _ = lis.Close() }() + + if err := svr.Start(p.Context, lis); err != nil { // The proxy stopped serving unexpectedly. Bring the app // down rather than linger in a non-serving state; Start // has already logged the cause. @@ -76,9 +92,10 @@ var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { })) // ProxyParams collects the fx-provided dependencies needed to construct and run -// the proxy [Server]. Context, Config, and Translator are required; Logger is -// optional and falls back to the default used by [New] when not supplied. -// [protoutil.Module] provides the Translator in the assembled application. +// the proxy [Server]. Context, Config, Translator, and Pool are required; +// Logger is optional and falls back to the default used by [New] when not +// supplied. [protoutil.Module] provides the Translator and [connect.Module] +// provides the Pool in the assembled application. type ProxyParams struct { fx.In Lifecycle fx.Lifecycle @@ -88,28 +105,83 @@ type ProxyParams struct { Context context.Context Config *config.Config Translator *protoutil.Translator + Pool *connect.Pool // Optional values Logger logger.Logger `optional:"true"` } +// upstreamResolver builds the [connect.Resolver] for an upstream. When neither +// the hostPort nor the TLS server name is templated it returns a static +// resolver whose connection is constructed eagerly (and reused for every +// request); otherwise it returns a DynamicResolver that renders the target and +// server name, and rebuilds credentials, per request. opts holds the +// request-independent dial options (namespace translation and outbound +// credentials). +func upstreamResolver(upstream *config.Upstream, opts []grpc.DialOption) (connect.Resolver, error) { + if upstream.IsTemplated() { + translator := func(s string) string { return s } + if upstream.Namespaces.Rules.Configured() { + translator = upstream.Namespaces.Rules.Remote + } + + // Share one loader across the per-request credentials so templated mTLS + // routing reads and parses the certificate and CA files once rather than + // on every request. Only ServerName varies per request. nil for non-mTLS + // upstreams, which have no files to load. + var loader *creds.CertLoader + if tls := upstream.Listen.TLS; tls != nil && tls.CA != "" { + loader = creds.NewCertLoader(tls.CA, tls.Cert, tls.Key) + } + + return NewDynamicResolver( + upstream, + WithRemoteNamespacer(translator), + WithOptionsFactory(func(data RouteData) ([]grpc.DialOption, error) { + creds, err := upstreamCreds(upstream, data.ResolvedServerName, loader).DialOption() + if err != nil { + return nil, err + } + + return append(slices.Clone(opts), creds), nil + }), + ) + } + + serverName := "" + if upstream.Listen.TLS != nil { + serverName = upstream.Listen.TLS.ServerName + } + + creds, err := upstreamCreds(upstream, serverName, nil).DialOption() + if err != nil { + return nil, fmt.Errorf("failed to build credentials for upstream %q: %w", upstream.Name, err) + } + + return connect.StaticResolver(upstream.Listen.HostPort, append(slices.Clone(opts), creds)...), nil +} + // upstreamCreds derives the credentials used to dial the upstream frontend from // the upstream TLS configuration: mutual TLS when a CA is set, server-verified // client TLS when TLS is configured without one, and insecure otherwise. -func upstreamCreds(upstream *config.Upstream) Credentials { +// serverName overrides the SNI/certificate-verification name; it is the +// upstream's static configured ServerName, or a per-request value rendered +// from a templated ServerName. loader, when non-nil, supplies the mTLS +// certificate material so it is loaded once and reused across the per-request +// credentials of a templated upstream; pass nil for a fixed-address upstream, +// whose credentials are built once. +func upstreamCreds(upstream *config.Upstream, serverName string, loader *creds.CertLoader) Credentials { tls := upstream.Listen.TLS if tls == nil { return creds.NewInsecure() } if tls.CA != "" { - return creds.NewMTLS( - tls.CA, - tls.Cert, - tls.Key, - creds.MTLSOptions{ServerName: tls.ServerName}, - ) + return creds.NewMTLS(tls.CA, tls.Cert, tls.Key, creds.MTLSOptions{ + ServerName: serverName, + Loader: loader, + }) } - return creds.NewClientTLS() + return creds.NewClientTLS(serverName) } diff --git a/internal/proxy/fx_test.go b/internal/proxy/fx_test.go index 6ce3006..2edc5e6 100644 --- a/internal/proxy/fx_test.go +++ b/internal/proxy/fx_test.go @@ -16,6 +16,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/config" "github.com/temporalio/temporal-proxy/internal/protoutil" "github.com/temporalio/temporal-proxy/internal/proxy" + "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/pkg/logger" "github.com/temporalio/temporal-proxy/pkg/testutil" "github.com/temporalio/temporal-proxy/pkg/validation" @@ -155,7 +156,7 @@ func TestModuleWithUpstreamCredentials(t *testing.T) { require.NoError(t, app.Err()) } -func TestModuleRejectsTemplatedUpstream(t *testing.T) { +func TestModuleAcceptsTemplatedUpstream(t *testing.T) { t.Parallel() app := newProxyApp(t, &config.Config{ @@ -164,8 +165,19 @@ func TestModuleRejectsTemplatedUpstream(t *testing.T) { }, }) - require.Error(t, app.Err()) - require.ErrorContains(t, app.Err(), "templated") + // Previously this failed with "templated upstreams are not yet supported". + // A templated hostPort is now resolved per-request via a templatedPlan, so + // construction (and the full start/stop lifecycle) succeeds without ever + // dialing the upstream. + require.NoError(t, app.Err()) + + startCtx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, app.Start(startCtx)) + + stopCtx, stopCancel := context.WithTimeout(t.Context(), 5*time.Second) + defer stopCancel() + require.NoError(t, app.Stop(stopCtx)) } func TestModuleInstallsNamespaceTranslation(t *testing.T) { @@ -190,14 +202,12 @@ func TestModuleInstallsNamespaceTranslation(t *testing.T) { func TestModuleRequiresTranslator(t *testing.T) { t.Parallel() - // proxy.Module requires a *protoutil.Translator. Without protoutil.Module - // (or another provider) the app must fail to build rather than run without - // one. app := fx.New( fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(&config.Config{ Upstreams: []config.Upstream{{Name: "primary", Listen: config.ListenConfig{HostPort: "127.0.0.1:47242"}}}, }), + connect.Module, proxy.Module, fx.NopLogger, ) @@ -211,6 +221,7 @@ func newProxyApp(t *testing.T, cfg *config.Config, opts ...fx.Option) *fx.App { base := []fx.Option{ fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), + connect.Module, protoutil.Module, proxy.Module, fx.NopLogger, diff --git a/internal/proxy/resolver.go b/internal/proxy/resolver.go new file mode 100644 index 0000000..64c9cc3 --- /dev/null +++ b/internal/proxy/resolver.go @@ -0,0 +1,210 @@ +package proxy + +import ( + "context" + "fmt" + "net" + "slices" + "strings" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/template" + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +type ( + // DynamicResolver is a [connect.Resolver] that renders an upstream's dial + // target (and optional TLS server name) per request from the local + // namespace and request metadata. It always reports IsStatic as false, so a + // [connect.Conn] built from it resolves lazily on every call. A non-templated + // hostPort renders to itself, so a DynamicResolver also serves upstreams with + // a fixed address. Construct one with NewDynamicResolver. + DynamicResolver struct { + name string + host *template.Template[template.UpstreamContext] + serverName *template.Template[template.UpstreamContext] + remote func(string) string + opts func(d RouteData) ([]grpc.DialOption, error) + } + + // RouteData is passed to the options factory once a request has been + // resolved. It carries the template context used for rendering plus the + // resolved TLS server name, so the factory can build dial options (e.g. + // credentials whose SNI depends on the rendered server name). + RouteData struct { + template.UpstreamContext + ResolvedServerName string + } + + // ResolverOption configures a DynamicResolver at construction. + ResolverOption func(*DynamicResolver) +) + +// NewDynamicResolver builds a DynamicResolver for up. It compiles the hostPort +// and TLS server-name templates (failing if either is malformed) and applies +// opts. By default the remote namespace equals the local one and no dial options +// are added; use WithRemoteNamespacer and WithOptionsFactory to change that. +func NewDynamicResolver(up *config.Upstream, opts ...ResolverOption) (*DynamicResolver, error) { + host, err := template.ParseUpstream(up.Listen.HostPort) + if err != nil { + return nil, fmt.Errorf("failed to parse upstream host template %q: %w", up.Name, err) + } + + tsn := "" + if up.Listen.TLS != nil { + tsn = up.Listen.TLS.ServerName + } + + serverName, err := template.ParseUpstream(tsn) + if err != nil { + return nil, fmt.Errorf("failed to parse upstream TLS server name template %q: %w", up.Name, err) + } + + r := &DynamicResolver{ + name: up.Name, + host: host, + serverName: serverName, + remote: func(s string) string { return s }, + opts: func(RouteData) ([]grpc.DialOption, error) { return nil, nil }, + } + + for _, opt := range opts { + opt(r) + } + + return r, nil +} + +// WithRemoteNamespacer sets the function that maps the local namespace to the +// remote one, making RemoteNamespace available to the templates. +func WithRemoteNamespacer(f func(string) string) ResolverOption { + return func(r *DynamicResolver) { r.remote = f } +} + +// WithOptionsFactory sets the function that produces the dial options for a +// resolved request. It receives the rendered host and server name via RouteData. +func WithOptionsFactory(f func(RouteData) ([]grpc.DialOption, error)) ResolverOption { + return func(r *DynamicResolver) { r.opts = f } +} + +// IsStatic reports that a DynamicResolver always resolves per request. +func (r *DynamicResolver) IsStatic() bool { + return false +} + +// Resolve renders the dial target and server name from ctx and returns the pool +// cache key, the dial target, and the dial options. The cache key combines the +// target and rendered server name so that two requests to the same address with +// different server names get distinct pooled connections. It fails with +// codes.Internal (naming the upstream and template) when a template fails to +// render, the rendered address is empty or malformed, or the options factory +// errors; nothing is dialed in those cases. +func (r *DynamicResolver) Resolve(ctx context.Context) (string, string, []grpc.DialOption, error) { + localNS := meta.NamespaceFrom(ctx) + data := template.UpstreamContext{ + LocalNamespace: localNS, + RemoteNamespace: r.remote(localNS), + Metadata: lastMetadataValues(ctx), + } + + target, err := r.host.Render(data) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to render hostPort %q: %v", + r.name, + r.host.String(), + err, + ) + } + + if invalidAddress(target) { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q rendered invalid hostPort %q from template %q", + r.name, + target, + r.host.String(), + ) + } + + svrName, err := r.serverName.Render(data) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to render serverName %q: %v", + r.name, + r.serverName.String(), + err, + ) + } + + opts, err := r.opts(RouteData{ + UpstreamContext: data, + ResolvedServerName: svrName, + }) + if err != nil { + return "", "", nil, status.Errorf( + codes.Internal, + "proxy: upstream %q failed to build dial options: %v", + r.name, + err, + ) + } + + return target + "\x00" + svrName, target, opts, nil +} + +// lastMetadataValues flattens the request's outgoing metadata to one value per +// key, keeping the last (most recently added) value. gRPC lowercases metadata +// keys, so template lookups must use lowercase keys. +func lastMetadataValues(ctx context.Context) map[string]string { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + return nil + } + + out := make(map[string]string, len(md)) + for k, v := range md { + if len(v) > 0 { + out[k] = v[len(v)-1] + } + } + + return out +} + +// invalidAddress reports whether addr is unusable as a dial target: empty, or +// missing a host label (e.g. ":7233" or ".acme:7233" produced when a referenced +// namespace or metadata value renders empty). It strips a leading gRPC scheme +// ("dns:///", "passthrough:///") and tolerates a single trailing dot so a +// fully-qualified name ("acme.example.:7233") is accepted. +func invalidAddress(addr string) bool { + if strings.TrimSpace(addr) == "" { + return true + } + + hp := addr + if i := strings.Index(hp, "://"); i >= 0 { + hp = hp[i+3:] + hp = strings.TrimPrefix(hp, "/") // handle the "dns:///" triple-slash form + } + + host, _, err := net.SplitHostPort(hp) + if err != nil { + host = hp // no port present; treat the whole remainder as the host + } + + host = strings.TrimSpace(host) + if host == "" { + return true + } + + host = strings.TrimSuffix(host, ".") + return slices.Contains(strings.Split(host, "."), "") +} diff --git a/internal/proxy/resolver_test.go b/internal/proxy/resolver_test.go new file mode 100644 index 0000000..f53b13c --- /dev/null +++ b/internal/proxy/resolver_test.go @@ -0,0 +1,149 @@ +package proxy_test + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/proxy" + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +func TestDynamicResolverIsStatic(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("u", "host:7233", "")) + require.NoError(t, err) + require.False(t, res.IsStatic()) +} + +func TestNewDynamicResolverRejectsBadTemplates(t *testing.T) { + t.Parallel() + + _, err := proxy.NewDynamicResolver(upstreamWith("u", "{{ .Nope }}:7233", "")) + require.Error(t, err) + require.ErrorContains(t, err, "host template") + + _, err = proxy.NewDynamicResolver(upstreamWith("u", "host:7233", "{{ .Nope }}")) + require.Error(t, err) + require.ErrorContains(t, err, "server name template") +} + +func TestDynamicResolverResolve(t *testing.T) { + t.Parallel() + + up := upstreamWith("cloud", "{{ .RemoteNamespace }}.acme.example:7233", "{{ .RemoteNamespace }}.sni.example") + + var got proxy.RouteData + res, err := proxy.NewDynamicResolver( + up, + proxy.WithRemoteNamespacer(func(s string) string { return s + "-remote" }), + proxy.WithOptionsFactory(func(d proxy.RouteData) ([]grpc.DialOption, error) { + got = d + return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil + }), + ) + require.NoError(t, err) + + key, target, opts, err := res.Resolve(meta.WithNamespace(t.Context(), "orders")) + require.NoError(t, err) + + require.Equal(t, "orders-remote.acme.example:7233", target) + require.Equal(t, "orders-remote.acme.example:7233\x00orders-remote.sni.example", key, + "cache key combines target and rendered server name") + require.Len(t, opts, 1) + + // The options factory sees the rendered server name and namespaces. + require.Equal(t, "orders-remote.sni.example", got.ResolvedServerName) + require.Equal(t, "orders", got.LocalNamespace) + require.Equal(t, "orders-remote", got.RemoteNamespace) +} + +func TestDynamicResolverUsesLastMetadataValue(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("m", `{{ index .Metadata "dc" }}.acme:7233`, "")) + require.NoError(t, err) + + md := metadata.MD{} + md.Append("dc", "old", "new") + ctx := metadata.NewOutgoingContext(t.Context(), md) + + _, target, _, err := res.Resolve(ctx) + require.NoError(t, err) + require.Equal(t, "new.acme:7233", target, "the last metadata value wins") +} + +func TestDynamicResolverFailsLoudOnOptionsError(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver( + upstreamWith("u", "host:7233", ""), + proxy.WithOptionsFactory(func(proxy.RouteData) ([]grpc.DialOption, error) { + return nil, errors.New("boom") + }), + ) + require.NoError(t, err) + + _, _, _, err = res.Resolve(meta.WithNamespace(t.Context(), "orders")) + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + require.ErrorContains(t, err, "failed to build dial options") +} + +func TestDynamicResolverRejectsInvalidAddresses(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + host string + wantErr bool + }{ + {"empty metadata leaves an empty label", `{{ index .Metadata "dc" }}.acme:7233`, true}, + {"leading dot", ".acme:7233", true}, + {"interior double dot", "acme..example:7233", true}, + {"missing host", ":7233", true}, + {"normal host", "orders.acme.example:7233", false}, + {"trailing-dot fqdn", "acme.example.:7233", false}, + {"ipv6 literal", "[::1]:7233", false}, + {"scheme prefixed", "dns:///host:7233", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + res, err := proxy.NewDynamicResolver(upstreamWith("u", tt.host, "")) + require.NoError(t, err) + + _, _, _, err = res.Resolve(meta.WithNamespace(t.Context(), "orders")) + if tt.wantErr { + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + return + } + + require.NoError(t, err) + }) + } +} + +func upstreamWith(name, hostPort, serverName string) *config.Upstream { + up := &config.Upstream{ + Name: name, + Listen: config.ListenConfig{HostPort: hostPort}, + } + + if serverName != "" { + up.Listen.TLS = &config.TLSConfig{ServerName: serverName} + } + + return up +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 3d678f0..b8e11c4 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -24,56 +24,35 @@ type ( DialOption() (grpc.DialOption, error) } - // Server proxies the Temporal WorkflowService. It dials an upstream frontend - // and re-serves it on a local unix socket, letting local workers connect - // without TLS while the upstream hop stays secured. + // Server proxies the Temporal WorkflowService. It re-serves an upstream + // frontend on a local unix socket, letting local workers connect without TLS + // while the upstream hop stays secured. The upstream connection(s) it + // forwards to are owned by the shared [connect.Pool], not by this Server. Server struct { svr *server.Server - conn *grpc.ClientConn - - host string // upstream hostPort path string // path to unix socket } // Options configures a [Server] at construction time. Options struct { - creds Credentials - logger logger.Logger - dialOpts []grpc.DialOption + logger logger.Logger } // Option configures a [Server] via [New]. Option func(*Options) ) -// New constructs a [Server] that forwards WorkflowService traffic to the -// upstream frontend at hostPort. The local listener is a unix socket whose path -// is derived from hostPort (see [Server.Start]). With no options it dials the -// upstream with insecure credentials and logs via a CLI logger. -func New(hostPort string, opts ...Option) (*Server, error) { - pops := &Options{ - creds: creds.NewInsecure(), - logger: logger.Default(), - } +// New constructs a Server that forwards WorkflowService traffic to the +// upstream reachable through cc. The local listener is a unix socket whose +// path is derived from hostPort. cc is typically a resolvingConn; the +// connection(s) it uses are owned by the shared pool, not by this Server. +func New(hostPort string, cc grpc.ClientConnInterface, opts ...Option) (*Server, error) { + pops := &Options{logger: logger.Default()} for _, opt := range opts { opt(pops) } - upstreamCreds, err := pops.creds.DialOption() - if err != nil { - return nil, fmt.Errorf("failed to generate outbound credentials: %w", err) - } - - dialOpts := append([]grpc.DialOption{upstreamCreds}, pops.dialOpts...) - - conn, err := grpc.NewClient(hostPort, dialOpts...) - if err != nil { - return nil, fmt.Errorf("failed to dial: %s, %w", hostPort, err) - } - - wfs, err := proxy.NewWorkflowServiceProxyServer(proxy.WorkflowServiceProxyOptions{ - Client: workflowservice.NewWorkflowServiceClient(conn), - }) + wfs, err := proxy.NewWorkflowServiceProxyServer(workflowProxyOptions(cc)) if err != nil { return nil, fmt.Errorf("failed to create workflowservice proxy: %w", err) } @@ -95,18 +74,7 @@ func New(hostPort string, opts ...Option) (*Server, error) { return nil, fmt.Errorf("failed to resolve socket path: %w", err) } - return &Server{ - svr: svr, - conn: conn, - host: hostPort, - path: path, - }, nil -} - -// WithCredentials sets the transport credentials used to dial the upstream -// frontend. -func WithCredentials(creds Credentials) Option { - return Option(func(o *Options) { o.creds = creds }) + return &Server{svr: svr, path: path}, nil } // WithLogger sets the logger used by the proxy. @@ -114,28 +82,30 @@ func WithLogger(log logger.Logger) Option { return Option(func(o *Options) { o.logger = log }) } -// WithDialOptions appends gRPC dial options applied to the outbound connection -// to the upstream frontend, in addition to the transport credentials. Outbound -// credentials and namespace translation are both wired this way. -func WithDialOptions(opts ...grpc.DialOption) Option { - return Option(func(o *Options) { o.dialOpts = append(o.dialOpts, opts...) }) -} - -// Start binds the local unix socket and serves until the proxy is stopped or -// ctx is cancelled. It first removes any socket left behind by a prior run. It -// blocks, so callers typically run it in its own goroutine. -func (s *Server) Start(ctx context.Context) error { +// Listen removes any socket left behind by a prior run and binds the proxy's +// local unix socket, returning the listener. Binding is separate from Start so +// callers can bind synchronously during startup (the socket is then listening, +// and the OS backlogs connections) before serving in the background, ensuring +// no request is routed to an unbound socket. +func (s *Server) Listen(ctx context.Context) (net.Listener, error) { // Remove any socket left behind by a prior run; otherwise the bind fails // with "address already in use". if err := os.Remove(s.path); err != nil && !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("failed to remove stale socket: unix://%s, %w", s.path, err) + return nil, fmt.Errorf("failed to remove stale socket: unix://%s, %w", s.path, err) } lis, err := (&net.ListenConfig{}).Listen(ctx, "unix", s.path) if err != nil { - return fmt.Errorf("failed to bind to socket: unix://%s, %w", s.path, err) + return nil, fmt.Errorf("failed to bind to socket: unix://%s, %w", s.path, err) } + return lis, nil +} + +// Start serves on lis until the proxy is stopped or ctx is cancelled. It +// blocks, so callers typically run it in its own goroutine after binding the +// listener with Listen. +func (s *Server) Start(ctx context.Context, lis net.Listener) error { return s.svr.Start(ctx, lis) } @@ -145,5 +115,14 @@ func (s *Server) Stop(ctx context.Context) error { return fmt.Errorf("failed to stop GRPC server: %w", err) } - return s.conn.Close() + return nil +} + +// workflowProxyOptions builds the options for the WorkflowService proxy +// server backed by cc. DisableHeaderForwarding is intentionally left false: +// the upstream proxy must forward incoming metadata (including the +// router-stamped namespace) onto the outbound call, since templated upstream +// resolution and namespace translation both depend on it. +func workflowProxyOptions(cc grpc.ClientConnInterface) proxy.WorkflowServiceProxyOptions { + return proxy.WorkflowServiceProxyOptions{Client: workflowservice.NewWorkflowServiceClient(cc)} } diff --git a/internal/proxy/server_test.go b/internal/proxy/server_test.go index 74c6694..e8fbdd5 100644 --- a/internal/proxy/server_test.go +++ b/internal/proxy/server_test.go @@ -2,7 +2,6 @@ package proxy_test import ( "context" - "errors" "os" "path/filepath" "testing" @@ -13,55 +12,21 @@ import ( "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/health/grpc_health_v1" - "github.com/temporalio/temporal-proxy/internal/auth" "github.com/temporalio/temporal-proxy/internal/proxy" - "github.com/temporalio/temporal-proxy/internal/transport/creds" "github.com/temporalio/temporal-proxy/internal/transport/socket" "github.com/temporalio/temporal-proxy/pkg/logger" ) -type failingCredentials struct { - err error -} - func TestNew(t *testing.T) { t.Parallel() t.Run("returns a server with default options", func(t *testing.T) { t.Parallel() - svr, err := proxy.New("127.0.0.1:7233") + svr, err := proxy.New("127.0.0.1:7233", upstreamConn(t, "127.0.0.1:7233")) require.NoError(t, err) require.NotNil(t, svr) }) - - t.Run("propagates credential errors", func(t *testing.T) { - t.Parallel() - - svr, err := proxy.New( - "127.0.0.1:7233", - proxy.WithCredentials(failingCredentials{err: errors.New("boom")}), - ) - require.Error(t, err) - require.Nil(t, svr) - require.ErrorContains(t, err, "outbound credentials") - require.ErrorContains(t, err, "boom") - }) -} - -func TestNewWithDialOptions(t *testing.T) { - t.Parallel() - - cp, err := auth.NewStaticCredentialProvider("k", "", "") - require.NoError(t, err) - - svr, err := proxy.New( - "127.0.0.1:7233", - proxy.WithCredentials(creds.NewClientTLS()), - proxy.WithDialOptions(auth.DialOptions(cp)...), - ) - require.NoError(t, err) - require.NotNil(t, svr) } func TestServerStartAndStop(t *testing.T) { @@ -73,14 +38,17 @@ func TestServerStartAndStop(t *testing.T) { const upstream = "127.0.0.1:17233" log := logger.NewTestLogger() - svr, err := proxy.New(upstream, proxy.WithLogger(log)) + svr, err := proxy.New(upstream, upstreamConn(t, upstream), proxy.WithLogger(log)) require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) defer cancel() + lis, err := svr.Listen(ctx) + require.NoError(t, err) + errCh := make(chan error, 1) - go func() { errCh <- svr.Start(ctx) }() + go func() { errCh <- svr.Start(ctx, lis) }() conn := dialUnix(t, upstream) defer func() { _ = conn.Close() }() @@ -119,14 +87,17 @@ func TestStartRemovesStaleSocket(t *testing.T) { require.NoError(t, os.WriteFile(path, []byte("stale"), 0o600)) t.Cleanup(func() { _ = os.Remove(path) }) - svr, err := proxy.New(upstream) + svr, err := proxy.New(upstream, upstreamConn(t, upstream)) require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) defer cancel() + lis, err := svr.Listen(ctx) + require.NoError(t, err) + errCh := make(chan error, 1) - go func() { errCh <- svr.Start(ctx) }() + go func() { errCh <- svr.Start(ctx, lis) }() conn := dialUnix(t, upstream) defer func() { _ = conn.Close() }() @@ -142,7 +113,7 @@ func TestStartRemovesStaleSocket(t *testing.T) { require.NoError(t, <-errCh) } -func TestStartReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { +func TestListenReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { t.Parallel() const upstream = "127.0.0.1:37233" @@ -150,26 +121,37 @@ func TestStartReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { path, err := socket.UnixPath(upstream) require.NoError(t, err) - // A non-empty directory at the socket path makes os.Remove fail, so Start + // A non-empty directory at the socket path makes os.Remove fail, so Listen // returns before it ever binds. require.NoError(t, os.Mkdir(path, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(path, "child"), nil, 0o600)) t.Cleanup(func() { _ = os.RemoveAll(path) }) - svr, err := proxy.New(upstream) + svr, err := proxy.New(upstream, upstreamConn(t, upstream)) require.NoError(t, err) - err = svr.Start(t.Context()) + _, err = svr.Listen(t.Context()) require.Error(t, err) require.ErrorContains(t, err, "failed to remove stale socket") } -func (f failingCredentials) DialOption() (grpc.DialOption, error) { - return nil, f.err +// upstreamConn returns a grpc.ClientConnInterface for New's cc argument. gRPC +// dials lazily, so this never opens a socket to upstream; the tests in this +// file never make an outbound RPC through it (they only exercise the local +// unix listener), so a plain client conn stands in for the pool-backed +// resolvingConn used in production. +func upstreamConn(t *testing.T, upstream string) grpc.ClientConnInterface { + t.Helper() + + conn, err := grpc.NewClient(upstream, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn } // dialUnix returns a client connection to the proxy's unix socket for the given -// upstream host. The socket path matches what proxy.Start binds. +// upstream host. The socket path matches what proxy.Listen binds. func dialUnix(t *testing.T, upstream string) *grpc.ClientConn { t.Helper() diff --git a/internal/proxy/translation_test.go b/internal/proxy/translation_test.go index f7ff37c..b21fc51 100644 --- a/internal/proxy/translation_test.go +++ b/internal/proxy/translation_test.go @@ -37,6 +37,15 @@ type ( } ) +func TestWorkflowProxyOptionsForwardsHeaders(t *testing.T) { + t.Parallel() + + // Guard: New must not disable header forwarding, or the upstream proxy would + // drop the router-stamped namespace header and templated resolution would + // break. + require.False(t, workflowProxyOptions(nil).DisableHeaderForwarding) +} + func TestUnaryClientInterceptorTranslatesBothDirections(t *testing.T) { t.Parallel() @@ -132,20 +141,29 @@ func TestOutboundNamespaceTranslation(t *testing.T) { go func() { _ = upstream.Serve(lis) }() t.Cleanup(upstream.Stop) - // Proxy dials the fake upstream, wrapping "local" -> "remote-local". + // Dial the fake upstream directly, wrapping "local" -> "remote-local"; this + // mirrors how fx folds translationDialOptions into a plan's dial options + // before the pool dials them. translator := protoutil.NewTranslator(protoregistry.GlobalFiles) - svr, err := New( - lis.Addr().String(), - WithDialOptions(translationDialOptions( + dialOpts := append( + []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, + translationDialOptions( translator, func(s string) string { return "remote-" + s }, func(s string) string { return "local-" + s }, - )...), + )..., ) + cc, err := grpc.NewClient(lis.Addr().String(), dialOpts...) + require.NoError(t, err) + t.Cleanup(func() { _ = cc.Close() }) + + svr, err := New(lis.Addr().String(), cc) require.NoError(t, err) ctx := t.Context() - go func() { _ = svr.Start(ctx) }() + srvLis, err := svr.Listen(ctx) + require.NoError(t, err) + go func() { _ = svr.Start(ctx, srvLis) }() t.Cleanup(func() { _ = svr.Stop(context.Background()) }) conn := dialUnixSocket(t, lis.Addr().String()) diff --git a/internal/router/fx.go b/internal/router/fx.go index 8c4897d..5a3e759 100644 --- a/internal/router/fx.go +++ b/internal/router/fx.go @@ -37,10 +37,8 @@ var Module = fx.Options(fx.Provide( return nil, fmt.Errorf("failed to resolve proxy socket path[%q]: %w", upstream.Name, err) } - conn, err := p.Pool.GetOrSet( - "unix://"+sockPath, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) + sock := "unix://" + sockPath + conn, err := p.Pool.ConnOrCreate(sock, sock, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { return nil, fmt.Errorf("failed to create upstream client[%q]: %w", upstream.Name, err) } diff --git a/internal/router/handler.go b/internal/router/handler.go index 2e0acef..604ae32 100644 --- a/internal/router/handler.go +++ b/internal/router/handler.go @@ -9,6 +9,8 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" + + "github.com/temporalio/temporal-proxy/internal/transport/meta" ) type ( @@ -76,6 +78,11 @@ func Handler(d Director, r Reflector, rep *Reporter) grpc.StreamHandler { namespace = r.Namespace(method, first.payload) } + // Carry the extracted namespace to the upstream proxy so it can resolve a + // templated address without re-parsing the payload. Set (not append) so a + // client-supplied value cannot influence routing. + outCtx = meta.WithNamespace(outCtx, namespace) + target, err := d.Resolve(ctx, method, namespace, maps.Clone(md)) if err != nil { return err diff --git a/internal/router/handler_test.go b/internal/router/handler_test.go index 5d8563d..00f4080 100644 --- a/internal/router/handler_test.go +++ b/internal/router/handler_test.go @@ -9,6 +9,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -27,6 +28,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/internal/router" "github.com/temporalio/temporal-proxy/internal/server" + "github.com/temporalio/temporal-proxy/internal/transport/meta" ) type ( @@ -448,6 +450,34 @@ tmprl_proxy_router_forwarding_errors_total{reason="stream_setup",upstream="prima `), "tmprl_proxy_router_forwarding_errors_total")) } +func TestHandlerStampsNamespace(t *testing.T) { + t.Parallel() + + upstream, got := capturingUpstream(t) + + m, _ := newTestReporter(t) + relayLis := bufconn.Listen(1024 * 1024) + relay := grpc.NewServer( + grpc.ForceServerCodecV2(router.Codec()), + grpc.UnknownServiceHandler(router.Handler(stubDirector{upstream: "u", cc: upstream}, &recordingReflector{ns: "orders"}, m)), + ) + serve(t, relay, relayLis) + + client := dialBufconn(t, relayLis) + + // The client sends a spoofed namespace header; the router must overwrite it + // with the value it extracted via the Reflector before forwarding. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(meta.NamespaceHeader, "spoofed")) + _ = client.Invoke(ctx, "/test.v1.Echo/Ping", &grpc_health_v1.HealthCheckRequest{}, new(grpc_health_v1.HealthCheckResponse)) + + select { + case ns := <-got: + require.Equal(t, "orders", ns) + case <-time.After(2 * time.Second): + t.Fatal("upstream never received a request") + } +} + func TestStatusError(t *testing.T) { t.Parallel() @@ -514,6 +544,28 @@ func (d *recordingDirector) snapshot() (calls int, method, namespace string, md return d.calls, d.method, d.namespace, d.md } +// capturingUpstream is an in-process gRPC server that records the namespace +// header from the first stream it receives. +func capturingUpstream(t *testing.T) (*grpc.ClientConn, <-chan string) { + t.Helper() + + got := make(chan string, 1) + lis := bufconn.Listen(1024 * 1024) + srv := grpc.NewServer(grpc.UnknownServiceHandler(func(_ any, ss grpc.ServerStream) error { + md, _ := metadata.FromIncomingContext(ss.Context()) + vals := md.Get(meta.NamespaceHeader) + if len(vals) == 0 { + got <- "" + } else { + got <- vals[len(vals)-1] + } + return nil + })) + serve(t, srv, lis) + + return dialBufconn(t, lis), got +} + func dialBufconn(t *testing.T, lis *bufconn.Listener) *grpc.ClientConn { t.Helper() conn, err := grpc.NewClient( diff --git a/internal/transport/connect/conn.go b/internal/transport/connect/conn.go new file mode 100644 index 0000000..1b4d821 --- /dev/null +++ b/internal/transport/connect/conn.go @@ -0,0 +1,129 @@ +package connect + +import ( + "context" + "fmt" + + "google.golang.org/grpc" +) + +type ( + // Conn is a [grpc.ClientConnInterface] that resolves its dial target per + // call through a Resolver and fetches (lazily creating) the underlying + // pooled connection through a ConnFactory. With a dynamic Resolver a single + // Conn fronts many physical connections (e.g. one per namespace); with a + // static Resolver it always resolves to the same one. Construct one with + // NewConn. + Conn struct { + factory ConnFactory + resolver Resolver + } + + // ConnFactory returns the pooled connection for a (key, target) pair, + // creating it on first use. [Pool.ConnOrCreate] satisfies this signature: + // the first argument is the logical cache key and the second is the dial + // address. + ConnFactory func(string, string, ...grpc.DialOption) (*grpc.ClientConn, error) + + // Resolver decides, per request, which connection a Conn should use. Resolve + // returns the pool cache key, the dial target, and the dial options for that + // connection. IsStatic reports whether the resolution is fixed for the life + // of the Conn: a static resolver is dialed eagerly when the Conn is created; + // a dynamic one is resolved lazily on every call. + Resolver interface { + IsStatic() bool + Resolve(context.Context) (string, string, []grpc.DialOption, error) + } + + // staticResolver resolves to a fixed address and options, ignoring the + // request context. Its cache key equals its dial target, so two static + // resolvers for the same address with different options would share one + // pooled connection; that does not arise here because upstream hostPorts are + // unique (enforced by config). Contrast the dynamic resolver in + // internal/proxy, which folds the rendered server name into its key. + staticResolver struct { + addr string + opts []grpc.DialOption + } +) + +// NewConn returns a Conn that resolves through r and dials through f. When r is +// static the pooled connection is created eagerly here, so a malformed target +// or bad dial option surfaces at construction rather than on the first request; +// the socket itself is not opened until first use (gRPC connects lazily). A +// dynamic resolver defers creation to the first call that resolves a given +// target. +func NewConn(f ConnFactory, r Resolver) (*Conn, error) { + cc := &Conn{ + factory: f, + resolver: r, + } + + // Static resolvers are eager loaded + if r.IsStatic() { + if _, err := cc.conn(context.Background()); err != nil { + return nil, fmt.Errorf("failed to initialize connection: %w", err) + } + } + + return cc, nil +} + +// StaticResolver returns a Resolver that always resolves to hostPort with the +// given dial options. It reports IsStatic as true, so a Conn built from it is +// created eagerly and reuses a single pooled connection keyed by hostPort. +func StaticResolver(hostPort string, opts ...grpc.DialOption) Resolver { + return &staticResolver{ + addr: hostPort, + opts: opts, + } +} + +// Invoke resolves the connection for this call and forwards the unary RPC to +// it, satisfying [grpc.ClientConnInterface]. +func (c *Conn) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error { + cc, err := c.conn(ctx) + if err != nil { + return err + } + + return cc.Invoke(ctx, method, args, reply, opts...) +} + +// NewStream resolves the connection for this call and opens the stream on it, +// satisfying [grpc.ClientConnInterface]. The target is resolved from ctx before +// any message is sent, so streaming and unary calls share one resolution path. +func (c *Conn) NewStream( + ctx context.Context, + desc *grpc.StreamDesc, + method string, + opts ...grpc.CallOption, +) (grpc.ClientStream, error) { + cc, err := c.conn(ctx) + if err != nil { + return nil, err + } + + return cc.NewStream(ctx, desc, method, opts...) +} + +// conn resolves the request and returns the pooled connection for it. +func (c *Conn) conn(ctx context.Context) (*grpc.ClientConn, error) { + key, target, opts, err := c.resolver.Resolve(ctx) + if err != nil { + return nil, err + } + + return c.factory(key, target, opts...) +} + +// IsStatic reports that a staticResolver never varies with the request. +func (r *staticResolver) IsStatic() bool { + return true +} + +// Resolve returns the fixed address as both the cache key and dial target, +// along with the configured dial options. +func (r *staticResolver) Resolve(context.Context) (string, string, []grpc.DialOption, error) { + return r.addr, r.addr, r.opts, nil +} diff --git a/internal/transport/connect/conn_test.go b/internal/transport/connect/conn_test.go new file mode 100644 index 0000000..180f2de --- /dev/null +++ b/internal/transport/connect/conn_test.go @@ -0,0 +1,126 @@ +package connect_test + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/temporalio/temporal-proxy/internal/transport/connect" +) + +type fakeResolver struct { + static bool + key, target string + opts []grpc.DialOption + err error +} + +func TestStaticResolver(t *testing.T) { + t.Parallel() + + r := connect.StaticResolver("host:7233", insecureOpt()) + require.True(t, r.IsStatic()) + + key, target, opts, err := r.Resolve(t.Context()) + require.NoError(t, err) + require.Equal(t, "host:7233", key, "cache key equals the address") + require.Equal(t, "host:7233", target, "dial target equals the address") + require.Len(t, opts, 1) +} + +func TestNewConnEagerLoadsStatic(t *testing.T) { + t.Parallel() + + pool := connect.NewPool() + t.Cleanup(func() { _ = pool.Close() }) + + _, err := connect.NewConn(pool.ConnOrCreate, connect.StaticResolver("passthrough:///eager", insecureOpt())) + require.NoError(t, err) + + // Eager load means the connection is already registered in the pool before + // any request is made. + conn, err := pool.Conn("passthrough:///eager") + require.NoError(t, err) + require.NotNil(t, conn) +} + +func TestNewConnDoesNotDialDynamic(t *testing.T) { + t.Parallel() + + calls := 0 + _, err := connect.NewConn(countingFactory(&calls, nil, nil), fakeResolver{static: false}) + require.NoError(t, err) + require.Zero(t, calls, "a dynamic resolver must not dial at construction") +} + +func TestNewConnStaticFactoryErrorFailsFast(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("boom") + + _, err := connect.NewConn( + countingFactory(&calls, nil, boom), + connect.StaticResolver("passthrough:///x", insecureOpt()), + ) + require.Error(t, err) + require.ErrorContains(t, err, "failed to initialize connection") + require.ErrorIs(t, err, boom) + require.Equal(t, 1, calls) +} + +func TestConnResolveErrorPropagates(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("resolve failed") + + c, err := connect.NewConn(countingFactory(&calls, nil, nil), fakeResolver{static: false, err: boom}) + require.NoError(t, err, "dynamic resolver is not resolved until a call is made") + require.ErrorIs(t, c.Invoke(t.Context(), "/svc/Method", nil, nil), boom) + + _, streamErr := c.NewStream(t.Context(), &grpc.StreamDesc{}, "/svc/Method") + require.ErrorIs(t, streamErr, boom) + require.Zero(t, calls, "the factory is never reached when resolution fails") +} + +func TestConnFactoryErrorPropagates(t *testing.T) { + t.Parallel() + + calls := 0 + boom := errors.New("dial failed") + + c, err := connect.NewConn( + countingFactory(&calls, nil, boom), + fakeResolver{static: false, key: "k", target: "t"}, + ) + require.NoError(t, err) + require.ErrorIs(t, c.Invoke(t.Context(), "/svc/Method", nil, nil), boom) + + _, streamErr := c.NewStream(t.Context(), &grpc.StreamDesc{}, "/svc/Method") + require.ErrorIs(t, streamErr, boom) + require.Equal(t, 2, calls, "both Invoke and NewStream resolve through the factory") +} + +func (f fakeResolver) IsStatic() bool { return f.static } + +func (f fakeResolver) Resolve(context.Context) (string, string, []grpc.DialOption, error) { + return f.key, f.target, f.opts, f.err +} + +// countingFactory returns a ConnFactory that records how many times it is +// invoked and always yields conn/err. +func countingFactory(calls *int, conn *grpc.ClientConn, err error) connect.ConnFactory { + return func(string, string, ...grpc.DialOption) (*grpc.ClientConn, error) { + *calls++ + return conn, err + } +} + +func insecureOpt() grpc.DialOption { + return grpc.WithTransportCredentials(insecure.NewCredentials()) +} diff --git a/internal/transport/connect/doc.go b/internal/transport/connect/doc.go index 291016d..b95a3c6 100644 --- a/internal/transport/connect/doc.go +++ b/internal/transport/connect/doc.go @@ -1,5 +1,10 @@ // Package connect manages a pool of reusable gRPC client connections keyed by -// host. Use NewPool to create a Pool, Set to register a connection for a host, -// Conn to retrieve it, GetOrSet to retrieve or lazily dial one, and Close to -// shut every connection down exactly once. +// a caller-supplied logical key, distinct from the dial target. Use NewPool to +// create a Pool, Set to register a connection for a key, Conn to retrieve it, +// ConnOrCreate to retrieve or create one on first use, and Close to shut every connection +// down exactly once. +// +// Conn wraps that pool as a [grpc.ClientConnInterface] whose dial target is +// chosen per call by a Resolver: StaticResolver for a fixed upstream, or a +// caller-supplied dynamic Resolver that varies the target with the request. package connect diff --git a/internal/transport/connect/pool.go b/internal/transport/connect/pool.go index 87fff25..1d96846 100644 --- a/internal/transport/connect/pool.go +++ b/internal/transport/connect/pool.go @@ -9,17 +9,21 @@ import ( ) var ( - // ErrDuplicateHost is returned by Pool.Set when a connection is already - // registered for the given host. - ErrDuplicateHost = errors.New("host already defined") + // ErrDuplicateKey is returned by Pool.Set when a connection is already + // registered for the given key. + ErrDuplicateKey = errors.New("key already defined") - // ErrHostNotFound is returned by Pool.Conn when no connection is registered - // for the given host. - ErrHostNotFound = errors.New("no connection for host") + // ErrKeyNotFound is returned by Pool.Conn when no connection is registered + // for the given key. + ErrKeyNotFound = errors.New("no connection for key") ) type ( - // Pool is a concurrency-safe set of gRPC client connections keyed by host. + // Pool is a concurrency-safe set of gRPC client connections keyed by a + // caller-supplied logical key, which is distinct from the dial target. + // Callers that need two connections to the same dial target (e.g. the same + // host:port with different TLS server names) must use different keys, or + // they will collapse onto whichever connection was dialed first. // The zero value is not usable; create one with NewPool. Pool struct { mu sync.RWMutex @@ -36,38 +40,41 @@ func NewPool() *Pool { } } -// Conn returns the connection registered for host. It returns ErrHostNotFound -// if no connection is registered for that host. -func (p *Pool) Conn(host string) (*grpc.ClientConn, error) { +// Conn returns the connection registered for key. It returns ErrKeyNotFound +// if no connection is registered for that key. +func (p *Pool) Conn(key string) (*grpc.ClientConn, error) { p.mu.RLock() - cn, ok := p.conns[host] + cn, ok := p.conns[key] p.mu.RUnlock() if !ok { - return nil, fmt.Errorf("%w: %s", ErrHostNotFound, host) + return nil, fmt.Errorf("%w: %s", ErrKeyNotFound, key) } return cn, nil } -// GetOrSet returns the connection registered for host, creating and registering -// one with grpc.NewClient(host, opts...) when none exists yet. If callers race to -// create the same host, each dials but only one connection is kept; the losers are -// closed and every caller receives the same *grpc.ClientConn. -func (p *Pool) GetOrSet(host string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { - if conn, _ := p.Conn(host); conn != nil { +// ConnOrCreate returns the connection registered for key, creating and registering +// one with grpc.NewClient(target, opts...) when none exists yet. key is the +// logical cache key and target is the dial address; callers that need distinct +// connections to the same target (e.g. identical host:port with different TLS +// server names) must pass distinct keys. If callers race to create the same +// key, each constructs a client but only one connection is kept; the losers are closed and +// every caller receives the same *grpc.ClientConn. +func (p *Pool) ConnOrCreate(key, target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { + if conn, _ := p.Conn(key); conn != nil { return conn, nil } - conn, err := grpc.NewClient(host, opts...) + conn, err := grpc.NewClient(target, opts...) if err != nil { - return nil, fmt.Errorf("failed to connect to %s, %w", host, err) + return nil, fmt.Errorf("failed to connect to %s, %w", target, err) } - if err := p.Set(host, conn); err != nil { - if errors.Is(err, ErrDuplicateHost) { - _ = conn.Close() // lost the race; don't leak our dial - return p.Conn(host) // return the connection that won + if err := p.Set(key, conn); err != nil { + if errors.Is(err, ErrDuplicateKey) { + _ = conn.Close() // lost the race; don't leak our dial + return p.Conn(key) // return the connection that won } return nil, err @@ -76,17 +83,17 @@ func (p *Pool) GetOrSet(host string, opts ...grpc.DialOption) (*grpc.ClientConn, return conn, nil } -// Set registers conn for host. It returns ErrDuplicateHost if a connection is -// already registered for that host, leaving the existing connection untouched. -func (p *Pool) Set(host string, conn *grpc.ClientConn) error { +// Set registers conn for key. It returns ErrDuplicateKey if a connection is +// already registered for that key, leaving the existing connection untouched. +func (p *Pool) Set(key string, conn *grpc.ClientConn) error { p.mu.Lock() defer p.mu.Unlock() - if _, ok := p.conns[host]; ok { - return fmt.Errorf("%w: %s", ErrDuplicateHost, host) + if _, ok := p.conns[key]; ok { + return fmt.Errorf("%w: %s", ErrDuplicateKey, key) } - p.conns[host] = conn + p.conns[key] = conn return nil } diff --git a/internal/transport/connect/pool_test.go b/internal/transport/connect/pool_test.go index d62cb34..a4f113f 100644 --- a/internal/transport/connect/pool_test.go +++ b/internal/transport/connect/pool_test.go @@ -15,24 +15,24 @@ import ( func TestPoolConn(t *testing.T) { t.Parallel() - t.Run("returns ErrHostNotFound when host is absent", func(t *testing.T) { + t.Run("returns ErrKeyNotFound when key is absent", func(t *testing.T) { t.Parallel() p := connect.NewPool() cn, err := p.Conn("missing") require.Nil(t, cn) - require.ErrorIs(t, err, connect.ErrHostNotFound) + require.ErrorIs(t, err, connect.ErrKeyNotFound) require.Contains(t, err.Error(), "missing") }) - t.Run("returns the connection when host is present", func(t *testing.T) { + t.Run("returns the connection when key is present", func(t *testing.T) { t.Parallel() p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, conn, cn) }) @@ -45,18 +45,18 @@ func TestPoolSet(t *testing.T) { t.Parallel() p := connect.NewPool() - require.NoError(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", newConn(t))) }) - t.Run("rejects a duplicate host", func(t *testing.T) { + t.Run("rejects a duplicate key", func(t *testing.T) { t.Parallel() p := connect.NewPool() - require.NoError(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", newConn(t))) - err := p.Set("host", newConn(t)) - require.ErrorIs(t, err, connect.ErrDuplicateHost) - require.Contains(t, err.Error(), "host") + err := p.Set("key", newConn(t)) + require.ErrorIs(t, err, connect.ErrDuplicateKey) + require.Contains(t, err.Error(), "key") }) t.Run("keeps the original connection after a duplicate is rejected", func(t *testing.T) { @@ -64,10 +64,10 @@ func TestPoolSet(t *testing.T) { p := connect.NewPool() original := newConn(t) - require.NoError(t, p.Set("host", original)) - require.Error(t, p.Set("host", newConn(t))) + require.NoError(t, p.Set("key", original)) + require.Error(t, p.Set("key", newConn(t))) - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, original, cn) }) @@ -97,7 +97,7 @@ func TestPoolClose(t *testing.T) { p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) // First close shuts the underlying connection down. A second close on // the conn itself would error, so this proves Close short-circuits via @@ -111,7 +111,7 @@ func TestPoolClose(t *testing.T) { p := connect.NewPool() conn := newConn(t) - require.NoError(t, p.Set("host", conn)) + require.NoError(t, p.Set("key", conn)) // Close the connection out from under the pool so the pool's own // Close observes an "already closed" error and surfaces it. @@ -132,7 +132,7 @@ func TestPoolConcurrent(t *testing.T) { const ( workers = 50 - hosts = 8 + keys = 8 ) var wg sync.WaitGroup @@ -143,23 +143,23 @@ func TestPoolConcurrent(t *testing.T) { go func(i int) { defer wg.Done() - host := fmt.Sprintf("host-%d", i%hosts) + key := fmt.Sprintf("key-%d", i%keys) conn := newConn(t) <-start // line every goroutine up so the operations actually overlap - // Multiple goroutines fight over each host: one wins, the rest get - // ErrDuplicateHost. Close the connection we couldn't hand off so it + // Multiple goroutines fight over each key: one wins, the rest get + // ErrDuplicateKey. Close the connection we couldn't hand off so it // doesn't leak. - if err := p.Set(host, conn); err != nil { - require.ErrorIs(t, err, connect.ErrDuplicateHost) + if err := p.Set(key, conn); err != nil { + require.ErrorIs(t, err, connect.ErrDuplicateKey) require.NoError(t, conn.Close()) } - // Conn may or may not find the host yet depending on scheduling; + // Conn may or may not find the key yet depending on scheduling; // both outcomes are valid, we only care that the read is race-free. - if _, err := p.Conn(host); err != nil { - require.ErrorIs(t, err, connect.ErrHostNotFound) + if _, err := p.Conn(key); err != nil { + require.ErrorIs(t, err, connect.ErrKeyNotFound) } // Close racing against Set/Conn must also be safe. closeOnce means @@ -172,37 +172,37 @@ func TestPoolConcurrent(t *testing.T) { wg.Wait() } -func TestPoolGetOrSet(t *testing.T) { +func TestPoolConnOrCreate(t *testing.T) { t.Parallel() insecureOpt := grpc.WithTransportCredentials(insecure.NewCredentials()) - t.Run("creates and registers a connection when host is absent", func(t *testing.T) { + t.Run("creates and registers a connection when key is absent", func(t *testing.T) { t.Parallel() p := connect.NewPool() t.Cleanup(func() { require.NoError(t, p.Close()) }) - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) require.NotNil(t, conn) // The connection is registered, so a subsequent read returns the same one. - cn, err := p.Conn("host") + cn, err := p.Conn("key") require.NoError(t, err) require.Same(t, conn, cn) }) - t.Run("returns the existing connection when host is present", func(t *testing.T) { + t.Run("returns the existing connection when key is present", func(t *testing.T) { t.Parallel() p := connect.NewPool() t.Cleanup(func() { require.NoError(t, p.Close()) }) existing := newConn(t) - require.NoError(t, p.Set("host", existing)) + require.NoError(t, p.Set("key", existing)) - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) require.Same(t, existing, conn) }) @@ -214,15 +214,49 @@ func TestPoolGetOrSet(t *testing.T) { t.Cleanup(func() { require.NoError(t, p.Close()) }) // No transport credentials option, so grpc.NewClient fails synchronously. - conn, err := p.GetOrSet("host") + conn, err := p.ConnOrCreate("key", "target") require.Nil(t, conn) require.ErrorContains(t, err, "failed to connect") }) + t.Run("same target with different keys yields distinct connections", func(t *testing.T) { + t.Parallel() + + // A templated serverName can vary independently of the dial address. + // Two logical keys sharing a target must not collapse onto the same + // pooled connection, or one caller's TLS identity would silently leak + // onto another's channel. + p := connect.NewPool() + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + connA, err := p.ConnOrCreate("key-a", "shared-target", insecureOpt) + require.NoError(t, err) + + connB, err := p.ConnOrCreate("key-b", "shared-target", insecureOpt) + require.NoError(t, err) + + require.NotSame(t, connA, connB) + }) + + t.Run("same key returns the same connection regardless of target", func(t *testing.T) { + t.Parallel() + + p := connect.NewPool() + t.Cleanup(func() { require.NoError(t, p.Close()) }) + + connA, err := p.ConnOrCreate("key", "target-a", insecureOpt) + require.NoError(t, err) + + connB, err := p.ConnOrCreate("key", "target-b", insecureOpt) + require.NoError(t, err) + + require.Same(t, connA, connB) + }) + t.Run("hands every concurrent caller the same connection", func(t *testing.T) { t.Parallel() - // Many goroutines race to create the same host. Only one dial can be + // Many goroutines race to create the same key. Only one dial can be // retained; the losers must be closed and every caller must receive the // winner. This only proves something under `go test -race`. p := connect.NewPool() @@ -238,7 +272,7 @@ func TestPoolGetOrSet(t *testing.T) { wg.Go(func() { <-start // line every goroutine up so the calls actually overlap - conn, err := p.GetOrSet("host", insecureOpt) + conn, err := p.ConnOrCreate("key", "target", insecureOpt) require.NoError(t, err) conns[i] = conn }) diff --git a/internal/transport/creds/mtls.go b/internal/transport/creds/mtls.go index 7c7cb0f..87b3d09 100644 --- a/internal/transport/creds/mtls.go +++ b/internal/transport/creds/mtls.go @@ -5,6 +5,7 @@ import ( "crypto/x509" "fmt" "os" + "sync" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -21,6 +22,7 @@ type ( certFile string keyFile string serverName string + loader *CertLoader } // MTLSOptions holds optional configuration for [MTLS] connections. @@ -29,6 +31,31 @@ type ( // certificate hostname. Useful when the server's certificate CN does not // match its dial address. ServerName string + + // Loader, when set, supplies the client certificate and CA pool used by + // [MTLS.DialOption]. Share one loader across the short-lived MTLS values + // built per request for a templated upstream so the certificate and CA + // files are read and parsed once rather than on every request. When nil, + // the MTLS loads (and caches) its own material on first use. + Loader *CertLoader + } + + // CertLoader reads and parses a client certificate/key pair and a CA + // certificate pool once, caching the result for reuse across [MTLS] values. + // It is safe for concurrent use: the parsed material is immutable after the + // first load, so a cached CA pool and certificate may be shared by many + // dial options. A rotated certificate on disk is not picked up until the + // process restarts, matching how a fixed-address upstream loads its material + // once at startup. + CertLoader struct { + caFile string + certFile string + keyFile string + + once sync.Once + cert tls.Certificate + caPool *x509.CertPool + err error } ) @@ -36,31 +63,33 @@ type ( // caFile and the client certificate/key pair from certFile and keyFile. opts // provides additional configuration; a zero-value [MTLSOptions] is valid. func NewMTLS(caFile, certFile, keyFile string, opts MTLSOptions) *MTLS { + loader := opts.Loader + if loader == nil { + loader = NewCertLoader(caFile, certFile, keyFile) + } + return &MTLS{ caFile: caFile, certFile: certFile, keyFile: keyFile, serverName: opts.ServerName, + loader: loader, } } +// NewCertLoader returns a [CertLoader] that loads the CA certificate from caFile +// and the client certificate/key pair from certFile and keyFile on first use. +func NewCertLoader(caFile, certFile, keyFile string) *CertLoader { + return &CertLoader{caFile: caFile, certFile: certFile, keyFile: keyFile} +} + // DialOption returns a [grpc.DialOption] that configures the outbound // connection with mutual TLS. The client presents its certificate/key pair and // verifies the server against the CA pool loaded from caFile. func (c *MTLS) DialOption() (grpc.DialOption, error) { - cert, err := tls.LoadX509KeyPair(c.certFile, c.keyFile) - if err != nil { - return nil, fmt.Errorf("failed to load client key pair: %w", err) - } - - ca := x509.NewCertPool() - caBytes, err := os.ReadFile(c.caFile) + cert, ca, err := c.loader.load() if err != nil { - return nil, fmt.Errorf("failed to load CA certificate: %w", err) - } - - if ok := ca.AppendCertsFromPEM(caBytes); !ok { - return nil, fmt.Errorf("failed to parse CA file: %s", c.caFile) + return nil, err } return grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{ @@ -137,3 +166,34 @@ func (c *MTLS) Validate() error { func (c *MTLS) Encrypted() bool { return true } + +// load reads and parses the client certificate/key pair and CA pool on the first +// call, caching the result (or the error) for every subsequent call. The +// returned certificate and CA pool are immutable and safe to share across dial +// options and goroutines. +func (l *CertLoader) load() (tls.Certificate, *x509.CertPool, error) { + l.once.Do(func() { + cert, err := tls.LoadX509KeyPair(l.certFile, l.keyFile) + if err != nil { + l.err = fmt.Errorf("failed to load client key pair: %w", err) + return + } + + ca := x509.NewCertPool() + caBytes, err := os.ReadFile(l.caFile) + if err != nil { + l.err = fmt.Errorf("failed to load CA certificate: %w", err) + return + } + + if ok := ca.AppendCertsFromPEM(caBytes); !ok { + l.err = fmt.Errorf("failed to parse CA file: %s", l.caFile) + return + } + + l.cert = cert + l.caPool = ca + }) + + return l.cert, l.caPool, l.err +} diff --git a/internal/transport/creds/mtls_test.go b/internal/transport/creds/mtls_test.go index fe1f442..35b3018 100644 --- a/internal/transport/creds/mtls_test.go +++ b/internal/transport/creds/mtls_test.go @@ -4,6 +4,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "math/big" + "os" "path/filepath" "testing" "time" @@ -249,3 +250,53 @@ func TestMTLS_Encrypted(t *testing.T) { t.Parallel() require.True(t, creds.NewMTLS("", "", "", creds.MTLSOptions{}).Encrypted()) } + +func TestCertLoader(t *testing.T) { + t.Parallel() + + t.Run("shared loader reads material once", func(t *testing.T) { + t.Parallel() + + caFile, certFile, keyFile := testutil.GenerateMTLSCerts(t) + loader := creds.NewCertLoader(caFile, certFile, keyFile) + + // Simulate two per-request credentials for the same upstream sharing the + // loader; only ServerName differs. + opt, err := creds.NewMTLS(caFile, certFile, keyFile, creds.MTLSOptions{ + ServerName: "first.example.com", + Loader: loader, + }).DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + + // Remove the CA file. A re-read would fail at the CA step, so a successful + // second DialOption proves the loader cached the parsed material. + require.NoError(t, os.Remove(caFile)) + + opt, err = creds.NewMTLS(caFile, certFile, keyFile, creds.MTLSOptions{ + ServerName: "second.example.com", + Loader: loader, + }).DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) + + t.Run("load error is cached", func(t *testing.T) { + t.Parallel() + + _, certFile, keyFile := testutil.GenerateMTLSCerts(t) + missingCA := filepath.Join(t.TempDir(), "missing.pem") + loader := creds.NewCertLoader(missingCA, certFile, keyFile) + + m := creds.NewMTLS(missingCA, certFile, keyFile, creds.MTLSOptions{Loader: loader}) + + opt, err := m.DialOption() + require.ErrorContains(t, err, "failed to load CA certificate") + require.Nil(t, opt) + + // The cached error is returned again without another disk read. + opt, err = m.DialOption() + require.ErrorContains(t, err, "failed to load CA certificate") + require.Nil(t, opt) + }) +} diff --git a/internal/transport/creds/tls.go b/internal/transport/creds/tls.go index 7314f32..544c56d 100644 --- a/internal/transport/creds/tls.go +++ b/internal/transport/creds/tls.go @@ -28,16 +28,18 @@ var preferredCipherSuites = []uint16{ // presents a certificate to the client; the client is not required to present // one. Minimum TLS version is 1.2. type TLS struct { - certFile string - keyFile string + certFile string + keyFile string + serverName string } // NewClientTLS returns a [TLS] credential suitable for outbound (client-side) // connections. The server's certificate is verified against the system root CA // pool; no client certificate is presented. Use [NewMTLS] when mutual -// authentication is required. -func NewClientTLS() *TLS { - return NewServerTLS("", "") +// authentication is required. serverName overrides the name used for SNI and +// certificate verification; when empty it defaults to the dial target's host. +func NewClientTLS(serverName string) *TLS { + return &TLS{serverName: serverName} } // NewServerTLS returns a [TLS] credential that loads the server certificate and @@ -51,10 +53,11 @@ func NewServerTLS(certFile, keyFile string) *TLS { // DialOption returns a [grpc.DialOption] that configures the outbound // connection with TLS, verifying the server's certificate against the system -// root CA pool. No client certificate is presented; use [MTLS] when mutual -// authentication is required. +// root CA pool. The configured server name (when non-empty) is used for SNI and +// verification; otherwise the dial target's host is used. No client certificate +// is presented; use [MTLS] when mutual authentication is required. func (c *TLS) DialOption() (grpc.DialOption, error) { - return grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), nil + return grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, c.serverName)), nil } // ServerOption returns a [grpc.ServerOption] that configures the server with diff --git a/internal/transport/creds/tls_test.go b/internal/transport/creds/tls_test.go index fcc3e19..3a0b7a5 100644 --- a/internal/transport/creds/tls_test.go +++ b/internal/transport/creds/tls_test.go @@ -17,10 +17,22 @@ import ( func TestTLS_DialOption(t *testing.T) { t.Parallel() - // DialOption does not load any files; file paths are irrelevant. - opt, err := creds.NewClientTLS().DialOption() - require.NoError(t, err) - require.NotNil(t, opt) + t.Run("without a server name", func(t *testing.T) { + t.Parallel() + + // DialOption does not load any files; file paths are irrelevant. + opt, err := creds.NewClientTLS("").DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) + + t.Run("with a server name", func(t *testing.T) { + t.Parallel() + + opt, err := creds.NewClientTLS("example.com").DialOption() + require.NoError(t, err) + require.NotNil(t, opt) + }) } func TestTLS_ServerOption(t *testing.T) { @@ -136,7 +148,7 @@ func TestTLS_Validate(t *testing.T) { // NewClientTLS leaves certFile empty; Validate is documented as // server-mode-only, but we still want to confirm it surfaces a clear // IO error rather than panicking. - err := creds.NewClientTLS().Validate() + err := creds.NewClientTLS("").Validate() require.ErrorContains(t, err, "failed to read PEM file") }) } @@ -145,6 +157,6 @@ func TestTLS_Encrypted(t *testing.T) { t.Parallel() // TLS encrypts the transport in both client and server modes. - require.True(t, creds.NewClientTLS().Encrypted()) + require.True(t, creds.NewClientTLS("").Encrypted()) require.True(t, creds.NewServerTLS("", "").Encrypted()) } diff --git a/internal/transport/meta/meta.go b/internal/transport/meta/meta.go new file mode 100644 index 0000000..9b37328 --- /dev/null +++ b/internal/transport/meta/meta.go @@ -0,0 +1,47 @@ +// Package meta defines the internal gRPC metadata contract carried between the +// router and the per-upstream proxy: the router stamps the request namespace it +// already extracted so the proxy can resolve a templated upstream address +// without parsing the payload again. It depends on no other internal packages. +package meta + +import ( + "context" + + "google.golang.org/grpc/metadata" +) + +// NamespaceHeader is the outgoing metadata key that carries the local (pre- +// translation) namespace from the router to the upstream proxy. +const NamespaceHeader = "x-temporal-proxy-namespace" + +// WithNamespace returns ctx with namespace set on its outgoing gRPC metadata, +// replacing any value already present for NamespaceHeader (so a client cannot +// influence routing by sending the header itself). +func WithNamespace(ctx context.Context, namespace string) context.Context { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + md = metadata.MD{} + } else { + md = md.Copy() + } + + md.Set(NamespaceHeader, namespace) + return metadata.NewOutgoingContext(ctx, md) +} + +// NamespaceFrom returns the namespace carried on ctx's outgoing metadata, or "" +// when absent. When multiple values are present the last (most recently added) +// wins. +func NamespaceFrom(ctx context.Context) string { + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + return "" + } + + vals := md.Get(NamespaceHeader) + if len(vals) == 0 { + return "" + } + + return vals[len(vals)-1] +} diff --git a/internal/transport/meta/meta_test.go b/internal/transport/meta/meta_test.go new file mode 100644 index 0000000..958a626 --- /dev/null +++ b/internal/transport/meta/meta_test.go @@ -0,0 +1,44 @@ +package meta_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc/metadata" + + "github.com/temporalio/temporal-proxy/internal/transport/meta" +) + +func TestWithNamespaceRoundTrips(t *testing.T) { + t.Parallel() + + ctx := meta.WithNamespace(t.Context(), "orders") + require.Equal(t, "orders", meta.NamespaceFrom(ctx)) +} + +func TestWithNamespaceOverwritesExisting(t *testing.T) { + t.Parallel() + + // A spoofed value already present in outgoing metadata is replaced. + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(meta.NamespaceHeader, "spoofed")) + ctx = meta.WithNamespace(ctx, "orders") + + md, _ := metadata.FromOutgoingContext(ctx) + require.Equal(t, []string{"orders"}, md.Get(meta.NamespaceHeader)) +} + +func TestNamespaceFromReturnsLastValue(t *testing.T) { + t.Parallel() + + ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs( + meta.NamespaceHeader, "old", + meta.NamespaceHeader, "new", + )) + require.Equal(t, "new", meta.NamespaceFrom(ctx)) +} + +func TestNamespaceFromAbsentIsEmpty(t *testing.T) { + t.Parallel() + + require.Equal(t, "", meta.NamespaceFrom(t.Context())) +} diff --git a/rfc/routing.md b/rfc/routing.md index cae60e0..ff1f816 100644 --- a/rfc/routing.md +++ b/rfc/routing.md @@ -2,83 +2,119 @@ How to handle routing to multiple upstream clusters based on namespace. -# Background +## Background -The proxy currently supports a single static upstream defined in config. However, we want a single proxy (tenant) to serve many namespaces, each potentially terminating at a different address/upstream. +The proxy currently supports a single static upstream defined in config. However, we want a single proxy (tenant) to +serve many namespaces, each potentially terminating at a different address/upstream. -This can be for a number of reasons, but notably, during migrations, some teams will want to move over namespace-by-namespace rather than cluster-by-cluster. They’d point all namespaces to the original upstream (typically the OSS frontend) and switch to another one as needed based on some routing rules. +This can be for a number of reasons, but notably, during migrations, some teams will want to move over +namespace-by-namespace rather than cluster-by-cluster. They’d point all namespaces to the original upstream (typically +the OSS frontend) and switch to another one as needed based on some routing rules. > [!NOTE] -> Cloud offers three relevant flavours of endpoint: per-namespace endpoints (`..tmprl.cloud:7233`), regional endpoints (`.aws.tmprl.cloud:7233`), and private-link endpoints (per-VPC hostnames). Operators also need to mix Cloud and self-hosted upstreams in a single proxy. > -> Ideally, we’d use the namespace endpoints since they’re already set up to handle multi-region traffic. However, there are times when that isn’t possible, so we should support all cases. +> Cloud offers three relevant flavours of endpoint: per-namespace endpoints (`..tmprl.cloud:7233`), +> regional endpoints (`.aws.tmprl.cloud:7233`), and private-link endpoints (per-VPC hostnames). Operators also +> need to mix Cloud and self-hosted upstreams in a single proxy. +> +> Ideally, we’d use the namespace endpoints since they’re already set up to handle multi-region traffic. However, there +> are times when that isn’t possible, so we should support all cases. -# Proposal +## Proposal -We need to support multiple named upstream configs in a single proxy and select among them (per request) using a set of routing rules. These rules will have access to the incoming namespace (before and after translation) and to gRPC metadata headers. +We need to support multiple named upstream configs in a single proxy and select among them (per request) using a set of +routing rules. These rules will have access to the incoming namespace (before and after translation) and to gRPC +metadata headers. -## Upstreams +### Upstreams -As part of this, we’ll need to exchange the single upstream we have today for multiple upstream definitions. Each upstream will have its `hostPort` and `tls.serverName` fields templated, allowing for the following replacements: +As part of this, we’ll need to exchange the single upstream we have today for multiple upstream definitions. Each +upstream will have its `hostPort` and `tls.serverName` fields templated, allowing for the following replacements: -* `{{ .LocalNamespace }}` \- The namespace before translation -* `{{ .RemoteNamespace }}` \- The namespace after translation -* `{{ .Metadata.DC }}` \- The `DC` metadata value (arbitrary) -* `{{ index .Metadata “x-cluster” }}` \- The `x-cluster` metadata value (arbitrary) +- `{{ .LocalNamespace }}` \- The namespace before translation +- `{{ .RemoteNamespace }}` \- The namespace after translation +- `{{ .Metadata.DC }}` \- The `DC` metadata value (arbitrary) +- `{{ index .Metadata “x-cluster” }}` \- The `x-cluster` metadata value (arbitrary) -To be clear, these are simply available, but not required. Static `hostPort/serverName` values are not a problem (see the example config [below](#configuration-schema)). +To be clear, these are simply available, but not required. Static `hostPort/serverName` values are not a problem (see +the example config [below](#configuration-schema)). -To minimize runtime issues, all static `hostPort` upstreams will establish eager connections at startup. Templated values are not predetermined and therefore cannot benefit from this. Those connections will be created lazily on first use. +To minimize runtime issues, all static `hostPort` upstreams will establish eager connections at startup. Templated +values are not predetermined and therefore cannot benefit from this. Those connections will be created lazily on first +use. -All connections will be maintained in a shared container (TBD) since connection establishment is expensive. Each connection will be a gRPC `ClientConn,` ensuring automatic redials and shared underlying channels via connection pooling. +All connections will be maintained in a shared container (TBD) since connection establishment is expensive. Each +connection will be a gRPC `ClientConn,` ensuring automatic redials and shared underlying channels via connection +pooling. -## Routing +### Routing -To match incoming requests to the appropriate upstream, we’ll need to define routing rules. In proxy config, we can define a set of rules that grant access to request information, which can be used to select an upstream connection. +To match incoming requests to the appropriate upstream, we’ll need to define routing rules. In proxy config, we can +define a set of rules that grant access to request information, which can be used to select an upstream connection. -These rules will be evaluated in the order in which they are defined. The first matching rule will be used to route the request, with one notable exception: requests without a namespace (system requests). These will be sent to a particular upstream defined as the `system` upstream. +These rules will be evaluated in the order in which they are defined. The first matching rule will be used to route the +request, with one notable exception: requests without a namespace (system requests). These will be sent to a particular +upstream defined as the `system` upstream. -The other special case is handling the absence of a match. It seems perfectly reasonable to reject the request with an error, but some operators may prefer to fall back to a specific upstream (e.g., receiving Temporal’s `NamespaceNotFound` rather than a proxy-specific error). +The other special case is handling the absence of a match. It seems perfectly reasonable to reject the request with an +error, but some operators may prefer to fall back to a specific upstream (e.g., receiving Temporal’s `NamespaceNotFound` +rather than a proxy-specific error). -### Rules +#### Rules -To route incoming requests to the appropriate upstream, we’ll need to define a mapping in the proxy config. The mapping can be done using the incoming namespace and/or gRPC metadata attached to the request. A rule is defined with a matcher that specifies the namespace and the metadata values to match, and an upstream to route to. +To route incoming requests to the appropriate upstream, we’ll need to define a mapping in the proxy config. The mapping +can be done using the incoming namespace and/or gRPC metadata attached to the request. A rule is defined with a matcher +that specifies the namespace and the metadata values to match, and an upstream to route to. -Rules use `AND` logic to match. This means that defining a rule for `namespace=test` that contains a metadata `key=value` pair requires both conditions to be true to match. While more complex logic could be added, it also introduces complexity and the risk of logic errors in the config. +Rules use `AND` logic to match. This means that defining a rule for `namespace=test` that contains a metadata +`key=value` pair requires both conditions to be true to match. While more complex logic could be added, it also +introduces complexity and the risk of logic errors in the config. It is an error to define a rule without a namespace or metadata values. This will fail validation on startup. -#### Namespace Rules +##### Namespace Rules -Since namespace translation is defined per upstream and we don’t yet know which upstream we’re using, the local namespace (before translation) is used to perform the match. When a namespace is not defined on the rule, no namespace-based matching will be performed. +Since namespace translation is defined per upstream and we don’t yet know which upstream we’re using, the local +namespace (before translation) is used to perform the match. When a namespace is not defined on the rule, no +namespace-based matching will be performed. -Namespace matches can be defined as string literals or simple globs. This supports exact-match, starts-with, ends-with, and contains semantics. Here are some valid examples: +Namespace matches can be defined as string literals or simple globs. This supports exact-match, starts-with, ends-with, +and contains semantics. Here are some valid examples: -* `my-namespace` -* `prod-*` -* `*-test` -* `*-test-*` +- `my-namespace` +- `prod-*` +- `*-test` +- `*-test-*` -#### Metadata Rules +##### Metadata Rules -Key/value pairs can be added to the match to validate incoming header metadata. When multiple conditions are defined, they are `AND`ed (i.e. all conditions must match). +Key/value pairs can be added to the match to validate incoming header metadata. When multiple conditions are defined, +they are `AND`ed (i.e. all conditions must match). -A header condition contains a map of keys and values. Keys are not case-sensitive (normalized to lowercase) but do not support wildcards. The intent here is to allow operators to make routing decisions based on known metadata (e.g. `X-Data-Centre`). Values follow the same convention as namespaces, with support for exact-match, starts-with, ends-with, and contains semantics. +A header condition contains a map of keys and values. Keys are not case-sensitive (normalized to lowercase) but do not +support wildcards. The intent here is to allow operators to make routing decisions based on known metadata (e.g. +`X-Data-Centre`). Values follow the same convention as namespaces, with support for exact-match, starts-with, ends-with, +and contains semantics. -### System Upstream +#### System Upstream -Not all RPCs have a namespace. For example, `GetClusterInfo` or `GetSystemInfo`. When these are called, we need to ensure they get sent to the appropriate upstream. To handle this, the routing config has a `system` field which can be set to a known upstream. +Not all RPCs have a namespace. For example, `GetClusterInfo` or `GetSystemInfo`. When these are called, we need to +ensure they get sent to the appropriate upstream. To handle this, the routing config has a `system` field which can be +set to a known upstream. When not defined, requests without a namespace fallback to the default upstream (see below). -### Default Upstream +#### Default Upstream -When a request cannot be matched, that is to say, no routing rules apply, the proxy will return an error (e.g. `RouteNotFound`). This isn’t always what operators desire. Sometimes they’d like a fallback that sends the request to a default upstream. +When a request cannot be matched, that is to say, no routing rules apply, the proxy will return an error (e.g. +`RouteNotFound`). This isn’t always what operators desire. Sometimes they’d like a fallback that sends the request to a +default upstream. -For these cases, a `default` upstream can be defined in the routing rules. This will ensure all requests are proxied regardless of namespace or metadata. +For these cases, a `default` upstream can be defined in the routing rules. This will ensure all requests are proxied +regardless of namespace or metadata. -## Configuration Schema +### Configuration Schema -Based on the information above, here’s an example of what the config could look like. +Based on the information above, here’s an example of what the config could look like. ```yaml hostPort: 0.0.0.0:7233