diff --git a/weed/iam/integration/iam_manager.go b/weed/iam/integration/iam_manager.go index ece1a54927..6e7ade42cf 100644 --- a/weed/iam/integration/iam_manager.go +++ b/weed/iam/integration/iam_manager.go @@ -53,12 +53,9 @@ type IAMManager struct { oidcRetryAgain bool // oidcRefreshMu serializes refreshes from reading the store to handing // STS the result, so an older snapshot cannot replace a newer one. - oidcRefreshMu sync.Mutex - // oidcManagedProviders caches providers refreshOIDCProvidersFrom built, - // keyed by record identity, so an unchanged record reuses its provider: - // rebuilding would drop the provider's consumed-token (jti) replay cache - // and let an already-used token be presented again. Guarded by - // oidcRefreshMu. + // oidcManagedProviders caches the provider built per stored record so a + // refresh reuses unchanged providers and keeps their JWKS cache. + oidcRefreshMu sync.Mutex oidcManagedProviders map[string]*oidcManagedProvider oidcAuditSink OIDCProviderAuditSink revocationStore SessionRevocationStore @@ -953,10 +950,6 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro m.oidcManagedProviders = make(map[string]*oidcManagedProvider) } seen := make(map[string]struct{}, len(records)) - // Records for one issuer with identical validation settings share a single - // provider — and with it the consumed-token (jti) replay store — so a token - // spent through one account's record stays spent through the other's too. - sharedProviders := make(map[string]*oidc.OIDCProvider) for _, rec := range records { if rec == nil || rec.URL == "" { continue @@ -967,25 +960,10 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro } seen[key] = struct{}{} fingerprint := oidcProviderFingerprint(rec) - configFingerprint := oidcProviderConfigFingerprint(rec) - var provider *oidc.OIDCProvider cached := m.oidcManagedProviders[key] - if shared, ok := sharedProviders[configFingerprint]; ok { - provider = shared - if cached != nil && cached.provider != nil && cached.provider != shared { - // The record's own provider is superseded; its consumed - // tokens move to the shared store so none reopen for replay. - shared.MergeReplayStateFrom(cached.provider) - } - m.oidcManagedProviders[key] = &oidcManagedProvider{fingerprint: fingerprint, provider: provider} - if cached != nil && cached.provider != nil && cached.provider != shared && !m.oidcProviderStillReferenced(cached.provider) { - if err := cached.provider.Shutdown(ctx); err != nil { - glog.Warningf("shutdown superseded OIDC provider %s: %v", rec.ARN, err) - } - } - } else if cached != nil && cached.fingerprint == fingerprint { + var provider *oidc.OIDCProvider + if cached != nil && cached.fingerprint == fingerprint { provider = cached.provider - sharedProviders[configFingerprint] = provider } else { var err error provider, err = buildOIDCProviderFromRecord(rec) @@ -994,20 +972,7 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro glog.Warningf("skip refreshing OIDC provider %s: %v", rec.ARN, err) continue } - if cached != nil && cached.provider != nil { - // The rebuild must keep rejecting tokens the replaced - // provider consumed; carry its replay store over, then stop - // its cleanup worker so only the new one sweeps the store — - // unless another record still references that provider. - provider.AdoptReplayStateFrom(cached.provider) - } - sharedProviders[configFingerprint] = provider m.oidcManagedProviders[key] = &oidcManagedProvider{fingerprint: fingerprint, provider: provider} - if cached != nil && cached.provider != nil && !m.oidcProviderStillReferenced(cached.provider) { - if err := cached.provider.Shutdown(ctx); err != nil { - glog.Warningf("shutdown replaced OIDC provider %s: %v", rec.ARN, err) - } - } } // Multiple records may share an issuer when each is scoped to a // different account; STS picks the right one at validation time @@ -1017,16 +982,9 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro Provider: provider, }) } - for key, managed := range m.oidcManagedProviders { + for key := range m.oidcManagedProviders { if _, ok := seen[key]; !ok { delete(m.oidcManagedProviders, key) - // A removed record's provider may be shared by records that are - // still present; shut it down only when nothing references it. - if managed.provider != nil && !m.oidcProviderStillReferenced(managed.provider) { - if err := managed.provider.Shutdown(ctx); err != nil { - glog.Warningf("shutdown removed OIDC provider %s: %v", key, err) - } - } } } m.stsService.SetIAMManagedOIDCProviders(byIssuer) @@ -1040,17 +998,6 @@ type oidcManagedProvider struct { provider *oidc.OIDCProvider } -// oidcProviderStillReferenced reports whether any managed record still points -// at the provider, so a shared provider is not shut down with a removed record. -func (m *IAMManager) oidcProviderStillReferenced(provider *oidc.OIDCProvider) bool { - for _, managed := range m.oidcManagedProviders { - if managed.provider == provider { - return true - } - } - return false -} - // oidcProviderFingerprint identifies the record content a provider was built // from, so a refresh rebuilds the provider only when that content changed. func oidcProviderFingerprint(rec *OIDCProviderRecord) string { @@ -1067,21 +1014,6 @@ func oidcProviderFingerprint(rec *OIDCProviderRecord) string { return hex.EncodeToString(sum[:]) } -// oidcProviderConfigFingerprint identifies the validation settings a provider -// is built from — everything buildOIDCProviderFromRecord consumes except the -// record's name — so records that differ only in identity reuse one provider. -func oidcProviderConfigFingerprint(rec *OIDCProviderRecord) string { - key, _ := json.Marshal(&oidc.OIDCConfig{ - Issuer: rec.URL, - ClientIDs: rec.ClientIDs, - Thumbprints: rec.Thumbprints, - AllowedPrincipalTagKeys: rec.AllowedPrincipalTagKeys, - PolicyClaim: rec.PolicyClaim, - }) - sum := sha256.Sum256(key) - return hex.EncodeToString(sum[:]) -} - // buildOIDCProviderFromRecord turns a stored record into a runtime // OIDCProvider. The provider name is the ARN so re-registration is // idempotent and never collides with static-config entries. diff --git a/weed/iam/integration/oidc_provider_persist_test.go b/weed/iam/integration/oidc_provider_persist_test.go index 26a6f81250..98a176f1be 100644 --- a/weed/iam/integration/oidc_provider_persist_test.go +++ b/weed/iam/integration/oidc_provider_persist_test.go @@ -560,92 +560,47 @@ func TestAFailureDuringARetryIsNotDropped(t *testing.T) { assert.GreaterOrEqual(t, store.successes(), 2, "the retry ended on its first success, dropping the refresh that failed while it ran") } -// A refresh must not rebuild a provider whose record is unchanged: rebuilding -// would drop the provider's consumed-token (jti) replay cache and let an -// already-used token be presented again. -func TestRefreshReusesProviderWhileRecordIsUnchanged(t *testing.T) { - ctx := context.Background() - store := NewMemoryOIDCProviderStore() +// A refresh caused by an unrelated record must not rebuild a provider whose +// own record is unchanged: rebuilding drops its JWKS cache, so a briefly +// unreachable issuer would fail exchanges its cached keys still allow. +func TestAnUnrelatedRefreshKeepsAProvidersRuntimeState(t *testing.T) { + store := &persistentTestStore{NewMemoryOIDCProviderStore()} mgr := startServer(t, store) - arn := arnOf(t, persistTestAPIIssuer) - require.NoError(t, store.StoreProvider(ctx, "", &OIDCProviderRecord{ - ARN: arn, URL: persistTestAPIIssuer, ClientIDs: []string{"aud"}, - })) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - first := mgr.oidcManagedProviders[arn].provider - require.NotNil(t, first) + createAPIProvider(t, mgr, persistTestAPIIssuer) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - assert.Same(t, first, mgr.oidcManagedProviders[arn].provider, - "an unchanged record was rebuilt, dropping its replay cache") + mgr.oidcRefreshMu.Lock() + before := mgr.oidcManagedProviders[arn] + mgr.oidcRefreshMu.Unlock() + require.NotNil(t, before) - // A changed record must rebuild so the new configuration applies. - require.NoError(t, store.StoreProvider(ctx, "", &OIDCProviderRecord{ - ARN: arn, URL: persistTestAPIIssuer, ClientIDs: []string{"aud", "aud2"}, - })) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - rebuilt := mgr.oidcManagedProviders[arn].provider - require.NotNil(t, rebuilt) - assert.NotSame(t, first, rebuilt, "a changed record kept the old provider") + createAPIProvider(t, mgr, persistTestStaticIssuer) // refreshes over an unrelated record - // Deleting the record drops the provider from the cache and the STS map. - require.NoError(t, store.DeleteProvider(ctx, "", arn)) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - assert.NotContains(t, mgr.oidcManagedProviders, arn) - assert.False(t, stsKnowsIssuer(t, mgr, persistTestAPIIssuer)) + mgr.oidcRefreshMu.Lock() + after := mgr.oidcManagedProviders[arn] + mgr.oidcRefreshMu.Unlock() + require.NotNil(t, after) + assert.Same(t, before.provider, after.provider, "an unchanged record's provider was rebuilt") } -// stsKnowsIssuerForAccount is stsKnowsIssuer through the account-scoped -// lookup used for providers that have no global record. -func stsKnowsIssuerForAccount(t *testing.T, mgr *IAMManager, issuer, accountID string) bool { - t.Helper() - tok, err := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{ - "iss": issuer, "sub": "probe", "aud": "aud", "exp": time.Now().Add(time.Hour).Unix(), - }).SignedString([]byte("not-the-providers-key")) - require.NoError(t, err) - _, _, err = mgr.GetSTSService().ValidateWebIdentityTokenForAccount(context.Background(), tok, accountID) - require.Error(t, err, "an unsigned-by-provider token for %s was accepted", issuer) - return !strings.Contains(err.Error(), "no identity provider registered") -} - -// Two records for one issuer with identical validation settings share a -// single provider — and with it the consumed-token (jti) replay store — so a -// token spent through one account's record stays spent through the other's. -// Removing one record of the pair must not shut the shared provider down. -func TestSameIssuerRecordsShareOneProvider(t *testing.T) { - ctx := context.Background() - store := NewMemoryOIDCProviderStore() +// A record whose validation settings change must get a new provider: the +// cached runtime cannot keep serving settings the store no longer holds. +func TestAChangedRecordRebuildsItsProvider(t *testing.T) { + store := &persistentTestStore{NewMemoryOIDCProviderStore()} mgr := startServer(t, store) + arn := arnOf(t, persistTestAPIIssuer) + createAPIProvider(t, mgr, persistTestAPIIssuer) - arnA := arnOf(t, persistTestAPIIssuer) + "-account-a" - arnB := arnOf(t, persistTestAPIIssuer) + "-account-b" - for _, rec := range []*OIDCProviderRecord{ - {ARN: arnA, URL: persistTestAPIIssuer, AccountID: "111111111111", ClientIDs: []string{"aud"}}, - {ARN: arnB, URL: persistTestAPIIssuer, AccountID: "222222222222", ClientIDs: []string{"aud"}}, - } { - require.NoError(t, store.StoreProvider(ctx, "", rec)) - } - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) + mgr.oidcRefreshMu.Lock() + before := mgr.oidcManagedProviders[arn].provider + mgr.oidcRefreshMu.Unlock() - shared := mgr.oidcManagedProviders[arnA].provider - require.NotNil(t, shared) - assert.Same(t, shared, mgr.oidcManagedProviders[arnB].provider, - "identical records for one issuer built separate replay stores") + require.NoError(t, mgr.UpdateOIDCProviderThumbprints(context.Background(), arn, + []string{"0123456789abcdef0123456789abcdef01234567"})) - // Removing one record of the pair keeps the shared provider alive. - require.NoError(t, store.DeleteProvider(ctx, "", arnA)) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - assert.Same(t, shared, mgr.oidcManagedProviders[arnB].provider, - "a removed sibling record shut down the shared provider") - assert.True(t, stsKnowsIssuerForAccount(t, mgr, persistTestAPIIssuer, "222222222222")) - - // The remaining record diverging rebuilds its own provider. - require.NoError(t, store.StoreProvider(ctx, "", &OIDCProviderRecord{ - ARN: arnB, URL: persistTestAPIIssuer, AccountID: "222222222222", ClientIDs: []string{"b-aud"}, - })) - require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store)) - assert.NotSame(t, shared, mgr.oidcManagedProviders[arnB].provider, - "a changed record kept the shared provider") - assert.True(t, stsKnowsIssuerForAccount(t, mgr, persistTestAPIIssuer, "222222222222")) + mgr.oidcRefreshMu.Lock() + after := mgr.oidcManagedProviders[arn] + mgr.oidcRefreshMu.Unlock() + require.NotNil(t, after) + assert.NotSame(t, before, after.provider, "a changed record kept its provider") } diff --git a/weed/iam/oidc/jti_cache.go b/weed/iam/oidc/jti_cache.go deleted file mode 100644 index 77dab0d044..0000000000 --- a/weed/iam/oidc/jti_cache.go +++ /dev/null @@ -1,410 +0,0 @@ -package oidc - -import ( - "context" - "crypto/sha256" - "encoding/hex" - "fmt" - "strings" - "sync" - "sync/atomic" - "time" - - "github.com/seaweedfs/seaweedfs/weed/glog" - "github.com/seaweedfs/seaweedfs/weed/iam/providers" -) - -const ( - // clockSkewTolerance matches the typical JWT library clock skew allowance (RFC 7519) - clockSkewTolerance = 5 * time.Minute - // defaultJTICacheSize is the default maximum number of JTI entries to cache - defaultJTICacheSize = 10000 - // defaultCleanupInterval is how often to scan for expired JTI entries - defaultCleanupInterval = 5 * time.Minute - // eagerCleanupMinInterval rate limits opportunistic sweeps near capacity - eagerCleanupMinInterval = 5 * time.Second - // tokenFingerprintPrefix marks replay-cache keys derived from the token's - // own fingerprint, used when the token carries no usable jti claim - tokenFingerprintPrefix = "fp:" -) - -// jtiEntry represents a cached JTI with metadata -type jtiEntry struct { - subject string - expiresAt time.Time - // uncounted marks an entry merged in past the capacity bound: it keeps - // rejecting replays but does not charge the count, so merged excess can - // neither reject new exchanges nor be double-counted when swept. - uncounted bool -} - -// jtiReplayCache wraps the consumed-token store that backs JTI replay -// protection. Each consumed token identifier is held until the token's exp -// plus clock skew, and a second presentation of the same identifier is -// rejected with ErrProviderTokenReplayed. -// -// The cache is bounded by maxSize and uses automatic TTL-based expiration: -// a background worker sweeps expired entries, an eager sweep runs when the -// cache nears capacity, and the consume path reclaims expired slots before -// rejecting a fresh token for capacity. -// -// Thread Safety: -// All methods are safe for concurrent use. The consume path uses atomic -// LoadOrStore to prevent TOCTOU races. -type jtiReplayCache struct { - name string // owning provider name, for log messages - - enabled atomic.Bool - maxSize atomic.Int64 - lastSweep atomic.Int64 // unix nano of the last sweep; rate limits eager cleanups - - // storeMu guards the store/count pair: replay-state handoffs (adoptFrom, - // mergeFrom) repoint them under the write lock while the consume path - // holds a read lock, so a consume can never land in a store the handoff - // already drained and abandoned. - // It is always acquired before sweepMu, never while holding it. - storeMu sync.RWMutex - store *sync.Map // map[string]*jtiEntry - atomic storage for consumed token keys - count *atomic.Int64 // entries in store; shared with the store - - // mu guards the lifecycle of the cleanup worker: cleanupCancel and - // cleanupDone. The worker itself only touches store/count, so holding - // mu while waiting for it is safe. - mu sync.Mutex - cleanupCancel context.CancelFunc - cleanupDone chan struct{} // Closed when cleanup goroutine exits - - // sweepMu serializes cache sweeps so at most one scan runs at a time. - sweepMu sync.Mutex -} - -func newJTIReplayCache(name string) *jtiReplayCache { - return &jtiReplayCache{ - name: name, - store: &sync.Map{}, - count: &atomic.Int64{}, - } -} - -// configure applies the replay-protection settings: a previous worker is -// always stopped, whether the new configuration enables protection or -// disables it. The consumed-token store survives re-configuration: resetting -// it would let already-seen tokens be replayed against the refreshed provider. -func (c *jtiReplayCache) configure(enabled bool, maxSize int64) { - c.mu.Lock() - c.stopWorkerLocked() - if enabled { - if maxSize <= 0 { - maxSize = defaultJTICacheSize - } - // the bound must land before the flag: a consume that sees enabled - // with a zero bound would reject the first token it registers - c.maxSize.Store(maxSize) - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan struct{}) - c.cleanupCancel = cancel - c.cleanupDone = done - c.enabled.Store(true) - c.mu.Unlock() - - go c.cleanupLoop(ctx, done, defaultCleanupInterval) - glog.V(2).Infof("OIDC provider %q: JTI replay protection enabled (max size: %d)", c.name, maxSize) - } else { - c.enabled.Store(false) - c.mu.Unlock() - glog.Warningf("OIDC provider %q: JTI replay protection is DISABLED", c.name) - } -} - -// stopWorkerLocked cancels the current cleanup goroutine and waits briefly -// for it to exit. Caller must hold mu. -func (c *jtiReplayCache) stopWorkerLocked() { - if c.cleanupCancel == nil { - return - } - c.cleanupCancel() - select { - case <-c.cleanupDone: - glog.V(3).Infof("OIDC provider %q: JTI cleanup goroutine exited", c.name) - case <-time.After(2 * time.Second): - glog.Warningf("OIDC provider %q: Timeout waiting for JTI cleanup goroutine", c.name) - } - c.cleanupCancel = nil - c.cleanupDone = nil -} - -// consume marks a token as used and rejects a second presentation of it. -// jti is the token's jti claim (may be empty); tokenExpiresAt is its exp. -// A no-op while protection is disabled. -func (c *jtiReplayCache) consume(token, jti, subject string, tokenExpiresAt time.Time) error { - if !c.enabled.Load() { - return nil - } - - key := jti - if key == "" { - // Providers that issue no jti cannot be tracked by claim, so the - // signed input keys the entry: a repeated presentation of the same - // claims is a replay even when the signature bytes differ (e.g. - // ECDSA s/N-s malleability). - signingInput := token - if i := strings.LastIndexByte(token, '.'); i > 0 { - signingInput = token[:i] - } - sum := sha256.Sum256([]byte(signingInput)) - key = tokenFingerprintPrefix + hex.EncodeToString(sum[:]) - } - - if tokenExpiresAt.IsZero() { - // a cache slot is freed only by expiry, so a token without exp - // would hold it forever and eventually exhaust the bound - return fmt.Errorf("%w: missing exp claim", providers.ErrProviderMissingClaims) - } - expiresAt := tokenExpiresAt.Add(clockSkewTolerance) - if !time.Now().Before(expiresAt) { - return fmt.Errorf("%w: token expired", providers.ErrProviderTokenExpired) - } - - entry := &jtiEntry{ - subject: subject, - expiresAt: expiresAt, - } - - // The read lock pins the store/count pair for the whole test-and-set - // so a replay-state handoff cannot repoint them mid-check and strand - // this consume in an abandoned store. - c.storeMu.RLock() - store, count := c.store, c.count - - // Atomic test-and-set detects replays before any capacity work runs, - // so a replayed token cannot trigger cleanup scans. - if _, loaded := store.LoadOrStore(key, entry); loaded { - c.storeMu.RUnlock() - glog.Warningf("OIDC provider %q: Token replay detected (jti: %s)", - c.name, jtiLogKey(key)) - return fmt.Errorf("%w: jti %s has already been used", - providers.ErrProviderTokenReplayed, jtiLogKey(key)) - } - - // The count is incremented as part of the insert, so concurrent - // validations cannot all slip under the limit: whoever pushes the - // count over evicts their own entry and is rejected. - newCount := count.Add(1) - maxSize := c.maxSize.Load() - if newCount > maxSize { - // Over the bound: expired entries may still occupy slots, so - // reclaim them with a synchronized sweep before rejecting. - c.sweepStore(store, count) - if count.Load() > maxSize { - if _, loaded := store.LoadAndDelete(key); loaded { - count.Add(-1) - } - c.storeMu.RUnlock() - glog.Errorf("OIDC provider %q: JTI cache FULL (%d entries) - rejecting token. "+ - "This may indicate a DoS attack or insufficient cache size. Consider increasing JTICacheMaxSize.", - c.name, maxSize) - return fmt.Errorf("JTI cache capacity exceeded") - } - } else if newCount >= maxSize*9/10 { - glog.Warningf("OIDC provider %q: JTI cache at %d%% capacity (%d/%d), triggering eager cleanup", - c.name, (newCount*100)/maxSize, newCount, maxSize) - go c.performEagerCleanup() - } - c.storeMu.RUnlock() - - glog.V(4).Infof("OIDC provider %q: Registered JTI %s (expires at %v, cache size: %d)", - c.name, jtiLogKey(key), expiresAt, newCount) - return nil -} - -// evictExpired removes entries whose expiry has passed and returns how -// many it deleted. Caller must hold sweepMu. -func (c *jtiReplayCache) evictExpired(now time.Time, store *sync.Map, count *atomic.Int64) int { - removed := 0 - store.Range(func(key, value interface{}) bool { - expiresAt := value.(*jtiEntry).expiresAt - if !expiresAt.IsZero() && now.After(expiresAt) { - // LoadAndDelete reports whether this call actually removed the - // entry, so concurrent deletes cannot decrement the count twice. - if v, loaded := store.LoadAndDelete(key); loaded { - if !v.(*jtiEntry).uncounted { - count.Add(-1) - } - removed++ - } - } - return true - }) - return removed -} - -// state returns the cache's current store/count pair. -func (c *jtiReplayCache) state() (*sync.Map, *atomic.Int64) { - c.storeMu.RLock() - defer c.storeMu.RUnlock() - return c.store, c.count -} - -// sweepStore runs one serialized expiry scan over the given store. It only -// takes sweepMu, so callers already holding storeMu may use it. -func (c *jtiReplayCache) sweepStore(store *sync.Map, count *atomic.Int64) { - c.sweepMu.Lock() - defer c.sweepMu.Unlock() - c.lastSweep.Store(time.Now().UnixNano()) - if removed := c.evictExpired(time.Now(), store, count); removed > 0 { - glog.V(4).Infof("OIDC provider %q: Cleaned up %d expired JTI entries (remaining: %d)", - c.name, removed, count.Load()) - } -} - -// sweepExpired runs a scan that is guaranteed to complete before the -// caller proceeds; the periodic worker uses it so the scan always covers -// the store currently in effect, not one captured earlier. -func (c *jtiReplayCache) sweepExpired() { - store, count := c.state() - c.sweepStore(store, count) -} - -// performEagerCleanup performs an opportunistic cleanup of expired JTI -// entries when the cache approaches capacity. At most one sweep runs at a -// time and no more often than eagerCleanupMinInterval, so a burst of -// requests near the limit cannot each turn into a full-cache scan. -func (c *jtiReplayCache) performEagerCleanup() { - if !c.enabled.Load() { - return - } - if time.Since(time.Unix(0, c.lastSweep.Load())) < eagerCleanupMinInterval { - return - } - store, count := c.state() - if !c.sweepMu.TryLock() { - return - } - defer c.sweepMu.Unlock() - c.lastSweep.Store(time.Now().UnixNano()) - - beforeCount := count.Load() - if removed := c.evictExpired(time.Now(), store, count); removed > 0 { - glog.Infof("OIDC provider %q: Eager cleanup removed %d expired JTI entries (%d -> %d)", - c.name, removed, beforeCount, count.Load()) - } else { - glog.Warningf("OIDC provider %q: Eager cleanup found no expired entries to remove. "+ - "Cache may be filled with valid long-lived tokens. Consider increasing JTICacheMaxSize.", - c.name) - } -} - -// cleanupLoop periodically removes expired JTI entries from the cache. -// This prevents memory growth by removing JTIs after their tokens have -// expired. Runs until ctx is cancelled; done is closed on exit. -func (c *jtiReplayCache) cleanupLoop(ctx context.Context, done chan struct{}, interval time.Duration) { - defer close(done) - - ticker := time.NewTicker(interval) - defer ticker.Stop() - - glog.V(3).Infof("OIDC provider %q: JTI cleanup goroutine started (interval: %v)", c.name, interval) - - for { - select { - case <-ctx.Done(): - glog.V(3).Infof("OIDC provider %q: JTI cleanup goroutine stopping", c.name) - return - case <-ticker.C: - c.sweepExpired() - } - } -} - -// stop gracefully stops the cleanup worker. -// Should be called when the provider is being shut down to prevent goroutine -// leaks. -func (c *jtiReplayCache) stop(ctx context.Context) error { - c.mu.Lock() - cancel, done := c.cleanupCancel, c.cleanupDone - c.mu.Unlock() - - if cancel == nil { - return nil - } - cancel() - - select { - case <-done: - glog.V(2).Infof("OIDC provider %q: Cleanup goroutine shutdown complete", c.name) - case <-ctx.Done(): - glog.Warningf("OIDC provider %q: Shutdown timeout waiting for cleanup goroutine: %v", c.name, ctx.Err()) - return ctx.Err() - } - - c.mu.Lock() - if c.cleanupDone == done { - c.cleanupCancel = nil - c.cleanupDone = nil - } - c.mu.Unlock() - return nil -} - -// adoptFrom takes over the consumed-token store of another cache, so tokens -// it already saw stay rejected across the rebuild. The caller must shut down -// the replaced provider; only one cleanup worker may sweep the shared store -// at a time. Call before the new provider serves validators; sweepMu -// excludes its own cleanup worker during the swap. -func (c *jtiReplayCache) adoptFrom(old *jtiReplayCache) { - if old == nil || old == c { - return - } - c.storeMu.Lock() - c.sweepMu.Lock() - c.store, c.count = old.store, old.count - c.sweepMu.Unlock() - c.storeMu.Unlock() -} - -// mergeFrom hands this cache's store to another cache and then folds the -// other's consumed-token entries in, so rejections the other recorded keep -// holding when its records move to a shared provider. Repointing first means -// tokens the other still consumes — in-flight exchanges or ones routed to it -// before the provider map updates — land in the shared store instead of a -// store this merge already drained. Unlike adoptFrom it swaps nothing of its -// own, so it is safe on a cache that is already serving validators. -func (c *jtiReplayCache) mergeFrom(other *jtiReplayCache) { - if other == nil || other == c { - return - } - c.storeMu.RLock() - defer c.storeMu.RUnlock() - store, count := c.store, c.count - - other.storeMu.Lock() - oldStore := other.store - if oldStore == store { - other.storeMu.Unlock() - return - } - other.store, other.count = store, count - other.storeMu.Unlock() - - c.sweepMu.Lock() - defer c.sweepMu.Unlock() - now := time.Now() - oldStore.Range(func(key, value interface{}) bool { - entry := value.(*jtiEntry) - if expiresAt := entry.expiresAt; !expiresAt.IsZero() && now.After(expiresAt) { - return true - } - // A consumed token is never dropped to fit the bound: merged entries - // stay but do not charge the count, so a store filled past the bound - // by the merge cannot reject new exchanges while it drains by expiry. - store.LoadOrStore(key, &jtiEntry{subject: entry.subject, expiresAt: entry.expiresAt, uncounted: true}) - return true - }) -} - -// jtiLogKey redacts a token identifier for logs and error strings: a stable -// fingerprint that still lets operators correlate repeated presentations. -func jtiLogKey(jti string) string { - sum := sha256.Sum256([]byte(jti)) - return hex.EncodeToString(sum[:8]) -} diff --git a/weed/iam/oidc/mock_provider.go b/weed/iam/oidc/mock_provider.go index 5b840eafa7..bc20a5f14f 100644 --- a/weed/iam/oidc/mock_provider.go +++ b/weed/iam/oidc/mock_provider.go @@ -75,17 +75,6 @@ func (m *MockOIDCProvider) Authenticate(ctx context.Context, token string) (*pro return identity, nil } -// AuthenticateSingleUse stays on the mock path: the embedded provider's -// method would bypass the mock's test data and demand a real JWKS. -func (m *MockOIDCProvider) AuthenticateSingleUse(ctx context.Context, token string) (*providers.ExternalIdentity, error) { - return m.Authenticate(ctx, token) -} - -// ValidateTokenSingleUse stays on the mock path; see AuthenticateSingleUse. -func (m *MockOIDCProvider) ValidateTokenSingleUse(ctx context.Context, token string) (*providers.TokenClaims, error) { - return m.ValidateToken(ctx, token) -} - // ValidateToken validates tokens using test data func (m *MockOIDCProvider) ValidateToken(ctx context.Context, token string) (*providers.TokenClaims, error) { if !m.initialized { diff --git a/weed/iam/oidc/oidc_provider.go b/weed/iam/oidc/oidc_provider.go index 6b115041d5..eba00d3559 100644 --- a/weed/iam/oidc/oidc_provider.go +++ b/weed/iam/oidc/oidc_provider.go @@ -27,20 +27,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/security" ) -// OIDCProvider implements OpenID Connect authentication with JTI replay protection. -// -// Replay Protection: -// The provider validates the 'jti' (JWT ID) claim to prevent token replay attacks. -// Each JTI is stored in memory until the token expires. If a token with a duplicate -// JTI is presented, it is rejected with ErrProviderTokenReplayed. -// -// The JTI cache is bounded by JTICacheMaxSize and uses automatic TTL-based expiration. -// A token without a usable jti claim is tracked by the token's own fingerprint -// instead, so a repeated presentation is still detected as a replay. -// -// Thread Safety: -// All methods are safe for concurrent use. The JTI cache uses atomic LoadOrStore -// operations to prevent TOCTOU race conditions. +// OIDCProvider implements OpenID Connect authentication type OIDCProvider struct { name string config *OIDCConfig @@ -57,25 +44,6 @@ type OIDCProvider struct { jwksFetchedAt time.Time resolvedJWKSUri string discoveryFailed bool - - // jti wraps the consumed-token cache behind the replay protection. It is - // allocated by NewOIDCProvider or lazily by replayCache, and survives - // re-initialization, so a config refresh cannot reopen already-consumed - // tokens for replay. jtiOnce makes a zero-value OIDCProvider safe. - jti *jtiReplayCache - jtiOnce sync.Once -} - -// replayCache returns the consumed-token cache, allocating it for providers -// built without NewOIDCProvider so the delegated operations never panic on a -// nil cache. -func (p *OIDCProvider) replayCache() *jtiReplayCache { - p.jtiOnce.Do(func() { - if p.jti == nil { - p.jti = newJTIReplayCache(p.name) - } - }) - return p.jti } // OIDCConfig holds OIDC provider configuration @@ -139,13 +107,6 @@ type OIDCConfig struct { // TLSInsecureSkipVerify controls whether to skip TLS verification. // WARNING: Should only be used in development/testing environments. Never use in production. TLSInsecureSkipVerify bool `json:"tlsInsecureSkipVerify,omitempty"` - - // JTIReplayProtectionEnabled enables token replay protection via JTI tracking (default: true) - // Use pointer to distinguish between "not set" (nil = default true) and "explicitly false" - JTIReplayProtectionEnabled *bool `json:"jtiReplayProtectionEnabled,omitempty"` - - // JTICacheMaxSize sets maximum number of JTI entries to cache (default: 10000) - JTICacheMaxSize int `json:"jtiCacheMaxSize,omitempty"` } // PrincipalTagsClaim is the AWS-defined namespace claim that carries @@ -358,7 +319,6 @@ func NewOIDCProvider(name string) *OIDCProvider { return &OIDCProvider{ name: name, httpClient: &http.Client{Timeout: 30 * time.Second, CheckRedirect: refuseDowngradeRedirect}, - jti: newJTIReplayCache(name), } } @@ -462,6 +422,9 @@ func (p *OIDCProvider) Initialize(config interface{}) error { transport := &http.Transport{ TLSClientConfig: tlsConfig, + // A replaced provider drops its transport without a close; bound + // idle connections so they cannot linger with their goroutines. + IdleConnTimeout: 90 * time.Second, } p.httpClient = &http.Client{ Timeout: 30 * time.Second, @@ -469,17 +432,6 @@ func (p *OIDCProvider) Initialize(config interface{}) error { CheckRedirect: refuseDowngradeRedirect, } - // Configure JTI replay protection last, so a config failure above cannot - // leave a running worker behind. A previous worker is always stopped, - // whether the new configuration enables protection or disables it. - // The consumed-token store survives re-initialization: resetting it would - // let already-seen tokens be replayed against the refreshed provider. - jtiEnabled := true - if oidcConfig.JTIReplayProtectionEnabled != nil { - jtiEnabled = *oidcConfig.JTIReplayProtectionEnabled - } - p.replayCache().configure(jtiEnabled, int64(oidcConfig.JTICacheMaxSize)) - // For testing, we'll skip the actual OIDC client initialization return nil } @@ -530,18 +482,6 @@ func (p *OIDCProvider) allowedAudiences() []string { // Authenticate authenticates a user with an OIDC token func (p *OIDCProvider) Authenticate(ctx context.Context, token string) (*providers.ExternalIdentity, error) { - return p.authenticate(ctx, token, false) -} - -// AuthenticateSingleUse authenticates and consumes the token so a second -// presentation is rejected as a replay. The web-identity exchange uses it, -// where a token buys a session exactly once; callers that validate the same -// token on every request use Authenticate. -func (p *OIDCProvider) AuthenticateSingleUse(ctx context.Context, token string) (*providers.ExternalIdentity, error) { - return p.authenticate(ctx, token, true) -} - -func (p *OIDCProvider) authenticate(ctx context.Context, token string, singleUse bool) (*providers.ExternalIdentity, error) { if !p.initialized { return nil, fmt.Errorf("provider not initialized") } @@ -551,7 +491,7 @@ func (p *OIDCProvider) authenticate(ctx context.Context, token string, singleUse } // Validate token and get claims - claims, err := p.validateToken(ctx, token, singleUse) + claims, err := p.ValidateToken(ctx, token) if err != nil { return nil, err } @@ -718,16 +658,6 @@ func (p *OIDCProvider) getUserInfoWithToken(ctx context.Context, userID, accessT // ValidateToken validates an OIDC JWT token func (p *OIDCProvider) ValidateToken(ctx context.Context, token string) (*providers.TokenClaims, error) { - return p.validateToken(ctx, token, false) -} - -// ValidateTokenSingleUse validates and consumes the token: a second -// presentation of it is rejected as a replay. See AuthenticateSingleUse. -func (p *OIDCProvider) ValidateTokenSingleUse(ctx context.Context, token string) (*providers.TokenClaims, error) { - return p.validateToken(ctx, token, true) -} - -func (p *OIDCProvider) validateToken(ctx context.Context, token string, singleUse bool) (*providers.TokenClaims, error) { if !p.initialized { return nil, fmt.Errorf("provider not initialized") } @@ -754,7 +684,8 @@ func (p *OIDCProvider) validateToken(ctx context.Context, token string, singleUs return nil, fmt.Errorf("failed to get public key: %v", err) } - // Parse and validate token with proper signature verification + // Parse and validate token with proper signature verification. exp is + // required: without it a leaked token could mint STS sessions forever. claims := jwt.MapClaims{} validatedToken, err := jwt.ParseWithClaims(token, claims, func(token *jwt.Token) (interface{}, error) { // Verify signing method @@ -764,7 +695,7 @@ func (p *OIDCProvider) validateToken(ctx context.Context, token string, singleUs default: return nil, fmt.Errorf("unsupported signing method: %v", token.Header["alg"]) } - }) + }, jwt.WithExpirationRequired()) if err != nil { // Use JWT library's typed errors for robust error checking @@ -860,16 +791,6 @@ func (p *OIDCProvider) validateToken(ctx context.Context, token string, singleUs tokenClaims.Claims[key] = value } - // Consume the token for replay protection. The check runs only for - // single-use callers: a bearer token is meant to be presented on every - // request, so validating it must not consume it. - if singleUse { - jti, _ := claims["jti"].(string) - if err := p.replayCache().consume(token, jti, subject, tokenClaims.ExpiresAt); err != nil { - return nil, err - } - } - return tokenClaims, nil } @@ -1298,36 +1219,3 @@ func (p *OIDCProvider) mapUserInfoToIdentity(userInfo map[string]interface{}) *p return identity } - -// Shutdown gracefully stops the JTI cleanup goroutine. -// Should be called when the provider is being shut down to prevent goroutine leaks. -func (p *OIDCProvider) Shutdown(ctx context.Context) error { - return p.replayCache().stop(ctx) -} - -// AdoptReplayStateFrom takes over the consumed-token store of the provider it -// replaces, so tokens it already saw stay rejected across the rebuild. The -// caller must shut down the replaced provider; only one cleanup worker may -// sweep the shared store at a time. Call before the new provider serves -// validators. -func (p *OIDCProvider) AdoptReplayStateFrom(old *OIDCProvider) { - if old == nil || old == p || old.jti == nil { - return - } - p.replayCache().adoptFrom(old.jti) -} - -// MergeReplayStateFrom hands this provider's store to another provider and -// then folds the other's consumed-token entries in, so rejections the other -// recorded keep holding when its records move to a shared provider. Repointing -// first means tokens the other still consumes — in-flight exchanges or ones -// routed to it before the provider map updates — land in the shared store -// instead of a store this merge already drained. Unlike AdoptReplayStateFrom -// it swaps nothing of its own, so it is safe on a provider that is already -// serving validators. -func (p *OIDCProvider) MergeReplayStateFrom(other *OIDCProvider) { - if other == nil || other == p { - return - } - p.replayCache().mergeFrom(other.replayCache()) -} diff --git a/weed/iam/oidc/oidc_provider_test.go b/weed/iam/oidc/oidc_provider_test.go index 29f7e3e656..83e28ed8cc 100644 --- a/weed/iam/oidc/oidc_provider_test.go +++ b/weed/iam/oidc/oidc_provider_test.go @@ -8,14 +8,9 @@ import ( "crypto/rsa" "encoding/base64" "encoding/json" - "errors" - "fmt" - "math/big" "net/http" "net/http/httptest" "strings" - "sync" - "sync/atomic" "testing" "time" @@ -183,6 +178,18 @@ func TestOIDCProviderJWTValidation(t *testing.T) { assert.Contains(t, err.Error(), "expired") }) + t.Run("missing exp", func(t *testing.T) { + token := createTestJWT(t, privateKey, jwt.MapClaims{ + "iss": server.URL, + "aud": "test-client", + "sub": "user123", + "iat": time.Now().Unix(), + }) + + _, err := provider.ValidateToken(context.Background(), token) + assert.Error(t, err) + }) + t.Run("invalid signature", func(t *testing.T) { // Create token with wrong key wrongKey, _ := generateTestKeys(t) @@ -552,42 +559,42 @@ func TestOIDCProviderUserInfo(t *testing.T) { // Helper functions for testing -func generateTestKeys(tb testing.TB) (*rsa.PrivateKey, *rsa.PublicKey) { +func generateTestKeys(t *testing.T) (*rsa.PrivateKey, *rsa.PublicKey) { privateKey, err := rsa.GenerateKey(rand.Reader, 2048) - require.NoError(tb, err) + require.NoError(t, err) return privateKey, &privateKey.PublicKey } -func generateTestECKeys(tb testing.TB) (*ecdsa.PrivateKey, *ecdsa.PublicKey) { +func generateTestECKeys(t *testing.T) (*ecdsa.PrivateKey, *ecdsa.PublicKey) { privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) - require.NoError(tb, err) + require.NoError(t, err) return privateKey, &privateKey.PublicKey } -func createTestJWT(tb testing.TB, privateKey *rsa.PrivateKey, claims jwt.MapClaims) string { +func createTestJWT(t *testing.T, privateKey *rsa.PrivateKey, claims jwt.MapClaims) string { token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims) token.Header["kid"] = "test-key-id" tokenString, err := token.SignedString(privateKey) - require.NoError(tb, err) + require.NoError(t, err) return tokenString } -func createTestECDSAJWT(tb testing.TB, privateKey *ecdsa.PrivateKey, claims jwt.MapClaims) string { +func createTestECDSAJWT(t *testing.T, privateKey *ecdsa.PrivateKey, claims jwt.MapClaims) string { token := jwt.NewWithClaims(jwt.SigningMethodES256, claims) token.Header["kid"] = "test-ec-key-id" tokenString, err := token.SignedString(privateKey) - require.NoError(tb, err) + require.NoError(t, err) return tokenString } -func encodePublicKey(tb testing.TB, publicKey *rsa.PublicKey) string { +func encodePublicKey(t *testing.T, publicKey *rsa.PublicKey) string { // Properly encode the RSA modulus (N) as base64url return base64.RawURLEncoding.EncodeToString(publicKey.N.Bytes()) } -func encodeECPublicKey(tb testing.TB, publicKey *ecdsa.PublicKey) (string, string) { +func encodeECPublicKey(t *testing.T, publicKey *ecdsa.PublicKey) (string, string) { // RFC 7518 §6.2.1.2 requires EC coordinates to be zero-padded to the full field size curveParams := publicKey.Curve.Params() size := (curveParams.BitSize + 7) / 8 @@ -602,7 +609,7 @@ func encodeECPublicKey(tb testing.TB, publicKey *ecdsa.PublicKey) (string, strin base64.RawURLEncoding.EncodeToString(yPadded) } -func setupOIDCTestServer(tb testing.TB, publicKey *rsa.PublicKey) *httptest.Server { +func setupOIDCTestServer(t *testing.T, publicKey *rsa.PublicKey) *httptest.Server { jwks := map[string]interface{}{ "keys": []map[string]interface{}{ { @@ -610,7 +617,7 @@ func setupOIDCTestServer(tb testing.TB, publicKey *rsa.PublicKey) *httptest.Serv "kid": "test-key-id", "use": "sig", "alg": "RS256", - "n": encodePublicKey(tb, publicKey), + "n": encodePublicKey(t, publicKey), "e": "AQAB", }, }, @@ -665,1066 +672,3 @@ func setupOIDCTestServer(tb testing.TB, publicKey *rsa.PublicKey) *httptest.Serv } })) } - -func TestOIDCProviderJTIReplayProtection(t *testing.T) { - // Setup mock OIDC provider with JTI protection enabled - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 100, - } - - err := provider.Initialize(config) - require.NoError(t, err) - - // Test 1: First use of token with JTI should succeed - jti := "unique-token-id-12345" - token1 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": jti, - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - claims1, err := provider.ValidateTokenSingleUse(context.Background(), token1) - require.NoError(t, err) - assert.Equal(t, "user123", claims1.Subject) - assert.Equal(t, jti, claims1.Claims["jti"]) - - // Test 2: Replay of same token should be rejected - claims2, err := provider.ValidateTokenSingleUse(context.Background(), token1) - assert.Error(t, err) - assert.Nil(t, claims2) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed)) - assert.NotContains(t, err.Error(), jti, "error should not expose the raw token identifier") - - // Test 3: Different token with different JTI should succeed - jti2 := "different-token-id-67890" - token2 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user456", - "jti": jti2, - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - claims3, err := provider.ValidateTokenSingleUse(context.Background(), token2) - require.NoError(t, err) - assert.Equal(t, "user456", claims3.Subject) - - // Test 4: Token without JTI is tracked by the token's own fingerprint, - // so presenting the same token again is still a replay - tokenNoJTI := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user789", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - claims4, err := provider.ValidateTokenSingleUse(context.Background(), tokenNoJTI) - require.NoError(t, err) - assert.Equal(t, "user789", claims4.Subject) - - _, err = provider.ValidateTokenSingleUse(context.Background(), tokenNoJTI) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "a jti-less token must not stay reusable while protection is enabled") - - // Test 5: Token with a JTI but no exp is rejected — a cache slot frees - // only by expiry, so an expiry-less entry would hold it forever and a - // burst of distinct tokens could exhaust the bound permanently. - tokenNoExp := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "userNoExp", - "jti": "no-exp-jti", - "iat": time.Now().Unix(), - }) - - claims5, err := provider.ValidateTokenSingleUse(context.Background(), tokenNoExp) - assert.Nil(t, claims5) - assert.True(t, errors.Is(err, providers.ErrProviderMissingClaims)) -} - -func TestOIDCProviderJTIFallbackResistsSignatureMalleability(t *testing.T) { - privateKey, publicKey := generateTestECKeys(t) - x, y := encodeECPublicKey(t, publicKey) - - jwks := map[string]interface{}{ - "keys": []map[string]interface{}{ - { - "kty": "EC", - "kid": "test-ec-key-id", - "use": "sig", - "alg": "ES256", - "crv": "P-256", - "x": x, - "y": y, - }, - }, - } - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/.well-known/openid_configuration" { - json.NewEncoder(w).Encode(map[string]interface{}{ - "issuer": "http://" + r.Host, - "jwks_uri": "http://" + r.Host + "/jwks", - }) - } else if r.URL.Path == "/jwks" { - json.NewEncoder(w).Encode(jwks) - } - })) - defer server.Close() - - provider := NewOIDCProvider("test-oidc-ecdsa-malleable") - err := provider.Initialize(&OIDCConfig{ - Issuer: server.URL, - ClientID: "test-client", - JWKSUri: server.URL + "/jwks", - }) - require.NoError(t, err) - - token := createTestECDSAJWT(t, privateKey, jwt.MapClaims{ - "iss": server.URL, - "aud": "test-client", - "sub": "user-malleable", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - _, err = provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - - // The same signed claims under a malleated ECDSA signature (s -> N-s) - // must still be a replay, not a fresh exchange. - malleated := malleateECDSAJWTSignature(t, token) - require.NotEqual(t, token, malleated) - - _, err = provider.ValidateTokenSingleUse(context.Background(), malleated) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "a signature-malleated jti-less token must not bypass replay detection") -} - -func malleateECDSAJWTSignature(tb testing.TB, tokenString string) string { - parts := strings.Split(tokenString, ".") - require.Len(tb, parts, 3) - sig, err := base64.RawURLEncoding.DecodeString(parts[2]) - require.NoError(tb, err) - half := len(sig) / 2 - r := new(big.Int).SetBytes(sig[:half]) - s := new(big.Int).SetBytes(sig[half:]) - s.Sub(elliptic.P256().Params().N, s) - malleated := make([]byte, len(sig)) - r.FillBytes(malleated[:half]) - s.FillBytes(malleated[half:]) - parts[2] = base64.RawURLEncoding.EncodeToString(malleated) - return strings.Join(parts, ".") -} - -func TestOIDCProviderJTIReplayProtectionDisabled(t *testing.T) { - // Setup provider with JTI protection DISABLED - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-no-jti") - jtiDisabled := false - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiDisabled, // Explicitly disabled - } - - err := provider.Initialize(config) - require.NoError(t, err) - - // Token with JTI can be used multiple times when protection is disabled - jti := "reusable-token-id" - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": jti, - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - // First use - should succeed - claims1, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - assert.Equal(t, "user123", claims1.Subject) - - // Second use - should ALSO succeed (no replay protection) - claims2, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - assert.Equal(t, "user123", claims2.Subject) -} - -func TestOIDCProviderJTIExpiration(t *testing.T) { - // Test that JTI cache entries expire with token - // This test also verifies default behavior (JTIReplayProtectionEnabled not set = enabled by default) - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc") - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - // JTIReplayProtectionEnabled not set - should default to true (secure by default) - } - - err := provider.Initialize(config) - require.NoError(t, err) - - // Create token that expires in 2 seconds - jti := "short-lived-token" - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": jti, - "exp": time.Now().Add(2 * time.Second).Unix(), - "iat": time.Now().Unix(), - }) - - // First use - should succeed - claims1, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - assert.Equal(t, "user123", claims1.Subject) - - // Immediate replay - should be rejected - _, err = provider.ValidateTokenSingleUse(context.Background(), token) - assert.Error(t, err) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed)) - - // Wait for token AND cache entry to expire - time.Sleep(3 * time.Second) - - // Now validation should fail due to EXPIRATION (not replay) - _, err = provider.ValidateTokenSingleUse(context.Background(), token) - assert.Error(t, err) - assert.True(t, errors.Is(err, providers.ErrProviderTokenExpired)) -} - -func TestOIDCProviderJTIReplayProtectionConcurrent(t *testing.T) { - // Test concurrent validation to ensure no TOCTOU race conditions - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-concurrent") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 1000, - } - require.NoError(t, provider.Initialize(config)) - - jti := "concurrent-test-jti" - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": jti, - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - // Launch 100 concurrent validations with the same token - const numGoroutines = 100 - var successCount, replayCount, otherErrorCount atomic.Int32 - - var wg sync.WaitGroup - wg.Add(numGoroutines) - - // All goroutines start at roughly the same time - startBarrier := make(chan struct{}) - - for i := 0; i < numGoroutines; i++ { - go func() { - defer wg.Done() - <-startBarrier // Wait for start signal - - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - if err == nil { - successCount.Add(1) - } else if errors.Is(err, providers.ErrProviderTokenReplayed) { - replayCount.Add(1) - } else { - otherErrorCount.Add(1) - t.Logf("Unexpected error: %v", err) - } - }() - } - - close(startBarrier) // Start all goroutines - wg.Wait() - - // Exactly one should succeed, all others should be replay errors - assert.Equal(t, int32(1), successCount.Load(), - "Exactly one validation should succeed (atomic test-and-set)") - assert.Equal(t, int32(numGoroutines-1), replayCount.Load(), - "All other validations should be replay errors") - assert.Equal(t, int32(0), otherErrorCount.Load(), - "No other errors should occur") -} - -func TestOIDCProviderJTICacheMaxSize(t *testing.T) { - // Test that cache respects max size limit - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-maxsize") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 10, // Small cache for testing - } - require.NoError(t, provider.Initialize(config)) - - // Add tokens up to the limit - for i := 0; i < 10; i++ { - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": fmt.Sprintf("jti-%d", i), - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err, "Should accept token %d", i) - } - - // Next token should be rejected due to cache full - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "jti-overflow", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - assert.Error(t, err) - assert.Contains(t, err.Error(), "cache capacity exceeded") -} - -func TestOIDCProviderJTICleanup(t *testing.T) { - // Test that cleanup goroutine logic correctly identifies expired entries - // NOTE: JWT library allows clock skew, so tokens can't be truly expired - // We test the cleanup logic without actually waiting for expiration - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-cleanup") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - } - require.NoError(t, provider.Initialize(config)) - defer provider.Shutdown(context.Background()) - - // Create token that expires in 1 hour - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "cleanup-test-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - // Validation should succeed - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - - // Check cache size is 1 - assert.Equal(t, int64(1), provider.jti.count.Load()) - - // Verify entry exists and check its expiration includes clock skew - val, ok := provider.jti.store.Load("cleanup-test-jti") - require.True(t, ok, "JTI entry should exist") - entry := val.(*jtiEntry) - - // Entry should expire around exp + clock skew tolerance (1 hour + 5 minutes) - expectedExpiry := time.Now().Add(time.Hour + 5*time.Minute) - assert.WithinDuration(t, expectedExpiry, entry.expiresAt, 10*time.Second, - "Entry expiry should be token exp + clock skew tolerance") - - // Verify entry is not yet expired (cleanup shouldn't remove it) - now := time.Now() - assert.True(t, now.Before(entry.expiresAt), - "Entry should not be expired yet") - - // Run the periodic cleanup with a short interval and an expired entry: - // both the entry and the count must actually be cleared - cleanupCtx, cleanupCancel := context.WithCancel(context.Background()) - cleanupDone := make(chan struct{}) - go provider.jti.cleanupLoop(cleanupCtx, cleanupDone, 20*time.Millisecond) - defer func() { - cleanupCancel() - <-cleanupDone - }() - - provider.jti.store.Store("expired-cleanup-jti", &jtiEntry{ - subject: "expired-user", - expiresAt: time.Now().Add(-time.Minute), - }) - provider.jti.count.Add(1) - - require.Eventually(t, func() bool { - _, stillThere := provider.jti.store.Load("expired-cleanup-jti") - return !stillThere && provider.jti.count.Load() == 1 // only the valid entry remains - }, 2*time.Second, 10*time.Millisecond, "periodic cleanup should evict the expired entry") -} - -func TestOIDCProviderReinitialization(t *testing.T) { - // Test that re-initialization properly cancels the old cleanup goroutine - // to prevent goroutine leaks - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-reinit") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - } - - // First initialization - require.NoError(t, provider.Initialize(config)) - require.NotNil(t, provider.jti.cleanupCancel, "First cleanup cancel should be set") - - // Create and validate a token to ensure JTI protection is working - token1 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user1", - "jti": "token-1", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := provider.ValidateTokenSingleUse(context.Background(), token1) - require.NoError(t, err, "First token should validate successfully") - - // Second initialization should not cause errors - require.NoError(t, provider.Initialize(config)) - require.NotNil(t, provider.jti.cleanupCancel, "Second cleanup cancel should be set") - - // The JTI cache survives re-initialization, so the consumed token is - // still detected as a replay rather than being usable again - _, err = provider.ValidateTokenSingleUse(context.Background(), token1) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "Re-initialization must not reopen a consumed token for replay") - - // Third initialization to ensure it works multiple times - require.NoError(t, provider.Initialize(config)) - require.NotNil(t, provider.jti.cleanupCancel, "Third cleanup cancel should be set") - - // Verify JTI replay protection still works after re-initialization - token2 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user2", - "jti": "token-2", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err = provider.ValidateTokenSingleUse(context.Background(), token2) - require.NoError(t, err, "New token should validate") - - // Replay should still be detected - _, err = provider.ValidateTokenSingleUse(context.Background(), token2) - assert.Error(t, err, "Replay should be detected") - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), "Should be a replay error") - - // Final cleanup - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - err = provider.Shutdown(ctx) - assert.NoError(t, err) -} - -func TestOIDCProviderJTIEagerCleanup(t *testing.T) { - // Test the eager cleanup mechanism when cache approaches capacity - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-eager-cleanup") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 10, // Small cache for testing - } - require.NoError(t, provider.Initialize(config)) - defer provider.Shutdown(context.Background()) - - // Manually add expired entries to the cache - for i := 0; i < 5; i++ { - jti := fmt.Sprintf("expired-jti-%d", i) - entry := &jtiEntry{ - subject: fmt.Sprintf("expired-user-%d", i), - expiresAt: time.Now().Add(-time.Hour), // Already expired - } - provider.jti.store.Store(jti, entry) - provider.jti.count.Add(1) - } - - assert.Equal(t, int64(5), provider.jti.count.Load(), "Should have 5 expired entries") - - // Perform eager cleanup - should remove all expired entries - provider.jti.performEagerCleanup() - - assert.Equal(t, int64(0), provider.jti.count.Load(), - "All expired entries should be removed by eager cleanup") - - // Now test that adding valid tokens works after cleanup - for i := 0; i < 9; i++ { - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": fmt.Sprintf("user-%d", i), - "jti": fmt.Sprintf("valid-jti-%d", i), - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - } - - // Should have 9 valid entries now (90% of capacity, triggers eager cleanup warning) - assert.Equal(t, int64(9), provider.jti.count.Load(), "Should have 9 valid entries") - - // 10th token should succeed (at capacity but not over) - token10 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user-10", - "jti": "valid-jti-10", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := provider.ValidateTokenSingleUse(context.Background(), token10) - require.NoError(t, err, "10th token should validate (at capacity)") - - // 11th token should fail (over capacity) - token11 := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user-11", - "jti": "valid-jti-11", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err = provider.ValidateTokenSingleUse(context.Background(), token11) - assert.Error(t, err, "11th token should fail (over capacity)") - assert.Contains(t, err.Error(), "capacity exceeded", "Should indicate capacity exceeded") -} - -func TestOIDCProviderConcurrentCleanupNoNegativeCount(t *testing.T) { - // Test that concurrent cleanups don't cause jtiCount to go negative - // due to double-decrement of the same entry - _, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-cleanup-race") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 100, - } - require.NoError(t, provider.Initialize(config)) - defer provider.Shutdown(context.Background()) - - // Manually add expired entries to the cache - for i := 0; i < 50; i++ { - jti := fmt.Sprintf("expired-jti-%d", i) - entry := &jtiEntry{ - subject: fmt.Sprintf("user-%d", i), - expiresAt: time.Now().Add(-time.Hour), // Already expired - } - provider.jti.store.Store(jti, entry) - provider.jti.count.Add(1) - } - - initialCount := provider.jti.count.Load() - assert.Equal(t, int64(50), initialCount, "Should have 50 expired entries") - - // Run multiple concurrent cleanups - var wg sync.WaitGroup - numCleanups := 10 - - for i := 0; i < numCleanups; i++ { - wg.Add(1) - go func() { - defer wg.Done() - provider.jti.performEagerCleanup() - }() - } - - wg.Wait() - - // Counter should be 0 (all entries removed), never negative - finalCount := provider.jti.count.Load() - assert.Equal(t, int64(0), finalCount, - "Counter should be 0 after concurrent cleanups, got %d", finalCount) - assert.GreaterOrEqual(t, finalCount, int64(0), - "Counter should never go negative, got %d", finalCount) - - // Verify store is actually empty - storeCount := 0 - provider.jti.store.Range(func(key, value interface{}) bool { - storeCount++ - return true - }) - assert.Equal(t, 0, storeCount, "Store should be empty") -} - -func TestOIDCProviderShutdown(t *testing.T) { - // Test graceful shutdown - _, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-shutdown") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - } - require.NoError(t, provider.Initialize(config)) - - // Shutdown should complete without error - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - // Measure shutdown time to ensure it doesn't block unnecessarily - start := time.Now() - err := provider.Shutdown(ctx) - duration := time.Since(start) - - assert.NoError(t, err) - // Shutdown should be quick (goroutine exits immediately on cancel) - // Allow 1 second for goroutine scheduling overhead - assert.Less(t, duration, 1*time.Second, - "Shutdown should not block for fixed timeout, took %v", duration) - - // Second shutdown should be idempotent - err = provider.Shutdown(ctx) - assert.NoError(t, err) -} - -func TestOIDCProviderShutdownTimeout(t *testing.T) { - // Test that shutdown respects caller's context timeout - _, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-shutdown-timeout") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - } - require.NoError(t, provider.Initialize(config)) - - // Use a very short timeout to test timeout handling - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond) - defer cancel() - - // Wait a bit to ensure context expires - time.Sleep(10 * time.Millisecond) - - err := provider.Shutdown(ctx) - // Should return context error if timeout occurs - if err != nil { - assert.ErrorIs(t, err, context.DeadlineExceeded, - "Should return deadline exceeded error when context times out") - } - // Note: goroutine may exit before timeout, so error is optional -} - -// Benchmark tests - -func BenchmarkJTIValidation(b *testing.B) { - // Benchmark JTI validation performance with unique tokens - privateKey, publicKey := generateTestKeys(b) - jwksServer := setupOIDCTestServer(b, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("bench-oidc") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: b.N + 1024, - } - require.NoError(b, provider.Initialize(config)) - defer provider.Shutdown(context.Background()) - - // Pre-generate unique tokens - tokens := make([]string, b.N) - for i := 0; i < b.N; i++ { - tokens[i] = createTestJWT(b, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": fmt.Sprintf("jti-%d", i), - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - } - - // Parallel workers draw distinct token indices so each iteration - // validates a new token rather than a replay or capacity rejection. - var tokenIndex atomic.Int64 - b.ResetTimer() - b.RunParallel(func(pb *testing.PB) { - for pb.Next() { - i := int(tokenIndex.Add(1)-1) % len(tokens) - if _, err := provider.ValidateTokenSingleUse(context.Background(), tokens[i]); err != nil { - b.Errorf("unexpected validation error: %v", err) - } - } - }) -} - -func BenchmarkJTIValidationWithReplay(b *testing.B) { - // Benchmark replay detection performance - privateKey, publicKey := generateTestKeys(b) - jwksServer := setupOIDCTestServer(b, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("bench-oidc-replay") - jtiEnabled := true - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - JTICacheMaxSize: 100000, - } - require.NoError(b, provider.Initialize(config)) - defer provider.Shutdown(context.Background()) - - // Create one token that will be replayed - token := createTestJWT(b, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "replay-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - // First validation to store the JTI - _, err := provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(b, err) - - b.ResetTimer() - b.RunParallel(func(pb *testing.PB) { - for pb.Next() { - // All subsequent validations should be fast replay rejections - provider.ValidateTokenSingleUse(context.Background(), token) - } - }) -} - -func BenchmarkJTIValidationNoProtection(b *testing.B) { - // Benchmark validation without JTI protection (baseline) - privateKey, publicKey := generateTestKeys(b) - jwksServer := setupOIDCTestServer(b, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("bench-oidc-no-jti") - jtiDisabled := false - config := &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiDisabled, - } - require.NoError(b, provider.Initialize(config)) - - token := createTestJWT(b, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "no-protection-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - b.ResetTimer() - b.RunParallel(func(pb *testing.PB) { - for pb.Next() { - provider.ValidateTokenSingleUse(context.Background(), token) - } - }) -} - -func TestOIDCProviderBearerValidationDoesNotConsume(t *testing.T) { - // Bearer authentication validates the same token on every request, so it - // must not consume it; only the single-use exchange rejects a repeat. - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - provider := NewOIDCProvider("test-oidc-bearer") - jtiEnabled := true - require.NoError(t, provider.Initialize(&OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTIReplayProtectionEnabled: &jtiEnabled, - })) - defer provider.Shutdown(context.Background()) - - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "bearer-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - - for i := 0; i < 3; i++ { - _, err := provider.ValidateToken(context.Background(), token) - require.NoError(t, err, "bearer validation must not consume the token") - } - - _, err := provider.Authenticate(context.Background(), token) - require.NoError(t, err, "bearer authentication must not consume the token") - - // The same token then exchanged single-use is tracked and a repeat rejected. - _, err = provider.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - _, err = provider.ValidateTokenSingleUse(context.Background(), token) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed)) - - // Even after the exchange consumed it, bearer validation still accepts: - // the token itself remains valid until it expires. - _, err = provider.ValidateToken(context.Background(), token) - require.NoError(t, err) -} - -func TestAdoptReplayStateFromKeepsConsumedTokensRejected(t *testing.T) { - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - newConfig := func() *OIDCConfig { - return &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - } - } - - old := NewOIDCProvider("test-oidc-old") - require.NoError(t, old.Initialize(newConfig())) - - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "consumed-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := old.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - - rebuilt := NewOIDCProvider("test-oidc-new") - require.NoError(t, rebuilt.Initialize(newConfig())) - defer rebuilt.Shutdown(context.Background()) - - rebuilt.AdoptReplayStateFrom(old) - require.NoError(t, old.Shutdown(context.Background())) - - _, err = rebuilt.ValidateTokenSingleUse(context.Background(), token) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "a token consumed before the rebuild must stay rejected") -} - -// A token consumed by one provider must stay rejected by another provider for -// the same issuer once the consumed entries merge over: pooling a record onto -// a shared provider cannot reopen the replay window. -func TestMergeReplayStateFromKeepsConsumedTokensRejected(t *testing.T) { - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - newConfig := func() *OIDCConfig { - return &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - } - } - - first := NewOIDCProvider("test-oidc-first") - require.NoError(t, first.Initialize(newConfig())) - defer first.Shutdown(context.Background()) - - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "consumed-elsewhere-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := first.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err) - - second := NewOIDCProvider("test-oidc-second") - require.NoError(t, second.Initialize(newConfig())) - defer second.Shutdown(context.Background()) - - second.MergeReplayStateFrom(first) - _, err = second.ValidateTokenSingleUse(context.Background(), token) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "a token the superseded provider consumed must stay rejected after the merge") -} - -// The superseded provider may keep serving exchanges while the provider map -// updates; tokens it consumes after the handoff must land in the shared store -// or they would reopen for replay. -func TestMergeReplayStateFromSharesLiveStore(t *testing.T) { - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - newConfig := func() *OIDCConfig { - return &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - } - } - - first := NewOIDCProvider("test-oidc-first") - require.NoError(t, first.Initialize(newConfig())) - defer first.Shutdown(context.Background()) - - second := NewOIDCProvider("test-oidc-second") - require.NoError(t, second.Initialize(newConfig())) - defer second.Shutdown(context.Background()) - - second.MergeReplayStateFrom(first) - - token := createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": "late-consume-jti", - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - _, err := first.ValidateTokenSingleUse(context.Background(), token) - require.NoError(t, err, "the superseded provider still accepts a fresh token") - - _, err = second.ValidateTokenSingleUse(context.Background(), token) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "a token consumed through the superseded provider after the handoff must stay rejected") -} - -// Merging two stores that are each near the bound must not reject new -// exchanges: consumed tokens are carried over without charging the bound, -// which still applies to tokens consumed through the live path. -func TestMergeReplayStateFromOverflowStillAdmits(t *testing.T) { - privateKey, publicKey := generateTestKeys(t) - jwksServer := setupOIDCTestServer(t, publicKey) - defer jwksServer.Close() - - newConfig := func() *OIDCConfig { - return &OIDCConfig{ - Issuer: jwksServer.URL, - ClientID: "test-client", - JWKSUri: jwksServer.URL + "/jwks", - JTICacheMaxSize: 4, - } - } - newToken := func(jti string) string { - return createTestJWT(t, privateKey, jwt.MapClaims{ - "iss": jwksServer.URL, - "aud": "test-client", - "sub": "user123", - "jti": jti, - "exp": time.Now().Add(time.Hour).Unix(), - "iat": time.Now().Unix(), - }) - } - - first := NewOIDCProvider("test-oidc-first") - require.NoError(t, first.Initialize(newConfig())) - defer first.Shutdown(context.Background()) - for _, jti := range []string{"first-a", "first-b", "first-c"} { - _, err := first.ValidateTokenSingleUse(context.Background(), newToken(jti)) - require.NoError(t, err) - } - - second := NewOIDCProvider("test-oidc-second") - require.NoError(t, second.Initialize(newConfig())) - defer second.Shutdown(context.Background()) - for _, jti := range []string{"second-a", "second-b", "second-c"} { - _, err := second.ValidateTokenSingleUse(context.Background(), newToken(jti)) - require.NoError(t, err) - } - - second.MergeReplayStateFrom(first) - - _, err := second.ValidateTokenSingleUse(context.Background(), newToken("first-a")) - assert.True(t, errors.Is(err, providers.ErrProviderTokenReplayed), - "merged consumed tokens must stay rejected") - - _, err = second.ValidateTokenSingleUse(context.Background(), newToken("fresh-jti")) - assert.NoError(t, err, "merged excess must not reject a new exchange") -} diff --git a/weed/iam/providers/errors.go b/weed/iam/providers/errors.go index 02ae7df532..eeac47c521 100644 --- a/weed/iam/providers/errors.go +++ b/weed/iam/providers/errors.go @@ -19,7 +19,4 @@ var ( // ErrProviderMissingClaims indicates that required claims are missing from the token ErrProviderMissingClaims = errors.New("provider: missing required claims") - - // ErrProviderTokenReplayed is returned when a token with duplicate JTI is detected - ErrProviderTokenReplayed = errors.New("provider: token has been replayed") ) diff --git a/weed/iam/sts/constants.go b/weed/iam/sts/constants.go index 3e6e24df39..cff6b352e1 100644 --- a/weed/iam/sts/constants.go +++ b/weed/iam/sts/constants.go @@ -42,23 +42,21 @@ const ( // Configuration Field Names const ( - ConfigFieldFilerAddress = "filerAddress" - ConfigFieldBasePath = "basePath" - ConfigFieldIssuer = "issuer" - ConfigFieldClientID = "clientId" - ConfigFieldClientIDs = "clientIds" - ConfigFieldThumbprints = "thumbprints" - ConfigFieldAllowedPrincipalTagKeys = "allowedPrincipalTagKeys" - ConfigFieldPolicyClaim = "policyClaim" - ConfigFieldClientSecret = "clientSecret" - ConfigFieldJWKSUri = "jwksUri" - ConfigFieldScopes = "scopes" - ConfigFieldUserInfoUri = "userInfoUri" - ConfigFieldRedirectUri = "redirectUri" - ConfigFieldTLSCACert = "tlsCaCert" - ConfigFieldTLSInsecureSkipVerify = "tlsInsecureSkipVerify" - ConfigFieldJTIReplayProtectionEnabled = "jtiReplayProtectionEnabled" - ConfigFieldJTICacheMaxSize = "jtiCacheMaxSize" + ConfigFieldFilerAddress = "filerAddress" + ConfigFieldBasePath = "basePath" + ConfigFieldIssuer = "issuer" + ConfigFieldClientID = "clientId" + ConfigFieldClientIDs = "clientIds" + ConfigFieldThumbprints = "thumbprints" + ConfigFieldAllowedPrincipalTagKeys = "allowedPrincipalTagKeys" + ConfigFieldPolicyClaim = "policyClaim" + ConfigFieldClientSecret = "clientSecret" + ConfigFieldJWKSUri = "jwksUri" + ConfigFieldScopes = "scopes" + ConfigFieldUserInfoUri = "userInfoUri" + ConfigFieldRedirectUri = "redirectUri" + ConfigFieldTLSCACert = "tlsCaCert" + ConfigFieldTLSInsecureSkipVerify = "tlsInsecureSkipVerify" ) // Error Messages diff --git a/weed/iam/sts/oidc_provider_lookup_test.go b/weed/iam/sts/oidc_provider_lookup_test.go index aba144be2b..104d622a7f 100644 --- a/weed/iam/sts/oidc_provider_lookup_test.go +++ b/weed/iam/sts/oidc_provider_lookup_test.go @@ -103,51 +103,3 @@ func TestLookupOIDCProviderForAccountFallsBackToStatic(t *testing.T) { t.Fatalf("expected static-config provider as last resort, got %v ok=%v", got, ok) } } - -// shutdownTrackingProvider records Shutdown calls so tests can observe which -// replaced providers the swap released. -type shutdownTrackingProvider struct { - stubIdentityProvider - shutdownCalls int32 -} - -func (s *shutdownTrackingProvider) Shutdown(context.Context) error { - s.shutdownCalls++ - return nil -} - -// A provider dropped from the managed map owns a background JTI cleanup -// worker; replacing the map must shut it down, while providers that stay in -// the map keep running. -func TestSetIAMManagedOIDCProvidersShutsDownDroppedProviders(t *testing.T) { - dropped := &shutdownTrackingProvider{stubIdentityProvider{name: "dropped"}, 0} - kept := &shutdownTrackingProvider{stubIdentityProvider{name: "kept"}, 0} - added := &shutdownTrackingProvider{stubIdentityProvider{name: "added"}, 0} - - s := &STSService{issuerToProvider: map[string]providers.IdentityProvider{}} - s.SetIAMManagedOIDCProviders(map[string][]ScopedOIDCProvider{ - "https://a.example.com": {{Provider: dropped}}, - "https://b.example.com": {{Provider: kept}, {AccountID: "111111111111", Provider: kept}}, - }) - - s.SetIAMManagedOIDCProviders(map[string][]ScopedOIDCProvider{ - "https://b.example.com": {{Provider: kept}}, - "https://c.example.com": {{Provider: added}}, - }) - - if dropped.shutdownCalls != 1 { - t.Fatalf("dropped provider was shut down %d times, want 1", dropped.shutdownCalls) - } - if kept.shutdownCalls != 0 { - t.Fatalf("kept provider was shut down %d times, want 0", kept.shutdownCalls) - } - if added.shutdownCalls != 0 { - t.Fatalf("newly added provider was shut down %d times, want 0", added.shutdownCalls) - } - - // Clearing the map releases the remaining providers. - s.SetIAMManagedOIDCProviders(nil) - if kept.shutdownCalls != 1 || added.shutdownCalls != 1 { - t.Fatalf("clearing the map shut down kept=%d added=%d, want 1 each", kept.shutdownCalls, added.shutdownCalls) - } -} diff --git a/weed/iam/sts/provider_factory.go b/weed/iam/sts/provider_factory.go index b68894eb58..63e27e8124 100644 --- a/weed/iam/sts/provider_factory.go +++ b/weed/iam/sts/provider_factory.go @@ -1,7 +1,6 @@ package sts import ( - "encoding/json" "fmt" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -155,21 +154,6 @@ func (f *ProviderFactory) convertToOIDCConfig(configMap map[string]interface{}) config.TLSInsecureSkipVerify = tlsInsecureSkipVerify } - if enabled, ok := configMap[ConfigFieldJTIReplayProtectionEnabled].(bool); ok { - config.JTIReplayProtectionEnabled = &enabled - } - - switch maxSize := configMap[ConfigFieldJTICacheMaxSize].(type) { - case float64: - config.JTICacheMaxSize = int(maxSize) - case int: - config.JTICacheMaxSize = maxSize - case json.Number: - if n, err := maxSize.Int64(); err == nil { - config.JTICacheMaxSize = int(n) - } - } - // Convert claims mapping if claimsMapInterface, ok := configMap["claimsMapping"]; ok { claimsMap, err := f.convertToStringMap(claimsMapInterface) diff --git a/weed/iam/sts/provider_factory_jti_test.go b/weed/iam/sts/provider_factory_jti_test.go deleted file mode 100644 index 93800113ff..0000000000 --- a/weed/iam/sts/provider_factory_jti_test.go +++ /dev/null @@ -1,31 +0,0 @@ -package sts - -import ( - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestConvertToOIDCConfigCarriesJTISettings(t *testing.T) { - factory := NewProviderFactory() - - config, err := factory.convertToOIDCConfig(map[string]interface{}{ - "issuer": "https://issuer.example", - "clientId": "client", - "jtiReplayProtectionEnabled": false, - "jtiCacheMaxSize": float64(42), - }) - require.NoError(t, err) - require.NotNil(t, config.JTIReplayProtectionEnabled) - assert.False(t, *config.JTIReplayProtectionEnabled) - assert.Equal(t, 42, config.JTICacheMaxSize) - - config, err = factory.convertToOIDCConfig(map[string]interface{}{ - "issuer": "https://issuer.example", - "clientId": "client", - }) - require.NoError(t, err) - assert.Nil(t, config.JTIReplayProtectionEnabled) - assert.Zero(t, config.JTICacheMaxSize) -} diff --git a/weed/iam/sts/sts_service.go b/weed/iam/sts/sts_service.go index 99f48dc14d..5b4f6ae311 100644 --- a/weed/iam/sts/sts_service.go +++ b/weed/iam/sts/sts_service.go @@ -522,35 +522,8 @@ func (s *STSService) SetIAMManagedOIDCProviders(byIssuer map[string][]ScopedOIDC } } s.iamManagedOIDCMu.Lock() - old := s.iamManagedOIDCByIssuer s.iamManagedOIDCByIssuer = cp s.iamManagedOIDCMu.Unlock() - - // Providers dropped by the swap may own background workers (the JTI - // cleanup goroutine); stop them so a replaced provider does not leak. - kept := make(map[providers.IdentityProvider]struct{}) - for _, scoped := range cp { - for _, sp := range scoped { - kept[sp.Provider] = struct{}{} - } - } - stopped := make(map[providers.IdentityProvider]struct{}) - for _, scoped := range old { - for _, sp := range scoped { - if _, ok := kept[sp.Provider]; ok { - continue - } - if _, ok := stopped[sp.Provider]; ok { - continue - } - stopped[sp.Provider] = struct{}{} - if sd, ok := sp.Provider.(interface{ Shutdown(context.Context) error }); ok { - if err := sd.Shutdown(context.Background()); err != nil { - glog.Warningf("OIDC provider %q shutdown after replacement: %v", sp.Provider.Name(), err) - } - } - } - } } // lookupOIDCProviderForAccount returns the provider that should validate @@ -624,7 +597,7 @@ func (s *STSService) AssumeRoleWithWebIdentity(ctx context.Context, request *Ass // caller passed a legacy or claim-based ARN, in which case lookup falls // back to a global (account-less) record only. roleAccountID := utils.ParseRoleARN(request.RoleArn).AccountID - externalIdentity, provider, err := s.validateWebIdentityToken(ctx, request.WebIdentityToken, roleAccountID, true) + externalIdentity, provider, err := s.validateWebIdentityToken(ctx, request.WebIdentityToken, roleAccountID) if err != nil { return nil, fmt.Errorf("failed to validate web identity token: %w", err) } @@ -1029,7 +1002,7 @@ func (s *STSService) validateAssumeRoleWithWebIdentityRequest(request *AssumeRol // validateWebIdentityToken validates the web identity token with strict issuer-to-provider mapping // SECURITY: JWT tokens with a specific issuer claim MUST only be validated by the provider for that issuer // SECURITY: This method only accepts JWT tokens. Non-JWT authentication must use AssumeRoleWithCredentials with explicit ProviderName. -func (s *STSService) validateWebIdentityToken(ctx context.Context, token, roleAccountID string, singleUse bool) (*providers.ExternalIdentity, providers.IdentityProvider, error) { +func (s *STSService) validateWebIdentityToken(ctx context.Context, token, roleAccountID string) (*providers.ExternalIdentity, providers.IdentityProvider, error) { // Try to extract issuer from JWT token for strict validation issuer, err := s.extractIssuerFromJWT(token) if err != nil { @@ -1052,10 +1025,8 @@ func (s *STSService) validateWebIdentityToken(ctx context.Context, token, roleAc return nil, nil, fmt.Errorf("no identity provider registered for issuer: %s", issuer) } - // Authenticate with the correct provider for this issuer. Only the STS - // exchange authenticates single-use: a bearer token is presented on every - // request, so consuming it here would reject all but the first one. - identity, err := authenticateOIDCProvider(ctx, provider, token, singleUse) + // Authenticate with the correct provider for this issuer + identity, err := provider.Authenticate(ctx, token) if err != nil { // Map provider errors to STS errors using errors.Is() for robust error checking // This eliminates fragile string matching and provides reliable error classification @@ -1087,29 +1058,14 @@ func (s *STSService) validateWebIdentityToken(ctx context.Context, token, roleAc // only, then static-config); call ValidateWebIdentityTokenForAccount when the assumed-role account // is known. func (s *STSService) ValidateWebIdentityToken(ctx context.Context, token string) (*providers.ExternalIdentity, providers.IdentityProvider, error) { - return s.validateWebIdentityToken(ctx, token, "", false) + return s.validateWebIdentityToken(ctx, token, "") } // ValidateWebIdentityTokenForAccount mirrors ValidateWebIdentityToken but // scopes the IAM-managed provider lookup to roleAccountID. Pass "" for // callers that don't yet know the account (e.g. claim-based mode). func (s *STSService) ValidateWebIdentityTokenForAccount(ctx context.Context, token, roleAccountID string) (*providers.ExternalIdentity, providers.IdentityProvider, error) { - return s.validateWebIdentityToken(ctx, token, roleAccountID, false) -} - -// authenticateOIDCProvider authenticates the token, consuming it for replay -// protection when the caller is the single-use STS exchange and the provider -// supports it. Other providers and non-exchange callers authenticate without -// consuming. -func authenticateOIDCProvider(ctx context.Context, provider providers.IdentityProvider, token string, singleUse bool) (*providers.ExternalIdentity, error) { - if singleUse { - if su, ok := provider.(interface { - AuthenticateSingleUse(context.Context, string) (*providers.ExternalIdentity, error) - }); ok { - return su.AuthenticateSingleUse(ctx, token) - } - } - return provider.Authenticate(ctx, token) + return s.validateWebIdentityToken(ctx, token, roleAccountID) } // extractIssuerFromJWT extracts the issuer (iss) claim from a JWT token without verification diff --git a/weed/s3api/s3api_sts_web_identity_http_test.go b/weed/s3api/s3api_sts_web_identity_http_test.go index ff05a11d9c..1aacb89fba 100644 --- a/weed/s3api/s3api_sts_web_identity_http_test.go +++ b/weed/s3api/s3api_sts_web_identity_http_test.go @@ -7,12 +7,10 @@ import ( "encoding/base64" "encoding/json" "encoding/xml" - "fmt" "net/http" "net/http/httptest" "net/url" "strings" - "sync/atomic" "testing" "time" @@ -99,7 +97,6 @@ func TestAssumeRoleWithWebIdentityOverHTTP(t *testing.T) { iam := &IdentityAccessManagement{iamIntegration: NewS3IAMIntegration(manager, "")} handlers := NewSTSHandlers(manager.GetSTSService(), iam) - var jtiCounter atomic.Int64 signOIDCToken := func(t *testing.T, subject string) string { t.Helper() token := jwt.NewWithClaims(jwt.SigningMethodRS256, jwt.MapClaims{ @@ -108,10 +105,6 @@ func TestAssumeRoleWithWebIdentityOverHTTP(t *testing.T) { "aud": "test-client", "exp": time.Now().Add(time.Hour).Unix(), "iat": time.Now().Unix(), - // a unique jti per token: the exchange consumes it, so two - // subtests minting the same claims in one second would otherwise - // collide on the token fingerprint and read as a replay - "jti": fmt.Sprintf("%s-%d", subject, jtiCounter.Add(1)), }) token.Header["kid"] = "web-identity-test-key" signed, err := token.SignedString(key) @@ -181,6 +174,25 @@ func TestAssumeRoleWithWebIdentityOverHTTP(t *testing.T) { }) } + // Kubernetes workload identity re-presents the same projected token until + // ~80% of its lifetime, so one token must exchange more than once. + t.Run("the same token exchanges twice", func(t *testing.T) { + form := url.Values{ + "Action": {"AssumeRoleWithWebIdentity"}, + "Version": {"2011-06-15"}, + "RoleArn": {trustedRoleArn}, + "RoleSessionName": {"web-session-reuse"}, + "WebIdentityToken": {signOIDCToken(t, "oidc-user-1")}, + } + for attempt := 1; attempt <= 2; attempt++ { + req := httptest.NewRequest(http.MethodPost, "http://sts.seaweedfs.test/", strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + rec := httptest.NewRecorder() + handlers.HandleSTSRequest(rec, req) + require.Equal(t, http.StatusOK, rec.Code, "attempt %d: %s", attempt, rec.Body.String()) + } + }) + t.Run("inline session policy is reported in PackedPolicySize", func(t *testing.T) { sessionPolicy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::bucket/*"]}]}` form := url.Values{