WIP: 🐛 authentication: add custom SA lookup with ttl cache for non-local clusters

cmd/kcp/kcp.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/kcp-dev/client-go/kubernetes"
 	"github.com/spf13/cobra"
 
 	"k8s.io/apimachinery/pkg/util/errors"
@@ -40,6 +41,7 @@ import (
 	"github.com/kcp-dev/kcp/pkg/embeddedetcd"
 	kcpfeatures "github.com/kcp-dev/kcp/pkg/features"
 	"github.com/kcp-dev/kcp/pkg/server"
+	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
 	"github.com/kcp-dev/kcp/sdk/cmd/help"
 )
 
@@ -85,7 +87,13 @@ func main() {
 		}
 	}
 
-	kcpOptions := options.NewOptions(rootDir)
+	// these are late initialized on option->config. Hence, we pass the pointers here.
+	var (
+		delayedKcpInformers      kcpinformers.SharedInformerFactory
+		delayedClusterKubeClient kubernetes.ClusterInterface
+	)
+
+	kcpOptions := options.NewOptions(rootDir, &delayedClusterKubeClient, &delayedKcpInformers)
 	kcpOptions.Server.GenericControlPlane.Logs.Verbosity = logsapiv1.VerbosityLevel(2)
 	kcpOptions.Server.Extra.AdditionalMappingsFile = additionalMappingsFile
 
@@ -132,6 +140,10 @@ func main() {
 				return err
 			}
 
+			// set the delayed client and informers, used in the service account lookup
+			delayedKcpInformers = serverConfig.KcpSharedInformerFactory
+			delayedClusterKubeClient = serverConfig.KubeClusterClient
+
 			completedConfig, err := serverConfig.Complete()
 			if err != nil {
 				return err

cmd/kcp/options/options.go

@@ -19,9 +19,12 @@ package options
 import (
 	"io"
 
+	"github.com/kcp-dev/client-go/kubernetes"
+
 	cliflag "k8s.io/component-base/cli/flag"
 
 	serveroptions "github.com/kcp-dev/kcp/pkg/server/options"
+	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
 )
 
 type Options struct {
@@ -34,11 +37,11 @@ type Options struct {
 
 type ExtraOptions struct{}
 
-func NewOptions(rootDir string) *Options {
+func NewOptions(rootDir string, delayedClusterKubeClient *kubernetes.ClusterInterface, delayedKcpInformers *kcpinformers.SharedInformerFactory) *Options {
 	opts := &Options{
 		Output: nil,
 
-		Server:  *serveroptions.NewOptions(rootDir),
+		Server:  *serveroptions.NewOptions(rootDir, delayedClusterKubeClient, delayedKcpInformers),
 		Generic: *NewGeneric(rootDir),
 		Extra:   ExtraOptions{},
 	}

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/go-logr/logr v1.4.2
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
+	github.com/jellydator/ttlcache/v3 v3.3.0
 	github.com/kcp-dev/apimachinery/v2 v2.0.1-0.20240817110845-a9eb9752bfeb
 	github.com/kcp-dev/client-go v0.0.0-20240912145314-f5949d81732a
 	github.com/kcp-dev/kcp/sdk v0.0.0-00010101000000-000000000000

go.sum

@@ -138,6 +138,8 @@ github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jellydator/ttlcache/v3 v3.3.0 h1:BdoC9cE81qXfrxeb9eoJi9dWrdhSuwXMAnHTbnBm4Wc=
+github.com/jellydator/ttlcache/v3 v3.3.0/go.mod h1:bj2/e0l4jRnQdrnSTaGTsh4GSXvMjQcy41i7th0GVGw=
 github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
 github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

pkg/server/options/options.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/kcp-dev/client-go/kubernetes"
 	"github.com/spf13/pflag"
 
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -34,6 +35,7 @@ import (
 	etcdoptions "github.com/kcp-dev/kcp/pkg/embeddedetcd/options"
 	kcpfeatures "github.com/kcp-dev/kcp/pkg/features"
 	"github.com/kcp-dev/kcp/pkg/server/options/batteries"
+	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
 )
 
 type Options struct {
@@ -91,7 +93,7 @@ type CompletedOptions struct {
 }
 
 // NewOptions creates a new Options with default parameters.
-func NewOptions(rootDir string) *Options {
+func NewOptions(rootDir string, delayedKubeClusterClient *kubernetes.ClusterInterface, delayedKcpInformers *kcpinformers.SharedInformerFactory) *Options {
 	o := &Options{
 		GenericControlPlane: *controlplaneapiserver.NewOptions(),
 		EmbeddedEtcd:        *etcdoptions.NewOptions(rootDir),
@@ -119,6 +121,7 @@ func NewOptions(rootDir string) *Options {
 	// override all the stuff
 	o.GenericControlPlane.SecureServing.ServerCert.CertDirectory = rootDir
 	o.GenericControlPlane.Authentication.ServiceAccounts.Issuers = []string{"https://kcp.default.svc"}
+	o.GenericControlPlane.Authentication.ServiceAccounts.OptionalTokenGetter = newServiceAccountTokenCache(delayedKubeClusterClient, delayedKcpInformers)
 	o.GenericControlPlane.Etcd.StorageConfig.Transport.ServerList = []string{"embedded"}
 	o.GenericControlPlane.Authorization = nil // we have our own
 

pkg/server/options/serviceaccounts.go

@@ -0,0 +1,165 @@
+/*
+Copyright 2025 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 options
+
+import (
+	"context"
+	"time"
+
+	"github.com/jellydator/ttlcache/v3"
+	kcpkubernetesinformers "github.com/kcp-dev/client-go/informers"
+	"github.com/kcp-dev/client-go/kubernetes"
+	"github.com/kcp-dev/logicalcluster/v3"
+
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	clientset "k8s.io/client-go/kubernetes"
+	kubecorev1lister "k8s.io/client-go/listers/core/v1"
+	"k8s.io/kubernetes/pkg/serviceaccount"
+
+	corev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/core/v1alpha1"
+	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
+	corev1alpha1listers "github.com/kcp-dev/kcp/sdk/client/listers/core/v1alpha1"
+)
+
+const (
+	// SuccessCacheTTL is the TTL to cache a successful lookup for remote clusters.
+	SuccessCacheTTL = 1 * time.Minute
+	// FailureCacheTTL is the TTL to cache a failed lookup for remote clusters.
+	FailureCacheTTL = 10 * time.Second
+)
+
+type cacheKey struct {
+	clusterName logicalcluster.Name
+	types.NamespacedName
+}
+
+// newServiceAccountTokenCache creates a new service account token cache backed
+// by ttl caches for all remote clusters, and local informers logic for local
+// clusters.
+func newServiceAccountTokenCache(delayedKubeClusterClient *kubernetes.ClusterInterface, delayedKcpInformers *kcpinformers.SharedInformerFactory) func(kubeInformers kcpkubernetesinformers.SharedInformerFactory) serviceaccount.ServiceAccountTokenClusterGetter {
+	return func(kubeInformers kcpkubernetesinformers.SharedInformerFactory) serviceaccount.ServiceAccountTokenClusterGetter {
+		return &serviceAccountTokenCache{
+			delayedKubeClusterClient: delayedKubeClusterClient,
+
+			delayedKcpInformers: delayedKcpInformers,
+			kubeInformers:       kubeInformers,
+
+			serviceAccountCache: ttlcache.New[cacheKey, *corev1.ServiceAccount](),
+			secretCache:         ttlcache.New[cacheKey, *corev1.Secret](),
+		}
+	}
+}
+
+type serviceAccountTokenCache struct {
+	delayedKubeClusterClient *kubernetes.ClusterInterface
+
+	delayedKcpInformers *kcpinformers.SharedInformerFactory
+	kubeInformers       kcpkubernetesinformers.SharedInformerFactory
+
+	serviceAccountCache *ttlcache.Cache[cacheKey, *corev1.ServiceAccount]
+	secretCache         *ttlcache.Cache[cacheKey, *corev1.Secret]
+}
+
+func (c *serviceAccountTokenCache) Cluster(clusterName logicalcluster.Name) serviceaccount.ServiceAccountTokenGetter {
+	return &serviceAccountTokenGetter{
+		kubeClient: (*c.delayedKubeClusterClient).Cluster(clusterName.Path()),
+
+		logicalClusters: (*c.delayedKcpInformers).Core().V1alpha1().LogicalClusters().Cluster(clusterName).Lister(),
+		serviceAccounts: c.kubeInformers.Core().V1().ServiceAccounts().Cluster(clusterName).Lister(),
+		secrets:         c.kubeInformers.Core().V1().Secrets().Cluster(clusterName).Lister(),
+
+		serviceAccountCache: c.serviceAccountCache,
+		secretCache:         c.secretCache,
+
+		clusterName: clusterName,
+	}
+}
+
+type serviceAccountTokenGetter struct {
+	kubeClient clientset.Interface
+
+	logicalClusters corev1alpha1listers.LogicalClusterLister
+	serviceAccounts kubecorev1lister.ServiceAccountLister
+	secrets         kubecorev1lister.SecretLister
+
+	serviceAccountCache *ttlcache.Cache[cacheKey, *corev1.ServiceAccount]
+	secretCache         *ttlcache.Cache[cacheKey, *corev1.Secret]
+
+	clusterName logicalcluster.Name
+}
+
+func (g *serviceAccountTokenGetter) GetServiceAccount(namespace, name string) (*corev1.ServiceAccount, error) {
+	// local cluster?
+	if _, err := g.logicalClusters.Get(corev1alpha1.LogicalClusterName); err != nil {
+		return g.serviceAccounts.ServiceAccounts(namespace).Get(name)
+	}
+
+	// cached?
+	if sa := g.serviceAccountCache.Get(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}); sa != nil && sa.Value() != nil {
+		return sa.Value(), nil
+	}
+
+	// fetch with external client
+	// TODO(sttts): here it's little racy, as we might fetch the service account multiple times.
+	sa, err := g.kubeClient.CoreV1().ServiceAccounts(namespace).Get(context.Background(), name, metav1.GetOptions{})
+	if err != nil && !kerrors.IsNotFound(err) {
+		return nil, err
+	} else if kerrors.IsNotFound(err) {
+		ttl := ttlcache.WithTTL[cacheKey, *corev1.ServiceAccount](FailureCacheTTL)
+		g.serviceAccountCache.GetOrSet(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}, nil, ttl)
+	}
+
+	g.serviceAccountCache.Set(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}, sa, SuccessCacheTTL)
+	return sa, nil
+}
+
+func (g *serviceAccountTokenGetter) GetPod(_, name string) (*corev1.Pod, error) {
+	return nil, kerrors.NewNotFound(schema.GroupResource{Group: "", Resource: "pods"}, name)
+}
+
+func (g *serviceAccountTokenGetter) GetSecret(namespace, name string) (*corev1.Secret, error) {
+	// local cluster?
+	if _, err := g.logicalClusters.Get(corev1alpha1.LogicalClusterName); err != nil {
+		return g.secrets.Secrets(namespace).Get(name)
+	}
+
+	// cached?
+	if secret := g.secretCache.Get(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}); secret != nil && secret.Value() != nil {
+		return secret.Value(), nil
+	}
+
+	// fetch with external client
+	// TODO(sttts): here it's little racy, as we might fetch the secret multiple times.
+	secret, err := g.kubeClient.CoreV1().Secrets(namespace).Get(context.Background(), name, metav1.GetOptions{})
+	if err != nil && !kerrors.IsNotFound(err) {
+		return nil, err
+	} else if kerrors.IsNotFound(err) {
+		ttl := ttlcache.WithTTL[cacheKey, *corev1.Secret](FailureCacheTTL)
+		g.secretCache.GetOrSet(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}, nil, ttl)
+	}
+
+	g.secretCache.Set(cacheKey{g.clusterName, types.NamespacedName{Namespace: namespace, Name: name}}, secret, SuccessCacheTTL)
+	return secret, nil
+}
+
+func (g *serviceAccountTokenGetter) GetNode(name string) (*corev1.Node, error) {
+	return nil, kerrors.NewNotFound(schema.GroupResource{Group: "", Resource: "nodes"}, name)
+}

test/e2e/authorizer/serviceaccounts_test.go

@@ -183,7 +183,7 @@ func TestServiceAccounts(t *testing.T) {
 			})
 
 			t.Run("Access another workspace in the same org", func(t *testing.T) {
-				t.Log("Create workspace with the same name ")
+				t.Log("Create workspace with the same name")
 				otherPath, _ := framework.NewWorkspaceFixture(t, server, orgPath)
 				_, err := kubeClusterClient.Cluster(otherPath).CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
 					ObjectMeta: metav1.ObjectMeta{
@@ -236,10 +236,7 @@ func TestServiceAccounts(t *testing.T) {
 				t.Log("Accessing other workspace with the (there foreign) service account should eventually work because it is authenticated")
 				framework.Eventually(t, func() (bool, string) {
 					_, err := saKubeClusterClient.Cluster(otherPath).CoreV1().ConfigMaps(namespace.Name).List(ctx, metav1.ListOptions{})
-					if err != nil {
-						return false, err.Error()
-					}
-					return true, ""
+					return err == nil, fmt.Sprintf("err = %v", err)
 				}, wait.ForeverTestTimeout, time.Millisecond*100)
 
 				t.Log("Taking away the authenticated access to the other workspace, restricting to only service accounts")