In absence of ingress filter, use directly connected worker (#2757)

* directly connected ingress filtering

internal/cmd/commands/server/worker_shutdown_reload_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/hashicorp/boundary/api/authmethods"
 	"github.com/hashicorp/boundary/api/authtokens"
 	"github.com/hashicorp/boundary/api/targets"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
+	_ "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets/tcp"
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/boundary/internal/tests/helper"
 	"github.com/hashicorp/boundary/internal/types/scope"
@@ -21,8 +23,6 @@ import (
 	"github.com/mitchellh/cli"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
-
-	_ "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets/tcp"
 )
 
 const shutdownReloadWorkerProvidedConfiguration = `
@@ -91,6 +91,7 @@ func TestServer_ShutdownWorker(t *testing.T) {
 	require.NotNil(tgtR)
 
 	// Authorize and connect
+	tg.SetupSuiteTargetFilters(t)
 	sess := helper.NewTestSession(ctx, t, tcl, tgt.Id)
 	sConn := sess.Connect(ctx, t)
 

internal/daemon/controller/handler.go

@@ -197,6 +197,7 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error {
 			c.StaticHostRepoFn,
 			c.VaultCredentialRepoFn,
 			c.StaticCredentialRepoFn,
+			c.downstreamWorkers,
 			c.workerStatusGracePeriod)
 		if err != nil {
 			return fmt.Errorf("failed to create target handler service: %w", err)

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

@@ -113,6 +113,7 @@ type Service struct {
 	staticHostRepoFn        common.StaticRepoFactory
 	vaultCredRepoFn         common.VaultCredentialRepoFactory
 	staticCredRepoFn        common.StaticCredentialRepoFactory
+	downstreams             common.Downstreamers
 	kmsCache                *kms.Kms
 	workerStatusGracePeriod *atomic.Int64
 }
@@ -131,6 +132,7 @@ func NewService(
 	staticHostRepoFn common.StaticRepoFactory,
 	vaultCredRepoFn common.VaultCredentialRepoFactory,
 	staticCredRepoFn common.StaticCredentialRepoFactory,
+	downstreams common.Downstreamers,
 	workerStatusGracePeriod *atomic.Int64,
 ) (Service, error) {
 	const op = "targets.NewService"
@@ -167,6 +169,7 @@ func NewService(
 		staticHostRepoFn:        staticHostRepoFn,
 		vaultCredRepoFn:         vaultCredRepoFn,
 		staticCredRepoFn:        staticCredRepoFn,
+		downstreams:             downstreams,
 		kmsCache:                kmsCache,
 		workerStatusGracePeriod: workerStatusGracePeriod,
 	}, nil
@@ -622,7 +625,7 @@ func (s Service) RemoveTargetCredentialSources(ctx context.Context, req *pbs.Rem
 
 // 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) {
+func AuthorizeSessionWithWorkerFilter(_ context.Context, t target.Target, selectedWorkers []*server.Worker, _ common.Downstreamers) ([]*server.Worker, error) {
 	if len(selectedWorkers) > 0 {
 		var eval *bexpr.Evaluator
 		var err error
@@ -732,7 +735,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		return nil, err
 	}
 
-	selectedWorkers, err = AuthorizeSessionWorkerFilterFn(t, selectedWorkers)
+	selectedWorkers, err = AuthorizeSessionWorkerFilterFn(ctx, t, selectedWorkers, s.downstreams)
 	if err != nil {
 		return nil, err
 	}

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

@@ -7,7 +7,6 @@ import (
 	"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"
@@ -40,6 +39,8 @@ func TestWorkerList_Addresses(t *testing.T) {
 
 func TestWorkerList_EgressFilter(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
+	SetupSuiteTargetFilters(t)
 	conn, _ := db.TestSetup(t, "postgres")
 	wrapper := db.TestWrapper(t)
 	kmsCache := kms.TestKms(t, conn, wrapper)
@@ -113,7 +114,7 @@ func TestWorkerList_EgressFilter(t *testing.T) {
 			if len(tc.filter) > 0 {
 				target.EgressWorkerFilter = tc.filter
 			}
-			out, err := AuthorizeSessionWithWorkerFilter(target, tc.in)
+			out, err := AuthorizeSessionWithWorkerFilter(ctx, target, tc.in, nil)
 			if tc.errContains != "" {
 				assert.Contains(err.Error(), tc.errContains)
 				assert.Nil(out)

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

@@ -105,7 +105,7 @@ func testService(t *testing.T, ctx context.Context, conn *db.DB, kms *kms.Kms, w
 	staticCredRepoFn := func() (*credstatic.Repository, error) {
 		return credstatic.NewRepository(context.Background(), rw, rw, kms)
 	}
-	return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
+	return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, nil, statusGracePeriod)
 }
 
 func TestGet(t *testing.T) {
@@ -2290,6 +2290,7 @@ func TestRemoveTargetCredentialSources(t *testing.T) {
 
 func TestAuthorizeSession(t *testing.T) {
 	ctx := context.Background()
+	targets.SetupSuiteTargetFilters(t)
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
 	wrapper := db.TestWrapper(t)
@@ -2469,7 +2470,7 @@ func TestAuthorizeSession(t *testing.T) {
 
 	statusGracePeriod := new(atomic.Int64)
 	statusGracePeriod.Store(int64(server.DefaultLiveness))
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, nil, statusGracePeriod)
 	require.NoError(t, err)
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -2584,6 +2585,8 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 	wrapper := db.TestWrapper(t)
 	kms := kms.TestKms(t, conn, wrapper)
 
+	targets.SetupSuiteTargetFilters(t)
+
 	sche := scheduler.TestScheduler(t, conn, wrapper)
 	err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
 	require.NoError(t, err)
@@ -2636,7 +2639,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 
 	statusGracePeriod := new(atomic.Int64)
 	statusGracePeriod.Store(int64(server.DefaultLiveness))
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, nil, statusGracePeriod)
 	require.NoError(t, err)
 
 	hc := static.TestCatalogs(t, conn, proj.GetPublicId(), 1)[0]
@@ -3160,6 +3163,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 
 func TestAuthorizeSession_Errors(t *testing.T) {
 	ctx := context.Background()
+	targets.SetupSuiteTargetFilters(t)
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
 	wrapper := db.TestWrapper(t)
@@ -3201,7 +3205,7 @@ func TestAuthorizeSession_Errors(t *testing.T) {
 
 	statusGracePeriod := new(atomic.Int64)
 	statusGracePeriod.Store(int64(server.DefaultLiveness))
-	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod)
+	s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, nil, statusGracePeriod)
 	require.NoError(t, err)
 
 	// Authorized user gets full permissions

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

@@ -0,0 +1,13 @@
+package targets
+
+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) {
+	oldFn := AuthorizeSessionWorkerFilterFn
+	AuthorizeSessionWorkerFilterFn = AuthorizeSessionWithWorkerFilter
+
+	t.Cleanup(func() {
+		AuthorizeSessionWorkerFilterFn = oldFn
+	})
+}

internal/tests/api/credentials/credentials_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/daemon/controller"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	_ "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets/tcp"
 	"github.com/hashicorp/boundary/internal/daemon/worker"
 	"github.com/hashicorp/boundary/internal/iam"
@@ -405,6 +406,9 @@ func TestUpdateAfterKeyRotation(t *testing.T) {
 	logger := hclog.New(&hclog.LoggerOptions{
 		Level: hclog.Trace,
 	})
+
+	tg.SetupSuiteTargetFilters(t)
+
 	tc := controller.NewTestController(t, &controller.TestControllerOpts{SchedulerRunJobInterval: 100 * time.Millisecond})
 	ctx := tc.Context()
 	t.Cleanup(tc.Shutdown)

internal/tests/cluster/session_cleanup_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/daemon/controller"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/worker"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/session"
@@ -75,6 +76,7 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin
 	return func(t *testing.T) {
 		t.Parallel()
 		require := require.New(t)
+		tg.SetupSuiteTargetFilters(t)
 		logger := hclog.New(&hclog.LoggerOptions{
 			Name:  t.Name(),
 			Level: hclog.Trace,
@@ -201,6 +203,7 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing
 	const op = "cluster.testWorkerSessionCleanupMulti"
 	return func(t *testing.T) {
 		t.Parallel()
+		tg.SetupSuiteTargetFilters(t)
 		require := require.New(t)
 		logger := hclog.New(&hclog.LoggerOptions{
 			Name:  t.Name(),

internal/tests/cluster/worker_bytesupdown_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/daemon/controller"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/worker"
 	"github.com/hashicorp/boundary/internal/tests/helper"
 	"github.com/hashicorp/dawdle"
@@ -20,6 +21,8 @@ func TestWorkerBytesUpDown(t *testing.T) {
 	t.Parallel()
 	require := require.New(t)
 
+	tg.SetupSuiteTargetFilters(t)
+
 	logger := hclog.New(&hclog.LoggerOptions{
 		Name:  t.Name(),
 		Level: hclog.Trace,

internal/tests/cluster/worker_proxy_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/daemon/controller"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/worker"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/tests/helper"
@@ -19,6 +20,9 @@ import (
 func TestWorkerSessionProxyMultipleConnections(t *testing.T) {
 	const op = "cluster.TestWorkerSessionMultipleConnections"
 	t.Parallel()
+
+	tg.SetupSuiteTargetFilters(t)
+
 	require := require.New(t)
 	logger := hclog.New(&hclog.LoggerOptions{
 		Name:  t.Name(),

internal/tests/cluster/x509_verification_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/daemon/controller"
+	tg "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
 	"github.com/hashicorp/boundary/internal/daemon/worker"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/tests/helper"
@@ -29,6 +30,7 @@ func TestCustomX509Verification_Client(t *testing.T) {
 	ctx := context.Background()
 	ec := event.TestEventerConfig(t, "TestWorkerReplay", event.TestWithObservationSink(t), event.TestWithSysSink(t))
 	testLock := &sync.Mutex{}
+	tg.SetupSuiteTargetFilters(t)
 	logger := hclog.New(&hclog.LoggerOptions{
 		Mutex: testLock,
 		Name:  "test",
@@ -174,6 +176,7 @@ func testCustomX509Verification_Server(ec event.TestConfig, certPool *x509.CertP
 		t.Parallel()
 		req := require.New(t)
 		ctx := context.Background()
+		tg.SetupSuiteTargetFilters(t)
 
 		conf, err := config.DevController()
 		conf.Eventing = &ec.EventerConfig