git.iko.soy/crush
on main help sign in
fix(auth): stop two Crush instances from invalidating each other's login
de67920Kieran Klukas
internal/config/refresh_singleflight_test.gonew file+175
compare with main
unifiedsplit
@@ -0,0 +1,175 @@
1+package config
2+
3+import (
4+ "context"
5+ "fmt"
6+ "os"
7+ "path/filepath"
8+ "sync"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+
13+ "github.com/charmbracelet/crush/internal/csync"
14+ "github.com/charmbracelet/crush/internal/oauth"
15+ "github.com/stretchr/testify/require"
16+)
17+
18+// newRefreshTestStore builds a ConfigStore whose hyper provider holds an
19+// expired OAuth token, persisted both in memory and on disk at configPath.
20+// Stores that share a configPath also share the per-provider refresh lock,
21+// which lets a single test process faithfully simulate two crush instances:
22+// lock.File opens a fresh descriptor per call, so two stores block each
23+// other on the same lock file exactly as two processes would.
24+func newRefreshTestStore(t *testing.T, configPath string, exchange func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error)) *ConfigStore {
25+ t.Helper()
26+
27+ expired := &oauth.Token{
28+ AccessToken: "at0",
29+ RefreshToken: "rt0",
30+ ExpiresIn: 3600,
31+ ExpiresAt: time.Now().Add(-time.Hour).Unix(),
32+ }
33+ configContent := `{
34+ "providers": {
35+ "hyper": {
36+ "oauth": {
37+ "access_token": "at0",
38+ "refresh_token": "rt0",
39+ "expires_in": 3600,
40+ "expires_at": ` + fmt.Sprintf("%d", expired.ExpiresAt) + `
41+ }
42+ }
43+ }
44+ }`
45+ require.NoError(t, os.WriteFile(configPath, []byte(configContent), 0o600))
46+
47+ providers := csync.NewMap[string, ProviderConfig]()
48+ providers.Set("hyper", ProviderConfig{
49+ ID: "hyper",
50+ Name: "Hyper",
51+ APIKey: expired.AccessToken,
52+ OAuthToken: expired,
53+ })
54+
55+ return &ConfigStore{
56+ config: &Config{Providers: providers},
57+ globalDataPath: configPath,
58+ workingDir: filepath.Dir(configPath),
59+ exchangeToken: exchange,
60+ }
61+}
62+
63+// TestRefreshOAuthToken_InProcessSingleFlight verifies that a storm of
64+// concurrent refresh calls for the same provider collapses into a single
65+// token exchange.
66+func TestRefreshOAuthToken_InProcessSingleFlight(t *testing.T) {
67+ t.Parallel()
68+
69+ configPath := filepath.Join(t.TempDir(), "crush.json")
70+
71+ var exchanges atomic.Int64
72+ store := newRefreshTestStore(t, configPath, func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
73+ exchanges.Add(1)
74+ time.Sleep(50 * time.Millisecond) // hold the flight open so peers join
75+ return &oauth.Token{
76+ AccessToken: "at1",
77+ RefreshToken: "rt1",
78+ ExpiresIn: 3600,
79+ ExpiresAt: time.Now().Add(time.Hour).Unix(),
80+ }, nil
81+ })
82+
83+ const goroutines = 20
84+ var wg sync.WaitGroup
85+ start := make(chan struct{})
86+ errs := make(chan error, goroutines)
87+ for range goroutines {
88+ wg.Go(func() {
89+ <-start
90+ errs <- store.RefreshOAuthToken(context.Background(), ScopeGlobal, "hyper")
91+ })
92+ }
93+ close(start)
94+ wg.Wait()
95+ close(errs)
96+
97+ for err := range errs {
98+ require.NoError(t, err)
99+ }
100+ require.Equal(t, int64(1), exchanges.Load(), "concurrent refreshes should collapse into one exchange")
101+
102+ pc, ok := store.config.Providers.Get("hyper")
103+ require.True(t, ok)
104+ require.Equal(t, "at1", pc.OAuthToken.AccessToken)
105+ require.Equal(t, "rt1", pc.OAuthToken.RefreshToken)
106+}
107+
108+// TestRefreshOAuthToken_CrossProcessAdopt verifies that when two instances
109+// share a credential, only one performs the token exchange and the other
110+// adopts the rotated token from disk rather than reusing the consumed
111+// refresh token. The fake exchange models a rotating provider: reusing a
112+// refresh token it has already rotated returns an error, so a second
113+// exchange would be observable as a failure.
114+func TestRefreshOAuthToken_CrossProcessAdopt(t *testing.T) {
115+ t.Parallel()
116+
117+ configPath := filepath.Join(t.TempDir(), "crush.json")
118+
119+ var (
120+ mu sync.Mutex
121+ current = "rt0" // the only refresh token the server will accept
122+ exchanges atomic.Int64
123+ reuseErrors atomic.Int64
124+ )
125+ exchange := func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
126+ mu.Lock()
127+ defer mu.Unlock()
128+ if refreshToken != current {
129+ reuseErrors.Add(1)
130+ return nil, fmt.Errorf("refresh token revoked")
131+ }
132+ exchanges.Add(1)
133+ time.Sleep(50 * time.Millisecond) // hold the lock so the peer must wait
134+ current = "rt1"
135+ return &oauth.Token{
136+ AccessToken: "at1",
137+ RefreshToken: "rt1",
138+ ExpiresIn: 3600,
139+ ExpiresAt: time.Now().Add(time.Hour).Unix(),
140+ }, nil
141+ }
142+
143+ // Two stores sharing the same config file and refresh lock = two
144+ // "processes".
145+ a := newRefreshTestStore(t, configPath, exchange)
146+ b := newRefreshTestStore(t, configPath, exchange)
147+
148+ var wg sync.WaitGroup
149+ start := make(chan struct{})
150+ errs := make(chan error, 2)
151+ for _, s := range []*ConfigStore{a, b} {
152+ wg.Go(func() {
153+ <-start
154+ errs <- s.RefreshOAuthToken(context.Background(), ScopeGlobal, "hyper")
155+ })
156+ }
157+ close(start)
158+ wg.Wait()
159+ close(errs)
160+
161+ for err := range errs {
162+ require.NoError(t, err)
163+ }
164+
165+ require.Equal(t, int64(1), exchanges.Load(), "only one instance should exchange")
166+ require.Equal(t, int64(0), reuseErrors.Load(), "no instance should reuse a rotated refresh token")
167+
168+ // Both instances converge on the rotated token.
169+ for name, s := range map[string]*ConfigStore{"a": a, "b": b} {
170+ pc, ok := s.config.Providers.Get("hyper")
171+ require.True(t, ok, name)
172+ require.Equal(t, "at1", pc.OAuthToken.AccessToken, name)
173+ require.Equal(t, "rt1", pc.OAuthToken.RefreshToken, name)
174+ }
175+}
internal/config/store.go+130−46
@@ -20,6 +20,7 @@ import (
20 "github.com/charmbracelet/crush/internal/oauth/hyper"
20 "github.com/charmbracelet/crush/internal/oauth/hyper"
21 "github.com/tidwall/gjson"
21 "github.com/tidwall/gjson"
22 "github.com/tidwall/sjson"
22 "github.com/tidwall/sjson"
23+ "golang.org/x/sync/singleflight"
23)
24)
24
25
25// configLockDeadline bounds how long lockConfig waits for the
26// configLockDeadline bounds how long lockConfig waits for the
@@ -27,6 +28,15 @@ import (
27// honest contention; longer suggests something is wedged.
28// honest contention; longer suggests something is wedged.
28const configLockDeadline = 5 * time.Second
29const configLockDeadline = 5 * time.Second
29
30
31+// refreshLockDeadline bounds how long RefreshOAuthToken waits for the
32+// per-provider cross-process refresh lock. It must exceed the token
33+// exchange HTTP timeout (30s) so that a peer mid-exchange is given time
34+// to finish and publish its result, which we then adopt instead of
35+// running our own exchange. Running our own would reuse an
36+// already-rotated refresh token and trip the provider's reuse detection,
37+// revoking the whole token family.
38+const refreshLockDeadline = 45 * time.Second
39+
30// fileSnapshot captures metadata about a config file at a point in time.
40// fileSnapshot captures metadata about a config file at a point in time.
31type fileSnapshot struct {
41type fileSnapshot struct {
32 Path string
42 Path string
@@ -80,6 +90,18 @@ type ConfigStore struct {
80
90
81 mu sync.Mutex // serialises config file writes
91 mu sync.Mutex // serialises config file writes
82 writeMu sync.Mutex // serialises in-memory config production (mutators + reload)
92 writeMu sync.Mutex // serialises in-memory config production (mutators + reload)
93+
94+ // refreshSF collapses concurrent in-process OAuth refreshes for the
95+ // same provider into a single attempt. Combined with the per-provider
96+ // cross-process refresh lock, it ensures only one token exchange runs
97+ // at a time. See RefreshOAuthToken.
98+ refreshSF singleflight.Group
99+
100+ // exchangeToken performs the provider-specific OAuth token exchange.
101+ // It is a field so tests can substitute a fake exchange without making
102+ // real network calls. Production code leaves it nil, and exchange falls
103+ // back to the real provider clients.
104+ exchangeToken func(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error)
83}
105}
84
106
85// Config returns the pure-data config struct (read-only after load).
107// Config returns the pure-data config struct (read-only after load).
@@ -480,64 +502,82 @@ func (s *ConfigStore) SetProviderAPIKey(scope Scope, providerID string, apiKey a
480
502
481// RefreshOAuthToken refreshes the OAuth token for the given provider.
503// RefreshOAuthToken refreshes the OAuth token for the given provider.
482//
504//
483−// It uses two-phase locking: the pre-check (reading the config file to
505+// Providers like Hyper rotate refresh tokens: each exchange consumes the
484−// see if another process already refreshed) happens under the config
506+// caller's refresh token, issues a new pair, and revokes the old one. If
485−// lock, then the HTTP exchange runs without any lock held, and finally
507+// two crush instances (or two goroutines) refresh concurrently with the
486−// the result is persisted via SetConfigFields (which acquires the lock
508+// same stored refresh token, the second exchange reuses an already-revoked
487−// internally). If the exchange fails — e.g. because another process
509+// token, trips the provider's reuse detection, and revokes the entire
488−// already rotated the refresh token — the disk is re-checked under lock
510+// token family — leaving both with dead tokens even though each refresh
489−// to recover the other process's token.
511+// "succeeded".
512+//
513+// To prevent that, refreshes are single-flighted at two levels:
514+//
515+// - In-process: refreshSF collapses concurrent goroutines for the same
516+// provider into one attempt.
517+// - Cross-process: a per-provider advisory lock is held across the whole
518+// read-decide-exchange-write cycle, so only one process exchanges at a
519+// time. A process that acquires the lock after a peer rotated finds the
520+// peer's fresh token on disk and adopts it instead of exchanging.
490func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, providerID string) error {
521func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, providerID string) error {
522+ key := fmt.Sprintf("%d\x00%s", scope, providerID)
523+ _, err, _ := s.refreshSF.Do(key, func() (any, error) {
524+ return nil, s.refreshOAuthTokenLocked(ctx, scope, providerID)
525+ })
526+ return err
527+}
528+
529+// refreshOAuthTokenLocked performs the cross-process single-flighted
530+// refresh. It is invoked through refreshSF, so at most one goroutine per
531+// provider runs it at a time within this process.
532+func (s *ConfigStore) refreshOAuthTokenLocked(ctx context.Context, scope Scope, providerID string) error {
491 cfg := s.Config()
533 cfg := s.Config()
492 providerConfig, exists := cfg.Providers.Get(providerID)
534 providerConfig, exists := cfg.Providers.Get(providerID)
493 if !exists {
535 if !exists {
494 return fmt.Errorf("provider %s not found", providerID)
536 return fmt.Errorf("provider %s not found", providerID)
495 }
537 }
496−
497 if providerConfig.OAuthToken == nil {
538 if providerConfig.OAuthToken == nil {
498 return fmt.Errorf("provider %s does not have an OAuth token", providerID)
539 return fmt.Errorf("provider %s does not have an OAuth token", providerID)
499 }
540 }
541+ entryToken := providerConfig.OAuthToken
500
542
501− // Phase 1: Pre-check under lock — did another process already refresh?
543+ // Acquire the per-provider cross-process refresh lock. This is a
502− release, lockErr := s.lockConfig(scope)
544+ // dedicated lock file, not the config-write lock, and it does not take
545+ // s.mu — so the network exchange below cannot stall unrelated config
546+ // operations. The deadline exceeds the exchange timeout so that a peer
547+ // mid-exchange has time to publish a token we can adopt. Lock ordering:
548+ // the refresh lock is always taken before the config-write lock (via
549+ // SetConfigFields), never the reverse, so no deadlock is possible.
550+ lockCtx, cancel := context.WithTimeout(ctx, refreshLockDeadline)
551+ defer cancel()
552+ release, lockErr := lock.File(lockCtx, s.refreshLockPath(providerID))
503 if lockErr != nil {
553 if lockErr != nil {
504− slog.Warn("Failed to lock config for pre-check, proceeding anyway", "provider", providerID, "error", lockErr)
554+ // Could not acquire the lock (peer wedged or deadline hit). Prefer a
505− } else {
555+ // usable token already on disk over forcing our own exchange, which
506− diskToken, err := s.loadTokenFromDisk(scope, providerID)
556+ // would risk reusing a rotated refresh token.
507− release()
557+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
508− if err != nil {
509− slog.Warn("Failed to read token from config file", "provider", providerID, "error", err)
558+ slog.Warn("Refresh lock unavailable; adopting token from disk", "provider", providerID, "error", lockErr)
510− } else if diskToken != nil && !diskToken.IsExpired() && diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
511− slog.Info("Using token refreshed by another session", "provider", providerID)
512 return s.applyToken(providerConfig, diskToken, providerID)
559 return s.applyToken(providerConfig, diskToken, providerID)
513 }
560 }
561+ return fmt.Errorf("acquire refresh lock for provider %s: %w", providerID, lockErr)
514 }
562 }
563+ defer release()
515
564
516− // Phase 2: HTTP exchange — no lock held.
565+ // Did a peer rotate the token while we waited for the lock? If disk now
517− var refreshedToken *oauth.Token
566+ // holds a different, unexpired token, adopt it instead of exchanging.
518− var refreshErr error
567+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
519− switch providerID {
568+ slog.Info("Adopting token refreshed by another session", "provider", providerID)
520− case string(catwalk.InferenceProviderCopilot):
569+ return s.applyToken(providerConfig, diskToken, providerID)
521− refreshedToken, refreshErr = copilot.RefreshToken(ctx, providerConfig.OAuthToken.RefreshToken)
522− case hyperp.Name:
523− refreshedToken, refreshErr = hyper.ExchangeToken(ctx, providerConfig.OAuthToken.RefreshToken)
524− default:
525− return fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
526 }
570 }
571+
572+ // Disk still holds our token (or no usable peer token exists) and we hold
573+ // the lock, so we are the sole exchanger. Perform the exchange.
574+ refreshedToken, refreshErr := s.exchange(ctx, providerID, entryToken.RefreshToken)
527 if refreshErr != nil {
575 if refreshErr != nil {
528− // Phase 3: Fallback — re-check disk under lock. The exchange may
529− // have failed because another process already rotated the refresh
576+ // The exchange may have failed because a peer rotated the refresh
530− // token.
577+ // token in a window we did not cover. Re-check disk and adopt.
531− if release, lockErr := s.lockConfig(scope); lockErr == nil {
578+ if diskToken := s.adoptableDiskToken(scope, providerID, entryToken); diskToken != nil {
532− diskToken, diskErr := s.loadTokenFromDisk(scope, providerID)
533− release()
534− if diskErr == nil &&
535− diskToken != nil &&
536− !diskToken.IsExpired() &&
537− diskToken.AccessToken != providerConfig.OAuthToken.AccessToken {
538− slog.Info("Using token refreshed by another session after exchange failure", "provider", providerID)
579+ slog.Info("Adopting token refreshed by another session after exchange failure", "provider", providerID)
539− return s.applyToken(providerConfig, diskToken, providerID)
580+ return s.applyToken(providerConfig, diskToken, providerID)
540− }
541 }
581 }
542 return fmt.Errorf("failed to refresh OAuth token for provider %s: %w", providerID, refreshErr)
582 return fmt.Errorf("failed to refresh OAuth token for provider %s: %w", providerID, refreshErr)
543 }
583 }
@@ -545,12 +585,9 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
545 slog.Info("Successfully refreshed OAuth token", "provider", providerID)
585 slog.Info("Successfully refreshed OAuth token", "provider", providerID)
546 providerConfig.OAuthToken = refreshedToken
586 providerConfig.OAuthToken = refreshedToken
547 providerConfig.APIKey = refreshedToken.AccessToken
587 providerConfig.APIKey = refreshedToken.AccessToken
548−
549− switch providerID {
550− case string(catwalk.InferenceProviderCopilot):
588+ if providerID == string(catwalk.InferenceProviderCopilot) {
551 providerConfig.SetupGitHubCopilot()
589 providerConfig.SetupGitHubCopilot()
552 }
590 }
553−
554 cfg.Providers.Set(providerID, providerConfig)
591 cfg.Providers.Set(providerID, providerConfig)
555
592
556 if err := s.SetConfigFields(scope, map[string]any{
593 if err := s.SetConfigFields(scope, map[string]any{
@@ -559,10 +596,57 @@ func (s *ConfigStore) RefreshOAuthToken(ctx context.Context, scope Scope, provid
559 }); err != nil {
596 }); err != nil {
560 return fmt.Errorf("failed to persist refreshed token: %w", err)
597 return fmt.Errorf("failed to persist refreshed token: %w", err)
561 }
598 }
562−
563 return nil
599 return nil
564}
600}
565
601
602+// adoptableDiskToken returns the on-disk token for the provider when it is
603+// usable and differs from entryToken — i.e. when another session has
604+// already refreshed it and we should adopt that result rather than running
605+// our own exchange. It returns nil when there is nothing newer to adopt.
606+func (s *ConfigStore) adoptableDiskToken(scope Scope, providerID string, entryToken *oauth.Token) *oauth.Token {
607+ diskToken, err := s.loadTokenFromDisk(scope, providerID)
608+ if err != nil {
609+ slog.Warn("Failed to read token from config file", "provider", providerID, "error", err)
610+ return nil
611+ }
612+ if diskToken == nil || diskToken.IsExpired() {
613+ return nil
614+ }
615+ if diskToken.AccessToken == entryToken.AccessToken {
616+ // Same token we started with; nobody refreshed since.
617+ return nil
618+ }
619+ return diskToken
620+}
621+
622+// exchange performs the provider-specific OAuth token exchange. Tests may
623+// override it via the exchangeToken field; production uses the real
624+// provider clients.
625+func (s *ConfigStore) exchange(ctx context.Context, providerID, refreshToken string) (*oauth.Token, error) {
626+ if s.exchangeToken != nil {
627+ return s.exchangeToken(ctx, providerID, refreshToken)
628+ }
629+ switch providerID {
630+ case string(catwalk.InferenceProviderCopilot):
631+ return copilot.RefreshToken(ctx, refreshToken)
632+ case hyperp.Name:
633+ return hyper.ExchangeToken(ctx, refreshToken)
634+ default:
635+ return nil, fmt.Errorf("OAuth refresh not supported for provider %s", providerID)
636+ }
637+}
638+
639+// refreshLockPath returns the path to the per-provider cross-process refresh
640+// lock file. Lock files live under a dedicated locks/ subdirectory of the
641+// data dir so they do not clutter the config directory. The file is created
642+// on demand by lock.File and is never removed (flock keys on inode, not
643+// path).
644+func (s *ConfigStore) refreshLockPath(providerID string) string {
645+ dir := filepath.Join(filepath.Dir(s.globalDataPath), "locks")
646+ _ = os.MkdirAll(dir, 0o755)
647+ return filepath.Join(dir, fmt.Sprintf("%s.refresh.lock", providerID))
648+}
649+
566// applyToken updates the in-memory provider config with the given token.
650// applyToken updates the in-memory provider config with the given token.
567func (s *ConfigStore) applyToken(providerConfig ProviderConfig, token *oauth.Token, providerID string) error {
651func (s *ConfigStore) applyToken(providerConfig ProviderConfig, token *oauth.Token, providerID string) error {
568 providerConfig.OAuthToken = token
652 providerConfig.OAuthToken = token