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) + } +}