[HYPERSHELL-297] Filter gateways by cluster_id for managed-cluster pull model

components/api-server/plugins/gateways/grpc_handler.go

@@ -3,6 +3,7 @@ package gateways
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
@@ -234,6 +235,22 @@ func (h *gatewayGRPCHandler) ListGateways(ctx context.Context, req *pb.ListGatew
 		Size: int64(size),
 	}
 
+	// A managed-cluster control-plane sets cluster_id to its own identity so it
+	// only ever lists the gateways it is responsible for provisioning. Filtering
+	// server-side keeps foreign gateways off the wire entirely (see WatchGateways,
+	// which applies the same cooperative scoping to the event stream).
+	//
+	// Validate before interpolating into the search DSL: cluster_id is
+	// request-supplied, so an unvalidated value (e.g. one containing a quote)
+	// could break the filter parse or broaden it. This applies the same field
+	// contract the create/update paths enforce on cluster_id.
+	if clusterID := req.GetClusterId(); clusterID != "" {
+		if err := grpcutil.ValidateStringField("cluster_id", clusterID, false); err != nil {
+			return nil, err
+		}
+		listArgs.Search = fmt.Sprintf("cluster_id = '%s'", clusterID)
+	}
+
 	var gateways []Gateway
 	paging, svcErr := h.generic.List(ctx, "id", listArgs, &gateways)
 	if svcErr != nil {
@@ -257,6 +274,15 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 		return status.Error(codes.Unavailable, "event broker not available")
 	}
 
+	// clusterFilter, when set, scopes this stream to a single managed cluster.
+	// The broker fans EVERY gateway out to EVERY subscriber, so without this a
+	// spoke would receive (and could act on) other clusters' gateways. This is
+	// cooperative scoping, not an enforced trust boundary: the server does not yet
+	// authenticate that the caller owns the claimed cluster_id (no per-caller
+	// RBAC), so any control-plane could pass any cluster_id. Enforcement is pending
+	// the managed-cluster caller-identity binding (remote gRPC TLS+OIDC dial).
+	clusterFilter := req.GetClusterId()
+
 	ctx := stream.Context()
 	sub, err := broker.Subscribe(ctx)
 	if err != nil {
@@ -296,7 +322,17 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 				gateway, svcErr := h.service.GetUnscoped(ctx, evt.SourceID)
 				if svcErr != nil {
 					glog.Warningf("WatchGateways: failed to load soft-deleted gateway %s: %v", evt.SourceID, svcErr)
+					// When a cluster filter is set we cannot attribute an
+					// unloadable delete to a cluster, so we must not leak it to a
+					// scoped subscriber. Skip it; the spoke's namespace GC still
+					// reaps the orphaned namespace.
+					if clusterFilter != "" {
+						continue
+					}
 				} else {
+					if clusterFilter != "" && gateway.ClusterId != clusterFilter {
+						continue
+					}
 					watchEvent.Gateway = gatewayToProto(gateway)
 				}
 			} else {
@@ -305,6 +341,9 @@ func (h *gatewayGRPCHandler) WatchGateways(req *pb.WatchGatewaysRequest, stream
 					glog.Warningf("WatchGateways: failed to load gateway %s: %v", evt.SourceID, svcErr)
 					continue
 				}
+				if clusterFilter != "" && gateway.ClusterId != clusterFilter {
+					continue
+				}
 				watchEvent.Gateway = gatewayToProto(gateway)
 			}
 

components/api-server/proto/hypershell/v1/gateways.proto

@@ -127,6 +127,11 @@ message DeleteGatewayRequest {
 message ListGatewaysRequest {
   int32 page = 1;
   int32 size = 2;
+  // cluster_id, when set, restricts results to gateways assigned to that
+  // managed cluster. A control-plane agent sets it to its own cluster identity
+  // so it only ever lists its cluster's gateways (managed-cluster pull model).
+  // Unset preserves the prior behaviour of listing every gateway.
+  optional string cluster_id = 3;
 }
 
 message ListGatewaysResponse {
@@ -136,7 +141,13 @@ message ListGatewaysResponse {
 
 message DeleteGatewayResponse {}
 
-message WatchGatewaysRequest {}
+message WatchGatewaysRequest {
+  // cluster_id, when set, restricts the stream to gateways assigned to that
+  // managed cluster (see ListGatewaysRequest.cluster_id). Because the event
+  // broker fans every gateway out to every subscriber, this filter is the
+  // security boundary that keeps a spoke's stream scoped to its own cluster.
+  optional string cluster_id = 1;
+}
 
 message WatchGatewaysResponse {
   EventType type = 1;

components/control-plane/cmd/hypershell-controller/main.go

@@ -52,6 +52,11 @@ func main() {
 
 	log.Printf("INFO hypershell-controller starting")
 	log.Printf("INFO grpc=%s api=%s namespace=%s database_provider=%s", cfg.GRPCServerAddr, cfg.APIServerURL, cfg.Namespace, cfg.DatabaseProvider)
+	if cfg.ClusterID != "" {
+		log.Printf("INFO managed-cluster mode: scoping gateway watch/seed/health to cluster_id=%s", cfg.ClusterID)
+	} else {
+		log.Printf("INFO single-cluster mode: handling all gateways (no cluster_id filter)")
+	}
 
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
@@ -281,7 +286,7 @@ func main() {
 		return watcher.WatchGatewayReleases(ctx, conn, releaseReconciler)
 	})
 	supervise("Gateway watch", func(ctx context.Context) error {
-		return watcher.WatchGateways(ctx, conn, gatewayReconciler)
+		return watcher.WatchGateways(ctx, conn, gatewayReconciler, cfg.ClusterID)
 	})
 	supervise("GatewayNetwork watch", func(ctx context.Context) error {
 		return watcher.WatchGatewayNetworks(ctx, conn, networkReconciler)
@@ -298,7 +303,7 @@ func main() {
 	// status synchronized with observed workload health (Running <-> Degraded).
 	// It requires an in-cluster Kubernetes client to observe Deployments.
 	if clientset != nil {
-		healthReconciler := reconciler.NewGatewayHealthReconciler(clientset, dynamicClient, conn, exposurePort, keycloakConfig)
+		healthReconciler := reconciler.NewGatewayHealthReconciler(clientset, dynamicClient, conn, exposurePort, keycloakConfig, cfg.ClusterID)
 		supervise("gateway health reconciler", healthReconciler.Run)
 		log.Printf("INFO gateway health reconciler launched")
 	} else {
@@ -310,7 +315,7 @@ func main() {
 	// its cache), instead of a repeated full-namespace pod LIST. It requires an
 	// in-cluster Kubernetes client to watch pods.
 	if clientset != nil {
-		sandboxCountReconciler := reconciler.NewSandboxCountReconciler(clientset, conn, 0)
+		sandboxCountReconciler := reconciler.NewSandboxCountReconciler(clientset, conn, 0, cfg.ClusterID)
 		supervise("sandbox count reconciler", sandboxCountReconciler.Run)
 		log.Printf("INFO sandbox count reconciler launched")
 	} else {
@@ -330,7 +335,7 @@ func main() {
 		// synchronously so the first sweep sees the freshly-labeled namespaces; it
 		// is best-effort and never blocks startup on failure.
 		backfillCtx, cancelBackfill := context.WithTimeout(ctx, instanceLabelBackfillTimeout)
-		reconciler.RunInstanceLabelBackfill(backfillCtx, clientset, conn, cfg.Namespace)
+		reconciler.RunInstanceLabelBackfill(backfillCtx, clientset, conn, cfg.Namespace, cfg.ClusterID)
 		cancelBackfill()
 
 		gcReconciler := reconciler.NewNamespaceGCReconciler(

components/control-plane/internal/config/config.go

@@ -27,6 +27,14 @@ type Config struct {
 	Namespace      string
 	LogLevel       string
 
+	// ClusterID is this control-plane's managed-cluster identity (a Gateway
+	// cluster_id / KSUID). When set, the control-plane restricts the gateways it
+	// watches, seeds, and health-checks to those whose cluster_id matches, so a
+	// managed-cluster spoke only ever provisions its own gateways (the pull
+	// model). Empty preserves the single-cluster behaviour of handling every
+	// gateway. Sourced from HYPERSHELL_CLUSTER_ID.
+	ClusterID string
+
 	// ServiceAccountProvisionerAddress is the in-cluster bind address for the
 	// internal service-account provisioner gRPC server. A NetworkPolicy restricts
 	// the port to the API server pod, so the channel is plaintext (no mTLS).
@@ -63,6 +71,7 @@ func Load() (*Config, error) {
 		APIServerURL:                     getEnv("HYPERSHELL_API_SERVER_URL", "http://localhost:8000"),
 		Namespace:                        getEnv("HYPERSHELL_NAMESPACE", "hypershell"),
 		LogLevel:                         strings.ToLower(getEnv("HYPERSHELL_LOG_LEVEL", "info")),
+		ClusterID:                        getEnv("HYPERSHELL_CLUSTER_ID", ""),
 		ServiceAccountProvisionerAddress: getEnv("HYPERSHELL_SERVICE_ACCOUNT_PROVISIONER_BIND_ADDRESS", ""),
 
 		NamespaceGCEnabled:     getEnvBool("GATEWAY_NAMESPACE_GC_ENABLED", true),

components/control-plane/internal/reconciler/backfill.go

@@ -41,11 +41,17 @@ import (
 // two generic management labels, they are indistinguishable from another
 // HyperShell's namespaces on a shared cluster, so claiming them is unsafe. Those
 // require manual cleanup.
-func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gwClient pb.GatewayServiceClient, instance string) (int, error) {
+func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gwClient pb.GatewayServiceClient, instance, clusterID string) (int, error) {
 	if instance == "" {
 		return 0, fmt.Errorf("refusing to backfill instance labels without a control-plane instance identity")
 	}
-	gateways, err := listAllGateways(ctx, gwClient)
+	// Scope the listing to this cluster's gateways (clusterID from
+	// HYPERSHELL_CLUSTER_ID; empty in single-cluster mode lists all). Unlike the
+	// namespace reaper -- which lists unfiltered because a superset only ever
+	// protects namespaces -- backfill WRITES this instance's label, so on a
+	// co-located cluster an unscoped list could stamp another instance's
+	// not-yet-labeled namespace. Filtering keeps a spoke to its own gateways.
+	gateways, err := listAllGateways(ctx, gwClient, clusterID)
 	if err != nil {
 		// A partial inventory would silently skip namespaces that need the label,
 		// leaving them to leak; fail so the caller can log and retry on the next
@@ -86,9 +92,9 @@ func BackfillInstanceLabels(ctx context.Context, client kubernetes.Interface, gw
 // but never aborts controller startup, because the periodic GC sweep still
 // functions for already-labeled namespaces and the backfill is retried on the
 // next restart.
-func RunInstanceLabelBackfill(ctx context.Context, client kubernetes.Interface, conn *grpc.ClientConn, instance string) {
+func RunInstanceLabelBackfill(ctx context.Context, client kubernetes.Interface, conn *grpc.ClientConn, instance, clusterID string) {
 	gwClient := pb.NewGatewayServiceClient(conn)
-	labeled, err := BackfillInstanceLabels(ctx, client, gwClient, instance)
+	labeled, err := BackfillInstanceLabels(ctx, client, gwClient, instance, clusterID)
 	if err != nil {
 		log.Printf("WARN instance-label backfill completed with errors (labeled=%d): %v", labeled, err)
 		return

components/control-plane/internal/reconciler/backfill_test.go

@@ -72,7 +72,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 			gatewayWithNamespace("id-empty", ""),
 		})
 
-		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell")
+		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "")
 		if err != nil {
 			t.Fatalf("BackfillInstanceLabels() error = %v", err)
 		}
@@ -104,7 +104,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 				return nil, fmt.Errorf("boom")
 			},
 		}
-		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell"); err == nil {
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", ""); err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want list failure")
 		}
 		got, err := client.CoreV1().Namespaces().Get(ctx, "openshell-a", metav1.GetOptions{})
@@ -133,7 +133,7 @@ func TestBackfillInstanceLabels(t *testing.T) {
 			gatewayWithNamespace("id-b", "openshell-b"),
 		})
 
-		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell")
+		labeled, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "")
 		if err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want a collected per-namespace error")
 		}
@@ -152,8 +152,25 @@ func TestBackfillInstanceLabels(t *testing.T) {
 	t.Run("refuses an empty instance identity", func(t *testing.T) {
 		client := fake.NewSimpleClientset()
 		gwClient := singlePageGatewayClient(nil)
-		if _, err := BackfillInstanceLabels(ctx, client, gwClient, ""); err == nil {
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "", ""); err == nil {
 			t.Fatalf("BackfillInstanceLabels() error = nil, want empty instance error")
 		}
 	})
+
+	t.Run("scopes the gateway listing to this cluster", func(t *testing.T) {
+		client := fake.NewSimpleClientset()
+		var gotClusterID *string
+		gwClient := &fakeGatewayClient{
+			listFn: func(ctx context.Context, in *pb.ListGatewaysRequest, opts ...grpc.CallOption) (*pb.ListGatewaysResponse, error) {
+				gotClusterID = in.ClusterId
+				return &pb.ListGatewaysResponse{Metadata: &pb.ListMeta{Total: 0}}, nil
+			},
+		}
+		if _, err := BackfillInstanceLabels(ctx, client, gwClient, "hypershell", "mc1"); err != nil {
+			t.Fatalf("BackfillInstanceLabels() error = %v", err)
+		}
+		if gotClusterID == nil || *gotClusterID != "mc1" {
+			t.Fatalf("cluster_id = %v, want \"mc1\" (backfill must scope to its own cluster)", gotClusterID)
+		}
+	})
 }