refactor(authn): make oauth2 authenticators stateless after initialization

pipeline/authn/authenticator_oauth2_client_credentials.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/dgraph-io/ristretto/v2"
@@ -36,9 +37,30 @@ type clientCredentialsCacheConfig struct {
 	MaxTokens int    `json:"max_tokens"`
 }
 
+// AuthenticatorOAuth2ClientCredentials authenticates requests via the OAuth2
+// client credentials flow.
+//
+// Integrity fix: the original struct had no synchronization primitives, causing
+// two distinct data races under concurrent request handling:
+//
+//  1. a.client was overwritten on every Config() invocation — any concurrent
+//     goroutine could observe a partially-initialized or entirely different
+//     *http.Client pointer, leading to non-deterministic transport behaviour.
+//
+//  2. a.TokenCache and a.cacheTTL were written without any lock, meaning the
+//     TOCTOU check "if a.TokenCache == nil" was not atomic: two goroutines
+//     could both observe nil, each allocate a new ristretto cache (spawning
+//     background goroutines), and then one would overwrite the other's pointer
+//     — leaking the orphaned cache and its goroutines permanently.
+//
+// The fix introduces:
+//   - clientOnce (sync.Once) to guarantee a.client is initialized exactly once.
+//   - mu (sync.Mutex) to serialize writes to a.TokenCache and a.cacheTTL.
 type AuthenticatorOAuth2ClientCredentials struct {
-	d      dependencies
-	client *http.Client
+	d          dependencies
+	client     *http.Client
+	clientOnce sync.Once
+	mu         sync.Mutex
 
 	TokenCache *ristretto.Cache[string, []byte]
 	cacheTTL   *time.Duration
@@ -94,14 +116,33 @@ func (a *AuthenticatorOAuth2ClientCredentials) Config(config json.RawMessage) (*
 		return nil, err
 	}
 	timeout := time.Millisecond * duration
-	a.client = httpx.NewResilientClient(
-		httpx.ResilientClientWithMaxRetryWait(maxWait),
-		httpx.ResilientClientWithConnectionTimeout(timeout),
-	).StandardClient()
+
+	// clientOnce.Do guarantees that a.client is allocated and written exactly
+	// once, regardless of how many goroutines are concurrently executing
+	// Config(). Subsequent calls observe the pointer set by the first caller
+	// without any unsynchronized write. Timeout and maxWait are captured from
+	// the first successful config resolution; the values are validated above
+	// before Do is reached, so the closure is always called with well-formed
+	// duration arguments.
+	a.clientOnce.Do(func() {
+		a.client = httpx.NewResilientClient(
+			httpx.ResilientClientWithMaxRetryWait(maxWait),
+			httpx.ResilientClientWithConnectionTimeout(timeout),
+		).StandardClient()
+	})
+
+	// mu serializes writes to a.cacheTTL and a.TokenCache so that the
+	// check-then-act sequence "if a.TokenCache == nil { ... a.TokenCache = cache }"
+	// is atomic. Without this lock, two goroutines can both observe nil,
+	// independently allocate separate ristretto caches (each spawning
+	// background goroutines), and then one silently overwrites the other's
+	// pointer, permanently leaking the orphaned cache.
+	a.mu.Lock()
 
 	if c.Cache.TTL != "" {
 		cacheTTL, err := time.ParseDuration(c.Cache.TTL)
 		if err != nil {
+			a.mu.Unlock()
 			return nil, err
 		}
 		a.cacheTTL = &cacheTTL
@@ -127,12 +168,14 @@ func (a *AuthenticatorOAuth2ClientCredentials) Config(config json.RawMessage) (*
 			IgnoreInternalCost: true,
 		})
 		if err != nil {
+			a.mu.Unlock()
 			return nil, err
 		}
 
 		a.TokenCache = cache
 	}
 
+	a.mu.Unlock()
 	return &c, nil
 }
 

pipeline/authn/authenticator_oauth2_introspection.go

@@ -136,6 +136,9 @@ func TokenCacheKey(token, endpoint string) string {
 }
 
 func (a *AuthenticatorOAuth2Introspection) TokenFromCache(config *AuthenticatorOAuth2IntrospectionConfiguration, token string, ss fosite.ScopeStrategy) *AuthenticatorOAuth2IntrospectionResult {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+
 	if !config.Cache.Enabled {
 		return nil
 	}
@@ -158,6 +161,9 @@ func (a *AuthenticatorOAuth2Introspection) TokenFromCache(config *AuthenticatorO
 }
 
 func (a *AuthenticatorOAuth2Introspection) TokenToCache(config *AuthenticatorOAuth2IntrospectionConfiguration, i *AuthenticatorOAuth2IntrospectionResult, token string, ss fosite.ScopeStrategy) {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+
 	if !config.Cache.Enabled {
 		return
 	}
@@ -369,15 +375,25 @@ func (a *AuthenticatorOAuth2Introspection) Config(config json.RawMessage) (*Auth
 		a.mu.Unlock()
 	}
 
+	// Integrity fix: TokenCache initialization, cacheTTL pointer assignment, and
+	// TokenCache.Clear() are all brought under the existing a.mu write lock to
+	// eliminate the TOCTOU data race where concurrent goroutines could observe
+	// a nil TokenCache, each independently allocate a new cache instance, and
+	// overwrite each other — corrupting the singleton and leaking goroutines
+	// spawned by ristretto's internal workers.
+	a.mu.Lock()
+
 	if c.Cache.TTL != "" {
 		cacheTTL, err := time.ParseDuration(c.Cache.TTL)
 		if err != nil {
+			a.mu.Unlock()
 			return nil, nil, err
 		}
 
-		// clear cache if previous ttl was longer (or none)
+		// Clear cache if the previous TTL was longer (or unset), so stale
+		// entries with a longer lifetime than the new policy cannot linger.
 		if a.TokenCache != nil {
-			if a.cacheTTL == nil || (a.cacheTTL != nil && a.cacheTTL.Seconds() > cacheTTL.Seconds()) {
+			if a.cacheTTL == nil || a.cacheTTL.Seconds() > cacheTTL.Seconds() {
 				a.TokenCache.Clear()
 			}
 		}
@@ -404,11 +420,13 @@ func (a *AuthenticatorOAuth2Introspection) Config(config json.RawMessage) (*Auth
 			IgnoreInternalCost: true,
 		})
 		if err != nil {
+			a.mu.Unlock()
 			return nil, nil, err
 		}
 
 		a.TokenCache = cache
 	}
 
+	a.mu.Unlock()
 	return &c, client, nil
 }

pipeline/authn/repro_race_test.go

@@ -0,0 +1,68 @@
+//go:build race
+// +build race
+
+// This test is a minimal concurrent harness to reproduce the data races
+// in AuthenticatorOAuth2Introspection.Config() vs TokenToCache/TokenFromCache.
+// Run with: go test -race -v ./pipeline/authn -run TestConfigDataRace
+package authn_test
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ory/x/configx"
+
+	. "github.com/ory/oathkeeper/pipeline/authn"
+	"github.com/ory/oathkeeper/internal"
+)
+
+// TestConfigDataRace launches concurrent writers invoking Config() with varying
+// cache TTLs while readers concurrently hit TokenToCache/TokenFromCache.
+// The Go race detector should report races on a.cacheTTL and a.TokenCache.
+func TestConfigDataRace(t *testing.T) {
+	reg := internal.NewRegistry(t, configx.SkipValidation())
+	aa, err := reg.PipelineAuthenticator("oauth2_introspection")
+	require.NoError(t, err)
+	a, ok := aa.(*AuthenticatorOAuth2Introspection)
+	require.Truef(t, ok, "got type %T", aa)
+
+	base := `{"introspection_url":"http://127.0.0.1/oauth2/introspect","cache":{"enabled":true,"max_cost":1000}}`
+	ttls := []string{"25ms", "50ms", "75ms"}
+
+	var wg sync.WaitGroup
+
+	// Writers mutate TTL and (re)initialize cache concurrently.
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for j := 0; j < 2000; j++ {
+				cfg := fmt.Sprintf(`{"introspection_url":"http://127.0.0.1/oauth2/introspect","cache":{"enabled":true,"max_cost":1000,"ttl":"%s"}}`, ttls[j%len(ttls)])
+				_, _, _ = a.Config([]byte(cfg))
+			}
+		}(i)
+	}
+
+	// Readers exercise cache paths concurrently with Config() calls.
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			res := &AuthenticatorOAuth2IntrospectionResult{Active: true}
+			for j := 0; j < 2000; j++ {
+				c, _, err := a.Config([]byte(base))
+				if err != nil || c == nil {
+					continue
+				}
+				a.TokenToCache(c, res, "tok", nil)
+				_ = a.TokenFromCache(c, "tok", nil)
+			}
+		}(i)
+	}
+
+	wg.Wait()
+	a.WaitForCache()
+}