diff --git a/go.mod b/go.mod index b76ebc2ad..90781b4ce 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/hashicorp/go-bexpr v0.1.10 github.com/hashicorp/go-cleanhttp v0.5.2 github.com/hashicorp/go-hclog v1.1.0 - github.com/hashicorp/go-kms-wrapping/v2 v2.0.4 + github.com/hashicorp/go-kms-wrapping/v2 v2.0.5 github.com/hashicorp/go-multierror v1.1.1 github.com/hashicorp/go-retryablehttp v0.7.0 github.com/hashicorp/go-rootcerts v1.0.2 @@ -40,7 +40,7 @@ require ( github.com/hashicorp/go-secure-stdlib/mlock v0.1.1 github.com/hashicorp/go-secure-stdlib/parseutil v0.1.2 github.com/hashicorp/go-secure-stdlib/password v0.1.1 - github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.0 + github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.1 github.com/hashicorp/go-secure-stdlib/reloadutil v0.1.1 github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 github.com/hashicorp/go-sockaddr v1.0.2 @@ -65,11 +65,13 @@ require ( github.com/pires/go-proxyproto v0.6.1 github.com/pkg/errors v0.9.1 github.com/posener/complete v1.2.3 + github.com/prometheus/client_golang v1.12.1 github.com/ryanuber/go-glob v1.0.0 - github.com/stretchr/testify v1.7.0 + github.com/stretchr/testify v1.7.1 github.com/zalando/go-keyring v0.2.1 go.uber.org/atomic v1.9.0 golang.org/x/crypto v0.0.0-20220331220935-ae2d96664a29 + golang.org/x/sync v0.0.0-20210220032951-036812b2e83c golang.org/x/sys v0.0.0-20220330033206-e17cdc41300f golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 golang.org/x/tools v0.1.10 @@ -85,10 +87,7 @@ require ( require github.com/hashicorp/go-dbw v0.0.0-20211215222256-2ff0d37184ff // this is a branch and should be updated before merging -require ( - github.com/prometheus/client_golang v1.12.1 - golang.org/x/sync v0.0.0-20210220032951-036812b2e83c -) +require github.com/hashicorp/nodeenrollment v0.0.0-20220427165109-a2f5dad593c7 require ( github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect diff --git a/go.sum b/go.sum index a2b2f40f2..ebc320bf9 100644 --- a/go.sum +++ b/go.sum @@ -367,8 +367,8 @@ github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjh github.com/hashicorp/go-kms-wrapping/entropy v0.1.0/go.mod h1:d1g9WGtAunDNpek8jUIEJnBlbgKS1N2Q61QkHiZyR1g= github.com/hashicorp/go-kms-wrapping/plugin/v2 v2.0.2 h1:it114Kjpk79JVh6AfxFMP5oi1gb+RPzcSLiG3zF/J0w= github.com/hashicorp/go-kms-wrapping/plugin/v2 v2.0.2/go.mod h1:b9mq0bG5xJ10ZcTEQJLZ22O9y71hIt2MGTLqROA/SAM= -github.com/hashicorp/go-kms-wrapping/v2 v2.0.4 h1:7BSAm60AqJom/TRb73wmaCzwco9A+Qolr/wCi0PORmY= -github.com/hashicorp/go-kms-wrapping/v2 v2.0.4/go.mod h1:sDQAfwJGv25uGPZA04x87ERglCG6avnRcBT9wYoMII8= +github.com/hashicorp/go-kms-wrapping/v2 v2.0.5 h1:rOFDv+3k05mnW0oaDLffhVUwg03Csn0mvfO98Wdd2bE= +github.com/hashicorp/go-kms-wrapping/v2 v2.0.5/go.mod h1:sDQAfwJGv25uGPZA04x87ERglCG6avnRcBT9wYoMII8= github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk= github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA= github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= @@ -400,8 +400,8 @@ github.com/hashicorp/go-secure-stdlib/parseutil v0.1.2 h1:Tz6v3Jb2DRnDCfifRSjYKG github.com/hashicorp/go-secure-stdlib/parseutil v0.1.2/go.mod h1:QmrqtbKuxxSWTN3ETMPuB+VtEiBJ/A9XhoYGv8E1uD8= github.com/hashicorp/go-secure-stdlib/password v0.1.1 h1:6JzmBqXprakgFEHwBgdchsjaA9x3GyjdI568bXKxa60= github.com/hashicorp/go-secure-stdlib/password v0.1.1/go.mod h1:9hH302QllNwu1o2TGYtSk8I8kTAN0ca1EHpwhm5Mmzo= -github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.0 h1:TszfJJ1vJWdvPkPxYivP5TUTkOaXziE1jTKGmfOck94= -github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.0/go.mod h1:Io+g8iYs//ESo5x2Gy1rTTYaIHUyCZD7h16K7DiOUI4= +github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.1 h1:s47ax4xEIStj1YV7TRz6zS8LlNrTU86aEhL8pfSOG4U= +github.com/hashicorp/go-secure-stdlib/pluginutil/v2 v2.0.1/go.mod h1:Mh8pP9IH9qnBQSSH3PYtSkGq3+KlIuw5kAS9W8++BLc= github.com/hashicorp/go-secure-stdlib/reloadutil v0.1.1 h1:SMGUnbpAcat8rIKHkBPjfv81yC46a8eCNZ2hsR2l1EI= github.com/hashicorp/go-secure-stdlib/reloadutil v0.1.1/go.mod h1:Ch/bf00Qnx77MZd49JRgHYqHQjtEmTgGU2faufpVZb0= github.com/hashicorp/go-secure-stdlib/strutil v0.1.1/go.mod h1:gKOamz3EwoIoJq7mlMIRBpVTAUn8qPCrEclOKKWhD3U= @@ -425,6 +425,8 @@ github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+l github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= +github.com/hashicorp/nodeenrollment v0.0.0-20220427165109-a2f5dad593c7 h1:pOKDBZ6t9IZCL4xXWt0hJw0TM6Mn0BrRk4NAu7sJKS4= +github.com/hashicorp/nodeenrollment v0.0.0-20220427165109-a2f5dad593c7/go.mod h1:yOOykLWw4G7O1Z7+6X3gzXcQCb/5au1Zna4d1nlQLmM= github.com/hashicorp/vault/api v1.3.1 h1:pkDkcgTh47PRjY1NEFeofqR4W/HkNUi9qIakESO2aRM= github.com/hashicorp/vault/api v1.3.1/go.mod h1:QeJoWxMFt+MsuWcYhmwRLwKEXrjwAFFywzhptMsTIUw= github.com/hashicorp/vault/sdk v0.1.13/go.mod h1:B+hVj7TpuQY1Y/GPbCpffmgd+tSEwvhkWnjtSYCaS2M= @@ -764,8 +766,9 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= github.com/stretchr/testify v1.6.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk= github.com/tj/assert v0.0.0-20171129193455-018094318fb0/go.mod h1:mZ9/Rh9oLWpLLDRpvE+3b7gP/C2YyLFYxNmcLnPTMe0= github.com/tj/assert v0.0.3 h1:Df/BlaZ20mq6kuai7f5z2TvPFiwC3xaWJSDQNiIS3Rk= diff --git a/internal/cmd/base/option.go b/internal/cmd/base/option.go index d5c4735e0..25a241886 100644 --- a/internal/cmd/base/option.go +++ b/internal/cmd/base/option.go @@ -22,24 +22,25 @@ type Option func(*Options) // Options - how Options are represented. type Options struct { - withNoTokenScope bool - withNoTokenValue bool - withSkipDatabaseDestruction bool - withSkipAuthMethodCreation bool - withSkipOidcAuthMethodCreation bool - withSkipScopesCreation bool - withSkipHostResourcesCreation bool - withSkipTargetCreation bool - withContainerImage string - withDialect string - withDatabaseTemplate string - withEventerConfig *event.EventerConfig - withEventFlags *EventFlags - withEventWrapper wrapping.Wrapper - withAttributeFieldPrefix string - withStatusCode int - withHostPlugin func() (string, plugin.HostPluginServiceClient) - withEventGating bool + withNoTokenScope bool + withNoTokenValue bool + withSkipDatabaseDestruction bool + withSkipAuthMethodCreation bool + withSkipOidcAuthMethodCreation bool + withSkipScopesCreation bool + withSkipHostResourcesCreation bool + withSkipTargetCreation bool + withContainerImage string + withDialect string + withDatabaseTemplate string + withEventerConfig *event.EventerConfig + withEventFlags *EventFlags + withEventWrapper wrapping.Wrapper + withAttributeFieldPrefix string + withStatusCode int + withHostPlugin func() (string, plugin.HostPluginServiceClient) + withEventGating bool + withSkipWorkerAuthKmsInstantiation bool } func getDefaultOptions() Options { @@ -188,3 +189,11 @@ func WithEventGating(with bool) Option { o.withEventGating = with } } + +// WithSkipWorkerAuthKmsInstantiation causes KMS methods used for worker-auth to +// not be processed, useful for dev mode +func WithSkipWorkerAuthKmsInstantiation(with bool) Option { + return func(o *Options) { + o.withSkipWorkerAuthKmsInstantiation = with + } +} diff --git a/internal/cmd/base/servers.go b/internal/cmd/base/servers.go index 2b0f4c0d6..99c2abf7b 100644 --- a/internal/cmd/base/servers.go +++ b/internal/cmd/base/servers.go @@ -517,7 +517,9 @@ func (b *Server) SetupListeners(ui cli.Ui, config *configutil.SharedConfig, allo // SetupKMSes takes in a parsed config, does some minor checking on purposes, // and sends each off to configutil to instantiate a wrapper. -func (b *Server) SetupKMSes(ctx context.Context, ui cli.Ui, config *config.Config) error { +func (b *Server) SetupKMSes(ctx context.Context, ui cli.Ui, config *config.Config, opt ...Option) error { + opts := getOpts(opt...) + sharedConfig := config.SharedConfig var pluginLogger hclog.Logger var err error @@ -527,7 +529,11 @@ func (b *Server) SetupKMSes(ctx context.Context, ui cli.Ui, config *config.Confi switch purpose { case "": return errors.New("KMS block missing 'purpose'") - case globals.KmsPurposeRoot, globals.KmsPurposeWorkerAuth, globals.KmsPurposeConfig: + case globals.KmsPurposeWorkerAuth: + if opts.withSkipWorkerAuthKmsInstantiation { + continue + } + case globals.KmsPurposeRoot, globals.KmsPurposeConfig: case globals.KmsPurposeRecovery: if config.Controller != nil && config.DevRecoveryKey != "" { kms.Config["key"] = config.DevRecoveryKey diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 5861b398a..668ab7b91 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -29,8 +29,10 @@ import ( "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/go-secure-stdlib/parseutil" "github.com/hashicorp/go-secure-stdlib/strutil" + "github.com/hashicorp/nodeenrollment/nodetypes" "github.com/mitchellh/cli" "github.com/posener/complete" + "go.uber.org/atomic" ) var ( @@ -51,39 +53,40 @@ type Command struct { controller *controller.Controller worker *worker.Worker - flagLogLevel string - flagLogFormat string - flagCombineLogs bool - flagLoginName string - flagPassword string - flagUnprivilegedLoginName string - flagUnprivilegedPassword string - flagIdSuffix string - flagHostAddress string - flagTargetDefaultPort int - flagTargetSessionMaxSeconds int - flagTargetSessionConnectionLimit int - flagControllerAPIListenAddr string - flagControllerClusterListenAddr string - flagControllerPublicClusterAddr string - flagControllerOnly bool - flagWorkerAuthKey string - flagWorkerProxyListenAddr string - flagWorkerPublicAddr string - flagOpsListenAddr string - flagUiPassthroughDir string - flagRecoveryKey string - flagDatabaseUrl string - flagContainerImage string - flagDisableDatabaseDestruction bool - flagEventFormat string - flagAudit string - flagObservations string - flagSysEvents string - flagEveryEventAllowFilters []string - flagEveryEventDenyFilters []string - flagCreateLoopbackHostPlugin bool - flagPluginExecutionDir string + flagLogLevel string + flagLogFormat string + flagCombineLogs bool + flagLoginName string + flagPassword string + flagUnprivilegedLoginName string + flagUnprivilegedPassword string + flagIdSuffix string + flagHostAddress string + flagTargetDefaultPort int + flagTargetSessionMaxSeconds int + flagTargetSessionConnectionLimit int + flagControllerAPIListenAddr string + flagControllerClusterListenAddr string + flagControllerPublicClusterAddr string + flagControllerOnly bool + flagWorkerAuthKey string + flagWorkerProxyListenAddr string + flagWorkerPublicAddr string + flagOpsListenAddr string + flagUiPassthroughDir string + flagRecoveryKey string + flagDatabaseUrl string + flagContainerImage string + flagDisableDatabaseDestruction bool + flagEventFormat string + flagAudit string + flagObservations string + flagSysEvents string + flagEveryEventAllowFilters []string + flagEveryEventDenyFilters []string + flagCreateLoopbackHostPlugin bool + flagPluginExecutionDir string + flagUseEphemeralKmsWorkerAuthMethod bool } func (c *Command) Synopsis() string { @@ -327,6 +330,11 @@ func (c *Command) Flags() *base.FlagSets { EnvVar: "BOUNDARY_DEV_PLUGIN_EXECUTION_DIR", Usage: "Specifies where Boundary should write plugins that it is executing; if not set defaults to system temp directory.", }) + f.BoolVar(&base.BoolVar{ + Name: "use-ephemeral-kms-worker-auth-method", + Target: &c.flagUseEphemeralKmsWorkerAuthMethod, + Usage: "If set, the original \"ephemeral\" method of worker auth will be used to connect the initial dev worker to the controller.", + }) f.BoolVar(&base.BoolVar{ Name: "create-loopback-host-plugin", @@ -540,7 +548,7 @@ func (c *Command) Run(args []string) int { if c.flagRecoveryKey != "" { c.Config.DevRecoveryKey = c.flagRecoveryKey } - if err := c.SetupKMSes(c.Context, c.UI, c.Config); err != nil { + if err := c.SetupKMSes(c.Context, c.UI, c.Config, base.WithSkipWorkerAuthKmsInstantiation(!c.flagUseEphemeralKmsWorkerAuthMethod)); err != nil { c.UI.Error(err.Error()) return base.CommandUserError } @@ -548,14 +556,16 @@ func (c *Command) Run(args []string) int { c.UI.Error("Controller KMS not found after parsing KMS blocks") return base.CommandUserError } - if c.WorkerAuthKms == nil { - c.UI.Error("Worker Auth KMS not found after parsing KMS blocks") - return base.CommandUserError + if c.flagUseEphemeralKmsWorkerAuthMethod { + if c.WorkerAuthKms == nil { + c.UI.Error("Worker Auth KMS not found after parsing KMS blocks") + return base.CommandUserError + } + c.InfoKeys = append(c.InfoKeys, "[Worker-Auth] AEAD Key Bytes") + c.Info["[Worker-Auth] AEAD Key Bytes"] = c.Config.DevWorkerAuthKey } c.InfoKeys = append(c.InfoKeys, "[Controller] AEAD Key Bytes") c.Info["[Controller] AEAD Key Bytes"] = c.Config.DevControllerKey - c.InfoKeys = append(c.InfoKeys, "[Worker-Auth] AEAD Key Bytes") - c.Info["[Worker-Auth] AEAD Key Bytes"] = c.Config.DevWorkerAuthKey c.InfoKeys = append(c.InfoKeys, "[Recovery] AEAD Key Bytes") c.Info["[Recovery] AEAD Key Bytes"] = c.Config.DevRecoveryKey @@ -613,12 +623,6 @@ func (c *Command) Run(args []string) int { } } - c.PrintInfo(c.UI) - if err := c.ReleaseLogGate(); err != nil { - c.UI.Error(fmt.Errorf("Error releasing event gate: %w", err).Error()) - return base.CommandCliError - } - { c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostAws, base.EnabledPluginHostAzure) conf := &controller.Config{ @@ -644,6 +648,8 @@ func (c *Command) Run(args []string) int { } } + errorEncountered := atomic.NewBool(false) + if !c.flagControllerOnly { conf := &worker.Config{ RawConfig: c.Config, @@ -669,6 +675,45 @@ func (c *Command) Run(args []string) int { } return base.CommandCliError } + + if !c.flagUseEphemeralKmsWorkerAuthMethod { + if c.worker.NodeeKeyId == "" { + c.UI.Error("No worker key ID found at worker registration time") + return base.CommandCliError + } + c.InfoKeys = append(c.InfoKeys, "worker key identifier") + c.Info["worker key identifier"] = c.worker.NodeeKeyId + go func() { + for { + select { + case <-c.Context.Done(): + return + case <-time.After(time.Second): + waitingNodes, err := c.controller.NodeeFileStorage.List(c.Context, (*nodetypes.NodeInformation)(nil)) + if err != nil { + c.UI.Error(fmt.Errorf("Error listing dev worker for authentication: %w", err).Error()) + errorEncountered.Store(true) + return + } + if len(waitingNodes) == 0 { + continue + } + if err := c.controller.AuthorizeNodeeWorker(c.worker.NodeeKeyId); err != nil { + c.UI.Error(fmt.Errorf("Error authorizing node: %w", err).Error()) + errorEncountered.Store(true) + return + } + return + } + } + }() + } + } + + c.PrintInfo(c.UI) + if err := c.ReleaseLogGate(); err != nil { + c.UI.Error(fmt.Errorf("Error releasing event gate: %w", err).Error()) + return base.CommandCliError } opsServer, err := ops.NewServer(c.Logger, c.controller, c.Listeners...) @@ -683,7 +728,7 @@ func (c *Command) Run(args []string) int { // Wait for shutdown shutdownTriggered := false - for !shutdownTriggered { + for !shutdownTriggered && !errorEncountered.Load() { select { case <-c.ShutdownCh: c.UI.Output("==> Boundary dev environment shutdown triggered, interrupt again to force") @@ -704,6 +749,15 @@ func (c *Command) Run(args []string) int { } }() + shutdownTriggered = true + + case <-c.SigUSR2Ch: + buf := make([]byte, 32*1024*1024) + n := runtime.Stack(buf[:], true) + event.WriteSysEvent(context.TODO(), op, "goroutine trace", "stack", string(buf[:n])) + } + + if shutdownTriggered { if c.Config.Controller != nil { c.opsServer.WaitIfHealthExists(c.Config.Controller.GracefulShutdownWaitDuration, c.UI) } @@ -722,13 +776,6 @@ func (c *Command) Run(args []string) int { if err != nil { c.UI.Error(fmt.Errorf("Failed to shutdown ops listeners: %w", err).Error()) } - - shutdownTriggered = true - - case <-c.SigUSR2Ch: - buf := make([]byte, 32*1024*1024) - n := runtime.Stack(buf[:], true) - event.WriteSysEvent(context.TODO(), op, "goroutine trace", "stack", string(buf[:n])) } } diff --git a/internal/cmd/commands/server/listener_reload_test.go b/internal/cmd/commands/server/listener_reload_test.go index 77398aa0d..4f19e884f 100644 --- a/internal/cmd/commands/server/listener_reload_test.go +++ b/internal/cmd/commands/server/listener_reload_test.go @@ -62,6 +62,7 @@ kms "aead" { listener "tcp" { purpose = "api" + address = "127.0.0.1:9500" tls_cert_file = "%s/bundle.pem" tls_key_file = "%s/bundle.pem" cors_enabled = true @@ -69,10 +70,12 @@ listener "tcp" { } listener "tcp" { + address = "127.0.0.1:9501" purpose = "cluster" } listener "tcp" { + address = "127.0.0.1:9502" purpose = "proxy" } ` @@ -131,7 +134,7 @@ func TestServer_ReloadListener(t *testing.T) { }() testCertificateSerial := func(serial string) { - conn, err := tls.Dial("tcp", "127.0.0.1:9200", &tls.Config{ + conn, err := tls.Dial("tcp", "127.0.0.1:9500", &tls.Config{ RootCAs: certPool, }) require.NoError(err) diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index bd9e3a2e4..eeadaf369 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -188,10 +188,6 @@ func (c *Command) Run(args []string) int { return base.CommandUserError } } - if c.WorkerAuthKms == nil { - c.UI.Error("Worker Auth KMS not found after parsing KMS blocks") - return base.CommandUserError - } if c.Config.DefaultMaxRequestDuration != 0 { globals.DefaultMaxRequestDuration = c.Config.DefaultMaxRequestDuration @@ -450,12 +446,6 @@ func (c *Command) Run(args []string) int { } }() - c.PrintInfo(c.UI) - if err := c.ReleaseLogGate(); err != nil { - c.UI.Error(fmt.Errorf("Error releasing event gate: %w", err).Error()) - return base.CommandCliError - } - if c.Config.Controller != nil { c.EnabledPlugins = append(c.EnabledPlugins, base.EnabledPluginHostAws, base.EnabledPluginHostAzure) if err := c.StartController(c.Context); err != nil { @@ -474,6 +464,28 @@ func (c *Command) Run(args []string) int { } return base.CommandCliError } + if c.WorkerAuthKms == nil { + if c.worker.NodeeKeyId == "" { + retErr := fmt.Errorf("No worker key ID found at worker registration time") + if err := c.worker.Shutdown(); err != nil { + c.UI.Error(retErr.Error()) + retErr = fmt.Errorf("Error shutting down worker: %w", err) + } + c.UI.Error(retErr.Error()) + if err := c.controller.Shutdown(); err != nil { + c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error()) + } + return base.CommandCliError + } + c.InfoKeys = append(c.InfoKeys, "worker key identifier") + c.Info["worker key identifier"] = c.worker.NodeeKeyId + } + } + + c.PrintInfo(c.UI) + if err := c.ReleaseLogGate(); err != nil { + c.UI.Error(fmt.Errorf("Error releasing event gate: %w", err).Error()) + return base.CommandCliError } opsServer, err := ops.NewServer(c.Logger, c.controller, c.Listeners...) diff --git a/internal/cmd/commands/server/worker_shutdown_reload_test.go b/internal/cmd/commands/server/worker_shutdown_reload_test.go index 9fafaee11..77a82c32e 100644 --- a/internal/cmd/commands/server/worker_shutdown_reload_test.go +++ b/internal/cmd/commands/server/worker_shutdown_reload_test.go @@ -125,6 +125,10 @@ func TestServer_ShutdownWorker(t *testing.T) { }() waitCh(t, workerCmd.startedCh) + // Give the worker time for initial status or we can fail because no worker + // is ready for the connection yet + time.Sleep(5 * time.Second) + // Set up the target ctx := context.Background() client := buildClient(t, shutdownReloadApiAddr) diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index 66da2faed..dc278af51 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -60,7 +60,7 @@ kms "aead" { listener "tcp" { purpose = "api" - address = "127.0.0.1:9203" + address = "127.0.0.1:9403" tls_disable = true cors_enabled = true cors_allowed_origins = ["*"] @@ -68,12 +68,12 @@ listener "tcp" { listener "tcp" { purpose = "cluster" - address = "127.0.0.1:9204" + address = "127.0.0.1:9404" } listener "tcp" { purpose = "proxy" - address = "127.0.0.1:9205" + address = "127.0.0.1:9405" } ` @@ -82,7 +82,7 @@ const tag1Config = ` worker { name = "dev-worker" description = "A default worker created in dev mode" - controllers = ["127.0.0.1:9204"] + controllers = ["127.0.0.1:9404"] tags { type = ["dev", "local"] } @@ -93,7 +93,7 @@ const tag2Config = ` worker { name = "dev-worker" description = "A default worker created in dev mode" - controllers = ["127.0.0.1:9204"] + controllers = ["127.0.0.1:9404"] tags { foo = ["bar", "baz"] } diff --git a/internal/servers/controller/controller.go b/internal/servers/controller/controller.go index 175dbb345..aeeb2b60f 100644 --- a/internal/servers/controller/controller.go +++ b/internal/servers/controller/controller.go @@ -36,6 +36,9 @@ import ( "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-secure-stdlib/mlock" "github.com/hashicorp/go-secure-stdlib/pluginutil/v2" + "github.com/hashicorp/nodeenrollment/noderegistration" + nodeefile "github.com/hashicorp/nodeenrollment/nodestorage/file" + "github.com/hashicorp/nodeenrollment/nodetypes" ua "go.uber.org/atomic" "google.golang.org/grpc" ) @@ -86,6 +89,9 @@ type Controller struct { // Used to signal the Health Service to start // replying to queries with "503 Service Unavailable". HealthService *health.Service + + // PoC: Testing bits for BYOW + NodeeFileStorage *nodeefile.FileStorage } func New(ctx context.Context, conf *Config) (*Controller, error) { @@ -299,6 +305,7 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { c.ConnectionRepoFn = func() (*session.ConnectionRepository, error) { return session.NewConnectionRepository(ctx, dbase, dbase, c.kms) } + return c, nil } @@ -309,6 +316,17 @@ func (c *Controller) Start() error { return nil } c.baseContext, c.baseCancel = context.WithCancel(context.Background()) + + var err error + c.NodeeFileStorage, err = nodeefile.NewFileStorage(c.baseContext) + if err != nil { + return err + } + _, err = nodetypes.RotateRootCertificates(c.baseContext, c.NodeeFileStorage) + if err != nil { + return err + } + if err := c.registerJobs(); err != nil { return fmt.Errorf("error registering jobs: %w", err) } @@ -391,3 +409,7 @@ func (c *Controller) Shutdown() error { func (c *Controller) WorkerStatusUpdateTimes() *sync.Map { return c.workerStatusUpdateTimes } + +func (c *Controller) AuthorizeNodeeWorker(keyId string) error { + return noderegistration.AuthorizeNode(c.baseContext, c.NodeeFileStorage, keyId) +} diff --git a/internal/servers/controller/handler.go b/internal/servers/controller/handler.go index 2f13dbac9..4a32ea81d 100644 --- a/internal/servers/controller/handler.go +++ b/internal/servers/controller/handler.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "io/ioutil" "net/http" "net/textproto" @@ -42,6 +43,7 @@ import ( "github.com/hashicorp/go-cleanhttp" "github.com/hashicorp/go-secure-stdlib/listenerutil" "github.com/hashicorp/go-secure-stdlib/strutil" + "github.com/hashicorp/nodeenrollment/nodetypes" "github.com/mr-tron/base58" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -65,6 +67,7 @@ func (c *Controller) apiHandler(props HandlerProperties) (http.Handler, error) { if err != nil { return nil, err } + mux.Handle("/v1/nodes", handleNodes(c)) mux.Handle("/v1/", grpcGwMux) mux.Handle("/", handleUi(c)) @@ -561,6 +564,68 @@ func wrapHandlerWithCallbackInterceptor(h http.Handler, c *Controller) http.Hand }) } +func handleNodes(c *Controller) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + switch req.Method { + case http.MethodGet: + type vals struct { + WaitingNodes []string `json:"waiting_nodes"` + } + var err error + var currVals vals + currVals.WaitingNodes, err = c.NodeeFileStorage.List(c.baseContext, (*nodetypes.NodeInformation)(nil)) + if err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + ret, err := json.Marshal(currVals) + if err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + _, err = w.Write(ret) + if err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + w.WriteHeader(200) + return + + case http.MethodPost: + body, err := io.ReadAll(req.Body) + req.Body.Close() + if err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + type val struct { + KeyId string `json:"key_id"` + } + var currVal val + if err := json.Unmarshal(body, &currVal); err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + if err := c.AuthorizeNodeeWorker(currVal.KeyId); err != nil { + _, _ = w.Write([]byte(err.Error())) + w.WriteHeader(500) + return + } + w.WriteHeader(204) + return + + default: + w.WriteHeader(http.StatusBadRequest) + return + } + }) +} + /* func WrapForwardedForHandler(h http.Handler, authorizedAddrs []*sockaddr.SockAddrMarshaler, rejectNotPresent, rejectNonAuthz bool, hopSkips int) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/servers/controller/handlers/workers/multihop_service.go b/internal/servers/controller/handlers/workers/multihop_service.go new file mode 100644 index 000000000..9b3fa4dbf --- /dev/null +++ b/internal/servers/controller/handlers/workers/multihop_service.go @@ -0,0 +1,45 @@ +package workers + +import ( + "context" + "fmt" + + pbs "github.com/hashicorp/nodeenrollment/multihop" + "github.com/hashicorp/nodeenrollment/nodeauth" + "github.com/hashicorp/nodeenrollment/noderegistration" + "github.com/hashicorp/nodeenrollment/nodetypes" +) + +type multihopServiceServer struct { + pbs.UnimplementedMultihopServiceServer + + currentParams nodeauth.CurrentParameterFactory +} + +func NewMultihopServiceServer( + currentParams nodeauth.CurrentParameterFactory, +) *multihopServiceServer { + return &multihopServiceServer{ + currentParams: currentParams, + } +} + +var _ pbs.MultihopServiceServer = (*multihopServiceServer)(nil) + +func (m *multihopServiceServer) FetchNodeCredentials(ctx context.Context, req *nodetypes.FetchNodeCredentialsRequest) (*nodetypes.FetchNodeCredentialsResponse, error) { + const op = "workers.(multihopServiceServer).FetchNodeCredentials" + _, storage, opt, err := m.currentParams() + if err != nil { + return nil, fmt.Errorf("%s: error getting current parameters: %w", op, err) + } + return noderegistration.FetchNodeCredentials(ctx, storage, req, opt...) +} + +func (m *multihopServiceServer) GenerateServerCertificates(ctx context.Context, req *nodetypes.GenerateServerCertificatesRequest) (*nodetypes.GenerateServerCertificatesResponse, error) { + const op = "workers.(multihopServiceServer).GenerateServerCertificates" + _, storage, opt, err := m.currentParams() + if err != nil { + return nil, fmt.Errorf("%s: error getting current parameters: %w", op, err) + } + return noderegistration.GenerateServerCertificates(ctx, storage, req, opt...) +} diff --git a/internal/servers/controller/intercepting_listener.go b/internal/servers/controller/intercepting_listener.go index 7cece060b..0088c8422 100644 --- a/internal/servers/controller/intercepting_listener.go +++ b/internal/servers/controller/intercepting_listener.go @@ -1,12 +1,15 @@ package controller import ( - "context" + "crypto/tls" "errors" "fmt" "net" + "strings" "github.com/hashicorp/boundary/internal/observability/event" + nodee "github.com/hashicorp/nodeenrollment" + "github.com/hashicorp/nodeenrollment/nodeauth" ) // interceptingListener allows us to validate the nonce from a connection before @@ -45,42 +48,54 @@ func newInterceptingListener(c *Controller, baseLn net.Listener) *interceptingLi func (m *interceptingListener) Accept() (net.Conn, error) { const op = "controller.(interceptingListener).Accept" - ctx := context.TODO() + ctx := m.c.baseContext conn, err := m.baseLn.Accept() if err != nil { + // We may already be closed but can't be sure, so just try anyways if conn != nil { if err := conn.Close(); err != nil { - event.WriteError(context.TODO(), op, err, event.WithInfoMsg("error closing worker connection")) + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) } } return nil, err } - nonce := make([]byte, 20) - read, err := conn.Read(nonce) - if err != nil { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + tlsConn := conn.(*tls.Conn) + switch { + case nodeauth.ContainsNodeAuthAlpnProto(tlsConn.ConnectionState().NegotiatedProtocol): + event.WriteSysEvent(ctx, op, "worker successfully authed", "key_id", nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId)) + return conn, nil + + case strings.HasPrefix(tlsConn.ConnectionState().NegotiatedProtocol, "v1workerauth"): + nonce := make([]byte, 20) + read, err := conn.Read(nonce) + if err != nil { + if err := conn.Close(); err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + } + return nil, fmt.Errorf("error reading nonce from connection: %w", err) } - return nil, fmt.Errorf("error reading nonce from connection: %w", err) - } - if read != len(nonce) { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + if read != len(nonce) { + if err := conn.Close(); err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + } + return nil, fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read) } - return nil, fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read) - } - workerInfoRaw, found := m.c.workerAuthCache.Load(string(nonce)) - if !found { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + workerInfoRaw, found := m.c.workerAuthCache.Load(string(nonce)) + if !found { + if err := conn.Close(); err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) + } + return nil, errors.New("did not find valid nonce for incoming worker") } - return nil, errors.New("did not find valid nonce for incoming worker") + workerInfo := workerInfoRaw.(*workerAuthEntry) + workerInfo.conn = tlsConn + event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name) + return tlsConn, nil + + default: + return nil, nodeauth.NewTempError(errors.New("unable to authenticate incoming connection")) } - workerInfo := workerInfoRaw.(*workerAuthEntry) - workerInfo.conn = conn - event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name) - return conn, nil } func (m *interceptingListener) Close() error { diff --git a/internal/servers/controller/listeners.go b/internal/servers/controller/listeners.go index aa9df9ed6..e0be9cb64 100644 --- a/internal/servers/controller/listeners.go +++ b/internal/servers/controller/listeners.go @@ -17,6 +17,9 @@ import ( "github.com/hashicorp/boundary/internal/servers/controller/handlers/workers" "github.com/hashicorp/boundary/internal/servers/controller/internal/metric" "github.com/hashicorp/go-multierror" + nodee "github.com/hashicorp/nodeenrollment" + "github.com/hashicorp/nodeenrollment/multihop" + "github.com/hashicorp/nodeenrollment/nodeauth" "google.golang.org/grpc" ) @@ -103,9 +106,21 @@ func (c *Controller) configureForApi(ln *base.ServerListener) ([]func(), error) } func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error) { - l := tls.NewListener(ln.ClusterListener, &tls.Config{ - GetConfigForClient: c.validateWorkerTls, - }) + l, err := nodeauth.NewInterceptingListener( + ln.ClusterListener, + &tls.Config{ + GetConfigForClient: c.validateWorkerTls, + }, + nodeauth.MakeCurrentParametersFactory( + c.baseContext, + nodee.NopTransactionStorage(c.NodeeFileStorage), + ), + nil, + nil, + ) + if err != nil { + return nil, fmt.Errorf("error instantiating node auth listener: %w", err) + } workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer) if err != nil { @@ -129,6 +144,12 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error c.workerStatusUpdateTimes, c.kms) pbs.RegisterServerCoordinationServiceServer(workerServer, workerService) pbs.RegisterSessionServiceServer(workerServer, workerService) + + multihopService := workers.NewMultihopServiceServer( + nodeauth.MakeCurrentParametersFactory(c.baseContext, nodee.NopTransactionStorage(c.NodeeFileStorage)), + ) + multihop.RegisterMultihopServiceServer(workerServer, multihopService) + metric.InitializeClusterCollectors(c.conf.PrometheusRegisterer, workerServer) ln.GrpcServer = workerServer diff --git a/internal/servers/controller/worker_tls_config.go b/internal/servers/controller/worker_tls_config.go index 29e20e12b..703e75f60 100644 --- a/internal/servers/controller/worker_tls_config.go +++ b/internal/servers/controller/worker_tls_config.go @@ -27,6 +27,7 @@ type workerAuthEntry struct { // been replayed, then stores connection information into the auth cache, which // is used by the intercepting listener to log information. func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config, error) { + const op = "controller.(Controller).validateWorkerTls" for _, p := range hello.SupportedProtos { switch { case strings.HasPrefix(p, "v1workerauth-"): @@ -50,7 +51,7 @@ func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config, return tlsConf, err } } - return nil, nil + return nil, errors.New("no supported protos found in hello") } // v1WorkerAuthConfig: diff --git a/internal/servers/worker/controller_connection.go b/internal/servers/worker/controller_connection.go index 210f31936..a1e0435a2 100644 --- a/internal/servers/worker/controller_connection.go +++ b/internal/servers/worker/controller_connection.go @@ -23,7 +23,12 @@ import ( "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/servers/worker/internal/metric" "github.com/hashicorp/go-secure-stdlib/base62" + nodee "github.com/hashicorp/nodeenrollment" + "github.com/hashicorp/nodeenrollment/multihop" + "github.com/hashicorp/nodeenrollment/nodeauth" "google.golang.org/grpc" + "google.golang.org/grpc/backoff" + "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" ) @@ -62,41 +67,62 @@ func (w *Worker) startControllerConnections() error { } func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, error) { - const op = "worker.(Worker).controllerDialerFunc" return func(ctx context.Context, addr string) (net.Conn, error) { - tlsConf, authInfo, err := w.workerAuthTLSConfig() - if err != nil { - return nil, fmt.Errorf("error creating tls config for worker auth: %w", err) - } - dialer := &net.Dialer{} - var nonTlsConn net.Conn + var conn net.Conn + var err error switch { - case strings.HasPrefix(addr, "/"): - nonTlsConn, err = dialer.DialContext(ctx, "unix", addr) + case w.conf.WorkerAuthKms != nil: + conn, err = w.v1KmsAuthDialFn(ctx, addr) default: - nonTlsConn, err = dialer.DialContext(ctx, "tcp", addr) + conn, err = nodeauth.AuthDialFn( + nodeauth.MakeCurrentParametersFactory( + ctx, + nodee.NopTransactionStorage(w.NodeeFileStorage), + ))(ctx, addr) } - if err != nil { - return nil, fmt.Errorf("unable to dial to controller: %w", err) + + if !w.everAuthenticated.Load() && err == nil && conn != nil { + w.everAuthenticated.Store(true) } - tlsConn := tls.Client(nonTlsConn, tlsConf) - written, err := tlsConn.Write([]byte(authInfo.ConnectionNonce)) - if err != nil { - if err := nonTlsConn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing connection after writing failure")) - } - return nil, fmt.Errorf("unable to write connection nonce: %w", err) - } - if written != len(authInfo.ConnectionNonce) { - if err := nonTlsConn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing connection after writing failure")) - } - return nil, fmt.Errorf("expected to write %d bytes of connection nonce, wrote %d", len(authInfo.ConnectionNonce), written) - } - return tlsConn, nil + + return conn, err } } +func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string) (net.Conn, error) { + const op = "worker.(Worker).v1KmsAuthDialFn" + tlsConf, authInfo, err := w.workerAuthTLSConfig() + if err != nil { + return nil, fmt.Errorf("error creating tls config for worker auth: %w", err) + } + dialer := &net.Dialer{} + var nonTlsConn net.Conn + switch { + case strings.HasPrefix(addr, "/"): + nonTlsConn, err = dialer.DialContext(ctx, "unix", addr) + default: + nonTlsConn, err = dialer.DialContext(ctx, "tcp", addr) + } + if err != nil { + return nil, fmt.Errorf("unable to dial to controller: %w", err) + } + tlsConn := tls.Client(nonTlsConn, tlsConf) + written, err := tlsConn.Write([]byte(authInfo.ConnectionNonce)) + if err != nil { + if err := nonTlsConn.Close(); err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing connection after writing failure")) + } + return nil, fmt.Errorf("unable to write connection nonce: %w", err) + } + if written != len(authInfo.ConnectionNonce) { + if err := nonTlsConn.Close(); err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error closing connection after writing failure")) + } + return nil, fmt.Errorf("expected to write %d bytes of connection nonce, wrote %d", len(authInfo.ConnectionNonce), written) + } + return tlsConn, nil +} + func (w *Worker) createClientConn(addr string) error { const op = "worker.(Worker).createClientConn" defaultTimeout := (time.Second + time.Nanosecond).String() @@ -113,18 +139,29 @@ func (w *Worker) createClientConn(addr string) error { } `, defaultTimeout) res := w.Resolver() - cc, err := grpc.DialContext(w.baseContext, - fmt.Sprintf("%s:///%s", res.Scheme(), addr), + dialOpts := []grpc.DialOption{ grpc.WithResolvers(res), grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)), grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(math.MaxInt32)), grpc.WithContextDialer(w.controllerDialerFunc()), - grpc.WithInsecure(), + grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(defServiceConfig), // Don't have the resolver reach out for a service config from the // resolver, use the one specified as default grpc.WithDisableServiceConfig(), + grpc.WithConnectParams(grpc.ConnectParams{ + Backoff: backoff.Config{ + BaseDelay: time.Second, + Multiplier: 1.2, + Jitter: 0.2, + MaxDelay: 3 * time.Second, + }, + }), + } + cc, err := grpc.DialContext(w.baseContext, + fmt.Sprintf("%s:///%s", res.Scheme(), addr), + dialOpts..., ) if err != nil { return fmt.Errorf("error dialing controller for worker auth: %w", err) @@ -132,8 +169,7 @@ func (w *Worker) createClientConn(addr string) error { w.controllerStatusConn.Store(pbs.NewServerCoordinationServiceClient(cc)) w.controllerSessionConn.Store(pbs.NewSessionServiceClient(cc)) - - event.WriteSysEvent(w.baseContext, op, "connected to controller", "address", addr) + w.controllerMultihopConn.Store(multihop.NewMultihopServiceClient(cc)) return nil } diff --git a/internal/servers/worker/listeners.go b/internal/servers/worker/listeners.go index ce259a700..f8a386247 100644 --- a/internal/servers/worker/listeners.go +++ b/internal/servers/worker/listeners.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log" + "math" "net" "net/http" "os" @@ -13,7 +14,13 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/boundary/internal/servers/controller/handlers/workers" "github.com/hashicorp/go-multierror" + nodee "github.com/hashicorp/nodeenrollment" + "github.com/hashicorp/nodeenrollment/multihop" + "github.com/hashicorp/nodeenrollment/nodeauth" + "github.com/hashicorp/nodeenrollment/nodetypes" + "google.golang.org/grpc" ) func (w *Worker) startListeners() error { @@ -27,21 +34,17 @@ func (w *Worker) startListeners() error { if err != nil { return fmt.Errorf("%s: unable to initialize std logger: %w", op, err) } - - 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) + if w.proxyListener == nil { + return fmt.Errorf("%s: nil proxy listener", op) } - for _, s := range servers { - s() + workerServer, err := w.configureForWorker(w.proxyListener, logger) + if err != nil { + return fmt.Errorf("%s: failed to configure for worker: %w", op, err) } + workerServer() + return nil } @@ -52,7 +55,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger) } cancelCtx := w.baseContext - server := &http.Server{ + httpServer := &http.Server{ Handler: handler, ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, @@ -61,80 +64,173 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger) return cancelCtx }, } - ln.HTTPServer = server + ln.HTTPServer = httpServer if ln.Config.HTTPReadHeaderTimeout > 0 { - server.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout + httpServer.ReadHeaderTimeout = ln.Config.HTTPReadHeaderTimeout } if ln.Config.HTTPReadTimeout > 0 { - server.ReadTimeout = ln.Config.HTTPReadTimeout + httpServer.ReadTimeout = ln.Config.HTTPReadTimeout } if ln.Config.HTTPWriteTimeout > 0 { - server.WriteTimeout = ln.Config.HTTPWriteTimeout + httpServer.WriteTimeout = ln.Config.HTTPWriteTimeout } if ln.Config.HTTPIdleTimeout > 0 { - server.IdleTimeout = ln.Config.HTTPIdleTimeout + httpServer.IdleTimeout = ln.Config.HTTPIdleTimeout } - l := tls.NewListener(ln.ProxyListener, &tls.Config{ - GetConfigForClient: w.getSessionTls, - }) + fetchCredsFn := func( + ctx context.Context, + _ nodee.Storage, + req *nodetypes.FetchNodeCredentialsRequest, + _ ...nodee.Option, + ) (*nodetypes.FetchNodeCredentialsResponse, error) { + client := w.controllerMultihopConn.Load() + if client == nil { + return nil, nodeauth.NewTempError(errors.New("error fetching controller connection, client is nil")) + } + multihopClient, ok := client.(multihop.MultihopServiceClient) + if !ok { + return nil, nodeauth.NewTempError(errors.New("client could not be understood as a multihop service client")) + } + return multihopClient.FetchNodeCredentials(ctx, req) + } - return func() { go server.Serve(l) }, nil + generateServerCertificatesFn := func( + ctx context.Context, + _ nodee.Storage, + req *nodetypes.GenerateServerCertificatesRequest, + _ ...nodee.Option, + ) (*nodetypes.GenerateServerCertificatesResponse, error) { + client := w.controllerMultihopConn.Load() + if client == nil { + return nil, nodeauth.NewTempError(errors.New("error fetching controller connection, client is nil")) + } + multihopClient, ok := client.(multihop.MultihopServiceClient) + if !ok { + return nil, nodeauth.NewTempError(errors.New("client could not be understood as a multihop service client")) + } + return multihopClient.GenerateServerCertificates(ctx, req) + } + + interceptingListener, err := nodeauth.NewInterceptingListener( + ln.ProxyListener, + &tls.Config{ + GetConfigForClient: w.getSessionTls, + }, + nodeauth.MakeCurrentParametersFactory( + w.baseContext, + nodee.NopTransactionStorage(w.NodeeFileStorage), + ), + fetchCredsFn, + generateServerCertificatesFn, + ) + if err != nil { + return nil, fmt.Errorf("error instantiating node auth listener: %w", err) + } + + w.nodeeTeeListener = nodeauth.NewTeeListener(interceptingListener) + + downstreamServer := grpc.NewServer( + grpc.MaxRecvMsgSize(math.MaxInt32), + grpc.MaxSendMsgSize(math.MaxInt32), + ) + multihopService := workers.NewMultihopServiceServer( + nodeauth.MakeCurrentParametersFactory(w.baseContext, nodee.NopTransactionStorage(w.NodeeFileStorage)), + ) + multihop.RegisterMultihopServiceServer(downstreamServer, multihopService) + + ln.GrpcServer = downstreamServer + + return func() { + go w.nodeeTeeListener.Start() + go httpServer.Serve(w.nodeeTeeListener.OtherListener()) + go ln.GrpcServer.Serve(w.nodeeTeeListener.NodeeListener()) + }, nil } func (w *Worker) stopServersAndListeners() error { - var closeErrors *multierror.Error - err := w.stopHttpServersAndListeners() - if err != nil { - closeErrors = multierror.Append(closeErrors, err) + var mg multierror.Group + mg.Go(w.stopHttpServer) + mg.Go(w.stopClusterGrpcServer) + + // FIXME (jeff): For some reason, unlike the controller, the grpc server + // really likes to hang on closing. Maybe because it's never served a + // connection? This is a workaround to force it until I can dig in. + var cancel context.CancelFunc + if w.nodeeTeeListener != nil { + var ctx context.Context + ctx, cancel = context.WithTimeout(w.baseContext, 2*time.Second) + go func() { + <-ctx.Done() + w.nodeeTeeListener.Stop() + cancel() + }() } - err = w.stopAnyListeners() - if err != nil { - closeErrors = multierror.Append(closeErrors, err) + stopErrors := mg.Wait() + + if w.nodeeTeeListener != nil { + cancel() + err := w.nodeeTeeListener.Stop() + if err != nil { + stopErrors = multierror.Append(stopErrors, err) + } } - return closeErrors.ErrorOrNil() + err := w.stopAnyListeners() + if err != nil { + stopErrors = multierror.Append(stopErrors, err) + } + + return stopErrors.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.ProxyListener.Close() - err = listenerCloseErrorCheck(ln.Config.Type, err) - if err != nil { - closeErrors = multierror.Append(closeErrors, err) - } +func (w *Worker) stopHttpServer() error { + if w.proxyListener == nil { + return nil } - return closeErrors.ErrorOrNil() + if w.proxyListener.HTTPServer == nil { + return nil + } + + ctx, cancel := context.WithTimeout(w.baseContext, w.proxyListener.Config.MaxRequestDuration) + w.proxyListener.HTTPServer.Shutdown(ctx) + cancel() + + return nil +} + +func (w *Worker) stopClusterGrpcServer() error { + if w.proxyListener == nil { + return nil + } + if w.proxyListener.GrpcServer == nil { + return nil + } + + w.proxyListener.GrpcServer.GracefulStop() + return nil } // 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 { + if w.proxyListener == nil { + return nil + } var closeErrors *multierror.Error - for _, ln := range w.listeners { - if ln == nil || ln.ProxyListener == nil { - continue - } - - err := ln.ProxyListener.Close() - err = listenerCloseErrorCheck(ln.Config.Type, err) - if err != nil { - closeErrors = multierror.Append(closeErrors, err) - } + var err error + if w.nodeeTeeListener != nil { + err = w.nodeeTeeListener.Stop() + } else if w.proxyListener.ProxyListener != nil { + err = w.proxyListener.ProxyListener.Close() + } + err = listenerCloseErrorCheck("proxy", err) + if err != nil { + closeErrors = multierror.Append(closeErrors, err) } return closeErrors.ErrorOrNil() diff --git a/internal/servers/worker/listeners_test.go b/internal/servers/worker/listeners_test.go index f8b6e2974..d48cb6827 100644 --- a/internal/servers/worker/listeners_test.go +++ b/internal/servers/worker/listeners_test.go @@ -3,13 +3,10 @@ package worker import ( "context" "fmt" - "io" "net" "net/http" "os" - "strconv" "testing" - "time" "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/go-secure-stdlib/listenerutil" @@ -18,30 +15,19 @@ import ( func TestStartListeners(t *testing.T) { testNonTlsRejected := func(t *testing.T, resp *http.Response, err error) { - require.NoError(t, err) - require.NotNil(t, resp) - defer resp.Body.Close() - body, err := io.ReadAll(resp.Body) - require.NoError(t, err) - require.Equal(t, "Client sent an HTTP request to an HTTPS server.\n", string(body)) + require.Error(t, err) } tests := []struct { - name string - listeners []*listenerutil.ListenerConfig - expErr bool - expErrMsg string - assertions func(t *testing.T, w *Worker, addrs []string) + name string + proxyListeners []*listenerutil.ListenerConfig + expErr bool + expErrMsg string + assertions func(t *testing.T, w *Worker, addr string) }{ { - name: "one tcp listener, one unix listener", - listeners: []*listenerutil.ListenerConfig{ - { - Type: "tcp", - Purpose: []string{"proxy"}, - Address: "127.0.0.1:0", - TLSDisable: true, - }, + name: "unix listener", + proxyListeners: []*listenerutil.ListenerConfig{ { Type: "unix", Purpose: []string{"proxy"}, @@ -50,68 +36,30 @@ func TestStartListeners(t *testing.T) { }, }, expErr: false, - assertions: func(t *testing.T, w *Worker, addrs []string) { - require.Len(t, addrs, 2) - - resp, err := http.Get("http://" + addrs[0] + "/v1/proxy/") - testNonTlsRejected(t, resp, err) - + assertions: func(t *testing.T, w *Worker, addr string) { cl := http.Client{ Transport: &http.Transport{ - Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", addrs[1]) }, + Dial: func(network, addr string) (net.Conn, error) { return net.Dial("unix", addr) }, }, } - resp, err = cl.Get("http://anything.domain/v1/proxy/") + resp, err := cl.Get("http://anything.domain/v1/proxy/") testNonTlsRejected(t, resp, err) }, }, { - name: "multiple tcp listeners, multiple unix listeners", - listeners: []*listenerutil.ListenerConfig{ + name: "tcp listener", + proxyListeners: []*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) - - resp, err := http.Get("http://" + addrs[0] + "/v1/proxy/") + assertions: func(t *testing.T, w *Worker, addr string) { + resp, err := http.Get("http://" + addr + "/v1/proxy/") testNonTlsRejected(t, resp, err) - - resp, err = http.Get("http://" + addrs[1] + "/v1/proxy/") - testNonTlsRejected(t, resp, err) - - 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) }, - }, - } - resp, err = cl.Get("http://anything.domain/v1/proxy") - testNonTlsRejected(t, resp, err) - } }, }, } @@ -119,7 +67,7 @@ func TestStartListeners(t *testing.T) { 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 + conf.RawConfig.SharedConfig.Listeners = tt.proxyListeners err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"proxy"}) require.NoError(t, err) @@ -136,12 +84,8 @@ func TestStartListeners(t *testing.T) { require.NoError(t, err) - addrs := make([]string, 0) - for _, l := range w.listeners { - addrs = append(addrs, l.ProxyListener.Addr().String()) - } if tt.assertions != nil { - tt.assertions(t, w, addrs) + tt.assertions(t, w, w.proxyListener.ProxyListener.Addr().String()) } }) } @@ -158,14 +102,14 @@ func TestStopHttpServersAndListeners(t *testing.T) { { name: "no listeners", workerFn: func(t *testing.T) *Worker { - return &Worker{listeners: []*base.ServerListener{}} + return &Worker{proxyListener: &base.ServerListener{}} }, expErr: false, }, { name: "nil listeners slice", workerFn: func(t *testing.T) *Worker { - return &Worker{listeners: nil} + return &Worker{proxyListener: nil} }, expErr: false, }, @@ -173,65 +117,42 @@ func TestStopHttpServersAndListeners(t *testing.T) { name: "listeners with nil http server", workerFn: func(t *testing.T) *Worker { return &Worker{ - listeners: []*base.ServerListener{ - {HTTPServer: nil}, - {HTTPServer: nil}, - {HTTPServer: nil}, + proxyListener: &base.ServerListener{ + HTTPServer: nil, }, } }, expErr: false, }, { - name: "multiple listeners", + name: "valid listener with http server", 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{ - { - ProxyListener: l1, - HTTPServer: s1, - Config: &listenerutil.ListenerConfig{Type: "tcp"}, - }, - { - ProxyListener: l2, - HTTPServer: s2, - Config: &listenerutil.ListenerConfig{Type: "tcp"}, - }, + proxyListener: &base.ServerListener{ + ProxyListener: l1, + HTTPServer: s1, + 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].ProxyListener), http.ErrServerClosed) - require.ErrorIs(t, w.listeners[1].HTTPServer.Serve(w.listeners[1].ProxyListener), http.ErrServerClosed) + require.ErrorIs(t, w.proxyListener.HTTPServer.Serve(w.proxyListener.ProxyListener), http.ErrServerClosed) // Asserts the underlying listeners are closed. - require.ErrorIs(t, w.listeners[0].ProxyListener.Close(), net.ErrClosed) - require.ErrorIs(t, w.listeners[1].ProxyListener.Close(), net.ErrClosed) + require.ErrorIs(t, w.proxyListener.ProxyListener.Close(), net.ErrClosed) }, expErr: false, }, @@ -240,7 +161,7 @@ func TestStopHttpServersAndListeners(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { w := tt.workerFn(t) - err := w.stopHttpServersAndListeners() + err := w.stopServersAndListeners() if tt.expErr { require.EqualError(t, err, tt.expErrStr) return @@ -265,66 +186,14 @@ func TestStopAnyListeners(t *testing.T) { { name: "nil listeners", workerFn: func(t *testing.T) *Worker { - return &Worker{listeners: nil} + return &Worker{proxyListener: 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: "non-empty but nil listeners", - workerFn: func(t *testing.T) *Worker { - return &Worker{listeners: []*base.ServerListener{ - {ProxyListener: nil}, {ProxyListener: nil}, {ProxyListener: 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"}, - ProxyListener: l1, - }, - { - Config: &listenerutil.ListenerConfig{Type: "tcp"}, - ProxyListener: l2, - }, - { - Config: &listenerutil.ListenerConfig{Type: "tcp"}, - ProxyListener: l3, - }, - }} - }, - assertions: func(t *testing.T, w *Worker) { - for i := range w.listeners { - ln := w.listeners[i] - require.ErrorIs(t, ln.ProxyListener.Close(), net.ErrClosed) - } + return &Worker{proxyListener: &base.ServerListener{}} }, expErr: false, }, diff --git a/internal/servers/worker/status.go b/internal/servers/worker/status.go index 45cb4496c..88b349f2e 100644 --- a/internal/servers/worker/status.go +++ b/internal/servers/worker/status.go @@ -90,6 +90,14 @@ func (w *Worker) WaitForNextSuccessfulStatusUpdate() error { func (w *Worker) sendWorkerStatus(cancelCtx context.Context) { const op = "worker.(Worker).sendWorkerStatus" + + // If we've never managed to successfully authenticate then we won't have + // any session information anyways and this will produce a ton of noise in + // observability, so suppress it + if !w.everAuthenticated.Load() { + return + } + // First send info as-is. We'll perform cleanup duties after we // get cancel/job change info back. var activeJobs []*pbs.JobStatus diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index beca2f74c..8f87cf731 100644 --- a/internal/servers/worker/worker.go +++ b/internal/servers/worker/worker.go @@ -25,6 +25,9 @@ import ( "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" + "github.com/hashicorp/nodeenrollment/nodeauth" + "github.com/hashicorp/nodeenrollment/noderegistration" + nodeefile "github.com/hashicorp/nodeenrollment/nodestorage/file" ua "go.uber.org/atomic" "google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver/manual" @@ -43,6 +46,7 @@ type Worker struct { tickerWg sync.WaitGroup controllerStatusConn *atomic.Value + everAuthenticated *ua.Bool lastStatusSuccess *atomic.Value workerStartTime time.Time @@ -51,7 +55,9 @@ type Worker struct { controllerSessionConn *atomic.Value sessionInfoMap *sync.Map - listeners []*base.ServerListener + controllerMultihopConn *atomic.Value + + proxyListener *base.ServerListener // Used to generate a random nonce for Controller connections nonceFn randFn @@ -62,7 +68,12 @@ type Worker struct { // This stores whether or not to send updated tags on the next status // request. It can be set via startup in New below, or (eventually) via // SIGHUP. - updateTags ua.Bool + updateTags *ua.Bool + + // PoC: Testing bits for BYOW + NodeeFileStorage *nodeefile.FileStorage + NodeeKeyId string + nodeeTeeListener *nodeauth.TeeListener // Test-specific options TestOverrideX509VerifyDnsName string @@ -75,16 +86,19 @@ func New(conf *Config) (*Worker, error) { metric.InitializeClusterClientCollectors(conf.PrometheusRegisterer) w := &Worker{ - conf: conf, - logger: conf.Logger.Named("worker"), - started: ua.NewBool(false), - controllerStatusConn: new(atomic.Value), - lastStatusSuccess: new(atomic.Value), - controllerResolver: new(atomic.Value), - controllerSessionConn: new(atomic.Value), - sessionInfoMap: new(sync.Map), - tags: new(atomic.Value), - nonceFn: base62.Random, + conf: conf, + logger: conf.Logger.Named("worker"), + started: ua.NewBool(false), + controllerStatusConn: new(atomic.Value), + everAuthenticated: ua.NewBool(false), + lastStatusSuccess: new(atomic.Value), + controllerResolver: new(atomic.Value), + controllerSessionConn: new(atomic.Value), + sessionInfoMap: new(sync.Map), + controllerMultihopConn: new(atomic.Value), + tags: new(atomic.Value), + updateTags: ua.NewBool(false), + nonceFn: base62.Random, } w.lastStatusSuccess.Store((*LastStatusInformation)(nil)) @@ -121,6 +135,7 @@ func New(conf *Config) (*Worker, error) { } } + var listenerCount int for i := range conf.Listeners { l := conf.Listeners[i] if l == nil || l.Config == nil || l.Config.Purpose == nil { @@ -131,11 +146,14 @@ func New(conf *Config) (*Worker, error) { } switch l.Config.Purpose[0] { case "proxy": - w.listeners = append(w.listeners, l) + if w.proxyListener == nil { + w.proxyListener = l + } + listenerCount++ } } - if len(w.listeners) == 0 { - return nil, fmt.Errorf("no proxy listeners found") + if listenerCount != 1 { + return nil, fmt.Errorf("exactly one proxy listener is required") } return w, nil @@ -143,13 +161,14 @@ func New(conf *Config) (*Worker, error) { func (w *Worker) Start() error { const op = "worker.(Worker).Start" + + w.baseContext, w.baseCancel = context.WithCancel(context.Background()) + if w.started.Load() { event.WriteSysEvent(w.baseContext, op, "already started, skipping") return nil } - w.baseContext, w.baseCancel = context.WithCancel(context.Background()) - scheme := strconv.FormatInt(time.Now().UnixNano(), 36) controllerResolver := manual.NewBuilderWithScheme(scheme) w.controllerResolver.Store(controllerResolver) @@ -157,6 +176,18 @@ func (w *Worker) Start() error { if err := w.startListeners(); err != nil { return fmt.Errorf("error starting worker listeners: %w", err) } + + var err error + w.NodeeFileStorage, err = nodeefile.NewFileStorage(w.baseContext) + if err != nil { + return err + } + + w.NodeeKeyId, err = noderegistration.GenerateNodeCredentialsForRegistration(w.baseContext, w.NodeeFileStorage) + if err != nil { + return err + } + if err := w.startControllerConnections(); err != nil { return fmt.Errorf("error making controller connections: %w", err) } diff --git a/internal/servers/worker/worker_test.go b/internal/servers/worker/worker_test.go index 8ae62a97d..2f04c2fc2 100644 --- a/internal/servers/worker/worker_test.go +++ b/internal/servers/worker/worker_test.go @@ -23,13 +23,13 @@ func TestWorkerNew(t *testing.T) { name: "nil listeners", in: &Config{Server: &base.Server{Listeners: nil}}, expErr: true, - expErrMsg: "no proxy listeners found", + expErrMsg: "exactly one proxy listener is required", }, { name: "zero listeners", in: &Config{Server: &base.Server{Listeners: []*base.ServerListener{}}}, expErr: true, - expErrMsg: "no proxy listeners found", + expErrMsg: "exactly one proxy listener is required", }, { name: "populated with nil values", @@ -43,7 +43,7 @@ func TestWorkerNew(t *testing.T) { }, }, expErr: true, - expErrMsg: "no proxy listeners found", + expErrMsg: "exactly one proxy listener is required", }, { name: "multiple purposes", @@ -60,7 +60,7 @@ func TestWorkerNew(t *testing.T) { expErrMsg: `found listener with multiple purposes "api,proxy"`, }, { - name: "valid listeners", + name: "too many proxy listeners", in: &Config{ Server: &base.Server{ Listeners: []*base.ServerListener{ @@ -71,10 +71,21 @@ func TestWorkerNew(t *testing.T) { }, }, }, - expErr: false, - assertions: func(t *testing.T, w *Worker) { - require.Len(t, w.listeners, 2) + expErr: true, + expErrMsg: "exactly one proxy listener is required", + }, + { + 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{"cluster"}}}, + }, + }, }, + expErr: false, }, { name: "worker nonce func is set",