From 1aad2627c831289c1119e19caa0e743f98b99fcd Mon Sep 17 00:00:00 2001 From: Todd Date: Mon, 25 Jul 2022 12:52:21 -0700 Subject: [PATCH] OSS side of second downstream connections (see enterprise PR 69) (#2292) --- Makefile | 2 +- internal/daemon/common/const.go | 3 + internal/daemon/common/eventing_listener.go | 62 +++++++ .../controller/intercepting_listener.go | 145 ----------------- internal/daemon/controller/kms_conn_router.go | 152 ++++++++++++++++++ internal/daemon/controller/listeners.go | 81 +++++++++- internal/daemon/worker/addressreceiver.go | 45 ++++++ internal/daemon/worker/auth_rotation_test.go | 10 +- .../daemon/worker/controller_connection.go | 46 +++--- internal/daemon/worker/listeners.go | 98 +++++------ internal/daemon/worker/status.go | 38 +++-- internal/daemon/worker/worker.go | 86 +++------- internal/daemon/worker/worker_test.go | 2 - internal/tests/cluster/worker_tagging_test.go | 6 +- 14 files changed, 460 insertions(+), 316 deletions(-) create mode 100644 internal/daemon/common/const.go create mode 100644 internal/daemon/common/eventing_listener.go delete mode 100644 internal/daemon/controller/intercepting_listener.go create mode 100644 internal/daemon/controller/kms_conn_router.go create mode 100644 internal/daemon/worker/addressreceiver.go diff --git a/Makefile b/Makefile index 805afb5d7..7cc4cbcbe 100644 --- a/Makefile +++ b/Makefile @@ -248,7 +248,7 @@ generate-database-dumps: test-ci: export CI_BUILD=1 test-ci: CGO_ENABLED=$(CGO_ENABLED) BUILD_TAGS='$(BUILD_TAGS)' sh -c "'$(CURDIR)/scripts/build.sh'" - ~/.go/bin/go test ./... -v $(TESTARGS) -json -cover -timeout 120m | tparse + ~/.go/bin/go test ./... -v $(TESTARGS) -json -cover -timeout 120m | tparse -follow .PHONY: test-sql test-sql: diff --git a/internal/daemon/common/const.go b/internal/daemon/common/const.go new file mode 100644 index 000000000..27b5e2925 --- /dev/null +++ b/internal/daemon/common/const.go @@ -0,0 +1,3 @@ +package common + +const ReverseGrpcConnectionAlpnValue = "the-downstream-dialer-plays-an-uno-reverse-card" diff --git a/internal/daemon/common/eventing_listener.go b/internal/daemon/common/eventing_listener.go new file mode 100644 index 000000000..105e3d713 --- /dev/null +++ b/internal/daemon/common/eventing_listener.go @@ -0,0 +1,62 @@ +package common + +import ( + "context" + "crypto/tls" + "net" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/observability/event" + nodee "github.com/hashicorp/nodeenrollment" +) + +// EventingListener simply sends an event when a worker has connected +// successfully +type eventingListener struct { + ctx context.Context + baseLn net.Listener +} + +func NewEventingListener(ctx context.Context, baseLn net.Listener) (net.Listener, error) { + const op = "common.(EventingListener).Accept" + switch { + case ctx == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "nil context") + case baseLn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "nil base listener") + } + return &eventingListener{ + ctx: ctx, + baseLn: baseLn, + }, nil +} + +func (e *eventingListener) Accept() (net.Conn, error) { + const op = "common.(EventingListener).Accept" + conn, err := e.baseLn.Accept() + if err != nil || conn == nil { + return conn, err + } + + tlsConn, ok := conn.(*tls.Conn) + if !ok { + return conn, err + } + + if tlsConn != nil && len(tlsConn.ConnectionState().PeerCertificates) > 0 { + keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId) + if err == nil { + event.WriteSysEvent(e.ctx, op, "worker successfully authenticated", "key_id", keyId) + } + } + + return conn, err +} + +func (e *eventingListener) Close() error { + return e.baseLn.Close() +} + +func (e *eventingListener) Addr() net.Addr { + return e.baseLn.Addr() +} diff --git a/internal/daemon/controller/intercepting_listener.go b/internal/daemon/controller/intercepting_listener.go deleted file mode 100644 index 3ae9831b7..000000000 --- a/internal/daemon/controller/intercepting_listener.go +++ /dev/null @@ -1,145 +0,0 @@ -package controller - -import ( - "crypto/tls" - "errors" - "fmt" - "net" - "strings" - - "github.com/hashicorp/boundary/internal/observability/event" - nodee "github.com/hashicorp/nodeenrollment" - "github.com/hashicorp/nodeenrollment/protocol" -) - -// 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 -} - -// interceptingListener allows us to validate the nonce from a connection before -// handing it off to 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. -// -// Note that this is semi-weak against a scenario where the value is decrypted -// later since a controller restart would clear the cache. We could store a list -// of seen nonces in the database, but since the original certificate was only -// good for 3 minutes and 30 seconds anyways, the decryption would need to -// happen within a short time window instead of much later. We can adjust this -// window if we want (or even make it tunable), or store values in the DB as -// well until the certificate expiration. -type interceptingListener struct { - baseLn net.Listener - c *Controller -} - -func newInterceptingListener(c *Controller, baseLn net.Listener) *interceptingListener { - ret := &interceptingListener{ - c: c, - baseLn: baseLn, - } - - return ret -} - -func (m *interceptingListener) Accept() (net.Conn, error) { - const op = "controller.(interceptingListener).Accept" - ctx := m.c.baseContext - conn, err := m.baseLn.Accept() - if err != nil { - // We may already be closed but can't be sure, so just try anyways - if conn != nil { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) - } - } - return nil, newTempError(err) - } - - var tlsConn *tls.Conn - switch c := conn.(type) { - case *protocol.Conn: - // If we so choose, at this point we can pull out the client's - // NextProtos with c.ClientNextProtos - tlsConn = c.Conn - case *tls.Conn: - tlsConn = c - default: - err := fmt.Errorf("unknown connection type %T received", c) - event.WriteError(ctx, op, err) - return nil, newTempError(err) - } - - switch { - case nodee.ContainsKnownAlpnProto(tlsConn.ConnectionState().NegotiatedProtocol): - keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId) - if err != nil { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) - } - return nil, newTempError(fmt.Errorf("error deriving key id from newly-authenticated node: %w", err)) - } - event.WriteSysEvent(ctx, op, "worker successfully authenticated", "key_id", keyId) - return conn, nil - - case strings.HasPrefix(tlsConn.ConnectionState().NegotiatedProtocol, "v1workerauth"): - nonce := make([]byte, 20) - read, err := conn.Read(nonce) - if err != nil { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) - } - return nil, newTempError(fmt.Errorf("error reading nonce from connection: %w", err)) - } - if read != len(nonce) { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) - } - return nil, newTempError(fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read)) - } - workerInfoRaw, found := m.c.workerAuthCache.Load(string(nonce)) - if !found { - if err := conn.Close(); err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection")) - } - return nil, newTempError(errors.New("did not find valid nonce for incoming worker")) - } - workerInfo := workerInfoRaw.(*workerAuthEntry) - workerInfo.conn = tlsConn - m.c.workerAuthCache.Delete(string(nonce)) - event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name, "description", workerInfo.Description, "proxy_address", workerInfo.ProxyAddress) - return protocol.NewConn(tlsConn, workerInfo.clientNextProtos), nil - - default: - return nil, newTempError(errors.New("unable to authenticate incoming connection")) - } -} - -func (m *interceptingListener) Close() error { - return m.baseLn.Close() -} - -func (m *interceptingListener) Addr() net.Addr { - return m.baseLn.Addr() -} diff --git a/internal/daemon/controller/kms_conn_router.go b/internal/daemon/controller/kms_conn_router.go new file mode 100644 index 000000000..80113eb01 --- /dev/null +++ b/internal/daemon/controller/kms_conn_router.go @@ -0,0 +1,152 @@ +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/observability/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 3475819b2..a909ee04d 100644 --- a/internal/daemon/controller/listeners.go +++ b/internal/daemon/controller/listeners.go @@ -3,7 +3,6 @@ package controller import ( "context" "crypto/tls" - "errors" "fmt" "math" "net" @@ -14,14 +13,27 @@ import ( grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware" "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/daemon/cluster/handlers" + "github.com/hashicorp/boundary/internal/daemon/common" "github.com/hashicorp/boundary/internal/daemon/controller/internal/metric" + "github.com/hashicorp/boundary/internal/errors" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/go-multierror" "github.com/hashicorp/nodeenrollment/multihop" + nodeenet "github.com/hashicorp/nodeenrollment/net" "github.com/hashicorp/nodeenrollment/protocol" "google.golang.org/grpc" ) +// the function that handles a secondary connection over a provided listener +var handleSecondaryConnection = closeListener + +func closeListener(_ context.Context, l net.Listener) error { + if l != nil { + return l.Close() + } + return nil +} + func (c *Controller) startListeners() error { servers := make([]func(), 0, len(c.conf.Listeners)) @@ -112,7 +124,10 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error return nil, fmt.Errorf("error fetching worker auth storage: %w", err) } - l, err := protocol.NewInterceptingListener( + // The cluster listener is shut down at server shutdown time so we do not + // need to handle individual listener shutdown. + var l net.Listener + l, err = protocol.NewInterceptingListener( &protocol.InterceptingListenerConfiguration{ Context: c.baseContext, Storage: workerAuthStorage, @@ -125,9 +140,63 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error return nil, fmt.Errorf("error instantiating node auth listener: %w", err) } + // Create split listener + splitListener, err := nodeenet.NewSplitListener(l) + if err != nil { + return nil, fmt.Errorf("error instantiating split listener: %w", err) + } + + // This handles connections coming in on the cluster port that are + // authenticated via nodeenrollment but not with any extra purpose; these + // are normal worker connections + nodeeAuthedListener, err := splitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto) + if err != nil { + return nil, fmt.Errorf("error instantiating node enrollment authed split listener: %w", err) + } + // This wraps the normal worker connections with something that events with + // the worker ID on successful auth/connection + eventingAuthedListener, err := common.NewEventingListener(c.baseContext, nodeeAuthedListener) + if err != nil { + return nil, fmt.Errorf("%s: error creating eventing 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(eventingAuthedListener); 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) + } + + // Connections coming in here are authed by nodeenrollment and are for the + // reverse grpc purpose + reverseGrpcListener, err := splitListener.GetListener(common.ReverseGrpcConnectionAlpnValue) + if err != nil { + return nil, fmt.Errorf("error instantiating reverse gprc connection split listener: %w", 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(reverseGrpcListener); 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) + } + workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer) if err != nil { - return nil, fmt.Errorf("error getting sub-listener for worker proto: %w", err) + return nil, fmt.Errorf("error getting request interceptor for worker proto: %w", err) } workerServer := grpc.NewServer( @@ -162,7 +231,11 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error ln.GrpcServer = workerServer - return func() { go ln.GrpcServer.Serve(newInterceptingListener(c, l)) }, nil + return func() { + go splitListener.Start() + go handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener) + go ln.GrpcServer.Serve(multiplexingAuthedListener) + }, nil } func (c *Controller) stopServersAndListeners() error { diff --git a/internal/daemon/worker/addressreceiver.go b/internal/daemon/worker/addressreceiver.go new file mode 100644 index 000000000..5e52e868b --- /dev/null +++ b/internal/daemon/worker/addressreceiver.go @@ -0,0 +1,45 @@ +package worker + +import ( + "context" + + "google.golang.org/grpc/resolver" + "google.golang.org/grpc/resolver/manual" +) + +var extraAddressReceivers = noopAddressReceivers + +func noopAddressReceivers(context.Context, *Worker) ([]addressReceiver, error) { + return nil, nil +} + +// addressReceiver allows the initializing and setting of addresses. Since a +// main use case of this interface is to use it in grpc dialing it satisfies +// a grpc resolver.Builder interface as well. +type addressReceiver interface { + InitialAddresses([]string) + SetAddresses([]string) +} + +// grpcResolverReceiver is an addressReceiver which wraps a grpc manual.Resolver +// InitialAddresses is passed onto the Resolver's InitialState method and +// SetAddresses is passed onto the Resolver's UpdateState call. +type grpcResolverReceiver struct { + *manual.Resolver +} + +func (r grpcResolverReceiver) getState(addrs []string) resolver.State { + var rAddrs []resolver.Address + for _, a := range addrs { + rAddrs = append(rAddrs, resolver.Address{Addr: a}) + } + return resolver.State{Addresses: rAddrs} +} + +func (r *grpcResolverReceiver) InitialAddresses(addrs []string) { + r.Resolver.InitialState(r.getState(addrs)) +} + +func (r *grpcResolverReceiver) SetAddresses(addrs []string) { + r.Resolver.UpdateState(r.getState(addrs)) +} diff --git a/internal/daemon/worker/auth_rotation_test.go b/internal/daemon/worker/auth_rotation_test.go index 1b9df50c6..b41266498 100644 --- a/internal/daemon/worker/auth_rotation_test.go +++ b/internal/daemon/worker/auth_rotation_test.go @@ -16,15 +16,12 @@ import ( "github.com/mr-tron/base58" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" ) // TestRotationTicking ensures that we see new credential information for a // worker on both the controller side and worker side in a periodic fashion func TestRotationTicking(t *testing.T) { - t.Parallel() - require, assert := require.New(t), assert.New(t) logger := hclog.New(&hclog.LoggerOptions{ Level: hclog.Trace, @@ -41,10 +38,16 @@ func TestRotationTicking(t *testing.T) { const rotationPeriod = 20 * time.Second + // names should not be set when using pki workers + wConf, err := config.DevWorker() + require.NoError(err) + wConf.Worker.Name = "" + wConf.Worker.InitialUpstreams = c.ClusterAddrs() w := worker.NewTestWorker(t, &worker.TestWorkerOpts{ InitialUpstreams: c.ClusterAddrs(), Logger: logger.Named("worker"), AuthRotationPeriod: rotationPeriod, + Config: wConf, }) t.Cleanup(w.Shutdown) @@ -109,7 +112,6 @@ func TestRotationTicking(t *testing.T) { // Stop and start the client connections to ensure the new credentials // are valid; if not, we won't establish a new connection and rotation // will fail - w.Worker().Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}}) require.NotNil(w.Worker().GrpcClientConn) require.NoError(w.Worker().GrpcClientConn.Close()) require.NoError(w.Worker().StartControllerConnections()) diff --git a/internal/daemon/worker/controller_connection.go b/internal/daemon/worker/controller_connection.go index 0d04800c5..2dff70650 100644 --- a/internal/daemon/worker/controller_connection.go +++ b/internal/daemon/worker/controller_connection.go @@ -8,7 +8,6 @@ import ( "encoding/base64" "encoding/json" "encoding/pem" - "errors" "fmt" "math" "math/big" @@ -22,6 +21,7 @@ import ( "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/daemon/worker/internal/metric" + "github.com/hashicorp/boundary/internal/errors" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-secure-stdlib/base62" @@ -41,11 +41,11 @@ const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud:9202" // connection client creation func (w *Worker) StartControllerConnections() error { const op = "worker.(Worker).StartControllerConnections" - initialAddrs := make([]resolver.Address, 0, len(w.conf.RawConfig.Worker.InitialUpstreams)) + initialAddrs := make([]string, 0, len(w.conf.RawConfig.Worker.InitialUpstreams)) for _, addr := range w.conf.RawConfig.Worker.InitialUpstreams { switch { case strings.HasPrefix(addr, "/"): - initialAddrs = append(initialAddrs, resolver.Address{Addr: addr}) + initialAddrs = append(initialAddrs, addr) default: host, port, err := net.SplitHostPort(addr) if err != nil && strings.Contains(err.Error(), "missing port in address") { @@ -54,40 +54,39 @@ func (w *Worker) StartControllerConnections() error { if err != nil { return fmt.Errorf("error parsing upstream address: %w", err) } - initialAddrs = append(initialAddrs, resolver.Address{Addr: net.JoinHostPort(host, port)}) + initialAddrs = append(initialAddrs, net.JoinHostPort(host, port)) } } if len(initialAddrs) == 0 { if w.conf.RawConfig.HcpbClusterId != "" { clusterAddress := fmt.Sprintf("%s%s", w.conf.RawConfig.HcpbClusterId, hcpbUrlSuffix) - initialAddrs = append(initialAddrs, resolver.Address{Addr: clusterAddress}) + initialAddrs = append(initialAddrs, clusterAddress) event.WriteSysEvent(w.baseContext, op, fmt.Sprintf("Setting HCP Boundary cluster address %s as upstream address", clusterAddress)) } else { - return errors.New("no initial upstream addresses found") + return errors.New(w.baseContext, errors.InvalidParameter, op, "no initial upstream addresses found") } } - w.Resolver().InitialState(resolver.State{ - Addresses: initialAddrs, - }) - if err := w.createClientConn(initialAddrs[0].Addr); err != nil { - return fmt.Errorf("error making client connection to upstream address %s: %w", initialAddrs[0].Addr, err) + for _, ar := range w.addressReceivers { + ar.InitialAddresses(initialAddrs) + } + if err := w.createClientConn(initialAddrs[0]); err != nil { + return fmt.Errorf("error making client connection to upstream address %s: %w", initialAddrs[0], err) } - return nil } -func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, error) { +func (w *Worker) controllerDialerFunc(extraAlpnProtos ...string) func(context.Context, string) (net.Conn, error) { const op = "worker.(Worker).controllerDialerFunc" return func(ctx context.Context, addr string) (net.Conn, error) { var conn net.Conn var err error switch { case w.conf.WorkerAuthKms != nil && !w.conf.DevUsePkiForUpstream: - conn, err = w.v1KmsAuthDialFn(ctx, addr) + conn, err = w.v1KmsAuthDialFn(ctx, addr, extraAlpnProtos...) default: - conn, err = protocol.Dial(ctx, w.WorkerAuthStorage, addr, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms)) + conn, err = protocol.Dial(ctx, w.WorkerAuthStorage, addr, nodeenrollment.WithWrapper(w.conf.WorkerAuthStorageKms), nodeenrollment.WithExtraAlpnProtos(extraAlpnProtos)) // No error and a valid connection means the WorkerAuthRegistrationRequest was populated // We can remove the stored workerAuthRequest file if err == nil && conn != nil { @@ -120,9 +119,9 @@ func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, } } -func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string) (net.Conn, error) { +func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string, extraAlpnProtos ...string) (net.Conn, error) { const op = "worker.(Worker).v1KmsAuthDialFn" - tlsConf, authInfo, err := w.workerAuthTLSConfig() + tlsConf, authInfo, err := w.workerAuthTLSConfig(extraAlpnProtos...) if err != nil { return nil, fmt.Errorf("error creating tls config for worker auth: %w", err) } @@ -169,7 +168,15 @@ func (w *Worker) createClientConn(addr string) error { ] } `, defaultTimeout) - res := w.Resolver() + var res resolver.Builder + for _, v := range w.addressReceivers { + if rec, ok := v.(*grpcResolverReceiver); ok { + res = rec.Resolver + } + } + if res == nil { + return errors.New(w.baseContext, errors.Internal, op, "unable to find a resolver.Builder amongst the address receivers") + } dialOpts := []grpc.DialOption{ grpc.WithResolvers(res), grpc.WithUnaryInterceptor(metric.InstrumentClusterClient()), @@ -204,7 +211,7 @@ func (w *Worker) createClientConn(addr string) error { return nil } -func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) { +func (w *Worker) workerAuthTLSConfig(extraAlpnProtos ...string) (*tls.Config, *base.WorkerAuthInfo, error) { var err error info := &base.WorkerAuthInfo{ Name: w.conf.RawConfig.Worker.Name, @@ -275,6 +282,7 @@ func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error } 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 diff --git a/internal/daemon/worker/listeners.go b/internal/daemon/worker/listeners.go index 3ee07516e..f42e47df0 100644 --- a/internal/daemon/worker/listeners.go +++ b/internal/daemon/worker/listeners.go @@ -3,7 +3,7 @@ package worker import ( "context" "crypto/tls" - "errors" + stderrers "errors" "fmt" "log" "math" @@ -14,7 +14,9 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/daemon/cluster/handlers" + "github.com/hashicorp/boundary/internal/daemon/common" "github.com/hashicorp/boundary/internal/daemon/worker/session" + "github.com/hashicorp/boundary/internal/errors" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-multierror" @@ -27,6 +29,16 @@ import ( "google.golang.org/grpc" ) +// the function that handles a secondary connection over a provided listener +var handleSecondaryConnection = closeListener + +func closeListener(_ context.Context, l net.Listener) error { + if l != nil { + return l.Close() + } + return nil +} + func (w *Worker) startListeners(sm session.Manager) error { const op = "worker.(Worker).startListeners" @@ -92,11 +104,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, ) (*types.FetchNodeCredentialsResponse, error) { client := w.controllerMultihopConn.Load() if client == nil { - return nil, temperror.New(errors.New("error fetching controller connection, client is nil")) + return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil")) } multihopClient, ok := client.(multihop.MultihopServiceClient) if !ok { - return nil, temperror.New(errors.New("client could not be understood as a multihop service client")) + return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client")) } // log.Println("proxying fetch node credentials request") return multihopClient.FetchNodeCredentials(ctx, req) @@ -110,11 +122,11 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, ) (*types.GenerateServerCertificatesResponse, error) { client := w.controllerMultihopConn.Load() if client == nil { - return nil, temperror.New(errors.New("error fetching controller connection, client is nil")) + return nil, temperror.New(stderrers.New("error fetching controller connection, client is nil")) } multihopClient, ok := client.(multihop.MultihopServiceClient) if !ok { - return nil, temperror.New(errors.New("client could not be understood as a multihop service client")) + return nil, temperror.New(stderrers.New("client could not be understood as a multihop service client")) } // log.Println("proxying generate server cert request") return multihopClient.GenerateServerCertificates(ctx, req) @@ -135,15 +147,30 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, return nil, fmt.Errorf("error instantiating node auth listener: %w", err) } + // Create split listener w.workerAuthSplitListener, err = nodeenet.NewSplitListener(interceptingListener) if err != nil { return nil, fmt.Errorf("error instantiating split listener: %w", err) } - workerListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto) + + // This handles connections coming in that are authenticated via + // nodeenrollment but not with any extra purpose; these are normal PKI + // worker connections + nodeeAuthListener, err := w.workerAuthSplitListener.GetListener(nodeenet.AuthenticatedNonSpecificNextProto) if err != nil { return nil, fmt.Errorf("error instantiating worker split listener: %w", err) } - nonWorkerListener, err := w.workerAuthSplitListener.GetListener(nodeenet.UnauthenticatedNextProto) + + // Connections that come into here are not authed by nodeenrollment so are + // proxy connections + proxyListener, err := w.workerAuthSplitListener.GetListener(nodeenet.UnauthenticatedNextProto) + if err != nil { + return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err) + } + + // Connections coming in here are authed by nodeenrollment and are for the + // reverse grpc purpose + reverseGrpcListener, err := w.workerAuthSplitListener.GetListener(common.ReverseGrpcConnectionAlpnValue) if err != nil { return nil, fmt.Errorf("error instantiating non-worker split listener: %w", err) } @@ -167,15 +194,16 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger, ln.GrpcServer = downstreamServer + eventingListener, err := common.NewEventingListener(cancelCtx, nodeeAuthListener) + if err != nil { + return nil, fmt.Errorf("%s: error creating eventing listener: %w", op, err) + } + return func() { go w.workerAuthSplitListener.Start() - go httpServer.Serve(nonWorkerListener) - go ln.GrpcServer.Serve( - &eventingListener{ - ctx: cancelCtx, - baseLn: workerListener, - }, - ) + go httpServer.Serve(proxyListener) + go ln.GrpcServer.Serve(eventingListener) + go handleSecondaryConnection(cancelCtx, reverseGrpcListener) }, nil } @@ -256,45 +284,3 @@ func listenerCloseErrorCheck(lnType string, err error) error { return err } - -type eventingListener struct { - ctx context.Context - baseLn net.Listener -} - -func (e *eventingListener) Accept() (net.Conn, error) { - const op = "worker.(eventingListener).Accept" - conn, err := e.baseLn.Accept() - if err != nil || conn == nil { - return conn, err - } - - // This is all best-effort; anything going wrong here shouldn't disrupt the - // connection, so on error simply stop trying to get to an event - var tlsConn *tls.Conn - switch c := conn.(type) { - case *protocol.Conn: - // If we so choose, at this point we can pull out the client's - // NextProtos with c.ClientNextProtos - tlsConn = c.Conn - case *tls.Conn: - tlsConn = c - } - - if tlsConn != nil && len(tlsConn.ConnectionState().PeerCertificates) > 0 { - keyId, err := nodee.KeyIdFromPkix(tlsConn.ConnectionState().PeerCertificates[0].SubjectKeyId) - if err == nil { - event.WriteSysEvent(e.ctx, op, "worker successfully authenticated", "key_id", keyId) - } - } - - return conn, err -} - -func (e *eventingListener) Close() error { - return e.baseLn.Close() -} - -func (e *eventingListener) Addr() net.Addr { - return e.baseLn.Addr() -} diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 05717abf2..214d9110a 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -14,7 +14,6 @@ import ( pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-secure-stdlib/strutil" - "google.golang.org/grpc/resolver" ) type LastStatusInformation struct { @@ -23,7 +22,7 @@ type LastStatusInformation struct { LastCalculatedUpstreams []string } -func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager session.Manager) { +func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager session.Manager, addrReceivers []addressReceiver) { const op = "worker.(Worker).startStatusTicking" r := rand.New(rand.NewSource(time.Now().UnixNano())) // This function exists to desynchronize calls to controllers from @@ -55,7 +54,7 @@ func (w *Worker) startStatusTicking(cancelCtx context.Context, sessionManager se continue } - w.sendWorkerStatus(cancelCtx, sessionManager) + w.sendWorkerStatus(cancelCtx, sessionManager, addrReceivers) timer.Reset(getRandomInterval()) } } @@ -99,7 +98,7 @@ func (w *Worker) WaitForNextSuccessfulStatusUpdate() error { return nil } -func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager) { +func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager, addressReceivers []addressReceiver) { const op = "worker.(Worker).sendWorkerStatus" // First send info as-is. We'll perform cleanup duties after we @@ -193,15 +192,12 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess } } else { w.updateTags.Store(false) - var addrs []resolver.Address - var newUpstreams []string + var addrs []string // This may be empty if we are in a multiple hop scenario if len(result.CalculatedUpstreams) > 0 { - addrs = make([]resolver.Address, 0, len(result.CalculatedUpstreams)) - newUpstreams = make([]string, 0, len(result.CalculatedUpstreams)) + addrs = make([]string, 0, len(result.CalculatedUpstreams)) for _, v := range result.CalculatedUpstreams { - addrs = append(addrs, resolver.Address{Addr: v.Address}) - newUpstreams = append(newUpstreams, v.Address) + addrs = append(addrs, v.Address) } } else if w.conf.RawConfig.HcpbClusterId != "" { @@ -212,27 +208,29 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess if err != nil { event.WriteError(hcpbWorkersCtx, op, err, event.WithInfoMsg("error fetching managed worker information")) } else { - addrs = make([]resolver.Address, 0, len(workersResp.Workers)) - newUpstreams = make([]string, 0, len(workersResp.Workers)) + addrs = make([]string, 0, len(workersResp.Workers)) for _, v := range workersResp.Workers { - addrs = append(addrs, resolver.Address{Addr: v.Address}) - newUpstreams = append(newUpstreams, v.Address) + addrs = append(addrs, v.Address) } } } if len(addrs) > 0 { lastStatus := w.lastStatusSuccess.Load().(*LastStatusInformation) // Compare upstreams; update resolver if there is a difference, and emit an event with old and new addresses - if lastStatus != nil && !strutil.EquivalentSlices(lastStatus.LastCalculatedUpstreams, newUpstreams) { - upstreamsMessage := fmt.Sprintf("Upstreams has changed; old upstreams were: %s, new upstreams are: %s", lastStatus.LastCalculatedUpstreams, newUpstreams) + if lastStatus != nil && !strutil.EquivalentSlices(lastStatus.LastCalculatedUpstreams, addrs) { + upstreamsMessage := fmt.Sprintf("Upstreams has changed; old upstreams were: %s, new upstreams are: %s", lastStatus.LastCalculatedUpstreams, addrs) event.WriteSysEvent(cancelCtx, op, upstreamsMessage) - w.Resolver().UpdateState(resolver.State{Addresses: addrs}) + for _, as := range addressReceivers { + as.SetAddresses(addrs) + } } else if lastStatus == nil { - w.Resolver().UpdateState(resolver.State{Addresses: addrs}) - event.WriteSysEvent(cancelCtx, op, fmt.Sprintf("Upstreams after first status set to: %s", newUpstreams)) + for _, as := range addressReceivers { + as.SetAddresses(addrs) + } + event.WriteSysEvent(cancelCtx, op, fmt.Sprintf("Upstreams after first status set to: %s", addrs)) } } - w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now(), LastCalculatedUpstreams: newUpstreams}) + w.lastStatusSuccess.Store(&LastStatusInformation{StatusResponse: result, StatusTime: time.Now(), LastCalculatedUpstreams: addrs}) for _, request := range result.GetJobsRequests() { switch request.GetRequestType() { diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index ce1e66faa..61100675a 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -6,7 +6,6 @@ import ( "crypto/rand" "crypto/tls" "crypto/x509" - "errors" "fmt" "strconv" "strings" @@ -14,10 +13,12 @@ import ( "sync/atomic" "time" + "github.com/hashicorp/boundary/internal/errors" pb "github.com/hashicorp/boundary/internal/gen/controller/servers" - "github.com/hashicorp/nodeenrollment" + nodeenet "github.com/hashicorp/nodeenrollment/net" "github.com/mr-tron/base58" + "google.golang.org/grpc/resolver/manual" "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/base" @@ -29,14 +30,10 @@ import ( "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" - "github.com/hashicorp/nodeenrollment/multihop" - nodeenet "github.com/hashicorp/nodeenrollment/net" nodeefile "github.com/hashicorp/nodeenrollment/storage/file" "github.com/hashicorp/nodeenrollment/types" ua "go.uber.org/atomic" "google.golang.org/grpc" - "google.golang.org/grpc/resolver" - "google.golang.org/grpc/resolver/manual" "google.golang.org/protobuf/proto" ) @@ -57,6 +54,9 @@ type Worker struct { // This is exported for tests. GrpcClientConn *grpc.ClientConn + // receives address updates and contains the grpc resolver. + addressReceivers []addressReceiver + sessionManager session.Manager controllerStatusConn *atomic.Value @@ -64,8 +64,6 @@ type Worker struct { lastStatusSuccess *atomic.Value workerStartTime time.Time - controllerResolver *atomic.Value - controllerMultihopConn *atomic.Value proxyListener *base.ServerListener @@ -94,6 +92,7 @@ type Worker struct { } func New(conf *Config) (*Worker, error) { + const op = "worker.New" metric.InitializeHttpCollectors(conf.PrometheusRegisterer) metric.InitializeWebsocketCollectors(conf.PrometheusRegisterer) metric.InitializeClusterClientCollectors(conf.PrometheusRegisterer) @@ -105,7 +104,6 @@ func New(conf *Config) (*Worker, error) { controllerStatusConn: new(atomic.Value), everAuthenticated: ua.NewBool(false), lastStatusSuccess: new(atomic.Value), - controllerResolver: new(atomic.Value), controllerMultihopConn: new(atomic.Value), tags: new(atomic.Value), updateTags: ua.NewBool(false), @@ -116,7 +114,7 @@ func New(conf *Config) (*Worker, error) { w.lastStatusSuccess.Store((*LastStatusInformation)(nil)) scheme := strconv.FormatInt(time.Now().UnixNano(), 36) controllerResolver := manual.NewBuilderWithScheme(scheme) - w.controllerResolver.Store(controllerResolver) + w.addressReceivers = []addressReceiver{&grpcResolverReceiver{controllerResolver}} if conf.RawConfig.Worker == nil { conf.RawConfig.Worker = new(config.Worker) @@ -178,6 +176,12 @@ func (w *Worker) Start() error { return nil } + extraAddrReceivers, err := extraAddressReceivers(w.baseContext, w) + if err != nil { + return errors.Wrap(w.baseContext, err, op) + } + w.addressReceivers = append(w.addressReceivers, extraAddrReceivers...) + if w.conf.WorkerAuthKms == nil || w.conf.DevUsePkiForUpstream { // In this section, we look for existing worker credentials. The two // variables below store whether to create new credentials and whether @@ -191,7 +195,7 @@ func (w *Worker) Start() error { w.WorkerAuthStorage, err = nodeefile.New(w.baseContext, nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.AuthStoragePath)) if err != nil { - return fmt.Errorf("error loading worker auth storage directory: %w", err) + return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error loading worker auth storage directory")) } var createNodeAuthCreds bool @@ -292,17 +296,16 @@ func (w *Worker) Start() error { } if err := w.StartControllerConnections(); err != nil { - return fmt.Errorf("error making controller connections: %w", err) + 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 fmt.Errorf("error creating session manager: %w", err) + return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager")) } if err := w.startListeners(w.sessionManager); err != nil { - return fmt.Errorf("error starting worker listeners: %w", err) + return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error starting worker listeners")) } // Rather than deal with some of the potential error conditions for Add on @@ -312,7 +315,7 @@ func (w *Worker) Start() error { w.tickerWg.Add(2) go func() { defer w.tickerWg.Done() - w.startStatusTicking(w.baseContext, w.sessionManager) + w.startStatusTicking(w.baseContext, w.sessionManager, w.addressReceivers) }() go func() { defer w.tickerWg.Done() @@ -339,9 +342,6 @@ func (w *Worker) Shutdown() error { // Stop listeners first to prevent new connections to the // controller. defer w.started.Store(false) - w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}}) - w.baseCancel() - if err := w.stopServersAndListeners(); err != nil { return fmt.Errorf("error stopping worker servers and listeners: %w", err) } @@ -370,7 +370,9 @@ func (w *Worker) Shutdown() error { // Proceed with remainder of shutdown. w.baseCancel() - w.Resolver().UpdateState(resolver.State{Addresses: []resolver.Address{}}) + for _, ar := range w.addressReceivers { + ar.SetAddresses(nil) + } w.started.Store(false) w.tickerWg.Wait() @@ -383,14 +385,6 @@ func (w *Worker) Shutdown() error { return nil } -func (w *Worker) Resolver() *manual.Resolver { - raw := w.controllerResolver.Load() - if raw == nil { - panic("nil resolver") - } - return raw.(*manual.Resolver) -} - func (w *Worker) ParseAndStoreTags(incoming map[string][]string) { if len(incoming) == 0 { w.tags.Store([]*pb.TagPair{}) @@ -409,40 +403,6 @@ func (w *Worker) ParseAndStoreTags(incoming map[string][]string) { w.updateTags.Store(true) } -// ControllerServerCoordinationServiceConn returns the underlying server coordination service client -func (w *Worker) ControllerServerCoordinationServiceConn() (pbs.ServerCoordinationServiceClient, error) { - rawConn := w.controllerStatusConn.Load() - if rawConn == nil { - return nil, errors.New("unable to load controller service coordination service connection") - } - statusClient, ok := rawConn.(pbs.ServerCoordinationServiceClient) - if !ok { - return nil, fmt.Errorf("invalid service coordination service client %T", rawConn) - } - if statusClient == nil { - return nil, fmt.Errorf("service coordination service client is nil") - } - - return statusClient, nil -} - -// ControllerMultihopConn returns the underlying multihop service client -func (w *Worker) ControllerMultihopConn() (multihop.MultihopServiceClient, error) { - rawConn := w.controllerMultihopConn.Load() - if rawConn == nil { - return nil, errors.New("unable to load controller multihop service connection") - } - multihopClient, ok := rawConn.(multihop.MultihopServiceClient) - if !ok { - return nil, fmt.Errorf("invalid multihop client %T", rawConn) - } - if multihopClient == nil { - return nil, fmt.Errorf("session multihop is nil") - } - - return multihopClient, nil -} - func (w *Worker) getSessionTls(sessionManager session.Manager) func(hello *tls.ClientHelloInfo) (*tls.Config, error) { const op = "worker.(Worker).getSessionTls" return func(hello *tls.ClientHelloInfo) (*tls.Config, error) { @@ -520,7 +480,7 @@ func (w *Worker) getSessionTls(sessionManager session.Manager) func(hello *tls.C // Go will not run this without at least one peer certificate, but // doesn't hurt to check if len(cs.PeerCertificates) == 0 { - return errors.New("no peer certificates provided") + return errors.New(ctx, errors.InvalidParameter, op, "no peer certificates provided") } _, err := cs.PeerCertificates[0].Verify(verifyOpts) return err diff --git a/internal/daemon/worker/worker_test.go b/internal/daemon/worker/worker_test.go index 31e77dbeb..0d249fee4 100644 --- a/internal/daemon/worker/worker_test.go +++ b/internal/daemon/worker/worker_test.go @@ -138,8 +138,6 @@ func TestWorkerNew(t *testing.T) { } func TestSetupWorkerAuthStorage(t *testing.T) { - t.Parallel() - ctx := context.Background() ts := db.TestWrapper(t) diff --git a/internal/tests/cluster/worker_tagging_test.go b/internal/tests/cluster/worker_tagging_test.go index 540f1d2f0..b75ea28b8 100644 --- a/internal/tests/cluster/worker_tagging_test.go +++ b/internal/tests/cluster/worker_tagging_test.go @@ -2,7 +2,6 @@ package cluster import ( "testing" - "time" "github.com/hashicorp/boundary/api/targets" "github.com/hashicorp/boundary/internal/cmd/config" @@ -60,6 +59,7 @@ func TestWorkerTagging(t *testing.T) { }) defer w1.Shutdown() w1Addr := w1.ProxyAddrs()[0] + w1.Worker().WaitForNextSuccessfulStatusUpdate() // Worker 2 conf, err = config.DevWorker() @@ -77,6 +77,7 @@ func TestWorkerTagging(t *testing.T) { }) defer w2.Shutdown() w2Addr := w2.ProxyAddrs()[0] + w2.Worker().WaitForNextSuccessfulStatusUpdate() // Worker 3 conf, err = config.DevWorker() @@ -95,7 +96,8 @@ func TestWorkerTagging(t *testing.T) { defer w3.Shutdown() w3Addr := w3.ProxyAddrs()[0] - time.Sleep(10 * time.Second) + w3.Worker().WaitForNextSuccessfulStatusUpdate() + expectWorkers(t, c1, w1, w2, w3) cases := []struct {