finish ut of sa token replacement (openyurtio#787)

* finish ut of sa token replacement
add tenant sa token subsitute

* add tenant sa token subsitute

.golangci.yaml

@@ -3,6 +3,7 @@
 
 # options for analysis running
 run:
+  go: '1.16'
   # default concurrency is a available CPU number
   concurrency: 4
 
@@ -44,4 +45,4 @@ linters:
     - gci
     - ineffassign
     - misspell
-    - vet
+    - vet

Makefile

@@ -52,7 +52,7 @@ vet:
 #   make release WHAT="yurthub yurt-controller-manager yurtctl-servant" ARCH="arm64 arm" REGION=cn
 #
 #   # compile all components with all architectures (i.e., amd64, arm64, arm)
-#   make relase
+#   make release
 release:
 	bash hack/make-rules/release-images.sh
 

cmd/yurthub/app/config/config.go

@@ -123,9 +123,10 @@ func Complete(options *options.YurtHubOptions) (*YurtHubConfiguration, error) {
 	if err != nil {
 		return nil, err
 	}
-	registerInformers(sharedFactory, yurtSharedFactory, workingMode, serviceTopologyFilterEnabled(options), options.NodePoolName, options.NodeName)
-
+	tenantNs := util.ParseTenantNs(options.YurtHubCertOrganizations)
+	registerInformers(sharedFactory, yurtSharedFactory, workingMode, serviceTopologyFilterEnabled(options), options.NodePoolName, options.NodeName, tenantNs)
 	filterManager, err := createFilterManager(options, sharedFactory, yurtSharedFactory, serializerManager, storageWrapper, us[0].Host, proxySecureServerDummyAddr, proxySecureServerAddr)
+
 	if err != nil {
 		klog.Errorf("could not create filter manager, %v", err)
 		return nil, err
@@ -226,7 +227,8 @@ func registerInformers(informerFactory informers.SharedInformerFactory,
 	yurtInformerFactory yurtinformers.SharedInformerFactory,
 	workingMode util.WorkingMode,
 	serviceTopologyFilterEnabled bool,
-	nodePoolName, nodeName string) {
+	nodePoolName, nodeName string,
+	tenantNs string) {
 	// skip construct node/nodePool informers if service topology filter disabled
 	if serviceTopologyFilterEnabled {
 		if workingMode == util.WorkingModeCloud {
@@ -258,6 +260,15 @@ func registerInformers(informerFactory informers.SharedInformerFactory,
 		return coreinformers.NewFilteredConfigMapInformer(client, util.YurtHubNamespace, resyncPeriod, nil, tweakListOptions)
 	}
 	informerFactory.InformerFor(&corev1.ConfigMap{}, newConfigmapInformer)
+
+	if tenantNs != "" {
+		newSecretInformer := func(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+
+			return coreinformers.NewFilteredSecretInformer(client, tenantNs, resyncPeriod, nil, nil)
+		}
+		informerFactory.InformerFor(&corev1.Secret{}, newSecretInformer)
+	}
+
 }
 
 // registerAllFilters by order, the front registered filter will be

cmd/yurthub/app/start.go

@@ -36,6 +36,7 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/network"
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy"
 	"github.com/openyurtio/openyurt/pkg/yurthub/server"
+	"github.com/openyurtio/openyurt/pkg/yurthub/tenant"
 	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
@@ -154,8 +155,13 @@ func Run(cfg *config.YurtHubConfiguration, stopCh <-chan struct{}) error {
 	}
 	trace++
 
+	klog.Infof("%d. new tenant sa manager", trace)
+	tenantMgr := tenant.New(cfg.YurtHubCertOrganizations, cfg.SharedFactory, stopCh)
+	trace++
+
 	klog.Infof("%d. new reverse proxy handler for remote servers", trace)
-	yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(cfg, cacheMgr, transportManager, healthChecker, certManager, stopCh)
+	yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(cfg, cacheMgr, transportManager, healthChecker, certManager, tenantMgr, stopCh)
+
 	if err != nil {
 		return fmt.Errorf("could not create reverse proxy handler, %v", err)
 	}

go.mod

@@ -26,6 +26,7 @@ require (
 	golang.org/x/sys v0.0.0-20210616094352-59db8d763f22
 	google.golang.org/grpc v1.27.1
 	gopkg.in/cheggaaa/pb.v1 v1.0.25
+	gopkg.in/square/go-jose.v2 v2.2.2
 	k8s.io/api v0.22.3
 	k8s.io/apimachinery v0.22.3
 	k8s.io/apiserver v0.20.11

go.sum

@@ -902,4 +902,4 @@ sigs.k8s.io/structured-merge-diff/v4 v4.1.2 h1:Hr/htKFmJEbtMgS/UD0N+gtgctAqz81t3
 sigs.k8s.io/structured-merge-diff/v4 v4.1.2/go.mod h1:j/nl6xW8vLS49O8YvXW1ocPhZawJtm+Yrr7PPRQ0Vg4=
 sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
 sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
-sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=

pkg/yurthub/proxy/proxy.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apiserver/pkg/endpoints/filters"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/server"
+	"k8s.io/klog/v2"
 
 	"github.com/openyurtio/openyurt/cmd/yurthub/app/config"
 	"github.com/openyurtio/openyurt/pkg/yurthub/cachemanager"
@@ -32,6 +33,7 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy/local"
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy/remote"
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy/util"
+	"github.com/openyurtio/openyurt/pkg/yurthub/tenant"
 	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
 	hubutil "github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
@@ -43,6 +45,7 @@ type yurtReverseProxy struct {
 	localProxy          *local.LocalProxy
 	cacheMgr            cachemanager.CacheManager
 	maxRequestsInFlight int
+	tenantMgr           tenant.Interface
 	stopCh              <-chan struct{}
 }
 
@@ -54,6 +57,7 @@ func NewYurtReverseProxyHandler(
 	transportMgr transport.Interface,
 	healthChecker healthchecker.HealthChecker,
 	certManager interfaces.YurtCertificateManager,
+	tenantMgr tenant.Interface,
 	stopCh <-chan struct{}) (http.Handler, error) {
 	cfg := &server.Config{
 		LegacyAPIGroupPrefixes: sets.NewString(server.DefaultLegacyAPIPrefix),
@@ -87,6 +91,7 @@ func NewYurtReverseProxyHandler(
 		localProxy:          localProxy,
 		cacheMgr:            cacheMgr,
 		maxRequestsInFlight: yurtHubCfg.MaxRequestInFlight,
+		tenantMgr:           tenantMgr,
 		stopCh:              stopCh,
 	}
 
@@ -105,7 +110,15 @@ func (p *yurtReverseProxy) buildHandlerChain(handler http.Handler) http.Handler
 	}
 	handler = util.WithMaxInFlightLimit(handler, p.maxRequestsInFlight)
 	handler = util.WithRequestClientComponent(handler)
+
+	if p.tenantMgr != nil && p.tenantMgr.GetTenantNs() != "" {
+		handler = util.WithSaTokenSubstitute(handler, p.tenantMgr)
+	} else {
+		klog.V(2).Info("tenant ns is empty, no need to substitute ")
+	}
+
 	handler = filters.WithRequestInfo(handler, p.resolver)
+
 	return handler
 }
 

pkg/yurthub/proxy/util/util.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The OpenYurt Authors.
+Copyright 2022 The OpenYurt Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -18,20 +18,24 @@ package util
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
 
+	"gopkg.in/square/go-jose.v2/jwt"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metainternalversionscheme "k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apiserver/pkg/authentication/serviceaccount"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/klog/v2"
 
 	"github.com/openyurtio/openyurt/pkg/yurthub/metrics"
+	"github.com/openyurtio/openyurt/pkg/yurthub/tenant"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
 
@@ -147,6 +151,7 @@ func WithRequestClientComponent(handler http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		ctx := req.Context()
 		if info, ok := apirequest.RequestInfoFrom(ctx); ok {
+
 			if info.IsResourceRequest {
 				var comp string
 				userAgent := strings.ToLower(req.Header.Get("User-Agent"))
@@ -295,3 +300,36 @@ func WithRequestTimeout(handler http.Handler) http.Handler {
 		handler.ServeHTTP(w, req)
 	})
 }
+
+func WithSaTokenSubstitute(handler http.Handler, tenantMgr tenant.Interface) http.Handler {
+
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+
+		if oldToken := util.ParseBearerToken(req.Header.Get("Authorization")); oldToken != "" { //bearer token is not empty&valid
+
+			if jsonWebToken, err := jwt.ParseSigned(oldToken); err != nil {
+
+				klog.Errorf("invaled bearer token %s, err: %v", oldToken, err)
+			} else {
+				oldClaim := jwt.Claims{}
+
+				if err := jsonWebToken.UnsafeClaimsWithoutVerification(&oldClaim); err == nil {
+
+					if tenantNs, _, err := serviceaccount.SplitUsername(oldClaim.Subject); err == nil {
+
+						if tenantMgr.GetTenantNs() != tenantNs && tenantNs == "kube-system" && tenantMgr.WaitForCacheSync() { // token is not from tenant's namespace
+							req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", tenantMgr.GetTenantToken()))
+							klog.V(2).Infof("replace token, old: %s, new: %s", oldToken, tenantMgr.GetTenantToken())
+						}
+
+					} else {
+						klog.Errorf("failed to parse tenant ns from token, token %s, sub: %s", oldToken, oldClaim.Subject)
+					}
+				}
+			}
+
+		}
+
+		handler.ServeHTTP(w, req)
+	})
+}