prevent concurrent JWT reuse

management/server/auth/session.go

@@ -7,9 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"time"
-
-	"github.com/eko/gocache/lib/v4/cache"
-	"github.com/eko/gocache/lib/v4/store"
 )
 
 const (
@@ -22,12 +19,17 @@ var (
 	ErrTokenExpired     = errors.New("JWT expired")
 )
 
-type SessionStore struct {
+// TokenCache atomically records used JWTs until their expiration.
-	cache *cache.Cache[string]
+type TokenCache interface {
+	SetNX(ctx context.Context, key, value string, ttl time.Duration) (bool, error)
 }
 
-func NewSessionStore(cacheStore store.StoreInterface) *SessionStore {
+type SessionStore struct {
-	return &SessionStore{cache: cache.New[string](cacheStore)}
+	cache TokenCache
+}
+
+func NewSessionStore(cacheStore TokenCache) *SessionStore {
+	return &SessionStore{cache: cacheStore}
 }
 
 // RegisterToken records a JWT until its exp time and rejects reuse.
@@ -38,19 +40,13 @@ func (s *SessionStore) RegisterToken(ctx context.Context, token string, expiresA
 	}
 
 	key := usedTokenKeyPrefix + hashToken(token)
-	_, err := s.cache.Get(ctx, key)
+	created, err := s.cache.SetNX(ctx, key, usedTokenMarker, ttl)
-	if err == nil {
+	if err != nil {
-		return ErrTokenAlreadyUsed
-	}
-
-	var notFound *store.NotFound
-	if !errors.As(err, &notFound) {
-		return fmt.Errorf("failed to lookup used token entry: %w", err)
-	}
-
-	if err := s.cache.Set(ctx, key, usedTokenMarker, store.WithExpiration(ttl)); err != nil {
 		return fmt.Errorf("failed to store used token entry: %w", err)
 	}
+	if !created {
+		return ErrTokenAlreadyUsed
+	}
 
 	return nil
 }

management/server/auth/session_test.go

@@ -2,6 +2,7 @@ package auth
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -38,6 +39,39 @@ func TestSessionStore_RegisterSameTokenTwiceIsRejected(t *testing.T) {
 	assert.ErrorIs(t, err, ErrTokenAlreadyUsed)
 }
 
+func TestSessionStore_ConcurrentRegistrationAllowsOneCaller(t *testing.T) {
+	s := newTestSessionStore(t)
+	ctx := context.Background()
+	const attempts = 100
+
+	start := make(chan struct{})
+	results := make(chan error, attempts)
+	for range attempts {
+		go func() {
+			<-start
+			results <- s.RegisterToken(ctx, "token", time.Now().Add(time.Hour))
+		}()
+	}
+	close(start)
+
+	succeeded := 0
+	alreadyUsed := 0
+	for range attempts {
+		err := <-results
+		switch {
+		case err == nil:
+			succeeded++
+		case errors.Is(err, ErrTokenAlreadyUsed):
+			alreadyUsed++
+		default:
+			require.NoError(t, err)
+		}
+	}
+
+	assert.Equal(t, 1, succeeded)
+	assert.Equal(t, attempts-1, alreadyUsed)
+}
+
 func TestSessionStore_RegisterDifferentTokensAreIndependent(t *testing.T) {
 	s := newTestSessionStore(t)
 	ctx := context.Background()
@@ -72,6 +106,23 @@ func TestSessionStore_EntryEvictsAtTTLAndAllowsReRegistration(t *testing.T) {
 	require.NoError(t, s.RegisterToken(ctx, token, time.Now().Add(time.Hour)))
 }
 
+type failingTokenCache struct {
+	err error
+}
+
+func (f failingTokenCache) SetNX(context.Context, string, string, time.Duration) (bool, error) {
+	return false, f.err
+}
+
+func TestSessionStore_CacheErrorIsReturned(t *testing.T) {
+	cacheErr := errors.New("cache unavailable")
+	s := NewSessionStore(failingTokenCache{err: cacheErr})
+
+	err := s.RegisterToken(context.Background(), "token", time.Now().Add(time.Hour))
+	require.Error(t, err)
+	assert.ErrorIs(t, err, cacheErr)
+}
+
 func TestHashToken_StableAndDoesNotLeak(t *testing.T) {
 	a := hashToken("tokenA")
 	b := hashToken("tokenB")