apiserver: set proxy auth info via request header

Signed-off-by: Iceber Gu <[email protected]>

Author: Iceber

pkg/kubeapiserver/apiserver.go

@@ -66,7 +66,9 @@ func NewDefaultConfig() *Config {
 }
 
 type ExtraConfig struct {
-	AllowedProxySubresources map[schema.GroupResource]sets.Set[string]
+	AllowPediaClusterConfigReuse    bool
+	ExtraProxyRequestHeaderPrefixes []string
+	AllowedProxySubresources        map[schema.GroupResource]sets.Set[string]
 }
 
 type Config struct {
@@ -149,15 +151,17 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/api/", resourceHandler)
 	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/apis/", resourceHandler)
 
-	controller := NewClusterResourceController(restManager, discoveryManager, c.InformerFactory.Cluster().V1alpha2().PediaClusters())
+	clusterInformer := c.InformerFactory.Cluster().V1alpha2().PediaClusters()
+	_ = NewClusterResourceController(restManager, discoveryManager, clusterInformer)
+
+	connector := proxyrest.NewProxyConnector(clusterInformer.Lister(), c.ExtraConfig.AllowPediaClusterConfigReuse, c.ExtraConfig.ExtraProxyRequestHeaderPrefixes)
 
 	methodSet := sets.New("GET")
-	for _, rest := range proxyrest.GetSubresourceRESTs(controller) {
+	for _, rest := range proxyrest.GetSubresourceRESTs(connector) {
 		allows := c.ExtraConfig.AllowedProxySubresources[rest.ParentGroupResource()]
 		if allows == nil || !allows.Has(rest.Subresource()) {
 			continue
 		}
-
 		if err := restManager.preRegisterSubresource(subresource{
 			gr:         rest.ParentGroupResource(),
 			kind:       rest.ParentKind(),
@@ -178,7 +182,7 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 		}
 	}
 
-	resourceHandler.proxy = proxyrest.NewRemoteProxyREST(c.GenericConfig.Serializer, controller)
+	resourceHandler.proxy = proxyrest.NewRemoteProxyREST(c.GenericConfig.Serializer, connector)
 	return genericserver, methods, nil
 }
 

pkg/kubeapiserver/clusterresource_controller.go

@@ -1,16 +1,11 @@
 package kubeapiserver
 
 import (
-	"context"
-	"errors"
-	"net/http"
 	"reflect"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
 
 	clusterv1alpha2 "github.com/clusterpedia-io/api/cluster/v1alpha2"
@@ -107,66 +102,6 @@ func (c *ClusterResourceController) removeCluster(name string) {
 	delete(c.clusterresources, name)
 }
 
-func (c *ClusterResourceController) resolveClusterRestConfig(name string) (*rest.Config, error) {
-	cluster, err := c.clusterLister.Get(name)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(cluster.Spec.Kubeconfig) != 0 {
-		clientconfig, err := clientcmd.NewClientConfigFromBytes(cluster.Spec.Kubeconfig)
-		if err != nil {
-			return nil, err
-		}
-		return clientconfig.ClientConfig()
-	}
-
-	if cluster.Spec.APIServer == "" {
-		return nil, errors.New("Cluster APIServer Endpoint is required")
-	}
-
-	if len(cluster.Spec.TokenData) == 0 &&
-		(len(cluster.Spec.CertData) == 0 || len(cluster.Spec.KeyData) == 0) {
-		return nil, errors.New("Cluster APIServer's Token or Cert is required")
-	}
-
-	config := &rest.Config{
-		Host: cluster.Spec.APIServer,
-	}
-
-	if len(cluster.Spec.CAData) != 0 {
-		config.TLSClientConfig.CAData = cluster.Spec.CAData
-	} else {
-		config.TLSClientConfig.Insecure = true
-	}
-
-	if len(cluster.Spec.CertData) != 0 && len(cluster.Spec.KeyData) != 0 {
-		config.TLSClientConfig.CertData = cluster.Spec.CertData
-		config.TLSClientConfig.KeyData = cluster.Spec.KeyData
-	}
-
-	if len(cluster.Spec.TokenData) != 0 {
-		config.BearerToken = string(cluster.Spec.TokenData)
-	}
-	return config, nil
-}
-
-func (c *ClusterResourceController) GetClusterDefaultConnection(ctx context.Context, name string) (string, http.RoundTripper, error) {
-	config, err := c.resolveClusterRestConfig(name)
-	if err != nil {
-		return "", nil, err
-	}
-	transport, err := rest.TransportFor(config)
-	if err != nil {
-		return "", nil, err
-	}
-	return config.Host, transport, nil
-}
-
-func (c *ClusterResourceController) GetClusterConnectionWithTLSConfig(ctx context.Context, name string) (string, http.RoundTripper, error) {
-	return "", nil, errors.New("CetClusterConnectionWithTLSConfig not implemented")
-}
-
 type resourceInfo struct {
 	Namespaced bool
 	Kind       string

pkg/kubeapiserver/options.go

@@ -4,17 +4,22 @@ import (
 	"fmt"
 	"strings"
 
+	proxyrest "github.com/clusterpedia-io/clusterpedia/pkg/kubeapiserver/resourcerest/proxy"
 	"github.com/spf13/pflag"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 type Options struct {
-	AllowedProxySubresources []string
+	AllowPediaClusterConfigForProxyRequest bool
+	AllowedProxySubresources               []string
+	ExtraProxyRequestHeaderPrefixes        []string
 }
 
 func NewOptions() *Options {
-	return &Options{}
+	return &Options{
+		ExtraProxyRequestHeaderPrefixes: []string{proxyrest.DefaultProxyRequestHeaderPrefix},
+	}
 }
 
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
@@ -33,6 +38,10 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 		"List of subresources that support proxying requests to the specified cluster, formatted as '[resource/subresource],[subresource],...'. "+
 		fmt.Sprintf("Supported proxy subresources include %q", strings.Join(resources, ",")),
 	)
+
+	fs.BoolVar(&o.AllowPediaClusterConfigForProxyRequest, "allow-pediacluster-config-for-proxy-request", o.AllowPediaClusterConfigForProxyRequest, ""+
+		"Allow proxy requests to use the cluster configuration from PediaCluster when authentication information cannot be obtained from the header.",
+	)
 }
 
 var supportedProxyCoreSubresources = map[string][]string{
@@ -75,5 +84,8 @@ func (o *Options) Config() (*ExtraConfig, error) {
 			return nil, fmt.Errorf("--allowed-proxy-subresources: unsupported subresources or invalid format %q", subresource)
 		}
 	}
-	return &ExtraConfig{AllowedProxySubresources: subresources}, nil
+	return &ExtraConfig{
+		AllowPediaClusterConfigReuse: o.AllowPediaClusterConfigForProxyRequest,
+		AllowedProxySubresources:     subresources,
+	}, nil
 }

pkg/kubeapiserver/resourcerest/proxy/proxy.go

@@ -19,8 +19,7 @@ import (
 )
 
 type ClusterConnectionGetter interface {
-	GetClusterDefaultConnection(ctx context.Context, cluster string) (string, http.RoundTripper, error)
-	GetClusterConnectionWithTLSConfig(ctx context.Context, cluster string) (string, http.RoundTripper, error)
+	GetClusterConnection(ctx context.Context, cluster string, req *http.Request) (string, http.RoundTripper, error)
 }
 
 type RemoteProxyREST struct {
@@ -33,7 +32,7 @@ func NewRemoteProxyREST(serializer runtime.NegotiatedSerializer, connGetter Clus
 }
 
 func (r *RemoteProxyREST) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
-	handler, err := proxyConn(req.Context(), r.connGetter, false, r, nil)
+	handler, err := proxyConn(req.Context(), r.connGetter, false, false, r, nil)
 	if err != nil {
 		r.Error(rw, req, err)
 	}
@@ -55,26 +54,28 @@ func proxyConn(ctx context.Context, connGetter ClusterConnectionGetter, upgradeR
 		return nil, errors.New("missing RequestInfo")
 	}
 
-	// TODO(iceber): need disconnect when the cluster authentication information changes
-	endpoint, transport, err := connGetter.GetClusterDefaultConnection(ctx, clusterName)
-	if err != nil {
-		return nil, err
-	}
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		// TODO(iceber): need disconnect when the cluster authentication information changes
+		endpoint, transport, err := connGetter.GetClusterConnection(ctx, clusterName, req)
+		if err != nil {
+			responder.Error(rw, req, err)
+			return
+		}
 
-	target, err := url.ParseRequestURI(endpoint + requestInfo.Path)
-	if err != nil {
-		return nil, err
-	}
-	target.RawQuery = request.RequestQueryFrom(ctx).Encode()
+		target, err := url.ParseRequestURI(endpoint + requestInfo.Path)
+		if err != nil {
+			responder.Error(rw, req, err)
+			return
+		}
+		target.RawQuery = request.RequestQueryFrom(ctx).Encode()
 
-	proxy := proxy.NewUpgradeAwareHandler(target, transport, false, upgradeRequired, responder)
-	proxy.UseLocationHost = true
+		proxy := proxy.NewUpgradeAwareHandler(target, transport, false, upgradeRequired, responder)
+		proxy.UseLocationHost = true
 
-	var handler http.Handler = proxy
-	if wrapProxy != nil {
-		handler = wrapProxy(proxy)
-	}
-	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		var handler http.Handler = proxy
+		if wrapProxy != nil {
+			handler = wrapProxy(proxy)
+		}
 		r := req.WithContext(req.Context())
 		r.Header = utilnet.CloneHeader(req.Header)
 		if auditID, _ := audit.AuditIDFrom(ctx); auditID != "" {

pkg/kubeapiserver/resourcerest/proxy/proxy_connector.go

@@ -0,0 +1,122 @@
+package proxy
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"k8s.io/client-go/rest"
+
+	clusterlister "github.com/clusterpedia-io/clusterpedia/pkg/generated/listers/cluster/v1alpha2"
+	"github.com/clusterpedia-io/clusterpedia/pkg/utils"
+)
+
+const DefaultProxyRequestHeaderPrefix = "X-Clusterpedia-Proxy-"
+
+type Connector struct {
+	allowConfigReuse    bool
+	extraHeaderPrefixes []string
+	clusterLister       clusterlister.PediaClusterLister
+}
+
+func NewProxyConnector(clusterLister clusterlister.PediaClusterLister, allowPediaClusterConfigReuse bool, extraHeaderPrefixes []string) ClusterConnectionGetter {
+	if len(extraHeaderPrefixes) == 0 {
+		extraHeaderPrefixes = []string{DefaultProxyRequestHeaderPrefix}
+	}
+	return &Connector{
+		allowConfigReuse:    allowPediaClusterConfigReuse,
+		extraHeaderPrefixes: extraHeaderPrefixes,
+		clusterLister:       clusterLister,
+	}
+}
+
+func (c *Connector) GetClusterConnection(ctx context.Context, name string, req *http.Request) (string, http.RoundTripper, error) {
+	cluster, err := c.clusterLister.Get(name)
+	if err != nil {
+		return "", nil, err
+	}
+	config := &rest.Config{}
+	if cluster.Status.APIServer != "" {
+		config.Host = cluster.Status.APIServer
+	} else if cluster.Spec.APIServer != "" {
+		config.Host = cluster.Spec.APIServer
+	} else {
+		return "", nil, errors.New(".Spec.APIServer and .Status.APIServer are empty")
+	}
+	config.TLSClientConfig.Insecure = true
+
+	var authInHeader bool
+	extra := map[string][]string{}
+
+	headers := req.Header.Clone()
+	for _, prefix := range c.extraHeaderPrefixes {
+		for headerName, vv := range headers {
+			if !(len(headerName) >= len(prefix) && strings.EqualFold(headerName[:len(prefix)], prefix)) {
+				continue
+			}
+
+			extraKey := unescapeExtraKey(strings.ToLower(headerName[len(prefix):]))
+			extra[extraKey] = append(extra[extraKey], vv...)
+			req.Header.Del(headerName)
+		}
+	}
+
+	for key, vals := range extra {
+		switch key {
+		case "ca":
+			authInHeader = true
+			if len(vals) > 0 {
+				config.TLSClientConfig.CAData, err = base64.StdEncoding.DecodeString(vals[0])
+				if err != nil {
+					return "", nil, err
+				}
+				config.TLSClientConfig.Insecure = false
+			}
+		case "token":
+			authInHeader = true
+			if len(vals) > 0 {
+				config.BearerToken = vals[0]
+			}
+		case "client-cert":
+			authInHeader = true
+			if len(vals) > 0 {
+				config.TLSClientConfig.CertData, err = base64.StdEncoding.DecodeString(vals[0])
+				if err != nil {
+					return "", nil, err
+				}
+			}
+		case "client-key":
+			authInHeader = true
+			if len(vals) > 0 {
+				config.TLSClientConfig.KeyData, err = base64.StdEncoding.DecodeString(vals[0])
+				if err != nil {
+					return "", nil, err
+				}
+			}
+		}
+	}
+
+	if !authInHeader && c.allowConfigReuse {
+		config, err = utils.BuildClusterRestConfig(cluster)
+		if err != nil {
+			return "", nil, err
+		}
+	}
+
+	transport, err := rest.TransportFor(config)
+	if err != nil {
+		return "", nil, err
+	}
+	return config.Host, transport, nil
+}
+
+func unescapeExtraKey(encodedKey string) string {
+	key, err := url.PathUnescape(encodedKey) // Decode %-encoded bytes.
+	if err != nil {
+		return encodedKey // Always record extra strings, even if malformed/unencoded.
+	}
+	return key
+}