diff --git a/cmd/sharded-test-server/frontproxy.go b/cmd/sharded-test-server/frontproxy.go index 02c06b89325..f635bef3b1f 100644 --- a/cmd/sharded-test-server/frontproxy.go +++ b/cmd/sharded-test-server/frontproxy.go @@ -39,6 +39,7 @@ import ( "k8s.io/klog/v2" kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" + kcptesting "github.com/kcp-dev/sdk/testing" kcptestingserver "github.com/kcp-dev/sdk/testing/server" "github.com/kcp-dev/sdk/testing/third_party/library-go/crypto" @@ -131,6 +132,17 @@ func startFrontProxy( return nil, fmt.Errorf("failed to write server cert: %w", err) } + // The front-proxy validates service account tokens itself, as it does in + // production: all shards sign with the one key under .kcp, and the proxy + // resolves the service account on the shard hosting its workspace + // through its index. With service account authentication enabled the + // proxy forwards nothing unauthenticated, so the static e2e users need + // their token file here as well. + tokenAuthFile, err := kcptesting.WriteDefaultTokenAuthFile(filepath.Join(workDirPath, ".kcp-front-proxy")) + if err != nil { + return nil, fmt.Errorf("failed to write token auth file: %w", err) + } + // run front-proxy command workdir, commandLine := kcptestingserver.Command("kcp-front-proxy", "front-proxy") commandLine = append(commandLine, @@ -155,6 +167,8 @@ func startFrontProxy( "--requestheader-allowed-names=kcp-front-proxy,kcp-mounts-proxy", fmt.Sprintf("--tls-cert-file=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "apiserver.crt")), fmt.Sprintf("--tls-private-key-file=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "apiserver.key")), + fmt.Sprintf("--service-account-key-file=%s", filepath.Join(workDirPath, ".kcp", "service-account.crt")), + fmt.Sprintf("--token-auth-file=%s", tokenAuthFile), "--secure-port=6443", "--v=4", ) diff --git a/docs/content/concepts/sharding/shards.md b/docs/content/concepts/sharding/shards.md index fd095707fba..1c0115a755f 100644 --- a/docs/content/concepts/sharding/shards.md +++ b/docs/content/concepts/sharding/shards.md @@ -132,9 +132,9 @@ front-proxy, add a `/services/` path mapping; any shard can be the backend. The front-proxy itself discovers shards through the Admin workspace, and reads the identities of the core APIs from the `apiidentities` registry there, so it starts and runs without the root shard: the kubeconfigs given -via `--shard-peer-kubeconfig` (falling back to the `--root-kubeconfig` -server) are seeds, used to bootstrap discovery and as a last-resort fallback -only. Every shard observed through the view joins the +via `--shard-peer-kubeconfig`, which is required, are seeds, used to +bootstrap discovery and as a last-resort fallback only; `--root-kubeconfig` +is ignored. Every shard observed through the view joins the failover peer set at runtime and decommissioned shards leave it; requests are distributed across the discovered shards, not the seeds. With short-lived shards the discovery channel therefore keeps working as long as any currently diff --git a/pkg/index/indextest/index.go b/pkg/index/indextest/index.go new file mode 100644 index 00000000000..82d437cd57f --- /dev/null +++ b/pkg/index/indextest/index.go @@ -0,0 +1,91 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package indextest provides a test double for index.Index. +package indextest + +import ( + "context" + "sync/atomic" + + "github.com/kcp-dev/logicalcluster/v3" + + "github.com/kcp-dev/kcp/pkg/index" +) + +var _ index.Index = &Index{} + +// Index is a map-based index.Index: it maps logical clusters to shards and +// shards to base URLs. Lookups resolve logical cluster names only, not +// workspace paths. The maps must not be changed once the Index is in use. +type Index struct { + // Clusters maps a logical cluster to the name of the shard hosting it. + Clusters map[logicalcluster.Name]string + // Shards maps a shard name to its base URL. + Shards map[string]string + // RecentlyMigratedClusters are the clusters RecentlyMigrated reports. + RecentlyMigratedClusters map[logicalcluster.Name]bool + + // Registered and Released count the ClusterContext calls and the + // calls of the cancel functions they returned. + Registered atomic.Int32 + Released atomic.Int32 +} + +// Lookup returns the shard hosting the cluster, with the cluster's URL on +// that shard. +func (f *Index) Lookup(path logicalcluster.Path) (index.Result, bool) { + name, ok := path.Name() + if !ok { + return index.Result{}, false + } + shard, ok := f.Clusters[name] + if !ok { + return index.Result{}, false + } + return index.Result{ + Shard: shard, + Cluster: name, + URL: f.Shards[shard] + name.Path().RequestPath(), + }, true +} + +// LookupURL is Lookup: the fake knows no mounts. +func (f *Index) LookupURL(path logicalcluster.Path) (index.Result, bool) { + return f.Lookup(path) +} + +// BaseURL returns the base URL of the shard. +func (f *Index) BaseURL(shardName string) (string, bool) { + url, ok := f.Shards[shardName] + return url, ok +} + +// RecentlyMigrated reports whether the cluster is in RecentlyMigratedClusters. +func (f *Index) RecentlyMigrated(cluster logicalcluster.Name) bool { + return f.RecentlyMigratedClusters[cluster] +} + +// ClusterContext returns a cancellable child of parent and counts the +// registration and its release. +func (f *Index) ClusterContext(parent context.Context, cluster logicalcluster.Name) (context.Context, context.CancelFunc) { + f.Registered.Add(1) + ctx, cancel := context.WithCancel(parent) + return ctx, func() { + f.Released.Add(1) + cancel() + } +} diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go index 8151aefcfe1..c5c024481f3 100644 --- a/pkg/proxy/config.go +++ b/pkg/proxy/config.go @@ -53,15 +53,14 @@ type ExtraConfig struct { // unknown. It is a readiness check: requests needing an identity fail // until it arrives, nothing blocks startup on it. CoreIdentitiesReady func() error - // RootShardConfig points at the root shard and is set only when - // --root-kubeconfig is given. Service account authentication still needs - // it. - RootShardConfig *rest.Config - ShardsConfig *rest.Config + ShardsConfig *rest.Config + // ServiceAccounts resolves service accounts and token secrets on the + // shard hosting them, through the index, for service account + // authentication. + ServiceAccounts *ShardRoutedServiceAccounts // PeersConfig points at the Admin workspace (/services/admin) of the // peer shards with round-robin and per-request failover, and is the only - // source for Shard discovery. When no peers are configured, the root - // kubeconfig's server acts as the seed peer. + // source for Shard discovery and the core identities. PeersConfig *rest.Config // Peers is the failover set behind PeersConfig: the configured seed // peers plus every discovered shard, fed at runtime from the shard @@ -98,22 +97,10 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error) var kcpShardIdentityRoundTripper func(rt http.RoundTripper) http.RoundTripper kcpShardIdentityRoundTripper, c.CoreIdentityHandler, c.CoreIdentitiesReady = bootstrap.NewCoreWildcardIdentities(bootstrap.KcpRootGroupExportNames, bootstrap.KcpRootGroupResourceExportNames) - if c.Options.RootKubeconfig != "" { - rootConfig, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{ExplicitPath: c.Options.RootKubeconfig}, nil).ClientConfig() - if err != nil { - return nil, fmt.Errorf("failed to load root kubeconfig: %w", err) - } - c.RootShardConfig = rest.CopyConfig(rootConfig) - c.RootShardConfig.Wrap(kcpShardIdentityRoundTripper) - } - var loopbackClientConfig *rest.Config if err := c.Options.SecureServing.ApplyTo(&c.ServingInfo, &loopbackClientConfig); err != nil { return nil, err } - if err := c.Options.Authentication.ApplyTo(ctx, &c.AuthenticationInfo, c.ServingInfo, c.RootShardConfig); err != nil { - return nil, err - } var err error c.ShardsConfig, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig( @@ -126,17 +113,16 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error) } c.ShardsConfig.Wrap(kcpShardIdentityRoundTripper) - // Shard discovery and the core identities always go through the Admin - // workspace of the peers; the root shard is never asked. Without - // configured peers, the root kubeconfig's server is the seed peer. - peerKubeconfigs := c.Options.ShardsPeerKubeconfigs - if len(peerKubeconfigs) == 0 { - if c.Options.RootKubeconfig == "" { - return nil, fmt.Errorf("--shard-peer-kubeconfig is required") - } - peerKubeconfigs = []string{c.Options.RootKubeconfig} + // Service account tokens name their logical cluster; the lookup finds the + // hosting shard through the index and asks it, with the shards credential. + c.ServiceAccounts = NewShardRoutedServiceAccounts(c.ShardsConfig) + if err := c.Options.Authentication.ApplyTo(ctx, &c.AuthenticationInfo, c.ServingInfo, c.ServiceAccounts); err != nil { + return nil, err } - c.PeersConfig, c.Peers, err = NewPeersConfig(peerKubeconfigs) + + // Shard discovery and the core identities always go through the Admin + // workspace of the peers; the root shard is never asked. + c.PeersConfig, c.Peers, err = NewPeersConfig(c.Options.ShardsPeerKubeconfigs) if err != nil { return nil, err } diff --git a/pkg/proxy/index/index_controller.go b/pkg/proxy/index/index_controller.go index 7e463c732e1..a1d30744bd9 100644 --- a/pkg/proxy/index/index_controller.go +++ b/pkg/proxy/index/index_controller.go @@ -23,6 +23,7 @@ import ( "time" "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/labels" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" @@ -73,6 +74,7 @@ func NewController( shardWorkspaceInformers: map[string]cache.SharedIndexInformer{}, shardLogicalClusterInformers: map[string]cache.SharedIndexInformer{}, shardWorkspaceStopCh: map[string]chan struct{}{}, + shardInformersSynced: map[string][]cache.InformerSynced{}, state: *index.New([]index.PathRewriter{ indexrewriters.UserRewriter, @@ -133,6 +135,9 @@ type Controller struct { shardWorkspaceInformers map[string]cache.SharedIndexInformer shardLogicalClusterInformers map[string]cache.SharedIndexInformer shardWorkspaceStopCh map[string]chan struct{} + // shardInformersSynced holds, per shard, the sync state of its + // informers; Ready consults it. + shardInformersSynced map[string][]cache.InformerSynced state index.State } @@ -274,6 +279,7 @@ func (c *Controller) process(ctx context.Context, key string) error { c.shardWorkspaceInformers[shard.Name] = wsInformer c.shardLogicalClusterInformers[shard.Name] = twInformer c.shardWorkspaceStopCh[shard.Name] = stopCh + c.shardInformersSynced[shard.Name] = []cache.InformerSynced{wsInformer.HasSynced, twInformer.HasSynced} go wsInformer.Run(stopCh) go twInformer.Run(stopCh) @@ -296,6 +302,47 @@ func (c *Controller) stopShard(shardName string) { delete(c.shardWorkspaceStopCh, shardName) delete(c.shardWorkspaceInformers, shardName) delete(c.shardLogicalClusterInformers, shardName) + delete(c.shardInformersSynced, shardName) +} + +// Ready reports whether the index knows the workspaces and logical clusters +// of every Shard: for each Shard the lister knows, the per-shard informers +// have been started and have completed their initial list. Until then a +// lookup of a cluster on a not yet indexed shard answers "not found", which +// the proxy turns into a 404 and the service account authenticator into a +// 401, so this is a readiness check rather than a liveness one. A Shard +// whose informers are not running yet, e.g. because its client cannot be +// built, keeps the index not ready. +// +// Cordoned Shards, those carrying the unschedulable annotation, are left +// out: cordoning is how a shard is put into maintenance, and a shard that +// is down for maintenance must not keep every proxy out of rotation, least +// of all on a proxy restart during the maintenance. Their workspaces are +// routable once their informers have synced. +func (c *Controller) Ready() error { + shards, err := c.shardLister.List(labels.Everything()) + if err != nil { + return fmt.Errorf("listing shards: %w", err) + } + + c.lock.RLock() + defer c.lock.RUnlock() + + for _, shard := range shards { + if _, cordoned := shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; cordoned { + continue + } + synced, found := c.shardInformersSynced[shard.Name] + if !found { + return fmt.Errorf("informers for shard %q are not started", shard.Name) + } + for _, hasSynced := range synced { + if !hasSynced() { + return fmt.Errorf("informers for shard %q have not synced", shard.Name) + } + } + } + return nil } func (c *Controller) Lookup(path logicalcluster.Path) (index.Result, bool) { diff --git a/pkg/proxy/index/index_controller_test.go b/pkg/proxy/index/index_controller_test.go index b0088aab13f..4218bd47896 100644 --- a/pkg/proxy/index/index_controller_test.go +++ b/pkg/proxy/index/index_controller_test.go @@ -28,19 +28,24 @@ import ( "github.com/kcp-dev/logicalcluster/v3" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + corev1alpha1listers "github.com/kcp-dev/sdk/client/listers/core/v1alpha1" "github.com/kcp-dev/kcp/pkg/index" ) func newTestController() *Controller { + shardIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) return &Controller{ queue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: "test"}, ), + shardIndexer: shardIndexer, + shardLister: corev1alpha1listers.NewShardLister(shardIndexer), shardWorkspaceInformers: map[string]cache.SharedIndexInformer{}, shardLogicalClusterInformers: map[string]cache.SharedIndexInformer{}, shardWorkspaceStopCh: map[string]chan struct{}{}, + shardInformersSynced: map[string][]cache.InformerSynced{}, state: *index.New(nil), } } @@ -111,3 +116,75 @@ func TestOnShardDelete(t *testing.T) { _, found := c.state.LookupURL(logicalcluster.NewPath("myws")) require.False(t, found, "expected the shard's clusters to be unroutable after delete") } + +// TestReady walks the readiness of the index through the life of a shard: +// unknown, known but informers not started, started but still syncing, +// synced, and gone. +func TestReady(t *testing.T) { + t.Parallel() + c := newTestController() + + require.NoError(t, c.Ready(), "an index without shards is ready") + + alpha := shardObj("alpha", "https://alpha") + require.NoError(t, c.shardIndexer.Add(alpha)) + err := c.Ready() + require.Error(t, err, "a shard without informers keeps the index not ready") + require.Contains(t, err.Error(), `"alpha"`) + require.Contains(t, err.Error(), "not started") + + synced := false + c.lock.Lock() + c.shardInformersSynced["alpha"] = []cache.InformerSynced{func() bool { return true }, func() bool { return synced }} + c.lock.Unlock() + err = c.Ready() + require.Error(t, err, "a shard whose informers have not synced keeps the index not ready") + require.Contains(t, err.Error(), "not synced") + + synced = true + require.NoError(t, c.Ready(), "all informers synced") + + beta := shardObj("beta", "https://beta") + require.NoError(t, c.shardIndexer.Add(beta)) + require.Error(t, c.Ready(), "a new shard is not ready until its informers are") + + c.stopShard("beta") + require.NoError(t, c.shardIndexer.Delete(beta)) + require.NoError(t, c.Ready(), "a deleted shard no longer counts") + + c.stopShard("alpha") + require.Error(t, c.Ready(), "stopping the informers of a shard that still exists makes the index not ready") +} + +// TestReadyCordonedShard checks that a cordoned shard, i.e. one in +// maintenance, does not hold the index back, and counts again once it is +// uncordoned. +func TestReadyCordonedShard(t *testing.T) { + t.Parallel() + c := newTestController() + + alpha := shardObj("alpha", "https://alpha") + require.NoError(t, c.shardIndexer.Add(alpha)) + c.lock.Lock() + c.shardInformersSynced["alpha"] = []cache.InformerSynced{func() bool { return true }} + c.lock.Unlock() + + beta := shardObj("beta", "https://beta") + require.NoError(t, c.shardIndexer.Add(beta)) + require.Error(t, c.Ready(), "an uncordoned shard that is not indexed keeps the index not ready") + + cordoned := beta.DeepCopy() + cordoned.Annotations = map[string]string{corev1alpha1.ShardUnschedulableAnnotationKey: "true"} + require.NoError(t, c.shardIndexer.Update(cordoned)) + require.NoError(t, c.Ready(), "a cordoned shard does not count") + + synced := false + c.lock.Lock() + c.shardInformersSynced["beta"] = []cache.InformerSynced{func() bool { return synced }} + c.lock.Unlock() + require.NoError(t, c.shardIndexer.Update(beta)) + require.Error(t, c.Ready(), "an uncordoned shard counts again") + + synced = true + require.NoError(t, c.Ready()) +} diff --git a/pkg/proxy/metrics_authorizer_test.go b/pkg/proxy/metrics_authorizer_test.go index 1458211a9d3..3d88226451a 100644 --- a/pkg/proxy/metrics_authorizer_test.go +++ b/pkg/proxy/metrics_authorizer_test.go @@ -39,8 +39,8 @@ func TestMetricsGroupAuthorizer(t *testing.T) { {name: "kcp admin may scrape", groups: []string{bootstrappolicy.SystemKcpAdminGroup, user.AllAuthenticated}, want: authorizer.DecisionAllow}, {name: "system:masters gets no opinion", groups: []string{bootstrappolicy.SystemMastersGroup}, want: authorizer.DecisionNoOpinion}, {name: "the metrics-reader group may scrape", groups: []string{bootstrappolicy.SystemKcpMetricsReader}, want: authorizer.DecisionAllow}, - // no opinion, not deny: a further authorizer in the union (the :root - // SubjectAccessReview while --root-kubeconfig is given) may still allow. + // no opinion rather than deny: the authorization filter treats it as + // forbidden, and a further authorizer could still be chained. {name: "an ordinary user gets no opinion", groups: []string{user.AllAuthenticated}, want: authorizer.DecisionNoOpinion}, {name: "no groups gets no opinion", groups: nil, want: authorizer.DecisionNoOpinion}, } { diff --git a/pkg/proxy/options/authentication.go b/pkg/proxy/options/authentication.go index 9bd5c7c1c0c..6d321cfae91 100644 --- a/pkg/proxy/options/authentication.go +++ b/pkg/proxy/options/authentication.go @@ -20,27 +20,29 @@ import ( "context" "fmt" "strings" - "time" "github.com/spf13/pflag" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apiserver/pkg/authentication/user" genericapiserver "k8s.io/apiserver/pkg/server" - "k8s.io/client-go/rest" "k8s.io/client-go/util/keyutil" - serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount" kubeoptions "k8s.io/kubernetes/pkg/kubeapiserver/options" "k8s.io/kubernetes/pkg/serviceaccount" - kcpkubernetesinformers "github.com/kcp-dev/client-go/informers" - kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" + kcpcorev1client "github.com/kcp-dev/client-go/kubernetes/typed/core/v1" kcpauthentication "github.com/kcp-dev/kcp/pkg/authentication" "github.com/kcp-dev/kcp/pkg/authorization/bootstrap" ) -const resyncPeriod = 10 * time.Hour +// ServiceAccountLookup resolves the service accounts and token secrets that +// service account tokens name, on whichever shard hosts them, and gives the +// legacy token validator a writer for its last-used annotations. +type ServiceAccountLookup interface { + serviceaccount.ServiceAccountTokenClusterGetter + SecretsWriter() kcpcorev1client.SecretClusterInterface +} // Authentication wraps BuiltInAuthenticationOptions so we can minimize the // dependencies on apiserver auth machinery, specifically by overriding the @@ -96,17 +98,15 @@ func (c *Authentication) tokenAuthEnabled() bool { return c.BuiltInOptions.TokenFile != nil && c.BuiltInOptions.TokenFile.TokenFile != "" } -// ServiceAccountAuthEnabled reports whether service account tokens are -// accepted, which needs a client to the root shard. -func (c *Authentication) ServiceAccountAuthEnabled() bool { - return c.serviceAccountAuthEnabled() -} - func (c *Authentication) serviceAccountAuthEnabled() bool { return c.BuiltInOptions.ServiceAccounts != nil && len(c.BuiltInOptions.ServiceAccounts.KeyFiles) != 0 } -func (c *Authentication) ApplyTo(ctx context.Context, authenticationInfo *genericapiserver.AuthenticationInfo, servingInfo *genericapiserver.SecureServingInfo, rootShardConfig *rest.Config) error { +// ApplyTo builds the authenticator. serviceAccounts resolves the service +// accounts and token secrets that service account tokens name, on whichever +// shard hosts them; it is only consulted when service account authentication +// is enabled. +func (c *Authentication) ApplyTo(ctx context.Context, authenticationInfo *genericapiserver.AuthenticationInfo, servingInfo *genericapiserver.SecureServingInfo, serviceAccounts ServiceAccountLookup) error { // Note BuiltInAuthenticationOptions.ApplyTo is not called, so we // can reduce the dependencies pulled in from auth methods which aren't enabled authenticatorConfig, err := c.BuiltInOptions.ToAuthenticationConfig() @@ -135,30 +135,16 @@ func (c *Authentication) ApplyTo(ctx context.Context, authenticationInfo *generi // Set for service account auth, if enabled if c.serviceAccountAuthEnabled() { - if rootShardConfig == nil { - return fmt.Errorf("service account authentication needs --root-kubeconfig") + if serviceAccounts == nil { + return fmt.Errorf("service account authentication needs a service account lookup") } authenticationInfo.APIAudiences = c.BuiltInOptions.APIAudiences if len(c.BuiltInOptions.ServiceAccounts.Issuers) != 0 && len(c.BuiltInOptions.APIAudiences) == 0 { authenticationInfo.APIAudiences = c.BuiltInOptions.ServiceAccounts.Issuers } - config := rest.CopyConfig(rootShardConfig) - tokenGetterClient, err := kcpkubernetesclientset.NewForConfig(config) - if err != nil { - return fmt.Errorf("failed to create client for ServiceAccountTokenGetter: %w", err) - } - - versionedInformers := kcpkubernetesinformers.NewSharedInformerFactory(tokenGetterClient, resyncPeriod) - - authenticatorConfig.ServiceAccountTokenGetter = serviceaccountcontroller.NewClusterGetterFromClient( - tokenGetterClient, - versionedInformers.Core().V1().Secrets().Lister(), - versionedInformers.Core().V1().ServiceAccounts().Lister(), - versionedInformers.Admissionregistration().V1().ValidatingWebhookConfigurations().Lister(), - versionedInformers.Admissionregistration().V1().MutatingWebhookConfigurations().Lister(), - ) - authenticatorConfig.SecretsWriter = tokenGetterClient.CoreV1().Secrets() + authenticatorConfig.ServiceAccountTokenGetter = serviceAccounts + authenticatorConfig.SecretsWriter = serviceAccounts.SecretsWriter() if len(c.BuiltInOptions.ServiceAccounts.KeyFiles) > 0 { // Load and set the public keys. diff --git a/pkg/proxy/options/options.go b/pkg/proxy/options/options.go index 3f6e64c8529..3f84a453073 100644 --- a/pkg/proxy/options/options.go +++ b/pkg/proxy/options/options.go @@ -30,10 +30,12 @@ type Options struct { Authentication Authentication MappingFile string RootDirectory string - // RootKubeconfig points at the root shard. + // RootKubeconfig is parsed but ignored. The proxy does not use the root + // shard: shard discovery, the core identities and service account lookups + // go through the peers, and --shard-peer-kubeconfig is required. A proxy + // started with --root-kubeconfig alone fails validation. // - // TODO: This is going to be removed once service account - // authentication no longer requires the root shard. + // Deprecated: ignored, to be removed. RootKubeconfig string ShardsKubeconfig string ShardsPeerKubeconfigs []string @@ -45,7 +47,6 @@ func NewOptions() *Options { o := &Options{ SecureServing: *apiserveroptions.NewSecureServingOptions().WithLoopback(), Authentication: *NewAuthentication(), - RootKubeconfig: "", RootDirectory: ".kcp", } @@ -63,12 +64,12 @@ func (o *Options) AddFlags(fss *cliflag.NamedFlagSets) { fs := fss.FlagSet("proxy") fs.StringVar(&o.MappingFile, "mapping-file", o.MappingFile, "Config file mapping paths to backends") fs.StringVar(&o.RootDirectory, "root-directory", o.RootDirectory, "Root directory.") - fs.StringVar(&o.RootKubeconfig, "root-kubeconfig", o.RootKubeconfig, "The path to the kubeconfig of the root shard. Only needed for service account authentication; without --shard-peer-kubeconfig its server also acts as the seed peer.") - if err := fs.MarkDeprecated("root-kubeconfig", "use --shard-peer-kubeconfig; the root shard is needed neither for shard discovery nor for the identities of the core APIs any more. It is still required for service account authentication."); err != nil { + fs.StringVar(&o.RootKubeconfig, "root-kubeconfig", "", "Ignored. The proxy does not use the root shard any more; --shard-peer-kubeconfig is required instead.") + if err := fs.MarkDeprecated("root-kubeconfig", "it is ignored; the proxy discovers shards, the core identities and service accounts through --shard-peer-kubeconfig, which is required."); err != nil { panic(err) } fs.StringVar(&o.ShardsKubeconfig, "shards-kubeconfig", o.ShardsKubeconfig, "The path to the kubeconfig used for communication with all shards. The server name if provided is replaced with a shard's hostname.") - fs.StringSliceVar(&o.ShardsPeerKubeconfigs, "shard-peer-kubeconfig", o.ShardsPeerKubeconfigs, "Repeatable path to a kubeconfig whose named clusters are seed peer shards; all files' clusters are merged into one seed list. The proxy discovers Shards and the identities of the core APIs through the Admin workspace (/services/admin) of the peers - requests are distributed round-robin with failover - and never talks to the root shard for them. If unset, the --root-kubeconfig server acts as the only seed peer. The seeds are used for bootstrapping and as a last-resort fallback only: once shards are discovered, requests go to the discovered shards, which join and leave the peer set at runtime, so the files do not have to track short-lived shards.") + fs.StringSliceVar(&o.ShardsPeerKubeconfigs, "shard-peer-kubeconfig", o.ShardsPeerKubeconfigs, "Repeatable path to a kubeconfig whose named clusters are seed peer shards; all files' clusters are merged into one seed list. The proxy discovers Shards and the identities of the core APIs through the Admin workspace (/services/admin) of the peers - requests are distributed round-robin with failover - and never talks to the root shard for them. The seeds are used for bootstrapping and as a last-resort fallback only: once shards are discovered, requests go to the discovered shards, which join and leave the peer set at runtime, so the files do not have to track short-lived shards.") fs.StringVar(&o.ProfilerAddress, "profiler-address", "", "[Address]:port to bind the profiler to") fs.StringSliceVar(&o.CorsAllowedOriginList, "cors-allowed-origins", o.CorsAllowedOriginList, "List of allowed origins for CORS, comma separated. An allowed origin can be a regular expression to support subdomain matching. If this list is empty CORS will not be enabled.") } @@ -101,12 +102,9 @@ func (o *Options) Validate() []error { if len(o.ShardsKubeconfig) == 0 { errs = append(errs, fmt.Errorf("--shards-kubeconfig is required")) } - if len(o.ShardsPeerKubeconfigs) == 0 && len(o.RootKubeconfig) == 0 { + if len(o.ShardsPeerKubeconfigs) == 0 { errs = append(errs, fmt.Errorf("--shard-peer-kubeconfig is required")) } - if o.Authentication.ServiceAccountAuthEnabled() && len(o.RootKubeconfig) == 0 { - errs = append(errs, fmt.Errorf("--root-kubeconfig is required for service account authentication")) - } errs = append(errs, o.SecureServing.Validate()...) errs = append(errs, o.Authentication.Validate()...) diff --git a/pkg/proxy/server.go b/pkg/proxy/server.go index 4b6aacd82a8..9d11b0cd158 100644 --- a/pkg/proxy/server.go +++ b/pkg/proxy/server.go @@ -30,7 +30,6 @@ import ( "k8s.io/apimachinery/pkg/runtime/serializer" "k8s.io/apiserver/pkg/authentication/authenticator" "k8s.io/apiserver/pkg/authorization/authorizer" - "k8s.io/apiserver/pkg/authorization/union" genericapifilters "k8s.io/apiserver/pkg/endpoints/filters" "k8s.io/apiserver/pkg/endpoints/request" genericfilters "k8s.io/apiserver/pkg/server/filters" @@ -40,7 +39,6 @@ import ( "k8s.io/component-base/metrics/legacyregistry" "k8s.io/klog/v2" - kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" adminv1alpha1 "github.com/kcp-dev/sdk/apis/admin/v1alpha1" "github.com/kcp-dev/sdk/apis/core" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" @@ -48,7 +46,6 @@ import ( kcpinformers "github.com/kcp-dev/sdk/client/informers/externalversions" "github.com/kcp-dev/kcp/pkg/authentication" - "github.com/kcp-dev/kcp/pkg/authorization/delegated" kcpfeatures "github.com/kcp-dev/kcp/pkg/features" frontproxyfilters "github.com/kcp-dev/kcp/pkg/proxy/filters" "github.com/kcp-dev/kcp/pkg/proxy/index" @@ -159,6 +156,7 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { // watching a number of resources on each. The controller is then also satisfying the Index // interface. s.IndexController = index.NewController(ctx, s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), getClientFunc) + s.CompletedConfig.ServiceAccounts.SetIndex(s.IndexController) handler, err := NewHandler(ctx, mappings) if err != nil { @@ -211,9 +209,14 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { handler = genericfilters.WithCORS(handler, c.Options.CorsAllowedOriginList, nil, nil, nil, "true") mux := http.NewServeMux() + // Readiness covers everything a request needs: the Shard informer, the + // core identities, and the workspace index having learned the clusters + // of every shard. Without the last one a path lookup 404s and a service + // account token 401s, so load balancers must hold traffic until then. healthz.InstallReadyzHandler(mux, healthz.NewInformerSyncHealthz(s.KcpSharedInformerFactory), healthz.NamedCheck("core-identities", func(*http.Request) error { return s.CompletedConfig.CoreIdentitiesReady() }), + healthz.NamedCheck("workspace-index", func(*http.Request) error { return s.IndexController.Ready() }), ) healthz.InstallLivezHandler(mux, healthz.PingHealthz) @@ -222,16 +225,9 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { // the endpoint is world-readable and leaks shard-wide operational data // (logical-cluster counts, client-cert expiry, request patterns). Members // of the system:kcp:admin and system:kcp:metrics-reader groups may - // scrape; with --root-kubeconfig a :root SubjectAccessReview - // is consulted as well, so the system:kcp:metrics-reader binding that - // authorizes scraping on the shards keeps working here during the - // deprecation. Registered before "/" so it takes precedence over the proxy + // scrape. Registered before "/" so it takes precedence over the proxy // handler for this exact path. - metricsHandler, err := newProtectedMetricsHandler(c.AuthenticationInfo.Authenticator, c.RootShardConfig, requestInfoFactory) - if err != nil { - return nil, fmt.Errorf("failed to create protected metrics handler: %w", err) - } - mux.Handle("/metrics", metricsHandler) + mux.Handle("/metrics", withMetricsAuthorization(legacyregistry.Handler(), c.AuthenticationInfo.Authenticator, newMetricsGroupAuthorizer(), requestInfoFactory)) mux.Handle("/", handler) s.Handler = mux @@ -239,28 +235,6 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { return s, nil } -func newProtectedMetricsHandler(auth authenticator.Request, rootShardConfig *restclient.Config, requestInfoResolver request.RequestInfoResolver) (http.Handler, error) { - authorizers := []union.NamedAuthorizer{{AuthorizerName: "MetricsGroups", Authorizer: newMetricsGroupAuthorizer()}} - if rootShardConfig != nil { - kubeClusterClient, err := kcpkubernetesclientset.NewForConfig(rootShardConfig) - if err != nil { - return nil, fmt.Errorf("failed to create root shard client for metrics authorization: %w", err) - } - authz, err := delegated.NewDelegatedAuthorizer(core.RootCluster, kubeClusterClient, delegated.Options{}) - if err != nil { - return nil, fmt.Errorf("failed to create metrics authorizer: %w", err) - } - authorizers = append(authorizers, union.NamedAuthorizer{AuthorizerName: "RootShardDelegated", Authorizer: authz}) - } - - unioned, err := union.New(authorizers...) - if err != nil { - return nil, fmt.Errorf("failed to create metrics authorizer union: %w", err) - } - - return withMetricsAuthorization(legacyregistry.Handler(), auth, unioned, requestInfoResolver), nil -} - func withMetricsAuthorization(h http.Handler, auth authenticator.Request, authz authorizer.Authorizer, requestInfoResolver request.RequestInfoResolver) http.Handler { scheme := runtime.NewScheme() metav1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"}) @@ -285,23 +259,33 @@ func (s preparedServer) Run(ctx context.Context) error { logger := klog.FromContext(ctx).WithValues("component", "proxy") // Both factories read the Admin workspace of the peers: any reachable - // shard suffices, the root shard is not special. The core identities - // arrive through the second one; until they do, requests that need them - // fail and readiness reports it, nothing blocks here on the root shard. + // shard suffices, the root shard is not special. Nothing is awaited + // before serving: /readyz reports the informers' sync, the core + // identities and the workspace index, and requests that need any of + // them fail until they are there. s.KcpSharedInformerFactory.Start(ctx.Done()) s.CoreIdentityInformerFactory.Start(ctx.Done()) - s.KcpSharedInformerFactory.WaitForCacheSync(ctx.Done()) - s.CoreIdentityInformerFactory.WaitForCacheSync(ctx.Done()) - if err := s.CompletedConfig.CoreIdentitiesReady(); err != nil { - logger.Info("starting without all core identities, waiting for the registry", "reason", err.Error()) - } - - // start indexes go s.IndexController.Start(ctx, 2) - + s.CompletedConfig.ServiceAccounts.Start(ctx) if s.AuthController != nil { go s.AuthController.Start(ctx, 2) } + // Wait for the sync off the serving path. Blocking here would keep the + // port closed for as long as no peer is reachable, so /livez would fail + // and the proxy be restarted in a loop instead of waiting for a shard, + // and /readyz could not tell why. Nothing waits on this goroutine and it + // gates nothing; it only logs. Traffic is gated by /readyz above + // (informer sync, core-identities, workspace-index), and requests needing + // anything not yet loaded fail until it is. + go func() { + s.KcpSharedInformerFactory.WaitForCacheSync(ctx.Done()) + s.CoreIdentityInformerFactory.WaitForCacheSync(ctx.Done()) + if err := s.CompletedConfig.CoreIdentitiesReady(); err != nil { + logger.Info("informers synced, core identities not all known yet", "reason", err.Error()) + return + } + logger.Info("informers synced, core identities known") + }() doneCh, _, err := s.CompletedConfig.ServingInfo.Serve(s.Handler, time.Second*60, ctx.Done()) if err != nil { diff --git a/pkg/proxy/serviceaccounts.go b/pkg/proxy/serviceaccounts.go new file mode 100644 index 00000000000..9080a9e1721 --- /dev/null +++ b/pkg/proxy/serviceaccounts.go @@ -0,0 +1,329 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "context" + "sync" + "time" + + admissionregistrationv1 "k8s.io/api/admissionregistration/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/watch" + corev1ac "k8s.io/client-go/applyconfigurations/core/v1" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + "k8s.io/client-go/rest" + "k8s.io/kubernetes/pkg/serviceaccount" + + kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" + kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" + kcpcorev1client "github.com/kcp-dev/client-go/kubernetes/typed/core/v1" + "github.com/kcp-dev/logicalcluster/v3" + + kcpindex "github.com/kcp-dev/kcp/pkg/index" + "github.com/kcp-dev/kcp/pkg/shardlookup" +) + +// lookupTimeout bounds a ServiceAccount or Secret GET against a shard. The +// GET is shared by every request waiting on the same key, so it does not +// run under any one request's context. +const lookupTimeout = 30 * time.Second + +// ShardRoutedServiceAccounts looks ServiceAccounts and their token Secrets up +// on the shard that hosts the logical cluster a service account token names, +// found through the proxy's index, using the shards credential. It is what +// lets the proxy validate service account tokens of workspaces on any shard +// without a connection to the root shard. +// +// It serves two roles for the authenticator: the token getter, through +// Cluster, and the writer the legacy token validator stamps last-used +// annotations with, through SecretsWriter. +// +// ServiceAccounts and Secrets are cached per logical cluster for +// shardlookup.DefaultSuccessTTL, failed lookups for DefaultFailureTTL, the +// same as the shards cache them for logical clusters on other shards. A +// deleted service account or token secret therefore keeps authenticating +// at the proxy for up to that long. Index misses are not cached: a cluster +// the index learns about is found on the next request. +type ShardRoutedServiceAccounts struct { + shardsConfig *rest.Config + + serviceAccounts *shardlookup.TTLCache[*corev1.ServiceAccount] + secrets *shardlookup.TTLCache[*corev1.Secret] + + mu sync.Mutex + index shardIndex // nil until the index controller exists + clients map[string]shardClient // by shard name +} + +// shardClient is a client for one shard, remembered with the base URL it +// was built for so that a shard moving to a new URL gets a new client. +type shardClient struct { + baseURL string + client kcpkubernetesclientset.ClusterInterface +} + +// NewShardRoutedServiceAccounts creates the lookup over the shards credential. +// The index is attached with SetIndex once the index controller exists; +// lookups before that fail with ServiceUnavailable. +// Call Start to evict expired entries in the background. +func NewShardRoutedServiceAccounts(shardsConfig *rest.Config) *ShardRoutedServiceAccounts { + return newShardRoutedServiceAccounts(shardsConfig, shardlookup.TTLOptions{}) +} + +func newShardRoutedServiceAccounts(shardsConfig *rest.Config, ttl shardlookup.TTLOptions) *ShardRoutedServiceAccounts { + return &ShardRoutedServiceAccounts{ + shardsConfig: shardsConfig, + serviceAccounts: shardlookup.NewTTLCacheWithOptions[*corev1.ServiceAccount](ttl), + secrets: shardlookup.NewTTLCacheWithOptions[*corev1.Secret](ttl), + clients: map[string]shardClient{}, + } +} + +// Start evicts expired cache entries until ctx is done. Lookups do not +// depend on it; without it expired entries stay in memory until replaced. +func (s *ShardRoutedServiceAccounts) Start(ctx context.Context) { + s.serviceAccounts.StartWithContext(ctx) + s.secrets.StartWithContext(ctx) +} + +// shardIndex is the part of the proxy's index the lookups need: which shard +// hosts a logical cluster, and where that shard is. +type shardIndex interface { + Lookup(path logicalcluster.Path) (kcpindex.Result, bool) + BaseURL(shardName string) (string, bool) +} + +// SetIndex attaches the index the lookups resolve shards through. +func (s *ShardRoutedServiceAccounts) SetIndex(index shardIndex) { + s.mu.Lock() + defer s.mu.Unlock() + s.index = index +} + +// clientFor returns a client for the shard hosting the logical cluster. +func (s *ShardRoutedServiceAccounts) clientFor(cluster logicalcluster.Name) (kcpkubernetesclientset.ClusterInterface, error) { + s.mu.Lock() + index := s.index + s.mu.Unlock() + if index == nil { + return nil, apierrors.NewServiceUnavailable("the shard index is not started yet") + } + + result, found := index.Lookup(cluster.Path()) + if !found || result.ErrorCode != 0 || result.Shard == "" { + return nil, apierrors.NewNotFound(corev1.Resource("logicalclusters"), cluster.String()) + } + baseURL, ok := index.BaseURL(result.Shard) + if !ok { + return nil, apierrors.NewNotFound(corev1.Resource("shards"), result.Shard) + } + + s.mu.Lock() + defer s.mu.Unlock() + if cached, ok := s.clients[result.Shard]; ok && cached.baseURL == baseURL { + return cached.client, nil + } + + // A miss means the shard set changed: a shard is new or moved. Shards + // are few and short-lived ones come and go, so this is the moment to + // drop the clients of shards the index no longer knows, or knows under + // another URL, instead of letting the cache grow for the life of the + // process. + for name, cached := range s.clients { + if current, ok := index.BaseURL(name); !ok || current != cached.baseURL { + delete(s.clients, name) + } + } + + config := rest.CopyConfig(s.shardsConfig) + config.Host = baseURL + client, err := kcpkubernetesclientset.NewForConfig(config) + if err != nil { + return nil, err + } + s.clients[result.Shard] = shardClient{baseURL: baseURL, client: client} + return client, nil +} + +// Cluster implements serviceaccount.ServiceAccountTokenClusterGetter. +func (s *ShardRoutedServiceAccounts) Cluster(cluster logicalcluster.Name) serviceaccount.ServiceAccountTokenGetter { + return clusterServiceAccounts{lookup: s, cluster: cluster} +} + +// SecretsWriter returns the secret client the legacy token validator writes +// through, routed the same way as the lookups. +func (s *ShardRoutedServiceAccounts) SecretsWriter() kcpcorev1client.SecretClusterInterface { + return routedSecrets{lookup: s} +} + +type clusterServiceAccounts struct { + lookup *ShardRoutedServiceAccounts + cluster logicalcluster.Name +} + +func (c clusterServiceAccounts) GetServiceAccount(ctx context.Context, namespace, name string) (*corev1.ServiceAccount, error) { + // The shard is resolved outside the cache so that an index miss is not + // cached as a failed lookup. + client, err := c.lookup.clientFor(c.cluster) + if err != nil { + return nil, err + } + key := kcpcache.ToClusterAwareKey(c.cluster.String(), namespace, name) + return c.lookup.serviceAccounts.Get(key, func() (*corev1.ServiceAccount, error) { + ctx, cancel := lookupContext(ctx) + defer cancel() + return client.Cluster(c.cluster.Path()).CoreV1().ServiceAccounts(namespace).Get(ctx, name, metav1.GetOptions{}) + }, nil) +} + +func (c clusterServiceAccounts) GetSecret(ctx context.Context, namespace, name string) (*corev1.Secret, error) { + client, err := c.lookup.clientFor(c.cluster) + if err != nil { + return nil, err + } + key := kcpcache.ToClusterAwareKey(c.cluster.String(), namespace, name) + return c.lookup.secrets.Get(key, func() (*corev1.Secret, error) { + ctx, cancel := lookupContext(ctx) + defer cancel() + return client.Cluster(c.cluster.Path()).CoreV1().Secrets(namespace).Get(ctx, name, metav1.GetOptions{}) + }, nil) +} + +// lookupContext keeps the values of the request context but not its +// cancellation: the GET is shared with every request waiting on the same +// key, and one of them going away must not fail the others. +func lookupContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), lookupTimeout) +} + +// GetPod and GetNode back bound tokens of pods and nodes, which do not exist +// in kcp. +func (c clusterServiceAccounts) GetPod(_ context.Context, namespace, name string) (*corev1.Pod, error) { + return nil, apierrors.NewNotFound(corev1.Resource("pods"), namespace+"/"+name) +} + +func (c clusterServiceAccounts) GetNode(_ context.Context, name string) (*corev1.Node, error) { + return nil, apierrors.NewNotFound(corev1.Resource("nodes"), name) +} + +// GetValidatingWebhookConfiguration and GetMutatingWebhookConfiguration back +// tokens bound to a webhook configuration, which lives in the logical cluster +// on the hosting shard like the service account itself. +func (c clusterServiceAccounts) GetValidatingWebhookConfiguration(ctx context.Context, name string) (*admissionregistrationv1.ValidatingWebhookConfiguration, error) { + client, err := c.lookup.clientFor(c.cluster) + if err != nil { + return nil, err + } + return client.Cluster(c.cluster.Path()).AdmissionregistrationV1().ValidatingWebhookConfigurations().Get(ctx, name, metav1.GetOptions{}) +} + +func (c clusterServiceAccounts) GetMutatingWebhookConfiguration(ctx context.Context, name string) (*admissionregistrationv1.MutatingWebhookConfiguration, error) { + client, err := c.lookup.clientFor(c.cluster) + if err != nil { + return nil, err + } + return client.Cluster(c.cluster.Path()).AdmissionregistrationV1().MutatingWebhookConfigurations().Get(ctx, name, metav1.GetOptions{}) +} + +// routedSecrets is a SecretClusterInterface that routes per logical cluster; +// listing or watching across clusters is not something the proxy does. +type routedSecrets struct { + lookup *ShardRoutedServiceAccounts +} + +func (r routedSecrets) Cluster(path logicalcluster.Path) kcpcorev1client.SecretsNamespacer { + cluster, ok := path.Name() + if !ok { + return erroringSecretsNamespacer{err: apierrors.NewBadRequest("secrets are addressed by logical cluster name, not path " + path.String())} + } + client, err := r.lookup.clientFor(cluster) + if err != nil { + return erroringSecretsNamespacer{err: err} + } + return clusterSecretsNamespacer{secrets: client.Cluster(cluster.Path()).CoreV1()} +} + +func (routedSecrets) List(context.Context, metav1.ListOptions) (*corev1.SecretList, error) { + return nil, apierrors.NewMethodNotSupported(corev1.Resource("secrets"), "list") +} + +func (routedSecrets) Watch(context.Context, metav1.ListOptions) (watch.Interface, error) { + return nil, apierrors.NewMethodNotSupported(corev1.Resource("secrets"), "watch") +} + +type clusterSecretsNamespacer struct { + secrets typedcorev1.SecretsGetter +} + +func (n clusterSecretsNamespacer) Namespace(namespace string) typedcorev1.SecretInterface { + return n.secrets.Secrets(namespace) +} + +// erroringSecretsNamespacer answers every call with the lookup error. +type erroringSecretsNamespacer struct { + err error +} + +func (n erroringSecretsNamespacer) Namespace(string) typedcorev1.SecretInterface { + return erroringSecrets(n) +} + +type erroringSecrets struct { + err error +} + +var _ typedcorev1.SecretInterface = erroringSecrets{} + +func (e erroringSecrets) Create(context.Context, *corev1.Secret, metav1.CreateOptions) (*corev1.Secret, error) { + return nil, e.err +} + +func (e erroringSecrets) Update(context.Context, *corev1.Secret, metav1.UpdateOptions) (*corev1.Secret, error) { + return nil, e.err +} + +func (e erroringSecrets) Delete(context.Context, string, metav1.DeleteOptions) error { + return e.err +} + +func (e erroringSecrets) DeleteCollection(context.Context, metav1.DeleteOptions, metav1.ListOptions) error { + return e.err +} + +func (e erroringSecrets) Get(context.Context, string, metav1.GetOptions) (*corev1.Secret, error) { + return nil, e.err +} + +func (e erroringSecrets) List(context.Context, metav1.ListOptions) (*corev1.SecretList, error) { + return nil, e.err +} + +func (e erroringSecrets) Watch(context.Context, metav1.ListOptions) (watch.Interface, error) { + return nil, e.err +} + +func (e erroringSecrets) Patch(context.Context, string, types.PatchType, []byte, metav1.PatchOptions, ...string) (*corev1.Secret, error) { + return nil, e.err +} + +func (e erroringSecrets) Apply(context.Context, *corev1ac.SecretApplyConfiguration, metav1.ApplyOptions) (*corev1.Secret, error) { + return nil, e.err +} diff --git a/pkg/proxy/serviceaccounts_authn_test.go b/pkg/proxy/serviceaccounts_authn_test.go new file mode 100644 index 00000000000..381dce7836c --- /dev/null +++ b/pkg/proxy/serviceaccounts_authn_test.go @@ -0,0 +1,185 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/client-go/rest" + "k8s.io/client-go/util/keyutil" + "k8s.io/kubernetes/pkg/apis/core" + "k8s.io/kubernetes/pkg/serviceaccount" + + "github.com/kcp-dev/logicalcluster/v3" + + "github.com/kcp-dev/kcp/pkg/index/indextest" + proxyoptions "github.com/kcp-dev/kcp/pkg/proxy/options" +) + +// TestServiceAccountAuthenticationThroughShards builds the proxy's +// authenticator the way NewConfig does, with service account authentication +// enabled and the shard-routed lookup behind it, and sends signed legacy and +// bound tokens of service accounts living on two different shards through +// it. It pins the wiring of ServiceAccountTokenGetter and SecretsWriter: a +// token is accepted only if its service account and secret are found on the +// shard hosting its logical cluster, and never through any other shard. +func TestServiceAccountAuthenticationThroughShards(t *testing.T) { + t.Parallel() + ctx := context.Background() + + // the signing key; the proxy only gets the public half + key, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + pem, err := keyutil.MarshalPrivateKeyToPEM(key) + require.NoError(t, err) + keyFile := filepath.Join(t.TempDir(), "sa.key") + require.NoError(t, os.WriteFile(keyFile, pem, 0600)) + + // two shards, each hosting one logical cluster with one service account + type account struct { + cluster logicalcluster.Name + shard *fakeShard + sa *corev1.ServiceAccount + secret *corev1.Secret + } + accounts := map[string]*account{ + "shard-a": {cluster: "ws1", shard: newFakeShard(t)}, + "shard-b": {cluster: "ws2", shard: newFakeShard(t)}, + } + for name, a := range accounts { + a.sa = &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{ + Name: "sa", Namespace: "default", UID: types.UID("uid-" + name), + Annotations: map[string]string{logicalcluster.AnnotationKey: a.cluster.String()}, + }} + a.secret = &corev1.Secret{ObjectMeta: metav1.ObjectMeta{ + Name: "sa-token", Namespace: "default", UID: types.UID("secret-uid-" + name), + Annotations: map[string]string{logicalcluster.AnnotationKey: a.cluster.String()}, + }, Data: map[string][]byte{}} + a.shard.add(a.sa, a.secret) + } + index := &indextest.Index{Clusters: map[logicalcluster.Name]string{}, Shards: map[string]string{}} + for name, a := range accounts { + index.Clusters[a.cluster] = name + index.Shards[name] = a.shard.URL + } + + // Uncached: every token below must reach its shard, and the legacy + // tokens are re-minted into the same secret. + lookup := newShardRoutedServiceAccounts(&rest.Config{Host: "https://kcp.io/fake", TLSClientConfig: rest.TLSClientConfig{Insecure: true}}, noCache) + lookup.SetIndex(index) + + options := proxyoptions.NewAuthentication() + options.BuiltInOptions.ServiceAccounts.KeyFiles = []string{keyFile} + var info genericapiserver.AuthenticationInfo + require.NoError(t, options.ApplyTo(ctx, &info, nil, lookup)) + require.NotNil(t, info.Authenticator) + + issuer := options.BuiltInOptions.ServiceAccounts.Issuers[0] + legacy, err := serviceaccount.JWTTokenGenerator(serviceaccount.LegacyIssuer, key) + require.NoError(t, err) + bound, err := serviceaccount.JWTTokenGenerator(issuer, key) + require.NoError(t, err) + + legacyToken := func(a *account) string { + claims, private := serviceaccount.LegacyClaims(*a.sa, *a.secret) + token, err := legacy.GenerateToken(ctx, claims, private) + require.NoError(t, err) + // the legacy validator compares the token with the secret's copy + a.shard.mu.Lock() + a.secret.Data[corev1.ServiceAccountTokenKey] = []byte(token) + a.shard.mu.Unlock() + return token + } + boundToken := func(a *account) string { + sa := core.ServiceAccount{ObjectMeta: a.sa.ObjectMeta} + secret := core.Secret{ObjectMeta: a.secret.ObjectMeta} + claims, private, err := serviceaccount.Claims(sa, nil, &secret, nil, nil, nil, 3600, 0, []string{issuer}, nil) + require.NoError(t, err) + token, err := bound.GenerateToken(ctx, claims, private) + require.NoError(t, err) + return token + } + authenticate := func(token string) (string, bool) { + req := httptest.NewRequestWithContext(ctx, http.MethodGet, "https://proxy/clusters/anything/api", http.NoBody) + req.Header.Set("Authorization", "Bearer "+token) + resp, ok, err := info.Authenticator.AuthenticateRequest(req) + if err != nil || !ok { + return "", false + } + return resp.User.GetName(), true + } + + for shardName, a := range accounts { + for kind, mint := range map[string]func(*account) string{"legacy": legacyToken, "bound": boundToken} { + t.Logf("%s token of the service account on %s authenticates through that shard", kind, shardName) + for _, other := range accounts { + other.shard.requests.Store(0) + } + name, ok := authenticate(mint(a)) + require.True(t, ok, "%s token of %s rejected", kind, shardName) + require.Equal(t, "system:serviceaccount:default:sa", name) + require.Positive(t, a.shard.requests.Load(), "the hosting shard must have been asked") + for otherName, other := range accounts { + if otherName != shardName { + require.Zero(t, other.shard.requests.Load(), "%s must not be asked about %s", otherName, shardName) + } + } + } + } + + t.Log("a token of a service account whose cluster the index does not know is rejected") + orphan := &account{cluster: "ws-unknown", shard: accounts["shard-a"].shard} + orphan.sa = accounts["shard-a"].sa.DeepCopy() + orphan.sa.Annotations[logicalcluster.AnnotationKey] = orphan.cluster.String() + orphan.secret = accounts["shard-a"].secret.DeepCopy() + _, ok := authenticate(boundToken(orphan)) + require.False(t, ok) + + t.Log("a token whose service account the hosting shard does not have is rejected") + stranger := &account{cluster: "ws1", shard: accounts["shard-a"].shard} + stranger.sa = accounts["shard-a"].sa.DeepCopy() + stranger.sa.Name = "nobody" + stranger.secret = accounts["shard-a"].secret.DeepCopy() + _, ok = authenticate(boundToken(stranger)) + require.False(t, ok) + + t.Log("a token signed by another key is rejected") + otherKey, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + forged, err := serviceaccount.JWTTokenGenerator(issuer, otherKey) + require.NoError(t, err) + a := accounts["shard-a"] + claims, private, err := serviceaccount.Claims(core.ServiceAccount{ObjectMeta: a.sa.ObjectMeta}, nil, &core.Secret{ObjectMeta: a.secret.ObjectMeta}, nil, nil, nil, 3600, 0, []string{issuer}, nil) + require.NoError(t, err) + token, err := forged.GenerateToken(ctx, claims, private) + require.NoError(t, err) + _, ok = authenticate(token) + require.False(t, ok) +} diff --git a/pkg/proxy/serviceaccounts_test.go b/pkg/proxy/serviceaccounts_test.go new file mode 100644 index 00000000000..ea5fd25ebf3 --- /dev/null +++ b/pkg/proxy/serviceaccounts_test.go @@ -0,0 +1,307 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/rest" + + "github.com/kcp-dev/logicalcluster/v3" + + "github.com/kcp-dev/kcp/pkg/index/indextest" + "github.com/kcp-dev/kcp/pkg/shardlookup" +) + +// noCache expires every cached lookup at once, for tests that count the +// requests reaching the shards. +var noCache = shardlookup.TTLOptions{SuccessTTL: time.Nanosecond, FailureTTL: time.Nanosecond} + +// fakeShard is a shard answering the lookups the authenticator makes for +// the service accounts and token secrets it holds, in any logical cluster. +type fakeShard struct { + *httptest.Server + requests atomic.Int32 + + mu sync.Mutex + accounts map[string]*corev1.ServiceAccount // by namespace/name + secrets map[string]*corev1.Secret // by namespace/name +} + +func newFakeShard(t *testing.T) *fakeShard { + t.Helper() + s := &fakeShard{ + accounts: map[string]*corev1.ServiceAccount{}, + secrets: map[string]*corev1.Secret{}, + } + s.Server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.requests.Add(1) + w.Header().Set("Content-Type", "application/json") + + // /clusters//api/v1/namespaces/// + parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), "/") + if len(parts) != 8 || parts[0] != "clusters" || parts[2] != "api" || parts[3] != "v1" || parts[4] != "namespaces" { + notFound(w) + return + } + key := parts[5] + "/" + parts[7] + + s.mu.Lock() + defer s.mu.Unlock() + switch { + case r.Method == http.MethodGet && parts[6] == "serviceaccounts": + if sa, ok := s.accounts[key]; ok { + _ = json.NewEncoder(w).Encode(sa) + return + } + case r.Method == http.MethodGet && parts[6] == "secrets": + if secret, ok := s.secrets[key]; ok { + _ = json.NewEncoder(w).Encode(secret) + return + } + case r.Method == http.MethodPatch && parts[6] == "secrets": + if secret, ok := s.secrets[key]; ok { + patched := secret.DeepCopy() + if patched.Annotations == nil { + patched.Annotations = map[string]string{} + } + patched.Annotations["patched"] = "true" + _ = json.NewEncoder(w).Encode(patched) + return + } + } + notFound(w) + })) + t.Cleanup(s.Close) + return s +} + +func notFound(w http.ResponseWriter) { + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"kind":"Status","apiVersion":"v1","status":"Failure","reason":"NotFound","code":404}`)) +} + +func (s *fakeShard) add(sa *corev1.ServiceAccount, secret *corev1.Secret) { + s.mu.Lock() + defer s.mu.Unlock() + if sa != nil { + s.accounts[sa.Namespace+"/"+sa.Name] = sa + } + if secret != nil { + s.secrets[secret.Namespace+"/"+secret.Name] = secret + } +} + +// cachedShards returns the names of the shards the lookup holds clients for. +func (s *ShardRoutedServiceAccounts) cachedShards() []string { + s.mu.Lock() + defer s.mu.Unlock() + names := make([]string, 0, len(s.clients)) + for name := range s.clients { + names = append(names, name) + } + return names +} + +// TestShardRoutedServiceAccounts walks the lookup through its states in +// order: before the index exists, with a known cluster, with an unknown one. +func TestShardRoutedServiceAccounts(t *testing.T) { + t.Parallel() + + shard := newFakeShard(t) + shard.add( + &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "sa", Namespace: "default", UID: "sa-uid"}}, + &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "token", Namespace: "default"}}, + ) + + config := &rest.Config{Host: "https://kcp.io/fake", TLSClientConfig: rest.TLSClientConfig{Insecure: true}} + lookup := NewShardRoutedServiceAccounts(config) + ctx := context.Background() + + t.Log("fails closed before the index exists") + _, err := lookup.Cluster("ws1").GetServiceAccount(ctx, "default", "sa") + require.True(t, apierrors.IsServiceUnavailable(err), "got %v", err) + + lookup.SetIndex(&indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws1": "shard-a"}, + Shards: map[string]string{"shard-a": shard.URL}, + }) + + t.Log("looks the service account and secret up on the hosting shard") + sa, err := lookup.Cluster("ws1").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.Equal(t, types.UID("sa-uid"), sa.UID) + secret, err := lookup.Cluster("ws1").GetSecret(ctx, "default", "token") + require.NoError(t, err) + require.Equal(t, "token", secret.Name) + + t.Log("writes the token secret on the hosting shard") + secret, err = lookup.SecretsWriter().Cluster(logicalcluster.NewPath("ws1")).Namespace("default").Patch(ctx, "token", types.MergePatchType, []byte(`{}`), metav1.PatchOptions{}) + require.NoError(t, err) + require.Equal(t, "true", secret.Annotations["patched"]) + require.Positive(t, shard.requests.Load(), "the shard must have been asked") + + t.Log("a cluster the index does not know is not found") + _, err = lookup.Cluster("unknown").GetServiceAccount(ctx, "default", "sa") + require.True(t, apierrors.IsNotFound(err), "got %v", err) + _, err = lookup.SecretsWriter().Cluster(logicalcluster.NewPath("unknown")).Namespace("default").Get(ctx, "token", metav1.GetOptions{}) + require.True(t, apierrors.IsNotFound(err), "got %v", err) + + t.Log("pods and nodes do not exist in kcp") + _, err = lookup.Cluster("ws1").GetPod(ctx, "default", "pod") + require.True(t, apierrors.IsNotFound(err)) + _, err = lookup.Cluster("ws1").GetNode(ctx, "node") + require.True(t, apierrors.IsNotFound(err)) + + t.Log("listing across clusters is not supported") + _, err = lookup.SecretsWriter().List(ctx, metav1.ListOptions{}) + require.True(t, apierrors.IsMethodNotSupported(err), "got %v", err) +} + +// TestShardRoutedServiceAccountsClientCache checks that the per-shard +// clients follow the shard set: one client per shard, replaced when the +// shard moves, dropped when the shard is gone. +func TestShardRoutedServiceAccountsClientCache(t *testing.T) { + t.Parallel() + + sa := &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "sa", Namespace: "default", UID: "sa-uid"}} + shardA, shardB, shardC := newFakeShard(t), newFakeShard(t), newFakeShard(t) + for _, s := range []*fakeShard{shardA, shardB, shardC} { + s.add(sa, nil) + } + + config := &rest.Config{Host: "https://kcp.io/fake", TLSClientConfig: rest.TLSClientConfig{Insecure: true}} + lookup := newShardRoutedServiceAccounts(config, noCache) + ctx := context.Background() + + lookup.SetIndex(&indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws1": "shard-a", "ws2": "shard-b"}, + Shards: map[string]string{"shard-a": shardA.URL, "shard-b": shardB.URL}, + }) + + t.Log("one client per shard, reused across lookups") + for range 3 { + _, err := lookup.Cluster("ws1").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + } + _, err := lookup.Cluster("ws2").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.ElementsMatch(t, []string{"shard-a", "shard-b"}, lookup.cachedShards()) + require.EqualValues(t, 3, shardA.requests.Load()) + require.EqualValues(t, 1, shardB.requests.Load()) + + t.Log("a shard the index forgot loses its client once another shard is looked up afresh") + lookup.SetIndex(&indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws2": "shard-b", "ws3": "shard-c"}, + Shards: map[string]string{"shard-b": shardB.URL, "shard-c": shardC.URL}, + }) + _, err = lookup.Cluster("ws2").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.ElementsMatch(t, []string{"shard-a", "shard-b"}, lookup.cachedShards(), "a cache hit does not prune") + _, err = lookup.Cluster("ws3").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.ElementsMatch(t, []string{"shard-b", "shard-c"}, lookup.cachedShards(), "the client of the forgotten shard is dropped") + require.EqualValues(t, 1, shardC.requests.Load()) + + t.Log("a shard that moved to a new URL gets a new client") + lookup.SetIndex(&indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws2": "shard-b", "ws3": "shard-c"}, + Shards: map[string]string{"shard-b": shardA.URL, "shard-c": shardC.URL}, + }) + before := shardA.requests.Load() + _, err = lookup.Cluster("ws2").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.ElementsMatch(t, []string{"shard-b", "shard-c"}, lookup.cachedShards()) + require.Equal(t, before+1, shardA.requests.Load(), "the request went to the shard's new URL") + require.EqualValues(t, 2, shardB.requests.Load(), "the old URL is not asked any more") +} + +// TestShardRoutedServiceAccountsResultCache checks that service accounts and +// secrets are cached per logical cluster, failed lookups included, and that +// index misses are not. +func TestShardRoutedServiceAccountsResultCache(t *testing.T) { + t.Parallel() + + shard := newFakeShard(t) + shard.add( + &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "sa", Namespace: "default", UID: "sa-uid"}}, + &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: "token", Namespace: "default"}}, + ) + + config := &rest.Config{Host: "https://kcp.io/fake", TLSClientConfig: rest.TLSClientConfig{Insecure: true}} + lookup := newShardRoutedServiceAccounts(config, shardlookup.TTLOptions{SuccessTTL: time.Hour, FailureTTL: time.Hour}) + ctx := context.Background() + + index := &indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws1": "shard-a", "ws2": "shard-a"}, + Shards: map[string]string{"shard-a": shard.URL}, + } + lookup.SetIndex(index) + + t.Log("repeated lookups of a service account and a secret ask the shard once each") + for range 3 { + _, err := lookup.Cluster("ws1").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + _, err = lookup.Cluster("ws1").GetSecret(ctx, "default", "token") + require.NoError(t, err) + } + require.EqualValues(t, 2, shard.requests.Load()) + + t.Log("the same name in another logical cluster is another entry") + _, err := lookup.Cluster("ws2").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.EqualValues(t, 3, shard.requests.Load()) + + t.Log("a service account the shard does not have is cached as not found") + for range 2 { + _, err = lookup.Cluster("ws1").GetServiceAccount(ctx, "default", "missing") + require.True(t, apierrors.IsNotFound(err), "got %v", err) + } + require.EqualValues(t, 4, shard.requests.Load()) + + t.Log("an index miss is not cached: the cluster is found once the index knows it") + _, err = lookup.Cluster("ws3").GetServiceAccount(ctx, "default", "sa") + require.True(t, apierrors.IsNotFound(err), "got %v", err) + lookup.SetIndex(&indextest.Index{ + Clusters: map[logicalcluster.Name]string{"ws3": "shard-a"}, + Shards: map[string]string{"shard-a": shard.URL}, + }) + _, err = lookup.Cluster("ws3").GetServiceAccount(ctx, "default", "sa") + require.NoError(t, err) + require.EqualValues(t, 5, shard.requests.Load()) + + t.Log("a cancelled request context does not fail the shared lookup") + cancelled, cancel := context.WithCancel(ctx) + cancel() + _, err = lookup.Cluster("ws3").GetSecret(cancelled, "default", "token") + require.NoError(t, err) + require.EqualValues(t, 6, shard.requests.Load()) +} diff --git a/pkg/server/proxy/lookup_test.go b/pkg/server/proxy/lookup_test.go index 637a8ca4d74..4c4c458ed2b 100644 --- a/pkg/server/proxy/lookup_test.go +++ b/pkg/server/proxy/lookup_test.go @@ -17,7 +17,6 @@ limitations under the License. package proxy import ( - "context" "net/http" "net/http/httptest" "testing" @@ -26,46 +25,10 @@ import ( "github.com/kcp-dev/logicalcluster/v3" - "github.com/kcp-dev/kcp/pkg/index" + "github.com/kcp-dev/kcp/pkg/index/indextest" apiexportbuilder "github.com/kcp-dev/kcp/pkg/virtual/apiexport/builder" ) -// fakeIndex is a test double for index.Index. -type fakeIndex struct { - result index.Result - found bool - baseURL string - recentlyMigrated bool - - registered int - released int -} - -func (f *fakeIndex) BaseURL(shardName string) (string, bool) { - return f.baseURL, f.baseURL != "" -} - -func (f *fakeIndex) Lookup(path logicalcluster.Path) (index.Result, bool) { - return f.result, f.found -} - -func (f *fakeIndex) LookupURL(path logicalcluster.Path) (index.Result, bool) { - return f.result, f.found -} - -func (f *fakeIndex) RecentlyMigrated(cluster logicalcluster.Name) bool { - return f.recentlyMigrated -} - -func (f *fakeIndex) ClusterContext(parent context.Context, cluster logicalcluster.Name) (context.Context, context.CancelFunc) { - f.registered++ - ctx, cancel := context.WithCancel(parent) - return ctx, func() { - f.released++ - cancel() - } -} - func TestClusterResolveHandlerMigration(t *testing.T) { t.Parallel() @@ -144,14 +107,10 @@ func TestClusterResolveHandlerMigration(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - fake := &fakeIndex{ - found: true, - recentlyMigrated: tt.recentlyMigrated, - result: index.Result{ - Cluster: logicalcluster.Name(cluster), - Shard: "shard-1", - URL: "https://shard-1.example/clusters/" + cluster, - }, + fake := &indextest.Index{ + Clusters: map[logicalcluster.Name]string{cluster: "shard-1"}, + Shards: map[string]string{"shard-1": "https://shard-1.example"}, + RecentlyMigratedClusters: map[logicalcluster.Name]bool{cluster: tt.recentlyMigrated}, } delegateCalled := false @@ -185,12 +144,13 @@ func TestClusterResolveHandlerMigration(t *testing.T) { if delegateCalled != tt.wantDelegate { t.Errorf("delegate called = %v, want %v", delegateCalled, tt.wantDelegate) } - if (fake.registered > 0) != tt.wantRegistered { - t.Errorf("registered = %d, want registered=%v", fake.registered, tt.wantRegistered) + registered, released := fake.Registered.Load(), fake.Released.Load() + if (registered > 0) != tt.wantRegistered { + t.Errorf("registered = %d, want registered=%v", registered, tt.wantRegistered) } // Any registered in-flight watch must be released when the request finishes. - if fake.registered != fake.released { - t.Errorf("registered (%d) != released (%d): in-flight watch not released", fake.registered, fake.released) + if registered != released { + t.Errorf("registered (%d) != released (%d): in-flight watch not released", registered, released) } }) } diff --git a/pkg/virtual/admin/authorizer/authorizer_test.go b/pkg/virtual/admin/authorizer/authorizer_test.go index 00ac95f562b..5173ccb6625 100644 --- a/pkg/virtual/admin/authorizer/authorizer_test.go +++ b/pkg/virtual/admin/authorizer/authorizer_test.go @@ -49,7 +49,7 @@ func TestAdminAuthorizer(t *testing.T) { }, { // The front-proxy discovers shards through this view with its - // --root-kubeconfig client certificate, which carries + // --shard-peer-kubeconfig client certificate, which carries // system:masters. Denying it here would break shard discovery, so // this case is the regression guard. name: "system:masters may read", diff --git a/test/e2e/logicalclustermigration/migration_server_test.go b/test/e2e/logicalclustermigration/migration_server_test.go index fd8edfe7b88..020b81d39a9 100644 --- a/test/e2e/logicalclustermigration/migration_server_test.go +++ b/test/e2e/logicalclustermigration/migration_server_test.go @@ -161,7 +161,7 @@ func privateMigrationServerWithRestart(t *testing.T, shardArgs map[string][]stri command = append(command, "--bind-address=127.0.0.1", "--secure-port="+proxyPort, "--root-directory="+filepath.Join(dataDir, "proxy"), - "--root-kubeconfig="+rootConfig, "--shard-peer-kubeconfig="+rootConfig, + "--shard-peer-kubeconfig="+rootConfig, "--shards-kubeconfig="+rootConfig, "--mapping-file="+mappingFile, "--client-ca-file="+clientCAFile, "--tls-cert-file="+certFile, "--tls-private-key-file="+keyFile, diff --git a/test/e2e/proxy/serviceaccounts_test.go b/test/e2e/proxy/serviceaccounts_test.go new file mode 100644 index 00000000000..6e8a08fbec0 --- /dev/null +++ b/test/e2e/proxy/serviceaccounts_test.go @@ -0,0 +1,164 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + + authenticationv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/utils/ptr" + + kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" + "github.com/kcp-dev/sdk/apis/core" + kcptesting "github.com/kcp-dev/sdk/testing" + kcptestinghelpers "github.com/kcp-dev/sdk/testing/helpers" + + "github.com/kcp-dev/kcp/test/e2e/framework" +) + +// TestServiceAccountsThroughFrontProxy checks that the front-proxy itself +// authenticates service account tokens of workspaces on every shard. The +// sharded-test-server gives the proxy the service account signing key, so a +// token is validated by the proxy, which looks the service account and its +// token secret up on the shard hosting the workspace through its workspace +// index, without a connection to the root shard. For a workspace pinned to +// each shard, a legacy and a bound token are sent through the proxy, which +// must report the service account identity and let it act with its RBAC; +// once the token secret is gone, both tokens must be rejected. +func TestServiceAccountsThroughFrontProxy(t *testing.T) { + t.Parallel() + framework.Suite(t, "control-plane") + + server := kcptesting.SharedKcpServer(t) + if len(server.ShardNames()) < 2 { + t.Skip("This test requires a multi-shard setup with a front-proxy.") + } + + // every request below goes through the front-proxy + cfg := server.BaseConfig(t) + kubeClusterClient, err := kcpkubernetesclientset.NewForConfig(cfg) + require.NoError(t, err) + + orgPath, _ := kcptesting.NewWorkspaceFixture(t, server, core.RootCluster.Path(), kcptesting.WithType(core.RootCluster.Path(), "organization")) + + for _, shard := range server.ShardNames() { + t.Run(shard, func(t *testing.T) { + t.Parallel() + ctx := t.Context() + + wsPath, _ := kcptesting.NewWorkspaceFixture(t, server, orgPath, kcptesting.WithShard(shard)) + + t.Log("Creating namespace, role and role binding for the default service account") + namespace, err := kubeClusterClient.Cluster(wsPath).CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "e2e-proxy-sa-"}, + }, metav1.CreateOptions{}) + require.NoError(t, err, "failed to create namespace") + _, err = kubeClusterClient.Cluster(wsPath).RbacV1().Roles(namespace.Name).Create(ctx, &rbacv1.Role{ + ObjectMeta: metav1.ObjectMeta{Name: "configmaps"}, + Rules: []rbacv1.PolicyRule{{Verbs: []string{"list"}, APIGroups: []string{""}, Resources: []string{"configmaps"}}}, + }, metav1.CreateOptions{}) + require.NoError(t, err, "failed to create role") + _, err = kubeClusterClient.Cluster(wsPath).RbacV1().RoleBindings(namespace.Name).Create(ctx, &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{Name: "configmaps"}, + Subjects: []rbacv1.Subject{{Kind: "ServiceAccount", Name: "default", Namespace: namespace.Name}}, + RoleRef: rbacv1.RoleRef{Kind: "Role", Name: "configmaps", APIGroup: rbacv1.GroupName}, + }, metav1.CreateOptions{}) + require.NoError(t, err, "failed to create role binding") + + t.Log("Waiting for the default service account") + kcptestinghelpers.Eventually(t, func() (bool, string) { + _, err := kubeClusterClient.Cluster(wsPath).CoreV1().ServiceAccounts(namespace.Name).Get(ctx, "default", metav1.GetOptions{}) + return err == nil, fmt.Sprintf("%v", err) + }, wait.ForeverTestTimeout, 100*time.Millisecond, "the default service account was not created") + + t.Log("Creating the legacy token secret and waiting for it to be filled") + _, err = kubeClusterClient.Cluster(wsPath).CoreV1().Secrets(namespace.Name).Create(ctx, &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "default-token", Annotations: map[string]string{corev1.ServiceAccountNameKey: "default"}}, + Type: corev1.SecretTypeServiceAccountToken, + }, metav1.CreateOptions{}) + require.NoError(t, err, "failed to create the token secret") + var tokenSecret *corev1.Secret + kcptestinghelpers.Eventually(t, func() (bool, string) { + s, err := kubeClusterClient.Cluster(wsPath).CoreV1().Secrets(namespace.Name).Get(ctx, "default-token", metav1.GetOptions{}) + if err != nil { + return false, err.Error() + } + tokenSecret = s + return len(s.Data[corev1.ServiceAccountTokenKey]) > 0, "token not filled yet" + }, wait.ForeverTestTimeout, 100*time.Millisecond, "the token secret was not filled") + + t.Log("Requesting a bound token") + boundToken, err := kubeClusterClient.Cluster(wsPath).CoreV1().ServiceAccounts(namespace.Name).CreateToken(ctx, "default", &authenticationv1.TokenRequest{ + Spec: authenticationv1.TokenRequestSpec{ + Audiences: []string{"https://kcp.default.svc"}, + ExpirationSeconds: ptr.To[int64](3600), + BoundObjectRef: &authenticationv1.BoundObjectReference{APIVersion: "v1", Kind: "Secret", Name: tokenSecret.Name, UID: tokenSecret.UID}, + }, + }, metav1.CreateOptions{}) + require.NoError(t, err, "failed to create a bound token") + + tokens := map[string]string{ + "legacy": string(tokenSecret.Data[corev1.ServiceAccountTokenKey]), + "bound": boundToken.Status.Token, + } + expectedUser := "system:serviceaccount:" + namespace.Name + ":default" + clients := map[string]kcpkubernetesclientset.ClusterInterface{} + for kind, token := range tokens { + client, err := kcpkubernetesclientset.NewForConfig(framework.ConfigWithToken(token, cfg)) + require.NoError(t, err) + clients[kind] = client + } + + for kind, client := range clients { + t.Logf("The %s token authenticates through the front-proxy as the service account", kind) + kcptestinghelpers.Eventually(t, func() (bool, string) { + review, err := client.Cluster(wsPath).AuthenticationV1().SelfSubjectReviews().Create(ctx, &authenticationv1.SelfSubjectReview{}, metav1.CreateOptions{}) + if err != nil { + return false, err.Error() + } + return review.Status.UserInfo.Username == expectedUser, fmt.Sprintf("authenticated as %q", review.Status.UserInfo.Username) + }, wait.ForeverTestTimeout, 100*time.Millisecond, "the %s token did not authenticate as %s", kind, expectedUser) + + t.Logf("The %s token acts with the service account's RBAC", kind) + kcptestinghelpers.Eventually(t, func() (bool, string) { + _, err := client.Cluster(wsPath).CoreV1().ConfigMaps(namespace.Name).List(ctx, metav1.ListOptions{}) + return err == nil, fmt.Sprintf("%v", err) + }, wait.ForeverTestTimeout, 100*time.Millisecond, "the %s token could not list configmaps", kind) + _, err := client.Cluster(wsPath).CoreV1().Secrets(namespace.Name).List(ctx, metav1.ListOptions{}) + require.True(t, apierrors.IsForbidden(err), "the %s token must not list secrets, got: %v", kind, err) + } + + t.Log("Deleting the token secret invalidates both tokens") + require.NoError(t, kubeClusterClient.Cluster(wsPath).CoreV1().Secrets(namespace.Name).Delete(ctx, tokenSecret.Name, metav1.DeleteOptions{})) + for kind, client := range clients { + kcptestinghelpers.Eventually(t, func() (bool, string) { + _, err := client.Cluster(wsPath).CoreV1().ConfigMaps(namespace.Name).List(ctx, metav1.ListOptions{}) + return apierrors.IsUnauthorized(err), fmt.Sprintf("%v", err) + }, wait.ForeverTestTimeout, 100*time.Millisecond, "the %s token was still accepted after its secret was deleted", kind) + } + }) + } +}