From 1e40b17c08cd47bbca078e64882fed9cfd0bf4fc Mon Sep 17 00:00:00 2001 From: "David Muto (pseudomuto)" Date: Tue, 6 Oct 2026 14:05:43 -0400 Subject: [PATCH 1/2] [dataplane]: Remove the per-upstream unix socket Since #198 the gateway hands each accepted stream straight to the upstream's Forwarder.Handle, so nothing in the proxy dials the per-upstream socket any more. It was kept for local workers dialing it directly, but that is not a supported access path: workers only ever connect through the gateway, and the socket's path was a hash under the temp dir that nothing exposed. This removes the socket and everything that existed to serve it: internal/transport/socket, proxy.Server, and Dataplane.SocketPath. Start now opens the static upstream connections and binds the gateway. Stop drains only the gateway, so the concurrent per-upstream drain and its deadline arithmetic go with it. --- e2e/inbound_auth_strip_test.go | 7 +- e2e/metric_labels_test.go | 9 +- ...et_test.go => upstream_credential_test.go} | 7 +- examples/authz/README.md | 6 +- examples/cloud/README.md | 4 +- examples/codecserver/README.md | 16 +- examples/kms/README.md | 21 +- examples/listnamespace/README.md | 4 +- internal/config/cloudapi.go | 8 +- internal/dataplane/abort_test.go | 31 +- internal/dataplane/dataplane.go | 73 ++--- .../dataplane/dataplanetest/dataplanetest.go | 30 -- .../dataplanetest/dataplanetest_test.go | 15 - internal/dataplane/dataplanetest/upstream.go | 4 +- internal/dataplane/inprocess_test.go | 29 -- internal/dataplane/lifecycle.go | 43 +-- internal/dataplane/lifecycle_test.go | 110 ++----- internal/metrics/doc.go | 6 +- internal/proxy/doc.go | 7 +- internal/proxy/encryption_test.go | 4 +- internal/proxy/forwarding_test.go | 66 +++- internal/proxy/server.go | 120 -------- internal/proxy/server_test.go | 288 ------------------ internal/proxy/translation_test.go | 31 +- internal/transport/connect/fx.go | 10 +- internal/transport/socket/socket.go | 54 ---- internal/transport/socket/socket_test.go | 84 ----- 27 files changed, 174 insertions(+), 913 deletions(-) rename e2e/{upstream_credential_socket_test.go => upstream_credential_test.go} (87%) delete mode 100644 internal/dataplane/inprocess_test.go delete mode 100644 internal/proxy/server.go delete mode 100644 internal/proxy/server_test.go delete mode 100644 internal/transport/socket/socket.go delete mode 100644 internal/transport/socket/socket_test.go diff --git a/e2e/inbound_auth_strip_test.go b/e2e/inbound_auth_strip_test.go index ee1d54c..05cc5fc 100644 --- a/e2e/inbound_auth_strip_test.go +++ b/e2e/inbound_auth_strip_test.go @@ -21,10 +21,9 @@ import ( // gateway but never reaches the upstream, and the upstream sees exactly one // authorization value, the configured static API key. // -// TestProxyAttachesUpstreamCredential (in upstream_credential_socket_test.go) -// dials the per-upstream proxy's socket directly and so cannot exercise the -// gateway or router at all; this test is the only one that proves both strips -// hold together end to end. +// TestProxyAttachesUpstreamCredential (in upstream_credential_test.go) runs +// without inbound auth, so it proves only the outbound attach; this test is the +// only one that proves both strips hold together end to end. func TestEndToEndInboundAuthStrippedOutboundCredentialAttached(t *testing.T) { t.Parallel() diff --git a/e2e/metric_labels_test.go b/e2e/metric_labels_test.go index f4b4f1b..e524730 100644 --- a/e2e/metric_labels_test.go +++ b/e2e/metric_labels_test.go @@ -21,10 +21,9 @@ import ( // reaches the labels of every request-scoped collector. // // The vault_ops assertion is the one that matters: that collector lives on the -// per-upstream hop, on the far side of a unix socket that context values do not -// cross. It passes only because the value is read from the metadata the gateway -// forwards, and it is what fails if this is ever reworked into a single -// interceptor stashing values in a context. +// per-upstream hop, in the forwarder's outbound call rather than the gateway's +// interceptors, so it is what fails if the caller's metadata is lost in the +// handoff from gateway to forwarder. func TestEndToEndMetadataLabels(t *testing.T) { t.Parallel() @@ -50,7 +49,7 @@ func TestEndToEndMetadataLabels(t *testing.T) { requireLabel(t, f, "test_server_requests_total", "tenant", "acme") requireLabel(t, f, "test_router_decisions_total", "tenant", "acme") - // Hop 2: across the socket. + // Hop 2: the forwarder's outbound call. requireLabel(t, f, "test_encryption_vault_ops_total", "tenant", "acme") } diff --git a/e2e/upstream_credential_socket_test.go b/e2e/upstream_credential_test.go similarity index 87% rename from e2e/upstream_credential_socket_test.go rename to e2e/upstream_credential_test.go index 844afb6..dca5e34 100644 --- a/e2e/upstream_credential_socket_test.go +++ b/e2e/upstream_credential_test.go @@ -14,8 +14,8 @@ import ( // TestProxyAttachesUpstreamCredential proves the static credential configured // on an upstream actually reaches that upstream: it stands up a fake // WorkflowService server over TLS, points a real dataplane at it -// with a static credential configured, drives a request through the proxy's -// local socket, and asserts the fake upstream observed the +// with a static credential configured, drives a request through the gateway, +// and asserts the fake upstream observed the // "authorization: Bearer " header. The static provider requires // transport security, so this exercises the same TLS + per-RPC credential // dial path production traffic takes; construction in isolation cannot prove @@ -32,8 +32,7 @@ func TestProxyAttachesUpstreamCredential(t *testing.T) { f := dataplanetest.Start(t, cfg) - conn := f.UpstreamConn(dataplanetest.DefaultUpstream) - _, err := workflowservice.NewWorkflowServiceClient(conn).GetSystemInfo( + _, err := f.Client().GetSystemInfo( f.Context(), &workflowservice.GetSystemInfoRequest{}, grpc.WaitForReady(true), ) require.NoError(t, err) diff --git a/examples/authz/README.md b/examples/authz/README.md index c2e8bf0..512eb27 100644 --- a/examples/authz/README.md +++ b/examples/authz/README.md @@ -124,15 +124,13 @@ go run ./cmd/proxy serve -c examples/authz/config.yaml ``` ```text -{"level":"warn","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-08-13T14:20:36-04:00","message":"Running with insecure credentials. Configure TLS for production use."} -{"level":"info","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-08-13T14:20:36-04:00","message":"Starting the server"} {"level":"warn","addr":"127.0.0.1:7234","time":"2026-08-13T14:20:36-04:00","message":"Running with insecure credentials. Configure TLS for production use."} {"level":"info","addr":"127.0.0.1:7234","time":"2026-08-13T14:20:36-04:00","message":"Starting the server"} {"level":"info","component":"metrics","addr":":9090","time":"2026-08-13T14:20:36-04:00","message":"Starting metrics server"} ``` -The two `Running with insecure credentials` warnings are expected: they describe the local gateway and an internal -socket, the plaintext hops on this machine. They say nothing about whether callers are being authorized. +The `Running with insecure credentials` warning is expected: it describes the local gateway, the plaintext hop on this +machine. It says nothing about whether callers are being authorized. Terminal 4, in `examples/authz`, starts the Worker with the Worker token: diff --git a/examples/cloud/README.md b/examples/cloud/README.md index ead0df4..ddc0df8 100644 --- a/examples/cloud/README.md +++ b/examples/cloud/README.md @@ -44,8 +44,8 @@ command changes to the repo root and runs it from source: cd ../../ && go run ./cmd/proxy serve -c examples/cloud/config.yaml ``` -The proxy logs `Running with insecure credentials` for the local gateway and its internal sockets. That is expected: -those are local hops. The connection to Temporal Cloud is TLS. +The proxy logs `Running with insecure credentials` for the local gateway. That is expected: it is a local hop. The +connection to Temporal Cloud is TLS. Start the Worker: diff --git a/examples/codecserver/README.md b/examples/codecserver/README.md index 6f4cd73..bd6a432 100644 --- a/examples/codecserver/README.md +++ b/examples/codecserver/README.md @@ -55,18 +55,16 @@ go run ./cmd/proxy serve -c examples/codecserver/config.yaml ``` ```text -{"level":"info","namespace":"default","uri":"testing://","time":"2026-09-30T14:54:14-04:00","message":"Registering crypto key"} -{"level":"warn","component":"codecserver","time":"2026-09-30T14:54:14-04:00","message":"Codec server is running without authentication, which is only allowed on a loopback bind"} -{"level":"info","component":"codecserver","addr":"127.0.0.1:8081","time":"2026-09-30T14:54:14-04:00","message":"Starting the codec server"} -{"level":"warn","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-09-30T14:54:14-04:00","message":"Running with insecure credentials. Configure TLS for production use."} -{"level":"info","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-09-30T14:54:14-04:00","message":"Starting the server"} -{"level":"warn","addr":"127.0.0.1:7234","time":"2026-09-30T14:54:14-04:00","message":"Running with insecure credentials. Configure TLS for production use."} -{"level":"info","addr":"127.0.0.1:7234","time":"2026-09-30T14:54:14-04:00","message":"Starting the server"} -{"level":"info","component":"metrics","addr":":9090","time":"2026-09-30T14:54:14-04:00","message":"Starting metrics server"} +{"level":"info","namespace":"default","uri":"testing://","time":"2026-10-06T14:34:19-04:00","message":"Registering crypto key"} +{"level":"warn","component":"codecserver","time":"2026-10-06T14:34:19-04:00","message":"Codec server is running without authentication, which is only allowed on a loopback bind"} +{"level":"info","component":"codecserver","addr":"127.0.0.1:8081","time":"2026-10-06T14:34:19-04:00","message":"Starting the codec server"} +{"level":"info","component":"metrics","addr":":9090","time":"2026-10-06T14:34:19-04:00","message":"Starting metrics server"} +{"level":"warn","addr":"127.0.0.1:7234","time":"2026-10-06T14:34:19-04:00","message":"Running with insecure credentials. Configure TLS for production use."} +{"level":"info","addr":"127.0.0.1:7234","time":"2026-10-06T14:34:19-04:00","message":"Starting the server"} ``` These warnings are expected here. The codec server runs without authentication because it is bound to loopback, and the -`insecure credentials` lines describe the plaintext gateway and an internal socket on this machine. +`insecure credentials` line describes the plaintext gateway on this machine. ## Start a Workflow through the gateway diff --git a/examples/kms/README.md b/examples/kms/README.md index d5ca7e2..47e25d8 100644 --- a/examples/kms/README.md +++ b/examples/kms/README.md @@ -82,18 +82,15 @@ KMS_API_KEY=example-token go run ./cmd/proxy serve -c examples/kms/config.yaml ``` ```text -{"level":"info","namespace":"default","uri":"extension://kms/payloads","time":"2026-09-16T14:20:03-04:00","message":"Registering crypto key"} -{"level":"warn","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-09-16T14:20:03-04:00","message":"Running with insecure credentials. Configure TLS for production use."} -{"level":"info","addr":"/var/folders/6m/x_q_q9nx6ns3ygnf48xzqrn80000gn/T/127-0-0-1-7233-6689a1b6.sock","time":"2026-09-16T14:20:03-04:00","message":"Starting the server"} -{"level":"info","component":"metrics","addr":":9090","time":"2026-09-16T14:20:03-04:00","message":"Starting metrics server"} -{"level":"warn","addr":"127.0.0.1:7234","time":"2026-09-16T14:20:03-04:00","message":"Running with insecure credentials. Configure TLS for production use."} -{"level":"info","addr":"127.0.0.1:7234","time":"2026-09-16T14:20:03-04:00","message":"Starting the server"} -``` - -The two `Running with insecure credentials` warnings are expected, not a sign anything is broken: they describe the -local gateway and an internal socket, the plaintext hops between the Worker/starter and the proxy on this machine. They -say nothing about whether payloads get sealed or about the TLS connection to the extension server; both of those are -unaffected. +{"level":"info","namespace":"default","uri":"extension://kms/payloads","time":"2026-10-06T14:34:38-04:00","message":"Registering crypto key"} +{"level":"info","component":"metrics","addr":":9090","time":"2026-10-06T14:34:38-04:00","message":"Starting metrics server"} +{"level":"warn","addr":"127.0.0.1:7234","time":"2026-10-06T14:34:38-04:00","message":"Running with insecure credentials. Configure TLS for production use."} +{"level":"info","addr":"127.0.0.1:7234","time":"2026-10-06T14:34:38-04:00","message":"Starting the server"} +``` + +The `Running with insecure credentials` warning is expected, not a sign anything is broken: it describes the local +gateway, the plaintext hop between the Worker/starter and the proxy on this machine. It says nothing about whether +payloads get sealed or about the TLS connection to the extension server; both of those are unaffected. Terminal 4, in `examples/kms`, starts the Worker: diff --git a/examples/listnamespace/README.md b/examples/listnamespace/README.md index e7f6214..486ff0a 100644 --- a/examples/listnamespace/README.md +++ b/examples/listnamespace/README.md @@ -76,8 +76,8 @@ The first log line confirms the translation is armed: "message":"translating methods to the Cloud API"} ``` -You will also see `Running with insecure credentials` a couple of times. That is expected: it refers to the unix sockets -the proxy uses internally between its own tiers, which never leave the machine. Both hops to Cloud are TLS. +You will also see `Running with insecure credentials`. That is expected: it refers to the local gateway, which never +leaves the machine. Both hops to Cloud are TLS. ## Test it diff --git a/internal/config/cloudapi.go b/internal/config/cloudapi.go index 6464835..41756fb 100644 --- a/internal/config/cloudapi.go +++ b/internal/config/cloudapi.go @@ -43,10 +43,10 @@ type APITranslations struct { // // See https://docs.temporal.io/ops. // -// It is deliberately not an entry in Upstreams: an upstream is a server - a -// socket, a proxy.Server, and a routing destination - and the control plane is -// only ever a client connection. Declaring it there would give it three things -// it cannot use and one it should not have: routability. +// It is deliberately not an entry in Upstreams: an upstream is a forwarder and +// a routing destination, and the control plane is only ever a client +// connection. Declaring it there would give it a forwarder it cannot use and +// routability it should not have. type CloudAPI struct { Listen ListenConfig `yaml:",inline"` Credentials *CredentialConfig `yaml:"credentials"` diff --git a/internal/dataplane/abort_test.go b/internal/dataplane/abort_test.go index 99cfc65..95a6fa4 100644 --- a/internal/dataplane/abort_test.go +++ b/internal/dataplane/abort_test.go @@ -13,14 +13,14 @@ import ( "github.com/temporalio/temporal-proxy/pkg/logger" ) -// TestAbortFiresOnceWhenServingStopsUnexpectedly closes the listeners out from -// under the serving goroutines, which is the only way into the unexpected-exit +// TestAbortFiresWhenServingStopsUnexpectedly closes the gateway's listener out +// from under its serving goroutine, which is the only way into the unexpected-exit // path: Stop and cancelling the serving context are both clean shutdowns, and a // taken port fails before serving starts. -func TestAbortFiresOnceWhenServingStopsUnexpectedly(t *testing.T) { +func TestAbortFiresWhenServingStopsUnexpectedly(t *testing.T) { t.Parallel() - aborts := make(chan error, 2) + aborts := make(chan error, 1) cfg := abortConfig(t) d := newTestDeps(t, cfg) @@ -37,13 +37,8 @@ func TestAbortFiresOnceWhenServingStopsUnexpectedly(t *testing.T) { t.Cleanup(func() { _ = dp.Stop(context.WithoutCancel(t.Context())) }) listeners := dp.Listeners() - require.Len(t, listeners, 2, "one upstream socket and the gateway") - require.Equal(t, "unix", listeners[0].Addr().Network(), "upstream sockets bind before the gateway") - require.Equal(t, "tcp", listeners[1].Addr().Network()) - - for _, lis := range listeners { - require.NoError(t, lis.Close()) - } + require.Len(t, listeners, 1, "only the gateway binds") + require.NoError(t, listeners[0].Close()) select { case err := <-aborts: @@ -52,14 +47,6 @@ func TestAbortFiresOnceWhenServingStopsUnexpectedly(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("Abort was not called after serving stopped") } - - // Both tiers stopped serving, but a caller is told once: the first report is - // the one that brings the process down. - select { - case err := <-aborts: - t.Fatalf("Abort fired more than once: %v", err) - case <-time.After(500 * time.Millisecond): - } } func TestAbortIsOptional(t *testing.T) { @@ -87,10 +74,8 @@ func TestAbortIsOptional(t *testing.T) { ) } -// abortConfig points the only upstream at a template, so the plane binds a -// socket and serves without anything having to be reachable during Start. The -// socket path is derived from the hostPort, so naming the test in it keeps -// parallel planes off each other's sockets. +// abortConfig points the only upstream at a template, so the plane serves +// without anything having to be reachable during Start. func abortConfig(t *testing.T) *config.Config { t.Helper() diff --git a/internal/dataplane/dataplane.go b/internal/dataplane/dataplane.go index f00e236..735432a 100644 --- a/internal/dataplane/dataplane.go +++ b/internal/dataplane/dataplane.go @@ -23,7 +23,6 @@ import ( "github.com/temporalio/temporal-proxy/internal/server" "github.com/temporalio/temporal-proxy/internal/services" "github.com/temporalio/temporal-proxy/internal/transport/connect" - "github.com/temporalio/temporal-proxy/internal/transport/socket" "github.com/temporalio/temporal-proxy/internal/version" "github.com/temporalio/temporal-proxy/pkg/crypto" "github.com/temporalio/temporal-proxy/pkg/logger" @@ -37,7 +36,6 @@ type ( ctx context.Context gateway *server.Server hostPort string - upstreams []*upstreamTier ready []*connect.Conn codecs *proxy.Codecs abort func(error) @@ -76,20 +74,11 @@ type ( key string } - - // upstreamTier is one upstream's forwarder, served in process to the gateway - // and on the socket it binds. - upstreamTier struct { - name string - path string - svr *proxy.Server - fw *proxy.Forwarder - } ) -// New validates cfg in full, compiles the routing table, derives each upstream's -// socket path once, and builds both tiers. ctx is long lived and drives each -// tier's health check; the context passed to Start bounds startup only. Neither +// New validates cfg in full, compiles the routing table, and builds the gateway +// and every upstream's forwarder. ctx is long lived and drives the gateway's +// health check; the context passed to Start bounds startup only. Neither // stops serving, which only Stop does. New binds nothing and dials nothing. Every // Prometheus collector is registered here, so New must be called once per // registry. @@ -167,24 +156,18 @@ func New(ctx context.Context, cfg *config.Config, opts ...Option) (*Dataplane, e for i := range cfg.Upstreams { up := &cfg.Upstreams[i] - path, err := socket.UnixPath(up.Listen.HostPort) - if err != nil { - return nil, fmt.Errorf("failed to resolve proxy socket path[%q]: %w", up.Name, err) - } - - tier, ready, err := newUpstreamTier(cfg, o, up, path, codecs) + fw, ready, err := newUpstreamForwarder(cfg, o, up, codecs) if err != nil { return nil, err } - dp.upstreams = append(dp.upstreams, tier) if ready != nil { dp.ready = append(dp.ready, ready) } // The gateway calls this upstream's forwarder directly with the stream it - // accepted. The socket stays for local workers that dial it. - handlers[up.Name] = tier.fw.Handle + // accepted. + handlers[up.Name] = fw.Handle } handler := router.Handler( @@ -268,8 +251,8 @@ func WithVault(v *crypto.Vault) Option { return Option(func(o *options) { o.vault = v }) } -// WithLogger sets the logger used by the dataplane and both tiers, defaulting to -// [logger.Default]. A nil logger keeps the default, so an absent optional +// WithLogger sets the logger used by the gateway and every upstream, defaulting +// to [logger.Default]. A nil logger keeps the default, so an absent optional // dependency can be passed straight through. func WithLogger(log logger.Logger) Option { return Option(func(o *options) { @@ -280,8 +263,8 @@ func WithLogger(log logger.Logger) Option { } // WithAbort sets a function called at most once, from the goroutine that was -// serving, when a tier stops for a reason other than Stop. It must not block and -// must not call back into the Dataplane. +// serving, when the gateway stops for a reason other than Stop. It must not +// block and must not call back into the Dataplane. func WithAbort(fn func(error)) Option { return Option(func(o *options) { o.abort = fn }) } @@ -300,18 +283,6 @@ func (d *Dataplane) Codecs() *proxy.Codecs { return d.codecs } -// SocketPath is the unix path the named upstream's proxy binds and the gateway -// dials. It is the single derivation of that path. -func (d *Dataplane) SocketPath(upstream string) (string, error) { - for _, up := range d.upstreams { - if up.name == upstream { - return up.path, nil - } - } - - return "", fmt.Errorf("dataplane: no upstream named %q", upstream) -} - // validate reports the first required dependency that is missing, by field // name, so a wiring mistake fails at construction rather than as a nil // dereference on the first request. @@ -338,16 +309,16 @@ func validate(ctx context.Context, cfg *config.Config, o *options) error { return nil } -// newUpstreamTier builds one upstream's proxy. It returns the tier, the -// connection to open eagerly at start (nil for a templated upstream, which -// resolves per request and has nothing to open yet), and any error. -func newUpstreamTier( +// newUpstreamForwarder builds one upstream's forwarder. It returns the +// forwarder, the connection to open eagerly at start (nil for a templated +// upstream, which resolves per request and has nothing to open yet), and any +// error. +func newUpstreamForwarder( cfg *config.Config, o *options, up *config.Upstream, - path string, codecs *proxy.Codecs, -) (*upstreamTier, *connect.Conn, error) { +) (*proxy.Forwarder, *connect.Conn, error) { // Request-independent dial options: namespace translation and outbound // credentials. Per-request credentials are added by the resolver. var dialOpts []grpc.DialOption @@ -428,16 +399,6 @@ func newUpstreamTier( return nil, nil, err } - svr, err := proxy.New( - up.Listen.HostPort, - fw, - proxy.WithLogger(o.logger), - proxy.WithSocketPath(path), - ) - if err != nil { - return nil, nil, fmt.Errorf("failed to create proxy for upstream %q: %w", up.Name, err) - } - // Only a static upstream holds a connection worth opening before serving; a // templated one resolves its target per request. var ready *connect.Conn @@ -445,7 +406,7 @@ func newUpstreamTier( ready = conn } - return &upstreamTier{name: up.Name, path: path, svr: svr, fw: fw}, ready, nil + return fw, ready, nil } // Resolve returns the overriding cache key with the wrapped resolver's target and diff --git a/internal/dataplane/dataplanetest/dataplanetest.go b/internal/dataplane/dataplanetest/dataplanetest.go index b49d20a..ed16dc2 100644 --- a/internal/dataplane/dataplanetest/dataplanetest.go +++ b/internal/dataplane/dataplanetest/dataplanetest.go @@ -100,18 +100,6 @@ func DeadUpstream(t *testing.T) string { return addr } -// DialUnix returns a client connection to the unix socket at path, closed when -// the test ends. -func DialUnix(t *testing.T, path string) *grpc.ClientConn { - t.Helper() - - conn, err := grpc.NewClient("unix://"+path, grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - t.Cleanup(func() { _ = conn.Close() }) - - return conn -} - // Start constructs a dataplane directly, the way the production fx module // does, and starts it. Every request is admitted unless [WithAuth] says // otherwise, and no vault is built, so a cfg configuring inbound auth or @@ -255,24 +243,6 @@ func (f *Fixture) Context() context.Context { return ctx } -// UpstreamConn is a client connection to the named upstream's own unix socket, -// the path a local worker bypassing the gateway would dial. -func (f *Fixture) UpstreamConn(name string) *grpc.ClientConn { - f.t.Helper() - - return DialUnix(f.t, f.SocketPath(name)) -} - -// SocketPath is the unix socket path the named upstream's proxy binds. -func (f *Fixture) SocketPath(name string) string { - f.t.Helper() - - path, err := f.dp.SocketPath(name) - require.NoError(f.t, err) - - return path -} - // applyDefaults fills in the fields every case would otherwise repeat. Routing // is deliberately untouched: an empty DefaultUpstream is indistinguishable from // an unset one, so filling it would quietly make a test of the unroutable path diff --git a/internal/dataplane/dataplanetest/dataplanetest_test.go b/internal/dataplane/dataplanetest/dataplanetest_test.go index 5c8b544..e7ccc28 100644 --- a/internal/dataplane/dataplanetest/dataplanetest_test.go +++ b/internal/dataplane/dataplanetest/dataplanetest_test.go @@ -104,18 +104,3 @@ func TestStartAppBuildsConfigDrivenCollaborators(t *testing.T) { _, err = f.Client().GetSystemInfo(bad, &workflowservice.GetSystemInfoRequest{}) require.Equal(t, codes.Unauthenticated, status.Code(err)) } - -func TestUpstreamConnReachesTheProxySocketDirectly(t *testing.T) { - t.Parallel() - - up := dataplanetest.NewUpstream(t) - f := dataplanetest.Start(t, dataplanetest.Config(up)) - - // The per-upstream socket is a supported access path for local workers, so - // it must serve without going through the gateway. - client := workflowservice.NewWorkflowServiceClient(f.UpstreamConn(dataplanetest.DefaultUpstream)) - - _, err := client.GetSystemInfo(f.Context(), &workflowservice.GetSystemInfoRequest{}, grpc.WaitForReady(true)) - require.NoError(t, err) - require.NotNil(t, up.Metadata()) -} diff --git a/internal/dataplane/dataplanetest/upstream.go b/internal/dataplane/dataplanetest/upstream.go index 26190d4..a3c5ab5 100644 --- a/internal/dataplane/dataplanetest/upstream.go +++ b/internal/dataplane/dataplanetest/upstream.go @@ -175,9 +175,7 @@ func (u *Upstream) record(ctx context.Context, req proto.Message) { } } -// newUpstream serves a fake frontend on an ephemeral loopback port. The -// ephemeral port also keeps the socket path the proxy derives from it unique -// across parallel tests. +// newUpstream serves a fake frontend on an ephemeral loopback port. func newUpstream(t *testing.T, creds credentials.TransportCredentials) *Upstream { t.Helper() diff --git a/internal/dataplane/inprocess_test.go b/internal/dataplane/inprocess_test.go deleted file mode 100644 index 54e801a..0000000 --- a/internal/dataplane/inprocess_test.go +++ /dev/null @@ -1,29 +0,0 @@ -package dataplane_test - -import ( - "os" - "testing" - - "github.com/stretchr/testify/require" - "go.temporal.io/api/workflowservice/v1" - - "github.com/temporalio/temporal-proxy/internal/dataplane/dataplanetest" -) - -func TestGatewayForwardsWithoutTheUpstreamSocket(t *testing.T) { - t.Parallel() - - // Guard: the gateway serves each upstream in process rather than over the - // upstream's unix socket. Both ends of that connection would be in this - // process, and a burst of cancellations under load can leave gRPC's reader - // and writer blocked on each other at both ends at once, hanging every - // request the gateway forwards. With the socket file gone, a gateway that - // still dialled it could not forward at all. - up := dataplanetest.NewUpstream(t) - f := dataplanetest.Start(t, dataplanetest.Config(up)) - - require.NoError(t, os.Remove(f.SocketPath(dataplanetest.DefaultUpstream))) - - _, err := f.Client().GetSystemInfo(f.Context(), &workflowservice.GetSystemInfoRequest{}) - require.NoError(t, err) -} diff --git a/internal/dataplane/lifecycle.go b/internal/dataplane/lifecycle.go index eab6b22..d12dda5 100644 --- a/internal/dataplane/lifecycle.go +++ b/internal/dataplane/lifecycle.go @@ -5,33 +5,17 @@ import ( "errors" "fmt" "net" - "sync" "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/pkg/logger/tag" ) -// Start binds and serves every upstream socket, opens every static upstream -// connection so an unreachable one fails startup, then binds and serves the -// gateway, in that order. It returns once the gateway is accepting. ctx bounds +// Start opens every static upstream connection so an unreachable one fails +// startup, then binds and serves the gateway. It returns once the gateway is accepting. ctx bounds // startup only and should carry a deadline, since it is what limits the wait for // an upstream to answer; the serving goroutines get the Context passed to New // instead. A failure part-way through stops whatever already started. func (d *Dataplane) Start(ctx context.Context) error { - for _, up := range d.upstreams { - // Bind synchronously so the socket is listening before the gateway routes - // anything to it, then serve in the background. - lis, err := up.svr.Listen(ctx) - if err != nil { - return d.rollback(ctx, fmt.Errorf("failed to start proxy for upstream %q: %w", up.name, err)) - } - - d.track(lis) - d.serve(fmt.Sprintf("upstream %q", up.name), func() error { - return up.svr.Start(d.ctx, lis) - }) - } - // A static upstream's connection is created during New, but gRPC does not // open a socket until it is used, so open them here: an unreachable upstream // fails startup instead of surfacing as request errors once the gateway is @@ -57,11 +41,8 @@ func (d *Dataplane) Start(ctx context.Context) error { return nil } -// Stop drains the gateway first, so no request is admitted for a tier that is -// going away, then every upstream proxy. Each tier's drain is bounded, so the -// upstreams go concurrently: they are independent, and serially their budgets -// would sum, which is how a shutdown overruns the lifecycle deadline and strands -// the hooks queued behind this one. +// Stop drains the gateway within its shutdown budget and closes every listener +// Start bound. func (d *Dataplane) Stop(ctx context.Context) error { d.mu.Lock() d.stopping = true @@ -74,22 +55,6 @@ func (d *Dataplane) Stop(ctx context.Context) error { errs = append(errs, err) } - // Indexed rather than appended, so the slot is written without coordinating - // and the report stays in upstream order. - upstreamErrs := make([]error, len(d.upstreams)) - - var wg sync.WaitGroup - for i, up := range d.upstreams { - wg.Go(func() { - if err := up.svr.Stop(ctx); err != nil { - upstreamErrs[i] = fmt.Errorf("upstream %q: %w", up.name, err) - } - }) - } - - wg.Wait() - errs = append(errs, upstreamErrs...) - // A graceful stop closes the listeners its server was serving on, but one // bound by a Start that failed before its goroutine reached Serve is not // among them. Closing here is what keeps that case from leaking a socket. diff --git a/internal/dataplane/lifecycle_test.go b/internal/dataplane/lifecycle_test.go index 5d05f5a..2f65641 100644 --- a/internal/dataplane/lifecycle_test.go +++ b/internal/dataplane/lifecycle_test.go @@ -7,6 +7,8 @@ import ( "time" "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/health/grpc_health_v1" "github.com/temporalio/temporal-proxy/internal/config" @@ -14,21 +16,12 @@ import ( "github.com/temporalio/temporal-proxy/internal/dataplane/dataplanetest" ) -func TestStartBindsUpstreamSocketsBeforeTheGatewayAccepts(t *testing.T) { +func TestStartReturnsOnceTheGatewayAccepts(t *testing.T) { t.Parallel() cfg := liveConfig(t) dp := startPlane(t, newTestDeps(t, cfg)) - // Every upstream socket must already accept by the time Start returns; the - // gateway routes to them immediately. - path, err := dp.SocketPath("primary") - require.NoError(t, err) - - conn, err := net.Dial("unix", path) - require.NoError(t, err, "upstream socket must be accepting once Start returns") - require.NoError(t, conn.Close()) - require.NotNil(t, dp.Addr()) gw, err := net.Dial("tcp", dp.Addr().String()) @@ -69,10 +62,8 @@ func TestStartFailsOnUnreachableStaticUpstream(t *testing.T) { "an unreachable static upstream must fail startup", ) - // A failed Start rolls back: the gateway never bound, and the upstream - // socket it did bind is closed again rather than left listening. + // A failed Start rolls back: the gateway never bound. require.Nil(t, dp.Addr()) - requireNotServing(t, dp, "primary") } func TestStartFailsWhenTheGatewayPortIsTaken(t *testing.T) { @@ -82,7 +73,7 @@ func TestStartFailsWhenTheGatewayPortIsTaken(t *testing.T) { first := startPlane(t, newTestDeps(t, cfg)) // A second plane on the address the first is already accepting on cannot - // bind, so its Start fails after its own upstream sockets are up. + // bind, so its Start fails after its upstream connections are open. cfg = liveConfig(t) cfg.Listen.HostPort = first.Addr().String() @@ -95,13 +86,9 @@ func TestStartFailsWhenTheGatewayPortIsTaken(t *testing.T) { require.ErrorContains(t, second.Start(ctx), "failed to create listener") require.Nil(t, second.Addr()) - requireNotServing(t, second, "primary") // Rolling the second plane back must not disturb the first. - path, err := first.SocketPath("primary") - require.NoError(t, err) - - conn, err := net.Dial("unix", path) + conn, err := net.Dial("tcp", first.Addr().String()) require.NoError(t, err, "the first plane must still be serving") require.NoError(t, conn.Close()) } @@ -147,17 +134,6 @@ func TestStopWithoutStart(t *testing.T) { require.NoError(t, dp.Stop(t.Context()), "Stop before Start has nothing to drain") } -func TestSocketPathUnknownUpstream(t *testing.T) { - t.Parallel() - - cfg := liveConfig(t) - dp, err := dataplane.New(t.Context(), cfg, newTestDeps(t, cfg).opts()...) - require.NoError(t, err) - - _, err = dp.SocketPath("nope") - require.ErrorContains(t, err, "nope") -} - func TestStartWithTemplatedAndStaticUpstreams(t *testing.T) { t.Parallel() @@ -169,37 +145,16 @@ func TestStartWithTemplatedAndStaticUpstreams(t *testing.T) { }) // Nothing is listening for the templated upstream, which is the point: it - // resolves per request, so it is excluded from the readiness wait and still - // gets a socket of its own. + // resolves per request, so it is excluded from the readiness wait and Start + // still succeeds. dp := startPlane(t, newTestDeps(t, cfg)) - - for _, name := range []string{"primary", "templated"} { - path, err := dp.SocketPath(name) - require.NoError(t, err) - - conn, err := net.Dial("unix", path) - require.NoError(t, err, "upstream %q must be accepting once Start returns", name) - require.NoError(t, conn.Close()) - } + require.NotNil(t, dp.Addr()) } -// TestStopClosesEveryUpstreamSocket proves a normal Stop tears down every -// upstream tier, not just the one Start bound most recently: each of two -// upstreams answers a real health check before Stop and refuses connections -// after it. -func TestStopClosesEveryUpstreamSocket(t *testing.T) { +func TestStopClosesTheGateway(t *testing.T) { t.Parallel() - cfg := testConfig() - // Neither upstream is named "primary", so the default upstream testConfig - // wires up must be cleared: nothing here routes through the gateway, but a - // stale reference still fails Config.Validate. - cfg.Routing = config.Routing{} - cfg.Upstreams = config.UpstreamList{ - {Name: "a", Listen: dataplanetest.NewUpstream(t).Listen()}, - {Name: "b", Listen: dataplanetest.NewUpstream(t).Listen()}, - } - + cfg := liveConfig(t) dp, err := dataplane.New(t.Context(), cfg, newTestDeps(t, cfg).opts()...) require.NoError(t, err) @@ -207,27 +162,28 @@ func TestStopClosesEveryUpstreamSocket(t *testing.T) { defer cancel() require.NoError(t, dp.Start(ctx)) - for _, name := range []string{"a", "b"} { - path, err := dp.SocketPath(name) - require.NoError(t, err) + addr := dp.Addr().String() + conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) - resp, err := grpc_health_v1.NewHealthClient(dataplanetest.DialUnix(t, path)).Check( - t.Context(), &grpc_health_v1.HealthCheckRequest{}, - ) - require.NoError(t, err, "upstream %q must be serving before Stop", name) - require.Equal(t, grpc_health_v1.HealthCheckResponse_SERVING, resp.GetStatus()) - } + resp, err := grpc_health_v1.NewHealthClient(conn).Check( + t.Context(), &grpc_health_v1.HealthCheckRequest{}, + ) + require.NoError(t, err, "the gateway must be serving before Stop") + require.Equal(t, grpc_health_v1.HealthCheckResponse_SERVING, resp.GetStatus()) require.NoError(t, dp.Stop(context.WithoutCancel(t.Context()))) - for _, name := range []string{"a", "b"} { - requireNotServing(t, dp, name) + gw, err := net.Dial("tcp", addr) + if err == nil { + _ = gw.Close() + t.Fatalf("the gateway is still accepting on %s", addr) } } // liveConfig is testConfig pointed at an upstream that is actually listening, -// which Start must reach before the gateway binds. The ephemeral port also -// keeps the derived unix socket path unique across parallel tests. +// which Start must reach before the gateway binds. func liveConfig(t *testing.T) *config.Config { t.Helper() @@ -256,19 +212,3 @@ func startPlane(t *testing.T, d testDeps) *dataplane.Dataplane { return dp } - -// requireNotServing asserts the named upstream's socket is gone, which is how a -// rolled-back Start or a completed Stop proves it leaked neither a listener nor -// a socket file. -func requireNotServing(t *testing.T, dp *dataplane.Dataplane, upstream string) { - t.Helper() - - path, err := dp.SocketPath(upstream) - require.NoError(t, err) - - conn, err := net.Dial("unix", path) - if err == nil { - _ = conn.Close() - t.Fatalf("upstream %q is still accepting on %s", upstream, path) - } -} diff --git a/internal/metrics/doc.go b/internal/metrics/doc.go index f05873f..887cf81 100644 --- a/internal/metrics/doc.go +++ b/internal/metrics/doc.go @@ -6,7 +6,7 @@ // // A reporter emitting while a request is in flight also takes a // [MetadataLabels], built from the configured header-to-name pairs, and carries -// those labels on its collectors. Values are read from the request's incoming metadata at each emit -// rather than resolved once and carried along, because the proxy forwards over a -// socket that context values do not cross while metadata does. +// those labels on its collectors. Values are read from the request's incoming +// metadata at each emit, which every hop's context already carries, so nothing +// has to resolve and stash them up front. package metrics diff --git a/internal/proxy/doc.go b/internal/proxy/doc.go index f93c347..9a44a4c 100644 --- a/internal/proxy/doc.go +++ b/internal/proxy/doc.go @@ -1,5 +1,4 @@ -// Package proxy serves every allowlisted service on a local unix socket, -// forwarding each request to an upstream Temporal Service over gRPC. The -// socket path is derived from the upstream host:port, so local workers connect -// without TLS while the upstream hop stays secured. +// Package proxy forwards every allowlisted service to an upstream Temporal +// Service over gRPC, applying namespace translation and payload codecs on the +// way. package proxy diff --git a/internal/proxy/encryption_test.go b/internal/proxy/encryption_test.go index 4ae0525..17aabe0 100644 --- a/internal/proxy/encryption_test.go +++ b/internal/proxy/encryption_test.go @@ -230,8 +230,8 @@ func TestEncryptionRecordsVaultOpsWithMetadataLabels(t *testing.T) { require.NoError(t, err) // The namespace reaches this hop as outgoing metadata, but a label is read - // from the incoming metadata the gateway forwarded over the socket, which is - // the only reason labels work on this side at all. + // from the incoming metadata, which the forwarder's call inherits from the + // gateway's request context. ctx := metadata.AppendToOutgoingContext(t.Context(), meta.NamespaceHeader, "ns1") ctx = metadata.NewIncomingContext(ctx, metadata.Pairs("x-tenant", "acme")) diff --git a/internal/proxy/forwarding_test.go b/internal/proxy/forwarding_test.go index cd4588a..851d9fa 100644 --- a/internal/proxy/forwarding_test.go +++ b/internal/proxy/forwarding_test.go @@ -1,24 +1,27 @@ package proxy_test -// These tests drive real requests through a real [proxy.Server] over its unix -// socket, so they use only the exported surface and live in the external test +// These tests drive real requests through a real [proxy.Forwarder] served over +// loopback TCP, so they use only the exported surface and live in the external test // package. forward_test.go stays in package proxy for the unit tests that need // the forwarder's unexported internals. import ( "context" "io" + "net" "testing" "github.com/stretchr/testify/require" workflowservice "go.temporal.io/api/workflowservice/v1" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/metadata" "google.golang.org/grpc/reflection" "google.golang.org/grpc/reflection/grpc_reflection_v1" "google.golang.org/grpc/status" + "github.com/temporalio/temporal-proxy/internal/proxy" "github.com/temporalio/temporal-proxy/internal/services" ) @@ -59,7 +62,7 @@ func TestStreamForwardsBidiReflection(t *testing.T) { // ServerReflectionInfo is the only streaming method across every forwardable // service, so it is the only way to exercise the streaming path at all. The // upstream is the sole reflection provider (the proxy's local server registers - // only the health service), so a response naming WorkflowService proves the + // nothing), so a response naming WorkflowService proves the // stream reached it and came back. addr := serveUpstream(t, func(s *grpc.Server) { workflowservice.RegisterWorkflowServiceServer(s, &metadataStampingService{}) @@ -135,3 +138,60 @@ func (*metadataStampingService) GetSystemInfo( return &workflowservice.GetSystemInfoResponse{}, nil } + +// forwarder builds a forwarder for the named services over a plain client conn +// to upstream, standing in for the pool-backed connection used in production. +func forwarder(t *testing.T, upstream string, allowed ...string) *proxy.Forwarder { + t.Helper() + + if len(allowed) == 0 { + allowed = services.Default() + } + + conn, err := grpc.NewClient(upstream, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + fw, err := proxy.NewForwarder(conn, services.NewAllowlist(allowed)) + require.NoError(t, err) + + return fw +} + +// serveUpstream starts a plaintext gRPC server on a loopback port, registers any +// services supplied, and returns its address for use as an upstream hostPort. +func serveUpstream(t *testing.T, register ...func(*grpc.Server)) string { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + svr := grpc.NewServer() + for _, reg := range register { + reg(svr) + } + + go func() { _ = svr.Serve(lis) }() + t.Cleanup(svr.Stop) + + return lis.Addr().String() +} + +// startProxy serves a forwarder for the named services to upstream on a +// loopback port and returns a client connection to it. +func startProxy(t *testing.T, upstream string, allowed ...string) *grpc.ClientConn { + t.Helper() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + svr := grpc.NewServer(grpc.UnknownServiceHandler(forwarder(t, upstream, allowed...).Handle)) + go func() { _ = svr.Serve(lis) }() + t.Cleanup(svr.Stop) + + conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go deleted file mode 100644 index 8109aaa..0000000 --- a/internal/proxy/server.go +++ /dev/null @@ -1,120 +0,0 @@ -package proxy - -import ( - "context" - "errors" - "fmt" - "net" - "os" - - "github.com/temporalio/temporal-proxy/internal/server" - "github.com/temporalio/temporal-proxy/internal/transport/creds" - "github.com/temporalio/temporal-proxy/internal/transport/socket" - "github.com/temporalio/temporal-proxy/pkg/logger" -) - -type ( - // Server proxies the Temporal WorkflowService. It re-serves an upstream - // Temporal Service 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 - path string // path to unix socket - } - - // Options configures a [Server] at construction time. - Options struct { - logger logger.Logger - socketPath string - } - - // Option configures a [Server] via [New]. - Option func(*Options) -) - -// New constructs a Server that hands every inbound method to fw, which forwards -// it to the upstream fw was built against. The local listener is a unix socket -// whose path is derived from hostPort. The connection(s) fw forwards over are -// owned by the shared pool, not by this Server. -func New(hostPort string, fw *Forwarder, opts ...Option) (*Server, error) { - pops := &Options{logger: logger.Default()} - for _, opt := range opts { - opt(pops) - } - - svr, err := server.New( - // NB: Hosting on local unix port, no need for TLS here. - server.WithCredentials(creds.NewListener(creds.Insecure())), - server.WithLogger(pops.logger), - server.WithUnknownServiceHandler(fw.Handle), - ) - if err != nil { - return nil, fmt.Errorf("failed to create proxy: %s, %w", hostPort, err) - } - - path := pops.socketPath - if path == "" { - p, err := socket.UnixPath(hostPort) - if err != nil { - return nil, fmt.Errorf("failed to resolve socket path: %w", err) - } - - path = p - } else if err := socket.ValidatePath(path); err != nil { - return nil, fmt.Errorf("invalid socket path: %w", err) - } - - return &Server{svr: svr, path: path}, nil -} - -// WithLogger sets the logger used by the proxy. -func WithLogger(log logger.Logger) Option { - return Option(func(o *Options) { o.logger = log }) -} - -// WithSocketPath sets the unix socket path the proxy binds, overriding the one -// derived from hostPort. A caller that also dials this socket passes the same -// value to both sides so the two cannot disagree. [New] rejects a path that -// exceeds the platform's sun_path limit. -func WithSocketPath(path string) Option { - return Option(func(o *Options) { o.socketPath = path }) -} - -// 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 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 nil, fmt.Errorf("failed to bind to socket: unix://%s, %w", s.path, err) - } - - return lis, nil -} - -// Start serves on lis until Stop is called; ctx is not what stops it, and is -// used only to drive the periodic health check. 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) -} - -// Stop shuts the proxy down, draining in-flight RPCs within the server's -// shutdown budget and dropping whatever is left. -func (s *Server) Stop(ctx context.Context) error { - if err := s.svr.Stop(ctx); err != nil { - return fmt.Errorf("failed to stop GRPC server: %w", err) - } - - return nil -} diff --git a/internal/proxy/server_test.go b/internal/proxy/server_test.go deleted file mode 100644 index b4f6bcd..0000000 --- a/internal/proxy/server_test.go +++ /dev/null @@ -1,288 +0,0 @@ -package proxy_test - -import ( - "context" - "net" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "github.com/stretchr/testify/require" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/health/grpc_health_v1" - - "github.com/temporalio/temporal-proxy/internal/proxy" - "github.com/temporalio/temporal-proxy/internal/services" - "github.com/temporalio/temporal-proxy/internal/transport/socket" - "github.com/temporalio/temporal-proxy/pkg/logger" -) - -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", forwarder(t, "127.0.0.1:7233")) - require.NoError(t, err) - require.NotNil(t, svr) - }) -} - -func TestServerStartAndStop(t *testing.T) { - t.Parallel() - - // A unique upstream host gives this test its own socket path so it can run in - // parallel with the others. The upstream is never dialed: the health service - // the proxy serves locally answers the Check below. - const upstream = "127.0.0.1:17233" - - log := logger.NewTestLogger() - svr, err := proxy.New(upstream, forwarder(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, lis) }() - - conn := dialUnix(t, upstream) - defer func() { _ = conn.Close() }() - - resp, err := grpc_health_v1.NewHealthClient(conn).Check( - t.Context(), - &grpc_health_v1.HealthCheckRequest{}, - grpc.WaitForReady(true), - ) - require.NoError(t, err) - require.Equal(t, grpc_health_v1.HealthCheckResponse_SERVING, resp.GetStatus()) - - // The supplied logger reaches the underlying server. - require.True(t, log.Contains("Starting the server"), "expected the injected logger to be used") - - require.NoError(t, svr.Stop(t.Context())) - - select { - case err := <-errCh: - require.NoError(t, err) - case <-time.After(time.Second): - t.Fatal("server did not stop after shutdown") - } -} - -func TestStartRemovesStaleSocket(t *testing.T) { - t.Parallel() - - // An ephemeral port gives this test its own socket path. A fixed one is shared - // with every past run on the machine, so a socket left behind by a run that was - // killed before it could shut down would occupy the path and make planting the - // stale socket below fail with "operation not supported on socket". - upstream := deadUpstream(t) - - path, err := socket.UnixPath(upstream) - require.NoError(t, err) - - // Clear the path first: a run killed before its cleanup leaves its own socket - // or directory behind, and planting over either fails. - require.NoError(t, os.RemoveAll(path)) - - // Leave a real socket behind, which is what a killed process leaves. Without - // removal the bind would fail with "address already in use" and the Check never - // succeeds. - stale, err := net.Listen("unix", path) - require.NoError(t, err) - unix, ok := stale.(*net.UnixListener) - require.True(t, ok) - unix.SetUnlinkOnClose(false) - require.NoError(t, unix.Close()) - t.Cleanup(func() { _ = os.Remove(path) }) - require.FileExists(t, path, "expected a stale socket to be planted") - - svr, err := proxy.New(upstream, forwarder(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, lis) }() - - conn := dialUnix(t, upstream) - defer func() { _ = conn.Close() }() - - _, err = grpc_health_v1.NewHealthClient(conn).Check( - t.Context(), - &grpc_health_v1.HealthCheckRequest{}, - grpc.WaitForReady(true), - ) - require.NoError(t, err) - - require.NoError(t, svr.Stop(t.Context())) - require.NoError(t, <-errCh) -} - -func TestNewWithSocketPathOverridesDerivedPath(t *testing.T) { - t.Parallel() - - // A directory under os.TempDir() rather than t.TempDir() keeps the path - // short: t.TempDir() embeds the full test name, which here would push the - // socket path past the sun_path limit socket.UnixPath enforces. - dir, err := os.MkdirTemp("", "socket-override") - require.NoError(t, err) - t.Cleanup(func() { _ = os.RemoveAll(dir) }) - - want := filepath.Join(dir, "override.sock") - - svr, err := proxy.New("127.0.0.1:7233", forwarder(t, "127.0.0.1:7233"), proxy.WithSocketPath(want)) - require.NoError(t, err) - - lis, err := svr.Listen(t.Context()) - require.NoError(t, err) - t.Cleanup(func() { _ = lis.Close() }) - - require.Equal(t, want, lis.Addr().String()) -} - -func TestNewRejectsOverlongSocketPath(t *testing.T) { - t.Parallel() - - path := filepath.Join(os.TempDir(), strings.Repeat("d", 120)+".sock") - - _, err := proxy.New("127.0.0.1:7233", forwarder(t, "127.0.0.1:7233"), proxy.WithSocketPath(path)) - require.ErrorContains(t, err, "invalid socket path") - require.ErrorContains(t, err, "exceeds limit") -} - -func TestListenReturnsErrorWhenStaleSocketCannotBeRemoved(t *testing.T) { - t.Parallel() - - // As above, an ephemeral port keeps the path this test's own, so a leftover - // socket cannot make the Mkdir below fail with "file exists". - upstream := deadUpstream(t) - - path, err := socket.UnixPath(upstream) - require.NoError(t, err) - - // Clear the path first, as above. - require.NoError(t, os.RemoveAll(path)) - - // 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, forwarder(t, upstream)) - require.NoError(t, err) - - _, err = svr.Listen(t.Context()) - require.Error(t, err) - require.ErrorContains(t, err, "failed to remove stale socket") -} - -// forwarder returns the Forwarder for New's fw argument, allowing the services -// named or the default set when none are. gRPC dials lazily, so the underlying -// client conn opens no socket to upstream until a request needs it, which lets -// the tests that only exercise the local unix listener pass an upstream that was -// never started. A plain client conn stands in for the pool-backed resolvingConn -// used in production. -func forwarder(t *testing.T, upstream string, allowed ...string) *proxy.Forwarder { - t.Helper() - - if len(allowed) == 0 { - allowed = services.Default() - } - - conn, err := grpc.NewClient(upstream, grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - t.Cleanup(func() { _ = conn.Close() }) - - fw, err := proxy.NewForwarder(conn, services.NewAllowlist(allowed)) - require.NoError(t, err) - - return fw -} - -// serveUpstream starts a plaintext gRPC server on a loopback port, registers any -// services supplied, and returns its address for use as an upstream hostPort. A -// static upstream's connection is opened on start, so an upstream pointing at -// nothing fails the lifecycle. The ephemeral port also keeps the proxy's derived -// socket path unique across parallel tests. -func serveUpstream(t *testing.T, register ...func(*grpc.Server)) string { - t.Helper() - - lis, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - - svr := grpc.NewServer() - for _, reg := range register { - reg(svr) - } - - go func() { _ = svr.Serve(lis) }() - t.Cleanup(svr.Stop) - - return lis.Addr().String() -} - -// deadUpstream returns a loopback address with nothing behind it, by taking a -// port from the kernel and immediately giving it back. -func deadUpstream(t *testing.T) string { - t.Helper() - - lis, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - - addr := lis.Addr().String() - require.NoError(t, lis.Close()) - - return addr -} - -// startProxy runs a proxy forwarding the named services to upstream and returns a -// client connection to its local unix socket. Stop takes a fresh context because -// the test's own is already cancelled by the time cleanups run. -func startProxy(t *testing.T, upstream string, allowed ...string) *grpc.ClientConn { - t.Helper() - - svr, err := proxy.New(upstream, forwarder(t, upstream, allowed...)) - require.NoError(t, err) - - ctx := t.Context() - lis, err := svr.Listen(ctx) - require.NoError(t, err) - - go func() { _ = svr.Start(ctx, lis) }() - t.Cleanup(func() { _ = svr.Stop(context.Background()) }) - - conn := dialUnix(t, upstream) - 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.Listen binds. -func dialUnix(t *testing.T, upstream string) *grpc.ClientConn { - t.Helper() - - path, err := socket.UnixPath(upstream) - require.NoError(t, err) - - conn, err := grpc.NewClient( - "unix://"+path, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - require.NoError(t, err) - return conn -} diff --git a/internal/proxy/translation_test.go b/internal/proxy/translation_test.go index b83a9a6..8778727 100644 --- a/internal/proxy/translation_test.go +++ b/internal/proxy/translation_test.go @@ -20,7 +20,6 @@ import ( "github.com/temporalio/temporal-proxy/internal/protoutil" "github.com/temporalio/temporal-proxy/internal/services" - "github.com/temporalio/temporal-proxy/internal/transport/socket" ) type ( @@ -152,18 +151,18 @@ func TestOutboundNamespaceTranslation(t *testing.T) { fw, err := NewForwarder(cc, services.NewAllowlist(services.Default())) require.NoError(t, err) - svr, err := New(lis.Addr().String(), fw) + proxyLis, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) + svr := grpc.NewServer(grpc.UnknownServiceHandler(fw.Handle)) + go func() { _ = svr.Serve(proxyLis) }() + t.Cleanup(svr.Stop) - ctx := t.Context() - srvLis, err := svr.Listen(ctx) + conn, err := grpc.NewClient(proxyLis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) require.NoError(t, err) - go func() { _ = svr.Start(ctx, srvLis) }() - t.Cleanup(func() { _ = svr.Stop(context.Background()) }) - - conn := dialUnixSocket(t, lis.Addr().String()) defer func() { _ = conn.Close() }() + ctx := t.Context() + client := workflowservice.NewWorkflowServiceClient(conn) require.Eventually(t, func() bool { @@ -196,19 +195,3 @@ func local(s string) string { return "local-" + s } func namespaceInfo(name string) *namespacepb.NamespaceInfo { return &namespacepb.NamespaceInfo{Name: name} } - -// dialUnixSocket returns a client connection to the proxy's unix socket for the -// given upstream host. The socket path matches what proxy.Start binds. -func dialUnixSocket(t *testing.T, upstream string) *grpc.ClientConn { - t.Helper() - - path, err := socket.UnixPath(upstream) - require.NoError(t, err) - - conn, err := grpc.NewClient( - "unix://"+path, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - require.NoError(t, err) - return conn -} diff --git a/internal/transport/connect/fx.go b/internal/transport/connect/fx.go index 6e34d1e..7fff153 100644 --- a/internal/transport/connect/fx.go +++ b/internal/transport/connect/fx.go @@ -5,11 +5,11 @@ import "go.uber.org/fx" // Module provides a *Pool and binds its lifecycle to the application, closing // every pooled connection on shutdown via an fx stop hook. // -// The pool is deliberately not opened as a whole on start. It also holds the -// router's loopback connections to sockets this application binds itself, which -// do not exist until the proxy start hooks run; waiting on those here would -// block on work a later hook has yet to do. Opening eager connections is the job -// of whoever owns one, through [WaitReady]. +// The pool is deliberately not opened as a whole on start. It also holds +// connections that are lazy on purpose, such as a templated upstream's, created +// per request, and the Cloud control plane's, opened on first use; waiting on +// those here would hold up startup for connections nothing needs yet. Opening +// eager connections is the job of whoever owns one, through [WaitReady]. var Module = fx.Options( fx.Provide(NewPool), fx.Invoke(func(p *Pool, lc fx.Lifecycle) { diff --git a/internal/transport/socket/socket.go b/internal/transport/socket/socket.go deleted file mode 100644 index 47be17b..0000000 --- a/internal/transport/socket/socket.go +++ /dev/null @@ -1,54 +0,0 @@ -// Package socket defines the addressing contract for the proxy's local unix -// socket. UnixPath derives the path once, from the upstream host:port, and the -// caller hands that same value to both the proxy, which listens on it, and the -// dialer, which connects to it. It depends on no other internal packages. -package socket - -import ( - "crypto/sha256" - "encoding/hex" - "fmt" - "os" - "path/filepath" - "strings" -) - -// maxUnixPath is a conservative cap on a unix socket path length. The OS limit -// on sockaddr_un.sun_path is 104 bytes on macOS/BSD and 108 on Linux including -// the null terminator; 103 is safe across both. -const maxUnixPath = 103 - -// UnixPath derives a stable, absolute unix socket path for hostPort, placed -// under os.TempDir(). It is deterministic for a given hostPort within a -// process, so a caller that derives it once and passes the same string to both -// the listener and the dialer never has the two disagree. It returns an error -// when the resulting path would exceed the platform's sun_path limit, rather -// than letting the OS silently truncate it (which would break that agreement). -func UnixPath(hostPort string) (string, error) { - sum := sha256.Sum256([]byte(hostPort)) - hash := hex.EncodeToString(sum[:])[:8] - slug := strings.Map(func(r rune) rune { - if r == ':' || r == '.' || r == '/' { - return '-' - } - return r - }, hostPort) - - path := filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.sock", slug[:min(32, len(slug))], hash)) - if err := ValidatePath(path); err != nil { - return "", err - } - - return path, nil -} - -// ValidatePath returns an error when path would exceed the platform's sun_path -// limit. Callers that supply their own socket path instead of deriving one with -// [UnixPath] use it to fail fast rather than at bind time. -func ValidatePath(path string) error { - if len(path) > maxUnixPath { - return fmt.Errorf("unix socket path %q (%d bytes) exceeds limit of %d", path, len(path), maxUnixPath) - } - - return nil -} diff --git a/internal/transport/socket/socket_test.go b/internal/transport/socket/socket_test.go deleted file mode 100644 index 750c47a..0000000 --- a/internal/transport/socket/socket_test.go +++ /dev/null @@ -1,84 +0,0 @@ -package socket_test - -import ( - "os" - "path/filepath" - "strings" - "testing" - - "github.com/stretchr/testify/require" - - "github.com/temporalio/temporal-proxy/internal/transport/socket" -) - -func TestUnixPath(t *testing.T) { - t.Parallel() - - t.Run("is deterministic for the same host:port", func(t *testing.T) { - t.Parallel() - a, err := socket.UnixPath("localhost:7233") - require.NoError(t, err) - b, err := socket.UnixPath("localhost:7233") - require.NoError(t, err) - require.Equal(t, a, b) - }) - - t.Run("differs for different host:port", func(t *testing.T) { - t.Parallel() - a, err := socket.UnixPath("localhost:7233") - require.NoError(t, err) - b, err := socket.UnixPath("localhost:7234") - require.NoError(t, err) - require.NotEqual(t, a, b) - }) - - t.Run("is an absolute path under TempDir ending in .sock", func(t *testing.T) { - t.Parallel() - - got, err := socket.UnixPath("dns:///cloud.example.com:443") - require.NoError(t, err) - require.True(t, filepath.IsAbs(got), "expected absolute path, got %q", got) - require.Equal(t, filepath.Clean(os.TempDir()), filepath.Dir(got)) - require.True(t, strings.HasSuffix(got, ".sock"), "got %q", got) - - // The host:port separators are sanitized in the filename. - require.NotContains(t, filepath.Base(got), ":") - }) -} - -func TestValidatePath(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - path string - wantErr bool - }{ - {name: "short path", path: "/tmp/proxy.sock"}, - {name: "at the limit", path: "/" + strings.Repeat("d", 102)}, - {name: "one byte over the limit", path: "/" + strings.Repeat("d", 103), wantErr: true}, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - err := socket.ValidatePath(tt.path) - if !tt.wantErr { - require.NoError(t, err) - return - } - - require.ErrorContains(t, err, "exceeds limit") - }) - } -} - -func TestUnixPathRejectsOverlongPath(t *testing.T) { - // Not parallel: mutates TMPDIR, which os.TempDir reads. - t.Setenv("TMPDIR", "/"+strings.Repeat("d", 200)) - - _, err := socket.UnixPath("localhost:7233") - require.Error(t, err) - require.ErrorContains(t, err, "exceeds limit") -} From 499744f705bc68c7fa00ea9575a6452e95e19fc7 Mon Sep 17 00:00:00 2001 From: Vaughan Andrews Date: Tue, 6 Oct 2026 14:57:08 -0700 Subject: [PATCH 2/2] [proxy]: Drop the forwarder's allowlist check The gateway's router.Handler rejects services outside the allowlist before routing, and with the per-upstream socket gone it is the forwarder's only caller, so the forwarder's own check can never fire. Forwarder no longer takes an allowlist. --- internal/dataplane/dataplane.go | 2 +- internal/proxy/doc.go | 5 +- internal/proxy/forward.go | 38 ++++----- internal/proxy/forward_test.go | 121 ++++++++++++----------------- internal/proxy/forwarding_test.go | 22 +++--- internal/proxy/translation_test.go | 3 +- 6 files changed, 75 insertions(+), 116 deletions(-) diff --git a/internal/dataplane/dataplane.go b/internal/dataplane/dataplane.go index 735432a..d5fb793 100644 --- a/internal/dataplane/dataplane.go +++ b/internal/dataplane/dataplane.go @@ -394,7 +394,7 @@ func newUpstreamForwarder( return nil, nil, err } - fw, err := proxy.NewForwarder(conn, o.allowlist, proxy.WithProtoTypes(o.types)) + fw, err := proxy.NewForwarder(conn, proxy.WithProtoTypes(o.types)) if err != nil { return nil, nil, err } diff --git a/internal/proxy/doc.go b/internal/proxy/doc.go index 9a44a4c..3591299 100644 --- a/internal/proxy/doc.go +++ b/internal/proxy/doc.go @@ -1,4 +1,3 @@ -// Package proxy forwards every allowlisted service to an upstream Temporal -// Service over gRPC, applying namespace translation and payload codecs on the -// way. +// Package proxy forwards requests to an upstream Temporal Service over gRPC, +// applying namespace translation and payload codecs on the way. package proxy diff --git a/internal/proxy/forward.go b/internal/proxy/forward.go index bd81427..e74fc21 100644 --- a/internal/proxy/forward.go +++ b/internal/proxy/forward.go @@ -23,15 +23,15 @@ import ( var transportHeaders = []string{"user-agent", ":authority", "content-type"} type ( - // Forwarder forwards any allowlisted method to a single upstream, typing each - // request and response from the proto registry rather than being generated per - // service. The typing is load-bearing: namespace translation and payload + // Forwarder forwards any method to a single upstream, typing each request and + // response from the proto registry rather than being generated per service. + // Deciding which services may be forwarded is the caller's job: the gateway + // rejects the rest before routing. The typing is load-bearing: namespace translation and payload // encryption are client interceptors on cc that operate on proto messages, so // an opaque byte passthrough (as the router uses) would silently skip both. // Resolved methods are cached, and a Forwarder is safe for concurrent use. Forwarder struct { cc grpc.ClientConnInterface - allowed services.Allowlist methods sync.Map // fullName -> *methodInfo types protoutil.Types } @@ -47,22 +47,17 @@ type ( } ) -// NewForwarder builds a Forwarder that forwards over cc every method belonging -// to a service a admits. It fails when cc or a is nil. By default methods are -// typed against the global proto registry; use [WithProtoTypes] to override it. -func NewForwarder(cc grpc.ClientConnInterface, a services.Allowlist, opts ...ForwarderOption) (*Forwarder, error) { +// NewForwarder builds a Forwarder that forwards over cc. It fails when cc is nil. +// By default methods are typed against the global proto registry; use +// [WithProtoTypes] to override it. +func NewForwarder(cc grpc.ClientConnInterface, opts ...ForwarderOption) (*Forwarder, error) { if cc == nil { return nil, fmt.Errorf("proxy: nil client connection passed to forwarder") } - if a == nil { - return nil, fmt.Errorf("proxy: nil allowlist passed to forwarder") - } - f := &Forwarder{ - cc: cc, - allowed: a, - types: protoregistry.GlobalTypes, + cc: cc, + types: protoregistry.GlobalTypes, } for _, opt := range opts { @@ -82,12 +77,9 @@ func WithProtoTypes(t protoutil.Types) ForwarderOption { } } -// Handle forwards one stream to the upstream, and suits -// [google.golang.org/grpc.UnknownServiceHandler]. A method whose service the -// [services.Allowlist] does not admit is rejected with Unimplemented before any -// upstream work, so the proxy answers as a server that does not implement it -// rather than revealing that an upstream might. Only methods present in the -// compiled descriptors can be forwarded; anything else is Unimplemented too. +// Handle forwards one stream to the upstream. Only methods present in the +// compiled descriptors can be forwarded; anything else is rejected with +// Unimplemented before any upstream work. func (f *Forwarder) Handle(_ any, ss grpc.ServerStream) error { ctx := ss.Context() method, err := rpc.FullMethod(ctx) @@ -95,10 +87,6 @@ func (f *Forwarder) Handle(_ any, ss grpc.ServerStream) error { return err } - if service := rpc.Service(method); !f.allowed.Allows(service) { - return status.Errorf(codes.Unimplemented, "unknown service %s", service) - } - info := f.lookup(method) if info == nil { return status.Errorf(codes.Unimplemented, "unknown method %s", method) diff --git a/internal/proxy/forward_test.go b/internal/proxy/forward_test.go index d349dde..648d12c 100644 --- a/internal/proxy/forward_test.go +++ b/internal/proxy/forward_test.go @@ -82,25 +82,20 @@ func TestForwardContextWithoutIncomingMetadata(t *testing.T) { func TestNewForwarderValidation(t *testing.T) { t.Parallel() - allowed := services.NewAllowlist(services.Default()) - tests := []struct { - name string - cc grpc.ClientConnInterface - allowed services.Allowlist - err string + name string + cc grpc.ClientConnInterface + err string }{ - {name: "nil client connection", allowed: allowed, err: "nil client connection"}, - {name: "nil allowlist", cc: &testutil.ClientConn{}, err: "nil allowlist"}, - {name: "both nil reports the connection first", err: "nil client connection"}, - {name: "a connection and an allowlist is enough", cc: &testutil.ClientConn{}, allowed: allowed}, + {name: "nil client connection", err: "nil client connection"}, + {name: "a connection is enough", cc: &testutil.ClientConn{}}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - fw, err := NewForwarder(tt.cc, tt.allowed) + fw, err := NewForwarder(tt.cc) if tt.err != "" { require.ErrorContains(t, err, tt.err) require.Nil(t, fw) @@ -119,12 +114,12 @@ func TestWithProtoTypes(t *testing.T) { custom := partialTypes{} - fw, err := NewForwarder(&testutil.ClientConn{}, services.NewAllowlist(services.Default()), WithProtoTypes(custom)) + fw, err := NewForwarder(&testutil.ClientConn{}, WithProtoTypes(custom)) require.NoError(t, err) require.Equal(t, custom, fw.types) // A nil registry leaves the default in place rather than disabling resolution. - fw, err = NewForwarder(&testutil.ClientConn{}, services.NewAllowlist(services.Default()), WithProtoTypes(nil)) + fw, err = NewForwarder(&testutil.ClientConn{}, WithProtoTypes(nil)) require.NoError(t, err) require.Equal(t, protoregistry.GlobalTypes, fw.types) } @@ -164,7 +159,7 @@ func TestResolveMethod(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - fw, err := NewForwarder(&testutil.ClientConn{}, services.NewAllowlist(services.Default()), WithProtoTypes(tt.types)) + fw, err := NewForwarder(&testutil.ClientConn{}, WithProtoTypes(tt.types)) require.NoError(t, err) got := fw.resolveMethod(tt.in) @@ -183,7 +178,7 @@ func TestResolveMethod(t *testing.T) { func TestLookupCachesResolvedMethods(t *testing.T) { t.Parallel() - fw, err := NewForwarder(&testutil.ClientConn{}, services.NewAllowlist(services.Default())) + fw, err := NewForwarder(&testutil.ClientConn{}) require.NoError(t, err) const method = "/" + services.WorkflowService + "/GetSystemInfo" @@ -210,7 +205,6 @@ func TestHandleErrors(t *testing.T) { name string method string noStream bool - allowed []string conn *testutil.ClientConn ss testutil.ServerStream code codes.Code @@ -223,71 +217,57 @@ func TestHandleErrors(t *testing.T) { msg: "no server transport stream", }, { - name: "a service that is not on the allowlist", - method: "/" + services.OperatorService + "/DeleteNamespace", - allowed: []string{services.WorkflowService}, - code: codes.Unimplemented, - msg: "unknown service " + services.OperatorService, - }, - { - name: "a method the compiled descriptors do not have", - method: "/" + services.WorkflowService + "/NoSuchMethod", - allowed: services.Default(), - code: codes.Unimplemented, - msg: "unknown method", + name: "a method the compiled descriptors do not have", + method: "/" + services.WorkflowService + "/NoSuchMethod", + code: codes.Unimplemented, + msg: "unknown method", }, { - name: "an upstream that rejects the unary call", - method: unaryMethod, - allowed: services.Default(), - conn: &testutil.ClientConn{InvokeErr: status.Error(codes.PermissionDenied, "denied")}, - code: codes.PermissionDenied, - msg: "denied", + name: "an upstream that rejects the unary call", + method: unaryMethod, + conn: &testutil.ClientConn{InvokeErr: status.Error(codes.PermissionDenied, "denied")}, + code: codes.PermissionDenied, + msg: "denied", }, { - name: "a caller that cannot receive the relayed header", - method: unaryMethod, - allowed: services.Default(), - conn: &testutil.ClientConn{Header: metadata.Pairs("x-upstream", "1")}, - ss: testutil.ServerStream{HeaderErr: errors.New("header refused")}, - code: codes.Internal, - msg: "header refused", + name: "a caller that cannot receive the relayed header", + method: unaryMethod, + conn: &testutil.ClientConn{Header: metadata.Pairs("x-upstream", "1")}, + ss: testutil.ServerStream{HeaderErr: errors.New("header refused")}, + code: codes.Internal, + msg: "header refused", }, { - name: "a caller that cannot receive the response", - method: unaryMethod, - allowed: services.Default(), - ss: testutil.ServerStream{SendErr: errors.New("broken pipe")}, - code: codes.Internal, - msg: "broken pipe", + name: "a caller that cannot receive the response", + method: unaryMethod, + ss: testutil.ServerStream{SendErr: errors.New("broken pipe")}, + code: codes.Internal, + msg: "broken pipe", }, { - name: "an upstream stream that cannot be opened", - method: streamMethod, - allowed: []string{services.Reflection}, - conn: &testutil.ClientConn{StreamErr: status.Error(codes.Unavailable, "upstream down")}, - code: codes.Unavailable, - msg: "upstream down", + name: "an upstream stream that cannot be opened", + method: streamMethod, + conn: &testutil.ClientConn{StreamErr: status.Error(codes.Unavailable, "upstream down")}, + code: codes.Unavailable, + msg: "upstream down", }, { - name: "an upstream stream that fails before its header", - method: streamMethod, - allowed: []string{services.Reflection}, - conn: &testutil.ClientConn{Stream: &testutil.ClientStream{HeaderErr: status.Error(codes.Aborted, "header failed")}}, - ss: testutil.ServerStream{RecvErr: io.EOF}, - code: codes.Aborted, - msg: "header failed", + name: "an upstream stream that fails before its header", + method: streamMethod, + conn: &testutil.ClientConn{Stream: &testutil.ClientStream{HeaderErr: status.Error(codes.Aborted, "header failed")}}, + ss: testutil.ServerStream{RecvErr: io.EOF}, + code: codes.Aborted, + msg: "header failed", }, { // The request pump reports first because the response pump is parked in // RecvMsg, so this pins the mapping of a caller-side stream failure. - name: "a caller whose request stream breaks", - method: streamMethod, - allowed: []string{services.Reflection}, - conn: &testutil.ClientConn{Stream: &testutil.ClientStream{BlockRecv: make(chan struct{})}}, - ss: testutil.ServerStream{RecvErr: errors.New("caller vanished")}, - code: codes.Internal, - msg: "caller vanished", + name: "a caller whose request stream breaks", + method: streamMethod, + conn: &testutil.ClientConn{Stream: &testutil.ClientStream{BlockRecv: make(chan struct{})}}, + ss: testutil.ServerStream{RecvErr: errors.New("caller vanished")}, + code: codes.Internal, + msg: "caller vanished", }, } @@ -304,7 +284,7 @@ func TestHandleErrors(t *testing.T) { t.Cleanup(func() { close(cs.BlockRecv) }) } - fw, err := NewForwarder(conn, services.NewAllowlist(tt.allowed)) + fw, err := NewForwarder(conn) require.NoError(t, err) ss := tt.ss @@ -329,10 +309,7 @@ func TestStreamTreatsWrappedEOFAsHalfClose(t *testing.T) { // otherwise successful call. wrapped := fmt.Errorf("transport closed: %w", io.EOF) - fw, err := NewForwarder( - &testutil.ClientConn{Stream: &testutil.ClientStream{RecvErr: wrapped}}, - services.NewAllowlist([]string{services.Reflection}), - ) + fw, err := NewForwarder(&testutil.ClientConn{Stream: &testutil.ClientStream{RecvErr: wrapped}}) require.NoError(t, err) ss := testutil.ServerStream{ diff --git a/internal/proxy/forwarding_test.go b/internal/proxy/forwarding_test.go index 851d9fa..21fb4de 100644 --- a/internal/proxy/forwarding_test.go +++ b/internal/proxy/forwarding_test.go @@ -68,7 +68,7 @@ func TestStreamForwardsBidiReflection(t *testing.T) { workflowservice.RegisterWorkflowServiceServer(s, &metadataStampingService{}) reflection.Register(s) }) - conn := startProxy(t, addr, services.Reflection) + conn := startProxy(t, addr) stream, err := grpc_reflection_v1.NewServerReflectionClient(conn).ServerReflectionInfo( t.Context(), @@ -139,20 +139,16 @@ func (*metadataStampingService) GetSystemInfo( return &workflowservice.GetSystemInfoResponse{}, nil } -// forwarder builds a forwarder for the named services over a plain client conn -// to upstream, standing in for the pool-backed connection used in production. -func forwarder(t *testing.T, upstream string, allowed ...string) *proxy.Forwarder { +// forwarder builds a forwarder over a plain client conn to upstream, standing in +// for the pool-backed connection used in production. +func forwarder(t *testing.T, upstream string) *proxy.Forwarder { t.Helper() - if len(allowed) == 0 { - allowed = services.Default() - } - conn, err := grpc.NewClient(upstream, grpc.WithTransportCredentials(insecure.NewCredentials())) require.NoError(t, err) t.Cleanup(func() { _ = conn.Close() }) - fw, err := proxy.NewForwarder(conn, services.NewAllowlist(allowed)) + fw, err := proxy.NewForwarder(conn) require.NoError(t, err) return fw @@ -177,15 +173,15 @@ func serveUpstream(t *testing.T, register ...func(*grpc.Server)) string { return lis.Addr().String() } -// startProxy serves a forwarder for the named services to upstream on a -// loopback port and returns a client connection to it. -func startProxy(t *testing.T, upstream string, allowed ...string) *grpc.ClientConn { +// startProxy serves a forwarder to upstream on a loopback port and returns a +// client connection to it. +func startProxy(t *testing.T, upstream string) *grpc.ClientConn { t.Helper() lis, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) - svr := grpc.NewServer(grpc.UnknownServiceHandler(forwarder(t, upstream, allowed...).Handle)) + svr := grpc.NewServer(grpc.UnknownServiceHandler(forwarder(t, upstream).Handle)) go func() { _ = svr.Serve(lis) }() t.Cleanup(svr.Stop) diff --git a/internal/proxy/translation_test.go b/internal/proxy/translation_test.go index 8778727..69b93eb 100644 --- a/internal/proxy/translation_test.go +++ b/internal/proxy/translation_test.go @@ -19,7 +19,6 @@ import ( "google.golang.org/protobuf/reflect/protoregistry" "github.com/temporalio/temporal-proxy/internal/protoutil" - "github.com/temporalio/temporal-proxy/internal/services" ) type ( @@ -148,7 +147,7 @@ func TestOutboundNamespaceTranslation(t *testing.T) { require.NoError(t, err) t.Cleanup(func() { _ = cc.Close() }) - fw, err := NewForwarder(cc, services.NewAllowlist(services.Default())) + fw, err := NewForwarder(cc) require.NoError(t, err) proxyLis, err := net.Listen("tcp", "127.0.0.1:0")