Remove the deprecated old KMS worker auth mechanism (#3935)

This has been superceded by the nodeenrollment-based KMS worker auth
mechanism. This commit only removes the ability to connect with this
mechanism; some upgrade logic will remain intact for a bit and
can be removed later.

globals/globals.go

@@ -12,6 +12,8 @@ const (
 	AnyAuthenticatedUserId = "u_auth"
 	AnonymousUserId        = "u_anon"
 	RecoveryUserId         = "u_recovery"
+
+	MinimumSupportedPostgresVersion = "12"
 )
 
 type (

go.mod

@@ -6,8 +6,6 @@ replace github.com/hashicorp/boundary/api => ./api
 
 replace github.com/hashicorp/boundary/sdk => ./sdk
 
-//replace github.com/hashicorp/nodeenrollment => ../nodeenrollment
-
 require (
 	github.com/fatih/color v1.15.0
 	github.com/fatih/structs v1.1.0

internal/cmd/base/internal/docker/supported.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/db/common"
 	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/ory/dockertest/v3"
@@ -113,7 +114,7 @@ func splitImage(opts Options) (string, string, error) {
 	switch separatedlen {
 	case 1:
 		if separated[0] == "postgres" {
-			return separated[0], "11", nil
+			return separated[0], globals.MinimumSupportedPostgresVersion, nil
 		}
 		return "", "", fmt.Errorf("valid reference format is repo:tag, if"+
 			" no tag provided then repo must be postgres, got: %s", opts.withContainerImage)

internal/cmd/config/config.go

@@ -299,9 +299,9 @@ type Worker struct {
 	// 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 indicates that the worker should use the
+	// pre-0.13 method of using KMSes to authenticate. This is currently only
+	// supported to throw an error if used telling people they need to upgrade.
 	UseDeprecatedKmsAuthMethod bool `hcl:"use_deprecated_kms_auth_method"`
 }
 
@@ -732,6 +732,10 @@ func Parse(d string) (*Config, error) {
 
 	// Parse worker tags
 	if result.Worker != nil {
+		if result.Worker.UseDeprecatedKmsAuthMethod {
+			return nil, fmt.Errorf("The flag 'use_deprecated_auth_method' is unsupported as of version 0.15.")
+		}
+
 		result.Worker.Name, err = parseutil.ParsePath(result.Worker.Name)
 		if err != nil && !errors.Is(err, parseutil.ErrNotAUrl) {
 			return nil, fmt.Errorf("Error parsing worker name: %w", err)

internal/credential/vault/supported.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/db/common"
 	"github.com/hashicorp/go-rootcerts"
 	vault "github.com/hashicorp/vault/api"
@@ -26,10 +27,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const DefaultVaultVersion = "1.7.2"
+const DefaultVaultVersion = "1.15.1"
 
 var (
-	vaultRepository    = "vault"
+	vaultRepository    = "hashicorp/vault"
 	postgresRepository = "postgres"
 )
 
@@ -226,7 +227,7 @@ func gotMountDatabase(t testing.TB, v *TestVaultServer, opt ...TestOption) *Test
 
 	dockerOptions := &dockertest.RunOptions{
 		Repository: postgresRepository,
-		Tag:        "11",
+		Tag:        globals.MinimumSupportedPostgresVersion,
 		Networks:   []*dockertest.Network{network},
 		Env:        []string{"POSTGRES_PASSWORD=password", "POSTGRES_DB=boundarytest"},
 	}
@@ -332,7 +333,7 @@ func gotMountDatabase(t testing.TB, v *TestVaultServer, opt ...TestOption) *Test
 	}
 	s, err := vc.Logical().Write(postgresConfPath, postgresConfOptions)
 	require.NoError(err)
-	require.NotEmpty(s)
+	require.Empty(s)
 
 	const (
 		vaultOpenedCreationStatement = `

internal/daemon/controller/controller.go

@@ -158,7 +158,7 @@ type Controller struct {
 	// replying to queries with "503 Service Unavailable".
 	HealthService *health.Service
 
-	pkiConnManager *cluster.DownstreamManager
+	downstreamConnManager *cluster.DownstreamManager
 
 	// ControllerExtension defines a std way to extend the controller
 	ControllerExtension intglobals.ControllerExtension
@@ -178,7 +178,7 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 		workerStatusUpdateTimes: new(sync.Map),
 		enabledPlugins:          conf.Server.EnabledPlugins,
 		apiListeners:            make([]*base.ServerListener, 0),
-		pkiConnManager:          cluster.NewDownstreamManager(),
+		downstreamConnManager:   cluster.NewDownstreamManager(),
 		workerStatusGracePeriod: new(atomic.Int64),
 		livenessTimeToStale:     new(atomic.Int64),
 	}
@@ -523,7 +523,7 @@ func (c *Controller) Start() error {
 		defer c.tickerWg.Done()
 		c.startCloseExpiredPendingTokens(c.baseContext)
 	}()
-	if err := c.startWorkerConnectionMaintenanceTicking(c.baseContext, c.tickerWg, c.pkiConnManager); err != nil {
+	if err := c.startWorkerConnectionMaintenanceTicking(c.baseContext, c.tickerWg, c.downstreamConnManager); err != nil {
 		return errors.Wrap(c.baseContext, err, op)
 	}
 

internal/daemon/controller/kms_conn_router.go

@@ -1,156 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: BUSL-1.1
-
-package controller
-
-import (
-	"context"
-	"crypto/tls"
-	"fmt"
-	"net"
-	"strings"
-
-	"github.com/hashicorp/boundary/internal/daemon/common"
-	"github.com/hashicorp/boundary/internal/errors"
-	"github.com/hashicorp/boundary/internal/event"
-	nodeenet "github.com/hashicorp/nodeenrollment/net"
-)
-
-// tempError is an error that satisfies the temporary error interface that is
-// internally used by gRPC to determine whether an error should cause a listener
-// to die. Any error that isn't an accept error is wrapped in this since one
-// connection failing TLS wise doesn't mean we don't want to accept any more...
-type tempError struct {
-	error
-}
-
-// newTempError is a "temporary" error
-func newTempError(inner error) tempError {
-	return tempError{error: inner}
-}
-
-func (t tempError) Temporary() bool {
-	return true
-}
-
-// startKmsConnRouter allows us to validate the nonce from a connection before
-// handing it off to other purposes, such as the gRPC server. It is expected
-// that the first thing a connection sends after successful TLS validation is
-// the nonce that was contained in the KMS-encrypted TLS info. The reason is for
-// replay attacks -- since the ALPN NextProtos are sent in the clear, we don't
-// want anyone sniffing the connection to simply replay the hello message and
-// gain access. By requiring the information encrypted within that message to
-// match the first bytes sent in the connection itself, we require that the
-// service making the incoming connection had to either be the service that did
-// the initial encryption, or had to be able to decrypt that against the same
-// KMS key. This means that KMS access is a requirement, and simple replay
-// itself is not sufficient.
-func startKmsConnRouter(
-	ctx context.Context,
-	c *Controller,
-	baseLn net.Listener,
-	authedListener,
-	reverseGrpcListener *nodeenet.MultiplexingListener,
-) error {
-	const op = "controller.startKmsAuthRouter"
-	switch {
-	case c == nil:
-		return errors.New(ctx, errors.InvalidParameter, op, "nil Controller")
-	case baseLn == nil:
-		return errors.New(ctx, errors.InvalidParameter, op, "nil baseLn")
-	case authedListener == nil:
-		return errors.New(ctx, errors.InvalidParameter, op, "nil authedListener")
-	case reverseGrpcListener == nil:
-		return errors.New(ctx, errors.InvalidParameter, op, "nil reverseGrpcListener")
-	}
-
-	go func() {
-		for {
-			conn, err := baseLn.Accept()
-			if err != nil {
-				if err == net.ErrClosed {
-					// We're done
-					return
-				}
-				// Conn may already be closed but can't be sure, so just try anyways
-				if conn != nil {
-					if err := conn.Close(); err != nil {
-						event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-					}
-				}
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error accepting kms connection"))
-				continue
-			}
-
-			if conn == nil {
-				// No idea why this would happen, but be safe before we operate
-				// on it below
-				event.WriteError(ctx, op, fmt.Errorf("connection is nil"))
-				continue
-			}
-
-			tlsConn, ok := conn.(*tls.Conn)
-			if !ok {
-				// This should never happen, but is just to be safe
-				event.WriteError(ctx, op, fmt.Errorf("connection is not a *tls.Conn"))
-				_ = conn.Close()
-				continue
-			}
-
-			if !strings.HasPrefix(tlsConn.ConnectionState().NegotiatedProtocol, "v1workerauth") {
-				// If we're here it hasn't been handled by PKI and can't be
-				// handled here. We should never actually get here...
-				event.WriteError(ctx, op, fmt.Errorf("connection is not a kms connection"))
-				_ = conn.Close()
-				continue
-			}
-
-			nonce := make([]byte, 20)
-			read, err := conn.Read(nonce)
-			if err != nil {
-				if err := conn.Close(); err != nil {
-					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-				}
-				event.WriteError(ctx, op, err, event.WithInfoMsg("error reading nonce from connection"))
-				continue
-			}
-			if read != len(nonce) {
-				if err := conn.Close(); err != nil {
-					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-				}
-				event.WriteError(ctx, op, fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read))
-				continue
-			}
-
-			workerInfoRaw, found := c.workerAuthCache.Load(string(nonce))
-			if !found {
-				if err := conn.Close(); err != nil {
-					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
-				}
-				event.WriteError(ctx, op, fmt.Errorf("did not find valid nonce for incoming worker"))
-				continue
-			}
-
-			workerInfo := workerInfoRaw.(*workerAuthEntry)
-			workerInfo.conn = tlsConn
-			c.workerAuthCache.Delete(string(nonce))
-			event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name, "description", workerInfo.Description, "proxy_address", workerInfo.ProxyAddress)
-
-			found = false
-			for _, proto := range workerInfo.clientNextProtos {
-				if proto == common.ReverseGrpcConnectionAlpnValue {
-					reverseGrpcListener.IngressConn(tlsConn, nil)
-					found = true
-					break
-				}
-			}
-			if found {
-				continue
-			}
-			// Didn't find specific ALPN values, so it's a normal grpc
-			// connection
-			authedListener.IngressConn(tlsConn, nil)
-		}
-	}()
-	return nil
-}

internal/daemon/controller/listeners.go

@@ -5,7 +5,6 @@ package controller
 
 import (
 	"context"
-	"crypto/tls"
 	stderrors "errors"
 	"fmt"
 	"math"
@@ -159,9 +158,6 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 			Context:      c.baseContext,
 			Storage:      workerAuthStorage,
 			BaseListener: ln.ClusterListener,
-			BaseTlsConfiguration: &tls.Config{
-				GetConfigForClient: c.validateWorkerTls,
-			},
 			Options: []nodee.Option{
 				nodee.WithLogger(eventLogger),
 				nodee.WithRegistrationWrapper(wrapperToUse),
@@ -185,26 +181,11 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		return nil, fmt.Errorf("error instantiating node enrollment authed split listener: %w", err)
 	}
 
-	// This wraps the normal pki worker connections with a listener which adds
-	// the worker key id of the connections to the controller's pkiConnManager.
-	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, nodeeAuthedListener, c.pkiConnManager, sourcePurpose(grpcListenerPurpose))
+	// This wraps connections with a listener which adds the worker key id of
+	// the connections to the controller's nodeeConnManager.
+	workerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, nodeeAuthedListener, c.downstreamConnManager, sourcePurpose(grpcListenerPurpose))
 	if err != nil {
-		return nil, fmt.Errorf("%s: error creating pki worker tracking listener: %w", op, err)
-	}
-
-	// Create a multiplexer to unify connections between PKI and KMS
-	multiplexingAuthedListener, err := nodeenet.NewMultiplexingListener(c.baseContext, nodeeAuthedListener.Addr())
-	if err != nil {
-		return nil, fmt.Errorf("error instantiating authed multiplexed listener: %w", err)
-	}
-	if err := multiplexingAuthedListener.IngressListener(pkiWorkerTrackingListener); err != nil {
-		return nil, fmt.Errorf("error adding authed evented listener to multiplexed listener: %w", err)
-	}
-	// Connections that came in on the cluster listener that are not authed via
-	// PKI need to be sent for KMS routing
-	nodeeUnauthedListener, err := splitListener.GetListener(nodeenet.UnauthenticatedNextProto)
-	if err != nil {
-		return nil, fmt.Errorf("error instantiating node enrollment unauthed split listener: %w", err)
+		return nil, fmt.Errorf("%s: error creating node enrollment worker tracking listener: %w", op, err)
 	}
 
 	// Connections coming in here are authed by nodeenrollment and are for the
@@ -214,25 +195,11 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		return nil, fmt.Errorf("error instantiating reverse gprc connection split listener: %w", err)
 	}
 
-	// This wraps the reverse grpc pki worker connections with a listener which
-	// adds the worker key id of the connections to the pkiConnManager.
-	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, reverseGrpcListener, c.pkiConnManager, sourcePurpose(reverseGrpcListenerPurpose))
+	// This wraps the reverse grpc connections with a listener which adds the
+	// worker key id of the connections to the nodeeConnManager.
+	revWorkerTrackingListener, err := cluster.NewTrackingListener(c.baseContext, reverseGrpcListener, c.downstreamConnManager, sourcePurpose(reverseGrpcListenerPurpose))
 	if err != nil {
-		return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err)
-	}
-
-	// Create a multiplexer to unify reverse grpc connections between PKI and KMS
-	multiplexingReverseGrpcListener, err := nodeenet.NewMultiplexingListener(c.baseContext, reverseGrpcListener.Addr())
-	if err != nil {
-		return nil, fmt.Errorf("error instantiating reverse grpc multiplexed listener: %w", err)
-	}
-	if err := multiplexingReverseGrpcListener.IngressListener(revPkiWorkerTrackingListener); err != nil {
-		return nil, fmt.Errorf("error adding reverse grpc listener to multiplexed listener: %w", err)
-	}
-
-	// Start routing KMS connections
-	if err := startKmsConnRouter(c.baseContext, c, nodeeUnauthedListener, multiplexingAuthedListener, multiplexingReverseGrpcListener); err != nil {
-		return nil, errors.Wrap(c.baseContext, err, op)
+		return nil, fmt.Errorf("%s: error creating reverse grpc worker tracking listener: %w", op, err)
 	}
 
 	workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer)
@@ -269,7 +236,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	ln.GrpcServer = workerServer
 
 	return func() {
-		err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(multiplexingReverseGrpcListener, reverseGrpcListenerPurpose),
+		err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(revWorkerTrackingListener, reverseGrpcListenerPurpose),
 			c.downstreamConns)
 		if err != nil {
 			event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
@@ -281,9 +248,9 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 			}
 		}()
 		go func() {
-			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, grpcListenerPurpose))
+			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(workerTrackingListener, grpcListenerPurpose))
 			if err != nil && !errors.Is(err, net.ErrClosed) {
-				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("multiplexingAuthedListener error"))
+				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("workerTrackingListener error"))
 			}
 		}()
 	}, nil

internal/daemon/controller/listeners_test.go

@@ -9,28 +9,36 @@ import (
 	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/base64"
-	"encoding/json"
 	"encoding/pem"
 	"fmt"
-	"io/ioutil"
+	"io"
+	"log"
 	"math/big"
 	"net"
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
+	"github.com/hashicorp/boundary/internal/daemon/cluster"
+	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/version"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/configutil/v2"
 	"github.com/hashicorp/go-secure-stdlib/listenerutil"
+	"github.com/hashicorp/nodeenrollment"
+	"github.com/hashicorp/nodeenrollment/protocol"
+	"github.com/hashicorp/nodeenrollment/storage/inmem"
+	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
-	"google.golang.org/protobuf/proto"
+	"google.golang.org/grpc/backoff"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func TestStartListeners(t *testing.T) {
@@ -240,17 +248,24 @@ func TestStartListeners(t *testing.T) {
 
 			ctx, cancel := context.WithCancel(context.Background())
 
+			debugEnabled := new(atomic.Bool)
+			debugEnabled.Store(false)
+			opts := &TestControllerOpts{
+				EnableEventing:             false,
+				WorkerAuthDebuggingEnabled: debugEnabled,
+			}
+
 			tc := &TestController{
 				t:              t,
 				ctx:            ctx,
 				cancel:         cancel,
+				opts:           opts,
 				shutdownDoneCh: make(chan struct{}),
 				shutdownOnce:   new(sync.Once),
-				opts:           new(TestControllerOpts),
 			}
 			t.Cleanup(func() { tc.Shutdown() })
 
-			conf := TestControllerConfig(t, ctx, tc, nil)
+			conf := TestControllerConfig(t, ctx, tc, opts)
 			conf.RawConfig.SharedConfig = &configutil.SharedConfig{Listeners: tt.listeners, DisableMlock: true}
 
 			err := conf.SetupListeners(nil, conf.RawConfig.SharedConfig, []string{"api", "cluster"})
@@ -718,11 +733,21 @@ func TestListenerCloseErrorCheck(t *testing.T) {
 }
 
 func clusterGrpcDialNoError(t *testing.T, c *Controller, network, addr string) {
-	grpcConn, err := grpc.Dial(addr,
-		grpc.WithInsecure(),
+	grpcConn, err := grpc.DialContext(
+		context.Background(),
+		addr,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithBlock(),
-		grpc.WithTimeout(5*time.Second),
 		grpc.WithContextDialer(clusterTestDialer(t, c, network)),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithConnectParams(grpc.ConnectParams{
+			Backoff: backoff.Config{
+				BaseDelay:  time.Second,
+				Multiplier: 1.2,
+				Jitter:     0.2,
+				MaxDelay:   3 * time.Second,
+			},
+		}),
 	)
 	require.NoError(t, err)
 	require.NoError(t, grpcConn.Close())
@@ -757,7 +782,7 @@ func testTlsHttpClient(t *testing.T, filePath string) *http.Client {
 	f, err := os.Open(filePath)
 	require.NoError(t, err)
 
-	certBytes, err := ioutil.ReadAll(f)
+	certBytes, err := io.ReadAll(f)
 	require.NoError(t, err)
 	require.NoError(t, f.Close())
 
@@ -795,68 +820,39 @@ func createTestCert(t *testing.T) ([]byte, ed25519.PublicKey, ed25519.PrivateKey
 
 func clusterTestDialer(t *testing.T, c *Controller, network string) func(context.Context, string) (net.Conn, error) {
 	return func(ctx context.Context, addr string) (net.Conn, error) {
-		certBytes, _, priv := createTestCert(t)
-
-		marshaledKey, err := x509.MarshalPKCS8PrivateKey(priv)
+		storage, err := inmem.New(ctx)
 		require.NoError(t, err)
 
-		nonce, err := base62.Random(20)
+		_, err = types.NewNodeCredentials(
+			ctx,
+			storage,
+		)
 		require.NoError(t, err)
 
-		// after 0.13.0 we need to include the version that matches between the
-		// controller and the worker otherwise worker tls auth with fail
-		v := version.Get().Version
-		if version.Get().VersionMetadata != "" {
-			v += "+" + version.Get().VersionMetadata
+		name, err := base62.Random(20)
+		require.NoError(t, err)
+
+		wci := &cluster.WorkerConnectionInfo{
+			BoundaryVersion:  version.Get().VersionNumber(),
+			Name:             strings.ToLower(name),
+			OperationalState: server.ActiveOperationalState.String(),
 		}
-		info := base.WorkerAuthInfo{
-			CertPEM:         pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certBytes}),
-			KeyPEM:          pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: marshaledKey}),
-			ConnectionNonce: nonce,
-			BoundaryVersion: v,
+		st, err := wci.AsConnectionStateStruct()
+		require.NoError(t, err)
+
+		conn, err := protocol.Dial(
+			ctx,
+			storage,
+			addr,
+			nodeenrollment.WithState(st),
+			nodeenrollment.WithRegistrationWrapper(c.conf.WorkerAuthKms),
+			nodeenrollment.WithLogger(c.conf.Logger),
+			nodeenrollment.WithWrappingRegistrationFlowApplicationSpecificParams(st),
+		)
+		if err != nil {
+			log.Println(err)
 		}
-		infoBytes, err := json.Marshal(info)
 		require.NoError(t, err)
-
-		encInfo, err := c.conf.WorkerAuthKms.Encrypt(ctx, infoBytes, nil)
-		require.NoError(t, err)
-
-		protoEncInfo, err := proto.Marshal(encInfo)
-		require.NoError(t, err)
-
-		b64alpn := base64.RawStdEncoding.EncodeToString(protoEncInfo)
-
-		var nextProtos []string
-		var count int
-		for i := 0; i < len(b64alpn); i += 230 {
-			end := i + 230
-			if end > len(b64alpn) {
-				end = len(b64alpn)
-			}
-			nextProtos = append(nextProtos, fmt.Sprintf("v1workerauth-%02d-%s", count, b64alpn[i:end]))
-			count++
-		}
-
-		cert, err := x509.ParseCertificate(certBytes)
-		require.NoError(t, err)
-
-		rootCAs := x509.NewCertPool()
-		rootCAs.AddCert(cert)
-
-		tlsCert, err := tls.X509KeyPair(info.CertPEM, info.KeyPEM)
-		require.NoError(t, err)
-
-		conn, err := tls.Dial(network, addr, &tls.Config{
-			Certificates: []tls.Certificate{tlsCert},
-			RootCAs:      rootCAs,
-			NextProtos:   nextProtos,
-			MinVersion:   tls.VersionTLS13,
-		})
-		require.NoError(t, err)
-
-		_, err = conn.Write([]byte(nonce))
-		require.NoError(t, err)
-
 		return conn, nil
 	}
 }

internal/daemon/controller/worker_tls_config.go

@@ -1,145 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: BUSL-1.1
-
-package controller
-
-import (
-	"context"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/base64"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"net"
-	"strings"
-
-	"github.com/hashicorp/boundary/internal/cmd/base"
-	"github.com/hashicorp/boundary/internal/event"
-	"github.com/hashicorp/boundary/internal/server"
-	"github.com/hashicorp/boundary/version"
-	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
-	"google.golang.org/protobuf/proto"
-)
-
-type workerAuthEntry struct {
-	*base.WorkerAuthInfo
-	conn             net.Conn
-	clientNextProtos []string
-}
-
-// validateWorkerTls is called by the Go TLS stack with client info. It calls
-// v1WorkerAuthConfig to validate the encryption and ensure the nonce has not
-// been replayed, then stores connection information into the auth cache, which
-// is used by the intercepting listener to log information.
-func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config, error) {
-	const op = "controller.(Controller).validateWorkerTls"
-	for _, p := range hello.SupportedProtos {
-		switch {
-		case strings.HasPrefix(p, "v1workerauth-"):
-			tlsConf, workerInfo, err := c.v1WorkerAuthConfig(hello.SupportedProtos)
-			if err == nil {
-				// Store info that can be retrieved in the intercepting
-				// listener.
-				//
-				// NOTE: This is not a load-or-store because we need to ensure
-				// replays can't happen across controllers, hence storing in the
-				// database in v1WorkerAuthConfig. Although this could serve as
-				// a secondary check, a load-or-store here keeps the test from
-				// ensuring that it is rejected at the database level via the
-				// nonce unique constraint, which is by far the more important
-				// thing to validate (although there are DB-level tests for that
-				// too).
-				authEntry := &workerAuthEntry{
-					WorkerAuthInfo: workerInfo,
-				}
-				if len(hello.SupportedProtos) > 0 {
-					authEntry.clientNextProtos = make([]string, len(hello.SupportedProtos))
-					copy(authEntry.clientNextProtos, hello.SupportedProtos)
-				}
-				c.workerAuthCache.Store(workerInfo.ConnectionNonce, authEntry)
-			}
-			return tlsConf, err
-		}
-	}
-	return nil, errors.New("no supported protos found in hello")
-}
-
-// v1WorkerAuthConfig:
-//
-// * Reads the information from ALPN, combining across protos
-// * Combines them to form the encrypted auth information
-// * Validates it by decrypting against the configured KMS
-// * Ensures that the nonce is unique to prevent replay attacks
-// * Returns the shared TLS configuration that is used to establish the connection
-func (c Controller) v1WorkerAuthConfig(protos []string) (*tls.Config, *base.WorkerAuthInfo, error) {
-	const op = "controller.(Controller).v1WorkerAuthConfig"
-	var firstMatchProto string
-	var encString string
-	for _, p := range protos {
-		if strings.HasPrefix(p, "v1workerauth-") {
-			// Strip that and the number
-			encString += strings.TrimPrefix(p, "v1workerauth-")[3:]
-			if firstMatchProto == "" {
-				firstMatchProto = p
-			}
-		}
-	}
-	if firstMatchProto == "" {
-		return nil, nil, errors.New("no matching proto found")
-	}
-	marshaledEncInfo, err := base64.RawStdEncoding.DecodeString(encString)
-	if err != nil {
-		return nil, nil, err
-	}
-	encInfo := new(wrapping.BlobInfo)
-	if err := proto.Unmarshal(marshaledEncInfo, encInfo); err != nil {
-		return nil, nil, err
-	}
-	if isNil(c.conf.WorkerAuthKms) {
-		return nil, nil, errors.New("worker auth kms not set")
-	}
-	marshaledInfo, err := c.conf.WorkerAuthKms.Decrypt(context.Background(), encInfo)
-	if err != nil {
-		return nil, nil, err
-	}
-	info := new(base.WorkerAuthInfo)
-	if err := json.Unmarshal(marshaledInfo, info); err != nil {
-		return nil, nil, err
-	}
-	if version.Get().VersionMetadata != "" {
-		incomingVer := version.FromVersionString(info.BoundaryVersion)
-		if (incomingVer == nil || incomingVer.VersionMetadata == "") && version.SupportsFeature(version.Binary, version.RequireVersionInWorkerInfo) {
-			err := errors.New("build contains version with metadata, incoming worker version is nil or has mismatched metadata")
-			event.WriteError(c.baseContext, op, err)
-			return nil, nil, err
-		}
-	}
-
-	// Check for replays
-	serversRepo, err := c.ServersRepoFn()
-	if err != nil {
-		return nil, nil, fmt.Errorf("unable to fetch server repo: %w", err)
-	}
-	if err := serversRepo.AddNonce(c.baseContext, info.ConnectionNonce, server.NoncePurposeWorkerAuth); err != nil {
-		return nil, nil, fmt.Errorf("unable to add connection nonce to database: %w", err)
-	}
-
-	rootCAs := x509.NewCertPool()
-	if ok := rootCAs.AppendCertsFromPEM(info.CertPEM); !ok {
-		return nil, info, errors.New("unable to add ca cert to cert pool")
-	}
-	tlsCert, err := tls.X509KeyPair(info.CertPEM, info.KeyPEM)
-	if err != nil {
-		return nil, info, err
-	}
-	tlsConfig := &tls.Config{
-		Certificates: []tls.Certificate{tlsCert},
-		ClientCAs:    rootCAs,
-		ClientAuth:   tls.RequireAndVerifyClientCert,
-		NextProtos:   []string{firstMatchProto},
-		MinVersion:   tls.VersionTLS13,
-	}
-
-	return tlsConfig, info, nil
-}

internal/daemon/worker/auth_rotation.go

@@ -25,11 +25,6 @@ const defaultAuthRotationResetDuration = 5 * time.Second
 func (w *Worker) startAuthRotationTicking(cancelCtx context.Context) {
 	const op = "worker.(Worker).startAuthRotationTicking"
 
-	if w.conf.RawConfig.Worker.UseDeprecatedKmsAuthMethod {
-		event.WriteSysEvent(cancelCtx, op, "using deprecated kms worker authentication method; pki auth rotation ticking not running")
-		return
-	}
-
 	event.WriteSysEvent(cancelCtx, op, "starting auth rotation ticking")
 
 	// This will start at 0 which will cause us to run immediately

internal/daemon/worker/controller_connection.go

@@ -5,16 +5,8 @@ package worker
 
 import (
 	"context"
-	"crypto/ed25519"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/base64"
-	"encoding/json"
-	"encoding/pem"
 	"fmt"
 	"math"
-	"math/big"
-	mathrand "math/rand"
 	"net"
 	"os"
 	"path/filepath"
@@ -33,7 +25,6 @@ import (
 	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/internal/util"
 	"github.com/hashicorp/boundary/version"
-	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/parseutil"
 	"github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/multihop"
@@ -44,7 +35,6 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/resolver"
-	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/structpb"
 )
 
@@ -96,10 +86,8 @@ func (w *Worker) StartControllerConnections() error {
 
 // upstreamDialerFunc dials an upstream server. extraAlpnProtos can be provided
 // to kms and pki connections and are used for identifying, on the server side,
-// the intended purpose of the connection.  upstreamDialerFunc takes an optional
-// stateProvidingFunction which, if the connection is made using the nodeenrollment
-// library, sends the state provided by it to the server and can be retrieved
-// from the resulting *protocol.Conn.
+// the intended purpose of the connection. The state provided by the worker to
+// the server can be retrieved from the resulting *protocol.Conn.
 func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) {
 	const op = "worker.(Worker).upstreamDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
@@ -119,32 +107,27 @@ func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Cont
 			event.WriteError(w.baseContext, op, err)
 			return nil, errors.Wrap(w.baseContext, err, op)
 		}
-		switch {
-		case w.conf.RawConfig.Worker.UseDeprecatedKmsAuthMethod:
-			conn, err = w.v1KmsAuthDialFn(ctx, addr, extraAlpnProtos...)
-		default:
-			conn, err = protocol.Dial(
-				ctx,
-				w.WorkerAuthStorage,
-				addr,
-				nodeenrollment.WithLogger(eventLogger),
-				nodeenrollment.WithState(st),
-				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
-				nodeenrollment.WithActivationToken(w.conf.RawConfig.Worker.ControllerGeneratedActivationToken),
-			)
-			// 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.conf.RawConfig.Worker.AuthStoragePath != "" {
-					workerAuthReqFilePath := filepath.Join(w.conf.RawConfig.Worker.AuthStoragePath, base.WorkerAuthReqFile)
-					// Intentionally ignoring any error removing this file
-					_ = os.Remove(workerAuthReqFilePath)
-				}
+		conn, err = protocol.Dial(
+			ctx,
+			w.WorkerAuthStorage,
+			addr,
+			nodeenrollment.WithLogger(eventLogger),
+			nodeenrollment.WithState(st),
+			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
+			nodeenrollment.WithActivationToken(w.conf.RawConfig.Worker.ControllerGeneratedActivationToken),
+		)
+		// 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.conf.RawConfig.Worker.AuthStoragePath != "" {
+				workerAuthReqFilePath := filepath.Join(w.conf.RawConfig.Worker.AuthStoragePath, base.WorkerAuthReqFile)
+				// Intentionally ignoring any error removing this file
+				_ = os.Remove(workerAuthReqFilePath)
 			}
 		}
 		switch {
@@ -183,40 +166,6 @@ func (w *Worker) upstreamDialerFunc(extraAlpnProtos ...string) func(context.Cont
 	}
 }
 
-func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string, extraAlpnProtos ...string) (net.Conn, error) {
-	const op = "worker.(Worker).v1KmsAuthDialFn"
-	tlsConf, authInfo, err := w.workerAuthTLSConfig(extraAlpnProtos...)
-	if err != nil {
-		return nil, fmt.Errorf("error creating tls config for worker auth: %w", err)
-	}
-	dialer := &net.Dialer{}
-	var nonTlsConn net.Conn
-	switch {
-	case strings.HasPrefix(addr, "/"):
-		nonTlsConn, err = dialer.DialContext(ctx, "unix", addr)
-	default:
-		nonTlsConn, err = dialer.DialContext(ctx, "tcp", addr)
-	}
-	if err != nil {
-		return nil, fmt.Errorf("unable to dial to upstream: %w", err)
-	}
-	tlsConn := tls.Client(nonTlsConn, tlsConf)
-	written, err := tlsConn.Write([]byte(authInfo.ConnectionNonce))
-	if err != nil {
-		if err := nonTlsConn.Close(); err != nil {
-			event.WriteError(w.baseContext, op, err, event.WithInfoMsg("error closing connection after writing failure"))
-		}
-		return nil, fmt.Errorf("unable to write connection nonce: %w", err)
-	}
-	if written != len(authInfo.ConnectionNonce) {
-		if err := nonTlsConn.Close(); err != nil {
-			event.WriteError(w.baseContext, op, err, event.WithInfoMsg("error closing connection after writing failure"))
-		}
-		return nil, fmt.Errorf("expected to write %d bytes of connection nonce, wrote %d", len(authInfo.ConnectionNonce), written)
-	}
-	return tlsConn, nil
-}
-
 func (w *Worker) createClientConn(addr string) error {
 	const op = "worker.(Worker).createClientConn"
 
@@ -295,112 +244,6 @@ func createDefaultGRPCDialOptions(res resolver.Builder, upstreamDialerFn func(co
 	return dialOpts
 }
 
-func (w *Worker) workerAuthTLSConfig(extraAlpnProtos ...string) (*tls.Config, *base.WorkerAuthInfo, error) {
-	var err error
-	info := &base.WorkerAuthInfo{
-		Name:            w.conf.RawConfig.Worker.Name,
-		Description:     w.conf.RawConfig.Worker.Description,
-		ProxyAddress:    w.conf.RawConfig.Worker.PublicAddr,
-		BoundaryVersion: version.Get().VersionNumber(),
-	}
-
-	info.ConnectionNonce, err = w.nonceFn(20)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	pubKey, privKey, err := ed25519.GenerateKey(w.conf.SecureRandomReader)
-	if err != nil {
-		return nil, nil, err
-	}
-	host, err := base62.Random(20)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	template := &x509.Certificate{
-		ExtKeyUsage: []x509.ExtKeyUsage{
-			x509.ExtKeyUsageServerAuth,
-			x509.ExtKeyUsageClientAuth,
-		},
-		DNSNames:              []string{host},
-		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageKeyAgreement | x509.KeyUsageCertSign,
-		SerialNumber:          big.NewInt(mathrand.Int63()),
-		NotBefore:             time.Now().Add(-30 * time.Second),
-		NotAfter:              time.Now().Add(globals.WorkerAuthNonceValidityPeriod),
-		BasicConstraintsValid: true,
-		IsCA:                  true,
-	}
-	certBytes, err := x509.CreateCertificate(w.conf.SecureRandomReader, template, template, pubKey, privKey)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	certPEMBlock := &pem.Block{
-		Type:  "CERTIFICATE",
-		Bytes: certBytes,
-	}
-	info.CertPEM = pem.EncodeToMemory(certPEMBlock)
-
-	marshaledKey, err := x509.MarshalPKCS8PrivateKey(privKey)
-	if err != nil {
-		return nil, nil, err
-	}
-	keyPEMBlock := &pem.Block{
-		Type:  "PRIVATE KEY",
-		Bytes: marshaledKey,
-	}
-	info.KeyPEM = pem.EncodeToMemory(keyPEMBlock)
-
-	// Marshal and encrypt
-	marshaledInfo, err := json.Marshal(info)
-	if err != nil {
-		return nil, nil, err
-	}
-	encInfo, err := w.conf.WorkerAuthKms.Encrypt(w.baseContext, marshaledInfo)
-	if err != nil {
-		return nil, nil, err
-	}
-	marshaledEncInfo, err := proto.Marshal(encInfo)
-	if err != nil {
-		return nil, nil, err
-	}
-	b64alpn := base64.RawStdEncoding.EncodeToString(marshaledEncInfo)
-	var nextProtos []string
-	nextProtos = append(nextProtos, extraAlpnProtos...)
-	var count int
-	for i := 0; i < len(b64alpn); i += 230 {
-		end := i + 230
-		if end > len(b64alpn) {
-			end = len(b64alpn)
-		}
-		nextProtos = append(nextProtos, fmt.Sprintf("v1workerauth-%02d-%s", count, b64alpn[i:end]))
-		count++
-	}
-
-	cert, err := x509.ParseCertificate(certBytes)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	// Build local tls config
-	rootCAs := x509.NewCertPool()
-	rootCAs.AddCert(cert)
-	tlsCert, err := tls.X509KeyPair(info.CertPEM, info.KeyPEM)
-	if err != nil {
-		return nil, nil, err
-	}
-	tlsConfig := &tls.Config{
-		ServerName:   host,
-		Certificates: []tls.Certificate{tlsCert},
-		RootCAs:      rootCAs,
-		NextProtos:   nextProtos,
-		MinVersion:   tls.VersionTLS13,
-	}
-
-	return tlsConfig, info, nil
-}
-
 // workerConnectionInfo returns the worker's cluster.WorkerConnectionInfo as
 // a struct suitable to be used as the state field in a *protocol.Conn
 func (w *Worker) workerConnectionInfo(addr string) (*structpb.Struct, error) {

internal/daemon/worker/listeners.go

@@ -224,8 +224,8 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	}
 
 	// This handles connections coming in that are authenticated via
-	// nodeenrollment but not with any extra purpose; these are normal PKI
-	// worker connections
+	// nodeenrollment but not with any extra purpose; these are normal worker
+	// connections
 	nodeeAuthListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto, nodee.WithNativeConns(true))
 	if err != nil {
 		return nil, fmt.Errorf("error instantiating worker split listener: %w", err)
@@ -245,11 +245,12 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		return nil, fmt.Errorf("error instantiating reverse grpc split listener: %w", err)
 	}
 
-	// This wraps the reverse grpc pki worker connections with a listener which
-	// adds the worker key id of the connections to the worker's pkiConnManager.
-	revPkiWorkerTrackingListener, err := cluster.NewTrackingListener(w.baseContext, reverseGrpcListener, w.pkiConnManager, sourcePurpose(reverseGrpcListenerPurpose))
+	// This wraps the reverse grpc worker connections with a listener which adds
+	// the worker key id of the connections to the worker's downstream
+	// ConnManager.
+	revWorkerTrackingListener, err := cluster.NewTrackingListener(w.baseContext, reverseGrpcListener, w.downstreamConnManager, sourcePurpose(reverseGrpcListenerPurpose))
 	if err != nil {
-		return nil, fmt.Errorf("%s: error creating reverse grpc pki worker tracking listener: %w", op, err)
+		return nil, fmt.Errorf("%s: error creating reverse grpc worker tracking listener: %w", op, err)
 	}
 
 	// Connections coming in here are authed by nodeenrollment and are for the
@@ -260,10 +261,10 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	}
 
 	// This wraps the web socket proxying worker connections with a listener which
-	// adds the worker key id of the connections to the worker's pkiConnManager.
-	dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.pkiConnManager, sourcePurpose(multihopProxyDataplaneListenerPurpose))
+	// adds the worker key id of the connections to the worker's downstreamConnManager.
+	dataPlaneProxyTrackingListener, err := cluster.NewTrackingListener(w.baseContext, dataPlaneProxyListener, w.downstreamConnManager, sourcePurpose(multihopProxyDataplaneListenerPurpose))
 	if err != nil {
-		return nil, fmt.Errorf("%s: error creating web socket proxying tracking listener: %w", op, err)
+		return nil, fmt.Errorf("%s: error creating websocket proxying tracking listener: %w", op, err)
 	}
 
 	statsHandler, err := metric.InstrumentClusterStatsHandler(w.baseContext)
@@ -287,19 +288,19 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 
 	ln.GrpcServer = downstreamServer
 
-	// This wraps the normal pki worker connections with a listener which adds
-	// the worker key id of the  connections to the worker's pkiConnManager.
-	pkiWorkerTrackingListener, err := cluster.NewTrackingListener(cancelCtx, nodeeAuthListener, w.pkiConnManager, sourcePurpose(grpcListenerPurpose))
+	// This wraps the normal worker connections with a listener which adds the
+	// worker key id of the connections to the worker's downstreamConnManager.
+	workerTrackingListener, err := cluster.NewTrackingListener(cancelCtx, nodeeAuthListener, w.downstreamConnManager, sourcePurpose(grpcListenerPurpose))
 	if err != nil {
-		return nil, fmt.Errorf("%s: error creating pki worker tracking listener: %w", op, err)
+		return nil, fmt.Errorf("%s: error creating worker tracking listener: %w", op, err)
 	}
 
 	return func() {
-		handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, reverseGrpcListenerPurpose),
+		handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revWorkerTrackingListener, reverseGrpcListenerPurpose),
 			metric.InstrumentWorkerClusterTrackingListener(dataPlaneProxyTrackingListener, multihopProxyDataplaneListenerPurpose), w.downstreamReceiver)
 		go w.workerAuthSplitListener.Start()
 		go httpServer.Serve(proxyListener)
-		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, grpcListenerPurpose))
+		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(workerTrackingListener, grpcListenerPurpose))
 	}, nil
 }
 

internal/daemon/worker/status.go

@@ -198,7 +198,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
 			event.WithInfoMsg("error making status request to controller"))
 	}
 	versionInfo := version.Get()
-	connectionState := w.pkiConnManager.Connected()
+	connectionState := w.downstreamConnManager.Connected()
 	result, err := client.Status(statusCtx, &pbs.StatusRequest{
 		Jobs: activeJobs,
 		WorkerStatus: &pb.ServerWorkerStatus{

internal/daemon/worker/testing.go

@@ -224,9 +224,6 @@ 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
 
@@ -285,7 +282,6 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker {
 		}
 	}
 
-	opts.Config.Worker.UseDeprecatedKmsAuthMethod = opts.UseDeprecatedKmsAuthMethod
 	if len(opts.InitialUpstreams) > 0 {
 		opts.Config.Worker.InitialUpstreams = opts.InitialUpstreams
 	}

internal/daemon/worker/worker.go

@@ -179,7 +179,7 @@ type Worker struct {
 
 	statusLock sync.Mutex
 
-	pkiConnManager *cluster.DownstreamManager
+	downstreamConnManager *cluster.DownstreamManager
 }
 
 func New(ctx context.Context, conf *Config) (*Worker, error) {
@@ -205,7 +205,7 @@ func New(ctx context.Context, conf *Config) (*Worker, error) {
 		nonceFn:                     base62.Random,
 		WorkerAuthCurrentKeyId:      new(ua.String),
 		operationalState:            new(atomic.Value),
-		pkiConnManager:              cluster.NewDownstreamManager(),
+		downstreamConnManager:       cluster.NewDownstreamManager(),
 		successfulStatusGracePeriod: new(atomic.Int64),
 		statusCallTimeoutDuration:   new(atomic.Int64),
 		upstreamConnectionState:     new(atomic.Value),
@@ -369,150 +369,147 @@ func (w *Worker) Start() error {
 		return nil
 	}
 
-	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.
-		//
-		// 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
-		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"))
-			}
+	// 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.
+	//
+	// 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
+	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.WithStorageWrapper(w.conf.WorkerAuthStorageKms))
+	switch {
+	case err == nil:
+		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
+			break
 		}
 
-		var createNodeAuthCreds bool
-		var createFetchRequest bool
-		nodeCreds, err := types.LoadNodeCredentials(
-			w.baseContext,
-			w.WorkerAuthStorage,
-			nodeenrollment.CurrentId,
-			nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms))
-		switch {
-		case err == nil:
-			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
-				break
-			}
-
-			// Check that we have valid creds, or that we have generated creds but
-			// simply are still waiting on authentication (in which case we don't
-			// want to invalidate what we've already sent)
-			var validCreds bool
-			switch len(nodeCreds.CertificateBundles) {
-			case 0:
-				// Still waiting on initial creds, so don't invalidate the request
-				// by creating new credentials. However, we will generate and
-				// display a new valid request in case the first was lost.
-				createFetchRequest = true
-
-			default:
-				now := time.Now()
-				for _, bundle := range nodeCreds.CertificateBundles {
-					if bundle.CertificateNotBefore.AsTime().Before(now) && bundle.CertificateNotAfter.AsTime().After(now) {
-						// If we have a certificate in its validity period,
-						// everything is fine
-						validCreds = true
-						break
-					}
-				}
-
-				// Certificates are both expired, so create new credentials and
-				// output a request based on those
-				createNodeAuthCreds = !validCreds
-				createFetchRequest = !validCreds
-			}
-
-		case errors.Is(err, nodeenrollment.ErrNotFound):
-			// Nothing was found on disk, so create
-			createNodeAuthCreds = true
+		// Check that we have valid creds, or that we have generated creds but
+		// simply are still waiting on authentication (in which case we don't
+		// want to invalidate what we've already sent)
+		var validCreds bool
+		switch len(nodeCreds.CertificateBundles) {
+		case 0:
+			// Still waiting on initial creds, so don't invalidate the request
+			// by creating new credentials. However, we will generate and
+			// display a new valid request in case the first was lost.
 			createFetchRequest = true
 
 		default:
-			// Some other type of error happened, bail out
-			return fmt.Errorf("error loading worker auth creds: %w", err)
+			now := time.Now()
+			for _, bundle := range nodeCreds.CertificateBundles {
+				if bundle.CertificateNotBefore.AsTime().Before(now) && bundle.CertificateNotAfter.AsTime().After(now) {
+					// If we have a certificate in its validity period,
+					// everything is fine
+					validCreds = true
+					break
+				}
+			}
+
+			// Certificates are both expired, so create new credentials and
+			// output a request based on those
+			createNodeAuthCreds = !validCreds
+			createFetchRequest = !validCreds
 		}
 
-		// Don't output a fetch request if an activation token has been
-		// provided. Technically we _could_ still output a fetch request, and it
-		// would be valid to do so, but if a token was provided it may well be
-		// confusing to a user if it seems like it was ignored because a fetch
-		// request was still output.
-		if actToken := w.conf.RawConfig.Worker.ControllerGeneratedActivationToken; actToken != "" {
-			createFetchRequest = false
-		}
+	case errors.Is(err, nodeenrollment.ErrNotFound):
+		// Nothing was found on disk, so create
+		createNodeAuthCreds = true
+		createFetchRequest = true
 
-		// NOTE: this block _must_ be before the `if createFetchRequest` block
-		// or the fetch request may have no credentials to work with
-		if createNodeAuthCreds {
-			nodeCreds, err = types.NewNodeCredentials(
-				w.baseContext,
-				w.WorkerAuthStorage,
-				nodeenrollment.WithRandomReader(w.conf.SecureRandomReader),
-				nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
-			)
-			if err != nil {
-				return fmt.Errorf("error generating new worker auth creds: %w", err)
-			}
-		}
+	default:
+		// Some other type of error happened, bail out
+		return fmt.Errorf("error loading worker auth creds: %w", err)
+	}
 
-		if createFetchRequest {
-			if nodeCreds == nil {
-				return fmt.Errorf("need to create fetch request but worker auth creds are nil: %w", err)
-			}
-			req, err := nodeCreds.CreateFetchNodeCredentialsRequest(w.baseContext, nodeenrollment.WithRandomReader(w.conf.SecureRandomReader))
-			if err != nil {
-				return fmt.Errorf("error creating worker auth fetch credentials request: %w", err)
-			}
-			reqBytes, err := proto.Marshal(req)
-			if err != nil {
-				return fmt.Errorf("error marshaling worker auth fetch credentials request: %w", err)
-			}
-			w.WorkerAuthRegistrationRequest = base58.FastBase58Encoding(reqBytes)
-			if err != nil {
-				return fmt.Errorf("error encoding worker auth registration request: %w", err)
-			}
-			currentKeyId, err := nodeenrollment.KeyIdFromPkix(nodeCreds.CertificatePublicKeyPkix)
-			if err != nil {
-				return fmt.Errorf("error deriving worker auth key id: %w", err)
-			}
-			w.WorkerAuthCurrentKeyId.Store(currentKeyId)
+	// Don't output a fetch request if an activation token has been
+	// provided. Technically we _could_ still output a fetch request, and it
+	// would be valid to do so, but if a token was provided it may well be
+	// confusing to a user if it seems like it was ignored because a fetch
+	// request was still output.
+	if actToken := w.conf.RawConfig.Worker.ControllerGeneratedActivationToken; actToken != "" {
+		createFetchRequest = false
+	}
+
+	// NOTE: this block _must_ be before the `if createFetchRequest` block
+	// or the fetch request may have no credentials to work with
+	if createNodeAuthCreds {
+		nodeCreds, err = types.NewNodeCredentials(
+			w.baseContext,
+			w.WorkerAuthStorage,
+			nodeenrollment.WithRandomReader(w.conf.SecureRandomReader),
+			nodeenrollment.WithStorageWrapper(w.conf.WorkerAuthStorageKms),
+		)
+		if err != nil {
+			return fmt.Errorf("error generating new worker auth creds: %w", err)
+		}
+	}
+
+	if createFetchRequest {
+		if nodeCreds == nil {
+			return fmt.Errorf("need to create fetch request but worker auth creds are nil: %w", err)
+		}
+		req, err := nodeCreds.CreateFetchNodeCredentialsRequest(w.baseContext, nodeenrollment.WithRandomReader(w.conf.SecureRandomReader))
+		if err != nil {
+			return fmt.Errorf("error creating worker auth fetch credentials request: %w", err)
+		}
+		reqBytes, err := proto.Marshal(req)
+		if err != nil {
+			return fmt.Errorf("error marshaling worker auth fetch credentials request: %w", err)
+		}
+		w.WorkerAuthRegistrationRequest = base58.FastBase58Encoding(reqBytes)
+		if err != nil {
+			return fmt.Errorf("error encoding worker auth registration request: %w", err)
 		}
-		// Regardless, we want to load the currentKeyId
 		currentKeyId, err := nodeenrollment.KeyIdFromPkix(nodeCreds.CertificatePublicKeyPkix)
 		if err != nil {
 			return fmt.Errorf("error deriving worker auth key id: %w", err)
 		}
 		w.WorkerAuthCurrentKeyId.Store(currentKeyId)
 	}
+	// Regardless, we want to load the currentKeyId
+	currentKeyId, err := nodeenrollment.KeyIdFromPkix(nodeCreds.CertificatePublicKeyPkix)
+	if err != nil {
+		return fmt.Errorf("error deriving worker auth key id: %w", err)
+	}
+	w.WorkerAuthCurrentKeyId.Store(currentKeyId)
 
 	if err := w.StartControllerConnections(); err != nil {
 		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error making controller connections"))
 	}
 
-	var err error
 	w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn))
 	if err != nil {
 		return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager"))

internal/tests/cluster/switch_kms_method_test.go

@@ -1,80 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: BUSL-1.1
-
-package cluster
-
-import (
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/hashicorp/boundary/internal/cmd/config"
-	"github.com/hashicorp/boundary/internal/daemon/controller"
-	"github.com/hashicorp/boundary/internal/daemon/worker"
-	"github.com/hashicorp/boundary/internal/event"
-	"github.com/hashicorp/boundary/internal/types/scope"
-	"github.com/hashicorp/go-hclog"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestWorkerUpgradeKmsAuthMethod(t *testing.T) {
-	ec := event.TestEventerConfig(t, "TestWorkerUpgradeKmsAuthMethod", event.TestWithObservationSink(t), event.TestWithSysSink(t))
-	testLock := &sync.Mutex{}
-	testLogger := hclog.New(&hclog.LoggerOptions{
-		Mutex: testLock,
-		Name:  "test",
-	})
-	require.NoError(t, event.InitSysEventer(testLogger, testLock, "use-TestWorkerUpgradeKmsAuthMethod", event.WithEventerConfig(&ec.EventerConfig)))
-
-	conf, err := config.DevController()
-	conf.Eventing = &ec.EventerConfig
-	require.NoError(t, err)
-	c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-		Config: conf,
-	})
-	defer c1.Shutdown()
-	serversRepo := c1.ServersRepo()
-
-	conf, err = config.DevWorker()
-	conf.Eventing = &ec.EventerConfig
-	require.NoError(t, err)
-	w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		Config:                     conf,
-		WorkerAuthKms:              c1.Config().WorkerAuthKms,
-		InitialUpstreams:           c1.ClusterAddrs(),
-		UseDeprecatedKmsAuthMethod: true,
-	})
-	defer w1.Shutdown()
-
-	// Give time for it to connect
-	time.Sleep(10 * time.Second)
-
-	require.NoError(t, c1.WaitForNextWorkerStatusUpdate(w1.Name()))
-
-	// Verify it's KMS type
-	workers, err := serversRepo.ListWorkers(c1.Context(), []string{scope.Global.String()})
-	require.NoError(t, err)
-	require.Len(t, workers, 1)
-	assert.Equal(t, "kms", workers[0].Type)
-	oldId := workers[0].PublicId
-
-	require.NoError(t, w1.Worker().Shutdown())
-
-	// Now, start up again, using the new method
-	w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		Config:           conf,
-		WorkerAuthKms:    c1.Config().WorkerAuthKms,
-		InitialUpstreams: c1.ClusterAddrs(),
-	})
-	defer w1.Shutdown()
-
-	require.NoError(t, c1.WaitForNextWorkerStatusUpdate(w1.Name()))
-
-	// Verify it's PKI type
-	workers, err = serversRepo.ListWorkers(c1.Context(), []string{scope.Global.String()})
-	require.NoError(t, err)
-	require.Len(t, workers, 1)
-	assert.Equal(t, "pki", workers[0].Type)
-	assert.Equal(t, oldId, workers[0].PublicId)
-}

internal/tests/cluster/worker_replay_test.go

@@ -1,93 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: BUSL-1.1
-
-package cluster
-
-import (
-	"os"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/hashicorp/boundary/internal/cmd/config"
-	"github.com/hashicorp/boundary/internal/daemon/controller"
-	"github.com/hashicorp/boundary/internal/daemon/worker"
-	"github.com/hashicorp/boundary/internal/event"
-	"github.com/hashicorp/boundary/internal/server"
-	"github.com/hashicorp/go-hclog"
-	"github.com/stretchr/testify/require"
-)
-
-func TestWorkerReplay(t *testing.T) {
-	ec := event.TestEventerConfig(t, "TestWorkerReplay", event.TestWithObservationSink(t), event.TestWithSysSink(t))
-	testLock := &sync.Mutex{}
-	testLogger := hclog.New(&hclog.LoggerOptions{
-		Mutex: testLock,
-		Name:  "test",
-	})
-	require.NoError(t, event.InitSysEventer(testLogger, testLock, "use-TestWorkerReplay", event.WithEventerConfig(&ec.EventerConfig)))
-
-	conf, err := config.DevController()
-	conf.Eventing = &ec.EventerConfig
-	require.NoError(t, err)
-	c1 := controller.NewTestController(t, &controller.TestControllerOpts{
-		Config: conf,
-	})
-	defer c1.Shutdown()
-
-	conf, err = config.DevWorker()
-	conf.Eventing = &ec.EventerConfig
-	require.NoError(t, err)
-	w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		Config:           conf,
-		WorkerAuthKms:    c1.Config().WorkerAuthKms,
-		InitialUpstreams: c1.ClusterAddrs(),
-		NonceFn: func(length int) (string, error) {
-			return "test_noncetest_nonce", nil
-		},
-		UseDeprecatedKmsAuthMethod: true,
-	})
-
-	// Give time for it to connect
-	time.Sleep(10 * time.Second)
-
-	// Assert that the worker has connected
-	logBuf, err := os.ReadFile(ec.AllEvents.Name())
-	require.NoError(t, err)
-	require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed"))
-
-	// Assert we have the expected nonce in the DB
-	nonces, err := c1.ServersRepo().ListNonces(c1.Context(), server.NoncePurposeWorkerAuth)
-	require.NoError(t, err)
-	require.Len(t, nonces, 1)
-	require.Equal(t, &server.Nonce{Nonce: "test_noncetest_nonce", Purpose: server.NoncePurposeWorkerAuth}, nonces[0])
-
-	require.NoError(t, w1.Worker().Shutdown())
-
-	// Now, start up again with the same nonce
-	worker.NewTestWorker(t, &worker.TestWorkerOpts{
-		Config:           conf,
-		WorkerAuthKms:    c1.Config().WorkerAuthKms,
-		InitialUpstreams: c1.ClusterAddrs(),
-		NonceFn: func(length int) (string, error) {
-			return "test_noncetest_nonce", nil
-		},
-		UseDeprecatedKmsAuthMethod: true,
-	})
-
-	// Give time for it to connect
-	time.Sleep(10 * time.Second)
-
-	// We should find only one nonce, and one successful worker authentication,
-	// both in the output and in the database
-	ec.AllEvents.Close()
-	logBuf, err = os.ReadFile(ec.AllEvents.Name())
-	require.NoError(t, err)
-	require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed"))
-
-	nonces, err = c1.ServersRepo().ListNonces(c1.Context(), server.NoncePurposeWorkerAuth)
-	require.NoError(t, err)
-	require.Len(t, nonces, 1)
-	require.Equal(t, &server.Nonce{Nonce: "test_noncetest_nonce", Purpose: server.NoncePurposeWorkerAuth}, nonces[0])
-}