fix(substrate): scope actors and templates to the watched namespaces

go/core/internal/service/system/service.go

@@ -51,7 +51,7 @@ func NewService(
 ) *Service {
 	return &Service{
 		kubeClient:         kubeClient,
-		observedNamespaces: slices.Clone(observedNamespaces),
+		observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))),
 		authorizer:         authorizer,
 		ateClient:          ateClient,
 	}
@@ -148,16 +148,6 @@ func namespacesFromNames(names []string) []Namespace {
 	return result
 }
 
-func (s *Service) substrateNamespaces(requested string) []string {
-	if requested != "" {
-		return []string{requested}
-	}
-	if len(s.observedNamespaces) > 0 {
-		return slices.Clone(s.observedNamespaces)
-	}
-	return []string{""}
-}
-
 func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev1alpha1.WorkerPool, error) {
 	var options []client.ListOption
 	if namespace != "" {
@@ -173,22 +163,24 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev
 }
 
 // substrateActorTemplates drains upstream pagination for the configuration-sized list.
-func (s *Service) substrateActorTemplates(ctx context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) {
-	templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace)
-	if err != nil {
-		return nil, err
-	}
-	templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI))
-	for _, template := range templatesFromAPI {
-		if template == nil {
-			continue
+func (s *Service) substrateActorTemplates(ctx context.Context, scope substrateScope) ([]*ateapipb.ActorTemplate, error) {
+	templates := []*ateapipb.ActorTemplate{}
+	for _, atespace := range scope.listNamespaces() {
+		templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace)
+		if err != nil {
+			return nil, err
+		}
+		for _, template := range templatesFromAPI {
+			if template == nil {
+				continue
+			}
+			// Only expose inventory fields: containers can contain resolved credentials.
+			templates = append(templates, &ateapipb.ActorTemplate{
+				Metadata:       template.GetMetadata(),
+				Status:         template.GetStatus(),
+				WorkerSelector: template.GetWorkerSelector(),
+			})
 		}
-		// Only expose inventory fields: containers can contain resolved credentials.
-		templates = append(templates, &ateapipb.ActorTemplate{
-			Metadata:       template.GetMetadata(),
-			Status:         template.GetStatus(),
-			WorkerSelector: template.GetWorkerSelector(),
-		})
 	}
 
 	slices.SortStableFunc(templates, func(left, right *ateapipb.ActorTemplate) int {

go/core/internal/service/system/service_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"testing"
@@ -46,11 +47,14 @@ type fakeATEClient struct {
 	pageSize int
 	// How many page reads each list has taken, so a test can assert that the token
 	// was followed rather than that the rows came back.
-	actorReads  int
-	workerReads int
+	actorReads        int
+	workerReads       int
+	actorAtespaces    []string
+	templateAtespaces []string
 }
 
 func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) {
+	client.templateAtespaces = append(client.templateAtespaces, atespace)
 	templates := []*ateapipb.ActorTemplate{}
 	for _, template := range client.templates {
 		if atespace == "" || template.GetMetadata().GetAtespace() == atespace {
@@ -62,6 +66,7 @@ func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace stri
 
 func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) {
 	client.actorReads++
+	client.actorAtespaces = append(client.actorAtespaces, atespace)
 	if err := client.readError(client.actorReads); err != nil {
 		return nil, "", err
 	}
@@ -527,7 +532,7 @@ func TestSubstrateScopesAreIndependent(t *testing.T) {
 		},
 		workers: []*ateapipb.Worker{{WorkerNamespace: "workers", WorkerPod: "pod"}},
 	}
-	service := system.NewService(kubeClient, []string{"workers"}, &pkgAuth.NoopAuthorizer{}, ateClient)
+	service := system.NewService(kubeClient, []string{"workers", "team-a", "shared"}, &pkgAuth.NoopAuthorizer{}, ateClient)
 	for _, tc := range []struct {
 		atespace string
 		count    int64
@@ -536,7 +541,6 @@ func TestSubstrateScopesAreIndependent(t *testing.T) {
 		{atespace: "", count: 2, running: 1},
 		{atespace: "team-a", count: 1, running: 1},
 		{atespace: "shared", count: 1, running: 0},
-		{atespace: "missing", count: 0, running: 0},
 	} {
 		t.Run(tc.atespace, func(t *testing.T) {
 			summary, err := service.GetSubstrateSummary(ctx, "", tc.atespace)
@@ -604,3 +608,177 @@ func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) {
 	assert.Empty(t, workers.ATEAPIError)
 	assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken)
 }
+
+func TestSubstrateTenantIsolation(t *testing.T) {
+	scheme := runtime.NewScheme()
+	require.NoError(t, atev1alpha1.AddToScheme(scheme))
+	ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}})
+	ateClient := &fakeATEClient{
+		pageSize: 1,
+		actors: []*ateapipb.Actor{
+			substrateActor("session-b", "tenant-b", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-b", "pod-b"),
+			substrateActor("session-a", "tenant-a", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""),
+			substrateActor("session-c", "tenant-c", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-c", "pod-c"),
+			substrateActor("session-b2", "tenant-b", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""),
+		},
+		templates: []*ateapipb.ActorTemplate{
+			{Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-a", Name: "agent-a"}},
+			{Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-b", Name: "agent-b"}},
+			{Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-c", Name: "agent-c"}},
+		},
+		workers: []*ateapipb.Worker{
+			{WorkerNamespace: "tenant-a", WorkerPod: "pod-a"},
+			{WorkerNamespace: "tenant-b", WorkerPod: "pod-b"},
+			{WorkerNamespace: "tenant-c", WorkerPod: "pod-c"},
+			{WorkerPod: "pod-unlabelled"},
+		},
+	}
+	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
+		&atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-a", Name: "pool-a"}},
+		&atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-b", Name: "pool-b"}},
+		&atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: "tenant-c", Name: "pool-c"}},
+	).Build()
+	newService := func(observedNamespaces []string) *system.Service {
+		return system.NewService(kubeClient, observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient)
+	}
+	tenantA := newService([]string{"tenant-a"})
+	tenantsAC := newService([]string{"tenant-a", "tenant-c"})
+	tenantsACDuplicated := newService([]string{"tenant-a", "tenant-c", "tenant-a"})
+	clusterWide := newService(nil)
+
+	listAllActors := func(t *testing.T, service *system.Service, atespace string) []string {
+		var names []string
+		request := &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: 1}}
+		for {
+			page, err := service.ListSubstrateActors(ctx, request)
+			require.NoError(t, err)
+			for _, actor := range page.Actors {
+				names = append(names, actor.GetMetadata().GetAtespace()+"/"+actor.GetMetadata().GetName())
+			}
+			if page.NextPageToken == "" {
+				return names
+			}
+			request.Page.PageToken = page.NextPageToken
+		}
+	}
+	workerPods := func(t *testing.T, service *system.Service) []string {
+		var pods []string
+		request := &apiv1.ListSubstrateWorkersRequest{Page: &apiv1.PageRequest{}}
+		for {
+			page, err := service.ListSubstrateWorkers(ctx, request)
+			require.NoError(t, err)
+			for _, worker := range page.Workers {
+				pods = append(pods, worker.GetWorkerPod())
+			}
+			if page.NextPageToken == "" {
+				return pods
+			}
+			request.Page.PageToken = page.NextPageToken
+		}
+	}
+	poolNames := func(summary system.SubstrateSummary) []string {
+		var names []string
+		for _, pool := range summary.WorkerPools {
+			names = append(names, pool.Namespace+"/"+pool.Name)
+		}
+		return names
+	}
+	templateNames := func(summary system.SubstrateSummary) []string {
+		var names []string
+		for _, template := range summary.ActorTemplates {
+			names = append(names, template.GetMetadata().GetName())
+		}
+		return names
+	}
+
+	t.Run("lists only its own actors", func(t *testing.T) {
+		assert.Equal(t, []string{"tenant-a/session-a"}, listAllActors(t, tenantA, ""))
+	})
+
+	t.Run("summarises only its own atespace", func(t *testing.T) {
+		summary, err := tenantA.GetSubstrateSummary(ctx, "", "")
+		require.NoError(t, err)
+		require.Empty(t, summary.ATEAPIError)
+		assert.Equal(t, []string{"agent-a"}, templateNames(summary))
+		assert.Equal(t, []string{"tenant-a/pool-a"}, poolNames(summary))
+		assert.Equal(t, int64(1), summary.ActorCount)
+		assert.Equal(t, int64(0), summary.RunningActorCount)
+		assert.Equal(t, int64(1), summary.WorkerCount)
+	})
+
+	t.Run("hides a worker that reports no namespace", func(t *testing.T) {
+		assert.Equal(t, []string{"pod-a"}, workerPods(t, tenantA))
+		assert.Equal(t, []string{"pod-a", "pod-c"}, workerPods(t, tenantsAC))
+	})
+
+	t.Run("a controller watching two namespaces lists only those atespaces", func(t *testing.T) {
+		assert.Equal(t, []string{"tenant-a/session-a", "tenant-c/session-c"}, listAllActors(t, tenantsAC, ""))
+
+		first, err := tenantsAC.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}})
+		require.NoError(t, err)
+		assert.Empty(t, first.Actors, "the first upstream page holds only a tenant-b actor")
+		assert.NotEmpty(t, first.NextPageToken, "a page filtered to nothing still leads to the rest")
+	})
+
+	t.Run("a controller watching two namespaces asks ate-api for only those atespaces", func(t *testing.T) {
+		ateClient.actorAtespaces, ateClient.templateAtespaces = nil, nil
+		summary, err := tenantsAC.GetSubstrateSummary(ctx, "", "")
+		require.NoError(t, err)
+		require.Empty(t, summary.ATEAPIError)
+		assert.Equal(t, int64(2), summary.ActorCount)
+		assert.Equal(t, int64(1), summary.RunningActorCount)
+		assert.ElementsMatch(t, []system.SubstrateActorStatusCount{
+			{State: ateapipb.ActorState_ACTOR_STATE_RUNNING, Count: 1},
+			{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, Count: 1},
+		}, summary.ActorStatusCounts)
+		assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary))
+		assert.Equal(t, []string{"tenant-a/pool-a", "tenant-c/pool-c"}, poolNames(summary))
+		assert.Equal(t, int64(2), summary.WorkerCount)
+
+		assert.Equal(t, []string{"tenant-a", "tenant-c"}, slices.Compact(slices.Sorted(slices.Values(ateClient.actorAtespaces))))
+		assert.ElementsMatch(t, []string{"tenant-a", "tenant-c"}, ateClient.templateAtespaces)
+	})
+
+	t.Run("a namespace listed twice in the config is counted once", func(t *testing.T) {
+		summary, err := tenantsACDuplicated.GetSubstrateSummary(ctx, "", "")
+		require.NoError(t, err)
+		require.Empty(t, summary.ATEAPIError)
+		assert.Equal(t, int64(2), summary.ActorCount)
+		assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary))
+		assert.Equal(t, []string{"tenant-a/pool-a", "tenant-c/pool-c"}, poolNames(summary))
+		assert.Equal(t, int64(2), summary.WorkerCount)
+	})
+
+	t.Run("refuses another tenant's atespace or namespace", func(t *testing.T) {
+		_, err := tenantA.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"})
+		assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err)
+		assert.Contains(t, serviceerrors.MessageOf(err), "not watched by this controller", "the caller sees only the message, not the cause")
+		_, err = tenantA.GetSubstrateSummary(ctx, "", "tenant-b")
+		assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err)
+		_, err = tenantA.GetSubstrateSummary(ctx, "tenant-b", "")
+		assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err)
+		_, err = tenantA.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "tenant-b"})
+		assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err)
+	})
+
+	t.Run("a cluster-wide install still sees every tenant", func(t *testing.T) {
+		summary, err := clusterWide.GetSubstrateSummary(ctx, "", "")
+		require.NoError(t, err)
+		assert.Equal(t, int64(4), summary.ActorCount)
+		assert.Len(t, summary.ActorTemplates, 3)
+		assert.Equal(t, []string{"tenant-a/pool-a", "tenant-b/pool-b", "tenant-c/pool-c"}, poolNames(summary))
+		assert.Equal(t, int64(4), summary.WorkerCount)
+		assert.Equal(t, []string{"pod-a", "pod-b", "pod-c", "pod-unlabelled"}, workerPods(t, clusterWide))
+		assert.Equal(t, []string{"tenant-b/session-b", "tenant-b/session-b2"}, listAllActors(t, clusterWide, "tenant-b"))
+	})
+
+	t.Run("a cluster-wide install counts nothing in an atespace with no rows", func(t *testing.T) {
+		summary, err := clusterWide.GetSubstrateSummary(ctx, "", "missing")
+		require.NoError(t, err)
+		assert.Empty(t, summary.ATEAPIError)
+		assert.Zero(t, summary.ActorCount)
+		assert.Zero(t, summary.RunningActorCount)
+		assert.Empty(t, summary.ActorStatusCounts)
+		assert.Empty(t, summary.ActorTemplates)
+	})
+}