iam/oidc: reject reused jti claims with a bounded replay cache (#11936)

* iam/oidc: reject reused jti claims with a bounded replay cache

* iam/sts: pass jti replay settings through convertToOIDCConfig

* iam/oidc: keep consumed jtis and stop the cleanup worker across re-init

Re-initializing a provider reset the JTI store, so an already-used token
could be presented again after a config refresh; the store is now
allocated once and survives Initialize. The cleanup worker is stopped on
every re-init (previously only when protection stayed enabled), is
started only after the rest of Initialize has succeeded, and Shutdown no
longer consumes a permanent sync.Once, so a re-initialized worker can be
stopped again.

* iam/oidc: enforce the jti replay cache bound for every token

Tokens without a usable jti claim skipped the cache entirely and stayed
reusable; they are now keyed by the token's own fingerprint so a repeat
presentation is still a replay. Tokens without exp previously failed as
expired on first use; their entry now gets a bounded retention. The
replay test-and-set now runs before any capacity work so replayed tokens
cannot each trigger a full-cache scan, the entry count is incremented as
part of the insert so concurrent validations cannot overshoot the bound,
and a guaranteed serialized sweep runs before a capacity rejection while
opportunistic sweeps are rate limited and confined to one at a time. Raw
token identifiers are no longer written to logs or errors, a stable
fingerprint is logged instead.

* iam/sts: shut down providers dropped by the managed-provider swap

Each managed OIDC provider runs a background JTI cleanup worker, but
SetIAMManagedOIDCProviders replaced the provider map without stopping
the old providers, so every refresh leaked goroutines. Providers absent
from the new map are now shut down; providers that stay keep running.

* iam/integration: reuse providers across refreshes for unchanged records

Every refresh rebuilt providers from their records, so a refreshed
provider forgot the JTIs it had already consumed and an already-used
token could be presented again. Records are now fingerprinted and an
unchanged record reuses its provider, keeping the consumed-token cache;
changed records are rebuilt, removed records are dropped, and the swap
in STS releases the workers of whatever is no longer referenced.

* iam/oidc: consume replay entries only on the single-use exchange path

Bearer authentication validates the same token on every request, so
consuming it in shared ValidateToken/Authenticate made every token
single-use and rejected all but the first request. The replay check now
runs only for the STS exchange via AuthenticateSingleUse; bearer paths
validate without consuming. A token without exp also no longer gets a
bounded replay entry that ages out while the token stays valid: its
entry is kept until the capacity bound evicts it, so it can never become
usable again.

* iam/oidc: keep the mock provider on the mock path for single-use calls

* iam/oidc: require exp on single-use tokens

A cache slot frees only by expiry, so an expiry-less entry would hold it
forever; a burst of distinct tokens could exhaust the bound permanently
and block every later exchange.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: give each web-identity test token a unique jti

Subtests minted tokens with identical claims inside one second, so the
exchange saw the same token fingerprint twice and the second call read
as a replay. Each exchange now carries its own jti.

* iam/oidc: carry the consumed-token replay store across provider rebuilds

A record change rebuilds the managed provider with a fresh JTI store, so a token consumed before the rebuild could be presented again until it expired. The new provider adopts the replaced provider's replay store before it is published, and the replaced provider is shut down so only one cleanup worker sweeps the shared store.

* iam/oidc: stop the cleanup worker of providers dropped from the store

A refresh removed the provider object from the managed cache but left its JTI cleanup goroutine running for the life of the process.

* iam/oidc: merge a provider's consumed replay entries into another's store

MergeReplayStateFrom folds one provider's consumed-token entries into
another's, skipping already-expired ones. Unlike AdoptReplayStateFrom it
swaps no pointers, so it is safe on a provider still serving validators.

* iam/integration: share one provider and replay store across identical issuer records

Records for the same issuer URL with identical validation settings now
reuse a single provider, so a token consumed through one account's
record is rejected through the other's — previously each record owned a
separate jti store and a spent token could be replayed cross-account.

A superseded provider's consumed entries merge into the shared store so
no rejection is lost, and a removed or rebuilt provider is only shut
down once no remaining record references it.

* iam/oidc: hand the superseded provider's live replay store to the shared one

* iam/oidc: carry merged replay entries without charging the capacity bound

* iam/oidc: key the jti-less replay fingerprint on the signed input

Hashing the whole token lets ECDSA signature malleability (s -> N-s)
mint a second cache key for identical claims, so a consumed token could
be replayed for another exchange. Key the fallback on header.payload
instead.

* iam/oidc: publish the jti cache bound before the enabled flag

Initialize stored jtiEnabled ahead of jtiMaxSize, so a provider flipped
from disabled to enabled exposed enabled=true with a zero bound for a
window; a single-use validation landing there would compute
newCount > maxSize and reject with "JTI cache capacity exceeded". Store
the bound first so the flag is the last thing published.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/iam/integration/iam_manager.go

@@ -53,7 +53,13 @@ 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
+	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 map[string]*oidcManagedProvider
 	oidcAuditSink        OIDCProviderAuditSink
 	revocationStore      SessionRevocationStore
 	filerAddressProvider func() string // Function to get current filer address
@@ -943,14 +949,65 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro
 		return fmt.Errorf("list OIDC providers: the store was replaced during the refresh")
 	}
 	byIssuer := make(map[string][]sts.ScopedOIDCProvider, len(records))
+	if m.oidcManagedProviders == nil {
+		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
 		}
-		provider, err := buildOIDCProviderFromRecord(rec)
-		if err != nil {
-			glog.Warningf("skip refreshing OIDC provider %s: %v", rec.ARN, err)
-			continue
+		key := rec.ARN
+		if key == "" {
+			key = rec.URL + "\x00" + rec.AccountID
+		}
+		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 {
+			provider = cached.provider
+			sharedProviders[configFingerprint] = provider
+		} else {
+			var err error
+			provider, err = buildOIDCProviderFromRecord(rec)
+			if err != nil {
+				delete(m.oidcManagedProviders, key)
+				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
@@ -960,10 +1017,71 @@ func (m *IAMManager) refreshOIDCProvidersFrom(ctx context.Context, store OIDCPro
 			Provider:  provider,
 		})
 	}
+	for key, managed := 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)
 	return nil
 }
 
+// oidcManagedProvider is a runtime provider together with the fingerprint of
+// the record it was built from.
+type oidcManagedProvider struct {
+	fingerprint string
+	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 {
+	key, _ := json.Marshal(struct {
+		ARN                     string
+		AccountID               string
+		URL                     string
+		ClientIDs               []string
+		Thumbprints             []string
+		AllowedPrincipalTagKeys []string
+		PolicyClaim             string
+	}{rec.ARN, rec.AccountID, rec.URL, rec.ClientIDs, rec.Thumbprints, rec.AllowedPrincipalTagKeys, rec.PolicyClaim})
+	sum := sha256.Sum256(key)
+	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.

weed/iam/integration/oidc_provider_persist_test.go

@@ -559,3 +559,93 @@ 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()
+	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)
+
+	require.NoError(t, mgr.refreshOIDCProvidersFrom(ctx, store))
+	assert.Same(t, first, mgr.oidcManagedProviders[arn].provider,
+		"an unchanged record was rebuilt, dropping its replay cache")
+
+	// 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")
+
+	// 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))
+}
+
+// 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()
+	mgr := startServer(t, store)
+
+	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))
+
+	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")
+
+	// 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"))
+}

weed/iam/oidc/mock_provider.go

@@ -75,6 +75,17 @@ 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 {

weed/iam/oidc/oidc_provider.go

@@ -6,6 +6,7 @@ import (
 	"crypto/elliptic"
 	"crypto/rsa"
 	"crypto/sha1"
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
@@ -19,6 +20,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/golang-jwt/jwt/v5"
@@ -27,7 +29,44 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/security"
 )
 
-// OIDCProvider implements OpenID Connect authentication
+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 jtiCount, so merged excess can
+	// neither reject new exchanges nor be double-counted when swept.
+	uncounted bool
+}
+
+// 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.
 type OIDCProvider struct {
 	name        string
 	config      *OIDCConfig
@@ -44,6 +83,32 @@ type OIDCProvider struct {
 	jwksFetchedAt   time.Time
 	resolvedJWKSUri string
 	discoveryFailed bool
+
+	// JTI replay protection fields. jtiStore and jtiCount are allocated once
+	// in NewOIDCProvider and survive re-initialization, so a config refresh
+	// cannot reopen already-consumed tokens for replay.
+	jtiEnabled   atomic.Bool
+	jtiStore     *sync.Map     // map[string]*jtiEntry - atomic storage for consumed token keys
+	jtiCount     *atomic.Int64 // entries in jtiStore; shared with the store
+	jtiMaxSize   atomic.Int64
+	jtiLastSweep atomic.Int64 // unix nano of the last sweep; rate limits eager cleanups
+
+	// jtiStoreMu guards the jtiStore/jtiCount pair: replay-state handoffs
+	// (AdoptReplayStateFrom, MergeReplayStateFrom) 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 jtiSweepMu, never while holding it.
+	jtiStoreMu sync.RWMutex
+
+	// jtiMu guards the lifecycle of the cleanup worker: jtiCleanupCancel and
+	// jtiCleanupDone. The worker itself only touches jtiStore/jtiCount, so
+	// holding jtiMu while waiting for it is safe.
+	jtiMu            sync.Mutex
+	jtiCleanupCancel context.CancelFunc
+	jtiCleanupDone   chan struct{} // Closed when cleanup goroutine exits
+
+	// jtiSweepMu serializes cache sweeps so at most one scan runs at a time.
+	jtiSweepMu sync.Mutex
 }
 
 // OIDCConfig holds OIDC provider configuration
@@ -107,6 +172,13 @@ 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
@@ -319,6 +391,8 @@ func NewOIDCProvider(name string) *OIDCProvider {
 	return &OIDCProvider{
 		name:       name,
 		httpClient: &http.Client{Timeout: 30 * time.Second, CheckRedirect: refuseDowngradeRedirect},
+		jtiStore:   &sync.Map{},
+		jtiCount:   &atomic.Int64{},
 	}
 }
 
@@ -429,10 +503,62 @@ 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.jtiMu.Lock()
+	p.stopJTICleanupWorkerLocked()
+	if jtiEnabled {
+		maxSize := int64(oidcConfig.JTICacheMaxSize)
+		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
+		p.jtiMaxSize.Store(maxSize)
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		p.jtiCleanupCancel = cancel
+		p.jtiCleanupDone = done
+		p.jtiEnabled.Store(true)
+		p.jtiMu.Unlock()
+
+		go p.cleanupExpiredJTIs(ctx, done, defaultCleanupInterval)
+		glog.V(2).Infof("OIDC provider %q: JTI replay protection enabled (max size: %d)", p.name, maxSize)
+	} else {
+		p.jtiEnabled.Store(false)
+		p.jtiMu.Unlock()
+		glog.Warningf("OIDC provider %q: JTI replay protection is DISABLED", p.name)
+	}
+
 	// For testing, we'll skip the actual OIDC client initialization
 	return nil
 }
 
+// stopJTICleanupWorkerLocked cancels the current cleanup goroutine and waits
+// briefly for it to exit. Caller must hold jtiMu.
+func (p *OIDCProvider) stopJTICleanupWorkerLocked() {
+	if p.jtiCleanupCancel == nil {
+		return
+	}
+	p.jtiCleanupCancel()
+	select {
+	case <-p.jtiCleanupDone:
+		glog.V(3).Infof("OIDC provider %q: JTI cleanup goroutine exited", p.name)
+	case <-time.After(2 * time.Second):
+		glog.Warningf("OIDC provider %q: Timeout waiting for JTI cleanup goroutine", p.name)
+	}
+	p.jtiCleanupCancel = nil
+	p.jtiCleanupDone = nil
+}
+
 // validateConfig validates the OIDC configuration
 func (p *OIDCProvider) validateConfig(config *OIDCConfig) error {
 	if config.Issuer == "" {
@@ -479,6 +605,18 @@ 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")
 	}
@@ -488,7 +626,7 @@ func (p *OIDCProvider) Authenticate(ctx context.Context, token string) (*provide
 	}
 
 	// Validate token and get claims
-	claims, err := p.ValidateToken(ctx, token)
+	claims, err := p.validateToken(ctx, token, singleUse)
 	if err != nil {
 		return nil, err
 	}
@@ -655,6 +793,16 @@ 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")
 	}
@@ -787,6 +935,87 @@ func (p *OIDCProvider) ValidateToken(ctx context.Context, token string) (*provid
 		tokenClaims.Claims[key] = value
 	}
 
+	// Extract and validate JTI claim 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 && p.jtiEnabled.Load() {
+		jti, _ := claims["jti"].(string)
+		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[:])
+		}
+
+		expiresAt := tokenClaims.ExpiresAt
+		if expiresAt.IsZero() {
+			// a cache slot is freed only by expiry, so a token without exp
+			// would hold it forever and eventually exhaust the bound
+			return nil, fmt.Errorf("%w: missing exp claim", providers.ErrProviderMissingClaims)
+		}
+		expiresAt = expiresAt.Add(clockSkewTolerance)
+		if !time.Now().Before(expiresAt) {
+			return nil, 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.
+		p.jtiStoreMu.RLock()
+		store, count := p.jtiStore, p.jtiCount
+
+		// 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 {
+			p.jtiStoreMu.RUnlock()
+			glog.Warningf("OIDC provider %q: Token replay detected (jti: %s)",
+				p.name, jtiLogKey(key))
+			return nil, 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 := p.jtiMaxSize.Load()
+		if newCount > maxSize {
+			// Over the bound: expired entries may still occupy slots, so
+			// reclaim them with a synchronized sweep before rejecting.
+			p.sweepJTIStore(store, count)
+			if count.Load() > maxSize {
+				if _, loaded := store.LoadAndDelete(key); loaded {
+					count.Add(-1)
+				}
+				p.jtiStoreMu.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.",
+					p.name, maxSize)
+				return nil, 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",
+				p.name, (newCount*100)/maxSize, newCount, maxSize)
+			go p.performEagerCleanup()
+		}
+		p.jtiStoreMu.RUnlock()
+
+		glog.V(4).Infof("OIDC provider %q: Registered JTI %s (expires at %v, cache size: %d)",
+			p.name, jtiLogKey(key), expiresAt, newCount)
+	}
+
 	return tokenClaims, nil
 }
 
@@ -1215,3 +1444,195 @@ func (p *OIDCProvider) mapUserInfoToIdentity(userInfo map[string]interface{}) *p
 
 	return identity
 }
+
+// evictExpiredJTIs removes entries whose expiry has passed and returns how
+// many it deleted. Caller must hold jtiSweepMu.
+func (p *OIDCProvider) evictExpiredJTIs(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
+}
+
+// replayState returns the provider's current store/count pair.
+func (p *OIDCProvider) replayState() (*sync.Map, *atomic.Int64) {
+	p.jtiStoreMu.RLock()
+	defer p.jtiStoreMu.RUnlock()
+	return p.jtiStore, p.jtiCount
+}
+
+// sweepJTIStore runs one serialized expiry scan over the given store. It only
+// takes jtiSweepMu, so callers already holding jtiStoreMu may use it.
+func (p *OIDCProvider) sweepJTIStore(store *sync.Map, count *atomic.Int64) {
+	p.jtiSweepMu.Lock()
+	defer p.jtiSweepMu.Unlock()
+	p.jtiLastSweep.Store(time.Now().UnixNano())
+	if removed := p.evictExpiredJTIs(time.Now(), store, count); removed > 0 {
+		glog.V(4).Infof("OIDC provider %q: Cleaned up %d expired JTI entries (remaining: %d)",
+			p.name, removed, count.Load())
+	}
+}
+
+// sweepExpiredJTIs runs a scan that is guaranteed to complete before the
+// caller proceeds; the capacity-rejection path uses it so a fresh token is
+// not refused while expired entries still hold slots.
+func (p *OIDCProvider) sweepExpiredJTIs() {
+	store, count := p.replayState()
+	p.sweepJTIStore(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 (p *OIDCProvider) performEagerCleanup() {
+	if !p.jtiEnabled.Load() {
+		return
+	}
+	if time.Since(time.Unix(0, p.jtiLastSweep.Load())) < eagerCleanupMinInterval {
+		return
+	}
+	store, count := p.replayState()
+	if !p.jtiSweepMu.TryLock() {
+		return
+	}
+	defer p.jtiSweepMu.Unlock()
+	p.jtiLastSweep.Store(time.Now().UnixNano())
+
+	beforeCount := count.Load()
+	if removed := p.evictExpiredJTIs(time.Now(), store, count); removed > 0 {
+		glog.Infof("OIDC provider %q: Eager cleanup removed %d expired JTI entries (%d -> %d)",
+			p.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.",
+			p.name)
+	}
+}
+
+// cleanupExpiredJTIs periodically removes expired JTI entries from the cache.
+// This prevents memory growth by removing JTIs after their tokens have expired.
+// Runs in a background goroutine until ctx is cancelled; done is closed on exit.
+func (p *OIDCProvider) cleanupExpiredJTIs(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)", p.name, interval)
+
+	for {
+		select {
+		case <-ctx.Done():
+			glog.V(3).Infof("OIDC provider %q: JTI cleanup goroutine stopping", p.name)
+			return
+		case <-ticker.C:
+			p.sweepExpiredJTIs()
+		}
+	}
+}
+
+// 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 {
+	p.jtiMu.Lock()
+	cancel, done := p.jtiCleanupCancel, p.jtiCleanupDone
+	p.jtiMu.Unlock()
+
+	if cancel == nil {
+		return nil
+	}
+	cancel()
+
+	select {
+	case <-done:
+		glog.V(2).Infof("OIDC provider %q: Cleanup goroutine shutdown complete", p.name)
+	case <-ctx.Done():
+		glog.Warningf("OIDC provider %q: Shutdown timeout waiting for cleanup goroutine: %v", p.name, ctx.Err())
+		return ctx.Err()
+	}
+
+	p.jtiMu.Lock()
+	if p.jtiCleanupDone == done {
+		p.jtiCleanupCancel = nil
+		p.jtiCleanupDone = nil
+	}
+	p.jtiMu.Unlock()
+	return nil
+}
+
+// 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; jtiSweepMu excludes its own cleanup worker during the swap.
+func (p *OIDCProvider) AdoptReplayStateFrom(old *OIDCProvider) {
+	if old == nil || old == p {
+		return
+	}
+	p.jtiStoreMu.Lock()
+	p.jtiSweepMu.Lock()
+	p.jtiStore, p.jtiCount = old.jtiStore, old.jtiCount
+	p.jtiSweepMu.Unlock()
+	p.jtiStoreMu.Unlock()
+}
+
+// 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.jtiStoreMu.RLock()
+	defer p.jtiStoreMu.RUnlock()
+	store, count := p.jtiStore, p.jtiCount
+
+	other.jtiStoreMu.Lock()
+	oldStore := other.jtiStore
+	if oldStore == store {
+		other.jtiStoreMu.Unlock()
+		return
+	}
+	other.jtiStore, other.jtiCount = store, count
+	other.jtiStoreMu.Unlock()
+
+	p.jtiSweepMu.Lock()
+	defer p.jtiSweepMu.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])
+}

weed/iam/providers/errors.go

@@ -19,4 +19,7 @@ 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")
 )

weed/iam/sts/constants.go

@@ -42,21 +42,23 @@ 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"
+	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"
 )
 
 // Error Messages

weed/iam/sts/oidc_provider_lookup_test.go

@@ -11,8 +11,8 @@ import (
 // lookupOIDCProviderForAccount; the lookup never calls Authenticate.
 type stubIdentityProvider struct{ name string }
 
-func (s *stubIdentityProvider) Name() string                       { return s.name }
-func (s *stubIdentityProvider) Initialize(interface{}) error       { return nil }
+func (s *stubIdentityProvider) Name() string                 { return s.name }
+func (s *stubIdentityProvider) Initialize(interface{}) error { return nil }
 func (s *stubIdentityProvider) Authenticate(context.Context, string) (*providers.ExternalIdentity, error) {
 	return nil, nil
 }
@@ -103,3 +103,51 @@ 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)
+	}
+}

weed/iam/sts/provider_factory.go

@@ -1,6 +1,7 @@
 package sts
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -154,6 +155,21 @@ 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)

weed/iam/sts/provider_factory_jti_test.go

@@ -0,0 +1,31 @@
+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)
+}

weed/iam/sts/sts_service.go

@@ -512,8 +512,35 @@ 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
@@ -587,7 +614,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)
+	externalIdentity, provider, err := s.validateWebIdentityToken(ctx, request.WebIdentityToken, roleAccountID, true)
 	if err != nil {
 		return nil, fmt.Errorf("failed to validate web identity token: %w", err)
 	}
@@ -981,7 +1008,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) (*providers.ExternalIdentity, providers.IdentityProvider, error) {
+func (s *STSService) validateWebIdentityToken(ctx context.Context, token, roleAccountID string, singleUse bool) (*providers.ExternalIdentity, providers.IdentityProvider, error) {
 	// Try to extract issuer from JWT token for strict validation
 	issuer, err := s.extractIssuerFromJWT(token)
 	if err != nil {
@@ -1004,8 +1031,10 @@ 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
-	identity, err := provider.Authenticate(ctx, token)
+	// 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)
 	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
@@ -1037,14 +1066,29 @@ 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, "")
+	return s.validateWebIdentityToken(ctx, token, "", false)
 }
 
 // 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)
+	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)
 }
 
 // extractIssuerFromJWT extracts the issuer (iss) claim from a JWT token without verification

weed/s3api/s3api_sts_web_identity_http_test.go

@@ -7,10 +7,12 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/xml"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -97,6 +99,7 @@ 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{
@@ -105,6 +108,10 @@ 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)