Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions framework/configstore/rdb.go
Original file line number Diff line number Diff line change
Expand Up @@ -5852,12 +5852,24 @@ func (s *RDBConfigStore) GetExpiringOauthTokens(ctx context.Context, before time
// worker re-selects a permanently-dead token on every tick (its expires_at
// stays in the past) and logs the same failure indefinitely; a dead grant
// needs re-authorization, not perpetual retries.
//
// Refresh is also limited to tokens whose oauth_config is referenced by
// at least one enabled MCP client: nothing consumes a token while every
// client using it is disabled (or gone), so background refresh would keep
// calling the identity provider forever for an unused connection. When a
// client is re-enabled or attached later, GetAccessToken refreshes inline
// on first use.
result := s.DB().WithContext(ctx).
Where("expires_at IS NOT NULL AND expires_at < ?", before).
Where("NOT EXISTS (?)",
s.DB().Model(&tables.TableOauthConfig{}).
Select("1").
Where("oauth_configs.token_id = oauth_tokens.id AND oauth_configs.status IN ?", []string{"expired", "revoked"})).
Where("EXISTS (?)",
s.DB().Model(&tables.TableMCPClient{}).
Select("1").
Joins("JOIN oauth_configs ON oauth_configs.id = config_mcp_clients.oauth_config_id").
Where("oauth_configs.token_id = oauth_tokens.id AND config_mcp_clients.disabled = ?", false)).
Find(&tokens)
if result.Error != nil {
return nil, fmt.Errorf("failed to get expiring tokens: %w", result.Error)
Expand Down
94 changes: 75 additions & 19 deletions framework/configstore/rdb_oauth2_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -59,50 +59,106 @@ func makeRefreshToken(id, familyID, clientID, hash string) *tables.TableOAuth2Re
}
}

// TestGetExpiringOauthTokens_ExcludesTerminalConfigs verifies the refresh worker
// query skips tokens whose oauth_config is already terminal (expired/revoked), so
// a permanently-dead grant is not retried — and re-logged — on every tick.
func TestGetExpiringOauthTokens_ExcludesTerminalConfigs(t *testing.T) {
s := setupRDBTestStore(t)
require.NoError(t, s.DB().AutoMigrate(&tables.TableOauthConfig{}, &tables.TableOauthToken{}))
ctx := context.Background()
// seedExpiringTokenFixtures installs the token/config/client helpers shared by
// the GetExpiringOauthTokens tests. Every token is created already-expired so
// only the config/client conditions decide whether it is selected.
func seedExpiringTokenFixtures(t *testing.T, s *RDBConfigStore) (mkToken func(id string), mkConfig func(id, tokenID, status, state string), mkClient func(name, oauthConfigID string, disabled bool)) {
t.Helper()
require.NoError(t, s.DB().AutoMigrate(&tables.TableOauthConfig{}, &tables.TableOauthToken{}, &tables.TableMCPClient{}))
past := time.Now().Add(-time.Hour)

mkToken := func(id string) {
mkToken = func(id string) {
require.NoError(t, s.DB().Create(&tables.TableOauthToken{
ID: id, AccessToken: "at-" + id, TokenType: "Bearer",
ExpiresAt: &past, CreatedAt: time.Now(), UpdatedAt: time.Now(),
}).Error)
}
mkConfig := func(id, tokenID, status, state string) {
mkConfig = func(id, tokenID, status, state string) {
require.NoError(t, s.DB().Create(&tables.TableOauthConfig{
ID: id, RedirectURI: "http://127.0.0.1/cb", State: state, Status: status,
TokenID: &tokenID, CreatedAt: time.Now(), UpdatedAt: time.Now(),
ExpiresAt: time.Now().Add(time.Hour),
}).Error)
}
mkClient = func(name, oauthConfigID string, disabled bool) {
require.NoError(t, s.DB().Create(&tables.TableMCPClient{
ClientID: "cid-" + name, Name: name, ConnectionType: "http",
AuthType: "oauth", OauthConfigID: &oauthConfigID, Disabled: disabled,
CreatedAt: time.Now(), UpdatedAt: time.Now(),
}).Error)
}
return mkToken, mkConfig, mkClient
}

func expiringTokenIDs(t *testing.T, s *RDBConfigStore) map[string]bool {
t.Helper()
got, err := s.GetExpiringOauthTokens(context.Background(), time.Now().Add(time.Minute))
require.NoError(t, err)
ids := make(map[string]bool, len(got))
for _, tk := range got {
ids[tk.ID] = true
}
return ids
}

// TestGetExpiringOauthTokens_ExcludesTerminalConfigs verifies the refresh worker
// query skips tokens whose oauth_config is already terminal (expired/revoked), so
// a permanently-dead grant is not retried — and re-logged — on every tick. Each
// config gets an enabled MCP client so status is the only deciding condition.
func TestGetExpiringOauthTokens_ExcludesTerminalConfigs(t *testing.T) {
s := setupRDBTestStore(t)
mkToken, mkConfig, mkClient := seedExpiringTokenFixtures(t, s)

mkToken("tok-live")
mkConfig("cfg-live", "tok-live", "authorized", "state-live")
mkClient("client-live", "cfg-live", false)
mkToken("tok-expired")
mkConfig("cfg-expired", "tok-expired", "expired", "state-expired")
mkClient("client-expired", "cfg-expired", false)
mkToken("tok-revoked")
mkConfig("cfg-revoked", "tok-revoked", "revoked", "state-revoked")
mkToken("tok-orphan") // no owning config at all

got, err := s.GetExpiringOauthTokens(ctx, time.Now().Add(time.Minute))
require.NoError(t, err)
mkClient("client-revoked", "cfg-revoked", false)

ids := make(map[string]bool, len(got))
for _, tk := range got {
ids[tk.ID] = true
}
assert.True(t, ids["tok-live"], "token with an authorized config should be refreshed")
assert.True(t, ids["tok-orphan"], "token with no config should still be returned")
ids := expiringTokenIDs(t, s)
assert.True(t, ids["tok-live"], "token with an authorized config and enabled client should be refreshed")
assert.False(t, ids["tok-expired"], "token with an expired config must be excluded")
assert.False(t, ids["tok-revoked"], "token with a revoked config must be excluded")
}

// TestGetExpiringOauthTokens_RequiresEnabledClient verifies the refresh worker
// only keeps tokens warm while at least one enabled MCP client references the
// owning oauth_config. Disabled-only and unreferenced configs are skipped —
// their tokens catch up via GetAccessToken's inline refresh on next use.
func TestGetExpiringOauthTokens_RequiresEnabledClient(t *testing.T) {
s := setupRDBTestStore(t)
mkToken, mkConfig, mkClient := seedExpiringTokenFixtures(t, s)

mkToken("tok-enabled")
mkConfig("cfg-enabled", "tok-enabled", "authorized", "state-enabled")
mkClient("client-enabled", "cfg-enabled", false)

mkToken("tok-disabled")
mkConfig("cfg-disabled", "tok-disabled", "authorized", "state-disabled")
mkClient("client-disabled", "cfg-disabled", true)

mkToken("tok-shared")
mkConfig("cfg-shared", "tok-shared", "authorized", "state-shared")
mkClient("client-shared-off", "cfg-shared", true)
mkClient("client-shared-on", "cfg-shared", false)

mkToken("tok-no-client")
mkConfig("cfg-no-client", "tok-no-client", "authorized", "state-no-client")

mkToken("tok-orphan") // no owning config at all

ids := expiringTokenIDs(t, s)
assert.True(t, ids["tok-enabled"], "token with an enabled client should be refreshed")
assert.False(t, ids["tok-disabled"], "token referenced only by a disabled client must be excluded")
assert.True(t, ids["tok-shared"], "config shared with at least one enabled client should be refreshed")
assert.False(t, ids["tok-no-client"], "token whose config has no client rows must be excluded")
assert.False(t, ids["tok-orphan"], "token with no owning config must be excluded")
}

func TestGetOAuth2SigningKey_AutoGeneratesAndIsStable(t *testing.T) {
s := setupOAuth2TestStore(t)
ctx := context.Background()
Expand Down
71 changes: 27 additions & 44 deletions framework/oauth2/sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"sync"
"time"

bifrost "github.com/maximhq/bifrost/core"
"github.com/maximhq/bifrost/core/schemas"
)

Expand All @@ -21,6 +22,9 @@ type TokenRefreshWorker struct {

// NewTokenRefreshWorker creates a new token refresh worker
func NewTokenRefreshWorker(provider *OAuth2Provider, logger schemas.Logger) *TokenRefreshWorker {
if logger == nil {
logger = bifrost.NewNoOpLogger()
}
if provider.configStore == nil {
logger.Warn("config store is nil, skipping token refresh worker")
return nil
Expand All @@ -39,9 +43,7 @@ func (w *TokenRefreshWorker) Start(ctx context.Context) {
runCtx, cancel := context.WithCancel(ctx)
w.cancel = cancel
go w.run(runCtx)
if w.logger != nil {
w.logger.Info("Token refresh worker started")
}
w.logger.Info("Token refresh worker started")
}

// Stop gracefully stops the token refresh worker. Safe to call multiple times
Expand All @@ -55,9 +57,7 @@ func (w *TokenRefreshWorker) Stop() {
w.cancel()
}
close(w.stopCh)
if w.logger != nil {
w.logger.Info("Token refresh worker stopped")
}
w.logger.Info("Token refresh worker stopped")
})
}

Expand Down Expand Up @@ -88,52 +88,42 @@ func (w *TokenRefreshWorker) refreshExpiredTokens(ctx context.Context) {
// Get tokens expiring before the threshold
tokens, err := w.provider.configStore.GetExpiringOauthTokens(ctx, expiryThreshold)
if err != nil {
if w.logger != nil {
w.logger.Error("Failed to get expiring tokens", "error", err)
}
w.logger.Error("Failed to get expiring tokens: %v", err)
return
}

if len(tokens) == 0 {
return
}

if w.logger != nil {
w.logger.Debug("Found expiring tokens to refresh: %d", len(tokens))
}
w.logger.Debug("Found expiring tokens to refresh: %d", len(tokens))

// Refresh each expiring token
for _, token := range tokens {
// Find the oauth_config that references this token
oauthConfig, err := w.provider.configStore.GetOauthConfigByTokenID(ctx, token.ID)
if err != nil {
if w.logger != nil {
w.logger.Error("Failed to find oauth config for token: %s, error: %s", token.ID, err.Error())
}
w.logger.Error("Failed to find oauth config for token: %s, error: %s", token.ID, err.Error())
continue
}

if oauthConfig == nil {
if w.logger != nil {
w.logger.Warn("No oauth config found for token: %s", token.ID)
}
w.logger.Warn("No oauth config found for token: %s", token.ID)
continue
}

// Attempt to refresh the token
// Attempt to refresh the token. Logged at Debug: transient failures
// (DNS, timeout, offline) recur on every tick and would spam the
// error log, while permanent rejections are already surfaced by the
// oauth_config status flipping to "expired" below.
if err := w.provider.RefreshAccessToken(ctx, oauthConfig.ID); err != nil {
if w.logger != nil {
w.logger.Error("Failed to refresh token", "oauth_config_id", oauthConfig.ID, "error", err)
}
w.logger.Debug("Failed to refresh token: oauth_config_id: %s, error: %s", oauthConfig.ID, err.Error())

// Only mark as expired for permanent auth rejections (e.g. invalid_grant, 401).
// Transient failures (DNS, timeout, offline) are skipped — the worker will
// retry on the next tick and the connection heals automatically when online.
w.provider.markExpiredIfPermanent(ctx, oauthConfig, err)
} else {
if w.logger != nil {
w.logger.Debug("Successfully refreshed token: %s", oauthConfig.ID)
}
w.logger.Debug("Successfully refreshed token: %s", oauthConfig.ID)
}
}
}
Expand Down Expand Up @@ -169,10 +159,11 @@ type PerUserOAuthSweepWorker struct {
// NewPerUserOAuthSweepWorker creates a sweep worker with sensible defaults.
// orphanRetention <= 0 disables the orphan-token sweep.
func NewPerUserOAuthSweepWorker(provider *OAuth2Provider, orphanRetention time.Duration, logger schemas.Logger) *PerUserOAuthSweepWorker {
if logger == nil {
logger = bifrost.NewNoOpLogger()
}
if provider == nil || provider.configStore == nil {
if logger != nil {
logger.Warn("per-user OAuth sweep worker not started: provider or config store is nil")
}
logger.Warn("per-user OAuth sweep worker not started: provider or config store is nil")
return nil
}
return &PerUserOAuthSweepWorker{
Expand All @@ -190,10 +181,8 @@ func (w *PerUserOAuthSweepWorker) Start(ctx context.Context) {
runCtx, cancel := context.WithCancel(ctx)
w.cancel = cancel
go w.run(runCtx)
if w.logger != nil {
w.logger.Info("Per-user OAuth sweep worker started (flow=%s, orphan=%s, retention=%s)",
w.flowSweepEvery, w.orphanSweepEvery, w.orphanRetention)
}
w.logger.Info("Per-user OAuth sweep worker started (flow=%s, orphan=%s, retention=%s)",
w.flowSweepEvery, w.orphanSweepEvery, w.orphanRetention)
}

// Stop gracefully stops the sweep worker. sync.Once guards against double-close
Expand All @@ -206,9 +195,7 @@ func (w *PerUserOAuthSweepWorker) Stop() {
w.cancel()
}
close(w.stopCh)
if w.logger != nil {
w.logger.Info("Per-user OAuth sweep worker stopped")
}
w.logger.Info("Per-user OAuth sweep worker stopped")
})
}

Expand Down Expand Up @@ -239,12 +226,10 @@ func (w *PerUserOAuthSweepWorker) run(ctx context.Context) {
func (w *PerUserOAuthSweepWorker) sweepExpiredFlows(ctx context.Context) {
n, err := w.provider.configStore.DeleteExpiredOauthUserSessions(ctx)
if err != nil {
if w.logger != nil {
w.logger.Error("per-user OAuth flow sweep failed: %v", err)
}
w.logger.Error("per-user OAuth flow sweep failed: %v", err)
return
}
if n > 0 && w.logger != nil {
if n > 0 {
w.logger.Debug("per-user OAuth flow sweep removed %d expired pending flows", n)
}
}
Expand All @@ -255,12 +240,10 @@ func (w *PerUserOAuthSweepWorker) sweepOrphanedTokens(ctx context.Context) {
}
n, err := w.provider.configStore.DeleteOrphanedOauthUserTokens(ctx, w.orphanRetention)
if err != nil {
if w.logger != nil {
w.logger.Error("per-user OAuth orphan-token sweep failed: %v", err)
}
w.logger.Error("per-user OAuth orphan-token sweep failed: %v", err)
return
}
if n > 0 && w.logger != nil {
if n > 0 {
w.logger.Info("per-user OAuth orphan-token sweep removed %d rows older than %s", n, w.orphanRetention)
}
}
Expand Down
Loading