Add storage bucket worker filter handling

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

@@ -658,7 +658,15 @@ func (s Service) RemoveTargetCredentialSources(ctx context.Context, req *pbs.Rem
 // If set, use the worker_filter or egress_worker_filter to filter the selected workers
 // and ensure we have workers available to service this request. The second return
 // argument is always nil.
-func AuthorizeSessionWithWorkerFilter(_ context.Context, t target.Target, selectedWorkers wl.WorkerList, _ string, _ common.Downstreamers) (wl.WorkerList, *server.Worker, error) {
+func AuthorizeSessionWithWorkerFilter(
+	_ context.Context,
+	t target.Target,
+	selectedWorkers wl.WorkerList,
+	_ string,
+	_ intglobals.ControllerExtension,
+	_ common.Downstreamers,
+	_ ...target.Option,
+) (wl.WorkerList, *server.Worker, error) {
 	if len(selectedWorkers) > 0 {
 		var eval *bexpr.Evaluator
 		var err error
@@ -890,7 +898,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 			"No workers are available to handle this session.")
 	}
 
-	selectedWorkers, protoWorker, err := AuthorizeSessionWorkerFilterFn(ctx, t, selectedWorkers, h, s.downstreams)
+	selectedWorkers, protoWorker, err := AuthorizeSessionWorkerFilterFn(ctx, t, selectedWorkers, h, s.controllerExt, s.downstreams)
 	if err != nil {
 		return nil, err
 	}

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

@@ -117,7 +117,7 @@ func TestWorkerList_EgressFilter(t *testing.T) {
 			if len(tc.filter) > 0 {
 				target.EgressWorkerFilter = tc.filter
 			}
-			out, protocolWorker, err := AuthorizeSessionWithWorkerFilter(ctx, target, tc.in, "", nil)
+			out, protocolWorker, err := AuthorizeSessionWithWorkerFilter(ctx, target, tc.in, "", nil, nil)
 			if tc.errContains != "" {
 				assert.Contains(err.Error(), tc.errContains)
 				assert.Nil(out)

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

@@ -3,7 +3,9 @@
 
 package targets
 
-import "testing"
+import (
+	"testing"
+)
 
 // SetupSuiteTargetFilters is used to ensure that OSS tests run from the ENT repo use the OSS level of target filtering
 func SetupSuiteTargetFilters(t *testing.T) {

internal/globals/globals.go

@@ -5,6 +5,7 @@ package globals
 
 import (
 	"context"
+	"net/netip"
 )
 
 // ControllerExtension defines the interface implemented
@@ -14,3 +15,9 @@ import (
 type ControllerExtension interface {
 	Start(context.Context) error
 }
+
+// This is an interface satisfied by net.DefaultResolver but can be replaced for
+// tests
+type NetIpResolver interface {
+	LookupNetIP(context.Context, string, string) ([]netip.Addr, error)
+}

internal/server/options.go

@@ -42,6 +42,7 @@ type options struct {
 	WithFetchNodeCredentialsRequest        *types.FetchNodeCredentialsRequest
 	withTestPkiWorkerAuthorized            bool
 	withTestPkiWorkerKeyId                 *string
+	withTestUseInputTagsAsApiTags          bool
 	withWorkerType                         WorkerType
 	withRoot                               RootInfo
 	withStopAfter                          uint
@@ -187,6 +188,15 @@ func WithTestPkiWorkerAuthorizedKeyId(id *string) Option {
 	}
 }
 
+// WithTestUseInputTagsAsApiTags tells NewWorker to set the set of input tags as
+// the api tags as well. This is useful for allowing a worker to have canonical
+// tags without having to store worker information in the database.
+func WithTestUseInputTagsAsApiTags(with bool) Option {
+	return func(o *options) {
+		o.withTestUseInputTagsAsApiTags = with
+	}
+}
+
 // WithWorkerType allows specifying a particular type of worker (kms, pki)
 // during lookup or listing
 func WithWorkerType(with WorkerType) Option {

internal/server/options_test.go

@@ -162,6 +162,13 @@ func Test_GetOpts(t *testing.T) {
 		opts.withNewIdFunc = nil
 		assert.Equal(opts, testOpts)
 	})
+	t.Run("WithTestUseInputTagsAsApiTags", func(t *testing.T) {
+		assert := assert.New(t)
+		testOpts := getDefaultOptions()
+		assert.False(testOpts.withTestUseInputTagsAsApiTags)
+		opts := GetOpts(WithTestUseInputTagsAsApiTags(true))
+		assert.True(opts.withTestUseInputTagsAsApiTags)
+	})
 	t.Run("WithWorkerType", func(t *testing.T) {
 		opts := getDefaultOptions()
 		assert.Empty(t, opts.withWorkerType)

internal/server/worker.go

@@ -106,7 +106,7 @@ type Worker struct {
 // not set any of the worker reported values.
 func NewWorker(scopeId string, opt ...Option) *Worker {
 	opts := GetOpts(opt...)
-	return &Worker{
+	worker := &Worker{
 		Worker: &store.Worker{
 			ScopeId:          scopeId,
 			Name:             opts.withName,
@@ -117,6 +117,10 @@ func NewWorker(scopeId string, opt ...Option) *Worker {
 		},
 		inputTags: opts.withWorkerTags,
 	}
+	if opts.withTestUseInputTagsAsApiTags {
+		worker.apiTags = worker.inputTags
+	}
+	return worker
 }
 
 // allocWorker will allocate a Worker
@@ -162,7 +166,7 @@ func (w *Worker) ActiveConnectionCount() uint32 {
 // CanonicalTags is the deduplicated set of tags contained on both the resource
 // set over the API as well as the tags reported by the worker itself. This
 // function is guaranteed to return a non-nil map.
-func (w *Worker) CanonicalTags() map[string][]string {
+func (w *Worker) CanonicalTags(opt ...Option) map[string][]string {
 	dedupedTags := make(map[Tag]struct{})
 	for _, t := range w.apiTags {
 		dedupedTags[*t] = struct{}{}

internal/session/testing.go

@@ -144,6 +144,13 @@ func TestSession(t testing.TB, conn *db.DB, rootWrapper wrapping.Wrapper, c Comp
 		TestSessionTargetAddress(t, conn, s.PublicId, s.TargetId)
 	}
 
+	if s.ProtocolWorkerId != "" {
+		p, err := NewSessionWorkerProtocol(ctx, s.PublicId, s.ProtocolWorkerId)
+		require.NoError(err)
+		err = rw.Create(ctx, p)
+		require.NoError(err)
+	}
+
 	ss, err := fetchStates(ctx, rw, s.PublicId, append(opts.withDbOpts, db.WithOrder("start_time desc"))...)
 	require.NoError(err)
 	s.States = ss

internal/target/options.go

@@ -4,8 +4,10 @@
 package target
 
 import (
+	"net"
 	"time"
 
+	intglobals "github.com/hashicorp/boundary/internal/globals"
 	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/types/subtypes"
 )
@@ -42,12 +44,14 @@ type options struct {
 	WithPermissions            []perms.Permission
 	WithPublicId               string
 	WithWorkerFilter           string
+	WithTestWorkerFilter       string
 	WithEgressWorkerFilter     string
 	WithIngressWorkerFilter    string
 	WithTargetIds              []string
 	WithAddress                string
 	WithStorageBucketId        string
 	WithEnableSessionRecording bool
+	WithNetResolver            intglobals.NetIpResolver
 }
 
 func getDefaultOptions() options {
@@ -70,9 +74,11 @@ func getDefaultOptions() options {
 		WithPermissions:            nil,
 		WithPublicId:               "",
 		WithWorkerFilter:           "",
+		WithTestWorkerFilter:       "",
 		WithEgressWorkerFilter:     "",
 		WithIngressWorkerFilter:    "",
 		WithAddress:                "",
+		WithNetResolver:            net.DefaultResolver,
 	}
 }
 
@@ -195,6 +201,13 @@ func WithWorkerFilter(filter string) Option {
 	}
 }
 
+// WithTestWorkerFilter provides an optional worker filter used only in testing
+func WithTestWorkerFilter(filter string) Option {
+	return func(o *options) {
+		o.WithTestWorkerFilter = filter
+	}
+}
+
 // WithEgressWorkerFilter provides an optional egress worker filter
 func WithEgressWorkerFilter(filter string) Option {
 	return func(o *options) {
@@ -245,3 +258,10 @@ func WithStorageBucketId(id string) Option {
 		o.WithStorageBucketId = id
 	}
 }
+
+// WithNetResolver provides an option to specify a custom DNS resolver
+func WithNetResolver(resolver intglobals.NetIpResolver) Option {
+	return func(o *options) {
+		o.WithNetResolver = resolver
+	}
+}

internal/target/options_test.go

@@ -125,6 +125,13 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.WithWorkerFilter = `"/foo" == "bar"`
 		assert.Equal(opts, testOpts)
 	})
+	t.Run("WithTestWorkerFilter", func(t *testing.T) {
+		assert := assert.New(t)
+		opts := GetOpts(WithTestWorkerFilter(`"/foo" == "bar"`))
+		testOpts := getDefaultOptions()
+		testOpts.WithTestWorkerFilter = `"/foo" == "bar"`
+		assert.Equal(opts, testOpts)
+	})
 	t.Run("WithEgressWorkerFilter", func(t *testing.T) {
 		assert := assert.New(t)
 		opts := GetOpts(WithEgressWorkerFilter(`"/foo" == "bar"`))

version/feature_manager.go

@@ -20,6 +20,7 @@ const (
 	CredentialLibraryVaultSubtype
 	UseTargetIdForHostId
 	RequireVersionInWorkerInfo
+	SshSessionRecording
 )
 
 var featureMap map[Feature]MetadataConstraint
@@ -68,6 +69,9 @@ func init() {
 	featureMap[RequireVersionInWorkerInfo] = MetadataConstraint{
 		Constraints: mustNewConstraints(">= 0.13.0"),
 	}
+	featureMap[SshSessionRecording] = MetadataConstraint{
+		Constraints: mustNewConstraints(">= 0.13.0"),
+	}
 }
 
 func mustNewConstraints(v string) gvers.Constraints {