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)