fix: add reload tke and ldap idp store periodically (#311)

* fix: add reload tke and ldap idp store periodically

* fix: wait sync rules before start auth controller

cmd/tke-auth-api/app/config/config.go

@@ -326,12 +326,12 @@ func setupCasbinEnforcer(authorizationOptions *options.AuthorizationOptions) (*c
 
 func setupDefaultConnector(versionInformers versionedinformers.SharedInformerFactory, auth *options.AuthOptions) error {
  log.Info("setup tke local connector", log.Any("tenantID", auth.InitTenantID))
- if _, ok := identityprovider.IdentityProvidersStore[auth.InitTenantID]; !ok {
+ if _, ok := identityprovider.GetIdentityProvider(auth.InitTenantID); !ok {
  defaultIDP, err := local.NewDefaultIdentityProvider(auth.InitTenantID, auth.InitIDPAdmins, versionInformers)
  if err != nil {
  return nil
  }
- identityprovider.IdentityProvidersStore[auth.InitTenantID] = defaultIDP
+ identityprovider.SetIdentityProvider(auth.InitTenantID, defaultIDP)
  }
 
  return nil
@@ -357,8 +357,8 @@ func setupLDAPConnector(auth *options.AuthOptions) error {
  return err
  }
 
- if _, ok := identityprovider.IdentityProvidersStore[auth.InitTenantID]; !ok {
- identityprovider.IdentityProvidersStore[auth.InitTenantID] = idp
+ if _, ok := identityprovider.GetIdentityProvider(auth.InitTenantID); !ok {
+ identityprovider.SetIdentityProvider(auth.InitTenantID, idp)
  }
 
  return nil

cmd/tke-auth-controller/app/controller.go

@@ -19,12 +19,14 @@
 package app
 
 import (
+ "fmt"
  "net/http"
  "time"
 
  "k8s.io/apimachinery/pkg/util/sets"
  "k8s.io/apimachinery/pkg/util/wait"
  "k8s.io/apiserver/pkg/server/mux"
+ "k8s.io/client-go/tools/cache"
  "tkestack.io/tke/pkg/util/log"
 )
 
@@ -58,6 +60,12 @@ func NewControllerInitializers() map[string]InitFunc {
 
 // StartControllers to start the controller.
 func StartControllers(ctx ControllerContext, controllers map[string]InitFunc, unsecuredMux *mux.PathRecorderMux) error {
+ go ctx.InformerFactory.Auth().V1().Rules().Informer().Run(ctx.Stop)
+ if ok := cache.WaitForCacheSync(ctx.Stop, ctx.InformerFactory.Auth().V1().Rules().Informer().HasSynced); !ok {
+ return fmt.Errorf("failed to wait for rules caches to sync")
+ }
+ _ = ctx.Enforcer.LoadPolicy()
+
  for controllerName, initFn := range controllers {
  if !ctx.IsControllerEnabled(controllerName) {
  log.Warnf("%q is disabled", controllerName)

pkg/auth/authentication/oidc/identityprovider/hook_oidc.go

@@ -30,8 +30,8 @@ import (
  "gopkg.in/square/go-jose.v2"
  "k8s.io/apimachinery/pkg/api/errors"
  genericapiserver "k8s.io/apiserver/pkg/server"
- authinternalclient "tkestack.io/tke/api/client/clientset/internalversion/typed/auth/internalversion"
 
+ authinternalclient "tkestack.io/tke/api/client/clientset/internalversion/typed/auth/internalversion"
  "tkestack.io/tke/pkg/apiserver/authentication/authenticator/oidc"
  "tkestack.io/tke/pkg/auth/authentication/authenticator"
  "tkestack.io/tke/pkg/util/log"
@@ -65,10 +65,11 @@ func NewDexHookHandler(ctx context.Context, authClient authinternalclient.AuthIn
 
 // PostStartHook provides a function that is called after the server has started.
 func (d *dexHookHandler) PostStartHook() (string, genericapiserver.PostStartHookFunc, error) {
- return "create-dex-server", func(_ genericapiserver.PostStartHookContext) error {
+ return "create-dex-server", func(context genericapiserver.PostStartHookContext) error {
  log.Info("start create dex server")
  // Ensure all identity providers defined exists in dex.
- for tenantID, idp := range IdentityProvidersStore {
+ idpMap := GetAllIdentityProviderMap()
+ for tenantID, idp := range idpMap {
  idp, err := idp.Store()
  if err != nil {
  log.Errorf("Get connector for tenant failed", log.String("tenantID", tenantID), log.Err(err))

pkg/auth/authentication/oidc/identityprovider/interface.go

@@ -20,6 +20,7 @@ package identityprovider
 
 import (
  "context"
+ "sync"
 
  "github.com/dexidp/dex/connector"
  dexlog "github.com/dexidp/dex/pkg/log"
@@ -38,8 +39,43 @@ type IdentityProvider interface {
  Store() (*auth.IdentityProvider, error)
 }
 
-// IdentityProvidersStore represents identity providers for every tenantID.
-var IdentityProvidersStore = make(map[string]IdentityProvider)
+var (
+ // identityProvidersStore represents identity providers for every tenantID.
+ identityProvidersStore = make(map[string]IdentityProvider)
+ mutex sync.RWMutex
+)
+
+func GetIdentityProvider(tenantID string) (IdentityProvider, bool) {
+ mutex.RLock()
+ defer mutex.RUnlock()
+
+ idp, ok := identityProvidersStore[tenantID]
+ return idp, ok
+}
+
+func SetIdentityProvider(tenantID string, provider IdentityProvider) {
+ mutex.Lock()
+ defer mutex.Unlock()
+ identityProvidersStore[tenantID] = provider
+}
+
+func DeleteIdentityProvider(tenantID string) {
+ mutex.Lock()
+ defer mutex.Unlock()
+ delete(identityProvidersStore, tenantID)
+}
+
+func GetAllIdentityProviderMap() map[string]IdentityProvider {
+ mutex.RLock()
+ defer mutex.RUnlock()
+
+ newMap := make(map[string]IdentityProvider)
+ for k, v := range identityProvidersStore {
+ newMap[k] = v
+ }
+
+ return newMap
+}
 
 // UserGetter is an object that can get the user that match the provided field and label criteria.
 type UserGetter interface {

pkg/auth/authentication/oidc/identityprovider/ldap/hook_ldap.go

@@ -20,10 +20,12 @@ package ldap
 
 import (
  "encoding/json"
+ "time"
 
  dexldap "github.com/dexidp/dex/connector/ldap"
  v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/fields"
+ "k8s.io/apimachinery/pkg/util/wait"
  genericapiserver "k8s.io/apiserver/pkg/server"
 
  authinternalclient "tkestack.io/tke/api/client/clientset/internalversion/typed/auth/internalversion"
@@ -44,31 +46,40 @@ func NewLdapHookHandler(authClient authinternalclient.AuthInterface) genericapis
 
 func (d *ldapHookHandler) PostStartHook() (string, genericapiserver.PostStartHookFunc, error) {
  return "load-ldap-idp", func(context genericapiserver.PostStartHookContext) error {
- tenantUserSelector := fields.AndSelectors(
- fields.OneTermEqualSelector("spec.type", ConnectorType),
- )
- conns, err := d.authClient.IdentityProviders().List(v1.ListOptions{FieldSelector: tenantUserSelector.String()})
- if err != nil {
- return err
- }
 
- for _, conn := range conns.Items {
- var ldapConfig dexldap.Config
- err = json.Unmarshal([]byte(conn.Spec.Config), &ldapConfig)
+ go wait.JitterUntil(func() {
+ tenantUserSelector := fields.AndSelectors(
+ fields.OneTermEqualSelector("spec.type", ConnectorType),
+ )
+ conns, err := d.authClient.IdentityProviders().List(v1.ListOptions{FieldSelector: tenantUserSelector.String()})
  if err != nil {
- log.Error("Unmarshal idp config failed", log.String("idp", conn.Spec.Name), log.Err(err))
- continue
+ log.Error("List ldap idp from registry failed", log.Err(err))
+ return
  }
 
- idp, err := NewLDAPIdentityProvider(ldapConfig, conn.Spec.Administrators, conn.Name)
- if err != nil {
- log.Error("NewLDAPIdentityProvider failed", log.String("idp", conn.Spec.Name), log.Err(err))
- continue
+ for _, conn := range conns.Items {
+ if _, ok := identityprovider.GetIdentityProvider(conn.Name); ok {
+ continue
+ }
+
+ var ldapConfig dexldap.Config
+ err = json.Unmarshal([]byte(conn.Spec.Config), &ldapConfig)
+ if err != nil {
+ log.Error("Unmarshal idp config failed", log.String("idp", conn.Spec.Name), log.Err(err))
+ continue
+ }
+
+ idp, err := NewLDAPIdentityProvider(ldapConfig, conn.Spec.Administrators, conn.Name)
+ if err != nil {
+ log.Error("NewLDAPIdentityProvider failed", log.String("idp", conn.Spec.Name), log.Err(err))
+ continue
+ }
+
+ identityprovider.SetIdentityProvider(conn.Name, idp)
+ log.Info("load ldap identity provider successfully", log.String("idp", conn.Name))
  }
 
- identityprovider.IdentityProvidersStore[conn.Name] = idp
- log.Info("load ldap identity provider successfully", log.String("idp", conn.Name))
- }
+ }, 30*time.Second, 0.0, false, context.StopCh)
 
  return nil
  }, nil

pkg/auth/authentication/oidc/identityprovider/local/hook_local.go

@@ -19,17 +19,19 @@
 package local
 
 import (
+ "time"
+
  v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/fields"
- "k8s.io/client-go/tools/cache"
- "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider"
- "tkestack.io/tke/pkg/util/log"
-
+ "k8s.io/apimachinery/pkg/util/wait"
  genericapiserver "k8s.io/apiserver/pkg/server"
+ "k8s.io/client-go/tools/cache"
 
  authinternalclient "tkestack.io/tke/api/client/clientset/internalversion/typed/auth/internalversion"
  versionedinformers "tkestack.io/tke/api/client/informers/externalversions"
  authv1informer "tkestack.io/tke/api/client/informers/externalversions/auth/v1"
+ "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider"
+ "tkestack.io/tke/pkg/util/log"
 )
 
 type localHookHandler struct {
@@ -56,24 +58,31 @@ func (d *localHookHandler) PostStartHook() (string, genericapiserver.PostStartHo
  log.Error("Failed to wait for local identity and group caches to sync")
  }
 
- tenantUserSelector := fields.AndSelectors(
- fields.OneTermEqualSelector("spec.type", ConnectorType),
- )
- conns, err := d.authClient.IdentityProviders().List(v1.ListOptions{FieldSelector: tenantUserSelector.String()})
- if err != nil {
- return err
- }
-
- for _, conn := range conns.Items {
- idp, err := NewDefaultIdentityProvider(conn.Name, conn.Spec.Administrators, d.versionedInformers)
+ go wait.JitterUntil(func() {
+ tenantUserSelector := fields.AndSelectors(
+ fields.OneTermEqualSelector("spec.type", ConnectorType),
+ )
+ conns, err := d.authClient.IdentityProviders().List(v1.ListOptions{FieldSelector: tenantUserSelector.String()})
  if err != nil {
- log.Error("NewDefaultIdentityProvider failed", log.String("idp", conn.Spec.Name), log.Err(err))
- continue
+ log.Error("List default idp from registry failed", log.Err(err))
+ return
  }
 
- identityprovider.IdentityProvidersStore[conn.Name] = idp
- log.Info("load local identity provider successfully", log.String("idp", conn.Name))
- }
+ for _, conn := range conns.Items {
+ if _, ok := identityprovider.GetIdentityProvider(conn.Name); ok {
+ continue
+ }
+
+ idp, err := NewDefaultIdentityProvider(conn.Name, conn.Spec.Administrators, d.versionedInformers)
+ if err != nil {
+ log.Error("NewDefaultIdentityProvider failed", log.String("idp", conn.Spec.Name), log.Err(err))
+ continue
+ }
+
+ identityprovider.SetIdentityProvider(conn.Name, idp)
+ log.Info("load local identity provider successfully", log.String("idp", conn.Name))
+ }
+ }, 30*time.Second, 0.0, false, context.StopCh)
 
  return nil
  }, nil

pkg/auth/registry/group/storage/storage.go

@@ -94,7 +94,7 @@ func (r *REST) Get(ctx context.Context, name string, options *metav1.GetOptions)
  tenantID, name = util.ParseTenantAndName(name)
  }
 
- idp, ok := identityprovider.IdentityProvidersStore[tenantID]
+ idp, ok := identityprovider.GetIdentityProvider(tenantID)
  if !ok {
  log.Error("Tenant has no related identity providers", log.String("tenantID", tenantID))
  return nil, apierrors.NewNotFound(auth.Resource("group"), name)
@@ -119,7 +119,7 @@ func (r *REST) List(ctx context.Context, options *metainternal.ListOptions) (run
  return &auth.GroupList{}, nil
  }
  }
- idp, ok := identityprovider.IdentityProvidersStore[tenantID]
+ idp, ok := identityprovider.GetIdentityProvider(tenantID)
  if !ok {
  log.Error("Tenant has no related identity providers", log.String("tenantID", tenantID))
  return &auth.GroupList{}, nil

pkg/auth/registry/identityprovider/storage/storage.go

@@ -22,22 +22,22 @@ import (
  "context"
  "encoding/json"
 
- "k8s.io/apimachinery/pkg/api/errors"
- "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider/ldap"
- "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider/local"
- "tkestack.io/tke/pkg/util/log"
-
  dexldap "github.com/dexidp/dex/connector/ldap"
+ "k8s.io/apimachinery/pkg/api/errors"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/runtime"
  "k8s.io/apiserver/pkg/registry/generic"
  "k8s.io/apiserver/pkg/registry/generic/registry"
  "k8s.io/apiserver/pkg/registry/rest"
+
  "tkestack.io/tke/api/auth"
  authinternalclient "tkestack.io/tke/api/client/clientset/internalversion/typed/auth/internalversion"
  versionedinformers "tkestack.io/tke/api/client/informers/externalversions"
  oidcidp "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider"
+ "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider/ldap"
+ "tkestack.io/tke/pkg/auth/authentication/oidc/identityprovider/local"
  "tkestack.io/tke/pkg/auth/registry/identityprovider"
+ "tkestack.io/tke/pkg/util/log"
 )
 
 // Storage includes storage for signing keys and all sub resources.
@@ -79,6 +79,13 @@ type REST struct {
  versionedInformers versionedinformers.SharedInformerFactory
 }
 
+var _ rest.ShortNamesProvider = &REST{}
+
+// ShortNames implements the ShortNamesProvider interface. Returns a list of short names for a resource.
+func (r *REST) ShortNames() []string {
+ return []string{"idp"}
+}
+
 func (r *REST) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
  var idp oidcidp.IdentityProvider
  var err error
@@ -108,7 +115,7 @@ func (r *REST) Create(ctx context.Context, obj runtime.Object, createValidation
 
  result, err := r.Store.Create(ctx, obj, createValidation, options)
  if err == nil && idp != nil {
- oidcidp.IdentityProvidersStore[idpObj.Name] = idp
+ oidcidp.SetIdentityProvider(idpObj.Name, idp)
  }
 
  return result, err

pkg/auth/registry/user/storage/storage.go

@@ -22,7 +22,6 @@ import (
  "context"
 
  "github.com/casbin/casbin/v2"
-
  apierrors "k8s.io/apimachinery/pkg/api/errors"
  metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -94,7 +93,7 @@ func (r *REST) Get(ctx context.Context, name string, options *metav1.GetOptions)
  tenantID, name = util.ParseTenantAndName(name)
  }
 
- idp, ok := identityprovider.IdentityProvidersStore[tenantID]
+ idp, ok := identityprovider.GetIdentityProvider(tenantID)
  if !ok {
  log.Error("Tenant has no related identity providers", log.String("tenantID", tenantID))
  return nil, apierrors.NewNotFound(auth.Resource("user"), name)
@@ -119,7 +118,7 @@ func (r *REST) List(ctx context.Context, options *metainternal.ListOptions) (run
  }
  }
 
- idp, ok := identityprovider.IdentityProvidersStore[tenantID]
+ idp, ok := identityprovider.GetIdentityProvider(tenantID)
  if !ok {
  log.Error("Tenant has no related identity providers", log.String("tenantID", tenantID))
  return &auth.UserList{}, nil