Check for RBAC before starting tunnel controllers

pkg/agent/tunnel/tunnel.go

@@ -20,6 +20,7 @@ import (
 	"github.com/rancher/remotedialer"
 	"github.com/sirupsen/logrus"
 	"github.com/yl2chen/cidranger"
+	authorizationv1 "k8s.io/api/authorization/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -86,6 +87,14 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 		if err := util.WaitForAPIServerReady(ctx, config.AgentConfig.KubeConfigKubelet, util.DefaultAPIServerReadyTimeout); err != nil {
 			logrus.Fatalf("Tunnel watches failed to wait for apiserver ready: %v", err)
 		}
+		if err := util.WaitForRBACReady(ctx, config.AgentConfig.KubeConfigK3sController, util.DefaultAPIServerReadyTimeout, authorizationv1.ResourceAttributes{
+			Namespace: metav1.NamespaceDefault,
+			Verb:      "list",
+			Resource:  "endpoints",
+		}, ""); err != nil {
+			logrus.Fatalf("Tunnel watches failed to wait for RBAC: %v", err)
+		}
+
 		close(apiServerReady)
 	}()
 
@@ -114,7 +123,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 			proxy.SetSupervisorDefault(addresses[0])
 			proxy.Update(addresses)
 		} else {
-			if endpoint, _ := client.CoreV1().Endpoints("default").Get(ctx, "kubernetes", metav1.GetOptions{}); endpoint != nil {
+			if endpoint, _ := client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); endpoint != nil {
 				if addresses := util.GetAddresses(endpoint); len(addresses) > 0 {
 					proxy.Update(addresses)
 				}

pkg/daemons/control/server.go

@@ -21,9 +21,6 @@ import (
 	"github.com/sirupsen/logrus"
 	authorizationv1 "k8s.io/api/authorization/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	authorizationv1client "k8s.io/client-go/kubernetes/typed/authorization/v1"
-	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/kubernetes/pkg/kubeapiserver/authorizer/modes"
 	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
 
@@ -367,37 +364,12 @@ func cloudControllerManager(ctx context.Context, cfg *config.Control) error {
 // If the CCM RBAC changes, the ResourceAttributes checked for by this function should
 // be modified to check for the most recently added privilege.
 func checkForCloudControllerPrivileges(ctx context.Context, runtime *config.ControlRuntime, timeout time.Duration) error {
-	restConfig, err := clientcmd.BuildConfigFromFlags("", runtime.KubeConfigAdmin)
-	if err != nil {
-		return err
-	}
-	authClient, err := authorizationv1client.NewForConfig(restConfig)
-	if err != nil {
-		return err
-	}
-	sar := &authorizationv1.SubjectAccessReview{
-		Spec: authorizationv1.SubjectAccessReviewSpec{
-			User: version.Program + "-cloud-controller-manager",
-			ResourceAttributes: &authorizationv1.ResourceAttributes{
-				Namespace: metav1.NamespaceSystem,
-				Verb:      "*",
-				Resource:  "daemonsets",
-				Group:     "apps",
-			},
-		},
-	}
-
-	err = wait.PollImmediate(time.Second, timeout, func() (bool, error) {
-		r, err := authClient.SubjectAccessReviews().Create(ctx, sar, metav1.CreateOptions{})
-		if err != nil {
-			return false, err
-		}
-		if r.Status.Allowed {
-			return true, nil
-		}
-		return false, nil
-	})
-	return err
+	return util.WaitForRBACReady(ctx, runtime.KubeConfigAdmin, timeout, authorizationv1.ResourceAttributes{
+		Namespace: metav1.NamespaceSystem,
+		Verb:      "*",
+		Resource:  "daemonsets",
+		Group:     "apps",
+	}, version.Program+"-cloud-controller-manager")
 }
 
 func waitForAPIServerHandlers(ctx context.Context, runtime *config.ControlRuntime) {

pkg/util/api.go

@@ -13,11 +13,14 @@ import (
 	"github.com/rancher/wrangler/pkg/merr"
 	"github.com/rancher/wrangler/pkg/schemes"
 	"github.com/sirupsen/logrus"
+	authorizationv1 "k8s.io/api/authorization/v1"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	clientset "k8s.io/client-go/kubernetes"
+	authorizationv1client "k8s.io/client-go/kubernetes/typed/authorization/v1"
 	coregetter "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
@@ -101,6 +104,84 @@ func WaitForAPIServerReady(ctx context.Context, kubeconfigPath string, timeout t
 	return nil
 }
 
+type genericAccessReviewRequest func(context.Context) (*authorizationv1.SubjectAccessReviewStatus, error)
+
+// WaitForRBACReady polls an AccessReview request until it returns an allowed response. If the user
+// and group are empty, it uses SelfSubjectAccessReview, otherwise SubjectAccessReview is used.  It
+// will return an error if the timeout expires, or nil if the SubjectAccessReviewStatus indicates
+// the access would be allowed.
+func WaitForRBACReady(ctx context.Context, kubeconfigPath string, timeout time.Duration, ra authorizationv1.ResourceAttributes, user string, groups ...string) error {
+	var lastErr error
+	restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
+	if err != nil {
+		return err
+	}
+	authClient, err := authorizationv1client.NewForConfig(restConfig)
+	if err != nil {
+		return err
+	}
+
+	var reviewFunc genericAccessReviewRequest
+	if len(user) == 0 && len(groups) == 0 {
+		reviewFunc = selfSubjectAccessReview(authClient, ra)
+	} else {
+		reviewFunc = subjectAccessReview(authClient, ra, user, groups)
+	}
+
+	err = wait.PollImmediateWithContext(ctx, time.Second, timeout, func(ctx context.Context) (bool, error) {
+		status, rerr := reviewFunc(ctx)
+		if rerr != nil {
+			lastErr = rerr
+			return false, nil
+		}
+		if status.Allowed {
+			return true, nil
+		}
+		lastErr = errors.New(status.Reason)
+		return false, nil
+	})
+
+	if err != nil {
+		return merr.NewErrors(err, lastErr)
+	}
+
+	return nil
+}
+
+// selfSubjectAccessReview returns a function that makes SelfSubjectAccessReview requests using the
+// provided client and attributes, returning a status or error.
+func selfSubjectAccessReview(authClient *authorizationv1client.AuthorizationV1Client, ra authorizationv1.ResourceAttributes) genericAccessReviewRequest {
+	return func(ctx context.Context) (*authorizationv1.SubjectAccessReviewStatus, error) {
+		r, err := authClient.SelfSubjectAccessReviews().Create(ctx, &authorizationv1.SelfSubjectAccessReview{
+			Spec: authorizationv1.SelfSubjectAccessReviewSpec{
+				ResourceAttributes: &ra,
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			return nil, err
+		}
+		return &r.Status, nil
+	}
+}
+
+// subjectAccessReview returns a function that makes SubjectAccessReview requests using the
+// provided client, attributes, user, and group, returning a status or error.
+func subjectAccessReview(authClient *authorizationv1client.AuthorizationV1Client, ra authorizationv1.ResourceAttributes, user string, groups []string) genericAccessReviewRequest {
+	return func(ctx context.Context) (*authorizationv1.SubjectAccessReviewStatus, error) {
+		r, err := authClient.SubjectAccessReviews().Create(ctx, &authorizationv1.SubjectAccessReview{
+			Spec: authorizationv1.SubjectAccessReviewSpec{
+				ResourceAttributes: &ra,
+				User:               user,
+				Groups:             groups,
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			return nil, err
+		}
+		return &r.Status, nil
+	}
+}
+
 func BuildControllerEventRecorder(k8s clientset.Interface, controllerName, namespace string) record.EventRecorder {
 	logrus.Infof("Creating %s event broadcaster", controllerName)
 	eventBroadcaster := record.NewBroadcaster()