From b65832bdc9bbbcf3682465b1be598627ac386206 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 00:46:14 -0700 Subject: [PATCH] iam/oidc: reject reused jti claims with a bounded replay cache (#11936) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 128 +- .../integration/oidc_provider_persist_test.go | 90 ++ weed/iam/oidc/mock_provider.go | 11 + weed/iam/oidc/oidc_provider.go | 425 ++++++- weed/iam/oidc/oidc_provider_test.go | 1092 ++++++++++++++++- weed/iam/providers/errors.go | 3 + weed/iam/sts/constants.go | 32 +- weed/iam/sts/oidc_provider_lookup_test.go | 52 +- weed/iam/sts/provider_factory.go | 16 + weed/iam/sts/provider_factory_jti_test.go | 31 + weed/iam/sts/sts_service.go | 56 +- .../s3api/s3api_sts_web_identity_http_test.go | 7 + 12 files changed, 1901 insertions(+), 42 deletions(-) create mode 100644 weed/iam/sts/provider_factory_jti_test.go diff --git a/weed/iam/integration/iam_manager.go b/weed/iam/integration/iam_manager.go index b3af9c7b36..ece1a54927 100644 --- a/weed/iam/integration/iam_manager.go +++ b/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. diff --git a/weed/iam/integration/oidc_provider_persist_test.go b/weed/iam/integration/oidc_provider_persist_test.go index 7190e2aa74..26a6f81250 100644 --- a/weed/iam/integration/oidc_provider_persist_test.go +++ b/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")) +} diff --git a/weed/iam/oidc/mock_provider.go b/weed/iam/oidc/mock_provider.go index bc20a5f14f..5b840eafa7 100644 --- a/weed/iam/oidc/mock_provider.go +++ b/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 { diff --git a/weed/iam/oidc/oidc_provider.go b/weed/iam/oidc/oidc_provider.go index 4f947cf2ba..c2ccf4ab8f 100644 --- a/weed/iam/oidc/oidc_provider.go +++ b/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]) +} diff --git a/weed/iam/oidc/oidc_provider_test.go b/weed/iam/oidc/oidc_provider_test.go index b226929d2f..3744ec3138 100644 --- a/weed/iam/oidc/oidc_provider_test.go +++ b/weed/iam/oidc/oidc_provider_test.go @@ -8,9 +8,14 @@ import ( "crypto/rsa" "encoding/base64" "encoding/json" + "errors" + "fmt" + "math/big" "net/http" "net/http/httptest" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -547,42 +552,42 @@ func TestOIDCProviderUserInfo(t *testing.T) { // Helper functions for testing -func generateTestKeys(t *testing.T) (*rsa.PrivateKey, *rsa.PublicKey) { +func generateTestKeys(tb testing.TB) (*rsa.PrivateKey, *rsa.PublicKey) { privateKey, err := rsa.GenerateKey(rand.Reader, 2048) - require.NoError(t, err) + require.NoError(tb, err) return privateKey, &privateKey.PublicKey } -func generateTestECKeys(t *testing.T) (*ecdsa.PrivateKey, *ecdsa.PublicKey) { +func generateTestECKeys(tb testing.TB) (*ecdsa.PrivateKey, *ecdsa.PublicKey) { privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) - require.NoError(t, err) + require.NoError(tb, err) return privateKey, &privateKey.PublicKey } -func createTestJWT(t *testing.T, privateKey *rsa.PrivateKey, claims jwt.MapClaims) string { +func createTestJWT(tb testing.TB, 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(t, err) + require.NoError(tb, err) return tokenString } -func createTestECDSAJWT(t *testing.T, privateKey *ecdsa.PrivateKey, claims jwt.MapClaims) string { +func createTestECDSAJWT(tb testing.TB, 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(t, err) + require.NoError(tb, err) return tokenString } -func encodePublicKey(t *testing.T, publicKey *rsa.PublicKey) string { +func encodePublicKey(tb testing.TB, publicKey *rsa.PublicKey) string { // Properly encode the RSA modulus (N) as base64url return base64.RawURLEncoding.EncodeToString(publicKey.N.Bytes()) } -func encodeECPublicKey(t *testing.T, publicKey *ecdsa.PublicKey) (string, string) { +func encodeECPublicKey(tb testing.TB, 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 @@ -597,7 +602,7 @@ func encodeECPublicKey(t *testing.T, publicKey *ecdsa.PublicKey) (string, string base64.RawURLEncoding.EncodeToString(yPadded) } -func setupOIDCTestServer(t *testing.T, publicKey *rsa.PublicKey) *httptest.Server { +func setupOIDCTestServer(tb testing.TB, publicKey *rsa.PublicKey) *httptest.Server { jwks := map[string]interface{}{ "keys": []map[string]interface{}{ { @@ -605,7 +610,7 @@ func setupOIDCTestServer(t *testing.T, publicKey *rsa.PublicKey) *httptest.Serve "kid": "test-key-id", "use": "sig", "alg": "RS256", - "n": encodePublicKey(t, publicKey), + "n": encodePublicKey(tb, publicKey), "e": "AQAB", }, }, @@ -660,3 +665,1066 @@ func setupOIDCTestServer(t *testing.T, publicKey *rsa.PublicKey) *httptest.Serve } })) } + +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.jtiCount.Load()) + + // Verify entry exists and check its expiration includes clock skew + val, ok := provider.jtiStore.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.cleanupExpiredJTIs(cleanupCtx, cleanupDone, 20*time.Millisecond) + defer func() { + cleanupCancel() + <-cleanupDone + }() + + provider.jtiStore.Store("expired-cleanup-jti", &jtiEntry{ + subject: "expired-user", + expiresAt: time.Now().Add(-time.Minute), + }) + provider.jtiCount.Add(1) + + require.Eventually(t, func() bool { + _, stillThere := provider.jtiStore.Load("expired-cleanup-jti") + return !stillThere && provider.jtiCount.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.jtiCleanupCancel, "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.jtiCleanupCancel, "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.jtiCleanupCancel, "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.jtiStore.Store(jti, entry) + provider.jtiCount.Add(1) + } + + assert.Equal(t, int64(5), provider.jtiCount.Load(), "Should have 5 expired entries") + + // Perform eager cleanup - should remove all expired entries + provider.performEagerCleanup() + + assert.Equal(t, int64(0), provider.jtiCount.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.jtiCount.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.jtiStore.Store(jti, entry) + provider.jtiCount.Add(1) + } + + initialCount := provider.jtiCount.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.performEagerCleanup() + }() + } + + wg.Wait() + + // Counter should be 0 (all entries removed), never negative + finalCount := provider.jtiCount.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.jtiStore.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 eeac47c521..02ae7df532 100644 --- a/weed/iam/providers/errors.go +++ b/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") ) diff --git a/weed/iam/sts/constants.go b/weed/iam/sts/constants.go index 9a4876069e..3e6e24df39 100644 --- a/weed/iam/sts/constants.go +++ b/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 diff --git a/weed/iam/sts/oidc_provider_lookup_test.go b/weed/iam/sts/oidc_provider_lookup_test.go index 0281e65b33..aba144be2b 100644 --- a/weed/iam/sts/oidc_provider_lookup_test.go +++ b/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) + } +} diff --git a/weed/iam/sts/provider_factory.go b/weed/iam/sts/provider_factory.go index 63e27e8124..b68894eb58 100644 --- a/weed/iam/sts/provider_factory.go +++ b/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) diff --git a/weed/iam/sts/provider_factory_jti_test.go b/weed/iam/sts/provider_factory_jti_test.go new file mode 100644 index 0000000000..93800113ff --- /dev/null +++ b/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) +} diff --git a/weed/iam/sts/sts_service.go b/weed/iam/sts/sts_service.go index e2fdcd6f17..9083b221e5 100644 --- a/weed/iam/sts/sts_service.go +++ b/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 diff --git a/weed/s3api/s3api_sts_web_identity_http_test.go b/weed/s3api/s3api_sts_web_identity_http_test.go index 1c045261c3..ff05a11d9c 100644 --- a/weed/s3api/s3api_sts_web_identity_http_test.go +++ b/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)