KMS-PKI Workers (#3101)

This adds a new mechanism for worker authentication that uses the third-party arbitration of the original KMS flow by leveraging recently-added capabilities in nodeenrollment to support a registration wrapper. This allows for workers to be registered to the system with just a name and a matching wrapper, while at the same time allowing these workers to be used for private Vault access and multi-hop as they contain per-worker encryption keys and support rotation. Nodes using this mode will generate and rotate credentials in-memory (using another recently added nodeenrollment capability) so will have a new set of keys on every startup. This has the side effect of ensuring that there is never a conflict or re-use of these credentials. Note that the normal test worker flow was using the now-deprecated KMS method; there is still a test that uses this (explicitly) but all other tests using test workers that did not specify a specific PKI flow now actually use this mechanism. This means there is fairly decent test coverage in a general sense, plus the specific tests updated/added for this feature. This also adds a new KMS type that can be used to have distinct KMSes for upstream authentication vs. accepting from downstream.
2023-03-30 16:20:28 -04:00
1 parent 16f48686a7
commit 163ce184b8
@@ -1,10 +1,17 @@
 # Boundary CHANGELOG

 Canonical reference for changes, improvements, and bugfixes for Boundary.
+
 ## Next

 ### Deprecations/Changes

+* With the introduction of the new KMS variant for worker registration (as
+  described below), using the deprecated behavior requires opting-in. This is
+  only recommended if compatibility with pre-0.13 workers using the KMS auth
+  method is required. Requiring opting in removes some potentially confusing
+  behavior for deciding when to use the old versus new mechanism. To opt in, add
+  `use_deprecated_kms_auth_method = true` to the `worker` config block.
 * When grants are added to roles additional validity checking is now performed.
  This extra validity checking is designed to reject grants that are not
  [documented grant
@@ -17,6 +24,12 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.

 ### New and Improved

+* KMS workers: KMS workers now have feature parity with PKI workers (they
+  support multi-hop and Vault private access) and support separate KMSes for
+  authenticating downstreams across different networks. See the [worker
+  configuration
+  documentation](https://developer.hashicorp.com/boundary/docs/configuration/worker)
+  for more information. ([PR](https://github.com/hashicorp/boundary/pull/3101))
 * roles: Perform additional validity checking on grants at submission time
  ([PR](https://github.com/hashicorp/boundary/pull/3081))

@@ -37,7 +50,7 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
  credential type. [PR](https://github.com/hashicorp/boundary/pull/2989)
 * ui: Fix credential library not saving correctly when trying to save it as a
  generic secrets type. ([PR](https://github.com/hashicorp/boundary-ui/pull/1640))
-* sessions: Fix tofu token retreival. ([PR](https://github.com/hashicorp/boundary/pull/3064))
+* sessions: Fix tofu token retrieval. ([PR](https://github.com/hashicorp/boundary/pull/3064))

 ## 0.12.0 (2023/01/24)

@@ -4,10 +4,11 @@
 package globals

 const (
-	KmsPurposeRoot              = "root"
-	KmsPurposePreviousRoot      = "previous-root"
-	KmsPurposeWorkerAuth        = "worker-auth"
-	KmsPurposeWorkerAuthStorage = "worker-auth-storage"
-	KmsPurposeRecovery          = "recovery"
-	KmsPurposeConfig            = "config"
+	KmsPurposeRoot                 = "root"
+	KmsPurposePreviousRoot         = "previous-root"
+	KmsPurposeWorkerAuth           = "worker-auth"
+	KmsPurposeWorkerAuthStorage    = "worker-auth-storage"
+	KmsPurposeRecovery             = "recovery"
+	KmsPurposeConfig               = "config"
+	KmsPurposeDownstreamWorkerAuth = "downstream-worker-auth"
 )
@@ -96,7 +96,7 @@ require (
 	github.com/hashicorp/cap/ldap v0.0.0-20230123181313-9c0fb924b0d9
 	github.com/hashicorp/go-kms-wrapping/extras/kms/v2 v2.0.0-20221122211539-47c893099f13
 	github.com/hashicorp/go-version v1.3.0
-	github.com/hashicorp/nodeenrollment v0.1.19
+	github.com/hashicorp/nodeenrollment v0.2.0
 	github.com/jimlambrt/gldap v0.1.2
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/mikesmitty/edkey v0.0.0-20170222072505-3356ea4e686a
@@ -180,7 +180,7 @@ require (
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
-	github.com/rogpeppe/go-internal v1.8.1 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/russross/blackfriday/v2 v2.0.1 // indirect
 	github.com/sethvargo/go-diceware v0.3.0 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
@@ -757,8 +757,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.1.19 h1:RZIYFVfjWlK8MZn7owEEmaoDLHMqROL2K9RkJz2Q5cA=
-github.com/hashicorp/nodeenrollment v0.1.19/go.mod h1:N5gYsm8mWiDfIw/j+1IQ6NBO1cWCmhPpvQ9GB1QUnsU=
+github.com/hashicorp/nodeenrollment v0.2.0 h1:79FbU+2MZaHgFxirQM1G9rU71FxCfXQ2hinEIR/xTco=
+github.com/hashicorp/nodeenrollment v0.2.0/go.mod h1:Do2FvaFHgnnMaqhdSTBX+TkQP0ep8VVQ0rOr+ZjUbJU=
 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=
@@ -1140,7 +1140,6 @@ github.com/pires/go-proxyproto v0.6.1/go.mod h1:Odh9VFOZJCf9G8cLW5o435Xf1J95Jw9G
 github.com/pkg/browser v0.0.0-20210706143420-7d21f8c997e2/go.mod h1:HKlIX3XHQyzLZPlr7++PzdhaXEj94dEiJgZDTsxEqUI=
 github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8/go.mod h1:HKlIX3XHQyzLZPlr7++PzdhaXEj94dEiJgZDTsxEqUI=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e h1:aoZm08cpOy4WuID//EZDgcC4zIxODThtZNPirFr42+A=
-github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1-0.20171018195549-f15c970de5b7/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -1202,8 +1201,8 @@ github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.6.2/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
-github.com/rogpeppe/go-internal v1.8.1 h1:geMPLpDpQOgVyCg5z5GoRwLHepNdb71NXb67XFkP+Eg=
-github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o=
+github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/rs/zerolog v1.13.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU=
 github.com/rs/zerolog v1.15.0/go.mod h1:xYTKnLHcpfU2225ny5qZjxnj9NvkumZYjJHlAThCjNc=
@@ -1952,7 +1951,6 @@ gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=
-gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
@@ -25,25 +25,24 @@ 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
-	withSkipWorkerAuthKmsInstantiation 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
 }

 func getDefaultOptions() Options {
@@ -192,11 +191,3 @@ 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
-	}
-}
@@ -18,6 +18,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"

@@ -76,12 +77,18 @@ type Server struct {
 	StderrLock *sync.Mutex
 	Eventer    *event.Eventer

-	RootKms              wrapping.Wrapper
-	WorkerAuthKms        wrapping.Wrapper
-	WorkerAuthStorageKms wrapping.Wrapper
-	RecoveryKms          wrapping.Wrapper
-	Kms                  *kms.Kms
-	SecureRandomReader   io.Reader
+	// NOTE: Unlike the other wrappers below, if set, DownstreamWorkerAuthKms
+	// should always be a PooledWrapper, so that we can allow multiple KMSes to
+	// accept downstream connections. As such it's made explicit here.
+	DownstreamWorkerAuthKms *multi.PooledWrapper
+	RootKms                 wrapping.Wrapper
+	WorkerAuthKms           wrapping.Wrapper
+	WorkerAuthStorageKms    wrapping.Wrapper
+	RecoveryKms             wrapping.Wrapper
+	Kms                     *kms.Kms
+	SecureRandomReader      io.Reader
+
+	WorkerAuthDebuggingEnabled *atomic.Bool

 	PrometheusRegisterer prometheus.Registerer

@@ -118,10 +125,6 @@ type Server struct {
 	DevTargetSessionConnectionLimit  int
 	DevLoopbackHostPluginId          string

-	// DevUsePkiForUpstream is a hint that we are in dev mode and have a worker
-	// auth KMS but want to use PKI for upstream connections
-	DevUsePkiForUpstream bool
-
 	EnabledPlugins []EnabledPlugin
 	HostPlugins    map[string]plgpb.HostPluginServiceClient

@@ -140,15 +143,16 @@ type Server struct {
 // NewServer creates a new Server.
 func NewServer(cmd *Command) *Server {
 	return &Server{
-		Command:              cmd,
-		ServerSideShutdownCh: make(chan struct{}),
-		InfoKeys:             make([]string, 0, 20),
-		Info:                 make(map[string]string),
-		SecureRandomReader:   rand.Reader,
-		ReloadFuncsLock:      new(sync.RWMutex),
-		ReloadFuncs:          make(map[string][]reloadutil.ReloadFunc),
-		StderrLock:           new(sync.Mutex),
-		PrometheusRegisterer: prometheus.DefaultRegisterer,
+		Command:                    cmd,
+		ServerSideShutdownCh:       make(chan struct{}),
+		InfoKeys:                   make([]string, 0, 20),
+		Info:                       make(map[string]string),
+		SecureRandomReader:         rand.Reader,
+		ReloadFuncsLock:            new(sync.RWMutex),
+		ReloadFuncs:                make(map[string][]reloadutil.ReloadFunc),
+		StderrLock:                 new(sync.Mutex),
+		WorkerAuthDebuggingEnabled: new(atomic.Bool),
+		PrometheusRegisterer:       prometheus.DefaultRegisterer,
 	}
 }

@@ -546,25 +550,22 @@ 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, opt ...Option) error {
-	opts := getOpts(opt...)
-
 	sharedConfig := config.SharedConfig
 	var pluginLogger hclog.Logger
 	var err error
 	var previousRootKms wrapping.Wrapper
+	purposeCount := map[string]uint{}
 	for _, kms := range sharedConfig.Seals {
 		for _, purpose := range kms.Purpose {
 			purpose = strings.ToLower(purpose)
+			purposeCount[purpose] = purposeCount[purpose] + 1
 			switch purpose {
 			case "":
 				return errors.New("KMS block missing 'purpose'")
-			case globals.KmsPurposeWorkerAuth:
-				if opts.withSkipWorkerAuthKmsInstantiation {
-					continue
-				}
 			case globals.KmsPurposeRoot,
 				globals.KmsPurposePreviousRoot,
 				globals.KmsPurposeConfig,
+				globals.KmsPurposeWorkerAuth,
 				globals.KmsPurposeWorkerAuthStorage:
 			case globals.KmsPurposeRecovery:
 				if config.Controller != nil && config.DevRecoveryKey != "" {
@@ -595,7 +596,7 @@ func (b *Server) SetupKMSes(ctx context.Context, ui cli.Ui, config *config.Confi
 					pluginutil.WithPluginsFilesystem(kms_plugin_assets.KmsPluginPrefix, kms_plugin_assets.FileSystem()),
 					pluginutil.WithPluginExecutionDirectory(config.Plugins.ExecutionDir),
 				),
-				configutil.WithLogger(pluginLogger.Named(kms.Type).With("purpose", purpose)),
+				configutil.WithLogger(pluginLogger.Named(kms.Type).With("purpose", fmt.Sprintf("%s-%d", purpose, purposeCount[purpose]))),
 			)
 			if wrapperConfigError != nil {
 				return fmt.Errorf(
@@ -641,6 +642,21 @@ func (b *Server) SetupKMSes(ctx context.Context, ui cli.Ui, config *config.Confi
 					return fmt.Errorf("Duplicate KMS block for purpose '%s'. You may need to remove all but the last KMS block for this purpose.", purpose)
 				}
 				b.WorkerAuthKms = wrapper
+			case globals.KmsPurposeDownstreamWorkerAuth:
+				if b.DownstreamWorkerAuthKms == nil {
+					b.DownstreamWorkerAuthKms, err = multi.NewPooledWrapper(ctx, wrapper)
+					if err != nil {
+						return fmt.Errorf("Error instantiating pooled wrapper for downstream worker auth: %w.", err)
+					}
+				} else {
+					added, err := b.DownstreamWorkerAuthKms.AddWrapper(ctx, wrapper)
+					if err != nil {
+						return fmt.Errorf("Error adding additional wrapper to downstream worker auth wrapper pool: %w.", err)
+					}
+					if !added {
+						return fmt.Errorf("Wrapper already added to downstream worker auth wrapper pool.")
+					}
+				}
 			case globals.KmsPurposeWorkerAuthStorage:
 				if b.WorkerAuthStorageKms != nil {
 					return fmt.Errorf("Duplicate KMS block for purpose '%s'. You may need to remove all but the last KMS block for this purpose.", purpose)
@@ -29,6 +29,7 @@ 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"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mitchellh/cli"
 	"github.com/mr-tron/base58"
@@ -37,6 +38,13 @@ import (
 	"google.golang.org/protobuf/proto"
 )

+const (
+	PkiControllerLedWorkerAuthMechanism = "pki-controller-led"
+	PkiWorkerLedWorkerAuthMechanism     = "pki-worker-led"
+	PkiKmsWorkerAuthMechanism           = "pki-kms"
+	PkiRandomWorkerAuthMechanism        = "pki"
+)
+
 var (
 	_ cli.Command             = (*Command)(nil)
 	_ cli.CommandAutocomplete = (*Command)(nil)
@@ -95,6 +103,7 @@ type Command struct {
 	flagWorkerAuthStorageDir         string
 	flagWorkerAuthStorageSkipCleanup bool
 	flagWorkerAuthRotationInterval   time.Duration
+	flagWorkerAuthDebuggingEnabled   bool
 }

 func (c *Command) Synopsis() string {
@@ -346,23 +355,27 @@ 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.StringVar(&base.StringVar{
 		Name:    "worker-auth-method",
 		Target:  &c.flagWorkerAuthMethod,
-		Default: "random-pki",
-		Usage:   `Allows specifying how the generated worker will authenticate to the controller. Valid values are "kms" for the KMS-based mechanism; "pki-controller-led" for the PKI mechanism via the server-led authorization flow; "pki-worker-led" for the PKI mechanism via the worker-led authorization flow; and "random-pki" to randomly choose one of the PKI methods.`,
+		Default: PkiRandomWorkerAuthMechanism,
+		Usage:   `Allows specifying how the generated worker will authenticate to the controller. Valid values are "kms" for the deprecated KMS-based mechanism; "pki-kms" for the new-style KMS mechanism; "pki-controller-led" for the PKI mechanism via the server-led authorization flow; "pki-worker-led" for the PKI mechanism via the worker-led authorization flow; and "pki" to randomly choose one of the PKI methods.`,
 	})
-
 	f.StringVar(&base.StringVar{
 		Name:   "worker-auth-storage-dir",
 		Target: &c.flagWorkerAuthStorageDir,
-		Usage:  "Specifies the directory to store worker authentication credentials in dev mode.",
+		Usage:  "Specifies the directory to store worker authentication credentials in dev mode. Setting this will make use of file storage at the specified location; otherwise in-memory storage or a temporary directory will be used.",
 	})
-
 	f.BoolVar(&base.BoolVar{
 		Name:   "worker-auth-storage-skip-cleanup",
 		Target: &c.flagWorkerAuthStorageSkipCleanup,
-		Usage:  "Prevents deletion of temp worker credential storage directory if set.",
+		Usage:  "Prevents deletion of worker credential storage directory if set. Has no effect unless worker-auth-storage-dir is specified.",
+	})
+	f.BoolVar(&base.BoolVar{
+		Name:   "worker-auth-enable-debugging",
+		Target: &c.flagWorkerAuthDebuggingEnabled,
+		Usage:  "Turn on debug logging of the worker authentication process.",
 	})

 	f.BoolVar(&base.BoolVar{
@@ -427,6 +440,8 @@ func (c *Command) Run(args []string) int {
 		}
 	}

+	c.WorkerAuthDebuggingEnabled.Store(c.flagWorkerAuthDebuggingEnabled)
+
 	c.DevLoginName = c.flagLoginName
 	c.DevPassword = c.flagPassword
 	c.DevUnprivilegedLoginName = c.flagUnprivilegedLoginName
@@ -614,26 +629,30 @@ 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
-	}
-	c.InfoKeys = append(c.InfoKeys, "[Worker-Auth] AEAD Key Bytes")
-	c.Info["[Worker-Auth] AEAD Key Bytes"] = c.Config.DevWorkerAuthKey

 	if c.flagWorkerAuthMethod != "kms" {
-		c.DevUsePkiForUpstream = true
-		// These must be unset for PKI
-		c.Config.Worker.Name = ""
-		c.Config.Worker.Description = ""
-	} else {
-		// This must be unset for KMS
-		c.WorkerAuthStorageKms = nil
+		// Flip a coin to decide between file storage and inmem. It's
+		// transparent to users, but keeps both exercised.
+		randStorage := rand.New(rand.NewSource(time.Now().UnixMicro())).Intn(2)
+		if randStorage == 0 {
+			const pattern = "nodeenrollment"
+			c.Config.Worker.AuthStoragePath, err = os.MkdirTemp("", pattern)
+			if err != nil {
+				c.UI.Error(fmt.Errorf("Error creating temp dir: %w", err).Error())
+				return base.CommandCliError
+			}
+			if !c.flagWorkerAuthStorageSkipCleanup && c.flagWorkerAuthStorageDir != "" {
+				c.ShutdownFuncs = append(c.ShutdownFuncs, func() error { return os.RemoveAll(c.Config.Worker.AuthStoragePath) })
+			}
+		}
 	}
+
 	c.InfoKeys = append(c.InfoKeys, "[Controller] AEAD Key Bytes")
 	c.Info["[Controller] AEAD Key Bytes"] = c.Config.DevControllerKey
 	c.InfoKeys = append(c.InfoKeys, "[Recovery] AEAD Key Bytes")
 	c.Info["[Recovery] AEAD Key Bytes"] = c.Config.DevRecoveryKey
+	c.InfoKeys = append(c.InfoKeys, "[Worker-Auth] AEAD Key Bytes")
+	c.Info["[Worker-Auth] AEAD Key Bytes"] = c.Config.DevWorkerAuthKey

 	// Initialize the listeners
 	if err := c.SetupListeners(c.UI, c.Config.SharedConfig, []string{"api", "cluster", "proxy", "ops"}); err != nil {
@@ -727,51 +746,58 @@ func (c *Command) Run(args []string) int {
 			c.worker.TestOverrideAuthRotationPeriod = c.flagWorkerAuthRotationInterval
 		}

-		// Note: this should be done before starting the worker so that the
-		// activation token is populated
-		var useWorkerLed bool
-		if c.flagWorkerAuthMethod != "kms" {
+		if c.flagWorkerAuthMethod == PkiRandomWorkerAuthMechanism {
 			// Flip a coin. Use one method or the other; it's transparent to
 			// users, but keeps both exercised.
-			if c.flagWorkerAuthMethod == "random-pki" {
-				randPki := rand.New(rand.NewSource(time.Now().UnixMicro())).Intn(2)
-				if randPki == 0 {
-					c.flagWorkerAuthMethod = "pki-controller-led"
-				} else {
-					c.flagWorkerAuthMethod = "pki-worker-led"
-				}
-			}
-			switch c.flagWorkerAuthMethod {
-			case "pki-controller-led":
-				// Controller-led
-				serversRepo, err := c.controller.ServersRepoFn()
-				if err != nil {
-					c.UI.Error(fmt.Errorf("Error instantiating server repo: %w", err).Error())
-					if err := c.controller.Shutdown(); err != nil {
-						c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error())
-					}
-					return base.CommandCliError
-				}
-
-				// Create the worker in the database and fetch an activation token
-				worker, err := serversRepo.CreateWorker(c.Context, &server.Worker{
-					Worker: &store.Worker{
-						ScopeId: scope.Global.String(),
-					},
-				}, server.WithCreateControllerLedActivationToken(true))
-				if err != nil {
-					c.UI.Error(fmt.Errorf("Error creating worker in database: %w", err).Error())
-					if err := c.controller.Shutdown(); err != nil {
-						c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error())
-					}
-					return base.CommandCliError
-				}
-
-				// Set the activation token in the config
-				conf.RawConfig.Worker.ControllerGeneratedActivationToken = worker.ControllerGeneratedActivationToken
-
+			randPki := rand.New(rand.NewSource(time.Now().UnixMicro())).Intn(3)
+			switch randPki {
+			case 0:
+				c.flagWorkerAuthMethod = PkiControllerLedWorkerAuthMechanism
+			case 1:
+				c.flagWorkerAuthMethod = PkiWorkerLedWorkerAuthMechanism
 			default:
-				useWorkerLed = true
+				c.flagWorkerAuthMethod = PkiKmsWorkerAuthMechanism
+			}
+		}
+		switch c.flagWorkerAuthMethod {
+		case PkiControllerLedWorkerAuthMechanism:
+			// Controller-led
+			serversRepo, err := c.controller.ServersRepoFn()
+			if err != nil {
+				c.UI.Error(fmt.Errorf("Error instantiating server repo: %w", err).Error())
+				if err := c.controller.Shutdown(); err != nil {
+					c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error())
+				}
+				return base.CommandCliError
+			}
+
+			// Create the worker in the database and fetch an activation token
+			worker, err := serversRepo.CreateWorker(c.Context, &server.Worker{
+				Worker: &store.Worker{
+					ScopeId: scope.Global.String(),
+				},
+			}, server.WithCreateControllerLedActivationToken(true))
+			if err != nil {
+				c.UI.Error(fmt.Errorf("Error creating worker in database: %w", err).Error())
+				if err := c.controller.Shutdown(); err != nil {
+					c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error())
+				}
+				return base.CommandCliError
+			}
+
+			// Set the activation token in the config and nil out the worker
+			// auth KMS so we don't use it via PKI-KMS
+			c.WorkerAuthKms = nil
+			conf.RawConfig.Worker.ControllerGeneratedActivationToken = worker.ControllerGeneratedActivationToken
+
+		case PkiWorkerLedWorkerAuthMechanism:
+			// Clear this out as presence of it causes PKI-KMS behavior
+			c.WorkerAuthKms = nil
+
+		case PkiKmsWorkerAuthMechanism, "kms":
+			if c.WorkerAuthKms == nil {
+				c.UI.Error("Worker Auth KMS not found after parsing KMS blocks")
+				return base.CommandUserError
 			}
 		}

@@ -792,28 +818,35 @@ func (c *Command) Run(args []string) int {
 			c.InfoKeys = append(c.InfoKeys, "worker auth current key id")
 			c.Info["worker auth current key id"] = c.worker.WorkerAuthCurrentKeyId.Load()
 			c.InfoKeys = append(c.InfoKeys, "worker auth storage path")
-			c.Info["worker auth storage path"] = c.worker.WorkerAuthStorage.BaseDir()
+			if c.Config.Worker.AuthStoragePath != "" {
+				c.Info["worker auth storage path"] = c.Config.Worker.AuthStoragePath
+			} else {
+				c.Info["worker auth storage path"] = "(in-memory)"
+			}

-			if useWorkerLed {
+			if c.flagWorkerAuthMethod == PkiWorkerLedWorkerAuthMechanism {
 				req := c.worker.WorkerAuthRegistrationRequest
 				if req == "" {
 					c.UI.Error("No worker auth registration request found at worker start time")
 					return base.CommandCliError
 				}

-				if err := c.StoreWorkerAuthReq(c.worker.WorkerAuthRegistrationRequest, c.worker.WorkerAuthStorage.BaseDir()); err != nil {
-					// Shutdown on failure
-					retErr := fmt.Errorf("Error storing worker auth request: %w", err)
-					if err := c.worker.Shutdown(); err != nil {
+				if c.Config.Worker.AuthStoragePath != "" {
+					if err := c.StoreWorkerAuthReq(c.worker.WorkerAuthRegistrationRequest, c.Config.Worker.AuthStoragePath); err != nil {
+						// Shutdown on failure
+						retErr := fmt.Errorf("Error storing worker auth request: %w", err)
+						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())
-						retErr = fmt.Errorf("Error shutting down worker: %w", err)
+						if err := c.controller.Shutdown(); err != nil {
+							c.UI.Error(fmt.Errorf("Error with controller shutdown: %w", err).Error())
+						}
+						return base.CommandCliError
 					}
-					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
 				}
+
 				go func() {
 					for {
 						select {
@@ -843,7 +876,6 @@ func (c *Command) Run(args []string) int {
 	if err != nil {
 		c.UI.Error(fmt.Errorf("Failed to start ops listeners: %w", err).Error())
 		return base.CommandCliError
-
 	}
 	c.opsServer = opsServer
 	c.opsServer.Start()
@@ -857,7 +889,11 @@ func (c *Command) Run(args []string) int {
 			c.UI.Error(fmt.Errorf("Error shutting down worker: %w", err).Error())
 		}
 		if !c.flagWorkerAuthStorageSkipCleanup && c.worker.WorkerAuthStorage != nil {
-			c.worker.WorkerAuthStorage.Cleanup()
+			if cleanable, ok := c.worker.WorkerAuthStorage.(nodeenrollment.CleanableStorage); ok {
+				if err := cleanable.Cleanup(c.Context); err != nil {
+					c.UI.Error(fmt.Errorf("Error cleaning up authentication storage: %w", err).Error())
+				}
+			}
 		}
 	}
 	workerGracefulShutdownFunc := func() {
@@ -177,6 +177,7 @@ func (c *Command) Run(args []string) int {
 		c.UI.Error(err.Error())
 		return base.CommandUserError
 	}
+	c.WorkerAuthDebuggingEnabled.Store(c.Config.EnableWorkerAuthDebugging)

 	base.StartMemProfiler(c.Context)

@@ -200,7 +201,7 @@ func (c *Command) Run(args []string) int {
 				return base.CommandUserError
 			}
 			if c.Config.Worker.Name != "" || c.Config.Worker.Description != "" {
-				c.UI.Error("Worker config cannot contain name or description when using PKI-based worker authentication; it must be set via the API.")
+				c.UI.Error("Worker config cannot contain name or description when using activation-token-based worker authentication; it must be set via the API.")
 				return base.CommandUserError
 			}
 		default:
@@ -212,6 +213,10 @@ func (c *Command) Run(args []string) int {
 				c.UI.Error("Worker has KMS auth info but also has a controller-generated activation token set, which is incompatible.")
 				return base.CommandUserError
 			}
+			if c.Config.Worker.AuthStoragePath != "" {
+				c.UI.Error("Worker has KMS auth info but also has an auth storage path set, which is incompatible.")
+				return base.CommandUserError
+			}
 		}
 		if c.Config.Controller != nil {
 			if c.Config.Worker.Name == c.Config.Controller.Name {
@@ -478,7 +483,7 @@ func (c *Command) Run(args []string) int {
 			c.Info["worker auth current key id"] = c.worker.WorkerAuthCurrentKeyId.Load()

 			// Write the WorkerAuth request to a file
-			if err := c.StoreWorkerAuthReq(c.worker.WorkerAuthRegistrationRequest, c.worker.WorkerAuthStorage.BaseDir()); err != nil {
+			if err := c.StoreWorkerAuthReq(c.worker.WorkerAuthRegistrationRequest, c.Config.Worker.AuthStoragePath); err != nil {
 				// Shutdown on failure
 				retErr := fmt.Errorf("Error storing worker auth request: %w", err)
 				if err := c.worker.Shutdown(); err != nil {
@@ -570,6 +575,7 @@ func (c *Command) reloadConfig() (*config.Config, int) {
 			return nil, base.CommandUserError
 		}
 	}
+
 	return cfg, 0
 }

@@ -618,12 +624,12 @@ func (c *Command) StartWorker() error {
 		return retErr
 	}

-	if c.WorkerAuthKms == nil || c.DevUsePkiForUpstream {
+	if c.WorkerAuthKms == nil {
 		if c.worker.WorkerAuthStorage == nil {
-			return fmt.Errorf("No worker auth storage found")
+			return fmt.Errorf("Chosen worker authentication method requires storage and no worker auth storage was configured")
 		}
 		c.InfoKeys = append(c.InfoKeys, "worker auth storage path")
-		c.Info["worker auth storage path"] = c.worker.WorkerAuthStorage.BaseDir()
+		c.Info["worker auth storage path"] = c.Config.Worker.AuthStoragePath
 	}

 	return nil
@@ -757,6 +763,8 @@ func (c *Command) WaitForInterrupt() int {
 				c.Logger.SetLevel(level)
 			}

+			c.WorkerAuthDebuggingEnabled.Store(newConf.EnableWorkerAuthDebugging)
+
 		RUNRELOADFUNCS:
 			if err := c.Reload(newConf); err != nil {
 				c.UI.Error(fmt.Errorf("Error(s) were encountered during reload: %w", err).Error())
@@ -7,7 +7,6 @@ package server

 import (
 	"fmt"
-	"path/filepath"
 	"sync"
 	"testing"
 	"time"
@@ -41,7 +40,6 @@ worker {
 	tags {
 		type = ["dev", "local"]
 	}
-	auth_storage_path = "%s"
 }
 `

@@ -57,18 +55,26 @@ func TestServer_ReloadInitialUpstreams(t *testing.T) {
 	// would be odd to have separate databases, but it makes it easy for the
 	// test to assert that the worker has connected to the second controller
 	// after a reload signal is sent.
-	testController := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
+	testController := controller.NewTestController(
+		t,
+		controller.WithWorkerAuthKms(workerAuthWrapper),
+		controller.WithRootKms(rootWrapper),
+		controller.WithRecoveryKms(recoveryWrapper),
+	)
 	defer testController.Shutdown()
-	testController2 := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
+	testController2 := controller.NewTestController(
+		t,
+		controller.WithWorkerAuthKms(workerAuthWrapper),
+		controller.WithRootKms(rootWrapper),
+		controller.WithRecoveryKms(recoveryWrapper),
+	)
 	defer testController2.Shutdown()
 	require.NotEqual(testController.Config().DatabaseUrl, testController2.Config().DatabaseUrl)

-	authStoragePath := t.TempDir()
-
 	wg := &sync.WaitGroup{}

 	cmd := testServerCommand(t, testServerCommandOpts{})
-	cmd.presetConfig = atomic.NewString(fmt.Sprintf(initialUpstreamConfig, key, testController.ClusterAddrs()[0], filepath.Join(authStoragePath, t.Name())))
+	cmd.presetConfig = atomic.NewString(fmt.Sprintf(initialUpstreamConfig, key, testController.ClusterAddrs()[0]))

 	wg.Add(1)
 	go func() {
@@ -108,7 +114,7 @@ pollFirstController:
 	}

 	// Reload the config after changing initial_upstreams to the second controller
-	cmd.presetConfig.Store(fmt.Sprintf(initialUpstreamConfig, key, testController2.ClusterAddrs()[0], filepath.Join(authStoragePath, t.Name())))
+	cmd.presetConfig.Store(fmt.Sprintf(initialUpstreamConfig, key, testController2.ClusterAddrs()[0]))
 	cmd.SighupCh <- struct{}{}
 	select {
 	case <-cmd.reloadedCh:
@@ -126,6 +132,7 @@ pollSecondController:
 			t.Fatalf("timeout wait for worker to connect to second controller")
 		case <-poll.C:
 			serversRepo, err := testController2.Controller().ServersRepoFn()
+			require.NoError(err)
 			w, err = serversRepo.LookupWorkerByName(testController2.Context(), "test")
 			require.NoError(err)
 			if w != nil {
@@ -17,7 +17,6 @@ import (
 	"crypto/rand"
 	"encoding/base64"
 	"fmt"
-	"path/filepath"
 	"sync"
 	"testing"
 	"time"
@@ -54,7 +53,6 @@ worker {
 	tags {
 		type = ["dev", "local"]
 	}
-	auth_storage_path = "%s"
 }
 `

@@ -66,7 +64,6 @@ worker {
 	tags {
 		foo = ["bar", "baz"]
 	}
-	auth_storage_path = "%s"
 }
 `

@@ -80,12 +77,10 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 	testController := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
 	defer testController.Shutdown()

-	authStoragePath := t.TempDir()
-
 	wg := &sync.WaitGroup{}

 	cmd := testServerCommand(t, testServerCommandOpts{})
-	cmd.presetConfig = atomic.NewString(fmt.Sprintf(workerBaseConfig+tag1Config, key, testController.ClusterAddrs()[0], filepath.Join(authStoragePath, "tag1")))
+	cmd.presetConfig = atomic.NewString(fmt.Sprintf(workerBaseConfig+tag1Config, key, testController.ClusterAddrs()[0]))

 	wg.Add(1)
 	go func() {
@@ -118,7 +113,7 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 	time.Sleep(10 * time.Second)
 	fetchWorkerTags("test", "type", []string{"dev", "local"})

-	cmd.presetConfig.Store(fmt.Sprintf(workerBaseConfig+tag2Config, key, testController.ClusterAddrs()[0], filepath.Join(authStoragePath, "tag2")))
+	cmd.presetConfig.Store(fmt.Sprintf(workerBaseConfig+tag2Config, key, testController.ClusterAddrs()[0]))

 	cmd.SighupCh <- struct{}{}
 	select {
@@ -139,8 +139,15 @@ type Config struct {
 	// Plugin-related options
 	Plugins Plugins `hcl:"plugins"`

-	// Internal field for use with HCP deployments. Used if controllers/ initial_upstreams is not set
+	// Internal field for use with HCP deployments. Used if controllers/
+	// initial_upstreams is not set.
 	HcpbClusterId string `hcl:"hcp_boundary_cluster_id"`
+
+	// Whether to log information from the worker authentication process. This
+	// shouldn't divulge any secrets but may contain information considered
+	// sensitive. This should only be enabled for debugging purposes, and can be
+	// toggled with SIGHUP.
+	EnableWorkerAuthDebugging bool `hcl:"enable_worker_auth_debugging"`
 }

 type Controller struct {
@@ -258,6 +265,11 @@ type Worker struct {
 	// token used to register this worker to the cluster. It can be a path, env
 	// var, or direct value.
 	ControllerGeneratedActivationToken string `hcl:"controller_generated_activation_token"`
+
+	// UseDeprecatedKmsAuthMethod indicates that the worker should use the pre-0.13
+	// method of using KMSes to authenticate. This should not be used when the
+	// controller version supports the new style.
+	UseDeprecatedKmsAuthMethod bool `hcl:"use_deprecated_kms_auth_method"`
 }

 type Database struct {
@@ -806,7 +818,9 @@ func Parse(d string) (*Config, error) {
 		}
 	}

-	sharedConfig, err := configutil.ParseConfig(d)
+	// Now that we can have multiple KMSes for downstream workers, allow an
+	// unlimited number of KMS blocks as we don't know how many might be defined
+	sharedConfig, err := configutil.ParseConfig(d, configutil.WithMaxKmsBlocks(-1))
 	if err != nil {
 		return nil, err
 	}
@@ -26,6 +26,14 @@ type WorkerConnectionInfo struct {
 	WorkerId string `mapstructure:"worker_id,omitempty"`
 	// The version of the worker that is dialing
 	BoundaryVersion string `mapstructure:"boundary_version,omitempty"`
+	// The name, used for wrapping registration flow
+	Name string `mapstructure:"name,omitempty"`
+	// The description, used for wrapping registration flow
+	Description string `mapstructure:"description,omitempty"`
+	// The public address, used for wrapping registration flow
+	PublicAddr string `mapstructure:"public_addr,omitempty"`
+	// The operational state, used for wrapping registration flow
+	OperationalState string `mapstructure:"operational_state,omitempty"`
 }

 // AsConnectionStateStruct builds a structPb in the format that is expected for
@@ -27,7 +27,19 @@ func TestWorkerConnectionInfo(t *testing.T) {
 		{WorkerId: "something"},
 		{UpstreamAddress: "something"},
 		{BoundaryVersion: "something"},
-		{WorkerId: "something", UpstreamAddress: "something else", BoundaryVersion: "something yet more"},
+		{Name: "something"},
+		{Description: "something"},
+		{PublicAddr: "something"},
+		{OperationalState: "something"},
+		{
+			WorkerId:         "something",
+			UpstreamAddress:  "something else",
+			BoundaryVersion:  "something yet more",
+			Name:             "some name",
+			Description:      "some description",
+			PublicAddr:       "some addr",
+			OperationalState: "some state",
+		},
 	}
 	for _, c := range connInfo {
 		st, err := c.AsConnectionStateStruct()
@@ -98,8 +98,6 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 	switch {
 	case wStat.GetName() == "" && wStat.GetKeyId() == "":
 		return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name and keyId are not set in the request; one is required.")
-	case wStat.GetName() != "" && wStat.GetKeyId() != "":
-		return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name and keyId are both set in the request; only one is allowed.")
 	case wStat.GetAddress() == "":
 		return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Address is not set but is required.")
 	}
@@ -1047,7 +1047,7 @@ func TestCreateWorkerLed(t *testing.T) {
 		// This happens on the worker
 		fileStorage, err := file.New(testCtx)
 		require.NoError(t, err)
-		defer fileStorage.Cleanup()
+		defer func() { fileStorage.Cleanup(testCtx) }()

 		nodeCreds, err := types.NewNodeCredentials(testCtx, fileStorage)
 		require.NoError(t, err)
@@ -1430,7 +1430,7 @@ func TestCreateControllerLed(t *testing.T) {
 		// This happens on the worker
 		fileStorage, err := file.New(testCtx)
 		require.NoError(t, err)
-		defer fileStorage.Cleanup()
+		defer func() { fileStorage.Cleanup(testCtx) }()

 		nodeCreds, err := types.NewNodeCredentials(testCtx, fileStorage)
 		require.NoError(t, err)
@@ -20,10 +20,12 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/util"
 	"github.com/hashicorp/go-multierror"
 	nodee "github.com/hashicorp/nodeenrollment"
 	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	"github.com/hashicorp/nodeenrollment/protocol"
+	"github.com/hashicorp/nodeenrollment/util/toggledlogger"
 	"google.golang.org/grpc"
 )

@@ -133,10 +135,25 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		return nil, fmt.Errorf("error fetching worker auth storage: %w", err)
 	}

+	eventLogger, err := event.NewHclogLogger(c.baseContext, c.conf.Eventer)
+	if err != nil {
+		event.WriteError(c.baseContext, op, err)
+		return nil, errors.Wrap(c.baseContext, err, op)
+	}
+	// Give the log a prefix
+	eventLogger = eventLogger.Named(fmt.Sprintf("workerauth_listener"))
+	// Wrap the log in a toggle so we can turn it on and off via config and
+	// SIGHUP
+	eventLogger = toggledlogger.NewToggledLogger(eventLogger, c.conf.WorkerAuthDebuggingEnabled)
+
+	wrapperToUse := c.conf.WorkerAuthKms
+	if !util.IsNil(c.conf.DownstreamWorkerAuthKms) {
+		wrapperToUse = c.conf.DownstreamWorkerAuthKms
+	}
+
 	// The cluster listener is shut down at server shutdown time so we do not
 	// need to handle individual listener shutdown.
-	var l net.Listener
-	l, err = protocol.NewInterceptingListener(
+	interceptingListener, err := protocol.NewInterceptingListener(
 		&protocol.InterceptingListenerConfiguration{
 			Context:      c.baseContext,
 			Storage:      workerAuthStorage,
@@ -144,13 +161,17 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 			BaseTlsConfiguration: &tls.Config{
 				GetConfigForClient: c.validateWorkerTls,
 			},
+			Options: []nodee.Option{
+				nodee.WithLogger(eventLogger),
+				nodee.WithRegistrationWrapper(wrapperToUse),
+			},
 		})
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating node auth listener: %w", err)
 	}

 	// Create split listener
-	splitListener, err := nodeenet.NewSplitListener(l)
+	splitListener, err := nodeenet.NewSplitListener(interceptingListener)
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating split listener: %w", err)
 	}
@@ -37,6 +37,7 @@ import (
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/go-hclog"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/hashicorp/go-kms-wrapping/v2/extras/multi"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/strutil"
 	"github.com/prometheus/client_golang/prometheus"
@@ -424,6 +425,9 @@ type TestControllerOpts struct {
 	// The worker auth KMS to use, or one will be created
 	WorkerAuthKms wrapping.Wrapper

+	// The downstream worker auth KMS to use, or one will be created
+	DownstreamWorkerAuthKms *multi.PooledWrapper
+
 	// The recovery KMS to use, or one will be created
 	RecoveryKms wrapping.Wrapper

@@ -670,16 +674,17 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController,
 	tc.b.DevUnprivilegedUserId = "u_" + strutil.Reverse(strings.TrimPrefix(tc.b.DevUserId, "u_"))

 	// Set up KMSes
-	switch {
-	case opts.RootKms != nil && opts.WorkerAuthKms != nil:
+	if err := tc.b.SetupKMSes(tc.b.Context, nil, opts.Config); err != nil {
+		t.Fatal(err)
+	}
+	if opts.RootKms != nil {
 		tc.b.RootKms = opts.RootKms
+	}
+	if opts.WorkerAuthKms != nil {
 		tc.b.WorkerAuthKms = opts.WorkerAuthKms
-	case opts.RootKms == nil && opts.WorkerAuthKms == nil:
-		if err := tc.b.SetupKMSes(tc.b.Context, nil, opts.Config); err != nil {
-			t.Fatal(err)
-		}
-	default:
-		t.Fatal("either controller and worker auth KMS must both be set, or neither")
+	}
+	if opts.DownstreamWorkerAuthKms != nil {
+		tc.b.DownstreamWorkerAuthKms = opts.DownstreamWorkerAuthKms
 	}
 	if opts.RecoveryKms != nil {
 		tc.b.RecoveryKms = opts.RecoveryKms
@@ -778,6 +783,7 @@ func (tc *TestController) AddClusterControllerMember(t testing.TB, opts *TestCon
 		DefaultOidcAuthMethodId:         tc.c.conf.DevOidcAuthMethodId,
 		RootKms:                         tc.c.conf.RootKms,
 		WorkerAuthKms:                   tc.c.conf.WorkerAuthKms,
+		DownstreamWorkerAuthKms:         tc.c.conf.DownstreamWorkerAuthKms,
 		RecoveryKms:                     tc.c.conf.RecoveryKms,
 		Name:                            opts.Name,
 		Logger:                          tc.c.conf.Logger,
@@ -42,7 +42,7 @@ type grpcResolverReceiver struct {
 }

 // IsDialingListener always returns
-func (_ *grpcResolverReceiver) Type() receiverType {
+func (*grpcResolverReceiver) Type() receiverType {
 	return grpcResolverReceiverType
 }

@@ -17,8 +17,9 @@ import (

 func (w *Worker) startAuthRotationTicking(cancelCtx context.Context) {
 	const op = "worker.(Worker).startAuthRotationTicking"
-	if w.conf.WorkerAuthKms != nil && !w.conf.DevUsePkiForUpstream {
-		event.WriteSysEvent(cancelCtx, op, "using kms worker authentication; pki auth rotation ticking not running")
+
+	if w.conf.RawConfig.Worker.UseDeprecatedKmsAuthMethod {
+		event.WriteSysEvent(cancelCtx, op, "using deprecated kms worker authentication method; pki auth rotation ticking not running")
 		return
 	}

@@ -47,7 +48,12 @@ func (w *Worker) startAuthRotationTicking(cancelCtx context.Context) {
 			resetDuration = defaultResetDuration

 			// Check if it's time to rotate and if not don't do anything
-			currentNodeCreds, err := types.LoadNodeCredentials(cancelCtx, w.WorkerAuthStorage, nodeenrollment.CurrentId, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms))
+			currentNodeCreds, err := types.LoadNodeCredentials(
+				cancelCtx,
+				w.WorkerAuthStorage,
+				nodeenrollment.CurrentId,
+				nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
+			)
 			if err != nil {
 				if errors.Is(err, nodeenrollment.ErrNotFound) {
 					// Be silent
@@ -139,7 +145,7 @@ func rotateWorkerAuth(ctx context.Context, w *Worker, currentNodeCreds *types.No
 		ctx,
 		w.WorkerAuthStorage,
 		randReaderOpt,
-		nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
+		nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
 		nodeenrollment.WithSkipStorage(true),
 	)
 	if err != nil {
@@ -189,7 +195,7 @@ func rotateWorkerAuth(ctx context.Context, w *Worker, currentNodeCreds *types.No
 		ctx,
 		w.WorkerAuthStorage,
 		fetchResp,
-		nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
+		nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
 	)
 	if err != nil {
 		return berrors.Wrap(ctx, err, op)
@@ -88,7 +88,12 @@ func TestRotationTicking(t *testing.T) {
 	require.NoError(err)
 	assert.Len(auths, 1)
 	// Fetch creds and store current key
-	currNodeCreds, err := types.LoadNodeCredentials(w.Context(), w.Worker().WorkerAuthStorage, nodeenrollment.CurrentId, nodeenrollment.WithWrapper(w.Config().WorkerAuthStorageKms))
+	currNodeCreds, err := types.LoadNodeCredentials(
+		w.Context(),
+		w.Worker().WorkerAuthStorage,
+		nodeenrollment.CurrentId,
+		nodeenrollment.WithStorageWrapper(w.Config().WorkerAuthStorageKms),
+	)
 	require.NoError(err)
 	currKey := currNodeCreds.CertificatePublicKeyPkix

@@ -106,7 +111,12 @@ func TestRotationTicking(t *testing.T) {
 		assert.Len(auths, 2)

 		// Fetch creds and compare current key
-		currNodeCreds, err := types.LoadNodeCredentials(w.Context(), w.Worker().WorkerAuthStorage, nodeenrollment.CurrentId, nodeenrollment.WithWrapper(w.Config().WorkerAuthStorageKms))
+		currNodeCreds, err := types.LoadNodeCredentials(
+			w.Context(),
+			w.Worker().WorkerAuthStorage,
+			nodeenrollment.CurrentId,
+			nodeenrollment.WithStorageWrapper(w.Config().WorkerAuthStorageKms),
+		)
 		require.NoError(err)
 		assert.NotEqual(currKey, currNodeCreds.CertificatePublicKeyPkix)
 		currKey = currNodeCreds.CertificatePublicKeyPkix
@@ -28,11 +28,13 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/multihop"
 	"github.com/hashicorp/nodeenrollment/protocol"
+	"github.com/hashicorp/nodeenrollment/util/toggledlogger"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials/insecure"
@@ -93,22 +95,34 @@ func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Cont
 	const op = "worker.(Worker).upstreamDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		var conn net.Conn
-		var err error
+		eventLogger, err := event.NewHclogLogger(w.baseContext, w.conf.Eventer)
+		if err != nil {
+			event.WriteError(w.baseContext, op, err)
+			return nil, errors.Wrap(w.baseContext, err, op)
+		}
+		// Give the log a prefix
+		eventLogger = eventLogger.Named(fmt.Sprintf("workerauth_dialer"))
+		// Wrap the log in a toggle so we can turn it on and off via config and
+		// SIGHUP
+		eventLogger = toggledlogger.NewToggledLogger(eventLogger, w.conf.WorkerAuthDebuggingEnabled)
+		st, err := w.workerConnectionInfo(addr)
+		if err != nil {
+			event.WriteError(w.baseContext, op, err)
+			return nil, errors.Wrap(w.baseContext, err, op)
+		}
 		switch {
-		case w.conf.WorkerAuthKms != nil && !w.conf.DevUsePkiForUpstream:
+		case w.conf.RawConfig.Worker.UseDeprecatedKmsAuthMethod:
 			conn, err = w.v1KmsAuthDialFn(ctx, addr, extraAlpnProtos...)
 		default:
-			var st *structpb.Struct
-			st, err = w.workerConnectionInfo(addr)
-			if err != nil {
-				event.WriteError(w.baseContext, op, err)
-			}
 			conn, err = protocol.Dial(
 				ctx,
 				w.WorkerAuthStorage,
 				addr,
+				nodeenrollment.WithLogger(eventLogger),
 				nodeenrollment.WithState(st),
-				nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
+				nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
+				nodeenrollment.WithRegistrationWrapper(w.conf.WorkerAuthKms),
+				nodeenrollment.WithWrappingRegistrationFlowApplicationSpecificParams(st),
 				nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos),
 				// If the activation token hasn't been populated, this won't do
 				// anything, and it won't do anything if it's already been used
@@ -117,8 +131,8 @@ func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Cont
 			// No error and a valid connection means the WorkerAuthRegistrationRequest was populated
 			// We can remove the stored workerAuthRequest file
 			if err == nil && conn != nil {
-				if w.WorkerAuthStorage.BaseDir() != "" {
-					workerAuthReqFilePath := filepath.Join(w.WorkerAuthStorage.BaseDir(), base.WorkerAuthReqFile)
+				if w.conf.RawConfig.Worker.AuthStoragePath != "" {
+					workerAuthReqFilePath := filepath.Join(w.conf.RawConfig.Worker.AuthStoragePath, base.WorkerAuthReqFile)
 					// Intentionally ignoring any error removing this file
 					_ = os.Remove(workerAuthReqFilePath)
 				}
@@ -134,18 +148,21 @@ func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Cont
 				// We don't event in this case, because the function retries
 				// often and will spam the logs while waiting on the user to
 				// transfer the worker-generated request over
+				return nil, errors.Wrap(w.baseContext, err, op)

 			default:
 				// In this case, event, so that the operator can understand that
 				// it was rejected
 				event.WriteError(ctx, op, fmt.Errorf("controller rejected activation token as invalid"))
+				return nil, errors.Wrap(w.baseContext, err, op)
 			}

 		default:
 			event.WriteError(ctx, op, err)
+			return nil, errors.Wrap(w.baseContext, err, op)
 		}

-		if err == nil && conn != nil {
+		if conn != nil {
 			if w.everAuthenticated.Load() == authenticationStatusNeverAuthenticated {
 				w.everAuthenticated.Store(authenticationStatusFirstAuthentication)
 			}
@@ -361,8 +378,12 @@ func (w *Worker) workerAuthTLSConfig(extraAlpnProtos ...string) (*tls.Config, *b
 func (w *Worker) workerConnectionInfo(addr string) (*structpb.Struct, error) {
 	const op = "worker.(Worker).workerConnectionInfo"
 	wci := &cluster.WorkerConnectionInfo{
-		UpstreamAddress: addr,
-		BoundaryVersion: version.Get().VersionNumber(),
+		UpstreamAddress:  addr,
+		BoundaryVersion:  version.Get().VersionNumber(),
+		Name:             w.conf.RawConfig.Worker.Name,
+		Description:      w.conf.RawConfig.Worker.Description,
+		PublicAddr:       w.conf.RawConfig.Worker.PublicAddr,
+		OperationalState: w.operationalState.Load().(server.OperationalState).String(),
 	}
 	if w.LastStatusSuccess() != nil && len(w.LastStatusSuccess().GetWorkerId()) > 0 {
 		// even though we wont have the worker the first time we dial, any
@@ -336,7 +336,7 @@ func (w *Worker) credDecryptFn(ctx context.Context) (proxyHandlers.DecryptFn, er
 	was := w.WorkerAuthStorage
 	var opts []nodeenrollment.Option
 	if !util.IsNil(w.conf.WorkerAuthStorageKms) {
-		opts = append(opts, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms))
+		opts = append(opts, nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms))
 	}
 	nodeCreds, err := types.LoadNodeCredentials(ctx, was, nodeenrollment.CurrentId, opts...)
 	if err != nil {
@@ -32,21 +32,21 @@ func TestGetHealth(t *testing.T) {
 		name             string
 		method           string
 		queryParams      string
-		expectedResponse opsservices.GetHealthResponse
+		expectedResponse *opsservices.GetHealthResponse
 		expCode          int
 	}{
 		{
 			name:             "healthy reply",
 			method:           http.MethodGet,
 			expCode:          http.StatusOK,
-			expectedResponse: opsservices.GetHealthResponse{},
+			expectedResponse: &opsservices.GetHealthResponse{},
 		},
 		{
 			name:        "healthy reply with worker info",
 			method:      http.MethodGet,
 			queryParams: "worker_info=1",
 			expCode:     http.StatusOK,
-			expectedResponse: opsservices.GetHealthResponse{
+			expectedResponse: &opsservices.GetHealthResponse{
 				WorkerProcessInfo: &pbhealth.HealthInfo{
 					State:              server.ActiveOperationalState.String(),
 					ActiveSessionCount: wrapperspb.UInt32(0),
@@ -58,7 +58,7 @@ func TestGetHealth(t *testing.T) {
 			method:           http.MethodPost,
 			queryParams:      "worker_info=1",
 			expCode:          http.StatusNotImplemented,
-			expectedResponse: opsservices.GetHealthResponse{},
+			expectedResponse: &opsservices.GetHealthResponse{},
 		},
 	}

@@ -6,7 +6,7 @@ package worker
 import (
 	"context"
 	"crypto/tls"
-	stderrers "errors"
+	stderrors "errors"
 	"fmt"
 	"log"
 	"math"
@@ -22,14 +22,18 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/worker/session"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/util"
 	"github.com/hashicorp/go-multierror"
 	nodee "github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/multihop"
 	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	"github.com/hashicorp/nodeenrollment/protocol"
+	"github.com/hashicorp/nodeenrollment/registration"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/hashicorp/nodeenrollment/util/temperror"
+	"github.com/hashicorp/nodeenrollment/util/toggledlogger"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 )

 const (
@@ -113,17 +117,49 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,

 	fetchCredsFn := func(
 		ctx context.Context,
-		_ nodee.Storage,
+		storage nodee.Storage,
 		req *types.FetchNodeCredentialsRequest,
-		_ ...nodee.Option,
+		opt ...nodee.Option,
 	) (*types.FetchNodeCredentialsResponse, error) {
+		switch {
+		case req == nil:
+			return nil, temperror.New(stderrors.New("nil request in multi-hop fetch function"))
+		case len(req.Bundle) == 0:
+			return nil, temperror.New(stderrors.New("empty bundle in multi-hop fetch function"))
+		}
+		// Check to see if there is encrypted registration info, if so we need
+		// to use our wrapper to decrypt it
+		reqInfo := new(types.FetchNodeCredentialsInfo)
+		if err := proto.Unmarshal(req.Bundle, reqInfo); err != nil {
+			return nil, temperror.New(fmt.Errorf("error unmarshaling request bundle in multi-hop fetch function: %w", err))
+		}
+		if len(reqInfo.WrappedRegistrationInfo) > 0 {
+			regInfo, err := registration.DecryptWrappedRegistrationInfo(ctx, reqInfo, opt...)
+			if err != nil {
+				return nil, temperror.New(fmt.Errorf("error during decryption of wrapped registration info in multi-hop fetch function: %w", err))
+			}
+			// We've successfully decrypted it using our registration wrapper;
+			// now we need to encrypt it to the server
+			nodeCreds, err := types.LoadNodeCredentials(ctx, storage, nodee.CurrentId, opt...)
+			if err != nil {
+				return nil, temperror.New(fmt.Errorf("error loading node credentials in multi-hop fetch function: %w", err))
+			}
+			req.RewrappingKeyId, err = nodee.KeyIdFromPkix(nodeCreds.CertificatePublicKeyPkix)
+			if err != nil {
+				return nil, temperror.New(fmt.Errorf("error deriving node credentials key id in multi-hop fetch function: %w", err))
+			}
+			req.RewrappedWrappingRegistrationFlowInfo, err = nodee.EncryptMessage(ctx, regInfo, nodeCreds)
+			if err != nil {
+				return nil, temperror.New(fmt.Errorf("error rewrapping registration information in multi-hop fetch function: %w", err))
+			}
+		}
 		client := w.controllerMultihopConn.Load()
 		if client == nil {
-			return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil"))
+			return nil, temperror.New(stderrors.New("error fetching controller connection, client is nil"))
 		}
 		multihopClient, ok := client.(multihop.MultihopServiceClient)
 		if !ok {
-			return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client"))
+			return nil, temperror.New(stderrors.New("client could not be understood as a multihop service client"))
 		}
 		return multihopClient.FetchNodeCredentials(ctx, req)
 	}
@@ -136,15 +172,31 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	) (*types.GenerateServerCertificatesResponse, error) {
 		client := w.controllerMultihopConn.Load()
 		if client == nil {
-			return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil"))
+			return nil, temperror.New(stderrors.New("error fetching controller connection, client is nil"))
 		}
 		multihopClient, ok := client.(multihop.MultihopServiceClient)
 		if !ok {
-			return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client"))
+			return nil, temperror.New(stderrors.New("client could not be understood as a multihop service client"))
 		}
 		return multihopClient.GenerateServerCertificates(ctx, req)
 	}

+	eventLogger, err := event.NewHclogLogger(w.baseContext, w.conf.Eventer)
+	if err != nil {
+		event.WriteError(w.baseContext, op, err)
+		return nil, errors.Wrap(w.baseContext, err, op)
+	}
+	// Give the log a prefix
+	eventLogger = eventLogger.Named(fmt.Sprintf("workerauth_listener"))
+	// Wrap the log in a toggle so we can turn it on and off via config and
+	// SIGHUP
+	eventLogger = toggledlogger.NewToggledLogger(eventLogger, w.conf.WorkerAuthDebuggingEnabled)
+
+	wrapperToUse := w.conf.WorkerAuthKms
+	if !util.IsNil(w.conf.DownstreamWorkerAuthKms) {
+		wrapperToUse = w.conf.DownstreamWorkerAuthKms
+	}
+
 	interceptingListener, err := protocol.NewInterceptingListener(
 		&protocol.InterceptingListenerConfiguration{
 			Context:      w.baseContext,
@@ -155,6 +207,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 			},
 			FetchCredsFunc:                 fetchCredsFn,
 			GenerateServerCertificatesFunc: generateServerCertificatesFn,
+			Options: []nodee.Option{
+				nodee.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
+				nodee.WithRegistrationWrapper(wrapperToUse),
+				nodee.WithLogger(eventLogger),
+			},
 		})
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating node auth listener: %w", err)
@@ -165,9 +165,9 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 	defer statusCancel()

 	keyId := w.WorkerAuthCurrentKeyId.Load()
-
-	if w.conf.RawConfig.Worker.Name == "" && keyId == "" {
-		event.WriteError(cancelCtx, op, errors.New("worker name and keyId are both empty; one is needed to identify a worker"),
+	switch {
+	case w.conf.RawConfig.Worker.Name == "" && keyId == "":
+		event.WriteError(cancelCtx, op, errors.New("worker name and keyId are both empty; at least one is needed to identify a worker"),
 			event.WithInfoMsg("error making status request to controller"))
 	}
 	versionInfo := version.Get()
@@ -10,6 +10,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -24,6 +25,7 @@ import (
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/go-hclog"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/hashicorp/go-kms-wrapping/v2/extras/multi"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mr-tron/base58"
 	"github.com/prometheus/client_golang/prometheus"
@@ -188,6 +190,9 @@ type TestWorkerOpts struct {
 	// The worker auth KMS to use, or one will be created
 	WorkerAuthKms wrapping.Wrapper

+	// The downstream worker auth KMS to use, or one will be created
+	DownstreamWorkerAuthKms *multi.PooledWrapper
+
 	// The worker credential storage KMS to use, or one will be created
 	WorkerAuthStorageKms wrapping.Wrapper

@@ -215,6 +220,12 @@ type TestWorkerOpts struct {

 	// If set, override the normal auth rotation period
 	AuthRotationPeriod time.Duration
+
+	// If set, will use the deprecated KMS auth method
+	UseDeprecatedKmsAuthMethod bool
+
+	// Toggle worker auth debugging
+	WorkerAuthDebuggingEnabled *atomic.Bool
 }

 func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
@@ -235,6 +246,7 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {

 	// Base server
 	tw.b = base.NewServer(nil)
+	tw.b.WorkerAuthDebuggingEnabled = opts.WorkerAuthDebuggingEnabled
 	tw.b.Command = &base.Command{
 		Context:    ctx,
 		ShutdownCh: make(chan struct{}),
@@ -252,6 +264,9 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 		}
 	}

+	if opts.UseDeprecatedKmsAuthMethod {
+		opts.Config.Worker.UseDeprecatedKmsAuthMethod = true
+	}
 	if len(opts.InitialUpstreams) > 0 {
 		opts.Config.Worker.InitialUpstreams = opts.InitialUpstreams
 	}
@@ -273,7 +288,6 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 	}
 	if opts.WorkerAuthStoragePath != "" {
 		opts.Config.Worker.AuthStoragePath = opts.WorkerAuthStoragePath
-		tw.b.DevUsePkiForUpstream = true
 	}
 	tw.name = opts.Config.Worker.Name

@@ -291,15 +305,17 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 	}

 	// Set up KMSes
-	switch {
-	case opts.WorkerAuthKms != nil:
+	if err := tw.b.SetupKMSes(tw.b.Context, nil, opts.Config); err != nil {
+		t.Fatal(err)
+	}
+	if opts.WorkerAuthKms != nil {
 		tw.b.WorkerAuthKms = opts.WorkerAuthKms
-	case opts.WorkerAuthStorageKms != nil:
+	}
+	if opts.WorkerAuthStorageKms != nil {
 		tw.b.WorkerAuthStorageKms = opts.WorkerAuthStorageKms
-	default:
-		if err := tw.b.SetupKMSes(tw.b.Context, nil, opts.Config); err != nil {
-			t.Fatal(err)
-		}
+	}
+	if opts.DownstreamWorkerAuthKms != nil {
+		tw.b.DownstreamWorkerAuthKms = opts.DownstreamWorkerAuthKms
 	}

 	// Ensure the listeners use random port allocation
@@ -362,11 +378,13 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts)
 	}
 	nextOpts := &TestWorkerOpts{
 		WorkerAuthKms:                       tw.w.conf.WorkerAuthKms,
+		DownstreamWorkerAuthKms:             tw.w.conf.DownstreamWorkerAuthKms,
 		WorkerAuthStorageKms:                tw.w.conf.WorkerAuthStorageKms,
 		Name:                                opts.Name,
 		InitialUpstreams:                    tw.UpstreamAddrs(),
 		Logger:                              tw.w.conf.Logger,
 		SuccessfulStatusGracePeriodDuration: opts.SuccessfulStatusGracePeriodDuration,
+		WorkerAuthDebuggingEnabled:          tw.w.conf.WorkerAuthDebuggingEnabled,
 	}
 	if nextOpts.Name == "" {
 		var err error
@@ -380,18 +398,37 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts)
 	return NewTestWorker(t, nextOpts)
 }

-// NewTestMultihopWorkers creates a KMS and PKI worker with the controller as an upstream, and a
-// child PKI worker as a downstream of the PKI worker connected to the controller.
-// Tags for the PKI and child PKI worker can be passed in, if desired
-func NewTestMultihopWorkers(t testing.TB, logger hclog.Logger, controllerContext context.Context, clusterAddrs []string,
-	workerAuthKms wrapping.Wrapper, serversRepoFn common.ServersRepoFactory, pkiTags,
-	childPkiTags map[string][]string,
-) (kmsWorker *TestWorker, pkiWorker *TestWorker, childPkiWorker *TestWorker) {
+// NewTestMultihopWorkers creates a PKI-KMS and PKI worker with the controller
+// as an upstream, and two child workers (one PKI, one KMS) as downstreams of
+// the initial workers (child PKI -> upstream PKI-KMS, child PKI-KMS -> upstream
+// PKI). Tags for the PKI and child PKI/KMS workers can be passed in, if
+// desired.
+func NewTestMultihopWorkers(t testing.TB,
+	logger hclog.Logger,
+	controllerContext context.Context,
+	clusterAddrs []string,
+	workerAuthKms wrapping.Wrapper,
+	serversRepoFn common.ServersRepoFactory,
+	pkiTags, childPkiTags, childKmsTags map[string][]string,
+	enableAuthDebugging *atomic.Bool,
+) (kmsWorker, pkiWorker, childPkiWorker, childKmsWorker *TestWorker) {
 	require := require.New(t)
+
+	// Create a few test wrappers for the child KMS worker to use
+	childDownstreamWrapper1 := db.TestWrapper(t)
+	childDownstreamWrapper2 := db.TestWrapper(t)
+	childDownstreamWrapper, err := multi.NewPooledWrapper(context.Background(), childDownstreamWrapper1)
+	require.NoError(err)
+	added, err := childDownstreamWrapper.AddWrapper(context.Background(), childDownstreamWrapper2)
+	require.NoError(err)
+	require.True(added)
+
 	kmsWorker = NewTestWorker(t, &TestWorkerOpts{
-		WorkerAuthKms:    workerAuthKms,
-		InitialUpstreams: clusterAddrs,
-		Logger:           logger.Named("kmsWorker"),
+		WorkerAuthKms:              workerAuthKms,
+		InitialUpstreams:           clusterAddrs,
+		Logger:                     logger.Named("kmsWorker"),
+		WorkerAuthDebuggingEnabled: enableAuthDebugging,
+		DownstreamWorkerAuthKms:    childDownstreamWrapper,
 	})

 	// names should not be set when using pki workers
@@ -403,9 +440,11 @@ func NewTestMultihopWorkers(t testing.TB, logger hclog.Logger, controllerContext
 	}
 	pkiWorkerConf.Worker.InitialUpstreams = clusterAddrs
 	pkiWorker = NewTestWorker(t, &TestWorkerOpts{
-		InitialUpstreams: clusterAddrs,
-		Logger:           logger.Named("pkiWorker"),
-		Config:           pkiWorkerConf,
+		InitialUpstreams:           clusterAddrs,
+		Logger:                     logger.Named("pkiWorker"),
+		Config:                     pkiWorkerConf,
+		DownstreamWorkerAuthKms:    childDownstreamWrapper,
+		WorkerAuthDebuggingEnabled: enableAuthDebugging,
 	})
 	t.Cleanup(pkiWorker.Shutdown)

@@ -436,9 +475,10 @@ func NewTestMultihopWorkers(t testing.TB, logger hclog.Logger, controllerContext
 	childPkiWorkerConf.Worker.InitialUpstreams = kmsWorker.ProxyAddrs()

 	childPkiWorker = NewTestWorker(t, &TestWorkerOpts{
-		InitialUpstreams: kmsWorker.ProxyAddrs(),
-		Logger:           logger.Named("childPkiWorker"),
-		Config:           childPkiWorkerConf,
+		InitialUpstreams:           kmsWorker.ProxyAddrs(),
+		Logger:                     logger.Named("childPkiWorker"),
+		Config:                     childPkiWorkerConf,
+		WorkerAuthDebuggingEnabled: enableAuthDebugging,
 	})

 	// Perform initial authentication of worker to controller
@@ -455,10 +495,28 @@ func NewTestMultihopWorkers(t testing.TB, logger hclog.Logger, controllerContext
 	}, server.WithFetchNodeCredentialsRequest(childPkiWorkerReq))
 	require.NoError(err)

+	childKmsWorkerConf, err := config.DevWorker()
+	require.NoError(err)
+	childKmsWorkerConf.Worker.Name = "child-kms-worker"
+	childKmsWorkerConf.Worker.Description = "child-kms-worker description"
+	// Set tags the same
+	if childKmsTags != nil {
+		childKmsWorkerConf.Worker.Tags = childKmsTags
+	}
+	childKmsWorkerConf.Worker.InitialUpstreams = kmsWorker.ProxyAddrs()
+
+	childKmsWorker = NewTestWorker(t, &TestWorkerOpts{
+		InitialUpstreams:           pkiWorker.ProxyAddrs(),
+		Logger:                     logger.Named("childKmsWorker"),
+		Config:                     childKmsWorkerConf,
+		WorkerAuthKms:              childDownstreamWrapper2,
+		WorkerAuthDebuggingEnabled: enableAuthDebugging,
+	})
+
 	// Sleep so that workers can startup and connect.
 	time.Sleep(10 * time.Second)

-	return kmsWorker, pkiWorker, childPkiWorker
+	return kmsWorker, pkiWorker, childPkiWorker, childKmsWorker
 }

 // NewAuthorizedPkiTestWorker creates a new test worker with the provided upstreams
@@ -10,6 +10,7 @@ import (
 	"crypto/x509"
 	"math/big"
 	"net"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -165,15 +166,19 @@ func TestNewTestMultihopWorkers(t *testing.T) {
 	t.Cleanup(c.Shutdown)
 	pkiTags := map[string][]string{"connected": {"directly"}}
 	childPkiTags := map[string][]string{"connected": {"multihop"}}
+	childKmsTags := map[string][]string{"connected": {"multihop"}}

-	kmsWorker, pkiWorker, childPkiWorker := NewTestMultihopWorkers(t, logger, c.Context(), c.ClusterAddrs(),
-		c.Config().WorkerAuthKms, c.Controller().ServersRepoFn, pkiTags, childPkiTags)
+	enableAuthDebugging := new(atomic.Bool)
+	enableAuthDebugging.Store(true)
+	kmsWorker, pkiWorker, childPkiWorker, childKmsWorker := NewTestMultihopWorkers(t, logger, c.Context(), c.ClusterAddrs(),
+		c.Config().WorkerAuthKms, c.Controller().ServersRepoFn, pkiTags, childPkiTags, childKmsTags, enableAuthDebugging)

 	srvRepo, err := c.Controller().ServersRepoFn()
-	workers, err := srvRepo.ListWorkers(ctx, []string{"global"})
-	assert.Len(t, workers, 3)
 	require.NoError(t, err)
-	var kmsW, pkiW, childPkiW *server.Worker
+	workers, err := srvRepo.ListWorkers(ctx, []string{"global"})
+	assert.Len(t, workers, 4)
+	require.NoError(t, err)
+	var kmsW, pkiW, childPkiW, childKmsW *server.Worker
 	for _, w := range workers {
 		switch w.GetAddress() {
 		case kmsWorker.ProxyAddrs()[0]:
@@ -182,18 +187,23 @@ func TestNewTestMultihopWorkers(t *testing.T) {
 			pkiW = w
 		case childPkiWorker.ProxyAddrs()[0]:
 			childPkiW = w
+		case childKmsWorker.ProxyAddrs()[0]:
+			childKmsW = w
 		}
 	}
-	assert.NotNil(t, kmsW)
-	assert.NotNil(t, pkiW)
-	assert.NotNil(t, childPkiW)
+	require.NotNil(t, kmsW)
+	require.NotNil(t, pkiW)
+	require.NotNil(t, childPkiW)
+	require.NotNil(t, childKmsW)

 	assert.NotZero(t, kmsW.GetLastStatusTime())
 	assert.NotZero(t, pkiW.GetLastStatusTime())
 	assert.NotZero(t, childPkiW.GetLastStatusTime())
+	assert.NotZero(t, childKmsW.GetLastStatusTime())

 	assert.Equal(t, pkiTags, pkiW.GetConfigTags())
 	assert.Equal(t, childPkiTags, childPkiW.GetConfigTags())
+	assert.Equal(t, childKmsTags, childKmsW.GetConfigTags())
 }

 func createTestCert(t *testing.T) ([]byte, ed25519.PublicKey, ed25519.PrivateKey) {
@@ -36,6 +36,7 @@ import (
 	"github.com/hashicorp/nodeenrollment"
 	nodeenet "github.com/hashicorp/nodeenrollment/net"
 	nodeefile "github.com/hashicorp/nodeenrollment/storage/file"
+	nodeeinmem "github.com/hashicorp/nodeenrollment/storage/inmem"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mr-tron/base58"
 	"github.com/prometheus/client_golang/prometheus"
@@ -124,7 +125,7 @@ type Worker struct {
 	updateTags *ua.Bool

 	// The storage for node enrollment
-	WorkerAuthStorage             *nodeefile.Storage
+	WorkerAuthStorage             nodeenrollment.Storage
 	WorkerAuthCurrentKeyId        *ua.String
 	WorkerAuthRegistrationRequest string
 	workerAuthSplitListener       *nodeenet.SplitListener
@@ -285,35 +286,50 @@ func (w *Worker) Start() error {
 		return nil
 	}

-	if w.conf.WorkerAuthKms == nil || w.conf.DevUsePkiForUpstream {
+	w.operationalState.Store(server.UnknownOperationalState)
+
+	if !w.conf.RawConfig.Worker.UseDeprecatedKmsAuthMethod {
 		// In this section, we look for existing worker credentials. The two
-		// variables below store whether to create new credentials and whether
-		// to create a fetch request so it can be displayed in the worker
-		// startup info. These may be different because if initial creds have
-		// been generated on the worker side but not yet authorized/fetched from
-		// the controller, we don't want to invalidate that request on restart
-		// by generating a new set of credentials. However it's safe to output a
-		// new fetch request so we do in fact do that.
+		// variables below store whether to create new credentials and whether to
+		// create a fetch request so it can be displayed in the worker startup info.
+		// These may be different because if initial creds have been generated on
+		// the worker side but not yet authorized/fetched from the controller, we
+		// don't want to invalidate that request on restart by generating a new set
+		// of credentials. However it's safe to output a new fetch request so we do
+		// in fact do that.
 		//
-		// Note that if a controller-generated activation token has been
-		// supplied, we do not output a fetch request; we attempt to use that
-		// directly later.
+		// Note that if a controller-generated activation token has been supplied,
+		// we do not output a fetch request; we attempt to use that directly later.
+		//
+		// If we have a stable storage path we use that; if no path is supplied
+		// (e.g. when using KMS) we use inmem storage.
 		var err error
-		w.WorkerAuthStorage, err = nodeefile.New(w.baseContext,
-			nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.AuthStoragePath))
-		if err != nil {
-			return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error loading worker auth storage directory"))
+		if w.conf.RawConfig.Worker.AuthStoragePath != "" {
+			w.WorkerAuthStorage, err = nodeefile.New(w.baseContext,
+				nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.AuthStoragePath))
+			if err != nil {
+				return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error loading worker auth storage directory"))
+			}
+		} else {
+			w.WorkerAuthStorage, err = nodeeinmem.New(w.baseContext)
+			if err != nil {
+				return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error loading in-mem worker auth storage"))
+			}
 		}

 		var createNodeAuthCreds bool
 		var createFetchRequest bool
-		nodeCreds, err := types.LoadNodeCredentials(w.baseContext, w.WorkerAuthStorage, nodeenrollment.CurrentId, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms))
+		nodeCreds, err := types.LoadNodeCredentials(
+			w.baseContext,
+			w.WorkerAuthStorage,
+			nodeenrollment.CurrentId,
+			nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms))
 		switch {
 		case err == nil:
-			// It's unclear why this would ever happen -- it shouldn't -- so
-			// this is simply safety against panics if something goes
-			// catastrophically wrong
 			if nodeCreds == nil {
+				// It's unclear why this would ever happen -- it shouldn't -- so
+				// this is simply safety against panics if something goes
+				// catastrophically wrong
 				event.WriteSysEvent(w.baseContext, op, "no error loading worker auth creds but nil creds, creating new creds for registration")
 				createNodeAuthCreds = true
 				createFetchRequest = true
@@ -374,7 +390,7 @@ func (w *Worker) Start() error {
 				w.baseContext,
 				w.WorkerAuthStorage,
 				nodeenrollment.WithRandomReader(w.conf.SecureRandomReader),
-				nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms),
+				nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
 			)
 			if err != nil {
 				return fmt.Errorf("error generating new worker auth creds: %w", err)
@@ -172,7 +172,7 @@ func TestSetupWorkerAuthStorage(t *testing.T) {
 	// Get an initial set of authorized node credentials
 	initStorage, err := nodeefile.New(ctx)
 	require.NoError(t, err)
-	t.Cleanup(initStorage.Cleanup)
+	t.Cleanup(func() { initStorage.Cleanup(ctx) })
 	_, err = rotation.RotateRootCertificates(ctx, initStorage)
 	require.NoError(t, err)
 	initNodeCreds, err := types.NewNodeCredentials(ctx, initStorage)
@@ -29,24 +29,27 @@ func TestDeleteConnectedWorkers(t *testing.T) {
 		Logger: logger.Named("controller"),
 	})
 	t.Cleanup(c.Shutdown)
-	_, directPkiWorker, multiHoppedPkiWorker := NewTestMultihopWorkers(t, logger, c.Context(), c.ClusterAddrs(),
-		c.Config().WorkerAuthKms, c.Controller().ServersRepoFn, nil, nil)
+	_, directPkiWorker, multiHoppedPkiWorker, multiHoppedKmsWorker := NewTestMultihopWorkers(t, logger, c.Context(), c.ClusterAddrs(),
+		c.Config().WorkerAuthKms, c.Controller().ServersRepoFn, nil, nil, nil, nil)

 	serverRepo, err := c.Controller().ServersRepoFn()
 	require.NoError(t, err)
 	workers, err := serverRepo.ListWorkers(ctx, []string{"global"}, server.WithLiveness(-1))
 	require.NoError(t, err)
-	var childWorker *server.Worker
-	var pkiWorker *server.Worker
+	var pkiWorker, childPkiWorker, childKmsWorker *server.Worker
 	for _, w := range workers {
 		if w.Type == "pki" && w.GetAddress() == multiHoppedPkiWorker.ProxyAddrs()[0] {
-			childWorker = w
+			childPkiWorker = w
+		}
+		if w.Type == "pki" && w.GetAddress() == multiHoppedKmsWorker.ProxyAddrs()[0] {
+			childKmsWorker = w
 		}
 		if w.Type == "pki" && w.GetAddress() == directPkiWorker.ProxyAddrs()[0] {
 			pkiWorker = w
 		}
 	}
-	require.NotNil(t, childWorker)
+	require.NotNil(t, childKmsWorker)
+	require.NotNil(t, childPkiWorker)
 	require.NotNil(t, pkiWorker)

 	cases := []struct {
@@ -55,10 +58,15 @@ func TestDeleteConnectedWorkers(t *testing.T) {
 		testWorker *TestWorker
 	}{
 		{
-			name:       "multi hop worker",
-			workerId:   childWorker.GetPublicId(),
+			name:       "multi hop pki worker",
+			workerId:   childPkiWorker.GetPublicId(),
 			testWorker: multiHoppedPkiWorker,
 		},
+		{
+			name:       "multi hop kms worker",
+			workerId:   childKmsWorker.GetPublicId(),
+			testWorker: multiHoppedKmsWorker,
+		},
 		{
 			name:       "directly connected worker",
 			workerId:   pkiWorker.GetPublicId(),
@@ -262,8 +262,6 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker id is not empty")
 	case worker.GetName() == "" && opts.withKeyId == "":
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; one is required")
-	case worker.GetName() != "" && opts.withKeyId != "":
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both set; no more than one is allowed")
 	case worker.OperationalState == "":
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "worker operational state is empty")
 	}
@@ -300,17 +298,16 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt
 			workerClone.PublicId = workerId
 			switch {
 			case opts.withKeyId != "":
-				// This case goes first in case we want to relax the restriction
-				// around both name and key ID being supplied to account for
-				// e.g. config processing bugs. In that case, a key ID being
-				// supplied should be a clear indicator that we are working with
-				// a PKI workerClone, and a lack of one a clear indication we are
+				// This case goes first because a key ID being supplied should
+				// be a clear indicator that we are working with a PKI
+				// workerClone, and a lack of one a clear indication we are
 				// working with a KMS workerClone.
 				//
 				// Note: unlike in the below case, this purposefully leaves out
-				// "description" since we want description changes for PKI-based
-				// workers to come via API only. We can't really guard on this
-				// in the DB so we need to be sure to not include it here.
+				// "description" since we want description changes for (non
+				// KMS-PKI) PKI-based workers to come via API only. We can't
+				// really guard on this in the DB so we need to be sure to not
+				// include it here.
 				n, err := w.Update(ctx, workerClone, []string{"address", "ReleaseVersion", "OperationalState"}, nil)
 				if err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update status of pki worker"))
@@ -583,7 +580,7 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to authorize node"))
 				}
 				nodeInfo.State = state
-				if err := StoreNodeInformationTx(ctx, w, databaseWrapper, nodeInfo); err != nil {
+				if err := StoreNodeInformationTx(ctx, read, w, r.kms, worker.ScopeId, nodeInfo); err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to store node information"))
 				}

@@ -380,18 +380,6 @@ func TestUpsertWorkerStatus(t *testing.T) {
 				assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
 			},
 		},
-		{
-			name: "name and key id provided",
-			repo: repo,
-			status: server.NewWorker(scope.Global.String(),
-				server.WithName("name and key id provided"),
-				server.WithAddress("someaddress")),
-			options: []server.Option{server.WithKeyId(pkiWorkerKeyId)},
-			errAssert: func(t *testing.T, err error) {
-				t.Helper()
-				assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err), err)
-			},
-		},
 		{
 			name: "no name or key id",
 			repo: repo,
@@ -469,6 +457,20 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		assert.Len(t, workers, 3)
 	})

+	t.Run("name and key id provided", func(t *testing.T) {
+		anotherStatus := server.NewWorker(scope.Global.String(),
+			server.WithName("name-and-keyid"),
+			server.WithAddress("address2"),
+			server.WithReleaseVersion("Boundary v0.11.0"),
+			server.WithKeyId(pkiWorkerKeyId))
+		_, err = repo.UpsertWorkerStatus(ctx, anotherStatus)
+		require.NoError(t, err)
+
+		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()})
+		require.NoError(t, err)
+		assert.Len(t, workers, 4)
+	})
+
 	t.Run("send shutdown status", func(t *testing.T) {
 		anotherStatus := server.NewWorker(scope.Global.String(),
 			server.WithName("another_test_worker"),
@@ -478,10 +480,10 @@ func TestUpsertWorkerStatus(t *testing.T) {
 		_, err = repo.UpsertWorkerStatus(ctx, anotherStatus)
 		require.NoError(t, err)

-		// Filtering out shutdown workers will remove the shutdown KMS and this shutdown worker, resulting in 1
+		// Filtering out shutdown workers will remove the shutdown KMS and this shutdown worker, resulting in 2
 		workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true))
 		require.NoError(t, err)
-		assert.Len(t, workers, 1)
+		assert.Len(t, workers, 2)
 	})
 }

@@ -1044,7 +1046,7 @@ func TestRepository_CreateWorker(t *testing.T) {
 					// This happens on the worker
 					fileStorage, err := file.New(testCtx)
 					require.NoError(t, err)
-					defer fileStorage.Cleanup()
+					defer func() { fileStorage.Cleanup(testCtx) }()

 					nodeCreds, err := types.NewNodeCredentials(testCtx, fileStorage)
 					require.NoError(t, err)
@@ -1136,7 +1138,7 @@ func TestRepository_CreateWorker(t *testing.T) {
 					// This happens on the worker
 					fileStorage, err := file.New(testCtx)
 					require.NoError(t, err)
-					defer fileStorage.Cleanup()
+					defer func() { fileStorage.Cleanup(testCtx) }()

 					nodeCreds, err := types.NewNodeCredentials(testCtx, fileStorage)
 					require.NoError(t, err)
@@ -14,14 +14,15 @@ import (
 	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/timestamppb"

+	"github.com/hashicorp/boundary/internal/daemon/cluster"
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/server/store"
 	"github.com/hashicorp/boundary/internal/types/scope"
+	"github.com/hashicorp/boundary/internal/util"
 	"github.com/hashicorp/go-dbw"
-	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
 	"github.com/hashicorp/nodeenrollment"
 	nodee "github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/types"
@@ -80,12 +81,9 @@ func (r *WorkerAuthRepositoryStorage) Store(ctx context.Context, msg nodee.Messa
 	switch t := msg.(type) {
 	case *types.NodeInformation:
 		// Encrypt the private key
-		databaseWrapper, err := r.kms.GetWrapper(ctx, scope.Global.String(), kms.KeyPurposeDatabase)
-		if err != nil {
-			return errors.Wrap(ctx, err, op)
-		}
+
 		if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(read db.Reader, w db.Writer) error {
-			return StoreNodeInformationTx(ctx, w, databaseWrapper, t)
+			return StoreNodeInformationTx(ctx, read, w, r.kms, scope.Global.String(), t)
 		}); err != nil {
 			return errors.Wrap(ctx, err, op)
 		}
@@ -114,18 +112,77 @@ func (r *WorkerAuthRepositoryStorage) Store(ctx context.Context, msg nodee.Messa
 // Node information is stored in two parts:
 // * the workerAuth record is stored with a reference to a worker
 // * certificate bundles are stored with a reference to the workerAuth record and issuing root certificate
-func StoreNodeInformationTx(ctx context.Context, writer db.Writer, databaseWrapper wrapping.Wrapper, node *types.NodeInformation, _ ...Option) error {
+func StoreNodeInformationTx(ctx context.Context, reader db.Reader, writer db.Writer, kmsCache *kms.Kms, scopeId string, node *types.NodeInformation, _ ...Option) error {
 	const op = "server.(WorkerAuthRepositoryStorage).storeNodeInformation"
+	if isNil(reader) {
+		return errors.New(ctx, errors.InvalidParameter, op, "missing reader")
+	}
 	if isNil(writer) {
 		return errors.New(ctx, errors.InvalidParameter, op, "missing writer")
 	}
-	if isNil(databaseWrapper) {
-		return errors.New(ctx, errors.InvalidParameter, op, "missing database wrapper")
+	if isNil(kmsCache) {
+		return errors.New(ctx, errors.InvalidParameter, op, "missing kms")
+	}
+	if scopeId == "" {
+		return errors.New(ctx, errors.InvalidParameter, op, "missing scope id")
 	}
 	if node == nil {
 		return errors.New(ctx, errors.InvalidParameter, op, "missing NodeInformation")
 	}

+	databaseWrapper, err := kmsCache.GetWrapper(ctx, scopeId, kms.KeyPurposeDatabase)
+	if err != nil {
+		return errors.Wrap(ctx, err, op)
+	}
+
+	var workerId string
+	if node.WrappingRegistrationFlowInfo != nil {
+		if util.IsNil(node.WrappingRegistrationFlowInfo.ApplicationSpecificParams) {
+			return errors.New(ctx, errors.InvalidParameter, op, "in wrapping registration flow but application specific parameters not provided")
+		}
+		// In this case we are using the wrapping registration flow and it's
+		// been validated; we need to upsert first to ensure that the worker
+		// exists in the database.
+		wci := new(cluster.WorkerConnectionInfo)
+		if err := mapstructure.Decode(node.WrappingRegistrationFlowInfo.ApplicationSpecificParams.AsMap(), wci); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		if wci.Name == "" {
+			return errors.New(ctx, errors.InvalidParameter, op, "in wrapping registration flow but worker name not provided")
+		}
+		if wci.BoundaryVersion == "" {
+			return errors.New(ctx, errors.InvalidParameter, op, "in wrapping registration flow but boundary version not provided")
+		}
+		wConf := NewWorker(scopeId,
+			WithName(wci.Name),
+			WithDescription(wci.Description),
+			WithReleaseVersion(fmt.Sprintf("Boundary v%s", wci.BoundaryVersion)),
+		)
+		wConf.Type = PkiWorkerType.String()
+		wConf.PublicId, err = newWorkerIdFromScopeAndName(ctx, scopeId, wci.Name)
+		if err != nil || wConf.PublicId == "" {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id"))
+		}
+		workerCreateConflict := &db.OnConflict{
+			Target: db.Columns{"public_id"},
+			Action: db.SetColumns([]string{"release_version", "name", "description"}),
+		}
+		var withRowsAffected int64
+		err = writer.Create(
+			ctx,
+			wConf,
+			db.WithOnConflict(workerCreateConflict),
+			db.WithReturnRowsAffected(&withRowsAffected),
+		)
+		switch {
+		case err != nil:
+			return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker"))
+		case withRowsAffected == 0:
+			return errors.New(ctx, errors.NotUnique, op, "error updating worker")
+		}
+		workerId = wConf.PublicId
+	}
+
 	nodeAuth := allocWorkerAuth()
 	nodeAuth.WorkerKeyIdentifier = node.Id
 	nodeAuth.WorkerEncryptionPubKey = node.EncryptionPublicKeyBytes
@@ -137,12 +194,15 @@ func StoreNodeInformationTx(ctx context.Context, writer db.Writer, databaseWrapp
 		return errors.Wrap(ctx, err, op)
 	}

-	// Get workerId from state passed in
-	var result workerAuthWorkerId
-	if err := mapstructure.Decode(node.State.AsMap(), &result); err != nil {
-		return errors.Wrap(ctx, err, op)
+	if workerId == "" {
+		// Get workerId from state passed in
+		var result workerAuthWorkerId
+		if err := mapstructure.Decode(node.State.AsMap(), &result); err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		workerId = result.WorkerId
 	}
-	nodeAuth.WorkerId = result.WorkerId
+	nodeAuth.WorkerId = workerId

 	if err := nodeAuth.ValidateNewWorkerAuth(ctx); err != nil {
 		return errors.Wrap(ctx, err, op)
@@ -24,6 +24,7 @@ import (
 	"github.com/hashicorp/nodeenrollment/registration"
 	"github.com/hashicorp/nodeenrollment/rotation"
 	"github.com/hashicorp/nodeenrollment/storage/file"
+	"github.com/hashicorp/nodeenrollment/storage/inmem"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mitchellh/mapstructure"
 	"github.com/mr-tron/base58"
@@ -314,8 +315,6 @@ func TestStoreNodeInformationTx(t *testing.T) {
 	kmsCache := kms.TestKms(t, conn, testWrapper)
 	// Ensures the global scope contains a valid root key
 	require.NoError(t, kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader)))
-	databaseWrapper, err := kmsCache.GetWrapper(context.Background(), scope.Global.String(), kms.KeyPurposeDatabase)
-	require.NoError(t, err)

 	testRootStorage, err := NewRepositoryStorage(testCtx, rw, rw, kmsCache)
 	require.NoError(t, err)
@@ -332,9 +331,9 @@ func TestStoreNodeInformationTx(t *testing.T) {

 	testNodeInfoFn := func() *types.NodeInformation {
 		// This happens on the worker
-		fileStorage, err := file.New(testCtx)
+		storage, err := inmem.New(testCtx)
 		require.NoError(t, err)
-		nodeCreds, err := types.NewNodeCredentials(testCtx, fileStorage)
+		nodeCreds, err := types.NewNodeCredentials(testCtx, storage)
 		require.NoError(t, err)

 		nodePubKey, err := curve25519.X25519(nodeCreds.EncryptionPrivateKeyBytes, curve25519.Basepoint)
@@ -353,10 +352,33 @@ func TestStoreNodeInformationTx(t *testing.T) {
 		return nodeInfo
 	}

+	// For swapping out key ID for wrapping registration flow
+	wrappingRegFlowStorage, err := inmem.New(testCtx)
+	require.NoError(t, err)
+	wrappingRegFlowNodeCreds, err := types.NewNodeCredentials(testCtx, wrappingRegFlowStorage)
+	require.NoError(t, err)
+	wrappingRegFlowNodeInfoFn := func() *types.NodeInformation {
+		ni := testNodeInfoFn()
+		wci, err := structpb.NewStruct(map[string]any{
+			"name":             "regflow-worker-name",
+			"description":      "regflow-worker-description",
+			"boundary_version": "0.13.0",
+		})
+		require.NoError(t, err)
+		ni.WrappingRegistrationFlowInfo = &types.WrappingRegistrationFlowInfo{
+			Nonce:                     ni.RegistrationNonce,
+			CertificatePublicKeyPkix:  ni.CertificatePublicKeyPkix,
+			ApplicationSpecificParams: wci,
+		}
+		return ni
+	}
+
 	tests := []struct {
 		name            string
+		reader          db.Reader
 		writer          db.Writer
-		databaseWrapper wrapping.Wrapper
+		scope           string
+		kms             *kms.Kms
 		node            *types.NodeInformation
 		wantErr         bool
 		wantErrIs       errors.Code
@@ -364,61 +386,70 @@ func TestStoreNodeInformationTx(t *testing.T) {
 	}{
 		{
 			name:            "missing-writer",
-			databaseWrapper: databaseWrapper,
+			reader:          rw,
+			scope:           scope.Global.String(),
+			kms:             kmsCache,
 			node:            testNodeInfoFn(),
 			wantErr:         true,
 			wantErrIs:       errors.InvalidParameter,
 			wantErrContains: "missing writer",
 		},
 		{
-			name:            "missing-wrapper",
+			name:            "missing-reader",
 			writer:          rw,
+			scope:           scope.Global.String(),
+			kms:             kmsCache,
 			node:            testNodeInfoFn(),
 			wantErr:         true,
 			wantErrIs:       errors.InvalidParameter,
-			wantErrContains: "missing database wrapper",
+			wantErrContains: "missing reader",
+		},
+		{
+			name:            "missing-scope",
+			reader:          rw,
+			writer:          rw,
+			kms:             kmsCache,
+			node:            testNodeInfoFn(),
+			wantErr:         true,
+			wantErrIs:       errors.InvalidParameter,
+			wantErrContains: "missing scope",
+		},
+		{
+			name:            "missing-kms",
+			reader:          rw,
+			writer:          rw,
+			scope:           scope.Global.String(),
+			node:            testNodeInfoFn(),
+			wantErr:         true,
+			wantErrIs:       errors.InvalidParameter,
+			wantErrContains: "missing kms",
 		},
 		{
 			name:            "missing-node",
+			reader:          rw,
 			writer:          rw,
-			databaseWrapper: databaseWrapper,
+			scope:           scope.Global.String(),
+			kms:             kmsCache,
 			wantErr:         true,
 			wantErrIs:       errors.InvalidParameter,
 			wantErrContains: "missing NodeInformation",
 		},
-		{
-			name:            "key-id-error",
-			writer:          rw,
-			databaseWrapper: &mockTestWrapper{err: errors.New(testCtx, errors.Internal, "testing", "key-id-error")},
-			node:            testNodeInfoFn(),
-			wantErr:         true,
-			wantErrIs:       errors.Encrypt,
-			wantErrContains: "reading cipher key id",
-		},
-		{
-			name:   "encrypt-error",
-			writer: rw,
-			databaseWrapper: &mockTestWrapper{
-				encryptError: true,
-				err:          errors.New(testCtx, errors.Encrypt, "testing", "encrypt-error"),
-			},
-			node:            testNodeInfoFn(),
-			wantErr:         true,
-			wantErrIs:       errors.Encrypt,
-			wantErrContains: "encrypt-error",
-		},
 		{
 			name:            "create-error-no-db",
+			reader:          rw,
+			scope:           scope.Global.String(),
 			writer:          &db.Db{},
-			databaseWrapper: databaseWrapper,
+			kms:             kmsCache,
 			node:            testNodeInfoFn(),
 			wantErr:         true,
 			wantErrContains: "db.Create",
 		},
 		{
-			name:            "create-error-validation",
-			writer:          rw,
-			databaseWrapper: databaseWrapper,
+			name:   "create-error-validation",
+			reader: rw,
+			writer: rw,
+			scope:  scope.Global.String(),
+			kms:    kmsCache,
 			node: func() *types.NodeInformation {
 				ni := testNodeInfoFn()
 				ni.State = nil
@@ -428,16 +459,69 @@ func TestStoreNodeInformationTx(t *testing.T) {
 			wantErrContains: "missing WorkerId",
 		},
 		{
-			name:            "success",
-			writer:          rw,
-			databaseWrapper: databaseWrapper,
-			node:            testNodeInfoFn(),
+			name:   "wrapflow-no-name",
+			reader: rw,
+			writer: rw,
+			scope:  scope.Global.String(),
+			kms:    kmsCache,
+			node: func() *types.NodeInformation {
+				ni := wrappingRegFlowNodeInfoFn()
+				ni.WrappingRegistrationFlowInfo.ApplicationSpecificParams.Fields["name"] = nil
+				ni.CertificatePublicKeyPkix = wrappingRegFlowNodeCreds.CertificatePublicKeyPkix
+				keyId, err := nodeenrollment.KeyIdFromPkix(ni.CertificatePublicKeyPkix)
+				require.NoError(t, err)
+				ni.Id = keyId
+				return ni
+			}(),
+			wantErr:         true,
+			wantErrContains: "in wrapping registration flow but worker name not provided",
+		},
+		{
+			name:   "wrapflow-no-version",
+			reader: rw,
+			writer: rw,
+			scope:  scope.Global.String(),
+			kms:    kmsCache,
+			node: func() *types.NodeInformation {
+				ni := wrappingRegFlowNodeInfoFn()
+				ni.WrappingRegistrationFlowInfo.ApplicationSpecificParams.Fields["boundary_version"] = nil
+				ni.CertificatePublicKeyPkix = wrappingRegFlowNodeCreds.CertificatePublicKeyPkix
+				keyId, err := nodeenrollment.KeyIdFromPkix(ni.CertificatePublicKeyPkix)
+				require.NoError(t, err)
+				ni.Id = keyId
+				return ni
+			}(),
+			wantErr:         true,
+			wantErrContains: "in wrapping registration flow but boundary version not provided",
+		},
+		{
+			name:   "success",
+			reader: rw,
+			writer: rw,
+			scope:  scope.Global.String(),
+			kms:    kmsCache,
+			node:   testNodeInfoFn(),
+		},
+		{
+			name:   "success-wrapflow",
+			reader: rw,
+			writer: rw,
+			scope:  scope.Global.String(),
+			kms:    kmsCache,
+			node: func() *types.NodeInformation {
+				ni := wrappingRegFlowNodeInfoFn()
+				ni.CertificatePublicKeyPkix = wrappingRegFlowNodeCreds.CertificatePublicKeyPkix
+				keyId, err := nodeenrollment.KeyIdFromPkix(ni.CertificatePublicKeyPkix)
+				require.NoError(t, err)
+				ni.Id = keyId
+				return ni
+			}(),
 		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
-			err := StoreNodeInformationTx(testCtx, tc.writer, tc.databaseWrapper, tc.node)
+			err := StoreNodeInformationTx(testCtx, tc.reader, tc.writer, tc.kms, tc.scope, tc.node)
 			if tc.wantErr {
 				require.Error(err)
 				if tc.wantErrIs != errors.Unknown {
@@ -46,6 +46,7 @@ func TestWorkerReplay(t *testing.T) {
 		NonceFn: func(length int) (string, error) {
 			return "test_noncetest_nonce", nil
 		},
+		UseDeprecatedKmsAuthMethod: true,
 	})

 	// Give time for it to connect
@@ -72,6 +73,7 @@ func TestWorkerReplay(t *testing.T) {
 		NonceFn: func(length int) (string, error) {
 			return "test_noncetest_nonce", nil
 		},
+		UseDeprecatedKmsAuthMethod: true,
 	})

 	// Give time for it to connect
@@ -48,7 +48,7 @@ func GetWrapperFromPath(ctx context.Context, path, purpose string, opt ...config
 }

 func GetWrapperFromHcl(ctx context.Context, inHcl, purpose string, opt ...configutil.Option) (wrapping.Wrapper, func() error, error) {
-	kmses, err := configutil.ParseKMSes(inHcl)
+	kmses, err := configutil.ParseKMSes(inHcl, configutil.WithMaxKmsBlocks(-1))
 	if err != nil {
 		return nil, nil, fmt.Errorf("Error parsing KMS HCL: %w", err)
 	}
@@ -6,23 +6,39 @@ description: |-
 ---

 # `worker` Stanza
+The `worker` stanza configures Boundary worker-specific parameters.

-The `worker` stanza configures Boundary worker-specific parameters. Boundary
-supports two different types of workers, differentiated by their means of
-authentication to Boundary:
+All workers within Boundary use certificates and encryption keys to identify
+themselves and protect data in transit. However, there are multiple different
+ways to register them so that registration of workers can fit into any workflow.
+Registration methods are referred to as `pki` and `kms` registration methods and
+the differences in how they are configured are in the sub-pages linked at the
+bottom of this page.

- [PKI Workers][] use certificates issued by Boundary to authenticate with controllers
- [KMS Workers][] use a shared KMS to authenticate with controllers
+Workers registered via the `pki` method ("PKI Workers") must be registered in
+the system via an API call, and require storage on disk to store the current set
+of credentials. Workers registering via the `kms` method ("KMS Workers")
+auto-register after successful authentication, making them an easy mechanism to
+use for automatic scaling, and meaning they are not required to store
+credentials locally; each time they connect the KMS is used to reauthenticate
+them.

-Different worker types have different configuration requirements, but share the common worker parameters listed below.
+~> Prior to version 0.13 of Boundary, workers had different properties and
+capabilities depending on the registration method. If you are using pre-0.13
+workers, with pre-0.13 upstreams please be sure to switch the documentation
+version to `0.12.x` for correct information. If you are using 0.13 workers with
+KMS-based authentication with pre-0.13 controllers, you _must_ set the
+`use_deprecated_kms_auth_method` value in the worker configuration. See the KMS
+page for more information.

 ## Common Worker Parameters
-The following fields are supported for all worker types:
+Regardless of registration mechanism, the following fields are supported.

 ```hcl
 worker {
  public_addr = "5.1.23.198"

+  # Mutually exclusive with hcp_boundary_cluster_id
  initial_upstreams = [
    "10.0.0.1",
    "10.0.0.2",
@@ -32,14 +48,17 @@ worker {
    type   = ["prod", "webservers"]
    region = ["us-east-1"]
  }
+
+  # HCP Boundary only
+  # hcp_boundary_cluster_id = "....."
 }
 ```

 - `public_addr` - Specifies the public host or IP address (and optionally port)
  at which the worker can be reached _by clients for proxying_. This defaults to
  the address of the listener marked for `proxy` purpose. This is especially
-  useful for cloud environments that do not bind a publicly accessible IP to a NIC
-  on the host directly, such as an Amazon EIP. 
+  useful for cloud environments that do not bind a publicly accessible IP to a
+  NIC on the host directly, such as an Amazon EIP. 
  
  This value can reference any of the following:
  - a direct address string
@@ -50,15 +69,118 @@ worker {
  reaching the boundary cluster. The port will default to `:9201` if not
  specified. This value can be a direct access string array with the addresses,
  or it can refer to a file on disk (`file://`) from which the addresses will be
-  read, or an env var (`env://`) from which the addresses will be read. When using
-  env or file, their contents must formatted as a JSON array: `["127.0.0.1",
-  "192.168.0.1", "10.0.0.1"]`
+  read, or an env var (`env://`) from which the addresses will be read. When
+  using env or file, their contents must formatted as a JSON array:
+  `["127.0.0.1", "192.168.0.1", "10.0.0.1"]`

 - `tags` - A map of key-value pairs where values are an array of strings. Most
-  commonly used for [filtering](/boundary/docs/concepts/filtering) targets a worker can
-  proxy via [worker tags](/boundary/docs/concepts/filtering/worker-tags). On `SIGHUP`, the
-  tags set here will be re-parsed and new values used. It can also be a string
+  commonly used for [filtering](/boundary/docs/concepts/filtering) targets a
+  worker can proxy via [worker
+  tags](/boundary/docs/concepts/filtering/worker-tags). On `SIGHUP`, the tags
+  set here will be re-parsed and new values used. It can also be a string
  referring to a file on disk (`file://`) or an env var (`env://`).

+- `hcp_boundary_cluster_id` - A string that can be used to configure PKI workers
+  to connect to your HCP Boundary cluster rather than specifying
+  `initial_upstreams`. This is currently only valid for workers using the PKI
+  registration method and for workers directly connected to HCP Boundary.
+
+## Multi-hop worker capabilities <sup>HCP only</sup>
+Multi-hop capabilities, including multi-hop sessions and Vault private access,
+is when a session or Vault credential request goes through more than one worker.
+To enable this, two or more workers must be connected to each other in some
+configuration. There are no limits on the amount of workers allowed in a
+multi-hop session configuration.
+
+It helps to think of “upstream” and “downstream” nodes in the context of
+multi-hope. If you view controllers as the “top” node of a multi-hop chain, any
+worker connected to a node is a "downstream" of that node; the node that any
+particular worker connects to (whether another worker or a controller) is the
+"upstream" of that node. For example, in the diagram below, Worker 2’s upstream
+is Worker 1, and its downstream is Worker 3.
+
+![multi-hop workers](/img/multi-hop-workers.png)
+
+You can deploy multi-hop workers in scenarios where inbound network traffic is
+not allowed. A worker in a private network can send outbound communication to
+its upstream worker, and create a reverse proxy to establish a session.
+
+You can configure [target worker filters][] with multi-hop workers to allow for
+fine-grained control on which workers handle ingress and egress for session
+traffic to a [target][]. Ingress worker filters determine which workers you
+connect with to initiate a session, and egress worker filters determine which
+workers are used to access targets.
+
+### Multi-hop worker requirements
+
+When you configure multi-hop sessions, there is an "ingress" worker, an "egress"
+worker, and any number of intermediary workers. Ingress, egress, and
+intermediary workers have the following requirements.
+
+#### Ingress worker requirements
+
+To proxy target connections, ingress workers require outbound access to the
+Boundary control plane and inbound access from clients.
+
+#### Intermediary worker requirements
+
+Intermediary workers require outbound access to an upstream worker. The upstream
+worker may be an ingress worker or another intermediary worker. Intermediary
+workers also require inbound access from a downstream worker. The downstream
+worker may be an egress worker or another intermediary worker.
+
+#### Egress worker requirements
+
+To proxy target connections, egress workers require outbound access to an
+upstream worker and outbound access to the destination host or service.
+
+## Complete configuration example
+
+```hcl
+listener "tcp" {
+	purpose = "proxy"
+	tls_disable = true
+	address = "127.0.0.1"
+}
+
+worker {
+  # Path for worker storage, assuming PKI registration. Must be unique across workers
+  auth_storage_path="/boundary/demo-worker-1"
+
+  # Workers typically need to reach upstreams on :9201
+  initial_upstreams = [
+    "10.0.0.1",
+    "10.0.0.2",
+    "10.0.0.3",
+  ]
+
+  public_addr = "myhost.mycompany.com"
+
+  tags {
+    type   = ["prod", "webservers"]
+    region = ["us-east-1"]
+  }
+}
+
+# The following KMS config is an example only
+# Use a production KMS such as AWS KMS for production installs
+kms "aead" {
+  purpose = "worker-auth-storage"
+	aead_type = "aes-gcm"
+	key = "X+IJMVT6OnsrIR6G/9OTcJSX+lM9FSPN"
+	key_id = "worker-auth-storage"
+}
+
+
+```
+
+## Tutorials
+
+Refer to the [Self-Managed Worker Registration with HCP Boundary](/boundary/tutorials/hcp-administration/hcp-manage-workers) tutorial to learn how to register and manage PKI workers.
+
+Refer to the [Manage Multi-Hop Sessions with HCP Boundary](/boundary/tutorials/hcp-administration/hcp-manage-multi-hop) tutorial to learn how to configure a multi-hop session.
+
 [kms workers]: /boundary/docs/configuration/worker/kms-worker
 [pki workers]: /boundary/docs/configuration/worker/pki-worker
+[target]: /boundary/docs/concepts/domain-model/targets
+[target worker filters]: /boundary/docs/concepts/filtering/worker-tags#target-worker-filtering
@@ -8,16 +8,26 @@ description: |-

 ## KMS Worker Configuration

-KMS Workers require a `name` field. This specifies a unique name of this worker within the Boundary cluster.
-The `name` value can be:
+This page describes configuration for workers that authenticate to upstreams
+using a shared KMS. This mechanism auto-registers the worker in addition to
+authenticating it, and does not require on-disk storage for credentials since
+each time it connects it reauthenticates using the trusted KMS.
+
+~> If using 0.13+ workers with pre-0.13 controllers you _must_ set the
+`use_deprecated_kms_auth_method` value to true. Additionally, a worker
+registered using this method _against a pre-0.13 controller_ will only be able
+to register directly to a controller, and cannot be used as part of multi-hop or
+Vault private access capabilities.
+
+KMS Workers require a `name` field. This specifies a unique name of this worker
+within the Boundary cluster and _must be unique across workers_. The `name`
+value can be:
 - a direct name string (must be all lowercase)
 - a reference to a file on disk (`file://`) from which the name is read
 - an env var (`env://`) from which the name is read.

-~> **Note:** Changing the `name` results in the creation of a new worker in the
-cluster and assignment of a new worker ID.
-
-KMS Workers accept an optional `description` field. The `description` value can be:
+KMS Workers accept an optional `description` field. The `description` value can
+be:
 - a direct description string
 - a reference to a file on disk (`file://`) from which the name is read
 - an env var (`env://`) from which the name is read.
@@ -27,38 +37,55 @@ worker {
  name = "example-worker"
  description = "An example worker"
  public_addr = "5.1.23.198"
+  # Uncomment if using 0.13 worker against pre-0.13 controller
+  # use_deprecated_kms_auth_method = true
 }
 ```

-KMS Workers also require a KMS block designated for `worker-auth`. This is the KMS configuration for
-authentication between the workers and controllers and must be present. Example (not safe for production!):
+KMS Workers also require a KMS block designated for `worker-auth`. This is the
+KMS configuration for authentication between the workers and controllers and
+must be present. Example (not safe for production!):

 ```hcl
-  kms "aead" {
+kms "aead" {
 	purpose = "worker-auth"
 	aead_type = "aes-gcm"
-	key = "8fZBjCUfN0TzjEGLQldGY4+iE9AkOvCfjh7+p0GtRBQ="
+	key = "X+IJMVT6OnsrIR6G/9OTcJSX+lM9FSPN"
 	key_id = "global_worker-auth"
 }
 ```

-This configuration must be the same for the worker-auth configuration for the controller if you're
-running the controller and worker as separate servers.
-
-Optionally, a KMS stanza for configuration encryption purpose can also be specified:
+The upstream controller or worker must have a `kms` block that references the
+same key and purpose. (If both a controller and worker are running as the same
+server process only one stanza is needed.) It is also possible to specify a
+`kms` block with the `downstream-worker-auth` purpose. If specified, this will
+be a separate KMS that can be used for authenticating new downstream nodes.
+Blocks with this purpose can be specified multiple times. This allows a single
+upstream node to authenticate with one key to its own upstream (via the
+`worker-auth` purpose) and then serve as an authenticating upstream to nodes
+across various networks, each with their own separate KMS system or key:

 ```hcl
-# Configuration encryption block: decrypts sensitive values in the
-# configuration file. See `boundary config [encrypt|decrypt] -h`.
 kms "aead" {
-  purpose   = "config"`
-  aead_type = "aes-gcm"
-  key       = "7xtkEoS5EXPbgynwd+dDLHopaCqK8cq0Rpep4eooaTs="
+	purpose = "downstream-worker-auth"
+	aead_type = "aes-gcm"
+	key = "XthZVtFtBD1Bw1XwAWhZKVrIwRhR7HcZ"
+	key_id = "iot-nodes-auth"
+}
+
+kms "aead" {
+	purpose = "downstream-worker-auth"
+	aead_type = "aes-gcm"
+	key = "OLFhJNbEb3umRjdhY15QKNEmNXokY1Iq"
+	key_id = "production-nodes-auth"
 }
 ```

-Boundary supports many kinds of KMS integrations. For a complete guide to all available
-KMS types, see our [KMS documentation](/boundary/docs/configuration/kms).
+In the examples above we are encoding key bytes directly in the configuration
+file. This is because we are using the `aead` method where you directly supply a
+key; in production you'd want to use a KMS such as AWS KMS, GCP CKMS, Azure
+KeyVault, or HashiCorp Vault. For a complete guide to all available KMS types,
+see our [KMS documentation](/boundary/docs/configuration/kms).

 # Complete Configuration Example

@@ -87,13 +114,15 @@ worker {
    type   = ["prod", "webservers"]
    region = ["us-east-1"]
  }
+
+  # use_deprecated_kms_auth_method = true
 }

 # must be same key as used on controller config
 kms "aead" {
 	purpose = "worker-auth"
 	aead_type = "aes-gcm"
-	key = "8fZBjCUfN0TzjEGLQldGY4+iE9AkOvCfjh7+p0GtRBQ="
+	key = "X+IJMVT6OnsrIR6G/9OTcJSX+lM9FSPN"
 	key_id = "global_worker-auth"
 }
 ```
@@ -7,11 +7,9 @@ description: |-


 # PKI worker configuration
-PKI Workers authenticate to Boundary using a certificate-based method, allowing
-for worker deployment without using a shared KMS.
-
-PKI Workers require an accessible directory defined by `auth_storage_path` for
-credential storage.
+PKI Workers authenticate to Boundary using an activation token. They require an
+accessible directory defined by `auth_storage_path` for credential storage and
+rotation.

 Example (not safe for production!):

@@ -62,14 +60,15 @@ on the CLI this would be via `boundary workers create worker-led
 -worker-generated-auth-token`. No values are needed in the configuration file.

 ## KMS configuration
-PKI Workers credentials can be encrypted by including an optional KMS stanza with the purpose `worker-auth-storage`.
+PKI Workers' credentials can be encrypted by including an optional KMS stanza
+with the purpose `worker-auth-storage`.

 Example (not safe for production!):
 ```hcl
 kms "aead" {
  purpose = "worker-auth-storage"
 	aead_type = "aes-gcm"
-	key = "8fZBjCUfN0TzjEGLQldGY4+iE9AkOvCfjh7+p0GtRBQ="
+	key = "X+IJMVT6OnsrIR6G/9OTcJSX+lM9FSPN"
 	key_id = "worker-auth-storage"
 }
 ```
@@ -78,99 +77,3 @@ kms "aead" {
 workers. These fields are only valid for [KMS Workers][]. `name` and
 `description` can only be set for PKI workers through the API.

-## Multi-hop sessions <sup>HCP only</sup>
-
-You can use the `hcp_boundary_cluster_id` field to configure PKI workers to connect to your HCP
-Boundary cluster or the `initial_upstreams` field to connect directly to other workers,
- when set to that worker's `public_addr` field.
-
-A multi-hop session is when two or more workers are connected, and there are multiple “hops” from a worker to
-the controller. There are no limits on the amount of workers allowed in a multi-hop session configuration.
-
-The multi-hop session introduces the concepts of “upstream” and “downstream” workers. If you view controllers
-as the “top” of a multi-hop chain, downstream workers reside below a worker in the chain, while upstream workers reside
-above a worker in the chain. For example, in the diagram below, Worker 2’s upstream is Worker 1, and its
-downstream is Worker 3.
-
-![multi-hop workers](/img/multi-hop-workers.png)
-
-You can deploy multi-hop workers in scenarios where inbound network traffic is not allowed. A worker in a private
-network can send outbound communication to its upstream worker, and create a reverse proxy to establish a session.
-
-You can configure [target worker filters][] with multi-hop workers to allow for fine-grained control on which workers handle
-ingress and egress for session traffic to a [target][]. Ingress worker filters determine which
-workers you connect with to initiate a session, and egress worker filters determine which workers are
-used to access targets.
-
-### Multi-hop worker requirements
-
-When you configure multi-hop sessions, there is an "ingress" worker, an "egress" worker, and any number of intermediary workers.
-Ingress, egress, and intermediary workers have the following requirements.
-
-#### Ingress worker requirements
-
-To proxy target connections, ingress workers require outbound access to the Boundary control plane and inbound access from clients.
-
-#### Intermediary worker requirements
-
-Intermediary workers require outbound access to an upstream worker.
-The upstream worker may be an ingress worker or another intermediary worker.
-Any upstream intermediary worker needs eventual connectivity to an ingress worker via trusted intermediaries.
-
-Intermediary workers also require inbound access from a downstream worker.
-The downstream worker may be an egress worker or another intermediary worker.
-Any downstream intermediary worker needs eventual connectivity to an egress worker via trusted intermediaries.
-
-#### Egress worker requirements
-
-To proxy target connections, egress workers require outbound access to an upstream worker and outbound access to the destination host.
-Inbound session connections from clients reach the egress worker via reverse proxy; the ingress worker is initiated from the egress worker.
-
-## Complete configuration example
-
-```hcl
-listener "tcp" {
-	purpose = "proxy"
-	tls_disable = true
-	address = "127.0.0.1"
-}
-
-worker {
-  # Path for worker storage. Must be unique across workers
-  auth_storage_path="/boundary/demo-worker-1"
-
-  # Workers typically need to reach upstreams on :9201
-  initial_upstreams = [
-    "10.0.0.1",
-    "10.0.0.2",
-    "10.0.0.3",
-  ]
-
-  public_addr = "myhost.mycompany.com"
-
-  tags {
-    type   = ["prod", "webservers"]
-    region = ["us-east-1"]
-  }
-}
-
-# The following KMS config is an example only
-# Use a production KMS such as AWS KMS for production installs
-kms "aead" {
-  purpose = "worker-auth-storage"
-	aead_type = "aes-gcm"
-	key = "8fZBjCUfN0TzjEGLQldGY4+iE9AkOvCfjh7+p0GtRBQ="
-	key_id = "worker-auth-storage"
-}
-
-
-```
-[kms workers]: /boundary/docs/configuration/worker/kms-worker
-[target]: /boundary/docs/concepts/domain-model/targets
-[target worker filters]: /boundary/docs/concepts/filtering/worker-tags#target-worker-filtering
-
-## Tutorials
-
-Refer to the [Self-Managed Worker Registration with HCP Boundary](/boundary/tutorials/hcp-administration/hcp-manage-workers) tutorial to learn how to register and manage PKI workers.
-
-Refer to the [Manage Multi-Hop Sessions with HCP Boundary](/boundary/tutorials/hcp-administration/hcp-manage-multi-hop) tutorial to learn how to configure a multi-hop session.