CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial BYOW auth PoC (#29)

Browse Source
This commit is contained in:
Jeff Mitchell committed 2022-05-09 13:28:17 -04:00
1 parent 8ed560fc3e
commit f748af7b02
21 files changed
+707 -404
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

go.mod
+6 -7
View File
@@ -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
go.sum
+8 -5
View File
@@ -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=
internal/cmd/base/option.go
+27 -18
View File
@@ -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
}
}
internal/cmd/base/servers.go
+8 -2
View File
@@ -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
internal/cmd/commands/dev/dev.go
+100 -53
View File
@@ -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]))
}
}
internal/cmd/commands/server/listener_reload_test.go
+4 -1
View File
@@ -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)
internal/cmd/commands/server/server.go
+22 -10
View File
@@ -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...)
internal/cmd/commands/server/worker_shutdown_reload_test.go
+4
View File
@@ -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)
internal/cmd/commands/server/worker_tags_reload_test.go
+5 -5
View File
@@ -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"]
}
internal/servers/controller/controller.go
+22
View File
@@ -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)
}
internal/servers/controller/handler.go
+65
View File
@@ -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) {
internal/servers/controller/handlers/workers/multihop_service.go
+45
View File
@@ -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...)
}
internal/servers/controller/intercepting_listener.go
+39 -24
View File
@@ -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 {
internal/servers/controller/listeners.go
+24 -3
View File
@@ -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
internal/servers/controller/worker_tls_config.go
+2 -1
View File
@@ -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:
internal/servers/worker/controller_connection.go
+68 -32
View File
@@ -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
}
internal/servers/worker/listeners.go
+153 -57
View File
@@ -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()
internal/servers/worker/listeners_test.go
+31 -162
View File
@@ -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,
},
internal/servers/worker/status.go
+8
View File
@@ -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
internal/servers/worker/worker.go
+48 -17
View File
@@ -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)
}
internal/servers/worker/worker_test.go
+18 -7
View File
@@ -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",