Make refresh token rotation atomic

pkg/auth/authorization_code.go

@@ -1,6 +1,7 @@
 package auth
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -362,46 +363,70 @@ func (s *authorizationServer) handleRefreshTokenGrant(w http.ResponseWriter, r *
 		return
 	}
 
+	// Copy every field we need out of the session while the lock is held, and
+	// never touch the *refreshSession pointer again afterward. A concurrent
+	// request presenting the same token can rotate or delete it at any point
+	// once we release the lock, so any later read through the pointer would
+	// race that request's writes to the same object.
 	s.refreshSessionsMu.RLock()
 	session, ok := s.refreshSessions[refreshToken]
+	var (
+		sessionClientID   string
+		sessionResource   string
+		sessionGitHubID   int64
+		sessionGitHub     string
+		sessionGitHubUser string
+		sessionOrgs       []string
+		sessionExpiresAt  time.Time
+	)
+	if ok {
+		sessionClientID = session.ClientID
+		sessionResource = session.Resource
+		sessionGitHubID = session.GitHubID
+		sessionGitHub = session.GitHubAccessToken
+		sessionGitHubUser = session.GitHubLogin
+		sessionOrgs = append([]string(nil), session.Orgs...)
+		sessionExpiresAt = session.ExpiresAt
+	}
 	s.refreshSessionsMu.RUnlock()
+
 	if !ok {
 		s.writeError(w, http.StatusBadRequest, "invalid_grant", "invalid refresh token")
 		return
 	}
 
-	if time.Now().After(session.ExpiresAt) {
+	if time.Now().After(sessionExpiresAt) {
 		s.refreshSessionsMu.Lock()
 		delete(s.refreshSessions, refreshToken)
 		s.refreshSessionsMu.Unlock()
 		s.writeError(w, http.StatusBadRequest, "invalid_grant", "refresh token expired")
 		return
 	}
 
-	if session.ClientID != clientID {
+	if sessionClientID != clientID {
 		s.writeError(w, http.StatusBadRequest, "invalid_grant", "parameter mismatch")
 		return
 	}
 
-	if resource != "" && session.Resource != resource {
+	if resource != "" && sessionResource != resource {
 		s.writeError(w, http.StatusBadRequest, "invalid_grant", "parameter mismatch")
 		return
 	}
 
-	githubToken := session.GitHubAccessToken
-	githubLogin := session.GitHubLogin
-	githubID := session.GitHubID
-	orgs := append([]string(nil), session.Orgs...)
+	githubToken := sessionGitHub
+	githubLogin := sessionGitHubUser
+	githubID := sessionGitHubID
+	orgs := sessionOrgs
 
 	if len(s.allowedOrgs) > 0 {
-		githubUser, err := s.github.GetUser(r.Context(), session.GitHubAccessToken)
+		githubUser, err := s.github.GetUser(r.Context(), sessionGitHub)
 		if err != nil {
-			s.log.WithError(err).WithField("login", session.GitHubLogin).Warn("Failed to verify GitHub org membership during refresh")
+			s.log.WithError(err).WithField("login", sessionGitHubUser).Warn("Failed to verify GitHub org membership during refresh")
 			s.writeError(w, http.StatusServiceUnavailable, "temporarily_unavailable", "could not verify organization membership")
 			return
 		}
 
-		if githubUser.ID != session.GitHubID {
+		if githubUser.ID != sessionGitHubID {
 			s.refreshSessionsMu.Lock()
 			delete(s.refreshSessions, refreshToken)
 			s.refreshSessionsMu.Unlock()
@@ -419,24 +444,22 @@ func (s *authorizationServer) handleRefreshTokenGrant(w http.ResponseWriter, r *
 
 		githubLogin = githubUser.Login
 		orgs = append([]string(nil), githubUser.Organizations...)
-
-		s.refreshSessionsMu.Lock()
-		if current := s.refreshSessions[refreshToken]; current != nil {
-			current.GitHubLogin = githubUser.Login
-			current.Orgs = append([]string(nil), githubUser.Organizations...)
-		}
-		s.refreshSessionsMu.Unlock()
 	}
 
-	accessToken, err := s.issueAccessToken(s.issuerURL, session.Resource, githubLogin, githubID, orgs)
+	accessToken, err := s.issueAccessToken(s.issuerURL, sessionResource, githubLogin, githubID, orgs)
 	if err != nil {
 		s.log.WithError(err).Error("Failed to sign refreshed token")
 		s.writeError(w, http.StatusInternalServerError, "server_error", "failed to create token")
 		return
 	}
 
-	newRefreshToken, err := s.rotateRefreshToken(refreshToken, session, githubLogin, githubID, githubToken, orgs)
+	newRefreshToken, err := s.rotateRefreshToken(refreshToken, sessionClientID, sessionResource, githubLogin, githubID, githubToken, orgs)
 	if err != nil {
+		if errors.Is(err, errRefreshTokenAlreadyConsumed) {
+			s.writeError(w, http.StatusBadRequest, "invalid_grant", "refresh token already used")
+			return
+		}
+
 		s.log.WithError(err).Error("Failed to rotate refresh session")
 		s.writeError(w, http.StatusInternalServerError, "server_error", "failed to rotate refresh token")
 		return

pkg/auth/refresh_token_concurrency_test.go

@@ -0,0 +1,142 @@
+package auth
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/ethpandaops/panda/pkg/auth/github"
+)
+
+// barrierGitHubClient blocks inside GetUser until told to proceed, so a test
+// can force two concurrent refresh requests to both pass the initial session
+// lookup before either one rotates the token.
+type barrierGitHubClient struct {
+	user    *github.GitHubUser
+	arrived chan struct{}
+	release chan struct{}
+}
+
+func (b *barrierGitHubClient) GetAuthorizationURL(_, _, _ string) string {
+	return "https://github.example.test/oauth"
+}
+
+func (b *barrierGitHubClient) ExchangeCode(_ context.Context, _, _ string) (*github.TokenResponse, error) {
+	return &github.TokenResponse{AccessToken: "github-access-token"}, nil
+}
+
+func (b *barrierGitHubClient) GetUser(_ context.Context, _ string) (*github.GitHubUser, error) {
+	b.arrived <- struct{}{}
+	<-b.release
+
+	return b.user, nil
+}
+
+func postRefreshToken(svc *authorizationServer, refreshToken string) (int, tokenResponseBody) {
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "http://internal-proxy/auth/token", strings.NewReader(url.Values{
+		"grant_type":    {"refresh_token"},
+		"refresh_token": {refreshToken},
+		"client_id":     {"panda"},
+		"resource":      {testIssuerURL},
+	}.Encode()))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	svc.handleToken(rec, req)
+
+	var body tokenResponseBody
+	_ = json.Unmarshal(rec.Body.Bytes(), &body)
+
+	return rec.Code, body
+}
+
+// TestConcurrentRefreshOfSameTokenRotatesExactlyOnce presents the same refresh
+// token from two concurrent requests. Only one may succeed in rotating it;
+// the other must see the token as already consumed rather than mint its own
+// independent successor.
+func TestConcurrentRefreshOfSameTokenRotatesExactlyOnce(t *testing.T) {
+	t.Parallel()
+
+	svc := newTestAuthorizationServer(t, []string{"ethpandaops"})
+	barrier := &barrierGitHubClient{
+		user: &github.GitHubUser{ID: 42, Login: "sam", Organizations: []string{"ethpandaops"}},
+		// Buffered so both goroutines can signal arrival without blocking on a
+		// reader that hasn't reached the receive yet.
+		arrived: make(chan struct{}, 2),
+		release: make(chan struct{}),
+	}
+	svc.github = barrier
+
+	refreshToken, err := svc.issueRefreshToken("panda", testIssuerURL, "sam", 42, "github-access-token", []string{"ethpandaops"})
+	if err != nil {
+		t.Fatalf("issueRefreshToken failed: %v", err)
+	}
+
+	type result struct {
+		status int
+		body   tokenResponseBody
+	}
+
+	results := make(chan result, 2)
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	for i := 0; i < 2; i++ {
+		go func() {
+			defer wg.Done()
+			status, body := postRefreshToken(svc, refreshToken)
+			results <- result{status, body}
+		}()
+	}
+
+	// Wait for both requests to reach the GitHub lookup, then release them so
+	// they attempt to rotate at the same time.
+	<-barrier.arrived
+	<-barrier.arrived
+	close(barrier.release)
+
+	wg.Wait()
+	close(results)
+
+	var successes, rejections int
+	var newTokens []string
+
+	for r := range results {
+		switch r.status {
+		case http.StatusOK:
+			successes++
+			if r.body.RefreshToken == "" {
+				t.Fatal("successful refresh did not return a new refresh token")
+			}
+			newTokens = append(newTokens, r.body.RefreshToken)
+		case http.StatusBadRequest:
+			rejections++
+		default:
+			t.Fatalf("unexpected status %d", r.status)
+		}
+	}
+
+	if successes != 1 || rejections != 1 {
+		t.Fatalf("expected exactly one success and one rejection, got %d successes and %d rejections", successes, rejections)
+	}
+
+	svc.refreshSessionsMu.RLock()
+	_, originalStillLive := svc.refreshSessions[refreshToken]
+	_, newTokenLive := svc.refreshSessions[newTokens[0]]
+	total := len(svc.refreshSessions)
+	svc.refreshSessionsMu.RUnlock()
+
+	if originalStillLive {
+		t.Fatal("original refresh token should have been consumed")
+	}
+	if !newTokenLive {
+		t.Fatal("the successful rotation's new refresh token should be live")
+	}
+	if total != 1 {
+		t.Fatalf("expected exactly one live refresh session after rotation, got %d", total)
+	}
+}

pkg/auth/token.go

@@ -5,13 +5,18 @@ import (
 	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"time"
 
 	"github.com/golang-jwt/jwt/v5"
 )
 
+// errRefreshTokenAlreadyConsumed is returned by rotateRefreshToken when the
+// presented refresh token has already been rotated by a concurrent request.
+var errRefreshTokenAlreadyConsumed = errors.New("refresh token already consumed")
+
 func (s *authorizationServer) issueAccessToken(
 	issuerURL, resource, githubLogin string, githubID int64, orgs []string,
 ) (string, error) {
@@ -77,10 +82,13 @@ func (s *authorizationServer) issueRefreshToken(
 	return refreshToken, nil
 }
 
+// rotateRefreshToken atomically consumes currentRefreshToken and replaces it
+// with a newly issued one. The presence check and the delete-then-insert
+// happen under a single lock, so a currentRefreshToken already consumed by a
+// concurrent request (returning errRefreshTokenAlreadyConsumed here) can never
+// be rotated twice into two independently live token families.
 func (s *authorizationServer) rotateRefreshToken(
-	currentRefreshToken string,
-	session *refreshSession,
-	githubLogin string,
+	currentRefreshToken, clientID, resource, githubLogin string,
 	githubID int64,
 	githubToken string,
 	orgs []string,
@@ -91,18 +99,23 @@ func (s *authorizationServer) rotateRefreshToken(
 	}
 
 	s.refreshSessionsMu.Lock()
+	defer s.refreshSessionsMu.Unlock()
+
+	if _, stillPresent := s.refreshSessions[currentRefreshToken]; !stillPresent {
+		return "", errRefreshTokenAlreadyConsumed
+	}
+
 	delete(s.refreshSessions, currentRefreshToken)
 	s.refreshSessions[newRefreshToken] = &refreshSession{
-		ClientID:          session.ClientID,
-		Resource:          session.Resource,
+		ClientID:          clientID,
+		Resource:          resource,
 		GitHubLogin:       githubLogin,
 		GitHubID:          githubID,
 		GitHubAccessToken: githubToken,
 		Orgs:              append([]string(nil), orgs...),
 		CreatedAt:         time.Now(),
 		ExpiresAt:         time.Now().Add(s.refreshTokenTTL),
 	}
-	s.refreshSessionsMu.Unlock()
 
 	return newRefreshToken, nil
 }