Add compare-and-set update for DCR credentials

pkg/authserver/storage/memory.go

@@ -2058,6 +2058,48 @@ func (s *MemoryStorage) UpdateDCRCredentialsIfPresent(_ context.Context, creds *
 	return cloneDCRCredentials(creds), nil
 }
 
+// UpdateDCRCredentialsIfUnchanged replaces the entry at creds.Key with creds
+// only when the stored entry still equals expected, returning ErrNotFound
+// (wrapped) when no entry exists and ErrDCRCredentialsChanged when it differs.
+// The compare and the write happen under s.mu, so no other writer can slip in
+// between them. Presence and ClientSecretExpiresAt handling are identical to
+// UpdateDCRCredentialsIfPresent.
+func (s *MemoryStorage) UpdateDCRCredentialsIfUnchanged(
+	_ context.Context, creds, expected *DCRCredentials,
+) (*DCRCredentials, error) {
+	if err := validateDCRCompareAndSet(creds, expected); err != nil {
+		return nil, err
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	existing, ok := s.dcrCredentials[creds.Key]
+	if !ok {
+		return nil, notFoundRFC6749Error("DCR credentials not found")
+	}
+	if !dcrCredentialsEqual(existing, expected) {
+		return nil, ErrDCRCredentialsChanged
+	}
+
+	s.dcrCredentials[creds.Key] = cloneDCRCredentials(creds)
+	return cloneDCRCredentials(creds), nil
+}
+
+// dcrCredentialsEqual reports whether a and b hold the same persisted values.
+// Time fields are compared with time.Time.Equal rather than ==, so two values
+// for the same instant compare equal regardless of monotonic-clock reading or
+// location.
+func dcrCredentialsEqual(a, b *DCRCredentials) bool {
+	ac, bc := *a, *b
+	if !ac.CreatedAt.Equal(bc.CreatedAt) || !ac.ClientSecretExpiresAt.Equal(bc.ClientSecretExpiresAt) {
+		return false
+	}
+	ac.CreatedAt, bc.CreatedAt = time.Time{}, time.Time{}
+	ac.ClientSecretExpiresAt, bc.ClientSecretExpiresAt = time.Time{}, time.Time{}
+	return ac == bc
+}
+
 // GetDCRCredentials retrieves DCR credentials by key.
 // Returns a defensive copy; returns ErrNotFound (wrapped) on miss.
 func (s *MemoryStorage) GetDCRCredentials(_ context.Context, key DCRKey) (*DCRCredentials, error) {

pkg/authserver/storage/memory_test.go

@@ -3026,6 +3026,144 @@ func TestMemoryStorage_DCRCredentials_UpdateCopyIsolatesCaller(t *testing.T) {
 	})
 }
 
+// TestMemoryStorage_DCRCredentials_UpdateIfUnchanged pins the
+// UpdateDCRCredentialsIfUnchanged compare-and-set contract: a row that still
+// equals expected is overwritten, a row that changed after expected was read
+// is refused with ErrDCRCredentialsChanged, and an absent row is refused with
+// ErrNotFound — in both refusal cases nothing is written.
+func TestMemoryStorage_DCRCredentials_UpdateIfUnchanged(t *testing.T) {
+	t.Parallel()
+
+	t.Run("unchanged row is overwritten", func(t *testing.T) {
+		withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+			key := dcrFixtureKey()
+			_, err := s.StoreDCRCredentialsIfAbsent(ctx, dcrCASFixture(key, "original"))
+			require.NoError(t, err)
+			expected, err := s.GetDCRCredentials(ctx, key)
+			require.NoError(t, err)
+
+			replacement := dcrCASFixture(key, "replacement")
+			got, err := s.UpdateDCRCredentialsIfUnchanged(ctx, replacement, expected)
+			require.NoError(t, err)
+			assert.Equal(t, *replacement, *got)
+
+			reread, err := s.GetDCRCredentials(ctx, key)
+			require.NoError(t, err)
+			assert.Equal(t, *replacement, *reread)
+		})
+	})
+
+	t.Run("changed row is refused and left intact", func(t *testing.T) {
+		withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+			key := dcrFixtureKey()
+			_, err := s.StoreDCRCredentialsIfAbsent(ctx, dcrCASFixture(key, "original"))
+			require.NoError(t, err)
+			stale, err := s.GetDCRCredentials(ctx, key)
+			require.NoError(t, err)
+
+			// A concurrent writer replaces the row after the stale read.
+			newer := dcrCASFixture(key, "newer")
+			_, err = s.UpdateDCRCredentialsIfPresent(ctx, newer)
+			require.NoError(t, err)
+
+			_, err = s.UpdateDCRCredentialsIfUnchanged(ctx, dcrCASFixture(key, "stale-writer"), stale)
+			require.ErrorIs(t, err, ErrDCRCredentialsChanged)
+
+			reread, err := s.GetDCRCredentials(ctx, key)
+			require.NoError(t, err)
+			assert.Equal(t, *newer, *reread, "a refused compare-and-set must not write")
+		})
+	})
+
+	t.Run("absent row returns not-found and does not create", func(t *testing.T) {
+		withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+			key := dcrFixtureKey()
+			_, err := s.UpdateDCRCredentialsIfUnchanged(ctx,
+				dcrCASFixture(key, "replacement"), dcrCASFixture(key, "original"))
+			requireNotFoundError(t, err)
+
+			_, getErr := s.GetDCRCredentials(ctx, key)
+			requireNotFoundError(t, getErr)
+		})
+	})
+
+	t.Run("equal instants in different locations compare equal", func(t *testing.T) {
+		withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+			key := dcrFixtureKey()
+			stored := dcrCASFixture(key, "original")
+			stored.CreatedAt = time.Now()
+			_, err := s.StoreDCRCredentialsIfAbsent(ctx, stored)
+			require.NoError(t, err)
+
+			expected := dcrCASFixture(key, "original")
+			expected.CreatedAt = stored.CreatedAt.In(time.FixedZone("elsewhere", 3600))
+			_, err = s.UpdateDCRCredentialsIfUnchanged(ctx, dcrCASFixture(key, "replacement"), expected)
+			require.NoError(t, err)
+		})
+	})
+}
+
+// TestMemoryStorage_DCRCredentials_UpdateIfUnchangedInvalidInput pins the
+// input contract: creds runs the shared validateDCRCredentialsForStore gate,
+// expected must be non-nil, and expected must address the same key as creds.
+func TestMemoryStorage_DCRCredentials_UpdateIfUnchangedInvalidInput(t *testing.T) {
+	t.Parallel()
+
+	key := dcrFixtureKey()
+	otherKey := key
+	otherKey.UpstreamID = "other-upstream"
+
+	tests := []struct {
+		name     string
+		creds    *DCRCredentials
+		expected *DCRCredentials
+	}{
+		{name: "nil creds", creds: nil, expected: dcrCASFixture(key, "original")},
+		{name: "nil expected", creds: dcrCASFixture(key, "replacement"), expected: nil},
+		{name: "mismatched key", creds: dcrCASFixture(key, "replacement"), expected: dcrCASFixture(otherKey, "original")},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+				_, err := s.UpdateDCRCredentialsIfUnchanged(ctx, tt.creds, tt.expected)
+				require.Error(t, err)
+				assert.ErrorIs(t, err, fosite.ErrInvalidRequest)
+			})
+		})
+	}
+}
+
+// TestMemoryStorage_DCRCredentials_UpdateIfUnchangedConcurrent pins that the
+// compare and the write are atomic: of N writers that all read the same row
+// and race to replace it, exactly one wins and the rest are refused.
+func TestMemoryStorage_DCRCredentials_UpdateIfUnchangedConcurrent(t *testing.T) {
+	withStorage(t, func(ctx context.Context, s *MemoryStorage) {
+		const goroutines = 8
+		key := dcrFixtureKey()
+		_, err := s.StoreDCRCredentialsIfAbsent(ctx, dcrCASFixture(key, "original"))
+		require.NoError(t, err)
+		expected, err := s.GetDCRCredentials(ctx, key)
+		require.NoError(t, err)
+
+		start := make(chan struct{})
+		errs := make([]error, goroutines)
+		var wg sync.WaitGroup
+		for g := range goroutines {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				<-start
+				_, errs[g] = s.UpdateDCRCredentialsIfUnchanged(ctx,
+					dcrCASFixture(key, fmt.Sprintf("secret-%d", g)), expected)
+			}()
+		}
+		close(start)
+		wg.Wait()
+
+		requireSingleDCRCASWinner(t, errs)
+	})
+}
+
 // TestMemoryStorage_DCRCredentials_StoreInvalidInputRejected pins the
 // fail-loud-on-invalid-input contract: nil creds, an unpopulated Key
 // (empty Issuer, UpstreamID, RedirectURI, or ScopesHash), and missing RFC 7591

pkg/authserver/storage/redis.go

@@ -1879,6 +1879,33 @@ type storedDCRCredentials struct {
 	ClientSecretExpiresAt int64 `json:"client_secret_expires_at"`
 }
 
+// newStoredDCRCredentials converts creds to its stored form, mirroring
+// newStoredUpstreamTokens. Time fields are truncated to Unix seconds; a zero
+// time.Time becomes the 0 "not set" sentinel.
+func newStoredDCRCredentials(creds *DCRCredentials) storedDCRCredentials {
+	stored := storedDCRCredentials{
+		KeyIssuer:               creds.Key.Issuer,
+		KeyUpstreamID:           creds.Key.UpstreamID,
+		KeyRedirectURI:          creds.Key.RedirectURI,
+		KeyScopesHash:           creds.Key.ScopesHash,
+		ProviderName:            creds.ProviderName,
+		ClientID:                creds.ClientID,
+		ClientSecret:            creds.ClientSecret,
+		TokenEndpointAuthMethod: creds.TokenEndpointAuthMethod,
+		RegistrationAccessToken: creds.RegistrationAccessToken,
+		RegistrationClientURI:   creds.RegistrationClientURI,
+		AuthorizationEndpoint:   creds.AuthorizationEndpoint,
+		TokenEndpoint:           creds.TokenEndpoint,
+	}
+	if !creds.CreatedAt.IsZero() {
+		stored.CreatedAt = creds.CreatedAt.Unix()
+	}
+	if !creds.ClientSecretExpiresAt.IsZero() {
+		stored.ClientSecretExpiresAt = creds.ClientSecretExpiresAt.Unix()
+	}
+	return stored
+}
+
 // toDCRCredentials decodes the stored form back into the public type, mirroring
 // storedUpstreamTokens.toUpstreamTokens. Zero epoch values become the zero
 // time.Time, preserving the "not set" sentinel.
@@ -2024,26 +2051,7 @@ func (s *RedisStorage) StoreDCRCredentialsIfAbsent(ctx context.Context, creds *D
 //
 // See StoreDCRCredentialsIfAbsent's docstring for the past-expiry rationale.
 func marshalDCRCredentialsForStore(creds *DCRCredentials) ([]byte, time.Duration, error) {
-	stored := storedDCRCredentials{
-		KeyIssuer:               creds.Key.Issuer,
-		KeyUpstreamID:           creds.Key.UpstreamID,
-		KeyRedirectURI:          creds.Key.RedirectURI,
-		KeyScopesHash:           creds.Key.ScopesHash,
-		ProviderName:            creds.ProviderName,
-		ClientID:                creds.ClientID,
-		ClientSecret:            creds.ClientSecret,
-		TokenEndpointAuthMethod: creds.TokenEndpointAuthMethod,
-		RegistrationAccessToken: creds.RegistrationAccessToken,
-		RegistrationClientURI:   creds.RegistrationClientURI,
-		AuthorizationEndpoint:   creds.AuthorizationEndpoint,
-		TokenEndpoint:           creds.TokenEndpoint,
-	}
-	if !creds.CreatedAt.IsZero() {
-		stored.CreatedAt = creds.CreatedAt.Unix()
-	}
-	if !creds.ClientSecretExpiresAt.IsZero() {
-		stored.ClientSecretExpiresAt = creds.ClientSecretExpiresAt.Unix()
-	}
+	stored := newStoredDCRCredentials(creds)
 
 	data, err := json.Marshal(stored) //nolint:gosec // G117 - internal Redis storage serialization, not exposed to users
 	if err != nil {
@@ -2186,6 +2194,85 @@ func (s *RedisStorage) UpdateDCRCredentialsIfPresent(ctx context.Context, creds
 	return cloneDCRCredentials(creds), nil
 }
 
+// UpdateDCRCredentialsIfUnchanged replaces the row at creds.Key with creds
+// only when the stored row still equals expected, returning ErrNotFound
+// (wrapped) when no row exists and ErrDCRCredentialsChanged when it differs.
+//
+// The compare and the write run in a WATCH/MULTI transaction on the single
+// row key — the same pattern StoreDCRCredentialsIfAbsent uses — so the
+// operation is atomic against concurrent writers and valid in standalone,
+// Sentinel and Cluster modes alike (one key means one hash slot). If another
+// writer touches the key between the GET and EXEC, EXEC aborts with
+// redis.TxFailedErr and the read-compare-write is retried, up to
+// maxDCRClaimRetries; the retry re-reads the row and so normally resolves to
+// ErrDCRCredentialsChanged.
+//
+// The comparison is between decoded stored forms (expected is normalised via
+// newStoredDCRCredentials), not raw JSON bytes, so it is unaffected by field
+// ordering or by a row written by an older encoder, and a value obtained from
+// GetDCRCredentials always matches the row it was read from despite the
+// one-second time precision of the stored form.
+//
+// Presence and TTL handling are identical to UpdateDCRCredentialsIfPresent:
+// an expired-but-present row is updatable, and the rewritten row's TTL is
+// derived from creds by marshalDCRCredentialsForStore. The write uses SET XX
+// for parity with that method, although WATCH already guarantees the key
+// still exists at EXEC.
+func (s *RedisStorage) UpdateDCRCredentialsIfUnchanged(
+	ctx context.Context, creds, expected *DCRCredentials,
+) (*DCRCredentials, error) {
+	if err := validateDCRCompareAndSet(creds, expected); err != nil {
+		return nil, err
+	}
+
+	key := redisDCRKey(s.keyPrefix, creds.Key)
+
+	data, ttl, err := marshalDCRCredentialsForStore(creds)
+	if err != nil {
+		return nil, err
+	}
+	want := newStoredDCRCredentials(expected)
+
+	txFn := func(tx *redis.Tx) error {
+		existingData, getErr := tx.Get(ctx, key).Bytes()
+		if errors.Is(getErr, redis.Nil) {
+			return notFoundRFC6749Error("DCR credentials not found")
+		}
+		if getErr != nil {
+			return fmt.Errorf("failed to get existing dcr credentials: %w", getErr)
+		}
+
+		var existing storedDCRCredentials
+		if unmarshalErr := json.Unmarshal(existingData, &existing); unmarshalErr != nil {
+			return fmt.Errorf("failed to unmarshal existing dcr credentials: %w", unmarshalErr)
+		}
+		if existing != want {
+			return ErrDCRCredentialsChanged
+		}
+
+		_, pipeErr := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+			pipe.SetArgs(ctx, key, data, redis.SetArgs{Mode: "XX", TTL: ttl})
+			return nil
+		})
+		return pipeErr
+	}
+
+	var watchErr error
+	for attempt := 0; attempt < maxDCRClaimRetries; attempt++ {
+		watchErr = s.client.Watch(ctx, txFn, key)
+		if watchErr == nil {
+			return cloneDCRCredentials(creds), nil
+		}
+		if errors.Is(watchErr, ErrNotFound) || errors.Is(watchErr, ErrDCRCredentialsChanged) {
+			return nil, watchErr
+		}
+		if !errors.Is(watchErr, redis.TxFailedErr) {
+			return nil, fmt.Errorf("failed to compare-and-set dcr credentials: %w", watchErr)
+		}
+	}
+	return nil, fmt.Errorf("failed to compare-and-set dcr credentials after %d attempts: %w", maxDCRClaimRetries, watchErr)
+}
+
 // GetDCRCredentials retrieves the credentials previously persisted under key.
 // Returns ErrNotFound (wrapped) when no entry exists. The returned value is a
 // fresh struct decoded from JSON, which acts as a defensive copy.