Kubesphere 权限认证代码阅读

chinazj · 2020年8月5日

Kubesphere 权限认证

kubernetes权限RBAC

kubernetes使用的RBAC的示意图如下：user1 可以使用role1和role2所允许的资源，user2只可以使用 role1 所允许的资源。我们有时候会称user为subject

user1 ------>rolebinding1<------ role1
user1 ------>rolebinding2<------ role2
user2 ------>rolebinding3<------ role1
user  ------>XXXXbinding  <------XXrole

在kubernetes，serviceaccount、user、group为user。clusterrole和role代表role。clusterrolebinding和rolebinding代表rolebinding。

Role 和 ClusterRole：角色和集群角色，这两个对象都包含 Rules，定义了这个role所能允许的资源。二者的区别在于，在 Role 中，定义的规则只适用于单个命名空间，也就是和 namespace 关联的，而 ClusterRole 是集群范围内的，因此定义的规则不受命名空间的约束。
Subject：主题，对应上面提到的user，集群中定义了3种类型的主题资源：
- User Account ：用户。Kubernetes没有 User Account 的 API 对象。但是我们可以用kubectl 创建密钥进行登录。
- Group：一组用户。
- Service Account ：服务帐号，通过Kubernetes API 来管理的一些用户帐号，和 namespace 进行关联的，适用于集群内部运行的应用程序，需要通过 API 来完成权限认证
RoleBinding和ClusterRoleBinding：将role和subject绑定。确定每一个subject能够拥有的资源权限

Kubesphere权限

Kubesphere新增了一个user、WorkspaceRole、GlobalRole 。user用于管理账号。WorkspaceRole、GlobalRole 是和role、clusterrole一样包含允许的资源。

为了方便理解，请将这四个role理解为并列的role。一个user通过binding来绑定不同role来获取不同资源权限。举例来说初始化的admin user同时包含cluster-admin(ClusterRole类型)和platform-admin(GlobalRole类型)。

User: 账号管理，不被namespace、workspace限制。
WorkspaceRole：workspace下的资源权限role。该role定义了在workspace下所允许的资源列表。被workspace限制。
GlobalRole：全局角色。该role定义了所允许的集群资源列表。如：user、workspace等资源。不被namespace、workspace限制。
role ：管理namespace下的资源。被namespace限制。
clusterrole：管理单个集群。该role定义了所允许集群的资源列表

验证权限的过程

填充请求，确定请求的范围，单集群还是多集群，是否是直发给k8s的请求。
验证用户，密码、token等方式。如果是无用户则定义为anonymous
如果是多集群，则需要转发。
确定该资源是否能被该用户获取。
1. 通过user，globalRoleBindings获取到globalRole。查看globalRole对该资源是否允许。
2. 通过user，workspaceRoleBindings获取到workspaceRole。查看workspaceRole对该资源是否允许。
3. 通过user，roleBindings获取到role。查看role对该资源是否允许。
4. 通过user，clusterRoleBindings获取到clusterRole。查看clusterRole对该资源是否允许。
  
  > 为了减少查询次数，通过if来过滤。比如workspace类型的资源绝不会出现role和clusterRole上
审计
如果是k8s的资源，类似于kubectl get pod -A的资源。则转发给kube-apiserver

代码分析

kubesphere使用iam模块来进行权限认证和权限获取。kubesphere使用go-restful开发web应用框架。权限认证和权限获取都是在filter进行的

权限代码如下。注意：filter使用封装handle的方式，类似于套娃一样，先被封装的却是最后运行的。所以下面请求的执行过程是：WithRequestInfo--->WithAuthentication--->WithMultipleClusterDispatcher--->WithAuthorization--->WithAuditing---->WithKubeAPIServer

func (s *APIServer) buildHandlerChain(stopCh <-chan struct{}) {
	//.....
	handler := s.Server.Handler
	handler = filters.WithKubeAPIServer(handler, s.KubernetesClient.Config(), &errorResponder{})

	if s.Config.AuditingOptions.Enable {
		handler = filters.WithAuditing(handler,
			audit.NewAuditing(s.InformerFactory, s.Config.AuditingOptions.WebhookUrl, stopCh))
	}
	//.....
	handler = filters.WithAuthorization(handler, authorizers)
	if s.Config.MultiClusterOptions.Enable {
		//.....
		handler = filters.WithMultipleClusterDispatcher(handler, clusterDispatcher)
	}

	loginRecorder := im.NewLoginRecorder(s.KubernetesClient.KubeSphere())
	//.....
	handler = filters.WithAuthentication(handler, authn, loginRecorder)
	handler = filters.WithRequestInfo(handler, requestInfoResolver)
	s.Server.Handler = handler
}

主要分为一下步骤

WithRequestInfo：填充请求信息
WithAuthentication：权限认证，password，匿名和token的方式
WithMultipleClusterDispatcher：多集群路径填补和转发。多集群时根据requestinfo，将信息发送给对应的集群。
WithAuthorization：权限获取，判断是否对集群资源有获取权限
WithAuditing：审计
WithKubeAPIServer：转发，如果是直接对k8s 的请求，就直接转发给k8s。

填充请求信息

请求信息如下：

type RequestInfo struct {
    // import k8s.io/apiserver/pkg/endpoints/request
	*k8srequest.RequestInfo
	//是否是k8s的请求，如果是k8s的请求，则会直接转发到kube-apiserver
	IsKubernetesRequest bool

	//请求资源所在的Workspace，可以为空
	Workspace string

	//请求资源所在的Cluster，如果是单集群则为空
	Cluster string

	//请求资源所在的devops项目
	DevOps string

	// 资源请求范围
	ResourceScope string
}

认证

有三个认证方式，anonymous、password、token。三个方法任意一个通过校验，就立刻返回。

anonymous的实现如下

   func (a *Authenticator) AuthenticateRequest(req *http.Request) (*authenticator.Response, bool, error) {
   	auth := strings.TrimSpace(req.Header.Get("Authorization"))
   	if auth == "" {
   		return &authenticator.Response{
   			User: &user.DefaultInfo{
   				Name:   user.Anonymous,
   				UID:    "",
   				Groups: []string{user.AllUnauthenticated},
   			},
   		}, true, nil
   	}
   	return nil, false, nil
   }

password的实现如下：从user资源中获取信息。

   func (im *passwordAuthenticator) Authenticate(username, password string) (authuser.Info, error) {
   	//从k8s 集群中查询到user资源
   	user, err := im.searchUser(username)
   	
   	providerOptions, ldapProvider := im.getLdapProvider()
   	//....
       //通过ldapProvider来判断账号密码是否正确
   	if ldapProvider != nil && username != constants.AdminUserName {
   		//
   		authenticated, err := ldapProvider.Authenticate(username, password)
   		//....
   		if authenticated != nil {
   			return &authuser.DefaultInfo{
   				Name: authenticated.Name,
   				UID:  string(authenticated.UID),
   			}, nil
   		}
   	}
   	//也可以通过use资源的EncryptedPassword，验证密码
   	if checkPasswordHash(password, user.Spec.EncryptedPassword) {
   		return &authuser.DefaultInfo{
   			Name: user.Name,
   			UID:  string(user.UID),
   		}, nil
   	}
   
   	return nil, AuthFailedIncorrectPassword
   }

token：使用jwt验证

   func (t tokenOperator) Verify(tokenStr string) (user.Info, error) {
       //调用jwt包验证
   	authenticated, tokenType, err := t.issuer.Verify(tokenStr)
   	//这里应该是过期时间，0为永不过期。这个为配置设置
   	if t.options.OAuthOptions.AccessTokenMaxAge == 0 ||
   		tokenType == token.StaticToken {
   		return authenticated, nil
   	}
       //redis验证
   	if err := t.tokenCacheValidate(authenticated.GetName(), tokenStr); err != nil {
   		klog.Error(err)
   		return nil, err
   	}
   	return authenticated, nil
   }

多集群路由转发和协议升级

只在多集群下运行此函数

func (c *clusterDispatch) Dispatch(w http.ResponseWriter, req *http.Request, handler http.Handler) {
	info, _ := request.RequestInfoFrom(req.Context())
	//cluster 是crd 资源。获取集群所有cluster信息
	cluster, err := c.clusterLister.Get(info.Cluster)
	//请求集群是主机集群，不需要通过代理
	if isClusterHostCluster(cluster) {
		req.URL.Path = strings.Replace(req.URL.Path, fmt.Sprintf("/clusters/%s", info.Cluster), "", 1)
		handler.ServeHTTP(w, req)
		return
	}
	//查询集群
	innCluster := c.getInnerCluster(cluster.Name)
	transport := http.DefaultTransport
	//替换url
	u := *req.URL
	u.Path = strings.Replace(u.Path, fmt.Sprintf("/clusters/%s", info.Cluster), "", 1)
	if cluster.Spec.Connection.Type == clusterv1alpha1.ConnectionTypeDirect &&
		len(cluster.Spec.Connection.KubeSphereAPIEndpoint) == 0 {
		u.Scheme = innCluster.kubernetesURL.Scheme
		u.Host = innCluster.kubernetesURL.Host
		u.Path = fmt.Sprintf(proxyURLFormat, u.Path)
		transport = innCluster.transport
         //...
	} else {
		u.Host = innCluster.kubesphereURL.Host
		u.Scheme = innCluster.kubesphereURL.Scheme
	}
    
	httpProxy := proxy.NewUpgradeAwareHandler(&u, transport, false, false, c)
	httpProxy.ServeHTTP(w, req)
}

权限获取

权限获取是通过RBAC方式。先调用Authorize

func (r *RBACAuthorizer) Authorize(requestAttributes authorizer.Attributes) (authorizer.Decision, string, error) {
	//调用visitRulesFor检查权限
	r.visitRulesFor(requestAttributes, ruleCheckingVisitor.visit)
	if ruleCheckingVisitor.allowed {
		return authorizer.DecisionAllow, ruleCheckingVisitor.reason, nil
	}
    //...
    //记录日志
    klog.Infof("...")
    //...
	return authorizer.DecisionNoOpinion, reason, nil
}

在调用visitRulesFor，

func (r *RBACAuthorizer) visitRulesFor(requestAttributes authorizer.Attributes, visitor func(source fmt.Stringer, regoPolicy string, rule *rbacv1.PolicyRule, err error) bool) {
	//查看globalRole是否有对此资源的获取权限
	if globalRoleBindings, err := r.am.ListGlobalRoleBindings(""); err != nil {
		//...
	} else {
		sourceDescriber := &globalRoleBindingDescriber{}
        //循环globalRoleBindings。找到该用户的globalRoleBindings。
		for _, globalRoleBinding := range globalRoleBindings {
			subjectIndex, applies := appliesTo(requestAttributes.GetUser(), globalRoleBinding.Subjects, "")
			if !applies {
                //如果不是这个找到该用户的globalRoleBindings，则continue
				continue
			}
            //根据globalRoleBinding上的roleref，获取到role。进一步获取到regoPolicy和rules。这是两个规则，只要有一个规则符合即可
			regoPolicy, rules, err := r.am.GetRoleReferenceRules(globalRoleBinding.RoleRef, "")
			//...
             //根据regoPolicy验证是否符合规则
			if !visitor(sourceDescriber, regoPolicy, nil, nil) {
				return
			}
             //根据rules验证是否符合规则
			for i := range rules {
				if !visitor(sourceDescriber, "", &rules[i], nil) {
					return
				}
			}
		}
        //...
	}
	
	if requestAttributes.GetResourceScope() == request.WorkspaceScope ||
		requestAttributes.GetResourceScope() == request.NamespaceScope ||
		requestAttributes.GetResourceScope() == request.DevOpsScope {
		//...
         // 代码部分删除，这部分主要是获取workspace
	    workspace, err = r.am.GetNamespaceControlledWorkspace(requestAttributes.GetNamespace()); err != nil 
		// 通过workspace获取workspaceRoleBindings。
		if workspaceRoleBindings, err := r.am.ListWorkspaceRoleBindings("", workspace); ...{
		     //...
		} else {
			//...轮训workspaceRoleBindings，找到workspaceRole，在进行权限检测。这部分和grobalrole一样
			for _, workspaceRoleBinding := range workspaceRoleBindings {
				subjectIndex, applies := appliesTo(requestAttributes.GetUser(), workspaceRoleBinding.Subjects, "")
				if !applies {
					continue
				}
				regoPolicy, rules, err := r.am.GetRoleReferenceRules(workspaceRoleBinding.RoleRef, "")
				//...
				if !visitor(sourceDescriber, regoPolicy, nil, nil) {
					return
				}
				for i := range rules {
					if !visitor(sourceDescriber, "", &rules[i], nil) {
						return
					}
				}
			}
		}
	}

	if requestAttributes.GetResourceScope() == request.NamespaceScope ||
		requestAttributes.GetResourceScope() == request.DevOpsScope {
		
		namespace := requestAttributes.GetNamespace()
		// 直接获取namespace，或者根据DevOps获取namespace
		if requestAttributes.GetResourceScope() == request.DevOpsScope {
			if relatedNamespace, err := r.am.GetDevOpsRelatedNamespace(requestAttributes.GetDevOps()); err != nil {
				if !visitor(nil, "", nil, err) {
					return
				}
			} else {
				namespace = relatedNamespace
			}
		}
		//根据namespace获取rolebinding
		if roleBindings, err := r.am.ListRoleBindings("", namespace); err != nil {
			if !visitor(nil, "", nil, err) {
				return
			}
		} else {
			sourceDescriber := &roleBindingDescriber{}
             //轮训roleBindings，找到role,检查role
			for _, roleBinding := range roleBindings {
				subjectIndex, applies := appliesTo(requestAttributes.GetUser(), roleBinding.Subjects, namespace)
				if !applies {
					continue
				}
				regoPolicy, rules, err := r.am.GetRoleReferenceRules(roleBinding.RoleRef, namespace)
				if err != nil {
					visitor(nil, "", nil, err)
					continue
				}
				sourceDescriber.binding = roleBinding
				sourceDescriber.subject = &roleBinding.Subjects[subjectIndex]
				if !visitor(sourceDescriber, regoPolicy, nil, nil) {
					return
				}
				for i := range rules {
					if !visitor(sourceDescriber, "", &rules[i], nil) {
						return
					}
				}
			}
		}
	}
	//获取所有clusterRoleBindings，轮训查找到clusterRole。检查规则
	if clusterRoleBindings, err := r.am.ListClusterRoleBindings(""); err != nil {
		if !visitor(nil, "", nil, err) {
			return
		}
	} else {
		sourceDescriber := &clusterRoleBindingDescriber{}
		for _, clusterRoleBinding := range clusterRoleBindings {
			subjectIndex, applies := appliesTo(requestAttributes.GetUser(), clusterRoleBinding.Subjects, "")
			if !applies {
				continue
			}
			regoPolicy, rules, err := r.am.GetRoleReferenceRules(clusterRoleBinding.RoleRef, "")
			if err != nil {
				visitor(nil, "", nil, err)
				continue
			}
			sourceDescriber.binding = clusterRoleBinding
			sourceDescriber.subject = &clusterRoleBinding.Subjects[subjectIndex]
			if !visitor(sourceDescriber, regoPolicy, nil, nil) {
				return
			}
			for i := range rules {
				if !visitor(sourceDescriber, "", &rules[i], nil) {
					return
				}
			}
		}
	}
}

检查规则的函数是visitor，如果regoPolicy符合就不检查rule

func (v *authorizingVisitor) visit(source fmt.Stringer, regoPolicy string, rule *rbacv1.PolicyRule, err error) bool {
    //调用open-policy-agent库检查权限
	if regoPolicy != "" && regoPolicyAllows(v.requestAttributes, regoPolicy) {
		v.allowed = true
		v.reason = fmt.Sprintf("RBAC: allowed by %s", source.String())
		return false
	}
    //调用k8s 接口实现rbac检查权限
	if rule != nil && ruleAllows(v.requestAttributes, rule) {
		v.allowed = true
		v.reason = fmt.Sprintf("RBAC: allowed by %s", source.String())
		return false
	}
	if err != nil {
		v.errors = append(v.errors, err)
	}
	return true
}

审计

主要记录请求日志，这部分主要使用k8s的Auditing接口。

func WithAuditing(handler http.Handler, a auditing.Auditing) http.Handler {

	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		//...
        //构建Event结构体
		e := a.LogRequestObject(req, info)
		if e != nil {
			resp := auditing.NewResponseCapture(w)
			handler.ServeHTTP(resp, req)
			go a.LogResponseObject(e, resp)
		} else {
			handler.ServeHTTP(w, req)
		}
	})
}

k8s集群资源转发

如果这个资源是向k8s请求的，则直接向k8s请求

func WithKubeAPIServer(handler http.Handler, config *rest.Config, failed proxy.ErrorResponder) http.Handler {
	//...
    //这部分为初始化内容，在请求时不执行。
	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		info, ok := request.RequestInfoFrom(req.Context())
		if !ok {
			err := errors.New("Unable to retrieve request info from request")
			klog.Error(err)
			responsewriters.InternalError(w, req, err)
		}
		// 确认是否为k8s请求
		if info.IsKubernetesRequest {
			s := *req.URL
			s.Host = kubernetes.Host
			s.Scheme = kubernetes.Scheme

			// make sure we don't override kubernetes's authorization
			req.Header.Del("Authorization")
            //转发
			httpProxy := proxy.NewUpgradeAwareHandler(&s, defaultTransport, true, false, failed)
			httpProxy.UpgradeTransport = proxy.NewUpgradeRequestRoundTripper(defaultTransport, defaultTransport)
			httpProxy.ServeHTTP(w, req)
			return
		}

		handler.ServeHTTP(w, req)
	})
}

gyx2350094959 · 2021年7月20日

企业空间创建也是根据kubesphere下rbac.go验证的吗

gyx2350094959 · 2021年7月20日

可以这么理解吗？只要是kapis的，都会用kubesphere的rbac代码验证，包括增删改查；api和apis的转发给k8s去验证

ycsk02 · 2022年1月11日

gyx2350094959

我理解，不是。

除了excludedPaths := []string{“/oauth/*”, “/kapis/config.kubesphere.io/*”, “/kapis/version”, “/kapis/metrics”}，这些API不经过rbac代码验证。其他API请求都会filters.WithAuthorization进行RBAC权限验证。经过权限验证过后，k8s的请求是通过filters.WithKubeAPIServer进行info.IsKubernetesRequest判断，kubernetes的请求会代理转发的k8s，这里请求k8s资源时，是使用的集群的kubeconfig来请求的。