Merge pull request #3967 from ntnn/inter-shard-0.30

Inter-shard

Author: kcp-ci-bot

cmd/sharded-test-server/cache.go

@@ -30,17 +30,19 @@ import (
 	"github.com/abiosoft/lineprefix"
 	"github.com/fatih/color"
 
+	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/klog/v2"
 
 	kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster"
 	kcptestingserver "github.com/kcp-dev/sdk/testing/server"
+	"github.com/kcp-dev/sdk/testing/third_party/library-go/crypto"
 
 	"github.com/kcp-dev/kcp/cmd/test-server/helpers"
 )
 
-func startCacheServer(ctx context.Context, logDirPath, workingDir, hostIP string, syntheticDelay time.Duration) (<-chan error, string, error) {
+func startCacheServer(ctx context.Context, logDirPath, workingDir, hostIP string, syntheticDelay time.Duration, clientCA *crypto.CA, clientCAPath string) (<-chan error, string, error) {
 	cyan := color.New(color.BgHiCyan, color.FgHiWhite).SprintFunc()
 	inverse := color.New(color.BgHiWhite, color.FgHiCyan).SprintFunc()
 	out := lineprefix.New(
@@ -52,6 +54,32 @@ func startCacheServer(ctx context.Context, logDirPath, workingDir, hostIP string
 		lineprefix.Color(color.New(color.FgHiWhite)),
 	)
 	cacheWorkingDir := filepath.Join(workingDir, ".kcp-cache")
+
+	// Generate a client certificate for accessing the cache server.
+	cacheClientCert := filepath.Join(cacheWorkingDir, "cache-client.crt")
+	cacheClientKey := filepath.Join(cacheWorkingDir, "cache-client.key")
+	if err := os.MkdirAll(cacheWorkingDir, 0755); err != nil {
+		return nil, "", err
+	}
+	_, err := clientCA.MakeClientCertificate(cacheClientCert, cacheClientKey,
+		&user.DefaultInfo{Name: "cache-client", Groups: []string{"system:masters"}}, 365)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to create cache client cert: %w", err)
+	}
+
+	// Use absolute paths for the client cert/key so they resolve correctly
+	// regardless of how the kubeconfig is loaded. ClientConfigLoadingRules
+	// resolves relative paths relative to the kubeconfig file, while
+	// LoadFromFile + NewNonInteractiveClientConfig uses them as-is from CWD.
+	absCacheClientCert, err := filepath.Abs(cacheClientCert)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to resolve absolute path for cache client cert: %w", err)
+	}
+	absCacheClientKey, err := filepath.Abs(cacheClientKey)
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to resolve absolute path for cache client key: %w", err)
+	}
+
 	cachePort := 8012
 	workdir, commandLine := kcptestingserver.Command("cache-server", "cache")
 	commandLine = append(
@@ -62,6 +90,7 @@ func startCacheServer(ctx context.Context, logDirPath, workingDir, hostIP string
 		"--embedded-etcd-peer-port=8011",
 		fmt.Sprintf("--secure-port=%d", cachePort),
 		fmt.Sprintf("--synthetic-delay=%s", syntheticDelay.String()),
+		fmt.Sprintf("--client-ca-file=%s", clientCAPath),
 	)
 	fmt.Fprintf(out, "running: %v\n", strings.Join(commandLine, " "))
 	cmd := exec.CommandContext(ctx, commandLine[0], commandLine[1:]...) //nolint:gosec
@@ -119,35 +148,40 @@ func startCacheServer(ctx context.Context, logDirPath, workingDir, hostIP string
 			continue
 		}
 
-		if _, err := os.Stat(cacheKubeconfigPath); os.IsNotExist(err) {
-			cacheServerCert, err := os.ReadFile(filepath.Join(cacheWorkingDir, "apiserver.crt"))
-			if err != nil {
-				return nil, "", err
-			}
-			cacheServerKubeConfig := clientcmdapi.Config{
-				Clusters: map[string]*clientcmdapi.Cluster{
-					"cache": {
-						Server:                   fmt.Sprintf("https://localhost:%d", cachePort),
-						CertificateAuthorityData: cacheServerCert,
-					},
+		cacheServerCert, err := os.ReadFile(filepath.Join(cacheWorkingDir, "apiserver.crt"))
+		if err != nil {
+			return nil, "", err
+		}
+		cacheServerKubeConfig := clientcmdapi.Config{
+			Clusters: map[string]*clientcmdapi.Cluster{
+				"cache": {
+					Server:                   fmt.Sprintf("https://localhost:%d", cachePort),
+					CertificateAuthorityData: cacheServerCert,
 				},
-				Contexts: map[string]*clientcmdapi.Context{
-					"cache": {
-						Cluster: "cache",
-					},
+			},
+			AuthInfos: map[string]*clientcmdapi.AuthInfo{
+				"cache": {
+					ClientCertificate: absCacheClientCert,
+					ClientKey:         absCacheClientKey,
 				},
-				CurrentContext: "cache",
-			}
-			if err := clientcmd.WriteToFile(cacheServerKubeConfig, cacheKubeconfigPath); err != nil {
-				return nil, "", err
-			}
+			},
+			Contexts: map[string]*clientcmdapi.Context{
+				"cache": {
+					Cluster:  "cache",
+					AuthInfo: "cache",
+				},
+			},
+			CurrentContext: "cache",
+		}
+		if err := clientcmd.WriteToFile(cacheServerKubeConfig, cacheKubeconfigPath); err != nil {
+			return nil, "", err
 		}
 
-		cacheServerKubeConfig, err := clientcmd.LoadFromFile(cacheKubeconfigPath)
+		loadedKubeConfig, err := clientcmd.LoadFromFile(cacheKubeconfigPath)
 		if err != nil {
 			return nil, "", err
 		}
-		cacheClientConfig := clientcmd.NewNonInteractiveClientConfig(*cacheServerKubeConfig, "cache", nil, nil)
+		cacheClientConfig := clientcmd.NewNonInteractiveClientConfig(*loadedKubeConfig, "cache", nil, nil)
 		cacheClientRestConfig, err := cacheClientConfig.ClientConfig()
 		if err != nil {
 			return nil, "", err

cmd/sharded-test-server/main.go

@@ -197,12 +197,10 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb
 	standaloneVW := sets.New[string](shardFlags...).Has("--run-virtual-workspaces=false")
 
 	cacheServerErrCh := make(chan indexErrTuple)
-	cacheServerConfigPath := ""
-	cacheServerCh, configPath, err := startCacheServer(ctx, logDirPath, workDirPath, hostIP.String(), cacheSyntheticDelay)
+	cacheServerCh, cacheServerConfigPath, err := startCacheServer(ctx, logDirPath, workDirPath, hostIP.String(), cacheSyntheticDelay, clientCA, filepath.Join(workDirPath, ".kcp", "client-ca.crt"))
 	if err != nil {
 		return fmt.Errorf("error starting the cache server: %w", err)
 	}
-	cacheServerConfigPath = configPath
 	go func() {
 		err := <-cacheServerCh
 		cacheServerErrCh <- indexErrTuple{0, err}

docs/content/concepts/sharding/cache-server.md

@@ -69,7 +69,15 @@ ctx = cacheclient.WithShardInContext(ctx, shard.New("cache"))
 
 ### Authorization/Authentication
 
-Not implemented at the moment
+The cache server can only be connected to using x509 client certificates
+with the `system:masters` group. It does not support authz webhook,
+service accounts or other means of authentication or authorization.
+
+This is because the cache server is only meant to be used by the shards
+directly, not by operators or humans.
+
+The only exception are the health check endpoints, which are always
+accessible without authentication.
 
 ### Built-in Resources
 
@@ -78,6 +86,7 @@ Out of the box, the server supports the following resources:
 - `apiresourceschemas`
 - `apiexports`
 - `shards`
+- `logicalclusters`
 
 All those resources are represented as CustomResourceDefinitions and
 stored in `system:cache:server` shard under `system:system-crds` cluster.

pkg/cache/server/config.go

@@ -126,10 +126,10 @@ func NewConfig(opts *cacheserveroptions.CompletedOptions, optionalLocalShardRest
 		}
 		serverConfig.LoopbackClientConfig = rest.CopyConfig(optionalLocalShardRestConfig)
 	}
-	if err := opts.Authentication.ApplyTo(&serverConfig.Config.Authentication, serverConfig.SecureServing, serverConfig.OpenAPIConfig); err != nil {
+	if err := opts.Authentication.ApplyTo(&serverConfig.Config.Authentication, serverConfig.Config.SecureServing); err != nil {
 		return nil, err
 	}
-	if err := opts.Authorization.ApplyTo(&serverConfig.Config.Authorization); err != nil {
+	if err := opts.Authorization.ApplyTo(&serverConfig.Config); err != nil {
 		return nil, err
 	}
 

pkg/cache/server/options/authentication.go

@@ -0,0 +1,84 @@
+/*
+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 options
+
+import (
+	"fmt"
+
+	"github.com/spf13/pflag"
+
+	"k8s.io/apiserver/pkg/authentication/authenticator"
+	"k8s.io/apiserver/pkg/authentication/request/anonymous"
+	"k8s.io/apiserver/pkg/authentication/request/union"
+	"k8s.io/apiserver/pkg/authentication/request/x509"
+	genericapiserver "k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+)
+
+type Authentication struct {
+	ClientCAFile string
+
+	// EmbeddedAuthenticator is an optional authenticator delegating to the
+	// parent server's authentication chain. Set by the shard server when
+	// the cache server runs embedded.
+	EmbeddedAuthenticator authenticator.Request
+}
+
+func NewAuthentication() *Authentication {
+	return &Authentication{}
+}
+
+func (o *Authentication) AddFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&o.ClientCAFile, "client-ca-file", o.ClientCAFile, "Path to a PEM-encoded certificate bundle. If set, any request presenting a client certificate signed by one of the authorities in the bundle is authenticated with an identity corresponding to the CommonName of the client certificate.")
+}
+
+func (o *Authentication) Validate() []error {
+	return nil
+}
+
+func (o *Authentication) ApplyTo(authenticationInfo *genericapiserver.AuthenticationInfo, servingInfo *genericapiserver.SecureServingInfo) error {
+	if o.ClientCAFile == "" && o.EmbeddedAuthenticator == nil {
+		// This validation cannot happen in .Validate because these
+		// options may be set by the shard embedding the cache server.
+		// For the standalone cache server it doesn't matter if it's
+		// validated here or in .Validate.
+		return fmt.Errorf("either --client-ca-file or an embedded authenticator must be configured")
+	}
+
+	var authenticators []authenticator.Request
+
+	if o.ClientCAFile != "" {
+		caProvider, err := dynamiccertificates.NewDynamicCAContentFromFile("client-ca", o.ClientCAFile)
+		if err != nil {
+			return fmt.Errorf("unable to load client CA file %q: %w", o.ClientCAFile, err)
+		}
+
+		if err := authenticationInfo.ApplyClientCert(caProvider, servingInfo); err != nil {
+			return fmt.Errorf("unable to apply client cert: %w", err)
+		}
+
+		authenticators = append(authenticators, x509.NewDynamic(caProvider.VerifyOptions, x509.CommonNameUserConversion))
+	}
+	if o.EmbeddedAuthenticator != nil {
+		authenticators = append(authenticators, o.EmbeddedAuthenticator)
+	}
+	authenticators = append(authenticators, anonymous.NewAuthenticator(nil))
+
+	authenticationInfo.Authenticator = union.New(authenticators...)
+
+	return nil
+}

pkg/cache/server/options/authorization.go

@@ -0,0 +1,68 @@
+/*
+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 options
+
+import (
+	"github.com/spf13/pflag"
+
+	"k8s.io/apiserver/pkg/authentication/user"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
+	"k8s.io/apiserver/pkg/authorization/authorizerfactory"
+	"k8s.io/apiserver/pkg/authorization/path"
+	"k8s.io/apiserver/pkg/authorization/union"
+	genericapiserver "k8s.io/apiserver/pkg/server"
+)
+
+type Authorization struct {
+	AlwaysAllowPaths  []string
+	AlwaysAllowGroups []string
+}
+
+func NewAuthorization() *Authorization {
+	return &Authorization{
+		AlwaysAllowPaths:  []string{"/healthz", "/readyz", "/livez"},
+		AlwaysAllowGroups: []string{user.SystemPrivilegedGroup},
+	}
+}
+
+func (o *Authorization) AddFlags(fs *pflag.FlagSet) {
+	fs.StringSliceVar(&o.AlwaysAllowPaths, "authorization-always-allow-paths", o.AlwaysAllowPaths,
+		"A list of HTTP paths to skip during authorization, i.e. these are authorized without contacting the 'core' kubernetes server.")
+}
+
+func (o *Authorization) Validate() []error {
+	return nil
+}
+
+func (o *Authorization) ApplyTo(config *genericapiserver.Config) error {
+	var authorizers []authorizer.Authorizer
+
+	if len(o.AlwaysAllowGroups) > 0 {
+		authorizers = append(authorizers, authorizerfactory.NewPrivilegedGroups(o.AlwaysAllowGroups...))
+	}
+
+	if len(o.AlwaysAllowPaths) > 0 {
+		a, err := path.NewAuthorizer(o.AlwaysAllowPaths)
+		if err != nil {
+			return err
+		}
+		authorizers = append(authorizers, a)
+	}
+
+	config.Authorization.Authorizer = union.New(authorizers...)
+	return nil
+}

pkg/cache/server/options/options.go

@@ -32,8 +32,8 @@ type Options struct {
 	ServerRunOptions *genericoptions.ServerRunOptions
 	Etcd             *genericoptions.EtcdOptions
 	SecureServing    *genericoptions.SecureServingOptionsWithLoopback
-	Authentication   *genericoptions.DelegatingAuthenticationOptions
-	Authorization    *genericoptions.DelegatingAuthorizationOptions
+	Authentication   *Authentication
+	Authorization    *Authorization
 	APIEnablement    *genericoptions.APIEnablementOptions
 	EmbeddedEtcd     etcdoptions.Options
 	SyntheticDelay   time.Duration
@@ -43,8 +43,8 @@ type completedOptions struct {
 	ServerRunOptions *genericoptions.ServerRunOptions
 	Etcd             *genericoptions.EtcdOptions
 	SecureServing    *genericoptions.SecureServingOptionsWithLoopback
-	Authentication   *genericoptions.DelegatingAuthenticationOptions
-	Authorization    *genericoptions.DelegatingAuthorizationOptions
+	Authentication   *Authentication
+	Authorization    *Authorization
 	APIEnablement    *genericoptions.APIEnablementOptions
 	EmbeddedEtcd     etcdoptions.CompletedOptions
 	SyntheticDelay   time.Duration
@@ -72,8 +72,8 @@ func NewOptions(rootDir string) *Options {
 		ServerRunOptions: genericoptions.NewServerRunOptions(),
 		Etcd:             genericoptions.NewEtcdOptions(storagebackend.NewDefaultConfig(kubeoptions.DefaultEtcdPathPrefix, nil)),
 		SecureServing:    genericoptions.NewSecureServingOptions().WithLoopback(),
-		Authentication:   genericoptions.NewDelegatingAuthenticationOptions(),
-		Authorization:    genericoptions.NewDelegatingAuthorizationOptions(),
+		Authentication:   NewAuthentication(),
+		Authorization:    NewAuthorization(),
 		APIEnablement:    genericoptions.NewAPIEnablementOptions(),
 		EmbeddedEtcd:     *etcdoptions.NewOptions(rootDir),
 	}
@@ -94,10 +94,6 @@ func (o *Options) Complete() (*CompletedOptions, error) {
 		o.EmbeddedEtcd.Enabled = true
 	}
 
-	// TODO: enable authN/Z stack
-	o.Authentication = nil
-	o.Authorization = nil
-
 	if err := o.SecureServing.MaybeDefaultWithSelfSignedCerts("localhost", nil, nil); err != nil {
 		return nil, err
 	}
@@ -116,5 +112,7 @@ func (o *Options) Complete() (*CompletedOptions, error) {
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	o.EmbeddedEtcd.AddFlags(fs)
 	o.SecureServing.AddFlags(fs)
+	o.Authentication.AddFlags(fs)
+	o.Authorization.AddFlags(fs)
 	fs.DurationVar(&o.SyntheticDelay, "synthetic-delay", 0, "The duration of time the cache server will inject a delay for to all inbound requests. Useful for testing.")
 }

pkg/server/cache.go

@@ -34,6 +34,7 @@ func (s *Server) installCacheServer(ctx context.Context) error {
 	// this could be reworked if we were providing Config/CompletedConfig for the additional servers
 	wasEmbeddedEtcdEnabled := s.Options.EmbeddedEtcd.Enabled
 	s.Options.Cache.Server.EmbeddedEtcd.Enabled = false
+	s.Options.Cache.Server.Authentication.EmbeddedAuthenticator = s.GenericConfig.Authentication.Authenticator
 	newCacheServerConfig, err := cacheserver.NewConfig(s.Options.Cache.Server, rest.CopyConfig(s.GenericConfig.LoopbackClientConfig))
 	if err != nil {
 		return err