From 55391dc24ca3e26e0d722961386b1f21395b07c0 Mon Sep 17 00:00:00 2001 From: Michael Tarassov Date: Sun, 26 Jul 2026 19:18:46 +0800 Subject: [PATCH] fix: serialize Grok token refresh to stop concurrent invalid_grant races GetAccessToken had no real locking around the actual refresh HTTP call: the mutex was only held long enough to copy fields, then released before the network round trip. When multiple requests arrived right at token expiry, each one read the same stale refresh_token and POSTed it to the provider concurrently. Since OAuth providers commonly rotate (single-use) refresh tokens, only one of those calls succeeds; the rest come back invalid_grant and the losing goroutines returned an error to their callers even though a sibling had just refreshed the credential successfully. Add a dedicated refreshMu that is held for the whole refresh operation, and after acquiring it re-check whether the entry is already fresh so a caller that lost the race for the lock just observes the token a sibling refreshed instead of hitting the provider again. ForceRefresh still always attempts a refresh (used after an upstream 401), but is now serialized against the same mutex so a burst of 401s does not also stampede the token endpoint. Added TestConcurrentGetAccessTokenDedupesRefresh, which simulates refresh token rotation and reliably failed against the old code (5/5 runs) before the fix. --- internal/auth/auth.go | 39 ++++++++++++++--- internal/auth/auth_test.go | 89 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 7 deletions(-) diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 9508f39..e83dbe6 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -75,6 +75,14 @@ type Manager struct { tokenEndpoint string writing bool // skip fsnotify events from our own writes + // refreshMu serializes the actual refresh HTTP round trip. Without it, + // concurrent callers racing GetAccessToken right at token expiry each read + // the same refresh_token and POST it to the provider concurrently; most + // OAuth providers rotate (single-use) refresh tokens, so only one of those + // requests succeeds and the rest fail with invalid_grant even though the + // manager holds a valid, freshly-refreshed token by the time they return. + refreshMu sync.Mutex + watcher *fsnotify.Watcher stopCh chan struct{} wg sync.WaitGroup @@ -363,7 +371,7 @@ func (m *Manager) GetAccessToken(ctx context.Context) (string, error) { // ForceRefresh refreshes regardless of expiry (e.g. after upstream 401). func (m *Manager) ForceRefresh(ctx context.Context) error { - return m.Refresh(ctx) + return m.doRefresh(ctx, true) } // StartProactiveRefresh periodically refreshes the access token before expiry @@ -412,9 +420,15 @@ func (m *Manager) proactiveTick() { m.log.Info("proactive token refresh ok", zap.Time("expires_at", m.ExpiresAt())) } -// Refresh exchanges the refresh_token for a new access token. +// Refresh exchanges the refresh_token for a new access token if it is at or +// near expiry. Concurrent calls are deduplicated: a caller that loses the +// race for refreshMu simply observes the token a sibling call just refreshed. func (m *Manager) Refresh(ctx context.Context) error { - err := m.refresh(ctx) + return m.doRefresh(ctx, false) +} + +func (m *Manager) doRefresh(ctx context.Context, force bool) error { + err := m.refresh(ctx, force) if m.metrics != nil { m.metrics.IncAuthRefresh(err == nil) } @@ -424,14 +438,25 @@ func (m *Manager) Refresh(ctx context.Context) error { return err } -func (m *Manager) refresh(ctx context.Context) error { - m.mu.Lock() - // Serialize refreshes +func (m *Manager) refresh(ctx context.Context, force bool) error { + // Only one refresh HTTP round trip may be in flight at a time (see + // refreshMu doc comment on the Manager struct). + m.refreshMu.Lock() + defer m.refreshMu.Unlock() + + m.mu.RLock() entry := m.entry mapKey := m.mapKey + skew := m.refreshSkew clientID := firstNonEmpty(entry.resolvedClientID(), m.clientID) issuer := firstNonEmpty(entry.resolvedIssuer(), m.issuer) - m.mu.Unlock() + m.mu.RUnlock() + + if !force && !entry.ExpiresAt.IsZero() && time.Now().Add(skew).Before(entry.ExpiresAt) { + // A concurrent caller already refreshed while we waited for refreshMu; + // the entry is fresh enough now, nothing more to do. + return nil + } if strings.TrimSpace(entry.RefreshToken) == "" { return errors.New("no refresh_token available") diff --git a/internal/auth/auth_test.go b/internal/auth/auth_test.go index 5fc89cd..5b0c690 100644 --- a/internal/auth/auth_test.go +++ b/internal/auth/auth_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "sync" "testing" "time" ) @@ -226,3 +227,91 @@ func writeTestAuthWithIssuer(t *testing.T, path, key, refresh string, exp time.T t.Fatal(err) } } + +// TestConcurrentGetAccessTokenDedupesRefresh guards against a real race: many +// requests arriving right at token expiry each observe the same stale +// refresh_token and, without serialization, would each POST it to the +// provider concurrently. Real OAuth providers commonly rotate (single-use) +// refresh tokens, so a second use of the same token is rejected — meaning the +// losing goroutine would fail even though a sibling goroutine successfully +// refreshed the credential moments earlier. This simulates that rotation +// behavior and asserts every concurrent caller still gets the fresh token, +// and that only one refresh request actually reaches the provider. +func TestConcurrentGetAccessTokenDedupesRefresh(t *testing.T) { + var mu sync.Mutex + seen := map[string]bool{} + hits := 0 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/.well-known/openid-configuration": + _ = json.NewEncoder(w).Encode(map[string]string{ + "token_endpoint": "http://" + r.Host + "/oauth/token", + }) + case r.URL.Path == "/oauth/token": + _ = r.ParseForm() + rt := r.Form.Get("refresh_token") + mu.Lock() + hits++ + reused := seen[rt] + seen[rt] = true + mu.Unlock() + if reused { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"invalid_grant"}`)) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "access_token": "new-access", + "refresh_token": "new-refresh", + "expires_in": 3600, + "token_type": "Bearer", + }) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + dir := t.TempDir() + path := filepath.Join(dir, "auth.json") + writeTestAuthWithIssuer(t, path, "old-access", "old-refresh", time.Now().Add(2*time.Second), srv.URL, "cid") + + m, err := NewManager(Options{ + Path: path, + Issuer: srv.URL, + ClientID: "cid", + RefreshSkew: 5 * time.Minute, + HTTPClient: srv.Client(), + }) + if err != nil { + t.Fatal(err) + } + + const n = 20 + var wg sync.WaitGroup + errs := make([]error, n) + toks := make([]string, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + toks[i], errs[i] = m.GetAccessToken(context.Background()) + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("goroutine %d: GetAccessToken failed: %v", i, err) + } + if toks[i] != "new-access" { + t.Fatalf("goroutine %d: got token %q, want new-access", i, toks[i]) + } + } + mu.Lock() + defer mu.Unlock() + if hits != 1 { + t.Fatalf("expected exactly 1 refresh request to reach the provider, got %d", hits) + } +}