diff --git a/controllers/auth_config_controller.go b/controllers/auth_config_controller.go index 888f9ebf8..0d7e7a37e 100644 --- a/controllers/auth_config_controller.go +++ b/controllers/auth_config_controller.go @@ -62,6 +62,7 @@ import ( const ( failedToCleanConfig = "failed to clean up all asynchronous workers" + failedToStartConfig = "failed to start all asynchronous workers" AuthConfigsReadyzSubpath = "authconfigs" ) @@ -187,12 +188,10 @@ func (r *AuthConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) ) span.AddEvent("authconfig.found") - // clean all async workers of the config, i.e. shuts down channels and goroutines - if err := r.cleanConfigs(ctx, resourceId); err != nil { - logger.Error(err, failedToCleanConfig) - span.RecordError(err) - } - + // translate first and only then swap: evaluators are built without starting any of their + // background workers, so a translation that fails leaves the config currently in the index + // untouched and still refreshing, instead of tearing it down before we know the new one is + // even valid translatedAuthConfig, err := r.translateAuthConfig(log.IntoContext(ctx, logger), &authConfig) if err != nil { r.StatusReport.Set(resourceId, api.StatusReasonInvalidResource, err.Error(), []string{}) @@ -201,6 +200,13 @@ func (r *AuthConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } + // the new config is good, so the one it replaces can be cleaned up, i.e. shuts down + // channels and goroutines + if err := r.cleanConfigs(ctx, resourceId); err != nil { + logger.Error(err, failedToCleanConfig) + span.RecordError(err) + } + // delete unused hosts from the index unusedHosts := utils.SubtractSlice(r.Index.FindKeys(resourceId), authConfig.Spec.Hosts) if len(unusedHosts) > 0 { @@ -227,6 +233,17 @@ func (r *AuthConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, err } + // only now, and only for a config that actually made it into the index. an authconfig whose + // hosts are all taken by another one translates just fine and is never indexed, so anything + // started before this point would run for a config that no request can ever reach and that + // cleanConfigs() would never find again + if len(linkedHosts) > 0 { + if err := r.startConfigs(ctx, translatedAuthConfig); err != nil { + logger.Error(err, failedToStartConfig) + span.RecordError(err) + } + } + span.SetAttributes( attribute.Int("authconfig.linked_hosts_count", len(linkedHosts)), attribute.Int("authconfig.loose_hosts_count", len(looseHosts)), @@ -245,6 +262,22 @@ func (r *AuthConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) return ctrl.Result{}, nil } +// startConfigs kicks off the asynchronous workers of a config that has been successfully +// translated and indexed, e.g. the openid connect configuration and external policy refreshers +func (r *AuthConfigReconciler) startConfigs(ctx context.Context, authConfig *evaluators.AuthConfig) error { + ctx, span := trace.NewSpan(ctx, "authconfig", "authconfig.start_configs") + defer span.End() + + if err := authConfig.Start(ctx); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "failed to start authconfig") + return err + } + + span.AddEvent("authconfig.started") + return nil +} + func (r *AuthConfigReconciler) cleanConfigs(ctx context.Context, resourceId string) error { ctx, span := trace.NewSpan(ctx, "authconfig", "authconfig.clean_configs") defer span.End() diff --git a/controllers/auth_config_controller_test.go b/controllers/auth_config_controller_test.go index eff155847..d788f5839 100644 --- a/controllers/auth_config_controller_test.go +++ b/controllers/auth_config_controller_test.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "os" + goruntime "runtime" "testing" + "time" api "github.com/kuadrant/authorino/api/v1beta3" "github.com/kuadrant/authorino/pkg/evaluators" @@ -446,3 +448,192 @@ func BenchmarkReconcileAuthConfig(b *testing.B) { b.StopTimer() assert.NilError(b, err) } + +func newTestAuthConfigWithRefresher() api.AuthConfig { + return api.AuthConfig{ + TypeMeta: metav1.TypeMeta{Kind: "AuthConfig", APIVersion: "authorino.kuadrant.io/v1beta3"}, + ObjectMeta: metav1.ObjectMeta{Name: "auth-config-1", Namespace: "authorino"}, + Spec: api.AuthConfigSpec{ + Hosts: []string{"echo-api"}, + Authentication: map[string]api.AuthenticationSpec{ + "keycloak": { + AuthenticationMethodSpec: api.AuthenticationMethodSpec{ + Jwt: &api.JwtAuthenticationSpec{ + IssuerUrl: "http://127.0.0.1:9001/auth/realms/demo", + TTL: 60, // starts the background OIDC refresher worker + }, + }, + }, + }, + }, + } +} + +// breaks translateAuthConfig() the same way a rotated or deleted credentialsRef does in a cluster. +// the metadata phase is translated after the identity phase, so the identity refresher is always +// built before the failure - authentication is a map and its iteration order is not deterministic +func breakTranslation(t *testing.T, k8sClient client.WithWatch, name types.NamespacedName) { + t.Helper() + authConfig := &api.AuthConfig{} + assert.NilError(t, k8sClient.Get(context.Background(), name, authConfig)) + authConfig.Spec.Metadata = map[string]api.MetadataSpec{ + "uma": { + MetadataMethodSpec: api.MetadataMethodSpec{ + Uma: &api.UmaMetadataSpec{ + Endpoint: "http://127.0.0.1:9001/auth/realms/demo", + Credentials: &v1.LocalObjectReference{Name: "no-such-secret"}, + }, + }, + }, + } + assert.NilError(t, k8sClient.Update(context.Background(), authConfig)) +} + +// A reconcile that fails in translateAuthConfig() returns before addToIndex(), so the config it +// just cleaned up is still the one in the index. The requeue then cleans that very same instance +// again, which used to close an already-closed channel and take the whole process down. +func TestReconcileCleansTheSameIndexedConfigTwiceWithoutPanicking(t *testing.T) { + authConfigIndex := index.NewIndex() + authConfig := newTestAuthConfigWithRefresher() + k8sClient := newTestK8sClient(&authConfig) + reconciler := newTestAuthConfigReconciler(k8sClient, authConfigIndex) + name := types.NamespacedName{Name: authConfig.Name, Namespace: authConfig.Namespace} + req := reconcile.Request{NamespacedName: name} + + _, err := reconciler.Reconcile(context.Background(), req) + assert.NilError(t, err) + assert.Check(t, authConfigIndex.Get("echo-api") != nil) + + breakTranslation(t, k8sClient, name) + + // reconcile #1: cleans up the indexed config, fails to translate, requeues + _, err = reconciler.Reconcile(context.Background(), req) + assert.ErrorContains(t, err, "no-such-secret") + assert.Check(t, authConfigIndex.Get("echo-api") != nil, "the last known good config should stay in the index") + + // reconcile #2: the requeue, cleaning up that same instance all over again + _, err = reconciler.Reconcile(context.Background(), req) + assert.ErrorContains(t, err, "no-such-secret") + assert.Check(t, authConfigIndex.Get("echo-api") != nil) +} + +// dropIdentity removes the identity evaluator from the resource. Kept apart from breakTranslation +// on purpose: the tests that check nothing gets orphaned need an identity in the failing config, +// or there would be no refresher to orphan in the first place and they would pass for free. +func dropIdentity(t *testing.T, k8sClient client.WithWatch, name types.NamespacedName) { + t.Helper() + authConfig := &api.AuthConfig{} + assert.NilError(t, k8sClient.Get(context.Background(), name, authConfig)) + authConfig.Spec.Authentication = nil + assert.NilError(t, k8sClient.Update(context.Background(), authConfig)) +} + +// A translation that fails must not start anything: the evaluators it built never reach the index, +// so nothing would ever clean them up and a persistently failing reconcile is requeued forever. +func TestReconcileStartsNoWorkersWhenTranslationFails(t *testing.T) { + authConfig := newTestAuthConfigWithRefresher() + k8sClient := newTestK8sClient(&authConfig) + reconciler := newTestAuthConfigReconciler(k8sClient, index.NewIndex()) + name := types.NamespacedName{Name: authConfig.Name, Namespace: authConfig.Namespace} + req := reconcile.Request{NamespacedName: name} + + breakTranslation(t, k8sClient, name) + + _, err := reconciler.Reconcile(context.Background(), req) + assert.ErrorContains(t, err, "no-such-secret") + time.Sleep(200 * time.Millisecond) + goruntime.GC() + before := goruntime.NumGoroutine() + + const reconciles = 20 + for i := 0; i < reconciles; i++ { + _, err := reconciler.Reconcile(context.Background(), req) + assert.ErrorContains(t, err, "no-such-secret") + } + + time.Sleep(500 * time.Millisecond) + goruntime.GC() + leaked := goruntime.NumGoroutine() - before + assert.Check(t, leaked < reconciles/2, "leaked %d goroutines over %d failed reconciles", leaked, reconciles) +} + +// An authconfig whose hosts are all taken by another one translates cleanly but is never indexed, +// and addToIndex() reports that with an empty linkedHosts and no error at all. Nothing may be +// started for it either, for exactly the same reason. +func TestReconcileStartsNoWorkersWhenNoHostIsLinked(t *testing.T) { + authConfigIndex := index.NewIndex() + + winner := newTestAuthConfigWithRefresher() + winner.Name = "auth-config-winner" + loser := newTestAuthConfigWithRefresher() + loser.Name = "auth-config-loser" + + k8sClient := newTestK8sClient(&winner, &loser) + reconciler := newTestAuthConfigReconciler(k8sClient, authConfigIndex) + winnerReq := reconcile.Request{NamespacedName: types.NamespacedName{Name: winner.Name, Namespace: winner.Namespace}} + loserReq := reconcile.Request{NamespacedName: types.NamespacedName{Name: loser.Name, Namespace: loser.Namespace}} + + _, err := reconciler.Reconcile(context.Background(), winnerReq) + assert.NilError(t, err) + + _, err = reconciler.Reconcile(context.Background(), loserReq) + assert.NilError(t, err) // a host collision is reported on the status, it is not a reconcile error + time.Sleep(200 * time.Millisecond) + goruntime.GC() + before := goruntime.NumGoroutine() + + const reconciles = 20 + for i := 0; i < reconciles; i++ { + _, err := reconciler.Reconcile(context.Background(), loserReq) + assert.NilError(t, err) + } + + time.Sleep(500 * time.Millisecond) + goruntime.GC() + leaked := goruntime.NumGoroutine() - before + assert.Check(t, leaked < reconciles/2, "leaked %d goroutines over %d reconciles of an unlinked authconfig", leaked, reconciles) +} + +// The config in the index is only torn down once its replacement is known to be good, so a failed +// translation leaves the last known good config both indexed AND still refreshing. +func TestReconcileKeepsTheIndexedConfigRunningWhenTranslationFails(t *testing.T) { + authConfigIndex := index.NewIndex() + authConfig := newTestAuthConfigWithRefresher() + k8sClient := newTestK8sClient(&authConfig) + reconciler := newTestAuthConfigReconciler(k8sClient, authConfigIndex) + name := types.NamespacedName{Name: authConfig.Name, Namespace: authConfig.Namespace} + req := reconcile.Request{NamespacedName: name} + + _, err := reconciler.Reconcile(context.Background(), req) + assert.NilError(t, err) + + indexed := authConfigIndex.Get("echo-api") + assert.Check(t, indexed != nil) + assert.Check(t, refresherRunning(indexed), "the indexed config should be refreshing once it is reconciled") + + breakTranslation(t, k8sClient, name) + // and take the identity away, so the version of the resource that fails to translate has no + // jwt evaluator of its own. a refresher still running below can then only have come from the + // config that was indexed before it, rather than from a config this reconcile put there + dropIdentity(t, k8sClient, name) + + _, err = reconciler.Reconcile(context.Background(), req) + assert.ErrorContains(t, err, "no-such-secret") + + stillIndexed := authConfigIndex.Get("echo-api") + assert.Check(t, stillIndexed != nil, "the last known good config should stay in the index") + assert.Check(t, refresherRunning(stillIndexed), "the refresher can only be the one from the previously indexed config, and it should still be running") +} + +func refresherRunning(authConfig *evaluators.AuthConfig) bool { + for _, evaluator := range authConfig.IdentityConfigs { + idConfig, ok := evaluator.(*evaluators.IdentityConfig) + if !ok || idConfig.JWTAuthentication == nil { + continue + } + if idConfig.JWTAuthentication.RefresherRunning() { + return true + } + } + return false +} diff --git a/pkg/auth/auth.go b/pkg/auth/auth.go index b70268840..88a437b9e 100644 --- a/pkg/auth/auth.go +++ b/pkg/auth/auth.go @@ -28,6 +28,15 @@ type AuthConfigEvaluator interface { Call(AuthPipeline, context.Context) (interface{}, error) } +type AuthConfigStarter interface { + // Start is used to give the different auth configs a chance to kick off anything that should + // only be running once the config is known to be valid and reachable, e.g. background workers + // that refresh remote state. Evaluators must not start those from their constructors: a config + // can fail to translate or lose every one of its hosts to another AuthConfig, and in neither + // case does it ever reach the index for anything to clean it up again. + Start(context.Context) error +} + type AuthConfigCleaner interface { // Clean is used to give the different auth configs chance to clean up anything internal to that config Clean(context.Context) error diff --git a/pkg/evaluators/authorization.go b/pkg/evaluators/authorization.go index 765d0d2c1..b1485aadf 100644 --- a/pkg/evaluators/authorization.go +++ b/pkg/evaluators/authorization.go @@ -120,6 +120,24 @@ func (config *AuthorizationConfig) MetricsEnabled() bool { return config.Metrics } +// impl:AuthConfigStarter + +func (config *AuthorizationConfig) Start(ctx context.Context) error { + if starter := config.getStarter(); starter != nil { + return starter.Start(log.IntoContext(ctx, log.FromContext(ctx).WithName("authorization"))) + } + return nil +} + +func (config *AuthorizationConfig) getStarter() auth.AuthConfigStarter { + switch { + case config.OPA != nil: + return config.OPA + default: + return nil + } +} + // impl:AuthConfigCleaner func (config *AuthorizationConfig) Clean(ctx context.Context) error { diff --git a/pkg/evaluators/authorization/opa.go b/pkg/evaluators/authorization/opa.go index 3f3f0597c..ef5b7a07d 100644 --- a/pkg/evaluators/authorization/opa.go +++ b/pkg/evaluators/authorization/opa.go @@ -69,22 +69,33 @@ func NewOPAAuthorization(policyName string, rego string, externalSource *OPAExte } o := &OPA{ - ExternalSource: externalSource, - AllValues: allValues, - regoVersion: regoVersion, - policyName: policyName, - policyUID: generatePolicyUID(policyName, rego, nonce), - opaContext: context.TODO(), + ExternalSource: externalSource, + AllValues: allValues, + regoVersion: regoVersion, + policyName: policyName, + policyUID: generatePolicyUID(policyName, rego, nonce), + opaContext: context.TODO(), + pullFromRegistry: pullFromRegistry, } if _, err := o.updateRego(rego, ctx, true); err != nil { return nil, err - } else { - if pullFromRegistry { - externalSource.setupRefresher(log.IntoContext(ctx, logger), o) - } - return o, nil } + + return o, nil +} + +// Start kicks off the refresh of the policy pulled from an external registry. The policy itself is +// downloaded and precompiled by the constructor, so an unreachable registry still fails the +// translation of the authconfig; only the periodic refresh waits until the config is indexed. +// impl: auth.AuthConfigStarter +func (opa *OPA) Start(ctx context.Context) error { + if !opa.pullFromRegistry || opa.ExternalSource == nil { + return nil + } + + opa.ExternalSource.start(log.IntoContext(ctx, log.FromContext(ctx).WithName("opa")), opa) + return nil } type OPA struct { @@ -92,11 +103,12 @@ type OPA struct { ExternalSource *OPAExternalSource AllValues bool - regoVersion opaParser.RegoVersion - opaContext context.Context - policy *rego.PreparedEvalQuery - policyName string - policyUID string + regoVersion opaParser.RegoVersion + opaContext context.Context + pullFromRegistry bool + policy *rego.PreparedEvalQuery + policyName string + policyUID string mu sync.RWMutex } @@ -229,8 +241,12 @@ type OPAExternalSource struct { Endpoint string SharedSecret string auth.AuthCredentials - TTL int - Timeout *int + TTL int + Timeout *int + + // guards refresher. deliberately not opa.mu: that one is read-locked by Call() on every + // request, and the refresher is none of its business + mu sync.Mutex refresher workers.Worker } @@ -285,6 +301,8 @@ func (ext *OPAExternalSource) downloadRegoDataFromUrl(ctx context.Context) (stri } } +// setupRefresher assigns ext.refresher and must be called with ext.mu held. The worker callback +// does not take ext.mu, and StartWorker only arms a ticker rather than calling it synchronously. func (ext *OPAExternalSource) setupRefresher(ctx context.Context, opa *OPA) { logger := log.FromContext(ctx).WithValues("policy", opa.policyName, "endpoint", ext.Endpoint) @@ -311,9 +329,28 @@ func (ext *OPAExternalSource) setupRefresher(ctx context.Context, opa *OPA) { } } +// start kicks off the refresher, unless one is already running. The check and the assignment happen +// under the same lock: released in between, two callers could both start one and leak whichever +// loses the race. +func (ext *OPAExternalSource) start(ctx context.Context, opa *OPA) { + ext.mu.Lock() + defer ext.mu.Unlock() + + if ext.refresher != nil { + return + } + + ext.setupRefresher(ctx, opa) +} + func (ext *OPAExternalSource) cleanupRefresher() error { - if ext.refresher == nil { + ext.mu.Lock() + refresher := ext.refresher + ext.refresher = nil + ext.mu.Unlock() + + if refresher == nil { return nil } - return ext.refresher.Stop() + return refresher.Stop() } diff --git a/pkg/evaluators/authorization/opa_test.go b/pkg/evaluators/authorization/opa_test.go index a4bed8049..73cfab730 100644 --- a/pkg/evaluators/authorization/opa_test.go +++ b/pkg/evaluators/authorization/opa_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "sync" "testing" "time" @@ -180,6 +181,10 @@ func TestOPAExternalUrlWithTTL(t *testing.T) { assert.NilError(t, err) assert.Check(t, strings.Contains(opa.GetRego(), "GET")) + // the policy is downloaded by the constructor, but the refresher is only started by the + // reconciler once the authconfig has been translated and indexed + assert.Check(t, opa.ExternalSource.refresher == nil) + assert.NilError(t, opa.Start(context.TODO())) assert.Check(t, opa.ExternalSource.refresher != nil) time.Sleep(4 * time.Second) @@ -305,3 +310,35 @@ func BenchmarkOPAAuthz(b *testing.B) { b.StopTimer() assert.NilError(b, err) } + +// Same as the jwt verifier: concurrent Start must not be able to spawn two refreshers. +func TestOPAConcurrentStart(t *testing.T) { + rego := `allow := true` + extServer := httptest.NewHttpServerMock(opaExtHttpServerMockAddr, map[string]httptest.HttpServerMockResponseFunc{ + "/rego": func() httptest.HttpServerMockResponse { + return httptest.HttpServerMockResponse{Status: 200, Body: rego} + }, + }) + defer extServer.Close() + + externalSource := &OPAExternalSource{ + Endpoint: "http://" + opaExtHttpServerMockAddr + "/rego", + AuthCredentials: auth.NewAuthCredential("", ""), + TTL: 60, + } + opa, err := NewOPAAuthorization("test-opa", "", externalSource, false, opaParser.RegoV1, 0, context.TODO()) + assert.NilError(t, err) + defer func() { _ = opa.Clean(context.Background()) }() + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + assert.NilError(t, opa.Start(context.TODO())) + }() + } + wg.Wait() + + assert.Check(t, opa.ExternalSource.refresher != nil) +} diff --git a/pkg/evaluators/config.go b/pkg/evaluators/config.go index a64785054..0e0eb4abb 100644 --- a/pkg/evaluators/config.go +++ b/pkg/evaluators/config.go @@ -40,32 +40,89 @@ func (config *AuthConfig) GetChallengeHeaders() []map[string]string { return challengeHeaders } -func (config *AuthConfig) Clean(ctx context.Context) error { +// Start kicks off whatever the evaluators of this config should only be running once the config is +// valid and reachable. Unlike Clean, it runs sequentially: there is nothing to parallelise here and +// it keeps the panic surface of the reconcile path small. +func (config *AuthConfig) Start(ctx context.Context) error { + var errs error + + for _, evaluator := range config.allEvaluators() { + if starter, ok := evaluator.(auth.AuthConfigStarter); ok { + if err := startEvaluator(ctx, evaluator, starter); err != nil { + errs = multierror.Append(errs, err) + } + } + } + + return errs +} + +// startEvaluator turns a panicking starter into an error. Start runs in the caller's goroutine, +// i.e. the reconcile one, where an unrecovered panic would take the whole process down. +func startEvaluator(ctx context.Context, evaluator auth.AuthConfigEvaluator, starter auth.AuthConfigStarter) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("recovered from panic starting evaluator %q: %v", evaluatorName(evaluator), r) + } + }() + + return starter.Start(ctx) +} + +func (config *AuthConfig) allEvaluators() []auth.AuthConfigEvaluator { evaluators := []auth.AuthConfigEvaluator{} evaluators = append(evaluators, config.IdentityConfigs...) evaluators = append(evaluators, config.MetadataConfigs...) evaluators = append(evaluators, config.AuthorizationConfigs...) evaluators = append(evaluators, config.ResponseConfigs...) evaluators = append(evaluators, config.CallbackConfigs...) + return evaluators +} + +func (config *AuthConfig) Clean(ctx context.Context) error { + evaluators := config.allEvaluators() - var errors error + // one slot per evaluator: every goroutine writes its own element, so there is no shared + // read-modify-write that could drop an error when two cleaners fail at the same time + cleanErrors := make([]error, len(evaluators)) var wait sync.WaitGroup wait.Add(len(evaluators)) - for _, evaluator := range evaluators { - go func(e auth.AuthConfigEvaluator) { + for i, evaluator := range evaluators { + go func(i int, e auth.AuthConfigEvaluator) { defer wait.Done() - if cleaner, ok := e.(auth.AuthConfigCleaner); ok { - if err := cleaner.Clean(ctx); err != nil { - errors = multierror.Append(errors, err) + // cleanup runs in its own goroutine, so a panic here would be unrecoverable by the + // caller and would take the whole process down with it + defer func() { + if r := recover(); r != nil { + // only the name of the config is reported: evaluators hold credentials in + // exported fields (OAuth2.ClientSecret, OPAExternalSource.SharedSecret) + cleanErrors[i] = fmt.Errorf("recovered from panic cleaning up evaluator %q: %v", evaluatorName(e), r) } + }() + if cleaner, ok := e.(auth.AuthConfigCleaner); ok { + cleanErrors[i] = cleaner.Clean(ctx) } - }(evaluator) + }(i, evaluator) } wait.Wait() - return errors + var errs error + for _, err := range cleanErrors { + if err != nil { + errs = multierror.Append(errs, err) + } + } + + return errs +} + +func evaluatorName(evaluator auth.AuthConfigEvaluator) string { + if named, ok := evaluator.(auth.NamedEvaluator); ok { + return named.GetName() + } + return "" } type DenyWith struct { diff --git a/pkg/evaluators/config_test.go b/pkg/evaluators/config_test.go index a728e7909..205401fd2 100644 --- a/pkg/evaluators/config_test.go +++ b/pkg/evaluators/config_test.go @@ -2,6 +2,8 @@ package evaluators import ( "context" + "fmt" + "strings" "testing" "github.com/kuadrant/authorino/pkg/auth" @@ -51,3 +53,71 @@ func TestCleanConfig(t *testing.T) { assert.Check(t, ev.cleaned) } } + +type failingCleaner struct{ name string } + +func (f *failingCleaner) Call(_ auth.AuthPipeline, _ context.Context) (interface{}, error) { + return nil, nil +} +func (f *failingCleaner) GetName() string { return f.name } +func (f *failingCleaner) Clean(_ context.Context) error { + return fmt.Errorf("failed to clean %s", f.name) +} + +type panickingCleaner struct{ failingCleaner } + +func (p *panickingCleaner) Clean(_ context.Context) error { + panic("boom") +} + +// Every cleaner that fails must be reported: the errors used to be accumulated with a shared +// read-modify-write, which races and can drop one. +func TestCleanReportsEveryFailure(t *testing.T) { + config := &AuthConfig{IdentityConfigs: []auth.AuthConfigEvaluator{ + &failingCleaner{name: "one"}, + &failingCleaner{name: "two"}, + &failingCleaner{name: "three"}, + &failingCleaner{name: "four"}, + }} + + err := config.Clean(context.TODO()) + assert.Check(t, err != nil) + + for _, name := range []string{"one", "two", "three", "four"} { + assert.Check(t, strings.Contains(err.Error(), "failed to clean "+name), "missing error for %q in %v", name, err) + } +} + +// A panicking cleaner must not take the process down, and must be reported rather than swallowed. +func TestCleanRecoversFromPanickingCleaner(t *testing.T) { + config := &AuthConfig{IdentityConfigs: []auth.AuthConfigEvaluator{ + &panickingCleaner{failingCleaner{name: "kaboom"}}, + &failingCleaner{name: "one"}, + }} + + err := config.Clean(context.TODO()) + assert.Check(t, err != nil) + assert.Check(t, strings.Contains(err.Error(), "recovered from panic"), "panic not reported: %v", err) + assert.Check(t, strings.Contains(err.Error(), "kaboom"), "evaluator name not reported: %v", err) + assert.Check(t, strings.Contains(err.Error(), "failed to clean one"), "other errors lost: %v", err) +} + +// A panicking starter must not take the reconcile goroutine, and therefore the process, down. +func TestStartRecoversFromPanickingStarter(t *testing.T) { + config := &AuthConfig{IdentityConfigs: []auth.AuthConfigEvaluator{ + &panickingStarter{name: "kaboom"}, + }} + + err := config.Start(context.TODO()) + assert.Check(t, err != nil) + assert.Check(t, strings.Contains(err.Error(), "recovered from panic"), "panic not reported: %v", err) + assert.Check(t, strings.Contains(err.Error(), "kaboom"), "evaluator name not reported: %v", err) +} + +type panickingStarter struct{ name string } + +func (p *panickingStarter) Call(_ auth.AuthPipeline, _ context.Context) (interface{}, error) { + return nil, nil +} +func (p *panickingStarter) GetName() string { return p.name } +func (p *panickingStarter) Start(_ context.Context) error { panic("boom") } diff --git a/pkg/evaluators/identity.go b/pkg/evaluators/identity.go index ac1c953e6..4d4c68877 100644 --- a/pkg/evaluators/identity.go +++ b/pkg/evaluators/identity.go @@ -149,6 +149,24 @@ func (config *IdentityConfig) GetConditions() jsonexp.Expression { return config.Conditions } +// impl:AuthConfigStarter + +func (config *IdentityConfig) Start(ctx context.Context) error { + if starter := config.getStarter(); starter != nil { + return starter.Start(log.IntoContext(ctx, log.FromContext(ctx).WithName("identity"))) + } + return nil +} + +func (config *IdentityConfig) getStarter() auth.AuthConfigStarter { + switch { + case config.JWTAuthentication != nil: + return config.JWTAuthentication + default: + return nil + } +} + // impl:AuthConfigCleaner func (config *IdentityConfig) Clean(ctx context.Context) error { diff --git a/pkg/evaluators/identity/jwt.go b/pkg/evaluators/identity/jwt.go index d4384c6f8..27c1f40df 100644 --- a/pkg/evaluators/identity/jwt.go +++ b/pkg/evaluators/identity/jwt.go @@ -65,6 +65,18 @@ func (j *JWTAuthentication) Call(pipeline auth.AuthPipeline, ctx gocontext.Conte return claims, nil } +// impl:auth.AuthConfigStarter +func (j *JWTAuthentication) Start(ctx gocontext.Context) error { + if j.verifier == nil { + return nil + } + starter, ok := j.verifier.(auth.AuthConfigStarter) + if !ok { + return nil + } + return starter.Start(ctx) +} + // impl:auth.AuthConfigCleaner func (j *JWTAuthentication) Clean(ctx gocontext.Context) error { if j.verifier == nil { @@ -93,6 +105,7 @@ type JWTVerifier interface { type oidcProviderVerifier struct { issuerUrl string + ttl int timeout *int mu sync.RWMutex @@ -100,17 +113,35 @@ type oidcProviderVerifier struct { refresher workers.Worker } +// NewOIDCProviderVerifier discovers the openid connect configuration straight away, so a broken +// issuer still surfaces while the authconfig is being translated, but leaves the periodic refresh +// to Start(), which the reconciler only calls once the config is indexed. func NewOIDCProviderVerifier(ctx gocontext.Context, issuerUrl string, ttl int, timeout *int) JWTVerifier { v := &oidcProviderVerifier{ issuerUrl: issuerUrl, + ttl: ttl, timeout: timeout, } ctxWithLogger := log.IntoContext(ctx, log.FromContext(ctx).WithName("jwt")) v.getOpenIdProvider(ctxWithLogger, false) - v.setupOpenIdProviderRefresh(ctxWithLogger, ttl) return v } +// impl: auth.AuthConfigStarter +func (v *oidcProviderVerifier) Start(ctx gocontext.Context) error { + v.mu.Lock() + defer v.mu.Unlock() + + // the check and the assignment have to happen under the same lock: released in between, two + // callers could both find no refresher and both start one, leaking whichever loses the race + if v.refresher != nil { + return nil + } + + v.setupOpenIdProviderRefresh(log.IntoContext(ctx, log.FromContext(ctx).WithName("jwt")), v.ttl) + return nil +} + func (v *oidcProviderVerifier) Verify(ctx gocontext.Context, rawIDToken string) (*oidc.IDToken, error) { provider := v.getOpenIdProvider(ctx, false) if provider == nil { @@ -152,10 +183,15 @@ func (v *oidcProviderVerifier) GetOpenIdUrl(ctx gocontext.Context, claim string) // Clean ensures the goroutine started by setupOpenIdProviderRefresh is cleaned up // impl: auth.AuthConfigCleaner func (v *oidcProviderVerifier) Clean(ctx gocontext.Context) error { - if v.refresher == nil { + v.mu.Lock() + refresher := v.refresher + v.refresher = nil + v.mu.Unlock() + + if refresher == nil { return nil } - return v.refresher.Stop() + return refresher.Stop() } // GetProvider returns the current OIDC provider in a thread-safe manner @@ -187,6 +223,11 @@ func (v *oidcProviderVerifier) getOpenIdProvider(ctx gocontext.Context, force bo return v.provider } +// setupOpenIdProviderRefresh assigns v.refresher and must be called with v.mu held. +// +// The worker callback takes v.mu itself, so this relies on StartWorker only arming a ticker and +// never invoking the callback synchronously. Do not make it fire immediately without moving this +// call out of the critical section first. func (v *oidcProviderVerifier) setupOpenIdProviderRefresh(ctx gocontext.Context, ttl int) { var err error @@ -227,3 +268,15 @@ func (v *jwksVerifier) Verify(ctx gocontext.Context, rawIDToken string) (*oidc.I func (v *jwksVerifier) Clean(_ gocontext.Context) error { return nil } + +// RefresherRunning reports whether the background refresher of the underlying verifier is running. +// Exposed so the reconciler tests can assert on the lifecycle of a config held in the index. +func (j *JWTAuthentication) RefresherRunning() bool { + verifier, ok := j.verifier.(*oidcProviderVerifier) + if !ok { + return false + } + verifier.mu.RLock() + defer verifier.mu.RUnlock() + return verifier.refresher != nil +} diff --git a/pkg/evaluators/identity/jwt_test.go b/pkg/evaluators/identity/jwt_test.go index 48f03a48a..64097eb29 100644 --- a/pkg/evaluators/identity/jwt_test.go +++ b/pkg/evaluators/identity/jwt_test.go @@ -151,6 +151,10 @@ func TestOIDCProviderVerifierRefresh(t *testing.T) { }(evaluator, context.Background()) verifier, _ := jwtVerifier.(*oidcProviderVerifier) + // the refresher is not started by the constructor any more: the reconciler starts it once the + // authconfig has been translated and indexed + assert.Check(t, verifier.refresher == nil) + assert.NilError(t, evaluator.Start(context.TODO())) assert.Check(t, verifier.refresher != nil) time.Sleep(4 * time.Second) @@ -254,3 +258,29 @@ func TestJWKSVerifierMalformedJWT(t *testing.T) { assert.Check(t, token == nil) assert.ErrorContains(t, err, "oidc: malformed jwt") } + +// Start must be safe to call concurrently: the check for an existing refresher and the assignment +// of a new one have to happen under the same lock, or two callers both start one and one leaks. +func TestOIDCProviderVerifierConcurrentStart(t *testing.T) { + authServer := httptest.NewHttpServerMock(oidcServerHost, map[string]httptest.HttpServerMockResponseFunc{ + "/.well-known/openid-configuration": func() httptest.HttpServerMockResponse { + return oidcServerMockResponse(1) + }, + }) + defer authServer.Close() + + verifier := NewOIDCProviderVerifier(context.TODO(), fmt.Sprintf("http://%v", oidcServerHost), 60, nil).(*oidcProviderVerifier) + defer func() { _ = verifier.Clean(context.TODO()) }() + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + assert.NilError(t, verifier.Start(context.TODO())) + }() + } + wg.Wait() + + assert.Check(t, verifier.refresher != nil) +} diff --git a/pkg/index/index.go b/pkg/index/index.go index bb3c43c2a..392534231 100644 --- a/pkg/index/index.go +++ b/pkg/index/index.go @@ -2,6 +2,7 @@ package index import ( "fmt" + "slices" "strings" "sync" @@ -73,7 +74,7 @@ func (c *authConfigTree) Set(id, key string, config evaluators.AuthConfig, overr AuthConfig: config, } err := c.root.set(revertKey(key), entry, override) - if err == nil { + if err == nil && !slices.Contains(c.keys[id], key) { c.keys[id] = append(c.keys[id], key) } return err @@ -84,9 +85,10 @@ func (c *authConfigTree) Delete(id string) { defer c.mu.Unlock() if keys, ok := c.keys[id]; ok { - for _, key := range keys { + for _, key := range slices.Clone(keys) { c.deleteKey(id, key) } + delete(c.keys, id) } } @@ -133,6 +135,20 @@ func (c *authConfigTree) deleteKey(id, key string) { if node, _ := c.root.longestCommonLabel(revertKey(key)); node != nil && node.entry != nil && node.entry.Id == id { node.entry = nil } + + // the key must go from the id -> keys map as well, otherwise FindKeys() keeps reporting hosts + // the resource no longer owns, and callers such as the reconciler's cleanConfigs() end up + // resolving to whichever AuthConfig owns that host now + if keys, ok := c.keys[id]; ok { + if i := slices.Index(keys, key); i >= 0 { + keys = slices.Delete(keys, i, i+1) + } + if len(keys) == 0 { + delete(c.keys, id) + } else { + c.keys[id] = keys + } + } } func newTreeNode(label string, parent *treeNode) *treeNode { diff --git a/pkg/index/index_test.go b/pkg/index/index_test.go index 1769c3f16..2f79ee9f6 100644 --- a/pkg/index/index_test.go +++ b/pkg/index/index_test.go @@ -152,3 +152,62 @@ func buildTestAuthConfig() evaluators.AuthConfig { AuthorizationConfigs: nil, } } + +func TestDeleteKeyPrunesTheKeysOfTheId(t *testing.T) { + c := newAuthConfigTree() + authConfig := buildTestAuthConfig() + + assert.NilError(t, c.Set("auth-1", "talker-api.nip.io", authConfig, false)) + assert.NilError(t, c.Set("auth-1", "echo-api.nip.io", authConfig, false)) + assert.Equal(t, len(c.FindKeys("auth-1")), 2) + + c.DeleteKey("auth-1", "talker-api.nip.io") + + // FindKeys() must not keep reporting a host the resource no longer owns: cleanConfigs() + // resolves the config to clean from FindKeys()[0] and would otherwise pick up whichever + // AuthConfig owns that host now + assert.DeepEqual(t, c.FindKeys("auth-1"), []string{"echo-api.nip.io"}) + assert.Check(t, c.Get("talker-api.nip.io") == nil) + assert.Check(t, c.Get("echo-api.nip.io") != nil) +} + +func TestDeleteRemovesTheIdFromTheKeys(t *testing.T) { + c := newAuthConfigTree() + authConfig := buildTestAuthConfig() + + assert.NilError(t, c.Set("auth-1", "talker-api.nip.io", authConfig, false)) + c.Delete("auth-1") + + assert.Equal(t, len(c.FindKeys("auth-1")), 0) + assert.Check(t, c.Empty()) +} + +func TestSetDoesNotDuplicateKeys(t *testing.T) { + c := newAuthConfigTree() + authConfig := buildTestAuthConfig() + + // every reconcile of an unchanged AuthConfig re-indexes the same hosts + for i := 0; i < 5; i++ { + assert.NilError(t, c.Set("auth-1", "talker-api.nip.io", authConfig, true)) + } + + assert.DeepEqual(t, c.FindKeys("auth-1"), []string{"talker-api.nip.io"}) +} + +func TestDeleteKeyDoesNotStealTheHostOfAnotherId(t *testing.T) { + c := newAuthConfigTree() + authConfig := buildTestAuthConfig() + + // auth-1 owns both hosts, then gets narrowed down to one of them + assert.NilError(t, c.Set("auth-1", "talker-api.nip.io", authConfig, false)) + assert.NilError(t, c.Set("auth-1", "echo-api.nip.io", authConfig, false)) + c.DeleteKey("auth-1", "talker-api.nip.io") + + // auth-2 legitimately takes over the released host (addToIndex() always sets with override) + assert.NilError(t, c.Set("auth-2", "talker-api.nip.io", authConfig, true)) + + id, found := c.FindId("talker-api.nip.io") + assert.Check(t, found) + assert.Equal(t, id, "auth-2") + assert.DeepEqual(t, c.FindKeys("auth-1"), []string{"echo-api.nip.io"}) +} diff --git a/pkg/workers/worker.go b/pkg/workers/worker.go index af6d2a51c..8b9dda0a3 100644 --- a/pkg/workers/worker.go +++ b/pkg/workers/worker.go @@ -3,6 +3,7 @@ package workers import ( "context" "fmt" + "sync" "time" ) @@ -26,8 +27,10 @@ type Worker interface { } type worker struct { - ctx context.Context - f func() + ctx context.Context + f func() + + mu sync.Mutex timer *time.Ticker done chan bool } @@ -44,19 +47,21 @@ func (w *worker) Start(interval int) error { duration := time.Duration(interval) * time.Second + w.mu.Lock() + defer w.mu.Unlock() + if w.timer != nil { w.timer.Stop() } - w.timer = time.NewTicker(duration) - + timer := time.NewTicker(duration) done := make(chan bool, 1) go func() { - defer w.timer.Stop() + defer timer.Stop() for { select { - case <-w.timer.C: + case <-timer.C: w.f() case <-w.ctx.Done(): return @@ -66,14 +71,22 @@ func (w *worker) Start(interval int) error { } }() + w.timer = timer w.done = done return nil } +// Stop is idempotent: stopping a worker that was never started, or that has already been +// stopped, is a no-op. Callers such as the AuthConfig reconciler clean up whatever config is +// currently in the index and cannot know whether that instance was cleaned up before. func (w *worker) Stop() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.done != nil { close(w.done) + w.done = nil } return nil } diff --git a/pkg/workers/worker_test.go b/pkg/workers/worker_test.go index 8fefa13ed..a12c45473 100644 --- a/pkg/workers/worker_test.go +++ b/pkg/workers/worker_test.go @@ -60,3 +60,41 @@ func TestStopWorker(t *testing.T) { mu.Unlock() assert.Equal(t, currentVal, 0) } + +func TestStopWorkerIsIdempotent(t *testing.T) { + worker, err := StartWorker(context.TODO(), 60, func() {}) + assert.NilError(t, err) + + // the AuthConfig reconciler cleans up whatever config is currently in the index and cannot + // know whether that same instance was already cleaned up by a previous (failed) reconcile + assert.NilError(t, worker.Stop()) + assert.NilError(t, worker.Stop()) + assert.NilError(t, worker.Stop()) +} + +func TestStopWorkerNeverStarted(t *testing.T) { + w := &worker{ctx: context.TODO(), f: func() {}} + assert.NilError(t, w.Stop()) +} + +func TestRestartWorkerAfterStop(t *testing.T) { + var mu sync.Mutex + val := 0 + worker, err := StartWorker(context.TODO(), 2, func() { + mu.Lock() + val += 1 + mu.Unlock() + }) + assert.NilError(t, err) + assert.NilError(t, worker.Stop()) + + assert.NilError(t, worker.Start(2)) + defer func() { _ = worker.Stop() }() + + time.Sleep(3 * time.Second) + + mu.Lock() + currentVal := val + mu.Unlock() + assert.Equal(t, currentVal, 1) +}