diff --git a/internal/servers/worker/listeners.go b/internal/servers/worker/listeners.go index 0342ae3d3..fd9f629cc 100644 --- a/internal/servers/worker/listeners.go +++ b/internal/servers/worker/listeners.go @@ -5,12 +5,14 @@ import ( "crypto/tls" "errors" "fmt" + "log" "net" "net/http" "os" "sync" "time" + "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/libs/alpnmux" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-multierror" @@ -18,7 +20,7 @@ import ( func (w *Worker) startListeners() error { const op = "worker.(Worker).startListeners" - servers := make([]func(), 0, len(w.conf.Listeners)) + e := event.SysEventer() if e == nil { return fmt.Errorf("%s: sys eventer not initialized", op) @@ -27,70 +29,15 @@ func (w *Worker) startListeners() error { if err != nil { return fmt.Errorf("%s: unable to initialize std logger: %w", op, err) } - for _, ln := range w.conf.Listeners { - for _, purpose := range ln.Config.Purpose { - switch purpose { - case "api", "cluster", "ops": - // We may have this in dev mode; ignore - continue - case "proxy": - // Do nothing; handle below - - default: - return fmt.Errorf("unknown listener purpose %q", purpose) - } - - handler, err := w.handler(HandlerProperties{ - ListenerConfig: ln.Config, - }) - if err != nil { - return fmt.Errorf("%s: %w", op, err) - } - - cancelCtx := w.baseContext - - server := &http.Server{ - Handler: handler, - ReadHeaderTimeout: 10 * time.Second, - ReadTimeout: 30 * time.Second, - ErrorLog: logger, - BaseContext: func(net.Listener) context.Context { - return cancelCtx - }, - } - ln.HTTPServer = server - - if ln.Config.HTTPReadHeaderTimeout > 0 { - server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout - } - if ln.Config.HTTPReadTimeout > 0 { - server.ReadTimeout = ln.Config.HTTPReadTimeout - } - if ln.Config.HTTPWriteTimeout > 0 { - server.WriteTimeout = ln.Config.HTTPWriteTimeout - } - if ln.Config.HTTPIdleTimeout > 0 { - server.IdleTimeout = ln.Config.HTTPIdleTimeout - } - - // Clear out in case this is a second start of the controller - ln.Mux.UnregisterProto(alpnmux.DefaultProto) - ln.Mux.UnregisterProto(alpnmux.NoProto) - l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{ - GetConfigForClient: w.getSessionTls, - }) - if err != nil { - return fmt.Errorf("error getting tls listener: %w", err) - } - if l == nil { - return errors.New("could not get tls listener") - } - - servers = append(servers, func() { - go server.Serve(l) - }) + servers := make([]func(), 0, len(w.listeners)) + for i := range w.listeners { + ln := w.listeners[i] + workerServer, err := w.configureForWorker(ln, logger) + if err != nil { + return fmt.Errorf("%s: failed to configure for worker: %w", op, err) } + servers = append(servers, workerServer) } for _, s := range servers { @@ -100,6 +47,53 @@ func (w *Worker) startListeners() error { return nil } +func (w *Worker) configureForWorker(ln *base.ServerListener, log *log.Logger) (func(), error) { + handler, err := w.handler(HandlerProperties{ListenerConfig: ln.Config}) + if err != nil { + return nil, err + } + + cancelCtx := w.baseContext + server := &http.Server{ + Handler: handler, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + ErrorLog: log, + BaseContext: func(net.Listener) context.Context { + return cancelCtx + }, + } + ln.HTTPServer = server + + if ln.Config.HTTPReadHeaderTimeout > 0 { + server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout + } + if ln.Config.HTTPReadTimeout > 0 { + server.ReadTimeout = ln.Config.HTTPReadTimeout + } + if ln.Config.HTTPWriteTimeout > 0 { + server.WriteTimeout = ln.Config.HTTPWriteTimeout + } + if ln.Config.HTTPIdleTimeout > 0 { + server.IdleTimeout = ln.Config.HTTPIdleTimeout + } + + // Clear out in case this is a second start of the controller + ln.Mux.UnregisterProto(alpnmux.DefaultProto) + ln.Mux.UnregisterProto(alpnmux.NoProto) + l, err := ln.Mux.RegisterProto(alpnmux.DefaultProto, &tls.Config{ + GetConfigForClient: w.getSessionTls, + }) + if err != nil { + return nil, fmt.Errorf("error getting tls listener: %w", err) + } + if l == nil { + return nil, errors.New("could not get tls listener") + } + + return func() { go server.Serve(l) }, nil +} + func (w *Worker) stopListeners() error { serverWg := new(sync.WaitGroup) for _, ln := range w.conf.Listeners { diff --git a/internal/servers/worker/listeners_test.go b/internal/servers/worker/listeners_test.go new file mode 100644 index 000000000..ebea51477 --- /dev/null +++ b/internal/servers/worker/listeners_test.go @@ -0,0 +1,134 @@ +package worker + +import ( + "context" + "io" + "net" + "net/http" + "testing" + + "github.com/hashicorp/go-secure-stdlib/listenerutil" + "github.com/stretchr/testify/require" +) + +func TestStartListeners(t *testing.T) { + tests := []struct { + name string + listeners []*listenerutil.ListenerConfig + expErr bool + expErrMsg string + assertions func(t *testing.T, w *Worker, addrs []string) + }{ + { + name: "one tcp listener, one unix listener", + listeners: []*listenerutil.ListenerConfig{ + { + Type: "tcp", + Purpose: []string{"proxy"}, + Address: "127.0.0.1:0", + TLSDisable: true, + }, + { + Type: "unix", + Purpose: []string{"proxy"}, + Address: "/tmp/boundary-worker-test-start-listeners.sock", + TLSDisable: true, + }, + }, + expErr: false, + assertions: func(t *testing.T, w *Worker, addrs []string) { + require.Len(t, addrs, 2) + + _, err := http.Get("http://" + addrs[0] + "/v1/proxy/") + require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation + + cl := http.Client{ + Transport: &http.Transport{ + Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", addrs[1]) }, + }, + } + _, err = cl.Get("http://anything.domain/v1/proxy/") + require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation + }, + }, + { + name: "multiple tcp listeners, multiple unix listeners", + listeners: []*listenerutil.ListenerConfig{ + { + Type: "tcp", + Purpose: []string{"proxy"}, + Address: "127.0.0.1:0", + TLSDisable: true, + }, + { + Type: "tcp", + Purpose: []string{"proxy"}, + Address: "127.0.0.1:0", + TLSDisable: true, + }, + { + Type: "unix", + Purpose: []string{"proxy"}, + Address: "/tmp/boundary-worker-test-start-listeners0.sock", + TLSDisable: true, + }, + { + Type: "unix", + Purpose: []string{"proxy"}, + Address: "/tmp/boundary-worker-test-start-listeners1.sock", + TLSDisable: true, + }, + }, + expErr: false, + assertions: func(t *testing.T, w *Worker, addrs []string) { + require.Len(t, addrs, 4) + + _, err := http.Get("http://" + addrs[0] + "/v1/proxy/") + require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation + + _, err = http.Get("http://" + addrs[1] + "/v1/proxy/") + require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation + + for _, proxyAddr := range []string{addrs[2], addrs[3]} { + cl := http.Client{ + Transport: &http.Transport{ + Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", proxyAddr) }, + }, + } + _, err = cl.Get("http://anything.domain/v1/proxy") + require.ErrorIs(t, err, io.EOF) // empty response because of worker tls request validation + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + conf := NewTestWorker(t, &TestWorkerOpts{DisableAutoStart: true}).w.conf + conf.RawConfig.SharedConfig.Listeners = tt.listeners + + err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"proxy"}) + require.NoError(t, err) + + w, err := New(conf) + require.NoError(t, err) + w.baseContext = context.Background() + + err = w.startListeners() + if tt.expErr { + require.EqualError(t, err, tt.expErrMsg) + return + } + + require.NoError(t, err) + + addrs := make([]string, 0) + for _, l := range w.listeners { + addrs = append(addrs, l.Mux.Addr().String()) + } + if tt.assertions != nil { + tt.assertions(t, w, addrs) + } + }) + } +}