Merge remote-tracking branch 'origin/llb-byow'

internal/daemon/controller/common/common.go

@@ -15,17 +15,18 @@ import (
 )
 
 type (
-	AuthTokenRepoFactory        = oidc.AuthTokenRepoFactory
-	VaultCredentialRepoFactory  = func() (*vault.Repository, error)
-	StaticCredentialRepoFactory = func() (*credstatic.Repository, error)
-	IamRepoFactory              func() (*iam.Repository, error)
-	OidcAuthRepoFactory         = oidc.OidcRepoFactory
-	PasswordAuthRepoFactory     func() (*password.Repository, error)
-	ServersRepoFactory          func() (*servers.Repository, error)
-	StaticRepoFactory           func() (*static.Repository, error)
-	PluginHostRepoFactory       func() (*pluginhost.Repository, error)
-	HostPluginRepoFactory       func() (*hostplugin.Repository, error)
-	SessionRepoFactory          func() (*session.Repository, error)
-	ConnectionRepoFactory       func() (*session.ConnectionRepository, error)
-	TargetRepoFactory           func() (*target.Repository, error)
+	AuthTokenRepoFactory         = oidc.AuthTokenRepoFactory
+	VaultCredentialRepoFactory   = func() (*vault.Repository, error)
+	StaticCredentialRepoFactory  = func() (*credstatic.Repository, error)
+	IamRepoFactory               func() (*iam.Repository, error)
+	OidcAuthRepoFactory          = oidc.OidcRepoFactory
+	PasswordAuthRepoFactory      func() (*password.Repository, error)
+	ServersRepoFactory           func() (*servers.Repository, error)
+	StaticRepoFactory            func() (*static.Repository, error)
+	PluginHostRepoFactory        func() (*pluginhost.Repository, error)
+	HostPluginRepoFactory        func() (*hostplugin.Repository, error)
+	SessionRepoFactory           func() (*session.Repository, error)
+	ConnectionRepoFactory        func() (*session.ConnectionRepository, error)
+	TargetRepoFactory            func() (*target.Repository, error)
+	WorkerAuthRepoStorageFactory func() (*servers.WorkerAuthRepositoryStorage, error)
 )

internal/daemon/controller/controller.go

@@ -28,6 +28,7 @@ import (
 	"github.com/hashicorp/boundary/internal/scheduler"
 	"github.com/hashicorp/boundary/internal/scheduler/job"
 	"github.com/hashicorp/boundary/internal/servers"
+	serversjob "github.com/hashicorp/boundary/internal/servers/job"
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/types/scope"
@@ -65,19 +66,20 @@ type Controller struct {
 	apiGrpcGatewayTicket  string
 
 	// Repo factory methods
-	AuthTokenRepoFn        common.AuthTokenRepoFactory
-	VaultCredentialRepoFn  common.VaultCredentialRepoFactory
-	StaticCredentialRepoFn common.StaticCredentialRepoFactory
-	IamRepoFn              common.IamRepoFactory
-	OidcRepoFn             common.OidcAuthRepoFactory
-	PasswordAuthRepoFn     common.PasswordAuthRepoFactory
-	ServersRepoFn          common.ServersRepoFactory
-	SessionRepoFn          common.SessionRepoFactory
-	ConnectionRepoFn       common.ConnectionRepoFactory
-	StaticHostRepoFn       common.StaticRepoFactory
-	PluginHostRepoFn       common.PluginHostRepoFactory
-	HostPluginRepoFn       common.HostPluginRepoFactory
-	TargetRepoFn           common.TargetRepoFactory
+	AuthTokenRepoFn         common.AuthTokenRepoFactory
+	VaultCredentialRepoFn   common.VaultCredentialRepoFactory
+	StaticCredentialRepoFn  common.StaticCredentialRepoFactory
+	IamRepoFn               common.IamRepoFactory
+	OidcRepoFn              common.OidcAuthRepoFactory
+	PasswordAuthRepoFn      common.PasswordAuthRepoFactory
+	ServersRepoFn           common.ServersRepoFactory
+	SessionRepoFn           common.SessionRepoFactory
+	ConnectionRepoFn        common.ConnectionRepoFactory
+	StaticHostRepoFn        common.StaticRepoFactory
+	PluginHostRepoFn        common.PluginHostRepoFactory
+	HostPluginRepoFn        common.HostPluginRepoFactory
+	TargetRepoFn            common.TargetRepoFactory
+	WorkerAuthRepoStorageFn common.WorkerAuthRepoStorageFactory
 
 	scheduler *scheduler.Scheduler
 
@@ -299,6 +301,10 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	c.ConnectionRepoFn = func() (*session.ConnectionRepository, error) {
 		return session.NewConnectionRepository(ctx, dbase, dbase, c.kms)
 	}
+	c.WorkerAuthRepoStorageFn = func() (*servers.WorkerAuthRepositoryStorage, error) {
+		return servers.NewRepositoryStorage(ctx, dbase, dbase, c.kms)
+	}
+
 	return c, nil
 }
 
@@ -308,7 +314,9 @@ func (c *Controller) Start() error {
 		event.WriteSysEvent(context.TODO(), op, "already started, skipping")
 		return nil
 	}
+
 	c.baseContext, c.baseCancel = context.WithCancel(context.Background())
+
 	if err := c.registerJobs(); err != nil {
 		return fmt.Errorf("error registering jobs: %w", err)
 	}
@@ -316,9 +324,9 @@ func (c *Controller) Start() error {
 		return fmt.Errorf("error starting controller listeners: %w", err)
 	}
 
-	// Upsert server before starting tickers and scheduler to ensure the server exists
-	if err := c.upsertServer(c.baseContext); err != nil {
-		return fmt.Errorf("error upserting server: %w", err)
+	// Upsert controller before starting tickers and scheduler to ensure the controller exists
+	if err := c.upsertController(c.baseContext); err != nil {
+		return fmt.Errorf("error upserting controller: %w", err)
 	}
 	if err := c.scheduler.Start(c.baseContext, c.schedulerWg); err != nil {
 		return fmt.Errorf("error starting scheduler: %w", err)
@@ -360,6 +368,9 @@ func (c *Controller) registerJobs() error {
 	if err := session.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.conf.StatusGracePeriodDuration); err != nil {
 		return err
 	}
+	if err := serversjob.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms); err != nil {
+		return err
+	}
 
 	return nil
 }

internal/daemon/controller/handler.go

@@ -35,6 +35,7 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/sessions"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/users"
+	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
 	"github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
@@ -224,6 +225,13 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error {
 		}
 		services.RegisterCredentialLibraryServiceServer(s, cl)
 	}
+	if _, ok := currentServices[services.WorkerService_ServiceDesc.ServiceName]; !ok {
+		ws, err := workers.NewService(c.baseContext, c.ServersRepoFn, c.IamRepoFn)
+		if err != nil {
+			return fmt.Errorf("failed to create worker handler service: %w", err)
+		}
+		services.RegisterWorkerServiceServer(s, ws)
+	}
 	if _, ok := currentServices[services.CredentialService_ServiceDesc.ServiceName]; !ok {
 		c, err := credentials.NewService(c.StaticCredentialRepoFn, c.IamRepoFn)
 		if err != nil {
@@ -288,6 +296,9 @@ func registerGrpcGatewayEndpoints(ctx context.Context, gwMux *runtime.ServeMux,
 	if err := services.RegisterCredentialLibraryServiceHandlerFromEndpoint(ctx, gwMux, gatewayTarget, dialOptions); err != nil {
 		return fmt.Errorf("failed to register credential library service handler: %w", err)
 	}
+	if err := services.RegisterWorkerServiceHandlerFromEndpoint(ctx, gwMux, gatewayTarget, dialOptions); err != nil {
+		return fmt.Errorf("failed to register worker service handler: %w", err)
+	}
 	if err := services.RegisterCredentialServiceHandlerFromEndpoint(ctx, gwMux, gatewayTarget, dialOptions); err != nil {
 		return fmt.Errorf("failed to register credential service handler: %w", err)
 	}

internal/daemon/controller/handlers/scopes/scope_service.go

@@ -22,6 +22,7 @@ import (
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/sessions"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/users"
+	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
 	"github.com/hashicorp/boundary/internal/errors"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	"github.com/hashicorp/boundary/internal/iam"
@@ -63,6 +64,7 @@ var (
 			resource.Role:       roles.CollectionActions,
 			resource.Scope:      CollectionActions,
 			resource.User:       users.CollectionActions,
+			resource.Worker:     workers.CollectionActions,
 		},
 
 		scope.Org.String(): {

internal/daemon/controller/handlers/scopes/scope_service_test.go

@@ -95,6 +95,12 @@ var globalAuthorizedCollectionActions = map[string]*structpb.ListValue{
 			structpb.NewStringValue("list"),
 		},
 	},
+	"workers": {
+		Values: []*structpb.Value{
+			structpb.NewStringValue("create:worker-led"),
+			structpb.NewStringValue("list"),
+		},
+	},
 }
 
 var orgAuthorizedCollectionActions = map[string]*structpb.ListValue{

internal/daemon/controller/handlers/targets/target_service.go

@@ -799,77 +799,27 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	// worker IDs below is used to contain their IDs in the same order. This is
 	// used to fetch tags for filtering. But we avoid allocation unless we
 	// actually need it.
-	var workers []*pb.WorkerInfo
-	var workerIds []string
-	hasWorkerFilter := len(t.GetWorkerFilter()) > 0
-	servers, err := serversRepo.ListServers(ctx, servers.ServerTypeWorker)
+	selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()})
 	if err != nil {
 		return nil, err
 	}
-	for _, v := range servers {
-		if hasWorkerFilter {
-			workerIds = append(workerIds, v.GetPrivateId())
-		}
-		workers = append(workers, &pb.WorkerInfo{Address: v.Address})
-	}
 
-	if hasWorkerFilter && len(workerIds) > 0 {
-		finalWorkers := make([]*pb.WorkerInfo, 0, len(workers))
-		// Fetch the tags for the given worker IDs
-		tags, err := serversRepo.ListTagsForServers(ctx, workerIds)
-		if err != nil {
-			return nil, err
-		}
-		// Build the map for filtering. This is similar to the filter map we
-		// built from the worker config, but with one extra level: a map of the
-		// worker's ID to its filter map.
-		tagMap := make(map[string]map[string][]string)
-		for _, tag := range tags {
-			currWorkerMap := tagMap[tag.ServerId]
-			if currWorkerMap == nil {
-				currWorkerMap = make(map[string][]string)
-				tagMap[tag.ServerId] = currWorkerMap
-			}
-			currWorkerMap[tag.Key] = append(currWorkerMap[tag.Key], tag.Value)
-			// We don't need to reinsert after the fact because maps are
-			// reference types, so we don't need to re-insert into tagMap
-		}
-
-		// Create the evaluator
+	if len(t.GetWorkerFilter()) > 0 && len(selectedWorkers) > 0 {
 		eval, err := bexpr.CreateEvaluator(t.GetWorkerFilter())
 		if err != nil {
 			return nil, err
 		}
-
-		// Iterate through the known worker IDs, and evaluate. If evaluation
-		// returns true, add to the final worker slice, which is assigned back
-		// to workers after this.
-		for i, worker := range workerIds {
-			filterInput := map[string]interface{}{
-				"name": worker,
-				"tags": tagMap[worker],
-			}
-			ok, err := eval.Evaluate(filterInput)
-			if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
-				return nil, handlers.ApiErrorWithCodeAndMessage(
-					codes.FailedPrecondition,
-					fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
-			}
-			if ok {
-				finalWorkers = append(finalWorkers, workers[i])
-			}
+		selectedWorkers, err = workerList(selectedWorkers).filtered(eval)
+		if err != nil {
+			return nil, err
 		}
-		workers = finalWorkers
 	}
-	if len(workers) == 0 {
+
+	if len(selectedWorkers) == 0 {
 		return nil, handlers.ApiErrorWithCodeAndMessage(
 			codes.FailedPrecondition,
 			"No workers are available to handle this session, or all have been filtered.")
 	}
-	workerAddresses := make([]string, 0, len(workers))
-	for _, worker := range workers {
-		workerAddresses = append(workerAddresses, worker.GetAddress())
-	}
 
 	requestedId := req.GetHostId()
 	staticHostRepo, err := s.staticHostRepoFn()
@@ -992,7 +942,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	if err != nil {
 		return nil, err
 	}
-	sess, privKey, err := sessionRepo.CreateSession(ctx, wrapper, sess, workerAddresses)
+	sess, privKey, err := sessionRepo.CreateSession(ctx, wrapper, sess, workerList(selectedWorkers).addresses())
 	if err != nil {
 		return nil, err
 	}
@@ -1091,7 +1041,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		PrivateKey:      privKey,
 		HostId:          chosenEndpoint.HostId,
 		Endpoint:        endpointUrl.String(),
-		WorkerInfo:      workers,
+		WorkerInfo:      workerList(selectedWorkers).workerInfos(),
 		ConnectionLimit: t.GetSessionConnectionLimit(),
 	}
 	marshaledSad, err := proto.Marshal(sad)
@@ -1966,3 +1916,46 @@ func validateAuthorizeSessionRequest(req *pbs.AuthorizeSessionRequest) error {
 	}
 	return nil
 }
+
+// workerList is a helper type to make the selection of workers clearer and more declarative.
+type workerList []*servers.Worker
+
+// addresses converts the slice of workers to a slice of their addresses
+func (w workerList) addresses() []string {
+	ret := make([]string, 0, len(w))
+	for _, worker := range w {
+		ret = append(ret, worker.GetAddress())
+	}
+	return ret
+}
+
+// workerInfos converts the slice of workers to a slice of their workerInfo protos
+func (w workerList) workerInfos() []*pb.WorkerInfo {
+	ret := make([]*pb.WorkerInfo, 0, len(w))
+	for _, worker := range w {
+		ret = append(ret, &pb.WorkerInfo{Address: worker.GetAddress()})
+	}
+	return ret
+}
+
+// filtered returns a new workerList where all elements contained in it are the
+// ones which from the original workerList that pass the evaluator's evaluation.
+func (w workerList) filtered(eval *bexpr.Evaluator) (workerList, error) {
+	var ret []*servers.Worker
+	for _, worker := range w {
+		filterInput := map[string]interface{}{
+			"name": worker.GetName(),
+			"tags": worker.CanonicalTags(),
+		}
+		ok, err := eval.Evaluate(filterInput)
+		if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
+			return nil, handlers.ApiErrorWithCodeAndMessage(
+				codes.FailedPrecondition,
+				fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
+		}
+		if ok {
+			ret = append(ret, worker)
+		}
+	}
+	return ret, nil
+}

internal/daemon/controller/handlers/targets/target_service_test.go

@@ -0,0 +1,79 @@
+package targets
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/types/scope"
+	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
+	"github.com/hashicorp/go-bexpr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerList_Addresses(t *testing.T) {
+	addresses := []string{
+		"test1",
+		"test2",
+		"test3",
+		"test4",
+	}
+	var workerInfos []*pb.WorkerInfo
+	var tested workerList
+	for _, a := range addresses {
+		workerInfos = append(workerInfos, &pb.WorkerInfo{Address: a})
+		tested = append(tested, servers.NewWorker(scope.Global.String(),
+			servers.WithName(a),
+			servers.WithAddress(a)))
+	}
+	assert.Equal(t, addresses, tested.addresses())
+	assert.Equal(t, workerInfos, tested.workerInfos())
+}
+
+func TestWorkerList_Filter(t *testing.T) {
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	var workers []*servers.Worker
+	for i := 0; i < 5; i++ {
+		switch {
+		case i%2 == 0:
+			workers = append(workers, servers.TestKmsWorker(t, conn, wrapper,
+				servers.WithName(fmt.Sprintf("test_worker_%d", i)),
+				servers.WithWorkerTags(&servers.Tag{
+					Key:   fmt.Sprintf("key%d", i),
+					Value: fmt.Sprintf("value%d", i),
+				})))
+		default:
+			workers = append(workers, servers.TestPkiWorker(t, conn, wrapper,
+				servers.WithName(fmt.Sprintf("test_worker_%d", i)),
+				servers.WithWorkerTags(&servers.Tag{
+					Key:   "key",
+					Value: "configvalue",
+				})))
+		}
+	}
+
+	{
+		ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[13]" and "configvalue2" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 0)
+	}
+	{
+		ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[12]" and "configvalue" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 1)
+	}
+	{
+		ev, err := bexpr.CreateEvaluator(`"configvalue" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 2)
+	}
+}

internal/daemon/controller/handlers/targets/tcp/target_service_test.go

@@ -16,12 +16,12 @@ import (
 	"github.com/hashicorp/boundary/internal/credential"
 	credstatic "github.com/hashicorp/boundary/internal/credential/static"
 	"github.com/hashicorp/boundary/internal/credential/vault"
+	cluster "github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/auth"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/credentiallibraries"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/credentials"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
 	"github.com/hashicorp/boundary/internal/db"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
@@ -2371,9 +2371,6 @@ func TestAuthorizeSession(t *testing.T) {
 	sessionRepoFn := func() (*session.Repository, error) {
 		return session.NewRepository(rw, rw, kms)
 	}
-	connectionRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
 	staticRepo, err := static.NewRepository(rw, rw, kms)
 	require.NoError(t, err)
 	staticHostRepoFn := func() (*static.Repository, error) {
@@ -2532,14 +2529,7 @@ func TestAuthorizeSession(t *testing.T) {
 			require.NoError(t, err)
 
 			// Tell our DB that there is a worker ready to serve the data
-			workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
-			_, err = workerService.Status(ctx, &spbs.StatusRequest{
-				Worker: &spb.Server{
-					PrivateId: "testworker",
-					Address:   "localhost:8457",
-				},
-			})
-			require.NoError(t, err)
+			servers.TestKmsWorker(t, conn, wrapper)
 
 			asRes1, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{
 				Id: tar.GetPublicId(),
@@ -2641,9 +2631,6 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 	sessionRepoFn := func() (*session.Repository, error) {
 		return session.NewRepository(rw, rw, kms)
 	}
-	connectionRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
 	staticHostRepoFn := func() (*static.Repository, error) {
 		return static.NewRepository(rw, rw, kms)
 	}
@@ -2887,14 +2874,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 			require.NoError(t, err)
 
 			// Tell our DB that there is a worker ready to serve the data
-			workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
-			_, err = workerService.Status(ctx, &spbs.StatusRequest{
-				Worker: &spb.Server{
-					PrivateId: "testworker",
-					Address:   "localhost:8457",
-				},
-			})
-			require.NoError(t, err)
+			servers.TestKmsWorker(t, conn, wrapper)
 
 			asRes, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{
 				Id: tar.GetPublicId(),
@@ -3002,11 +2982,12 @@ func TestAuthorizeSession_Errors(t *testing.T) {
 	store := vault.TestCredentialStore(t, conn, wrapper, proj.GetPublicId(), v.Addr, tok, sec.Auth.Accessor)
 
 	workerExists := func(tar target.Target) (version uint32) {
-		workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
+		workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
 		_, err := workerService.Status(context.Background(), &spbs.StatusRequest{
-			Worker: &spb.Server{
-				PrivateId: "testworker",
-				Address:   "localhost:123",
+			WorkerStatus: &spb.ServerWorkerStatus{
+				PublicId: "w_1234567890",
+				Name:     "w_1234567890",
+				Address:  "localhost:123",
 			},
 		})
 		require.NoError(t, err)

internal/daemon/controller/handlers/workers/worker_service.go

@@ -2,432 +2,616 @@ package workers
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"sync"
-	"time"
+	"strings"
 
+	"github.com/hashicorp/boundary/globals"
+	"github.com/hashicorp/boundary/internal/daemon/controller/auth"
 	"github.com/hashicorp/boundary/internal/daemon/controller/common"
+	"github.com/hashicorp/boundary/internal/daemon/controller/common/scopeids"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/kms"
-	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/errors"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
+	"github.com/hashicorp/boundary/internal/perms"
+	"github.com/hashicorp/boundary/internal/requests"
 	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/session"
+	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/action"
 	"github.com/hashicorp/boundary/internal/types/resource"
-	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
-	"github.com/hashicorp/go-bexpr"
+	"github.com/hashicorp/boundary/internal/types/scope"
+	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/workers"
+	"github.com/hashicorp/go-secure-stdlib/strutil"
+	"github.com/hashicorp/nodeenrollment/types"
+	"github.com/mr-tron/base58"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
-type workerServiceServer struct {
-	pbs.UnimplementedServerCoordinationServiceServer
-	pbs.UnimplementedSessionServiceServer
-
-	serversRepoFn    common.ServersRepoFactory
-	sessionRepoFn    common.SessionRepoFactory
-	connectionRepoFn common.ConnectionRepoFactory
-	updateTimes      *sync.Map
-	kms              *kms.Kms
-}
-
-func NewWorkerServiceServer(
-	serversRepoFn common.ServersRepoFactory,
-	sessionRepoFn common.SessionRepoFactory,
-	connectionRepoFn common.ConnectionRepoFactory,
-	updateTimes *sync.Map,
-	kms *kms.Kms,
-) *workerServiceServer {
-	return &workerServiceServer{
-		serversRepoFn:    serversRepoFn,
-		sessionRepoFn:    sessionRepoFn,
-		connectionRepoFn: connectionRepoFn,
-		updateTimes:      updateTimes,
-		kms:              kms,
-	}
-}
+const (
+	PkiWorkerType = "pki"
+	KmsWorkerType = "kms"
+)
 
 var (
-	_ pbs.SessionServiceServer            = &workerServiceServer{}
-	_ pbs.ServerCoordinationServiceServer = &workerServiceServer{}
+	maskManager handlers.MaskManager
+
+	// IdActions contains the set of actions that can be performed on
+	// individual resources
+	IdActions = action.ActionSet{
+		action.NoOp,
+		action.Read,
+		action.Update,
+		action.Delete,
+	}
+
+	// CollectionActions contains the set of actions that can be performed on
+	// this collection
+	CollectionActions = action.ActionSet{
+		action.CreateWorkerLed,
+		action.List,
+	}
 )
 
-func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusRequest) (*pbs.StatusResponse, error) {
-	const op = "workers.(workerServiceServer).Status"
-	// TODO: on the worker, if we get errors back from this repeatedly, do we
-	// terminate all sessions since we can't know if they were canceled?
-	ws.updateTimes.Store(req.Worker.PrivateId, time.Now())
-	serverRepo, err := ws.serversRepoFn()
+func init() {
+	var err error
+	if maskManager, err = handlers.NewMaskManager(handlers.MaskDestination{&store.Worker{}}, handlers.MaskSource{&pb.Worker{}}); err != nil {
+		panic(err)
+	}
+}
+
+// Service handles request as described by the pbs.WorkerServiceServer interface.
+type Service struct {
+	pbs.UnimplementedWorkerServiceServer
+
+	repoFn    common.ServersRepoFactory
+	iamRepoFn common.IamRepoFactory
+}
+
+// NewService returns a worker service which handles worker related requests to boundary.
+func NewService(ctx context.Context, repo common.ServersRepoFactory, iamRepoFn common.IamRepoFactory) (Service, error) {
+	const op = "workers.NewService"
+	if repo == nil {
+		return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing servers repository")
+	}
+	if iamRepoFn == nil {
+		return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing iam repository")
+	}
+	return Service{repoFn: repo, iamRepoFn: iamRepoFn}, nil
+}
+
+var _ pbs.WorkerServiceServer = Service{}
+
+// ListWorkers implements the interface pbs.WorkerServiceServer.
+func (s Service) ListWorkers(ctx context.Context, req *pbs.ListWorkersRequest) (*pbs.ListWorkersResponse, error) {
+	if err := validateListRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetScopeId(), action.List)
+	if authResults.Error != nil {
+		// If it's forbidden, and it's a recursive request, and they're
+		// successfully authenticated but just not authorized, keep going as we
+		// may have authorization on downstream scopes. Or, if they've not
+		// authenticated, still process in case u_anon has permissions.
+		if (authResults.Error == handlers.ForbiddenError() || authResults.Error == handlers.UnauthenticatedError()) &&
+			req.GetRecursive() &&
+			authResults.AuthenticationFinished {
+		} else {
+			return nil, authResults.Error
+		}
+	}
+
+	scopeIds, scopeInfoMap, err := scopeids.GetListingScopeIds(
+		ctx, s.iamRepoFn, authResults, req.GetScopeId(), resource.Worker, req.GetRecursive())
 	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to store worker status: %v", err)
+		return nil, err
 	}
-	sessRepo, err := ws.sessionRepoFn()
-	connectionRepo, err := ws.connectionRepoFn()
+	// If no scopes match, return an empty response
+	if len(scopeIds) == 0 {
+		return &pbs.ListWorkersResponse{}, nil
+	}
+
+	ul, err := s.listFromRepo(ctx, scopeIds)
 	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting sessions repo"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err)
+		return nil, err
 	}
-	req.Worker.Type = resource.Worker.String()
-	controllers, _, err := serverRepo.UpsertServer(ctx, req.Worker, servers.WithUpdateTags(req.GetUpdateTags()))
+	if len(ul) == 0 {
+		return &pbs.ListWorkersResponse{}, nil
+	}
+
+	filter, err := handlers.NewFilter(req.GetFilter())
 	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err)
+		return nil, err
 	}
-	ret := &pbs.StatusResponse{
-		Controllers: controllers,
+	finalItems := make([]*pb.Worker, 0, len(ul))
+	res := perms.Resource{
+		Type: resource.Worker,
+	}
+	for _, item := range ul {
+		res.Id = item.GetPublicId()
+		res.ScopeId = item.GetScopeId()
+		authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings()
+		if len(authorizedActions) == 0 {
+			continue
+		}
+
+		outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId)
+		outputOpts := make([]handlers.Option, 0, 3)
+		outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+		if outputFields.Has(globals.ScopeField) {
+			outputOpts = append(outputOpts, handlers.WithScope(scopeInfoMap[item.GetScopeId()]))
+		}
+		if outputFields.Has(globals.AuthorizedActionsField) {
+			outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions))
+		}
+
+		item, err := toProto(ctx, item, outputOpts...)
+		if err != nil {
+			return nil, err
+		}
+
+		if filter.Match(item) {
+			finalItems = append(finalItems, item)
+		}
+	}
+	return &pbs.ListWorkersResponse{Items: finalItems}, nil
+}
+
+// GetWorker implements the interface pbs.WorkerServiceServer.
+func (s Service) GetWorker(ctx context.Context, req *pbs.GetWorkerRequest) (*pbs.GetWorkerResponse, error) {
+	const op = "workers.(Service).GetWorker"
+
+	if err := validateGetRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.Read)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.getFromRepo(ctx, req.GetId())
+	if err != nil {
+		return nil, err
 	}
 
-	stateReport := make([]session.StateReport, 0, len(req.GetJobs()))
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
 
-	for _, jobStatus := range req.GetJobs() {
-		switch jobStatus.Job.GetType() {
-		case pbs.JOBTYPE_JOBTYPE_SESSION:
-			si := jobStatus.GetJob().GetSessionInfo()
-			if si == nil {
-				return nil, status.Error(codes.Internal, "Error getting session info at status time")
-			}
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
+	}
 
-			switch si.Status {
-			case pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
-				pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED:
-				// No need to see about canceling anything
-				continue
-			}
+	item, err := toProto(ctx, w, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
 
-			sr := session.StateReport{
-				SessionId:     si.GetSessionId(),
-				ConnectionIds: make([]string, 0, len(si.GetConnections())),
-			}
-			for _, conn := range si.GetConnections() {
-				switch conn.Status {
-				case pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_AUTHORIZED,
-					pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED:
-					sr.ConnectionIds = append(sr.ConnectionIds, conn.GetConnectionId())
+	return &pbs.GetWorkerResponse{Item: item}, nil
+}
+
+// CreateWorker implements the interface pbs.WorkerServiceServer.
+func (s Service) CreateWorkerLed(ctx context.Context, req *pbs.CreateWorkerLedRequest) (*pbs.CreateWorkerLedResponse, error) {
+	const op = "workers.(Service).CreateWorkerLed"
+
+	if err := validateCreateRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetItem().GetScopeId(), action.CreateWorkerLed)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	reqBytes, err := base58.FastBase58Decoding(req.GetItem().WorkerGeneratedAuthToken.GetValue())
+	if err != nil {
+		return nil, fmt.Errorf("%s: error decoding node_credentials_token: %w", op, err)
+	}
+	// Decode the proto into the request
+	creds := new(types.FetchNodeCredentialsRequest)
+	if err := proto.Unmarshal(reqBytes, creds); err != nil {
+		return nil, fmt.Errorf("%s: error unmarshaling node_credentials_token: %w", op, err)
+	}
+	created, err := s.createInRepo(ctx, req.GetItem(), creds)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error creating worker: %w", op, err)
+	}
+
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
+
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, created.GetPublicId(), IdActions).Strings()))
+	}
+
+	item, err := toProto(ctx, created, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbs.CreateWorkerLedResponse{Item: item}, nil
+}
+
+// DeleteWorker implements the interface pbs.WorkerServiceServer.
+func (s Service) DeleteWorker(ctx context.Context, req *pbs.DeleteWorkerRequest) (*pbs.DeleteWorkerResponse, error) {
+	if err := validateDeleteRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.Delete)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	_, err := s.deleteFromRepo(ctx, req.GetId())
+	if err != nil {
+		return nil, err
+	}
+	return nil, nil
+}
+
+// UpdateWorker implements the interface pbs.WorkerServiceServer.
+func (s Service) UpdateWorker(ctx context.Context, req *pbs.UpdateWorkerRequest) (*pbs.UpdateWorkerResponse, error) {
+	const op = "workers.(Service).UpdateWorker"
+
+	if err := validateUpdateRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.Update)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.updateInRepo(ctx, authResults.Scope.GetId(), req.GetId(), req.GetUpdateMask().GetPaths(), req.GetItem())
+	if err != nil {
+		return nil, err
+	}
+
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
+
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
+	}
+
+	item, err := toProto(ctx, w, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbs.UpdateWorkerResponse{Item: item}, nil
+}
+
+func (s Service) listFromRepo(ctx context.Context, scopeIds []string) ([]*servers.Worker, error) {
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+	wl, err := repo.ListWorkers(ctx, scopeIds, servers.WithLiveness(-1))
+	if err != nil {
+		return nil, err
+	}
+	return wl, nil
+}
+
+func (s Service) getFromRepo(ctx context.Context, id string) (*servers.Worker, error) {
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+	w, err := repo.LookupWorker(ctx, id)
+	if err != nil {
+		if errors.IsNotFoundError(err) {
+			return nil, handlers.NotFoundErrorf("Worker %q doesn't exist.", id)
+		}
+		return nil, err
+	}
+	if w == nil {
+		return nil, handlers.NotFoundErrorf("Worker %q doesn't exist.", id)
+	}
+	return w, nil
+}
+
+func (s Service) createInRepo(ctx context.Context, worker *pb.Worker, creds *types.FetchNodeCredentialsRequest) (*servers.Worker, error) {
+	const op = "workers.(Service).createInRepo"
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+	newWorker := servers.NewWorker(
+		worker.GetScopeId(),
+		servers.WithName(worker.GetName().GetValue()),
+		servers.WithDescription(worker.GetDescription().GetValue()),
+	)
+	retWorker, err := repo.CreateWorker(ctx, newWorker, servers.WithFetchNodeCredentialsRequest(creds))
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to create worker"))
+	}
+	return retWorker, nil
+}
+
+func (s Service) deleteFromRepo(ctx context.Context, id string) (bool, error) {
+	const op = "workers.(Service).deleteFromRepo"
+	repo, err := s.repoFn()
+	if err != nil {
+		return false, err
+	}
+	rows, err := repo.DeleteWorker(ctx, id)
+	if err != nil {
+		if errors.IsNotFoundError(err) {
+			return false, nil
+		}
+		return false, errors.Wrap(ctx, err, op, errors.WithMsg("unable to delete worker"))
+	}
+	return rows > 0, nil
+}
+
+func (s Service) updateInRepo(ctx context.Context, scopeId, id string, mask []string, item *pb.Worker) (*servers.Worker, error) {
+	const op = "workers.(Service).updateInRepo"
+	var opts []servers.Option
+	if desc := item.GetDescription(); desc != nil {
+		opts = append(opts, servers.WithDescription(desc.GetValue()))
+	}
+	if name := item.GetName(); name != nil {
+		opts = append(opts, servers.WithName(name.GetValue()))
+	}
+	w := servers.NewWorker(scopeId, opts...)
+	w.PublicId = id
+	dbMask := maskManager.Translate(mask)
+	if len(dbMask) == 0 {
+		return nil, handlers.InvalidArgumentErrorf("No valid fields included in the update mask.", map[string]string{"update_mask": "No valid fields provided in the update mask."})
+	}
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+	out, rowsUpdated, err := repo.UpdateWorker(ctx, w, item.GetVersion(), dbMask)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to update worker"))
+	}
+	if rowsUpdated == 0 {
+		return nil, handlers.NotFoundErrorf("Worker %q doesn't exist or incorrect version provided.", id)
+	}
+	return out, nil
+}
+
+func (s Service) authResult(ctx context.Context, id string, a action.Type) auth.VerifyResults {
+	res := auth.VerifyResults{}
+	repo, err := s.repoFn()
+	if err != nil {
+		res.Error = err
+		return res
+	}
+
+	var parentId string
+	opts := []auth.Option{auth.WithType(resource.Worker), auth.WithAction(a)}
+	switch a {
+	case action.List, action.CreateWorkerLed:
+		parentId = id
+	default:
+		w, err := repo.LookupWorker(ctx, id)
+		if err != nil {
+			res.Error = err
+			return res
+		}
+		if w == nil {
+			res.Error = handlers.NotFoundError()
+			return res
+		}
+		parentId = w.GetScopeId()
+		opts = append(opts, auth.WithId(id))
+	}
+	opts = append(opts, auth.WithScopeId(parentId))
+	return auth.Verify(ctx, opts...)
+}
+
+func toProto(ctx context.Context, in *servers.Worker, opt ...handlers.Option) (*pb.Worker, error) {
+	const op = "workers.toProto"
+	opts := handlers.GetOpts(opt...)
+	if opts.WithOutputFields == nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "output fields not found when building worker proto")
+	}
+	outputFields := *opts.WithOutputFields
+
+	out := pb.Worker{}
+	if outputFields.Has(globals.IdField) {
+		out.Id = in.GetPublicId()
+	}
+	if outputFields.Has(globals.ScopeIdField) {
+		out.ScopeId = in.GetScopeId()
+	}
+	if outputFields.Has(globals.DescriptionField) && in.GetDescription() != "" {
+		out.Description = wrapperspb.String(in.GetDescription())
+	}
+	if outputFields.Has(globals.NameField) && in.GetName() != "" {
+		out.Name = wrapperspb.String(in.GetName())
+	}
+	if outputFields.Has(globals.CreatedTimeField) {
+		out.CreatedTime = in.GetCreateTime().GetTimestamp()
+	}
+	if outputFields.Has(globals.UpdatedTimeField) {
+		out.UpdatedTime = in.GetUpdateTime().GetTimestamp()
+	}
+	if outputFields.Has(globals.VersionField) {
+		out.Version = in.GetVersion()
+	}
+	if outputFields.Has(globals.ScopeField) {
+		out.Scope = opts.WithScope
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		out.AuthorizedActions = opts.WithAuthorizedActions
+		if in.Type == KmsWorkerType && out.AuthorizedActions != nil {
+			// KMS workers cannot be updated through the API
+			allActions := out.AuthorizedActions
+			out.AuthorizedActions = make([]string, 0, len(allActions))
+			for _, act := range allActions {
+				if act != action.Update.String() {
+					out.AuthorizedActions = append(out.AuthorizedActions, act)
 				}
 			}
-			stateReport = append(stateReport, sr)
+		}
+	}
+	if outputFields.Has(globals.AddressField) && in.GetAddress() != "" {
+		out.Address = in.GetAddress()
+	}
+	if outputFields.Has(globals.TypeField) && in.GetType() != "" {
+		out.Type = in.GetType()
+	}
+	if outputFields.Has(globals.LastStatusTimeField) {
+		out.LastStatusTime = in.GetLastStatusTime().GetTimestamp()
+	}
+	if outputFields.Has(globals.ActiveConnectionCountField) {
+		out.ActiveConnectionCount = &wrapperspb.UInt32Value{Value: in.ActiveConnectionCount()}
+	}
+	if outputFields.Has(globals.ConfigTagsField) && len(in.GetConfigTags()) > 0 {
+		var err error
+		out.ConfigTags, err = tagsToMapProto(in.GetConfigTags())
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing config tags proto"))
+		}
+	}
+	if outputFields.Has(globals.CanonicalTagsField) && len(in.CanonicalTags()) > 0 {
+		var err error
+		out.CanonicalTags, err = tagsToMapProto(in.CanonicalTags())
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing canonical tags proto"))
 		}
 	}
 
-	notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, req.Worker.PrivateId, stateReport)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error comparing state of sessions for worker: %s: %v", req.Worker.PrivateId, err)
-	}
-	for _, na := range notActive {
-		var connChanges []*pbs.Connection
-		for _, connId := range na.ConnectionIds {
-			connChanges = append(connChanges, &pbs.Connection{
-				ConnectionId: connId,
-				Status:       session.StatusClosed.ProtoVal(),
-			})
-		}
-		ret.JobsRequests = append(ret.JobsRequests, &pbs.JobChangeRequest{
-			Job: &pbs.Job{
-				Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-				JobInfo: &pbs.Job_SessionInfo{
-					SessionInfo: &pbs.SessionJobInfo{
-						SessionId:   na.SessionId,
-						Status:      na.Status.ProtoVal(),
-						Connections: connChanges,
-					},
-				},
-			},
-			RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE,
-		})
-	}
+	return &out, nil
+}
 
+func tagsToMapProto(in map[string][]string) (map[string]*structpb.ListValue, error) {
+	b := make(map[string][]interface{})
+	for k, v := range in {
+		result := make([]interface{}, 0, len(v))
+		for _, t := range v {
+			result = append(result, t)
+		}
+		b[k] = result
+	}
+	ret := make(map[string]*structpb.ListValue)
+	var err error
+	for k, v := range b {
+		ret[k], err = structpb.NewList(v)
+		if err != nil {
+			return nil, err
+		}
+	}
 	return ret, nil
 }
 
-func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.LookupSessionRequest) (*pbs.LookupSessionResponse, error) {
-	const op = "workers.(workerServiceServer).LookupSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error getting session repo: %v", err)
-	}
-
-	sessionInfo, authzSummary, err := sessRepo.LookupSession(ctx, req.GetSessionId())
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error looking up session: %v", err)
-	}
-	if sessionInfo == nil {
-		return nil, status.Error(codes.PermissionDenied, "Unknown session ID.")
-	}
-	if len(sessionInfo.States) == 0 {
-		return nil, status.Error(codes.Internal, "Empty session states during lookup.")
-	}
-
-	if sessionInfo.WorkerFilter != "" {
-		if req.ServerId == "" {
-			event.WriteError(ctx, op, errors.New("worker filter enabled for session but got no server ID from worker"))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Did not receive server ID when looking up session but filtering is enabled: %v", err)
-		}
-		serversRepo, err := ws.serversRepoFn()
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo"))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err)
-		}
-		tags, err := serversRepo.ListTagsForServers(ctx, []string{req.ServerId})
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up tags for server", "server_id", req.ServerId))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err)
-		}
-		// Build the map for filtering.
-		tagMap := make(map[string][]string)
-		for _, tag := range tags {
-			tagMap[tag.Key] = append(tagMap[tag.Key], tag.Value)
-			// We don't need to reinsert after the fact because maps are
-			// reference types, so we don't need to re-insert into tagMap
-		}
-
-		// Create the evaluator
-		eval, err := bexpr.CreateEvaluator(sessionInfo.WorkerFilter)
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error creating worker filter evaluator", "server_id", req.ServerId))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error creating worker filter evaluator: %v", err)
-		}
-		filterInput := map[string]interface{}{
-			"name": req.ServerId,
-			"tags": tagMap,
-		}
-		ok, err := eval.Evaluate(filterInput)
-		if err != nil {
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-				fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
-		}
-		if !ok {
-			return nil, handlers.ApiErrorWithCodeAndMessage(
-				codes.FailedPrecondition,
-				"Worker filter expression precludes this worker from serving this session")
-		}
-	}
-
-	creds, err := sessRepo.ListSessionCredentials(ctx, sessionInfo.ScopeId, sessionInfo.PublicId)
-	if err != nil {
-		return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-			fmt.Sprintf("Error retrieving session credentials: %s", err))
-	}
-	var workerCreds []*pbs.Credential
-	for _, c := range creds {
-		m := &pbs.Credential{}
-		err = proto.Unmarshal(c, m)
-		if err != nil {
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-				fmt.Sprintf("Error unmarshaling credentials: %s", err))
-		}
-		workerCreds = append(workerCreds, m)
-	}
-
-	resp := &pbs.LookupSessionResponse{
-		Authorization: &targets.SessionAuthorizationData{
-			SessionId:   sessionInfo.GetPublicId(),
-			Certificate: sessionInfo.Certificate,
-		},
-		Status:          sessionInfo.States[0].Status.ProtoVal(),
-		Version:         sessionInfo.Version,
-		TofuToken:       string(sessionInfo.TofuToken),
-		Endpoint:        sessionInfo.Endpoint,
-		Expiration:      sessionInfo.ExpirationTime.Timestamp,
-		ConnectionLimit: sessionInfo.ConnectionLimit,
-		ConnectionsLeft: authzSummary.ConnectionLimit,
-		HostId:          sessionInfo.HostId,
-		HostSetId:       sessionInfo.HostSetId,
-		TargetId:        sessionInfo.TargetId,
-		UserId:          sessionInfo.UserId,
-		Credentials:     workerCreds,
-	}
-	if resp.ConnectionsLeft != -1 {
-		resp.ConnectionsLeft -= int32(authzSummary.CurrentConnectionCount)
-	}
-
-	wrapper, err := ws.kms.GetWrapper(ctx, sessionInfo.ScopeId, kms.KeyPurposeSessions, kms.WithKeyId(sessionInfo.KeyId))
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error getting sessions wrapper: %v", err)
-	}
-
-	// Derive the private key, which should match. Deriving on both ends allows
-	// us to not store it in the DB.
-	_, resp.Authorization.PrivateKey, err = session.DeriveED25519Key(ctx, wrapper, sessionInfo.UserId, sessionInfo.GetPublicId())
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error deriving session key: %v", err)
-	}
-
-	return resp, nil
+// A validateX method should exist for each method above.  These methods do not make calls to any backing service but enforce
+// requirements on the structure of the request.  They verify that:
+//  * The path passed in is correctly formatted
+//  * All required parameters are set
+//  * There are no conflicting parameters provided
+func validateGetRequest(req *pbs.GetWorkerRequest) error {
+	return handlers.ValidateGetRequest(handlers.NoopValidatorFn, req, servers.WorkerPrefix)
 }
 
-func (ws *workerServiceServer) CancelSession(ctx context.Context, req *pbs.CancelSessionRequest) (*pbs.CancelSessionResponse, error) {
-	const op = "workers.(workerServiceServer).CancelSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
+func validateListRequest(req *pbs.ListWorkersRequest) error {
+	badFields := map[string]string{}
+	if req.GetScopeId() != scope.Global.String() {
+		badFields["scope_id"] = "Must be 'global' when listing."
 	}
-
-	ses, _, err := sessRepo.LookupSession(ctx, req.GetSessionId())
-	if err != nil {
-		return nil, err
+	if _, err := handlers.NewFilter(req.GetFilter()); err != nil {
+		badFields["filter"] = fmt.Sprintf("This field could not be parsed. %v", err)
 	}
-
-	ses, err = sessRepo.CancelSession(ctx, req.GetSessionId(), ses.Version)
-	if err != nil {
-		return nil, err
+	if len(badFields) > 0 {
+		return handlers.InvalidArgumentErrorf("Error in provided request.", badFields)
 	}
-
-	return &pbs.CancelSessionResponse{
-		Status: ses.States[0].Status.ProtoVal(),
-	}, nil
+	return nil
 }
 
-func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.ActivateSessionRequest) (*pbs.ActivateSessionResponse, error) {
-	const op = "workers.(workerServiceServer).ActivateSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	sessionInfo, sessionStates, err := sessRepo.ActivateSession(
-		ctx,
-		req.GetSessionId(),
-		req.GetVersion(),
-		req.GetWorkerId(),
-		resource.Worker.String(),
-		[]byte(req.GetTofuToken()))
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error looking up session: %v", err)
-	}
-	if sessionInfo == nil {
-		return nil, status.Error(codes.PermissionDenied, "Unknown session ID.")
-	}
-	if len(sessionStates) == 0 {
-		return nil, status.Error(codes.Internal, "Invalid session state in activate response.")
-	}
-
-	return &pbs.ActivateSessionResponse{
-		Status: sessionStates[0].Status.ProtoVal(),
-	}, nil
+func validateDeleteRequest(req *pbs.DeleteWorkerRequest) error {
+	return handlers.ValidateDeleteRequest(handlers.NoopValidatorFn, req, servers.WorkerPrefix)
 }
 
-func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs.AuthorizeConnectionRequest) (*pbs.AuthorizeConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).AuthorizeConnection"
-	connectionRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	sessionRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	connectionInfo, connStates, authzSummary, err := session.AuthorizeConnection(ctx, sessionRepo, connectionRepo, req.GetSessionId(), req.GetWorkerId())
-	if err != nil {
-		return nil, err
-	}
-	if connectionInfo == nil {
-		return nil, status.Error(codes.Internal, "Invalid authorize connection response.")
-	}
-	if len(connStates) == 0 {
-		return nil, status.Error(codes.Internal, "Invalid connection state in authorize response.")
-	}
-
-	ret := &pbs.AuthorizeConnectionResponse{
-		ConnectionId:    connectionInfo.GetPublicId(),
-		Status:          connStates[0].Status.ProtoVal(),
-		ConnectionsLeft: authzSummary.ConnectionLimit,
-	}
-	if ret.ConnectionsLeft != -1 {
-		ret.ConnectionsLeft -= int32(authzSummary.CurrentConnectionCount)
-	}
-
-	return ret, nil
+func validateUpdateRequest(req *pbs.UpdateWorkerRequest) error {
+	return handlers.ValidateUpdateRequest(req, req.GetItem(), func() map[string]string {
+		badFields := map[string]string{}
+		if req.GetItem().GetAddress() != "" {
+			badFields[globals.AddressField] = "This is a read only field."
+		}
+		if req.GetItem().GetCanonicalTags() != nil {
+			badFields[globals.CanonicalAddressField] = "This is a read only field."
+		}
+		if req.GetItem().GetConfigTags() != nil {
+			badFields[globals.TagsField] = "This is a read only field."
+		}
+		nameString := req.GetItem().GetName().String()
+		if !strutil.Printable(nameString) {
+			badFields[globals.NameField] = "Contains non-printable characters"
+		}
+		if strings.ToLower(nameString) != nameString {
+			badFields[globals.NameField] = "Must be all lowercase."
+		}
+		descriptionString := req.GetItem().GetDescription().String()
+		if !strutil.Printable(descriptionString) {
+			badFields[globals.DescriptionField] = "Contains non-printable characters."
+		}
+		return badFields
+	}, servers.WorkerPrefix)
 }
 
-func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (*pbs.ConnectConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).ConnectConnection"
-	connRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	connectionInfo, connStates, err := connRepo.ConnectConnection(ctx, session.ConnectWith{
-		ConnectionId:       req.GetConnectionId(),
-		ClientTcpAddress:   req.GetClientTcpAddress(),
-		ClientTcpPort:      req.GetClientTcpPort(),
-		EndpointTcpAddress: req.GetEndpointTcpAddress(),
-		EndpointTcpPort:    req.GetEndpointTcpPort(),
-		UserClientIp:       req.GetUserClientIp(),
+func validateCreateRequest(req *pbs.CreateWorkerLedRequest) error {
+	return handlers.ValidateCreateRequest(req.GetItem(), func() map[string]string {
+		const (
+			mustBeGlobalMsg  = "Must be 'global'"
+			cannotBeEmptyMsg = "Cannot be empty."
+			readOnlyFieldMsg = "This is a read only field."
+		)
+		badFields := map[string]string{}
+		if scope.Global.String() != req.GetItem().GetScopeId() {
+			badFields[globals.ScopeIdField] = mustBeGlobalMsg
+		}
+		// FIXME: in the future, we won't require this token since we'll support
+		// the server led flow where a token is returned when a worker is created.
+		if req.GetItem().WorkerGeneratedAuthToken == nil {
+			badFields[globals.WorkerGeneratedAuthTokenField] = cannotBeEmptyMsg
+		}
+		if req.GetItem().Address != "" {
+			badFields[globals.CanonicalAddressField] = readOnlyFieldMsg
+		}
+		if req.GetItem().CanonicalTags != nil {
+			badFields[globals.CanonicalTagsField] = readOnlyFieldMsg
+		}
+		if req.GetItem().ConfigTags != nil {
+			badFields[globals.ConfigTagsField] = readOnlyFieldMsg
+		}
+		if req.GetItem().LastStatusTime != nil {
+			badFields[globals.LastStatusTimeField] = readOnlyFieldMsg
+		}
+		if req.GetItem().AuthorizedActions != nil {
+			badFields[globals.AuthorizedActionsField] = readOnlyFieldMsg
+		}
+		nameString := req.GetItem().GetName().String()
+		if !strutil.Printable(nameString) {
+			badFields[globals.NameField] = "Name contains non-printable characters."
+		}
+		if strings.ToLower(nameString) != nameString {
+			badFields[globals.NameField] = "Name must be all lowercase."
+		}
+		descriptionString := req.GetItem().GetDescription().String()
+		if !strutil.Printable(descriptionString) {
+			badFields[globals.DescriptionField] = "Description contains non-printable characters."
+		}
+		return badFields
 	})
-	if err != nil {
-		return nil, err
-	}
-	if connectionInfo == nil {
-		return nil, status.Error(codes.Internal, "Invalid connect connection response.")
-	}
-
-	return &pbs.ConnectConnectionResponse{
-		Status: connStates[0].Status.ProtoVal(),
-	}, nil
-}
-
-func (ws *workerServiceServer) CloseConnection(ctx context.Context, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).CloseConnection"
-	numCloses := len(req.GetCloseRequestData())
-	if numCloses == 0 {
-		return &pbs.CloseConnectionResponse{}, nil
-	}
-
-	closeWiths := make([]session.CloseWith, 0, numCloses)
-	closeIds := make([]string, 0, numCloses)
-
-	for _, v := range req.GetCloseRequestData() {
-		closeIds = append(closeIds, v.GetConnectionId())
-		closeWiths = append(closeWiths, session.CloseWith{
-			ConnectionId: v.GetConnectionId(),
-			BytesUp:      v.GetBytesUp(),
-			BytesDown:    v.GetBytesDown(),
-			ClosedReason: session.ClosedReason(v.GetReason()),
-		})
-	}
-	connRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting connection repo: %v", err)
-	}
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	closeInfos, err := session.CloseConnections(ctx, sessRepo, connRepo, closeWiths)
-	if err != nil {
-		return nil, err
-	}
-	if closeInfos == nil {
-		return nil, status.Error(codes.Internal, "Invalid close connection response.")
-	}
-
-	closeData := make([]*pbs.CloseConnectionResponseData, 0, numCloses)
-	for _, v := range closeInfos {
-		if v.Connection == nil {
-			return nil, status.Errorf(codes.Internal, "No connection found while closing one of the connection IDs: %v", closeIds)
-		}
-		if len(v.ConnectionStates) == 0 {
-			return nil, status.Errorf(codes.Internal, "No connection states found while closing one of the connection IDs: %v", closeIds)
-		}
-		closeData = append(closeData, &pbs.CloseConnectionResponseData{
-			ConnectionId: v.Connection.GetPublicId(),
-			Status:       v.ConnectionStates[0].Status.ProtoVal(),
-		})
-	}
-
-	ret := &pbs.CloseConnectionResponse{
-		CloseResponseData: closeData,
-	}
-
-	return ret, nil
 }

internal/daemon/controller/handlers/workers/worker_service_status_test.go

@@ -1,509 +0,0 @@
-package workers_test
-
-import (
-	"context"
-	"sync"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	"github.com/hashicorp/boundary/internal/authtoken"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
-	"github.com/hashicorp/boundary/internal/db"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/host/static"
-	"github.com/hashicorp/boundary/internal/iam"
-	"github.com/hashicorp/boundary/internal/kms"
-	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/session"
-	"github.com/hashicorp/boundary/internal/target"
-	"github.com/hashicorp/boundary/internal/target/tcp"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestStatus(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_controller1",
-		Type:      "controller",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_worker1",
-		Type:      "worker",
-		Address:   "127.0.0.1",
-	})
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := session.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu)
-	require.NoError(t, err)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PrivateId)
-	require.NoError(t, err)
-
-	cases := []struct {
-		name       string
-		wantErr    bool
-		wantErrMsg string
-		req        *pbs.StatusRequest
-		want       *pbs.StatusResponse
-	}{
-		{
-			name:    "No Sessions",
-			wantErr: false,
-			req: &pbs.StatusRequest{
-				Worker: worker1,
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Type:      "controller",
-						Address:   "127.0.0.1",
-					},
-				},
-			},
-		},
-		{
-			name:    "Still Active",
-			wantErr: false,
-			req: &pbs.StatusRequest{
-				Worker: worker1,
-				Jobs: []*pbs.JobStatus{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-									Connections: []*pbs.Connection{
-										{
-											ConnectionId: connection.PublicId,
-											Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Type:      "controller",
-						Address:   "127.0.0.1",
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-
-			got, err := s.Status(ctx, tc.req)
-			if tc.wantErr {
-				require.Error(err)
-				assert.Nil(got)
-				assert.Equal(tc.wantErrMsg, err.Error())
-				return
-			}
-			assert.Empty(
-				cmp.Diff(
-					tc.want,
-					got,
-					cmpopts.IgnoreUnexported(
-						pbs.StatusResponse{},
-						servers.Server{},
-						pbs.JobChangeRequest{},
-						pbs.Job{},
-						pbs.Job_SessionInfo{},
-						pbs.SessionJobInfo{},
-						pbs.Connection{},
-					),
-					cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-				),
-			)
-		})
-	}
-}
-
-func TestStatusSessionClosed(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_controller1",
-		Type:      "controller",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_worker1",
-		Type:      "worker",
-		Address:   "127.0.0.1",
-	})
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := session.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu)
-	require.NoError(t, err)
-	sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu2 := session.TestTofu(t)
-	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, worker1.Type, tofu2)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PrivateId)
-	require.NoError(t, err)
-
-	cases := []struct {
-		name       string
-		wantErr    bool
-		wantErrMsg string
-		setupFn    func(t *testing.T)
-		req        *pbs.StatusRequest
-		want       *pbs.StatusResponse
-	}{
-		{
-			name:    "Connection Canceled",
-			wantErr: false,
-			setupFn: func(t *testing.T) {
-				_, err := repo.CancelSession(ctx, sess2.PublicId, sess.Version)
-				require.NoError(t, err)
-			},
-			req: &pbs.StatusRequest{
-				Worker: worker1,
-				Jobs: []*pbs.JobStatus{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess2.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-									Connections: []*pbs.Connection{
-										{
-											ConnectionId: connection.PublicId,
-											Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Type:      "controller",
-						Address:   "127.0.0.1",
-					},
-				},
-				JobsRequests: []*pbs.JobChangeRequest{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess2.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
-								},
-							},
-						},
-						RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE,
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-
-			if tc.setupFn != nil {
-				tc.setupFn(t)
-			}
-			got, err := s.Status(ctx, tc.req)
-			if tc.wantErr {
-				require.Error(err)
-				assert.Nil(got)
-				assert.Equal(tc.wantErrMsg, err.Error())
-				return
-			}
-			assert.Empty(
-				cmp.Diff(
-					tc.want,
-					got,
-					cmpopts.IgnoreUnexported(
-						pbs.StatusResponse{},
-						servers.Server{},
-						pbs.JobChangeRequest{},
-						pbs.Job{},
-						pbs.Job_SessionInfo{},
-						pbs.SessionJobInfo{},
-						pbs.Connection{},
-					),
-					cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-				),
-			)
-		})
-	}
-}
-
-func TestStatusDeadConnection(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_controller1",
-		Type:      "controller",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertServer(ctx, &servers.Server{
-		PrivateId: "test_worker1",
-		Type:      "worker",
-		Address:   "127.0.0.1",
-	})
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms, session.WithWorkerStateDelay(0))
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := session.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu)
-	require.NoError(t, err)
-	sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu2 := session.TestTofu(t)
-	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, worker1.Type, tofu2)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PrivateId)
-	require.NoError(t, err)
-	deadConn, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker1.PrivateId)
-	require.NoError(t, err)
-	require.NotEqual(t, deadConn.PublicId, connection.PublicId)
-
-	req := &pbs.StatusRequest{
-		Worker: worker1,
-		Jobs: []*pbs.JobStatus{
-			{
-				Job: &pbs.Job{
-					Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-					JobInfo: &pbs.Job_SessionInfo{
-						SessionInfo: &pbs.SessionJobInfo{
-							SessionId: sess.PublicId,
-							Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-							Connections: []*pbs.Connection{
-								{
-									ConnectionId: connection.PublicId,
-									Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	want := &pbs.StatusResponse{
-		Controllers: []*servers.Server{
-			{
-				PrivateId: "test_controller1",
-				Type:      "controller",
-				Address:   "127.0.0.1",
-			},
-		},
-	}
-
-	got, err := s.Status(ctx, req)
-	assert.Empty(t,
-		cmp.Diff(
-			want,
-			got,
-			cmpopts.IgnoreUnexported(
-				pbs.StatusResponse{},
-				servers.Server{},
-				pbs.JobChangeRequest{},
-				pbs.Job{},
-				pbs.Job_SessionInfo{},
-				pbs.SessionJobInfo{},
-				pbs.Connection{},
-			),
-			cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-		),
-	)
-
-	gotConn, states, err := connRepo.LookupConnection(ctx, deadConn.PublicId)
-	require.NoError(t, err)
-	assert.Equal(t, session.ConnectionSystemError, session.ClosedReason(gotConn.ClosedReason))
-	assert.Equal(t, 2, len(states))
-	assert.Nil(t, states[0].EndTime)
-	assert.Equal(t, session.StatusClosed, states[0].Status)
-}

internal/daemon/controller/intercepting_listener.go

@@ -1,12 +1,14 @@
 package controller
 
 import (
-	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"net"
+	"strings"
 
 	"github.com/hashicorp/boundary/internal/observability/event"
+	nodee "github.com/hashicorp/nodeenrollment"
 )
 
 // tempError is an error that satisfies the temporary error interface that is
@@ -62,42 +64,61 @@ func newInterceptingListener(c *Controller, baseLn net.Listener) *interceptingLi
 
 func (m *interceptingListener) Accept() (net.Conn, error) {
 	const op = "controller.(interceptingListener).Accept"
-	ctx := context.TODO()
+	ctx := m.c.baseContext
 	conn, err := m.baseLn.Accept()
 	if err != nil {
+		// We may already be closed but can't be sure, so just try anyways
 		if conn != nil {
 			if err := conn.Close(); err != nil {
-				event.WriteError(context.TODO(), op, err, event.WithInfoMsg("error closing worker connection"))
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
 			}
 		}
 		return nil, newTempError(err)
 	}
 
-	nonce := make([]byte, 20)
-	read, err := conn.Read(nonce)
-	if err != nil {
-		if err := conn.Close(); err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
+	tlsConn := conn.(*tls.Conn)
+	switch {
+	case 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))
 		}
-		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"))
+		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))
 		}
-		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"))
+		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))
 		}
-		return nil, newTempError(errors.New("did not find valid nonce for incoming worker"))
+		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
+		event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name, "description", workerInfo.Description, "proxy_address", workerInfo.ProxyAddress)
+		return tlsConn, nil
+
+	default:
+		return nil, newTempError(errors.New("unable to authenticate incoming connection"))
 	}
-	workerInfo := workerInfoRaw.(*workerAuthEntry)
-	workerInfo.conn = conn
-	event.WriteSysEvent(ctx, op, "worker successfully authed", "name", workerInfo.Name)
-	return conn, nil
 }
 
 func (m *interceptingListener) Close() error {

internal/daemon/controller/listeners.go

@@ -13,10 +13,12 @@ import (
 
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/hashicorp/boundary/internal/cmd/base"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
+	"github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/nodeenrollment/multihop"
+	"github.com/hashicorp/nodeenrollment/protocol"
 	"google.golang.org/grpc"
 )
 
@@ -103,9 +105,25 @@ func (c *Controller) configureForApi(ln *base.ServerListener) ([]func(), error)
 }
 
 func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error) {
-	l := tls.NewListener(ln.ClusterListener, &tls.Config{
-		GetConfigForClient: c.validateWorkerTls,
-	})
+	const op = "controller.configureForCluster"
+
+	workerAuthStorage, err := c.WorkerAuthRepoStorageFn()
+	if err != nil {
+		return nil, fmt.Errorf("error fetching worker auth storage: %w", err)
+	}
+
+	l, err := protocol.NewInterceptingListener(
+		&protocol.InterceptingListenerConfiguration{
+			Context:      c.baseContext,
+			Storage:      workerAuthStorage,
+			BaseListener: ln.ClusterListener,
+			BaseTlsConfiguration: &tls.Config{
+				GetConfigForClient: c.validateWorkerTls,
+			},
+		})
+	if err != nil {
+		return nil, fmt.Errorf("error instantiating node auth listener: %w", err)
+	}
 
 	workerReqInterceptor, err := workerRequestInfoInterceptor(c.baseContext, c.conf.Eventer)
 	if err != nil {
@@ -125,10 +143,21 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		),
 	)
 
-	workerService := workers.NewWorkerServiceServer(c.ServersRepoFn, c.SessionRepoFn, c.ConnectionRepoFn,
+	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.SessionRepoFn, c.ConnectionRepoFn,
 		c.workerStatusUpdateTimes, c.kms)
 	pbs.RegisterServerCoordinationServiceServer(workerServer, workerService)
 	pbs.RegisterSessionServiceServer(workerServer, workerService)
+
+	multihopService, err := handlers.NewMultihopServiceServer(
+		workerAuthStorage,
+		true,
+		nil,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("%s: error creating multihop service handler: %w", op, err)
+	}
+	multihop.RegisterMultihopServiceServer(workerServer, multihopService)
+
 	metric.InitializeClusterCollectors(c.conf.PrometheusRegisterer, workerServer)
 
 	ln.GrpcServer = workerServer

internal/daemon/controller/testing.go

@@ -735,10 +735,10 @@ func (tc *TestController) AddClusterControllerMember(t testing.TB, opts *TestCon
 // WaitForNextWorkerStatusUpdate waits for the next status check from a worker to
 // come in. If it does not come in within the default status grace
 // period, this function returns an error.
-func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error {
+func (tc *TestController) WaitForNextWorkerStatusUpdate(workerStatusName string) error {
 	const op = "controller.(TestController).WaitForNextWorkerStatusUpdate"
 	ctx := context.TODO()
-	event.WriteSysEvent(ctx, op, "waiting for next status report from worker", "worker", workerId)
+	event.WriteSysEvent(ctx, op, "waiting for next status report from worker", "worker", workerStatusName)
 	waitStatusStart := time.Now()
 	ctx, cancel := context.WithTimeout(tc.ctx, tc.b.StatusGracePeriodDuration)
 	defer cancel()
@@ -777,7 +777,7 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error {
 				return false
 			}
 
-			if workerStatusUpdateId == workerId {
+			if workerStatusUpdateId == workerStatusName {
 				waitStatusCurrent = workerStatusUpdateTime
 				return false
 			}
@@ -795,10 +795,10 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error {
 	}
 
 	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error waiting for next status report from worker", "worker", workerId))
+		event.WriteError(ctx, op, err, event.WithInfoMsg("error waiting for next status report from worker", "worker", workerStatusName))
 		return err
 	}
-	event.WriteSysEvent(ctx, op, "waiting for next status report from worker received successfully", "worker", workerId)
+	event.WriteSysEvent(ctx, op, "waiting for next status report from worker received successfully", "worker", workerStatusName)
 	return nil
 }
 

internal/daemon/controller/tickers.go

@@ -5,10 +5,10 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/hashicorp/boundary/internal/servers/store"
+
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/observability/event"
-	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/types/resource"
 )
 
 // In the future we could make this configurable
@@ -30,7 +30,7 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) {
 			return
 
 		case <-timer.C:
-			if err := c.upsertServer(cancelCtx); err != nil {
+			if err := c.upsertController(cancelCtx); err != nil {
 				event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error fetching repository for status update"))
 			}
 			timer.Reset(statusInterval)
@@ -38,21 +38,18 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) {
 	}
 }
 
-func (c *Controller) upsertServer(ctx context.Context) error {
-	const op = "controller.(Controller).upsertServer"
-	server := &servers.Server{
-		PrivateId:   c.conf.RawConfig.Controller.Name,
-		Name:        c.conf.RawConfig.Controller.Name,
-		Type:        resource.Controller.String(),
-		Description: c.conf.RawConfig.Controller.Description,
-		Address:     c.conf.RawConfig.Controller.PublicClusterAddr,
+func (c *Controller) upsertController(ctx context.Context) error {
+	const op = "controller.(Controller).upsertController"
+	controller := &store.Controller{
+		PrivateId: c.conf.RawConfig.Controller.Name,
+		Address:   c.conf.RawConfig.Controller.PublicClusterAddr,
 	}
 	repo, err := c.ServersRepoFn()
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("error fetching repository for status update"))
 	}
 
-	_, _, err = repo.UpsertServer(ctx, server)
+	_, err = repo.UpsertController(ctx, controller)
 	if err != nil {
 		return errors.Wrap(ctx, err, op, errors.WithMsg("error performing status update"))
 	}

internal/daemon/controller/worker_tls_config.go

@@ -27,6 +27,7 @@ type workerAuthEntry struct {
 // been replayed, then stores connection information into the auth cache, which
 // is used by the intercepting listener to log information.
 func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config, error) {
+	const op = "controller.(Controller).validateWorkerTls"
 	for _, p := range hello.SupportedProtos {
 		switch {
 		case strings.HasPrefix(p, "v1workerauth-"):
@@ -50,7 +51,7 @@ func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config,
 			return tlsConf, err
 		}
 	}
-	return nil, nil
+	return nil, errors.New("no supported protos found in hello")
 }
 
 // v1WorkerAuthConfig: