diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index aa625fa94..5861b398a 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -659,7 +659,7 @@ func (c *Command) Run(args []string) int { if err := c.worker.Start(); err != nil { retErr := fmt.Errorf("Error starting worker: %w", err) - if err := c.worker.Shutdown(false); err != nil { + if err := c.worker.Shutdown(); err != nil { c.UI.Error(retErr.Error()) retErr = fmt.Errorf("Error shutting down worker: %w", err) } @@ -709,7 +709,7 @@ func (c *Command) Run(args []string) int { } if !c.flagControllerOnly { - if err := c.worker.Shutdown(false); err != nil { + if err := c.worker.Shutdown(); err != nil { c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error()) } } diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 129ce2909..bd9e3a2e4 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -636,7 +636,7 @@ func (c *Command) StartWorker() error { if err := c.worker.Start(); err != nil { retErr := fmt.Errorf("Error starting worker: %w", err) - if err := c.worker.Shutdown(false); err != nil { + if err := c.worker.Shutdown(); err != nil { c.UI.Error(retErr.Error()) retErr = fmt.Errorf("Error shutting down worker: %w", err) } @@ -677,7 +677,7 @@ func (c *Command) WaitForInterrupt() int { // Do worker shutdown if c.Config.Worker != nil { - if err := c.worker.Shutdown(false); err != nil { + if err := c.worker.Shutdown(); err != nil { c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error()) } } diff --git a/internal/servers/worker/controller_connection.go b/internal/servers/worker/controller_connection.go index 023a2e896..1b1402d95 100644 --- a/internal/servers/worker/controller_connection.go +++ b/internal/servers/worker/controller_connection.go @@ -138,17 +138,13 @@ func (w *Worker) createClientConn(addr string) error { func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) { var err error info := &base.WorkerAuthInfo{ - Name: w.conf.RawConfig.Worker.Name, - Description: w.conf.RawConfig.Worker.Description, - ConnectionNonce: w.testReusedAuthNonce, + Name: w.conf.RawConfig.Worker.Name, + Description: w.conf.RawConfig.Worker.Description, } - if info.ConnectionNonce == "" { - if info.ConnectionNonce, err = base62.Random(20); err != nil { - return nil, nil, err - } - if w.testReuseAuthNonces { - w.testReusedAuthNonce = info.ConnectionNonce - } + + info.ConnectionNonce, err = w.nonceFn(20) + if err != nil { + return nil, nil, err } pubKey, privKey, err := ed25519.GenerateKey(w.conf.SecureRandomReader) diff --git a/internal/servers/worker/listeners.go b/internal/servers/worker/listeners.go index 0342ae3d3..25521aef0 100644 --- a/internal/servers/worker/listeners.go +++ b/internal/servers/worker/listeners.go @@ -5,12 +5,13 @@ 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 +19,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 +28,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,38 +46,127 @@ func (w *Worker) startListeners() error { return nil } -func (w *Worker) stopListeners() error { - serverWg := new(sync.WaitGroup) - for _, ln := range w.conf.Listeners { - localLn := ln - serverWg.Add(1) - go func() { - defer serverWg.Done() - - shutdownKill, shutdownKillCancel := context.WithTimeout(w.baseContext, localLn.Config.MaxRequestDuration) - defer shutdownKillCancel() - - if localLn.HTTPServer != nil { - _ = localLn.HTTPServer.Shutdown(shutdownKill) - } - }() +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 } - serverWg.Wait() - var retErr *multierror.Error - if !w.conf.RawConfig.DevController { - for _, ln := range w.conf.Listeners { - if err := ln.Mux.Close(); err != nil { - if _, ok := err.(*os.PathError); ok && ln.Config.Type == "unix" { - // The rmListener probably tried to remove the file but it - // didn't exist, ignore the error; this is a conflict - // between rmListener and the default Go behavior of - // removing auto-vivified Unix domain sockets. - } else { - retErr = multierror.Append(retErr, 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) stopServersAndListeners() error { + var closeErrors *multierror.Error + err := w.stopHttpServersAndListeners() + if err != nil { + closeErrors = multierror.Append(closeErrors, err) + } + + err = w.stopAnyListeners() + if err != nil { + closeErrors = multierror.Append(closeErrors, err) + } + + return closeErrors.ErrorOrNil() +} + +func (w *Worker) stopHttpServersAndListeners() error { + var closeErrors *multierror.Error + for i := range w.listeners { + ln := w.listeners[i] + if ln.HTTPServer == nil { + continue + } + + ctx, cancel := context.WithTimeout(w.baseContext, ln.Config.MaxRequestDuration) + ln.HTTPServer.Shutdown(ctx) + cancel() + + err := ln.Mux.Close() + err = listenerCloseErrorCheck(ln.Config.Type, err) + if err != nil { + multierror.Append(closeErrors, err) } } - return retErr.ErrorOrNil() + + return closeErrors.ErrorOrNil() +} + +// stopAnyListeners does a final once over the known +// listeners to make sure we didn't miss any; +// expected to run at the end of stopServersAndListeners. +func (w *Worker) stopAnyListeners() error { + var closeErrors *multierror.Error + for _, ln := range w.listeners { + if ln == nil || ln.Mux == nil { + continue + } + + err := ln.Mux.Close() + err = listenerCloseErrorCheck(ln.Config.Type, err) + if err != nil { + multierror.Append(closeErrors, err) + } + } + + return closeErrors.ErrorOrNil() +} + +// listenerCloseErrorCheck does some validation on an error returned +// by a net.Listener's Close function, and ignores a few cases +// where we don't actually want an error to be returned. +func listenerCloseErrorCheck(lnType string, err error) error { + if errors.Is(err, net.ErrClosed) { + // Ignore net.ErrClosed - The listener was already closed, + // so there's nothing else to do. + return nil + } + if _, ok := err.(*os.PathError); ok && lnType == "unix" { + // The underlying rmListener probably tried to remove + // the file but it didn't exist, ignore the error; + // this is a conflict between rmListener and the + // default Go behavior of removing auto-vivified + // Unix domain sockets. + return nil + } + + return err } diff --git a/internal/servers/worker/listeners_test.go b/internal/servers/worker/listeners_test.go new file mode 100644 index 000000000..46f4a8689 --- /dev/null +++ b/internal/servers/worker/listeners_test.go @@ -0,0 +1,389 @@ +package worker + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "os" + "strconv" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/cmd/base" + "github.com/hashicorp/boundary/internal/libs/alpnmux" + "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) + } + }) + } +} + +func TestStopHttpServersAndListeners(t *testing.T) { + tests := []struct { + name string + workerFn func(t *testing.T) *Worker + assertions func(t *testing.T, w *Worker) + expErr bool + expErrStr string + }{ + { + name: "no listeners", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: []*base.ServerListener{}} + }, + expErr: false, + }, + { + name: "nil listeners slice", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: nil} + }, + expErr: false, + }, + { + name: "listeners with nil http server", + workerFn: func(t *testing.T) *Worker { + return &Worker{ + listeners: []*base.ServerListener{ + {HTTPServer: nil}, + {HTTPServer: nil}, + {HTTPServer: nil}, + }, + } + }, + expErr: false, + }, + { + name: "multiple listeners", + workerFn: func(t *testing.T) *Worker { + l1, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + l2, err := net.Listen("unix", "/tmp/boundary-worker-TestStopHttpServersAndListeners-"+strconv.FormatInt(time.Now().UnixNano(), 10)) + require.NoError(t, err) + + s1 := &http.Server{} + s2 := &http.Server{} + + go s1.Serve(l1) + go s2.Serve(l2) + + // Make sure they're up + _, err = http.Get("http://" + l1.Addr().String()) + require.NoError(t, err) + + c := http.Client{Transport: &http.Transport{ + Dial: func(network, addr string) (net.Conn, error) { + return net.Dial("unix", l2.Addr().String()) + }, + }} + _, err = c.Get("http://random.domain") + require.NoError(t, err) + + return &Worker{ + baseContext: context.Background(), + listeners: []*base.ServerListener{ + { + ALPNListener: l1, + HTTPServer: s1, + Mux: alpnmux.New(l1), + Config: &listenerutil.ListenerConfig{Type: "tcp"}, + }, + { + ALPNListener: l2, + HTTPServer: s2, + Mux: alpnmux.New(l2), + Config: &listenerutil.ListenerConfig{Type: "tcp"}, + }, + }, + } + }, + assertions: func(t *testing.T, w *Worker) { + // Asserts the HTTP Servers are closed. + require.ErrorIs(t, w.listeners[0].HTTPServer.Serve(w.listeners[0].ALPNListener), http.ErrServerClosed) + require.ErrorIs(t, w.listeners[1].HTTPServer.Serve(w.listeners[1].ALPNListener), http.ErrServerClosed) + + // Asserts the underlying listeners are closed. + require.ErrorIs(t, w.listeners[0].Mux.Close(), net.ErrClosed) + require.ErrorIs(t, w.listeners[1].Mux.Close(), net.ErrClosed) + }, + expErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := tt.workerFn(t) + err := w.stopHttpServersAndListeners() + if tt.expErr { + require.EqualError(t, err, tt.expErrStr) + return + } + + require.NoError(t, err) + if tt.assertions != nil { + tt.assertions(t, w) + } + }) + } +} + +func TestStopAnyListeners(t *testing.T) { + tests := []struct { + name string + workerFn func(t *testing.T) *Worker + assertions func(t *testing.T, w *Worker) + expErr bool + expErrStr string + }{ + { + name: "nil listeners", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: nil} + }, + expErr: false, + }, + { + name: "no listeners", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: []*base.ServerListener{}} + }, + expErr: false, + }, + { + name: "listeners slice with nil server listeners", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: []*base.ServerListener{nil, nil, nil}} + }, + expErr: false, + }, + { + name: "listeners with nil mux", + workerFn: func(t *testing.T) *Worker { + return &Worker{listeners: []*base.ServerListener{ + {Mux: nil}, {Mux: nil}, {Mux: nil}, + }} + }, + expErr: false, + }, + { + name: "multiple listeners, including a closed one", + workerFn: func(t *testing.T) *Worker { + l1, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + l2, err := net.Listen("unix", "/tmp/boundary-worker-TestStopAnyListeners-"+strconv.FormatInt(time.Now().UnixNano(), 10)) + require.NoError(t, err) + + l3, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + require.NoError(t, l3.Close()) + + return &Worker{listeners: []*base.ServerListener{ + { + Config: &listenerutil.ListenerConfig{Type: "tcp"}, + Mux: alpnmux.New(l1), + }, + { + Config: &listenerutil.ListenerConfig{Type: "tcp"}, + Mux: alpnmux.New(l2), + }, + { + Config: &listenerutil.ListenerConfig{Type: "tcp"}, + Mux: alpnmux.New(l3), + }, + }} + }, + assertions: func(t *testing.T, w *Worker) { + for i := range w.listeners { + ln := w.listeners[i] + require.ErrorIs(t, ln.Mux.Close(), net.ErrClosed) + } + }, + expErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := tt.workerFn(t) + err := c.stopAnyListeners() + if tt.expErr { + require.EqualError(t, err, tt.expErrStr) + return + } + + require.NoError(t, err) + if tt.assertions != nil { + tt.assertions(t, c) + } + }) + } +} + +func TestListenerCloseErrorCheck(t *testing.T) { + tests := []struct { + name string + lnType string + err error + expErr error + }{ + { + name: "nil err", + lnType: "tcp", + err: nil, + expErr: nil, + }, + { + name: "net.Closed", + lnType: "tcp", + err: net.ErrClosed, + expErr: nil, + }, + { + name: "path err not unix type", + lnType: "tcp", + err: &os.PathError{Op: "test"}, + expErr: &os.PathError{Op: "test"}, + }, + { + name: "path err unix type", + lnType: "unix", + err: &os.PathError{Op: "test"}, + expErr: nil, + }, + { + name: "literally anything else", + lnType: "tcp", + err: fmt.Errorf("oops I errored"), + expErr: fmt.Errorf("oops I errored"), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := listenerCloseErrorCheck(tt.lnType, tt.err) + require.Equal(t, tt.expErr, err) + }) + } +} diff --git a/internal/servers/worker/testing.go b/internal/servers/worker/testing.go index 17b21e6cf..e13d3aacc 100644 --- a/internal/servers/worker/testing.go +++ b/internal/servers/worker/testing.go @@ -148,7 +148,7 @@ func (tw *TestWorker) Shutdown() { tw.cancel() if tw.w != nil { - if err := tw.w.Shutdown(false); err != nil { + if err := tw.w.Shutdown(); err != nil { tw.t.Error(err) } } @@ -183,8 +183,9 @@ type TestWorkerOpts struct { // connection cannot be made back to the controller StatusGracePeriodDuration time.Duration - // Whether to set the replay value - EnableAuthReplay bool + // Overrides worker's nonceFn, for cases where we want to have control + // over the nonce we send to the Controller + NonceFn randFn } func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker { @@ -280,8 +281,8 @@ func NewTestWorker(t *testing.T, opts *TestWorkerOpts) *TestWorker { t.Fatal(err) } - if opts.EnableAuthReplay { - tw.w.testReuseAuthNonces = true + if opts.NonceFn != nil { + tw.w.nonceFn = opts.NonceFn } if !opts.DisableAutoStart { diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index f7ca0163e..c9659918a 100644 --- a/internal/servers/worker/worker.go +++ b/internal/servers/worker/worker.go @@ -14,18 +14,22 @@ import ( "sync/atomic" "time" + "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/cmd/config" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/boundary/internal/servers/worker/session" "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" ua "go.uber.org/atomic" "google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver/manual" ) +type randFn func(length int) (string, error) + type Worker struct { conf *Config logger hclog.Logger @@ -45,6 +49,11 @@ type Worker struct { controllerSessionConn *atomic.Value sessionInfoMap *sync.Map + listeners []*base.ServerListener + + // Used to generate a random nonce for Controller connections + nonceFn randFn + // We store the current set in an atomic value so that we can add // reload-on-sighup behavior later tags *atomic.Value @@ -52,10 +61,6 @@ type Worker struct { // request. It can be set via startup in New below, or (eventually) via // SIGHUP. updateTags ua.Bool - - // Test-related values - testReuseAuthNonces bool - testReusedAuthNonce string } func New(conf *Config) (*Worker, error) { @@ -69,6 +74,7 @@ func New(conf *Config) (*Worker, error) { controllerSessionConn: new(atomic.Value), sessionInfoMap: new(sync.Map), tags: new(atomic.Value), + nonceFn: base62.Random, } w.lastStatusSuccess.Store((*LastStatusInformation)(nil)) @@ -105,6 +111,23 @@ func New(conf *Config) (*Worker, error) { } } + for i := range conf.Listeners { + l := conf.Listeners[i] + if l == nil || l.Config == nil || l.Config.Purpose == nil { + continue + } + if len(l.Config.Purpose) != 1 { + return nil, fmt.Errorf("found listener with multiple purposes %q", strings.Join(l.Config.Purpose, ",")) + } + switch l.Config.Purpose[0] { + case "proxy": + w.listeners = append(w.listeners, l) + } + } + if len(w.listeners) == 0 { + return nil, fmt.Errorf("no proxy listeners found") + } + return w, nil } @@ -144,7 +167,7 @@ func (w *Worker) Start() error { // listeners, useful for tests if we want to stop and start a worker. In order // to create new listeners we'd have to migrate listener setup logic here -- // doable, but work for later. -func (w *Worker) Shutdown(skipListeners bool) error { +func (w *Worker) Shutdown() error { const op = "worker.(Worker).Shutdown" if !w.started.Load() { event.WriteSysEvent(w.baseContext, op, "already shut down, skipping") @@ -156,10 +179,9 @@ func (w *Worker) Shutdown(skipListeners bool) error { defer w.started.Store(false) w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}}) w.baseCancel() - if !skipListeners { - if err := w.stopListeners(); err != nil { - return fmt.Errorf("error stopping worker listeners: %w", err) - } + + if err := w.stopServersAndListeners(); err != nil { + return fmt.Errorf("error stopping worker servers and listeners: %w", err) } // Shut down all connections. diff --git a/internal/servers/worker/worker_test.go b/internal/servers/worker/worker_test.go new file mode 100644 index 000000000..8ae62a97d --- /dev/null +++ b/internal/servers/worker/worker_test.go @@ -0,0 +1,114 @@ +package worker + +import ( + "testing" + + "github.com/hashicorp/boundary/internal/cmd/base" + "github.com/hashicorp/boundary/internal/cmd/config" + "github.com/hashicorp/go-hclog" + "github.com/hashicorp/go-secure-stdlib/configutil/v2" + "github.com/hashicorp/go-secure-stdlib/listenerutil" + "github.com/stretchr/testify/require" +) + +func TestWorkerNew(t *testing.T) { + tests := []struct { + name string + in *Config + expErr bool + expErrMsg string + assertions func(t *testing.T, w *Worker) + }{ + { + name: "nil listeners", + in: &Config{Server: &base.Server{Listeners: nil}}, + expErr: true, + expErrMsg: "no proxy listeners found", + }, + { + name: "zero listeners", + in: &Config{Server: &base.Server{Listeners: []*base.ServerListener{}}}, + expErr: true, + expErrMsg: "no proxy listeners found", + }, + { + name: "populated with nil values", + in: &Config{ + Server: &base.Server{ + Listeners: []*base.ServerListener{ + nil, + {Config: nil}, + {Config: &listenerutil.ListenerConfig{Purpose: nil}}, + }, + }, + }, + expErr: true, + expErrMsg: "no proxy listeners found", + }, + { + name: "multiple purposes", + in: &Config{ + Server: &base.Server{ + Listeners: []*base.ServerListener{ + nil, + {Config: nil}, + {Config: &listenerutil.ListenerConfig{Purpose: []string{"api", "proxy"}}}, + }, + }, + }, + expErr: true, + expErrMsg: `found listener with multiple purposes "api,proxy"`, + }, + { + name: "valid listeners", + in: &Config{ + Server: &base.Server{ + Listeners: []*base.ServerListener{ + {Config: &listenerutil.ListenerConfig{Purpose: []string{"api"}}}, + {Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}}, + {Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}}, + {Config: &listenerutil.ListenerConfig{Purpose: []string{"cluster"}}}, + }, + }, + }, + expErr: false, + assertions: func(t *testing.T, w *Worker) { + require.Len(t, w.listeners, 2) + }, + }, + { + name: "worker nonce func is set", + in: &Config{ + Server: &base.Server{ + Listeners: []*base.ServerListener{ + {Config: &listenerutil.ListenerConfig{Purpose: []string{"proxy"}}}, + }, + }, + }, + expErr: false, + assertions: func(t *testing.T, w *Worker) { + require.NotNil(t, w.nonceFn) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // New() panics if these aren't set + tt.in.Logger = hclog.Default() + tt.in.RawConfig = &config.Config{SharedConfig: &configutil.SharedConfig{DisableMlock: true}} + + w, err := New(tt.in) + if tt.expErr { + require.EqualError(t, err, tt.expErrMsg) + require.Nil(t, w) + return + } + + require.NoError(t, err) + if tt.assertions != nil { + tt.assertions(t, w) + } + }) + } +} diff --git a/internal/tests/cluster/ipv6_listener_test.go b/internal/tests/cluster/ipv6_listener_test.go index 4f60f2bed..4f0527ce6 100644 --- a/internal/tests/cluster/ipv6_listener_test.go +++ b/internal/tests/cluster/ipv6_listener_test.go @@ -11,12 +11,11 @@ import ( "github.com/hashicorp/boundary/internal/servers/controller" "github.com/hashicorp/boundary/internal/servers/worker" "github.com/hashicorp/go-hclog" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) func TestIPv6Listener(t *testing.T) { - assert, require := assert.New(t), require.New(t) + require := require.New(t) logger := hclog.New(&hclog.LoggerOptions{ Level: hclog.Trace, }) @@ -40,28 +39,7 @@ func TestIPv6Listener(t *testing.T) { }) defer c1.Shutdown() - expectWorkers := func(c *controller.TestController, workers ...*worker.TestWorker) { - updateTimes := c.Controller().WorkerStatusUpdateTimes() - workerMap := map[string]*worker.TestWorker{} - for _, w := range workers { - workerMap[w.Name()] = w - } - updateTimes.Range(func(k, v interface{}) bool { - require.NotNil(k) - require.NotNil(v) - if workerMap[k.(string)] == nil { - // We don't remove from updateTimes currently so if we're not - // expecting it we'll see an out-of-date entry - return true - } - assert.WithinDuration(time.Now(), v.(time.Time), 60*time.Second) - delete(workerMap, k.(string)) - return true - }) - assert.Empty(workerMap) - } - - expectWorkers(c1) + expectWorkers(t, c1) wconf, err := config.DevWorker() require.NoError(err) @@ -75,7 +53,7 @@ func TestIPv6Listener(t *testing.T) { defer w1.Shutdown() time.Sleep(10 * time.Second) - expectWorkers(c1, w1) + expectWorkers(t, c1, w1) c2 := c1.AddClusterControllerMember(t, &controller.TestControllerOpts{ Logger: c1.Config().Logger.ResetNamed("c2"), @@ -83,12 +61,12 @@ func TestIPv6Listener(t *testing.T) { defer c2.Shutdown() time.Sleep(10 * time.Second) - expectWorkers(c2, w1) + expectWorkers(t, c2, w1) - require.NoError(w1.Worker().Shutdown(true)) + require.NoError(w1.Worker().Shutdown()) time.Sleep(10 * time.Second) - expectWorkers(c1) - expectWorkers(c2) + expectWorkers(t, c1) + expectWorkers(t, c2) client, err := api.NewClient(nil) require.NoError(err) diff --git a/internal/tests/cluster/multi_controller_worker_test.go b/internal/tests/cluster/multi_controller_worker_test.go index cb39bb630..deadb22da 100644 --- a/internal/tests/cluster/multi_controller_worker_test.go +++ b/internal/tests/cluster/multi_controller_worker_test.go @@ -54,21 +54,31 @@ func TestMultiControllerMultiWorkerConnections(t *testing.T) { expectWorkers(t, c1, w1, w2) expectWorkers(t, c2, w1, w2) - require.NoError(w1.Worker().Shutdown(true)) + require.NoError(w1.Worker().Shutdown()) time.Sleep(10 * time.Second) expectWorkers(t, c1, w2) expectWorkers(t, c2, w2) - require.NoError(w1.Worker().Start()) + w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialControllers: c1.ClusterAddrs(), + Logger: logger.Named("w1"), + }) + defer w1.Shutdown() + time.Sleep(10 * time.Second) expectWorkers(t, c1, w1, w2) expectWorkers(t, c2, w1, w2) require.NoError(c2.Controller().Shutdown()) time.Sleep(10 * time.Second) - expectWorkers(t, c2, w1, w2) + expectWorkers(t, c1, w1, w2) + + c2 = c1.AddClusterControllerMember(t, &controller.TestControllerOpts{ + Logger: c1.Config().Logger.ResetNamed("c2"), + }) + defer c2.Shutdown() - require.NoError(c1.Controller().Start()) time.Sleep(10 * time.Second) expectWorkers(t, c1, w1, w2) expectWorkers(t, c2, w1, w2) diff --git a/internal/tests/cluster/unix_listener_test.go b/internal/tests/cluster/unix_listener_test.go index c7248a6e9..ed8958fbf 100644 --- a/internal/tests/cluster/unix_listener_test.go +++ b/internal/tests/cluster/unix_listener_test.go @@ -14,12 +14,11 @@ import ( "github.com/hashicorp/boundary/internal/servers/controller" "github.com/hashicorp/boundary/internal/servers/worker" "github.com/hashicorp/go-hclog" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) func TestUnixListener(t *testing.T) { - assert, require := assert.New(t), require.New(t) + require := require.New(t) logger := hclog.New(&hclog.LoggerOptions{ Level: hclog.Trace, }) @@ -53,28 +52,7 @@ func TestUnixListener(t *testing.T) { }) defer c1.Shutdown() - expectWorkers := func(c *controller.TestController, workers ...*worker.TestWorker) { - updateTimes := c.Controller().WorkerStatusUpdateTimes() - workerMap := map[string]*worker.TestWorker{} - for _, w := range workers { - workerMap[w.Name()] = w - } - updateTimes.Range(func(k, v interface{}) bool { - require.NotNil(k) - require.NotNil(v) - if workerMap[k.(string)] == nil { - // We don't remove from updateTimes currently so if we're not - // expecting it we'll see an out-of-date entry - return true - } - assert.WithinDuration(time.Now(), v.(time.Time), 60*time.Second) - delete(workerMap, k.(string)) - return true - }) - assert.Empty(workerMap) - } - - expectWorkers(c1) + expectWorkers(t, c1) wconf, err := config.DevWorker() require.NoError(err) @@ -88,11 +66,11 @@ func TestUnixListener(t *testing.T) { defer w1.Shutdown() time.Sleep(10 * time.Second) - expectWorkers(c1, w1) + expectWorkers(t, c1, w1) - require.NoError(w1.Worker().Shutdown(true)) + require.NoError(w1.Worker().Shutdown()) time.Sleep(10 * time.Second) - expectWorkers(c1) + expectWorkers(t, c1) require.NoError(c1.Controller().Shutdown()) c1 = controller.NewTestController(t, &controller.TestControllerOpts{ @@ -103,7 +81,7 @@ func TestUnixListener(t *testing.T) { defer c1.Shutdown() time.Sleep(10 * time.Second) - expectWorkers(c1) + expectWorkers(t, c1) client, err := api.NewClient(nil) require.NoError(err) diff --git a/internal/tests/cluster/worker_replay_test.go b/internal/tests/cluster/worker_replay_test.go index 76be8b0cb..3182ee6d3 100644 --- a/internal/tests/cluster/worker_replay_test.go +++ b/internal/tests/cluster/worker_replay_test.go @@ -13,7 +13,6 @@ import ( "github.com/hashicorp/boundary/internal/servers/controller" "github.com/hashicorp/boundary/internal/servers/worker" "github.com/hashicorp/go-hclog" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -41,24 +40,49 @@ func TestWorkerReplay(t *testing.T) { Config: conf, WorkerAuthKms: c1.Config().WorkerAuthKms, InitialControllers: c1.ClusterAddrs(), - EnableAuthReplay: true, + NonceFn: func(length int) (string, error) { + return "test_noncetest_nonce", nil + }, }) // Give time for it to connect time.Sleep(10 * time.Second) - require.NoError(t, w1.Worker().Shutdown(true)) - // Now, start up again - require.NoError(t, w1.Worker().Start()) + // Assert that the worker has connected + logBuf, err := os.ReadFile(ec.AllEvents.Name()) + require.NoError(t, err) + require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed")) + + // Assert we have the expected nonce in the DB + nonces, err := c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth) + require.NoError(t, err) + require.Len(t, nonces, 1) + require.Equal(t, &servers.Nonce{Nonce: "test_noncetest_nonce", Purpose: servers.NoncePurposeWorkerAuth}, nonces[0]) + + require.NoError(t, w1.Worker().Shutdown()) + + // Now, start up again with the same nonce + w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialControllers: c1.ClusterAddrs(), + NonceFn: func(length int) (string, error) { + return "test_noncetest_nonce", nil + }, + }) + + // Give time for it to connect time.Sleep(10 * time.Second) // We should find only one nonce, and one successful worker authentication, // both in the output and in the database ec.AllEvents.Close() - logBuf, err := os.ReadFile(ec.AllEvents.Name()) + logBuf, err = os.ReadFile(ec.AllEvents.Name()) require.NoError(t, err) - assert.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed")) - nonces, err := c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth) + require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed")) + + nonces, err = c1.ServersRepo().ListNonces(c1.Context(), servers.NoncePurposeWorkerAuth) require.NoError(t, err) - assert.Len(t, nonces, 1) + require.Len(t, nonces, 1) + require.Equal(t, &servers.Nonce{Nonce: "test_noncetest_nonce", Purpose: servers.NoncePurposeWorkerAuth}, nonces[0]) }