feat(targets): Addition of egress and ingress worker filters (#2654)

* feat(targets): Addition of egress and ingress worker filters

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

@@ -91,8 +91,16 @@ var (
 		action.Create,
 		action.List,
 	}
+
+	ValidateIngressWorkerFilterFn  = IngressWorkerFilterUnsupported
+	AuthorizeSessionWorkerFilterFn = AuthorizeSessionWithWorkerFilter
+	WorkerFilterDeprecationMessage = "This field is deprecated. Use egress_filter instead."
 )
 
+func IngressWorkerFilterUnsupported(string) error {
+	return fmt.Errorf("Ingress Worker Filter field is not supported in OSS")
+}
+
 // Service handles request as described by the pbs.TargetServiceServer interface.
 type Service struct {
 	pbs.UnsafeTargetServiceServer
@@ -612,6 +620,39 @@ func (s Service) RemoveTargetCredentialSources(ctx context.Context, req *pbs.Rem
 	return &pbs.RemoveTargetCredentialSourcesResponse{Item: item}, nil
 }
 
+// If set, use the worker_filter or egress_worker_filter to filtere the selected workers
+// and ensure we have workers available to service this request.
+func AuthorizeSessionWithWorkerFilter(t target.Target, selectedWorkers []*server.Worker) ([]*server.Worker, error) {
+	if len(selectedWorkers) > 0 {
+		var eval *bexpr.Evaluator
+		var err error
+		switch {
+		case len(t.GetEgressWorkerFilter()) > 0:
+			eval, err = bexpr.CreateEvaluator(t.GetEgressWorkerFilter())
+		case len(t.GetWorkerFilter()) > 0:
+			eval, err = bexpr.CreateEvaluator(t.GetWorkerFilter())
+		default: // No filter
+			return selectedWorkers, nil
+		}
+		if err != nil {
+			return nil, err
+		}
+
+		selectedWorkers, err = workerList(selectedWorkers).filtered(eval)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if len(selectedWorkers) == 0 {
+		return nil, handlers.ApiErrorWithCodeAndMessage(
+			codes.FailedPrecondition,
+			"No workers are available to handle this session, or all have been filtered.")
+	}
+
+	return selectedWorkers, nil
+}
+
 func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSessionRequest) (*pbs.AuthorizeSessionResponse, error) {
 	const op = "targets.(Service).AuthorizeSession"
 	if err := validateAuthorizeSessionRequest(req); err != nil {
@@ -691,21 +732,9 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		return nil, err
 	}
 
-	if len(t.GetWorkerFilter()) > 0 && len(selectedWorkers) > 0 {
-		eval, err := bexpr.CreateEvaluator(t.GetWorkerFilter())
-		if err != nil {
-			return nil, err
-		}
-		selectedWorkers, err = workerList(selectedWorkers).filtered(eval)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if len(selectedWorkers) == 0 {
-		return nil, handlers.ApiErrorWithCodeAndMessage(
-			codes.FailedPrecondition,
-			"No workers are available to handle this session, or all have been filtered.")
+	selectedWorkers, err = AuthorizeSessionWorkerFilterFn(t, selectedWorkers)
+	if err != nil {
+		return nil, err
 	}
 
 	// Randomize the workers
@@ -824,18 +853,20 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	expTime := timestamppb.Now()
 	expTime.Seconds += int64(t.GetSessionMaxSeconds())
 	sessionComposition := session.ComposedOf{
-		UserId:             authResults.UserId,
-		HostId:             chosenEndpoint.HostId,
-		TargetId:           t.GetPublicId(),
-		HostSetId:          chosenEndpoint.SetId,
-		AuthTokenId:        authResults.AuthTokenId,
-		ProjectId:          authResults.Scope.Id,
-		Endpoint:           endpointUrl.String(),
-		ExpirationTime:     &timestamp.Timestamp{Timestamp: expTime},
-		ConnectionLimit:    t.GetSessionConnectionLimit(),
-		WorkerFilter:       t.GetWorkerFilter(),
-		DynamicCredentials: dynCreds,
-		StaticCredentials:  staticCreds,
+		UserId:              authResults.UserId,
+		HostId:              chosenEndpoint.HostId,
+		TargetId:            t.GetPublicId(),
+		HostSetId:           chosenEndpoint.SetId,
+		AuthTokenId:         authResults.AuthTokenId,
+		ProjectId:           authResults.Scope.Id,
+		Endpoint:            endpointUrl.String(),
+		ExpirationTime:      &timestamp.Timestamp{Timestamp: expTime},
+		ConnectionLimit:     t.GetSessionConnectionLimit(),
+		WorkerFilter:        t.GetWorkerFilter(),
+		EgressWorkerFilter:  t.GetEgressWorkerFilter(),
+		IngressWorkerFilter: t.GetIngressWorkerFilter(),
+		DynamicCredentials:  dynCreds,
+		StaticCredentials:   staticCreds,
 	}
 
 	sess, err := session.New(sessionComposition)
@@ -1000,8 +1031,11 @@ func (s Service) createInRepo(ctx context.Context, item *pb.Target) (target.Targ
 	if item.GetSessionConnectionLimit() != nil {
 		opts = append(opts, target.WithSessionConnectionLimit(item.GetSessionConnectionLimit().GetValue()))
 	}
-	if item.GetWorkerFilter() != nil {
-		opts = append(opts, target.WithWorkerFilter(item.GetWorkerFilter().GetValue()))
+	if item.GetEgressWorkerFilter() != nil {
+		opts = append(opts, target.WithEgressWorkerFilter(item.GetEgressWorkerFilter().GetValue()))
+	}
+	if item.GetIngressWorkerFilter() != nil {
+		opts = append(opts, target.WithIngressWorkerFilter(item.GetIngressWorkerFilter().GetValue()))
 	}
 
 	attr, err := subtypeRegistry.newAttribute(target.SubtypeFromType(item.GetType()), item.GetAttrs())
@@ -1043,9 +1077,16 @@ func (s Service) updateInRepo(ctx context.Context, scopeId, id string, mask []st
 	if item.GetSessionConnectionLimit() != nil {
 		opts = append(opts, target.WithSessionConnectionLimit(item.GetSessionConnectionLimit().GetValue()))
 	}
-	if filter := item.GetWorkerFilter(); filter != nil {
+	// worker_filter is deprecated, but we allow users who have migrated with a worker_filter value to update it.
+	if workerFilter := item.GetWorkerFilter(); workerFilter != nil {
 		opts = append(opts, target.WithWorkerFilter(item.GetWorkerFilter().GetValue()))
 	}
+	if egressFilter := item.GetEgressWorkerFilter(); egressFilter != nil {
+		opts = append(opts, target.WithEgressWorkerFilter(item.GetEgressWorkerFilter().GetValue()))
+	}
+	if ingressFilter := item.GetIngressWorkerFilter(); ingressFilter != nil {
+		opts = append(opts, target.WithIngressWorkerFilter(item.GetIngressWorkerFilter().GetValue()))
+	}
 	subtype := target.SubtypeFromId(id)
 
 	attr, err := subtypeRegistry.newAttribute(subtype, item.GetAttrs())
@@ -1355,6 +1396,12 @@ func toProto(ctx context.Context, in target.Target, hostSources []target.HostSou
 	if outputFields.Has(globals.WorkerFilterField) && in.GetWorkerFilter() != "" {
 		out.WorkerFilter = wrapperspb.String(in.GetWorkerFilter())
 	}
+	if outputFields.Has(globals.EgressWorkerFilterField) && in.GetEgressWorkerFilter() != "" {
+		out.EgressWorkerFilter = wrapperspb.String(in.GetEgressWorkerFilter())
+	}
+	if outputFields.Has(globals.IngressWorkerFilterField) && in.GetIngressWorkerFilter() != "" {
+		out.IngressWorkerFilter = wrapperspb.String(in.GetIngressWorkerFilter())
+	}
 	if outputFields.Has(globals.ScopeField) {
 		out.Scope = opts.WithScope
 	}
@@ -1461,9 +1508,18 @@ func validateCreateRequest(req *pbs.CreateTargetRequest) error {
 		} else if target.SubtypeFromType(req.GetItem().GetType()) == "" {
 			badFields[globals.TypeField] = "Unknown type provided."
 		}
-		if filter := req.GetItem().GetWorkerFilter(); filter != nil {
-			if _, err := bexpr.CreateEvaluator(filter.GetValue()); err != nil {
-				badFields[globals.WorkerFilterField] = "Unable to successfully parse filter expression."
+		if workerFilter := req.GetItem().GetWorkerFilter(); workerFilter != nil {
+			badFields[globals.WorkerFilterField] = WorkerFilterDeprecationMessage
+		}
+		if egressFilter := req.GetItem().GetEgressWorkerFilter(); egressFilter != nil {
+			if _, err := bexpr.CreateEvaluator(egressFilter.GetValue()); err != nil {
+				badFields[globals.EgressWorkerFilterField] = "Unable to successfully parse egress filter expression."
+			}
+		}
+		if ingressFilter := req.GetItem().GetIngressWorkerFilter(); ingressFilter != nil {
+			err := ValidateIngressWorkerFilterFn(ingressFilter.GetValue())
+			if err != nil {
+				badFields[globals.IngressWorkerFilterField] = err.Error()
 			}
 		}
 
@@ -1504,10 +1560,30 @@ func validateUpdateRequest(req *pbs.UpdateTargetRequest) error {
 		if req.GetItem().GetSessionMaxSeconds() != nil && req.GetItem().GetSessionMaxSeconds().GetValue() == 0 {
 			badFields[globals.SessionMaxSecondsField] = "This must be greater than zero."
 		}
-		if filter := req.GetItem().GetWorkerFilter(); filter != nil {
-			if _, err := bexpr.CreateEvaluator(filter.GetValue()); err != nil {
+		// worker_filter is mutually exclusive from ingress and egress filter
+		workerFilterFound := false
+		if workerFilter := req.GetItem().GetWorkerFilter(); workerFilter != nil {
+			if _, err := bexpr.CreateEvaluator(workerFilter.GetValue()); err != nil {
 				badFields[globals.WorkerFilterField] = "Unable to successfully parse filter expression."
 			}
+			workerFilterFound = true
+		}
+		if egressFilter := req.GetItem().GetEgressWorkerFilter(); egressFilter != nil {
+			if workerFilterFound {
+				badFields[globals.EgressWorkerFilterField] = "Cannot set worker_filter and egress_filter; they are mutually exclusive fields."
+			}
+			if _, err := bexpr.CreateEvaluator(egressFilter.GetValue()); err != nil {
+				badFields[globals.EgressWorkerFilterField] = "Unable to successfully parse egress filter expression."
+			}
+		}
+		if ingressFilter := req.GetItem().GetIngressWorkerFilter(); ingressFilter != nil {
+			if workerFilterFound {
+				badFields[globals.IngressWorkerFilterField] = "Cannot set worker_filter and ingress_filter; they are mutually exclusive fields."
+			}
+			err := ValidateIngressWorkerFilterFn(ingressFilter.GetValue())
+			if err != nil {
+				badFields[globals.IngressWorkerFilterField] = err.Error()
+			}
 		}
 		subtype := target.SubtypeFromId(req.GetId())
 		_, err := subtypeRegistry.get(subtype)

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

@@ -9,9 +9,10 @@ import (
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/server"
+	"github.com/hashicorp/boundary/internal/target/targettest"
+	"github.com/hashicorp/boundary/internal/target/targettest/store"
 	"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"
 )
@@ -35,11 +36,13 @@ func TestWorkerList_Addresses(t *testing.T) {
 	assert.Equal(t, workerInfos, tested.workerInfos())
 }
 
-func TestWorkerList_Filter(t *testing.T) {
+func TestWorkerList_EgressFilter(t *testing.T) {
+	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
 	wrapper := db.TestWrapper(t)
 	kmsCache := kms.TestKms(t, conn, wrapper)
 	require.NoError(t, kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader)))
+
 	var workers []*server.Worker
 	for i := 0; i < 5; i++ {
 		switch {
@@ -60,25 +63,66 @@ func TestWorkerList_Filter(t *testing.T) {
 		}
 	}
 
-	{
-		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)
+	cases := []struct {
+		name        string
+		in          []*server.Worker
+		out         []*server.Worker
+		filter      string
+		errContains string
+	}{
+		{
+			name:        "no-workers",
+			in:          []*server.Worker{},
+			out:         []*server.Worker{},
+			filter:      "",
+			errContains: "No workers are available to handle this session, or all have been filtered",
+		},
+		{
+			name: "no-filter",
+			in:   workers,
+			out:  workers,
+		},
+		{
+			name:        "filter-no-matches",
+			in:          workers,
+			out:         workers,
+			filter:      `"/name" matches "test_worker_[13]" and "configvalue2" in "/tags/key"`,
+			errContains: "No workers are available to handle this session, or all have been filtered",
+		},
+		{
+			name:   "filter-one-match",
+			in:     workers,
+			out:    []*server.Worker{workers[1]},
+			filter: `"/name" matches "test_worker_[12]" and "configvalue" in "/tags/key"`,
+		},
+		{
+			name:   "filter-two-matches",
+			in:     workers,
+			out:    []*server.Worker{workers[1], workers[3]},
+			filter: `"configvalue" in "/tags/key"`,
+		},
 	}
-	{
-		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)
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			target := &targettest.Target{
+				Target: &store.Target{},
+			}
+			if len(tc.filter) > 0 {
+				target.EgressWorkerFilter = tc.filter
+			}
+			out, err := AuthorizeSessionWithWorkerFilter(target, tc.in)
+			if tc.errContains != "" {
+				assert.Contains(err.Error(), tc.errContains)
+				assert.Nil(out)
+				return
+			}
+
+			require.NoError(err)
+			require.Len(out, len(tc.out))
+			for i, exp := range tc.out {
+				assert.Equal(exp.Name, out[i].Name)
+			}
+		})
 	}
 }

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

@@ -534,6 +534,10 @@ func TestCreate(t *testing.T) {
 	_ = iam.TestUserRole(t, conn, r.GetPublicId(), at.GetIamUserId())
 	_ = iam.TestRoleGrant(t, conn, r.GetPublicId(), "id=*;type=*;actions=*")
 
+	// Ensure we are using the OSS worker filter function
+	workerFilterFn := targets.AuthorizeSessionWorkerFilterFn
+	targets.AuthorizeSessionWorkerFilterFn = targets.AuthorizeSessionWithWorkerFilter
+
 	cases := []struct {
 		name string
 		req  *pbs.CreateTargetRequest
@@ -552,7 +556,7 @@ func TestCreate(t *testing.T) {
 						DefaultPort: wrapperspb.UInt32(2),
 					},
 				},
-				WorkerFilter: wrapperspb.String(`type == "bar"`),
+				EgressWorkerFilter: wrapperspb.String(`type == "bar"`),
 			}},
 			res: &pbs.CreateTargetResponse{
 				Uri: fmt.Sprintf("targets/%s_", tcp.TargetPrefix),
@@ -570,18 +574,18 @@ func TestCreate(t *testing.T) {
 					SessionMaxSeconds:      wrapperspb.UInt32(28800),
 					SessionConnectionLimit: wrapperspb.Int32(-1),
 					AuthorizedActions:      testAuthorizedActions,
-					WorkerFilter:           wrapperspb.String(`type == "bar"`),
+					EgressWorkerFilter:     wrapperspb.String(`type == "bar"`),
 				},
 			},
 		},
 		{
 			name: "Create a target with no port",
 			req: &pbs.CreateTargetRequest{Item: &pb.Target{
-				ScopeId:      proj.GetPublicId(),
-				Name:         wrapperspb.String("name"),
-				Description:  wrapperspb.String("desc"),
-				Type:         tcp.Subtype.String(),
-				WorkerFilter: wrapperspb.String(`type == "bar"`),
+				ScopeId:            proj.GetPublicId(),
+				Name:               wrapperspb.String("name"),
+				Description:        wrapperspb.String("desc"),
+				Type:               tcp.Subtype.String(),
+				EgressWorkerFilter: wrapperspb.String(`type == "bar"`),
 			}},
 			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
 		},
@@ -643,7 +647,23 @@ func TestCreate(t *testing.T) {
 		{
 			name: "Invalid worker filter expression",
 			req: &pbs.CreateTargetRequest{Item: &pb.Target{
-				WorkerFilter: wrapperspb.String("bad expression"),
+				EgressWorkerFilter: wrapperspb.String("bad expression"),
+			}},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Deprecated worker filter",
+			req: &pbs.CreateTargetRequest{Item: &pb.Target{
+				WorkerFilter: wrapperspb.String(`"/name" matches "test-worker"`),
+			}},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Ingress filter unsupported on OSS",
+			req: &pbs.CreateTargetRequest{Item: &pb.Target{
+				IngressWorkerFilter: wrapperspb.String(`"/name" matches "test-worker"`),
 			}},
 			res: nil,
 			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
@@ -687,6 +707,8 @@ func TestCreate(t *testing.T) {
 			assert.Empty(cmp.Diff(got, tc.res, protocmp.Transform()), "CreateTarget(%q)\n got response %q\n, wanted %q\n", tc.req, got, tc.res)
 		})
 	}
+	// Reset worker filter func
+	targets.AuthorizeSessionWorkerFilterFn = workerFilterFn
 }
 
 func TestUpdate(t *testing.T) {
@@ -757,6 +779,12 @@ func TestUpdate(t *testing.T) {
 	tested, err := testService(t, context.Background(), conn, kms, wrapper)
 	require.NoError(t, err, "Failed to create a new host set service.")
 
+	// Ensure we are using the OSS worker filter functions
+	workerFilterFn := targets.AuthorizeSessionWorkerFilterFn
+	targets.AuthorizeSessionWorkerFilterFn = targets.AuthorizeSessionWithWorkerFilter
+	validateIngressFn := targets.ValidateIngressWorkerFilterFn
+	targets.ValidateIngressWorkerFilterFn = targets.IngressWorkerFilterUnsupported
+
 	cases := []struct {
 		name string
 		req  *pbs.UpdateTargetRequest
@@ -1054,6 +1082,33 @@ func TestUpdate(t *testing.T) {
 			res: nil,
 			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
 		},
+		{
+			name: "Ingress filter unsupported",
+			req: &pbs.UpdateTargetRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"ingress_worker_filter"},
+				},
+				Item: &pb.Target{
+					IngressWorkerFilter: wrapperspb.String(`type == "bar"`),
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Can't update worker filter and egress filter",
+			req: &pbs.UpdateTargetRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"worker_filter", "egress_worker_filter"},
+				},
+				Item: &pb.Target{
+					WorkerFilter:       wrapperspb.String(`type == "bar"`),
+					EgressWorkerFilter: wrapperspb.String(`type == "bar"`),
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -1095,6 +1150,9 @@ func TestUpdate(t *testing.T) {
 			assert.Empty(cmp.Diff(got, tc.res, protocmp.Transform()), "UpdateTarget(%q) got response %q, wanted %q", req, got, tc.res)
 		})
 	}
+	// Reset worker filter funcs
+	targets.AuthorizeSessionWorkerFilterFn = workerFilterFn
+	targets.ValidateIngressWorkerFilterFn = validateIngressFn
 }
 
 func TestUpdate_BadVersion(t *testing.T) {