From 9332179a018eb37242a74f7b49cd2e7888541010 Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Fri, 12 Jan 2024 11:58:12 -0500 Subject: [PATCH] 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 | 2 + go.mod | 2 - .../cmd/base/internal/docker/supported.go | 3 +- internal/cmd/config/config.go | 10 +- internal/credential/vault/supported.go | 9 +- internal/daemon/controller/controller.go | 6 +- internal/daemon/controller/kms_conn_router.go | 156 ----------- internal/daemon/controller/listeners.go | 55 +--- internal/daemon/controller/listeners_test.go | 126 +++++---- .../daemon/controller/worker_tls_config.go | 145 ---------- internal/daemon/worker/auth_rotation.go | 5 - .../daemon/worker/controller_connection.go | 203 ++------------ internal/daemon/worker/listeners.go | 31 +-- internal/daemon/worker/status.go | 2 +- internal/daemon/worker/testing.go | 4 - internal/daemon/worker/worker.go | 247 +++++++++--------- .../tests/cluster/switch_kms_method_test.go | 80 ------ internal/tests/cluster/worker_replay_test.go | 93 ------- 18 files changed, 253 insertions(+), 926 deletions(-) delete mode 100644 internal/daemon/controller/kms_conn_router.go delete mode 100644 internal/daemon/controller/worker_tls_config.go delete mode 100644 internal/tests/cluster/switch_kms_method_test.go delete mode 100644 internal/tests/cluster/worker_replay_test.go diff --git a/globals/globals.go b/globals/globals.go index d213a3d01..1c712ca30 100644 --- a/globals/globals.go +++ b/globals/globals.go @@ -12,6 +12,8 @@ const ( AnyAuthenticatedUserId = "u_auth" AnonymousUserId = "u_anon" RecoveryUserId = "u_recovery" + + MinimumSupportedPostgresVersion = "12" ) type ( diff --git a/go.mod b/go.mod index db932e83a..0e8686af9 100644 --- a/go.mod +++ b/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 diff --git a/internal/cmd/base/internal/docker/supported.go b/internal/cmd/base/internal/docker/supported.go index fc6e81240..dc4b92546 100644 --- a/internal/cmd/base/internal/docker/supported.go +++ b/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) diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index b8db696d0..d0da2f5f7 100644 --- a/internal/cmd/config/config.go +++ b/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) diff --git a/internal/credential/vault/supported.go b/internal/credential/vault/supported.go index e4a81def5..fbd30a43f 100644 --- a/internal/credential/vault/supported.go +++ b/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 = ` diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index b885d6c83..53a055370 100644 --- a/internal/daemon/controller/controller.go +++ b/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) } diff --git a/internal/daemon/controller/kms_conn_router.go b/internal/daemon/controller/kms_conn_router.go deleted file mode 100644 index c354eb933..000000000 --- a/internal/daemon/controller/kms_conn_router.go +++ /dev/null @@ -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 -} diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index 5c46583dc..9843f6a8c 100644 --- a/internal/daemon/controller/listeners.go +++ b/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 diff --git a/internal/daemon/controller/listeners_test.go b/internal/daemon/controller/listeners_test.go index a10106594..a1b446e29 100644 --- a/internal/daemon/controller/listeners_test.go +++ b/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 } } diff --git a/internal/daemon/controller/worker_tls_config.go b/internal/daemon/controller/worker_tls_config.go deleted file mode 100644 index bed741a76..000000000 --- a/internal/daemon/controller/worker_tls_config.go +++ /dev/null @@ -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 -} diff --git a/internal/daemon/worker/auth_rotation.go b/internal/daemon/worker/auth_rotation.go index 6599fb507..16e6becb6 100644 --- a/internal/daemon/worker/auth_rotation.go +++ b/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 diff --git a/internal/daemon/worker/controller_connection.go b/internal/daemon/worker/controller_connection.go index e2c566c92..9321b6f41 100644 --- a/internal/daemon/worker/controller_connection.go +++ b/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) { diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index 303ca7dfb..9b4652020 100644 --- a/internal/daemon/worker/listeners.go +++ b/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 } diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 88a908699..150ac0804 100644 --- a/internal/daemon/worker/status.go +++ b/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{ diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index 227639fc5..5db707aab 100644 --- a/internal/daemon/worker/testing.go +++ b/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 } diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 4d5a598c5..ccf8be140 100644 --- a/internal/daemon/worker/worker.go +++ b/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")) diff --git a/internal/tests/cluster/switch_kms_method_test.go b/internal/tests/cluster/switch_kms_method_test.go deleted file mode 100644 index 56a5732eb..000000000 --- a/internal/tests/cluster/switch_kms_method_test.go +++ /dev/null @@ -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) -} diff --git a/internal/tests/cluster/worker_replay_test.go b/internal/tests/cluster/worker_replay_test.go deleted file mode 100644 index 2a6587545..000000000 --- a/internal/tests/cluster/worker_replay_test.go +++ /dev/null @@ -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]) -}