fix(substrate): limit Substrate views to watched namespaces

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

@@ -32,7 +32,9 @@ type ATEClient interface {
 }
 
 type Service struct {
-	kubeClient         client.Client
+	kubeClient client.Client
+	// Sorted and unique: the order Substrate pages are read in. A repeat must not read
+	// an atespace twice. Empty means every namespace.
 	observedNamespaces []string
 	authorizer         auth.Authorizer
 	ateClient          ATEClient
@@ -51,7 +53,7 @@ func NewService(
 ) *Service {
 	return &Service{
 		kubeClient:         kubeClient,
-		observedNamespaces: slices.Clone(observedNamespaces),
+		observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))),
 		authorizer:         authorizer,
 		ateClient:          ateClient,
 	}
@@ -172,24 +174,27 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev
 	return workerPoolList.Items, nil
 }
 
-// substrateActorTemplates drains upstream pagination for the configuration-sized list.
+// substrateActorTemplates lists the templates of the requested atespace, or of each
+// watched one. The list is configuration-sized.
 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
+	templates := []*ateapipb.ActorTemplate{}
+	for _, atespace := range s.substrateNamespaces(atespace) {
+		page, err := s.ateClient.ListActorTemplates(ctx, atespace)
+		if err != nil {
+			return nil, err
+		}
+		for _, template := range page {
+			// ate-api scoped the list already. This is a second guard.
+			if template != nil && (atespace == "" || template.GetMetadata().GetAtespace() == atespace) {
+				// Only expose inventory fields: containers can contain resolved credentials.
+				templates = append(templates, &ateapipb.ActorTemplate{
+					Metadata:       template.GetMetadata(),
+					Status:         template.GetStatus(),
+					SandboxConfig:  template.GetSandboxConfig(),
+					WorkerSelector: template.GetWorkerSelector(),
+				})
+			}
 		}
-		// Only expose inventory fields: containers can contain resolved credentials.
-		templates = append(templates, &ateapipb.ActorTemplate{
-			Metadata:       template.GetMetadata(),
-			Status:         template.GetStatus(),
-			SandboxConfig:  template.GetSandboxConfig(),
-			WorkerSelector: template.GetWorkerSelector(),
-		})
 	}
 
 	slices.SortStableFunc(templates, func(left, right *ateapipb.ActorTemplate) int {

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

@@ -1,9 +1,11 @@
 package system_test
 
 import (
+	"cmp"
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"testing"
@@ -12,6 +14,8 @@ import (
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -44,40 +48,96 @@ type fakeATEClient struct {
 	// for, so a fake that always fills the request would hide a caller that stops
 	// following the token as soon as it has enough rows.
 	pageSize int
+	// The ceiling of a worker page, when it differs from pageSize.
+	workerPageSize 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
+	// Every call in order, so a test can assert on what reached ate-api.
+	calls []ateCall
+	// ignoreAtespace answers an actor or template list with the rows of every
+	// atespace, as an ate-api that drops the filter would.
+	ignoreAtespace bool
+	// failCall fails the call with this 1-based number with failErr.
+	failCall int
+	failErr  error
+}
+
+// ateCall is one ate-api call: the method, the atespace, the token and the page size.
+type ateCall struct {
+	method, scope, token string
+	size                 int32
+}
+
+// record notes a call and answers with the error it is set up to fail with.
+func (client *fakeATEClient) record(method, scope, token string, size int32) error {
+	client.calls = append(client.calls, ateCall{method: method, scope: scope, token: token, size: size})
+	if client.failCall == len(client.calls) {
+		return client.failErr
+	}
+	return client.err
 }
 
 func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) {
+	if err := client.record("ListActorTemplates", atespace, "", 0); err != nil {
+		return nil, err
+	}
 	templates := []*ateapipb.ActorTemplate{}
 	for _, template := range client.templates {
-		if atespace == "" || template.GetMetadata().GetAtespace() == atespace {
+		if atespace == "" || client.ignoreAtespace || template.GetMetadata().GetAtespace() == atespace {
 			templates = append(templates, template)
 		}
 	}
-	return templates, client.err
+	return templates, nil
 }
 
 func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) {
 	client.actorReads++
-	if err := client.readError(client.actorReads); err != nil {
+	if err := client.record("ListActorsPage", atespace, pageToken, pageSize); err != nil {
 		return nil, "", err
 	}
-	return fakePage(actorsInAtespace(client.actors, atespace), client.pageSize, pageSize, pageToken)
+	scope := atespace
+	if client.ignoreAtespace {
+		scope = ""
+	}
+	return actorPage(actorsInAtespace(client.actors, scope), atespace, client.pageSize, pageSize, pageToken)
 }
 
 func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) {
 	client.workerReads++
-	if err := client.readError(client.workerReads); err != nil {
+	if err := client.record("ListWorkersPage", "", pageToken, pageSize); err != nil {
 		return nil, "", err
 	}
-	return fakePage(client.workers, client.pageSize, pageSize, pageToken)
+	return fakePage(client.workers, cmp.Or(client.workerPageSize, client.pageSize), pageSize, pageToken)
 }
 
-func (client *fakeATEClient) readError(int) error {
-	return client.err
+// actorPage pages actors the way ate-api does: in atespace and name order, after the
+// last row of the token, with a token bound to the atespace it was issued for. A token
+// sent with another atespace is refused, as ate-api refuses it.
+func actorPage(actors []*ateapipb.Actor, atespace string, ceiling int, requested int32, pageToken string) ([]*ateapipb.Actor, string, error) {
+	key := func(actor *ateapipb.Actor) string {
+		return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName()
+	}
+	rows := slices.SortedStableFunc(slices.Values(actors), func(left, right *ateapipb.Actor) int {
+		return strings.Compare(key(left), key(right))
+	})
+	prefix := "upstream:" + atespace + ":"
+	if pageToken != "" {
+		last, ok := strings.CutPrefix(pageToken, prefix)
+		if !ok {
+			return nil, "", status.Error(codes.InvalidArgument, "invalid page_token")
+		}
+		rows = slices.DeleteFunc(rows, func(actor *ateapipb.Actor) bool { return key(actor) <= last })
+	}
+	pageSize := ceiling
+	if requested > 0 && (ceiling <= 0 || int(requested) < ceiling) {
+		pageSize = int(requested)
+	}
+	if pageSize <= 0 || pageSize >= len(rows) {
+		return rows, "", nil
+	}
+	return rows[:pageSize], prefix + key(rows[pageSize-1]), nil
 }
 
 // fakePage slices rows the way ate-api pages them: an opaque token, empty on the last
@@ -266,6 +326,7 @@ func TestListSubstrateWorkers(t *testing.T) {
 		{WorkerNamespace: "team", WorkerPool: "pool", WorkerPod: "worker-0", Status: &ateapipb.WorkerStatus{Allocated: &ateapipb.WorkerResources{Actors: 2}}},
 		{WorkerNamespace: "other", WorkerPool: "pool", WorkerPod: "worker-1"},
 		{WorkerNamespace: "team", WorkerPool: "pool", WorkerPod: "worker-2"},
+		{WorkerPool: "pool", WorkerPod: "names-no-namespace"},
 	}}
 	service := system.NewService(nil, nil, &pkgAuth.NoopAuthorizer{}, ateClient)
 
@@ -529,7 +590,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, nil, &pkgAuth.NoopAuthorizer{}, ateClient)
 	for _, tc := range []struct {
 		atespace string
 		count    int64
@@ -573,36 +634,72 @@ func TestSubstrateScopesAreIndependent(t *testing.T) {
 	assert.Same(t, actor, page.Actors[0])
 }
 
-type emptyPageATEClient struct {
-	fakeATEClient
-	t *testing.T
-}
-
-func (c *emptyPageATEClient) ListActorsPage(_ context.Context, atespace string, size int32, token string) ([]*ateapipb.Actor, string, error) {
-	assert.Equal(c.t, "team", atespace)
-	assert.Equal(c.t, int32(7), size)
-	assert.Equal(c.t, "upstream:opaque/cursor=", token)
-	return nil, "upstream:next/cursor=", nil
+// A controller that watches namespaces must not read another tenant's namespace or
+// atespace, and must not ask ate-api or Kubernetes for it.
+func TestSubstrateRefusesUnwatchedNamespaces(t *testing.T) {
+	ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}})
+	for _, tc := range []struct {
+		name string
+		read func(*system.Service) error
+	}{
+		{"actors in another atespace", func(service *system.Service) error {
+			_, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"})
+			return err
+		}},
+		{"actors in a watched atespace with other case", func(service *system.Service) error {
+			_, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "Tenant-A"})
+			return err
+		}},
+		{"actors in a watched atespace with a trailing space", func(service *system.Service) error {
+			_, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-a "})
+			return err
+		}},
+		{"a summary of another atespace", func(service *system.Service) error {
+			_, err := service.GetSubstrateSummary(ctx, "", "tenant-b")
+			return err
+		}},
+		{"a summary of another namespace", func(service *system.Service) error {
+			_, err := service.GetSubstrateSummary(ctx, "tenant-b", "")
+			return err
+		}},
+		{"a summary of a watched namespace and another atespace", func(service *system.Service) error {
+			_, err := service.GetSubstrateSummary(ctx, "tenant-a", "tenant-b")
+			return err
+		}},
+		{"workers in another namespace", func(service *system.Service) error {
+			_, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "tenant-b"})
+			return err
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			kubeLists := 0
+			ateClient := pagingATEClient()
+			kubeClient := pagingKubeClient(t, interceptor.Funcs{List: func(ctx context.Context, c ctrlclient.WithWatch, list ctrlclient.ObjectList, opts ...ctrlclient.ListOption) error {
+				kubeLists++
+				return c.List(ctx, list, opts...)
+			}})
+			service := system.NewService(kubeClient, []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient)
+
+			err := tc.read(service)
+			assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), "err = %v", err)
+			assert.Empty(t, ateClient.calls, "a refused read must not reach ate-api")
+			assert.Zero(t, kubeLists, "a refused read must not list WorkerPools")
+		})
+	}
 }
 
-func (c *emptyPageATEClient) ListWorkersPage(_ context.Context, size int32, token string) ([]*ateapipb.Worker, string, error) {
-	assert.Equal(c.t, int32(7), size)
-	assert.Equal(c.t, "upstream:opaque/cursor=", token)
-	return []*ateapipb.Worker{{WorkerNamespace: "other"}}, "upstream:next/cursor=", nil
+func actorNames(actors []*ateapipb.Actor) []string {
+	names := []string{}
+	for _, actor := range actors {
+		names = append(names, actor.GetMetadata().GetName())
+	}
+	return names
 }
 
-func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) {
-	ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}})
-	service := system.NewService(nil, []string{"team"}, &pkgAuth.NoopAuthorizer{}, &emptyPageATEClient{t: t})
-	page := &apiv1.PageRequest{Limit: 7, PageToken: "upstream:opaque/cursor="}
-	actors, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "team", Page: page})
-	require.NoError(t, err)
-	assert.Empty(t, actors.Actors)
-	assert.Empty(t, actors.ATEAPIError)
-	assert.Equal(t, "upstream:next/cursor=", actors.NextPageToken)
-	workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Page: page})
-	require.NoError(t, err)
-	assert.Empty(t, workers.Workers)
-	assert.Empty(t, workers.ATEAPIError)
-	assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken)
+func templateNames(templates []*ateapipb.ActorTemplate) []string {
+	names := []string{}
+	for _, template := range templates {
+		names = append(names, template.GetMetadata().GetName())
+	}
+	return names
 }