From 0faac5d435bfe5a2b8586410cbe4328bb07de980 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 20:16:29 -0400 Subject: [PATCH 01/12] test(substrate): reproduce one tenant reading another's Substrate inventory Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index c6e19b1dd1..5af4cefc89 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -606,3 +606,77 @@ func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) { assert.Empty(t, workers.ATEAPIError) assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken) } + +// Two kagent installs, each watching only its own namespace, share one ate-api. Each +// must see only its own atespace, and refuse a request naming the other. +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, "", ""), + }, + templates: []*ateapipb.ActorTemplate{ + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-a", Name: "agent-a"}}, + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-b", Name: "agent-b"}}, + }, + workers: []*ateapipb.Worker{ + {WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, + {WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + }, + } + tenantA := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + t.Run("lists only its own actors", func(t *testing.T) { + var names []string + request := &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}} + for { + page, err := tenantA.ListSubstrateActors(ctx, request) + require.NoError(t, err) + for _, actor := range page.Actors { + names = append(names, actor.GetMetadata().GetAtespace()+"/"+actor.GetMetadata().GetName()) + } + if page.NextPageToken == "" { + break + } + request.Page.PageToken = page.NextPageToken + } + assert.Equal(t, []string{"tenant-a/session-a"}, names) + }) + + 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) + require.Len(t, summary.ActorTemplates, 1) + assert.Equal(t, "agent-a", summary.ActorTemplates[0].GetMetadata().GetName()) + assert.Equal(t, int64(1), summary.ActorCount) + assert.Equal(t, int64(0), summary.RunningActorCount) + assert.Equal(t, int64(1), 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) + _, 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) { + clusterWide := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), nil, &pkgAuth.NoopAuthorizer{}, ateClient) + summary, err := clusterWide.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + assert.Equal(t, int64(2), summary.ActorCount) + assert.Len(t, summary.ActorTemplates, 2) + page, err := clusterWide.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + require.NoError(t, err) + require.Len(t, page.Actors, 1) + }) +} From 4ce42c30a72b6a2aab7d46f3e31facf6942244c6 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 20:21:09 -0400 Subject: [PATCH 02/12] fix(substrate): scope actors and templates to the watched namespaces A kagent that watches only its own namespace listed every tenant's actors and ActorTemplates from a shared Substrate, and served a foreign atespace or namespace when a caller named one. Resolve one scope from the watched namespaces for every Substrate read. An empty request means the watched namespaces, and a namespace or atespace outside them is refused with PermissionDenied. A cluster-wide install still sees everything. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 16 +-- .../internal/service/system/service_test.go | 3 +- go/core/internal/service/system/substrate.go | 121 ++++++++++++------ 3 files changed, 85 insertions(+), 55 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index f3dd63e716..a4aaf3aca9 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -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,14 +163,14 @@ 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) +func (s *Service) substrateActorTemplates(ctx context.Context, scope substrateScope) ([]*ateapipb.ActorTemplate, error) { + templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, scope.atespace()) if err != nil { return nil, err } templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI)) for _, template := range templatesFromAPI { - if template == nil { + if template == nil || !scope.contains(template.GetMetadata().GetAtespace()) { continue } // Only expose inventory fields: containers can contain resolved credentials. diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 5af4cefc89..d4ac94ecc6 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -529,7 +529,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 @@ -538,7 +538,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) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 33349bbff2..a5c225b9e3 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -76,7 +76,14 @@ type SubstrateActorStatusCount struct { // ate-api reports no totals, so every count costs a walk of its pages. The walk holds // one page at a time and keeps only tallies. func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, atespace string) (SubstrateSummary, error) { - namespaces, err := s.substrateScope(ctx, requestedNamespace) + if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { + return SubstrateSummary{}, err + } + workerScope, err := s.resolveSubstrateScope(requestedNamespace) + if err != nil { + return SubstrateSummary{}, err + } + actorScope, err := s.resolveSubstrateScope(atespace) if err != nil { return SubstrateSummary{}, err } @@ -88,7 +95,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a ComputedAt: time.Now().UTC(), } - for _, namespace := range namespaces { + for _, namespace := range workerScope.listNamespaces() { workerPools, err := s.listWorkerPools(ctx, namespace) if err != nil { return SubstrateSummary{}, serviceerrors.NewInternal("Failed to list substrate resources from Kubernetes", err) @@ -99,18 +106,16 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return strings.Compare(left.Namespace+"/"+left.Name, right.Namespace+"/"+right.Name) }) - allowAll, allowed := substrateScopeFilter(namespaces) - // Three independent reads: none gates the others, so one failure leaves the rest // counted rather than zeroing the whole summary. - if templates, err := s.substrateActorTemplates(ctx, atespace); err != nil { + if templates, err := s.substrateActorTemplates(ctx, actorScope); err != nil { result.recordATEError(ctx, err) } else { result.ActorTemplates = templates } statusCounts := map[ateapipb.ActorState]int64{} - if err := s.walkActors(ctx, atespace, func(actor *ateapipb.Actor) { + if err := s.walkActors(ctx, actorScope, func(actor *ateapipb.Actor) { if actor == nil { return } @@ -124,7 +129,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a } if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if worker == nil || !allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if !workerScope.containsWorker(worker) { return } result.WorkerCount++ @@ -145,19 +150,24 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors returns one upstream page without changing its order or token. +// ListSubstrateActors filters one upstream page by atespace and preserves its order and token. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { - if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { + scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) + if err != nil { return SubstrateActorPage{}, err } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, input.GetAtespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + actors, next, err := s.ateClient.ListActorsPage(ctx, scope.atespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) return result, nil } - result.Actors = actors + for _, actor := range actors { + if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { + result.Actors = append(result.Actors, actor) + } + } result.NextPageToken = next return result, nil } @@ -165,7 +175,7 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs // ListSubstrateWorkers filters one upstream page by namespace and preserves its token. // A page with no workers in scope may still have a next page. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { - namespaces, err := s.substrateScope(ctx, input.GetNamespace()) + scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { return SubstrateWorkerPage{}, err } @@ -176,9 +186,8 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) return result, nil } - allowAll, allowed := substrateScopeFilter(namespaces) for _, worker := range workers { - if worker != nil && allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if scope.containsWorker(worker) { result.Workers = append(result.Workers, worker) } } @@ -186,12 +195,16 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// walkActors visits every actor in the requested atespace, one page at a time. -func (s *Service) walkActors(ctx context.Context, atespace string, visit func(*ateapipb.Actor)) error { +// walkActors visits every actor in scope, one page at a time. +func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { - return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) + return s.ateClient.ListActorsPage(ctx, scope.atespace(), pageSize, pageToken) } - return walkSubstrate(ctx, read, visit) + return walkSubstrate(ctx, read, func(actor *ateapipb.Actor) { + if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { + visit(actor) + } + }) } // walkWorkers calls visit for every worker ate-api holds, one page at a time. @@ -225,36 +238,64 @@ func walkSubstrate[Row any]( return fmt.Errorf("ate-api did not finish paging after %d pages", maxATEPagesPerWalk) } -// substrateScope authorizes the caller and resolves the namespaces a read covers. -// ATE-only actor reads authorize independently of Kubernetes scope. -func (s *Service) substrateScope(ctx context.Context, requestedNamespace string) ([]string, error) { +// substrateScope is the set of namespaces a Substrate read covers. An atespace is named +// after the namespace of the agent it serves, so the same set bounds actors and templates. +type substrateScope struct { + all bool + namespaces []string +} + +// authorizeSubstrateScope authorizes the caller and resolves the scope of a read. +func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) (substrateScope, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { - return nil, err + return substrateScope{}, err } - return s.substrateNamespaces(requestedNamespace), nil + return s.resolveSubstrateScope(requested) } -// substrateScopeFilter turns resolved namespaces into the pair every row filter here -// takes: whether every namespace is in scope, and the set that is when it is not. -func substrateScopeFilter(namespaces []string) (bool, map[string]struct{}) { - allowAll := len(namespaces) == 1 && namespaces[0] == "" - allowed := make(map[string]struct{}, len(namespaces)) - for _, namespace := range namespaces { - if namespace != "" { - allowed[namespace] = struct{}{} - } +// resolveSubstrateScope narrows the watched namespaces to the requested one. A +// controller that watches only some namespaces refuses any other. +func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error) { + switch { + case requested == "" && len(s.observedNamespaces) == 0: + return substrateScope{all: true}, nil + case requested == "": + return substrateScope{namespaces: slices.Clone(s.observedNamespaces)}, nil + case len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, requested): + return substrateScope{namespaces: []string{requested}}, nil + default: + return substrateScope{}, serviceerrors.NewPermissionDenied("Not authorized", fmt.Errorf("namespace %q is not watched by this controller", requested)) + } +} + +func (scope substrateScope) contains(namespace string) bool { + return scope.all || slices.Contains(scope.namespaces, namespace) +} + +// containsWorker also admits a worker that reports no namespace. +func (scope substrateScope) containsWorker(worker *ateapipb.Worker) bool { + if worker == nil { + return false + } + namespace := strings.TrimSpace(worker.GetWorkerNamespace()) + return namespace == "" || scope.contains(namespace) +} + +// atespace is the atespace to ask ate-api for: the only one in scope, or every +// atespace ("") for the caller to filter. +func (scope substrateScope) atespace() string { + if len(scope.namespaces) == 1 { + return scope.namespaces[0] } - return allowAll, allowed + return "" } -// allowedWorkerNamespace filters workers by their Kubernetes namespace. -func allowedWorkerNamespace(namespace string, allowAll bool, allowed map[string]struct{}) bool { - namespace = strings.TrimSpace(namespace) - if allowAll || namespace == "" { - return true +// listNamespaces is the namespaces to list Kubernetes objects in, "" meaning all. +func (scope substrateScope) listNamespaces() []string { + if scope.all { + return []string{""} } - _, ok := allowed[namespace] - return ok + return scope.namespaces } func substratePageSize(requested int32) int32 { From 93f555f9c611403f5605fddaa0cff7ea3d3566a9 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 21:00:34 -0400 Subject: [PATCH 03/12] test(substrate): cover multi-namespace scope, unlabelled workers and refusal wording A controller watching tenant-a and tenant-c must never page tenant-b actors to the caller, must ask ate-api only for its own atespaces when it builds the summary, and must hide a worker that reports no namespace. The refusal message must say the namespace is not watched, since the gRPC layer sends only the message. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 112 +++++++++++++++--- 1 file changed, 97 insertions(+), 15 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index d4ac94ecc6..09a90c5844 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "strconv" "strings" "testing" @@ -48,9 +49,13 @@ type fakeATEClient struct { // was followed rather than that the rows came back. actorReads int workerReads int + // The atespaces ate-api was asked for, since asking for "" reads every tenant's rows. + 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 +67,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 } @@ -606,8 +612,8 @@ func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) { assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken) } -// Two kagent installs, each watching only its own namespace, share one ate-api. Each -// must see only its own atespace, and refuse a request naming the other. +// Kagent installs that each watch only some namespaces share one ate-api. Each must see +// only the atespaces it watches, and refuse a request naming another. func TestSubstrateTenantIsolation(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, atev1alpha1.AddToScheme(scheme)) @@ -617,49 +623,116 @@ func TestSubstrateTenantIsolation(t *testing.T) { 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_PAUSED, "", ""), }, 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"}, }, } - tenantA := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + newService := func(observedNamespaces []string) *system.Service { + return system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient) + } + tenantA := newService([]string{"tenant-a"}) + tenantsAC := newService([]string{"tenant-a", "tenant-c"}) + clusterWide := newService(nil) - t.Run("lists only its own actors", func(t *testing.T) { + listAllActors := func(t *testing.T, service *system.Service, atespace string) []string { var names []string - request := &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 1}} + request := &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: 1}} for { - page, err := tenantA.ListSubstrateActors(ctx, request) + 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 == "" { - break + 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 } - assert.Equal(t, []string{"tenant-a/session-a"}, 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) - require.Len(t, summary.ActorTemplates, 1) - assert.Equal(t, "agent-a", summary.ActorTemplates[0].GetMetadata().GetName()) + assert.Equal(t, []string{"agent-a"}, templateNames(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, 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("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", "") @@ -669,13 +742,22 @@ func TestSubstrateTenantIsolation(t *testing.T) { }) t.Run("a cluster-wide install still sees every tenant", func(t *testing.T) { - clusterWide := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), nil, &pkgAuth.NoopAuthorizer{}, ateClient) summary, err := clusterWide.GetSubstrateSummary(ctx, "", "") require.NoError(t, err) - assert.Equal(t, int64(2), summary.ActorCount) - assert.Len(t, summary.ActorTemplates, 2) - page, err := clusterWide.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + assert.Equal(t, int64(4), summary.ActorCount) + assert.Len(t, summary.ActorTemplates, 3) + 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) - require.Len(t, page.Actors, 1) + 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) }) } From 0f23fc5a31f27b989034a5c872f7ec37cd560276 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 21:02:39 -0400 Subject: [PATCH 04/12] fix(substrate): ask ate-api per watched atespace and hide unlabelled workers The summary for a controller watching several namespaces asked ate-api for every atespace and filtered the rows itself, so ate-api served every tenant's actors and templates to it. It now walks actors and drains templates once per atespace in scope. ListSubstrateActors still reads one upstream page across all atespaces and filters it, because one upstream token cannot span several requests. A worker that reports no namespace now belongs only to a cluster-wide scope, and the refusal message names the namespace that is not watched, since the gRPC layer sends only the message. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 32 ++++++----- .../internal/service/system/service_test.go | 7 +-- go/core/internal/service/system/substrate.go | 57 ++++++------------- 3 files changed, 36 insertions(+), 60 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index a4aaf3aca9..9f35284df5 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -164,22 +164,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, scope substrateScope) ([]*ateapipb.ActorTemplate, error) { - templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, scope.atespace()) - if err != nil { - return nil, err - } - templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI)) - for _, template := range templatesFromAPI { - if template == nil || !scope.contains(template.GetMetadata().GetAtespace()) { - continue + 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(), + 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 { diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 09a90c5844..2dcdd17de2 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -47,9 +47,8 @@ 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 - // The atespaces ate-api was asked for, since asking for "" reads every tenant's rows. + actorReads int + workerReads int actorAtespaces []string templateAtespaces []string } @@ -612,8 +611,6 @@ func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) { assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken) } -// Kagent installs that each watch only some namespaces share one ate-api. Each must see -// only the atespaces it watches, and refuse a request naming another. func TestSubstrateTenantIsolation(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, atev1alpha1.AddToScheme(scheme)) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index a5c225b9e3..6c15366762 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -76,10 +76,7 @@ type SubstrateActorStatusCount struct { // ate-api reports no totals, so every count costs a walk of its pages. The walk holds // one page at a time and keeps only tallies. func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, atespace string) (SubstrateSummary, error) { - if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { - return SubstrateSummary{}, err - } - workerScope, err := s.resolveSubstrateScope(requestedNamespace) + workerScope, err := s.authorizeSubstrateScope(ctx, requestedNamespace) if err != nil { return SubstrateSummary{}, err } @@ -129,7 +126,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a } if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if !workerScope.containsWorker(worker) { + if worker == nil || !workerScope.contains(worker.GetWorkerNamespace()) { return } result.WorkerCount++ @@ -156,8 +153,12 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs if err != nil { return SubstrateActorPage{}, err } + atespace := "" + if len(scope.namespaces) == 1 { + atespace = scope.namespaces[0] + } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, scope.atespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + actors, next, err := s.ateClient.ListActorsPage(ctx, atespace, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) @@ -187,7 +188,7 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } for _, worker := range workers { - if scope.containsWorker(worker) { + if worker != nil && scope.contains(worker.GetWorkerNamespace()) { result.Workers = append(result.Workers, worker) } } @@ -195,16 +196,16 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// walkActors visits every actor in scope, one page at a time. func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { - read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { - return s.ateClient.ListActorsPage(ctx, scope.atespace(), pageSize, pageToken) - } - return walkSubstrate(ctx, read, func(actor *ateapipb.Actor) { - if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { - visit(actor) + for _, atespace := range scope.listNamespaces() { + read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { + return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) } - }) + if err := walkSubstrate(ctx, read, visit); err != nil { + return err + } + } + return nil } // walkWorkers calls visit for every worker ate-api holds, one page at a time. @@ -238,14 +239,11 @@ func walkSubstrate[Row any]( return fmt.Errorf("ate-api did not finish paging after %d pages", maxATEPagesPerWalk) } -// substrateScope is the set of namespaces a Substrate read covers. An atespace is named -// after the namespace of the agent it serves, so the same set bounds actors and templates. type substrateScope struct { all bool namespaces []string } -// authorizeSubstrateScope authorizes the caller and resolves the scope of a read. func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) (substrateScope, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { return substrateScope{}, err @@ -253,8 +251,6 @@ func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) return s.resolveSubstrateScope(requested) } -// resolveSubstrateScope narrows the watched namespaces to the requested one. A -// controller that watches only some namespaces refuses any other. func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error) { switch { case requested == "" && len(s.observedNamespaces) == 0: @@ -264,7 +260,7 @@ func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error case len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, requested): return substrateScope{namespaces: []string{requested}}, nil default: - return substrateScope{}, serviceerrors.NewPermissionDenied("Not authorized", fmt.Errorf("namespace %q is not watched by this controller", requested)) + return substrateScope{}, serviceerrors.NewPermissionDenied(fmt.Sprintf("Namespace %q is not watched by this controller", requested), nil) } } @@ -272,25 +268,6 @@ func (scope substrateScope) contains(namespace string) bool { return scope.all || slices.Contains(scope.namespaces, namespace) } -// containsWorker also admits a worker that reports no namespace. -func (scope substrateScope) containsWorker(worker *ateapipb.Worker) bool { - if worker == nil { - return false - } - namespace := strings.TrimSpace(worker.GetWorkerNamespace()) - return namespace == "" || scope.contains(namespace) -} - -// atespace is the atespace to ask ate-api for: the only one in scope, or every -// atespace ("") for the caller to filter. -func (scope substrateScope) atespace() string { - if len(scope.namespaces) == 1 { - return scope.namespaces[0] - } - return "" -} - -// listNamespaces is the namespaces to list Kubernetes objects in, "" meaning all. func (scope substrateScope) listNamespaces() []string { if scope.all { return []string{""} From f8f6a654505b28091dd0054ecc365600bdec0f99 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:04 -0400 Subject: [PATCH 05/12] test(substrate): actor page tokens must name only watched atespaces A controller watching tenant-a and tenant-c must page through its own actors in watched-atespace order, scope every ate-api read to one watched atespace, never hand out an empty page with a token, and never return a token that names a tenant-b actor. A token it did not issue is InvalidArgument, without echoing the token and without an ate-api call. The cluster-wide pass-through test now runs cluster-wide, and the assertion that a filtered-out first page still leads on is dropped: that page is the leak. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 9 +- .../service/system/substrate_paging_test.go | 122 ++++++++++++++++++ 2 files changed, 124 insertions(+), 7 deletions(-) create mode 100644 go/core/internal/service/system/substrate_paging_test.go diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index 2dcdd17de2..cbfd881812 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -597,14 +597,14 @@ func (c *emptyPageATEClient) ListWorkersPage(_ context.Context, size int32, toke 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}) + service := system.NewService(nil, nil, &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}) + workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: page}) require.NoError(t, err) assert.Empty(t, workers.Workers) assert.Empty(t, workers.ATEAPIError) @@ -701,11 +701,6 @@ func TestSubstrateTenantIsolation(t *testing.T) { 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) { diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go new file mode 100644 index 0000000000..324e48a024 --- /dev/null +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -0,0 +1,122 @@ +package system_test + +import ( + "context" + "encoding/base64" + "fmt" + "strings" + "testing" + + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" + "github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors" + "github.com/kagent-dev/kagent/go/core/internal/service/system" + pkgAuth "github.com/kagent-dev/kagent/go/core/pkg/auth" +) + +func pagingContext(t *testing.T) context.Context { + return pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) +} + +// pagingATEClient mixes three tenants' rows in ate-api order, two rows a page. Every +// tenant-b name holds "tenant-b", so a test can look for it in any text. +func pagingATEClient() *fakeATEClient { + client := &fakeATEClient{pageSize: 2} + for _, actor := range []string{"tenant-a/a-1", "tenant-b/b-1", "tenant-c/c-1", "tenant-a/a-2", "tenant-b/b-2", "tenant-c/c-2", "tenant-a/a-3"} { + atespace, name, _ := strings.Cut(actor, "/") + client.actors = append(client.actors, substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + } + return client +} + +// pagingService watches tenant-a and tenant-c, listed out of order and with a repeat. +func pagingService(ateClient system.ATEClient) *system.Service { + return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) +} + +// assertOwnToken fails when a page token names a tenant-b row, or when an empty page +// carries a token: that would tell the caller a hidden row exists. +func assertOwnToken(t *testing.T, rows int, pageToken string) { + t.Helper() + raw, _ := base64.RawURLEncoding.DecodeString(pageToken) + assert.NotContains(t, pageToken+string(raw), "tenant-b") + if rows == 0 { + assert.Empty(t, pageToken, "an empty page must not lead on") + } +} + +func walkActorNames(t *testing.T, service *system.Service, atespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Actors), page.NextPageToken) + names = append(names, actorNames(page.Actors)...) + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +func TestListSubstrateActorsPagesWatchedAtespaces(t *testing.T) { + for atespace, want := range map[string][]string{ + "": {"a-1", "a-2", "a-3", "c-1", "c-2"}, + "tenant-c": {"c-1", "c-2"}, + } { + for _, limit := range []int32{1, 2, 3, 4, 100} { + t.Run(fmt.Sprintf("%q/%d", atespace, limit), func(t *testing.T) { + ateClient := pagingATEClient() + assert.Equal(t, want, walkActorNames(t, pagingService(ateClient), atespace, limit)) + assert.Subset(t, []string{"tenant-a", "tenant-c"}, ateClient.actorAtespaces, "every ate-api read is scoped to a watched atespace") + }) + } + } +} + +func actorNames(actors []*ateapipb.Actor) []string { + names := []string{} + for _, actor := range actors { + names = append(names, actor.GetMetadata().GetName()) + } + return names +} + +func actorPageToken(raw string) string { + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// A token this controller did not issue never reaches ate-api and never restarts the +// walk at page one. Its error repeats neither the token nor a tenant-b name. +func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + atespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "no separator": {pageToken: actorPageToken("tenant-a")}, + "empty atespace": {pageToken: actorPageToken("/upstream:2")}, + "unwatched atespace": {pageToken: actorPageToken("tenant-b/upstream:2")}, + "outside the request": {atespace: "tenant-c", pageToken: actorPageToken("tenant-a/upstream:2")}, + "ate-api actor token": {pageToken: actorPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, + "over 1 KiB, else good": {pageToken: actorPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: tc.atespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.actorReads) + }) + } +} From a1cfe05051b40ed84426d2ecc30ca4d0db8e442e Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:28 -0400 Subject: [PATCH 06/12] fix(substrate): page actors through watched atespaces only With two or more watched namespaces, ListSubstrateActors asked ate-api for every atespace and passed its page token through. ate-api tokens are readable base64 JSON holding the last row's atespace and name, so a token could name another tenant's actor, and a page filtered to nothing still carried a token. A controller that watches some namespaces now reads each watched atespace with its own scoped list, in sorted order, and returns its own token: the atespace to read and ate-api's token inside it. A token it did not issue, or one longer than 1 KiB, is InvalidArgument and never reaches ate-api. NewService sorts and dedupes the watched list, since tokens resume in its order. A cluster-wide controller still passes ate-api's page and token through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/service.go | 5 +- go/core/internal/service/system/substrate.go | 66 ++++++++++++++++++-- 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 9f35284df5..7389221ebd 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -32,7 +32,8 @@ type ATEClient interface { } type Service struct { - kubeClient client.Client + kubeClient client.Client + // Sorted and unique: actor page tokens resume in this order. Empty means every namespace. observedNamespaces []string authorizer auth.Authorizer ateClient ATEClient @@ -51,7 +52,7 @@ func NewService( ) *Service { return &Service{ kubeClient: kubeClient, - observedNamespaces: slices.Clone(observedNamespaces), + observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))), authorizer: authorizer, ateClient: ateClient, } diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 6c15366762..1d5eeb6dd5 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -2,6 +2,7 @@ package system import ( "context" + "encoding/base64" "fmt" "maps" "slices" @@ -147,18 +148,28 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors filters one upstream page by atespace and preserves its order and token. +// ListSubstrateActors returns one page of actors. A cluster-wide controller passes +// ate-api's page and token through. A controller that watches some namespaces reads +// each watched atespace with its own scoped list, so no page token names an actor of +// a namespace it does not watch. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) if err != nil { return SubstrateActorPage{}, err } - atespace := "" - if len(scope.namespaces) == 1 { - atespace = scope.namespaces[0] + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + var actors []*ateapipb.Actor + var next string + if len(s.observedNamespaces) == 0 { + actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) + } else { + start, inner, tokenErr := parseActorPageToken(scope.namespaces, token) + if tokenErr != nil { + return SubstrateActorPage{}, tokenErr + } + actors, next, err = s.listScopedActors(ctx, scope.namespaces[start:], inner, limit) } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, atespace, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) @@ -173,6 +184,51 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } +// The longest actor page token accepted. This controller's own tokens are far shorter. +const maxActorPageTokenBytes = 1024 + +// actorPageToken is the actor page token of a controller that watches some +// namespaces: the atespace to read and ate-api's token inside it. An atespace holds no "/". +func actorPageToken(atespace, inner string) string { + return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) +} + +// parseActorPageToken returns where in atespaces a page starts, and ate-api's token there. +func parseActorPageToken(atespaces []string, token string) (int, string, error) { + if token == "" { + return 0, "", nil + } + decoded, err := base64.RawURLEncoding.DecodeString(token) + atespace, inner, ok := strings.Cut(string(decoded), "/") + start := slices.Index(atespaces, atespace) + if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { + // The token is the caller's text and can name anything, so the error omits it. + return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) + } + return start, inner, nil +} + +// listScopedActors fills one page from the atespaces in order, each with its own scoped +// list, starting at ate-api's inner token in the first. +func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { + var actors []*ateapipb.Actor + for i, atespace := range atespaces { + rows, next, err := s.ateClient.ListActorsPage(ctx, atespace, limit-int32(len(actors)), inner) + if err != nil { + return nil, "", err + } + actors = append(actors, rows...) + switch { + case next != "": + return actors, actorPageToken(atespace, next), nil + case int32(len(actors)) >= limit && i+1 < len(atespaces): + return actors, actorPageToken(atespaces[i+1], ""), nil + } + inner = "" + } + return actors, "", nil +} + // ListSubstrateWorkers filters one upstream page by namespace and preserves its token. // A page with no workers in scope may still have a next page. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { From 44e1d06259fbfc7f83c318ec1a1246b4d5a32b84 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:44:59 -0400 Subject: [PATCH 07/12] test(substrate): worker page tokens and errors must name only watched workers A controller watching tenant-a and tenant-c must page its own workers once each, in worker name order whatever order ate-api sends, never hand out an empty page with a token, and never send its own token to ate-api. An ate-api worker list error must reach the caller as its gRPC code only, since the text can name another tenant's worker. The tenant isolation fixtures now name their workers, as ate-api does. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 8 +-- .../service/system/substrate_paging_test.go | 52 +++++++++++++++++++ 2 files changed, 56 insertions(+), 4 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index cbfd881812..c5cc4512fc 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -629,10 +629,10 @@ func TestSubstrateTenantIsolation(t *testing.T) { {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"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-a"}, WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-b"}, WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-c"}, WorkerNamespace: "tenant-c", WorkerPod: "pod-c"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-unlabelled"}, WorkerPod: "pod-unlabelled"}, }, } newService := func(observedNamespaces []string) *system.Service { diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index 324e48a024..64283a987a 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -10,6 +10,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" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" @@ -30,6 +32,10 @@ func pagingATEClient() *fakeATEClient { atespace, name, _ := strings.Cut(actor, "/") client.actors = append(client.actors, substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) } + for _, worker := range []string{"tenant-a/u4", "tenant-b/u2", "tenant-a/u1", "tenant-c/u3", "tenant-a/u6", "tenant-b/u5"} { + namespace, uid, _ := strings.Cut(worker, "/") + client.workers = append(client.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: uid + "-" + namespace}, WorkerNamespace: namespace}) + } return client } @@ -120,3 +126,49 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { }) } } + +func walkWorkerNames(t *testing.T, service *system.Service, namespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: namespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Workers), page.NextPageToken) + for _, worker := range page.Workers { + names = append(names, worker.GetMetadata().GetName()) + } + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +// ate-api cannot list one namespace's workers, so its token can name any worker. The +// fake refuses a token it did not issue, so a kagent token sent to it fails the walk. +func TestListSubstrateWorkersPagesWatchedNamespaces(t *testing.T) { + for namespace, want := range map[string][]string{ + "": {"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}, + "tenant-a": {"u1-tenant-a", "u4-tenant-a", "u6-tenant-a"}, + } { + for _, limit := range []int32{1, 2, 3, 100} { + t.Run(fmt.Sprintf("%q/%d", namespace, limit), func(t *testing.T) { + assert.Equal(t, want, walkWorkerNames(t, pagingService(pagingATEClient()), namespace, limit)) + }) + } + } +} + +func TestListSubstrateWorkersHidesATEAPIErrorText(t *testing.T) { + ateClient := pagingATEClient() + ateClient.err = status.Error(codes.Internal, "worker u2-tenant-b is down") + + page, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{}) + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", page.ATEAPIError) + assert.Empty(t, page.Workers) + assert.Empty(t, page.NextPageToken) +} From 8d31900af8b949005cba8cb4b8d2f7ca97290633 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:45:28 -0400 Subject: [PATCH 08/12] fix(substrate): page workers by name and hide ate-api worker errors A controller that watches some namespaces passed ate-api's worker page token through. The token holds the last worker's name, so it could name another tenant's worker, and a page filtered to nothing still carried a token. Such a controller now drains ate-api's workers, keeps the ones in scope, and pages them in worker name order. Its token is the last worker's name: a cursor only, never sent to ate-api. ate-api worker list errors reach the caller as the gRPC code only, for the list and the summary, since the text can name another tenant's worker; the full error is logged. A cluster-wide controller still passes ate-api's page, token and error through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/substrate.go | 50 ++++++++++++++++++-- 1 file changed, 46 insertions(+), 4 deletions(-) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 1d5eeb6dd5..b791329add 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -11,6 +11,7 @@ import ( atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/status" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" "github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors" @@ -229,15 +230,21 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne return actors, "", nil } -// ListSubstrateWorkers filters one upstream page by namespace and preserves its token. -// A page with no workers in scope may still have a next page. +// ListSubstrateWorkers returns one page of workers. A cluster-wide controller filters +// ate-api's page and passes its token through. A controller that watches some +// namespaces pages its own workers with its own token, since ate-api's can name any +// worker. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { return SubstrateWorkerPage{}, err } + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + if len(s.observedNamespaces) > 0 { + return s.listWatchedWorkers(ctx, scope, limit, token), nil + } result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} - workers, next, err := s.ateClient.ListWorkersPage(ctx, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + workers, next, err := s.ateClient.ListWorkersPage(ctx, limit, token) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) @@ -252,6 +259,31 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } +// listWatchedWorkers fills one page with the workers in scope, in worker name order. +// The token is the name of the last worker on the previous page: a cursor only, never +// sent to ate-api. +func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, limit int32, after string) SubstrateWorkerPage { + result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} + var workers []*ateapipb.Worker + if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { + if scope.contains(worker.GetWorkerNamespace()) && worker.GetMetadata().GetName() > after { + workers = append(workers, worker) + } + }); err != nil { + result.ATEAPIError = err.Error() + return result + } + slices.SortFunc(workers, func(left, right *ateapipb.Worker) int { + return strings.Compare(left.GetMetadata().GetName(), right.GetMetadata().GetName()) + }) + if int32(len(workers)) > limit { + workers = workers[:limit] + result.NextPageToken = workers[limit-1].GetMetadata().GetName() + } + result.Workers = workers + return result +} + func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { for _, atespace := range scope.listNamespaces() { read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { @@ -265,8 +297,18 @@ func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit fu } // walkWorkers calls visit for every worker ate-api holds, one page at a time. +// +// NOTE: ate-api cannot list one namespace's workers, so a controller that watches some +// drains every tenant's workers per request. Use an ate-api worker namespace filter +// when one lands. func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { - return walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + if err != nil && len(s.observedNamespaces) > 0 { + // ate-api's error text can name another tenant's worker, so the caller gets the code. + logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) + return fmt.Errorf("ate-api could not list workers: %s", status.Code(err)) + } + return err } // walkSubstrate calls visit for every row ate-api holds, holding one page at a time From c5cbbc9fa318a4c9c17d1ab90b3a24dec046d2ed Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Sun, 18 Oct 2026 23:46:20 -0400 Subject: [PATCH 09/12] docs(substrate): describe paging for controllers that watch some namespaces The proto comments, the generated Go and TypeScript, the UI page type and the Playwright deferral notes said an empty atespace lists all atespaces and that every page passes Substrate's token through. A controller that watches some namespaces lists only its watched atespaces, issues its own tokens, and never returns an empty page with a token. The atespace filter placeholder now reads "All watched atespaces". Signed-off-by: Danielle Mustillo --- go/api/gen/kagent/api/v1/system.pb.go | 19 +++++++++++++------ proto/kagent/api/v1/system.proto | 19 +++++++++++++------ ui/playwright/DEFERRED.md | 12 +++++++----- ui/src/api/domain/substrate.ts | 6 +++++- ui/src/generated/kagent/api/v1/system_pb.ts | 19 +++++++++++++------ ui/src/pages/SubstratePage.tsx | 2 +- 6 files changed, 52 insertions(+), 25 deletions(-) diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 5ae0980ae8..13e953fa5d 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -396,7 +396,8 @@ type GetSubstrateSummaryRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // Kubernetes namespace for workers and pools. Empty uses watched namespaces. Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty lists watched atespaces (all when + // the controller watches all namespaces). Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -614,10 +615,12 @@ func (x *GetSubstrateSummaryResponse) GetActorTemplates() []*ateapipb.ActorTempl return nil } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. type ListSubstrateActorsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty lists watched atespaces (all when the + // controller watches all namespaces). Atespace string `protobuf:"bytes,6,opt,name=atespace,proto3" json:"atespace,omitempty"` // `limit` is capped at 100 in common.proto and zero means the server's default; // above the cap is refused rather than clamped. @@ -674,7 +677,9 @@ type ListSubstrateActorsResponse struct { state protoimpl.MessageState `protogen:"open.v1"` AteApiError string `protobuf:"bytes,2,opt,name=ate_api_error,json=ateApiError,proto3" json:"ate_api_error,omitempty"` Actors []*ateapipb.Actor `protobuf:"bytes,3,rep,name=actors,proto3" json:"actors,omitempty"` - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. Only a last page can be empty. Page *PageResponse `protobuf:"bytes,4,opt,name=page,proto3" json:"page,omitempty"` ComputedAt *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=computed_at,json=computedAt,proto3" json:"computed_at,omitempty"` unknownFields protoimpl.UnknownFields @@ -739,8 +744,10 @@ func (x *ListSubstrateActorsResponse) GetComputedAt() *timestamppb.Timestamp { return nil } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers, in worker name order. A controller that watches all namespaces +// filters one Substrate page, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces; only its last page can be empty. type ListSubstrateWorkersRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index b8e7cbecdf..0eb8c62bbb 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -62,7 +62,8 @@ message GetSubstrateSummaryRequest { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" }]; - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty lists watched atespaces (all when + // the controller watches all namespaces). string atespace = 2 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" @@ -94,9 +95,11 @@ message GetSubstrateSummaryResponse { repeated ateapi.ActorTemplate actor_templates = 11; } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. message ListSubstrateActorsRequest { - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty lists watched atespaces (all when the + // controller watches all namespaces). string atespace = 6 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" @@ -109,13 +112,17 @@ message ListSubstrateActorsRequest { message ListSubstrateActorsResponse { string ate_api_error = 2; repeated ateapi.Actor actors = 3; - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. Only a last page can be empty. PageResponse page = 4; google.protobuf.Timestamp computed_at = 5; } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers, in worker name order. A controller that watches all namespaces +// filters one Substrate page, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces; only its last page can be empty. message ListSubstrateWorkersRequest { string namespace = 1 [(buf.validate.field).string = { max_len: 63 diff --git a/ui/playwright/DEFERRED.md b/ui/playwright/DEFERRED.md index be664c4f12..530fd90a6f 100644 --- a/ui/playwright/DEFERRED.md +++ b/ui/playwright/DEFERRED.md @@ -219,11 +219,13 @@ when it holds one page of them, and only the proto says which. | `ListSubstrateActors` | `ListSubstrateActorsRequest { atespace, page }` | a page | `string filter` and a sort field enum — **and ate-api has to grow them first** | | `ListSubstrateWorkers` | `ListSubstrateWorkersRequest { namespace, page }` | a page | the same two, on the same condition | -**Substrate actor and worker lists use upstream pagination.** Each list request makes -one ate-api call and preserves its order and continuation token. Actors use upstream -atespace filtering. Worker namespace filtering applies only to the returned page, so an -empty worker page may still have a next token. The UI labels counts as "on this page" -and offers no actor/worker text search or column sorting. +**Substrate actor and worker lists use upstream pagination.** A controller that watches +every namespace makes one ate-api call per list request and passes its order and +continuation token through. Worker namespace filtering then applies only to the returned +page, so an empty worker page may still have a next token. A controller that watches some +namespaces reads actors with one scoped ate-api list per watched atespace and returns its +own token. It reads every ate-api worker page and returns a worker-name cursor. The UI +labels counts as "on this page" and offers no actor/worker text search or column sorting. Two capabilities remain deferred until Substrate supports them: diff --git a/ui/src/api/domain/substrate.ts b/ui/src/api/domain/substrate.ts index 09e3d1d3d1..2afe547b0b 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -94,7 +94,11 @@ export interface Timed { computedAt?: string; } -/** One upstream page; an empty page may still have a continuation token. */ +/** + * One page. Only a last page is empty, except a workers page from a controller that + * watches all namespaces: it filters one upstream page, so it can be empty and still + * have a continuation token. + */ interface SubstratePage extends Timed { /** Upstream read failure. No rows or continuation token accompany it. */ ateApiError?: string; diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 504a6cdba8..492410d558 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -184,7 +184,8 @@ export type GetSubstrateSummaryRequest = Message<"kagent.api.v1.GetSubstrateSumm namespace: string; /** - * ATE atespace for actors and templates. Empty lists all atespaces. + * ATE atespace for actors and templates. Empty lists watched atespaces (all when + * the controller watches all namespaces). * * @generated from field: string atespace = 2; */ @@ -291,13 +292,15 @@ export const GetSubstrateSummaryResponseSchema: GenMessage & { /** - * ATE atespace of the actor itself. Empty lists all atespaces. + * ATE atespace of the actor itself. Empty lists watched atespaces (all when the + * controller watches all namespaces). * * @generated from field: string atespace = 6; */ @@ -334,7 +337,9 @@ export type ListSubstrateActorsResponse = Message<"kagent.api.v1.ListSubstrateAc actors: Actor[]; /** - * Upstream continuation token. A page can be empty even when another page exists. + * Continuation token. A controller that watches all namespaces passes Substrate's + * token through. A controller that watches some issues its own, which names only + * watched data. Only a last page can be empty. * * @generated from field: kagent.api.v1.PageResponse page = 4; */ @@ -354,8 +359,10 @@ export const ListSubstrateActorsResponseSchema: GenMessage Date: Mon, 19 Oct 2026 00:18:32 -0400 Subject: [PATCH 10/12] docs(substrate): trim paging comments and correct empty-page wording Drop comments that restate the code. The proto now says that an actor page can be empty when another page exists, and gives the worker order for each mode. Signed-off-by: Danielle Mustillo --- go/api/gen/kagent/api/v1/system.pb.go | 8 +++--- go/core/internal/service/system/service.go | 3 +-- go/core/internal/service/system/substrate.go | 27 +++---------------- .../service/system/substrate_paging_test.go | 9 ------- proto/kagent/api/v1/system.proto | 8 +++--- ui/src/api/domain/substrate.ts | 5 ---- ui/src/generated/kagent/api/v1/system_pb.ts | 8 +++--- 7 files changed, 17 insertions(+), 51 deletions(-) diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 13e953fa5d..0549b3c927 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -679,7 +679,7 @@ type ListSubstrateActorsResponse struct { Actors []*ateapipb.Actor `protobuf:"bytes,3,rep,name=actors,proto3" json:"actors,omitempty"` // Continuation token. A controller that watches all namespaces passes Substrate's // token through. A controller that watches some issues its own, which names only - // watched data. Only a last page can be empty. + // watched data. A page can be empty even when another page exists. Page *PageResponse `protobuf:"bytes,4,opt,name=page,proto3" json:"page,omitempty"` ComputedAt *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=computed_at,json=computedAt,proto3" json:"computed_at,omitempty"` unknownFields protoimpl.UnknownFields @@ -744,10 +744,10 @@ func (x *ListSubstrateActorsResponse) GetComputedAt() *timestamppb.Timestamp { return nil } -// One page of workers, in worker name order. A controller that watches all namespaces -// filters one Substrate page, which may be empty and still have a next token. A +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A // controller that watches some reads every Substrate worker and returns only those of -// its watched namespaces; only its last page can be empty. +// its watched namespaces, in worker name order; only its last page can be empty. type ListSubstrateWorkersRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index 7389221ebd..fd316d4a51 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -32,8 +32,7 @@ type ATEClient interface { } type Service struct { - kubeClient client.Client - // Sorted and unique: actor page tokens resume in this order. Empty means every namespace. + kubeClient client.Client observedNamespaces []string authorizer auth.Authorizer ateClient ATEClient diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index b791329add..ddaffc49f7 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -149,10 +149,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors returns one page of actors. A cluster-wide controller passes -// ate-api's page and token through. A controller that watches some namespaces reads -// each watched atespace with its own scoped list, so no page token names an actor of -// a namespace it does not watch. +// ListSubstrateActors returns one page of actors. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) if err != nil { @@ -185,16 +182,12 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } -// The longest actor page token accepted. This controller's own tokens are far shorter. const maxActorPageTokenBytes = 1024 -// actorPageToken is the actor page token of a controller that watches some -// namespaces: the atespace to read and ate-api's token inside it. An atespace holds no "/". func actorPageToken(atespace, inner string) string { return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) } -// parseActorPageToken returns where in atespaces a page starts, and ate-api's token there. func parseActorPageToken(atespaces []string, token string) (int, string, error) { if token == "" { return 0, "", nil @@ -203,14 +196,11 @@ func parseActorPageToken(atespaces []string, token string) (int, string, error) atespace, inner, ok := strings.Cut(string(decoded), "/") start := slices.Index(atespaces, atespace) if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { - // The token is the caller's text and can name anything, so the error omits it. return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) } return start, inner, nil } -// listScopedActors fills one page from the atespaces in order, each with its own scoped -// list, starting at ate-api's inner token in the first. func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { var actors []*ateapipb.Actor for i, atespace := range atespaces { @@ -230,10 +220,7 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne return actors, "", nil } -// ListSubstrateWorkers returns one page of workers. A cluster-wide controller filters -// ate-api's page and passes its token through. A controller that watches some -// namespaces pages its own workers with its own token, since ate-api's can name any -// worker. +// ListSubstrateWorkers returns one page of workers. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { @@ -259,9 +246,6 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// listWatchedWorkers fills one page with the workers in scope, in worker name order. -// The token is the name of the last worker on the previous page: a cursor only, never -// sent to ate-api. func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, limit int32, after string) SubstrateWorkerPage { result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} var workers []*ateapipb.Worker @@ -297,14 +281,11 @@ func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit fu } // walkWorkers calls visit for every worker ate-api holds, one page at a time. -// -// NOTE: ate-api cannot list one namespace's workers, so a controller that watches some -// drains every tenant's workers per request. Use an ate-api worker namespace filter -// when one lands. +// ate-api's ListWorkersRequest has no namespace filter, so a controller that watches +// some namespaces still reads every tenant's workers. func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) if err != nil && len(s.observedNamespaces) > 0 { - // ate-api's error text can name another tenant's worker, so the caller gets the code. logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) return fmt.Errorf("ate-api could not list workers: %s", status.Code(err)) } diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index 64283a987a..20c279b386 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -24,8 +24,6 @@ func pagingContext(t *testing.T) context.Context { return pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) } -// pagingATEClient mixes three tenants' rows in ate-api order, two rows a page. Every -// tenant-b name holds "tenant-b", so a test can look for it in any text. func pagingATEClient() *fakeATEClient { client := &fakeATEClient{pageSize: 2} for _, actor := range []string{"tenant-a/a-1", "tenant-b/b-1", "tenant-c/c-1", "tenant-a/a-2", "tenant-b/b-2", "tenant-c/c-2", "tenant-a/a-3"} { @@ -39,13 +37,10 @@ func pagingATEClient() *fakeATEClient { return client } -// pagingService watches tenant-a and tenant-c, listed out of order and with a repeat. func pagingService(ateClient system.ATEClient) *system.Service { return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) } -// assertOwnToken fails when a page token names a tenant-b row, or when an empty page -// carries a token: that would tell the caller a hidden row exists. func assertOwnToken(t *testing.T, rows int, pageToken string) { t.Helper() raw, _ := base64.RawURLEncoding.DecodeString(pageToken) @@ -100,8 +95,6 @@ func actorPageToken(raw string) string { return base64.RawURLEncoding.EncodeToString([]byte(raw)) } -// A token this controller did not issue never reaches ate-api and never restarts the -// walk at page one. Its error repeats neither the token nor a tenant-b name. func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { for name, tc := range map[string]struct { atespace string @@ -147,8 +140,6 @@ func walkWorkerNames(t *testing.T, service *system.Service, namespace string, li return nil } -// ate-api cannot list one namespace's workers, so its token can name any worker. The -// fake refuses a token it did not issue, so a kagent token sent to it fails the walk. func TestListSubstrateWorkersPagesWatchedNamespaces(t *testing.T) { for namespace, want := range map[string][]string{ "": {"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}, diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index 0eb8c62bbb..64d4aadd1b 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -114,15 +114,15 @@ message ListSubstrateActorsResponse { repeated ateapi.Actor actors = 3; // Continuation token. A controller that watches all namespaces passes Substrate's // token through. A controller that watches some issues its own, which names only - // watched data. Only a last page can be empty. + // watched data. A page can be empty even when another page exists. PageResponse page = 4; google.protobuf.Timestamp computed_at = 5; } -// One page of workers, in worker name order. A controller that watches all namespaces -// filters one Substrate page, which may be empty and still have a next token. A +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A // controller that watches some reads every Substrate worker and returns only those of -// its watched namespaces; only its last page can be empty. +// its watched namespaces, in worker name order; only its last page can be empty. message ListSubstrateWorkersRequest { string namespace = 1 [(buf.validate.field).string = { max_len: 63 diff --git a/ui/src/api/domain/substrate.ts b/ui/src/api/domain/substrate.ts index 2afe547b0b..f13391950e 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -94,11 +94,6 @@ export interface Timed { computedAt?: string; } -/** - * One page. Only a last page is empty, except a workers page from a controller that - * watches all namespaces: it filters one upstream page, so it can be empty and still - * have a continuation token. - */ interface SubstratePage extends Timed { /** Upstream read failure. No rows or continuation token accompany it. */ ateApiError?: string; diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 492410d558..088a107c96 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -339,7 +339,7 @@ export type ListSubstrateActorsResponse = Message<"kagent.api.v1.ListSubstrateAc /** * Continuation token. A controller that watches all namespaces passes Substrate's * token through. A controller that watches some issues its own, which names only - * watched data. Only a last page can be empty. + * watched data. A page can be empty even when another page exists. * * @generated from field: kagent.api.v1.PageResponse page = 4; */ @@ -359,10 +359,10 @@ export const ListSubstrateActorsResponseSchema: GenMessage Date: Mon, 19 Oct 2026 00:42:46 -0400 Subject: [PATCH 11/12] test(substrate): expose row keys in fake page tokens and probe forged tokens The fake ate-api now hands out base64 JSON tokens naming the page's last row, as ate-api does, so assertOwnToken can see a token that would name another tenant's row. New tests: scoped worker page tokens must be refused unless kagent issued them, a scoped summary hides ate-api worker error text, forged tokens never reach or return an unwatched namespace, and the longest names fit the page token. Signed-off-by: Danielle Mustillo --- .../internal/service/system/service_test.go | 56 +++++--- .../service/system/substrate_paging_test.go | 133 ++++++++++++++++-- 2 files changed, 163 insertions(+), 26 deletions(-) diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index c5cc4512fc..52768ac8bf 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -2,11 +2,10 @@ package system_test import ( "context" + "encoding/base64" + "encoding/json" "errors" - "fmt" "slices" - "strconv" - "strings" "testing" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" @@ -70,7 +69,7 @@ func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, if err := client.readError(client.actorReads); err != nil { return nil, "", err } - return fakePage(actorsInAtespace(client.actors, atespace), client.pageSize, pageSize, pageToken) + return fakePage(actorsInAtespace(client.actors, atespace), actorKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) { @@ -78,42 +77,61 @@ func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, if err := client.readError(client.workerReads); err != nil { return nil, "", err } - return fakePage(client.workers, client.pageSize, pageSize, pageToken) + return fakePage(client.workers, workerKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) readError(int) error { return client.err } -// fakePage slices rows the way ate-api pages them: an opaque token, empty on the last -// page, and never more rows than its own ceiling however many were asked for. -func fakePage[T any](rows []T, ceiling int, requested int32, pageToken string) ([]T, string, error) { +type fakeCursor struct { + Last string `json:"last"` + Offset int `json:"offset"` +} + +// fakePage slices rows the way ate-api pages them: never more rows than its own +// ceiling however many were asked for, and a token that is empty on the last page and +// otherwise base64 JSON naming the page's last row, as ate-api's does. Carrying the +// row key lets a test see a token that would tell one tenant about another's rows. +func fakePage[T any](rows []T, key func(T) string, ceiling int, requested int32, pageToken string) ([]T, string, error) { pageSize := ceiling if requested > 0 && (ceiling <= 0 || int(requested) < ceiling) { pageSize = int(requested) } start := 0 if pageToken != "" { - parsed, err := strconv.Atoi(strings.TrimPrefix(pageToken, "upstream:")) - if err != nil { - return nil, "", fmt.Errorf("invalid page token %q", pageToken) + var cursor fakeCursor + raw, err := base64.StdEncoding.DecodeString(pageToken) + if err != nil || json.Unmarshal(raw, &cursor) != nil { + return nil, "", errors.New("invalid page token") } - start = parsed - } - if start > len(rows) { - start = len(rows) + start = max(cursor.Offset, 0) } + start = min(start, len(rows)) if pageSize <= 0 { return rows[start:], "", nil } end := min(start+pageSize, len(rows)) next := "" if end < len(rows) { - next = "upstream:" + strconv.Itoa(end) + next = fakePageToken(key(rows[end-1]), end) } return rows[start:end], next, nil } +func fakePageToken(last string, offset int) string { + raw, _ := json.Marshal(fakeCursor{Last: last, Offset: offset}) + return base64.StdEncoding.EncodeToString(raw) +} + +func actorKey(actor *ateapipb.Actor) string { + return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName() +} + +func workerKey(worker *ateapipb.Worker) string { + return worker.GetMetadata().GetName() +} + func TestCurrentUser(t *testing.T) { service := system.NewService(nil, nil, nil, nil) claims := map[string]any{"sub": "user-1", "groups": []any{"admins"}} @@ -278,7 +296,7 @@ func TestListSubstrateWorkers(t *testing.T) { require.NoError(t, err) require.Len(t, page.Workers, 1) assert.Equal(t, "worker-0", page.Workers[0].WorkerPod) - assert.Equal(t, "upstream:2", page.NextPageToken) + assert.Equal(t, fakePageToken("", 2), page.NextPageToken) assert.Equal(t, 1, ateClient.workerReads) next, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: &apiv1.PageRequest{Limit: 2, PageToken: page.NextPageToken}}) require.NoError(t, err) @@ -457,9 +475,13 @@ func TestGetSubstrateSummaryReadsAreIndependent(t *testing.T) { // counted while the pods they sit on cannot. type failingWorkersATEClient struct { fakeATEClient + workerErr error } func (client *failingWorkersATEClient) ListWorkersPage(context.Context, int32, string) ([]*ateapipb.Worker, string, error) { + if client.workerErr != nil { + return nil, "", client.workerErr + } return nil, "", errors.New("workers unavailable") } diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go index 20c279b386..bcda6ef98e 100644 --- a/go/core/internal/service/system/substrate_paging_test.go +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -4,14 +4,18 @@ import ( "context" "encoding/base64" "fmt" + "math/rand/v2" "strings" "testing" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "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" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" @@ -41,10 +45,24 @@ func pagingService(ateClient system.ATEClient) *system.Service { return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) } +// tokenText is a page token with every base64 layer under it decoded, so a row key +// nested in an ate-api cursor inside a kagent cursor is still visible. +func tokenText(token string) string { + text := token + for _, encoding := range []*base64.Encoding{base64.RawURLEncoding, base64.StdEncoding} { + if raw, err := encoding.DecodeString(token); err == nil && len(raw) > 0 { + text += "\n" + tokenText(string(raw)) + } + } + if _, inner, ok := strings.Cut(token, "/"); ok { + text += "\n" + tokenText(inner) + } + return text +} + func assertOwnToken(t *testing.T, rows int, pageToken string) { t.Helper() - raw, _ := base64.RawURLEncoding.DecodeString(pageToken) - assert.NotContains(t, pageToken+string(raw), "tenant-b") + assert.NotContains(t, tokenText(pageToken), "tenant-b") if rows == 0 { assert.Empty(t, pageToken, "an empty page must not lead on") } @@ -91,7 +109,7 @@ func actorNames(actors []*ateapipb.Actor) []string { return names } -func actorPageToken(raw string) string { +func scopedPageToken(raw string) string { return base64.RawURLEncoding.EncodeToString([]byte(raw)) } @@ -101,12 +119,12 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { pageToken string }{ "not base64": {pageToken: "upstream:2"}, - "no separator": {pageToken: actorPageToken("tenant-a")}, - "empty atespace": {pageToken: actorPageToken("/upstream:2")}, - "unwatched atespace": {pageToken: actorPageToken("tenant-b/upstream:2")}, - "outside the request": {atespace: "tenant-c", pageToken: actorPageToken("tenant-a/upstream:2")}, - "ate-api actor token": {pageToken: actorPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, - "over 1 KiB, else good": {pageToken: actorPageToken("tenant-a/" + strings.Repeat("x", 800))}, + "no separator": {pageToken: scopedPageToken("tenant-a")}, + "empty atespace": {pageToken: scopedPageToken("/upstream:2")}, + "unwatched atespace": {pageToken: scopedPageToken("tenant-b/upstream:2")}, + "outside the request": {atespace: "tenant-c", pageToken: scopedPageToken("tenant-a/upstream:2")}, + "ate-api actor token": {pageToken: scopedPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, } { t.Run(name, func(t *testing.T) { ateClient := pagingATEClient() @@ -120,6 +138,31 @@ func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { } } +func TestListSubstrateWorkersRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + namespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "ate-api worker token": {pageToken: fakePageToken("u2-tenant-b", 2)}, + "bare worker name": {pageToken: scopedPageToken("u1-tenant-a")}, + "no worker name": {pageToken: scopedPageToken("tenant-a/")}, + "unwatched namespace": {pageToken: scopedPageToken("tenant-b/u2-tenant-b")}, + "outside the request": {namespace: "tenant-c", pageToken: scopedPageToken("tenant-a/u1-tenant-a")}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: tc.namespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.workerReads) + }) + } +} + func walkWorkerNames(t *testing.T, service *system.Service, namespace string, limit int32) []string { t.Helper() names := []string{} @@ -163,3 +206,75 @@ func TestListSubstrateWorkersHidesATEAPIErrorText(t *testing.T) { assert.Empty(t, page.Workers) assert.Empty(t, page.NextPageToken) } + +func TestGetSubstrateSummaryHidesATEAPIWorkerErrorText(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + ateClient := &failingWorkersATEClient{fakeATEClient: *pagingATEClient()} + ateClient.workerErr = status.Error(codes.Internal, "worker u2-tenant-b is down") + service := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + summary, err := service.GetSubstrateSummary(pagingContext(t), "", "") + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", summary.ATEAPIError) + assert.Equal(t, int64(3), summary.ActorCount, "the actor walk is independent of the worker walk") +} + +func TestSubstrateForgedTokensStayInWatchedNamespaces(t *testing.T) { + forged := []string{ + scopedPageToken("tenant-b/"), + scopedPageToken("tenant-a/" + fakePageToken("tenant-b/b-1", 1)), + scopedPageToken("tenant-a/" + fakePageToken("", -5)), + scopedPageToken("tenant-a/" + fakePageToken("", 1<<40)), + scopedPageToken("/" + fakePageToken("", 0)), + fakePageToken("tenant-a/a-1", 1), + "!!!", + } + random := rand.New(rand.NewPCG(1, 1)) + for range 300 { + raw := make([]byte, random.IntN(80)) + for i := range raw { + raw[i] = "tenant-abc/xyz=+"[random.IntN(16)] + } + forged = append(forged, scopedPageToken(string(raw)), string(raw)) + } + + ateClient := pagingATEClient() + service := pagingService(ateClient) + watched := []string{"tenant-a", "tenant-c"} + for _, scope := range []string{"", "tenant-a", "tenant-c"} { + for _, pageToken := range forged { + actors, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, actor := range actors.Actors { + require.Contains(t, watched, actor.GetMetadata().GetAtespace()) + } + require.NotContains(t, tokenText(actors.NextPageToken), "tenant-b") + + workers, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, worker := range workers.Workers { + require.Contains(t, watched, worker.GetWorkerNamespace()) + } + require.NotContains(t, tokenText(workers.NextPageToken), "tenant-b") + } + } + assert.Subset(t, watched, ateClient.actorAtespaces, "every ate-api actor read is scoped to a watched atespace") +} + +func TestSubstrateLongestNamesFitThePageToken(t *testing.T) { + namespace := strings.Repeat("n", 63) + ateClient := &fakeATEClient{pageSize: 1} + for _, suffix := range []string{"0", "1"} { + ateClient.actors = append(ateClient.actors, substrateActor(strings.Repeat("a", 252)+suffix, namespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + ateClient.workers = append(ateClient.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: strings.Repeat("w", 252) + suffix}, WorkerNamespace: namespace}) + } + service := system.NewService(nil, []string{namespace}, &pkgAuth.NoopAuthorizer{}, ateClient) + + assert.Len(t, walkActorNames(t, service, "", 1), 2) + assert.Len(t, walkWorkerNames(t, service, "", 1), 2) +} From 6051b77ea54315113f70878916ce1f3c5fc5df46 Mon Sep 17 00:00:00 2001 From: Danielle Mustillo Date: Mon, 19 Oct 2026 00:43:31 -0400 Subject: [PATCH 12/12] fix(substrate): validate worker page tokens A controller that watches some namespaces used any worker page_token raw as the name to resume after, so an ate-api token or junk silently returned a wrong or short page. Worker tokens now share the actor tokens' shape (base64url of a watched namespace, a slash, then the cursor) and the same 1 KiB cap, and anything else is InvalidArgument. The length check now runs before decoding. Cluster-wide installs still pass ate-api's tokens through. Signed-off-by: Danielle Mustillo --- go/core/internal/service/system/substrate.go | 45 ++++++++++++++------ 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index ddaffc49f7..67a9d94798 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -161,7 +161,7 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs if len(s.observedNamespaces) == 0 { actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) } else { - start, inner, tokenErr := parseActorPageToken(scope.namespaces, token) + start, inner, tokenErr := parseScopedPageToken(scope.namespaces, token) if tokenErr != nil { return SubstrateActorPage{}, tokenErr } @@ -182,23 +182,32 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs return result, nil } -const maxActorPageTokenBytes = 1024 +const maxPageTokenBytes = 1024 -func actorPageToken(atespace, inner string) string { - return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) +// scopedPageToken is the cursor a controller that watches some namespaces hands out: +// the watched namespace a page ended in, then where to resume. +func scopedPageToken(namespace, cursor string) string { + return base64.RawURLEncoding.EncodeToString([]byte(namespace + "/" + cursor)) } -func parseActorPageToken(atespaces []string, token string) (int, string, error) { +func parseScopedPageToken(namespaces []string, token string) (int, string, error) { if token == "" { return 0, "", nil } + if len(token) > maxPageTokenBytes { + return 0, "", invalidPageToken() + } decoded, err := base64.RawURLEncoding.DecodeString(token) - atespace, inner, ok := strings.Cut(string(decoded), "/") - start := slices.Index(atespaces, atespace) - if len(token) > maxActorPageTokenBytes || err != nil || !ok || start < 0 { - return 0, "", serviceerrors.NewInvalidArgument("Invalid page token", nil) + namespace, cursor, ok := strings.Cut(string(decoded), "/") + start := slices.Index(namespaces, namespace) + if err != nil || !ok || start < 0 { + return 0, "", invalidPageToken() } - return start, inner, nil + return start, cursor, nil +} + +func invalidPageToken() error { + return serviceerrors.NewInvalidArgument("Invalid page token", nil) } func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { @@ -211,9 +220,9 @@ func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inne actors = append(actors, rows...) switch { case next != "": - return actors, actorPageToken(atespace, next), nil + return actors, scopedPageToken(atespace, next), nil case int32(len(actors)) >= limit && i+1 < len(atespaces): - return actors, actorPageToken(atespaces[i+1], ""), nil + return actors, scopedPageToken(atespaces[i+1], ""), nil } inner = "" } @@ -228,7 +237,14 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub } limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() if len(s.observedNamespaces) > 0 { - return s.listWatchedWorkers(ctx, scope, limit, token), nil + _, after, err := parseScopedPageToken(scope.namespaces, token) + if err == nil && token != "" && after == "" { + err = invalidPageToken() + } + if err != nil { + return SubstrateWorkerPage{}, err + } + return s.listWatchedWorkers(ctx, scope, limit, after), nil } result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} workers, next, err := s.ateClient.ListWorkersPage(ctx, limit, token) @@ -262,7 +278,8 @@ func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, }) if int32(len(workers)) > limit { workers = workers[:limit] - result.NextPageToken = workers[limit-1].GetMetadata().GetName() + last := workers[limit-1] + result.NextPageToken = scopedPageToken(last.GetWorkerNamespace(), last.GetMetadata().GetName()) } result.Workers = workers return result