AuthorizeConnection with filters (#2734)

* AuthorizeConnection with filters

internal/daemon/controller/common/common.go

@@ -27,3 +27,11 @@ type (
 	ConnectionRepoFactory        func() (*session.ConnectionRepository, error)
 	WorkerAuthRepoStorageFactory func() (*server.WorkerAuthRepositoryStorage, error)
 )
+
+// Downstreamers provides at least a minimum interface that must be met by a
+// Controller.downstreamWorkers field which is far better than allowing any (empty
+// interface)
+type Downstreamers interface {
+	// RootId returns the root ID of the downstreamers' graph
+	RootId() string
+}

internal/daemon/controller/controller.go

@@ -67,19 +67,11 @@ type downstreamWorkersTicker interface {
 	StartDownstreamWorkersTicking(context.Context, int) error
 }
 
-// downstreamers provides at least a minimum interface that must be met by a
-// Controller.downstreamWorkers field which is far better than allowing any (empty
-// interface)
-type downstreamers interface {
-	// RootId returns the root ID of the downstreamers' graph
-	RootId() string
-}
-
 var (
 	downstreamRouterFactory func() downstreamRouter
 
-	downstreamersFactory           func(context.Context, string) (downstreamers, error)
-	downstreamWorkersTickerFactory func(context.Context, string, downstreamers, downstreamRouter) (downstreamWorkersTicker, error)
+	downstreamersFactory           func(context.Context, string) (common.Downstreamers, error)
+	downstreamWorkersTickerFactory func(context.Context, string, common.Downstreamers, downstreamRouter) (downstreamWorkersTicker, error)
 	commandClientFactory           func(context.Context, *Controller) error
 )
 
@@ -97,7 +89,7 @@ type Controller struct {
 	workerAuthCache *sync.Map
 
 	// downstream workers and routes to those workers
-	downstreamWorkers downstreamers
+	downstreamWorkers common.Downstreamers
 	downstreamRoutes  downstreamRouter
 
 	apiListeners    []*base.ServerListener

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

@@ -15,6 +15,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/credential/vault"
+	wl "github.com/hashicorp/boundary/internal/daemon/common"
 	"github.com/hashicorp/boundary/internal/daemon/controller/auth"
 	"github.com/hashicorp/boundary/internal/daemon/controller/common"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
@@ -38,7 +39,6 @@ import (
 	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
 	"github.com/hashicorp/go-bexpr"
 	"github.com/hashicorp/go-secure-stdlib/strutil"
-	"github.com/mitchellh/pointerstructure"
 	"github.com/mr-tron/base58"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/protobuf/proto"
@@ -638,7 +638,7 @@ func AuthorizeSessionWithWorkerFilter(t target.Target, selectedWorkers []*server
 			return nil, err
 		}
 
-		selectedWorkers, err = workerList(selectedWorkers).filtered(eval)
+		selectedWorkers, err = wl.WorkerList(selectedWorkers).Filtered(eval)
 		if err != nil {
 			return nil, err
 		}
@@ -877,7 +877,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	if err != nil {
 		return nil, err
 	}
-	sess, err = sessionRepo.CreateSession(ctx, wrapper, sess, workerList(selectedWorkers).addresses())
+	sess, err = sessionRepo.CreateSession(ctx, wrapper, sess, wl.WorkerList(selectedWorkers).Addresses())
 	if err != nil {
 		return nil, err
 	}
@@ -976,7 +976,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		PrivateKey:      sess.CertificatePrivateKey,
 		HostId:          chosenEndpoint.HostId,
 		Endpoint:        endpointUrl.String(),
-		WorkerInfo:      workerList(selectedWorkers).workerInfos(),
+		WorkerInfo:      wl.WorkerList(selectedWorkers).WorkerInfos(),
 		ConnectionLimit: t.GetSessionConnectionLimit(),
 	}
 	marshaledSad, err := proto.Marshal(sad)
@@ -1882,46 +1882,3 @@ 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 []*server.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 []*server.Worker
-	for _, worker := range w {
-		filterInput := map[string]any{
-			"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

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/hashicorp/boundary/internal/daemon/common"
+
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/server"
@@ -25,15 +27,15 @@ func TestWorkerList_Addresses(t *testing.T) {
 		"test4",
 	}
 	var workerInfos []*pb.WorkerInfo
-	var tested workerList
+	var tested common.WorkerList
 	for _, a := range addresses {
 		workerInfos = append(workerInfos, &pb.WorkerInfo{Address: a})
 		tested = append(tested, server.NewWorker(scope.Global.String(),
 			server.WithName(a),
 			server.WithAddress(a)))
 	}
-	assert.Equal(t, addresses, tested.addresses())
-	assert.Equal(t, workerInfos, tested.workerInfos())
+	assert.Equal(t, addresses, tested.Addresses())
+	assert.Equal(t, workerInfos, tested.WorkerInfos())
 }
 
 func TestWorkerList_EgressFilter(t *testing.T) {

internal/daemon/controller/rpc_registration.go

@@ -34,7 +34,7 @@ func registerControllerServerCoordinationService(ctx context.Context, c *Control
 	}
 
 	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn,
-		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
+		c.SessionRepoFn, c.ConnectionRepoFn, c.downstreamWorkers, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
 	pbs.RegisterServerCoordinationServiceServer(server, workerService)
 	return nil
 }
@@ -52,7 +52,7 @@ func registerControllerSessionService(ctx context.Context, c *Controller, server
 	}
 
 	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn,
-		c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
+		c.SessionRepoFn, c.ConnectionRepoFn, c.downstreamWorkers, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale)
 	pbs.RegisterSessionServiceServer(server, workerService)
 	return nil
 }