feat: extract DigestReplayGuard from Vault RequestAuthorizer (#21663)

prashantkumar1982 · web-flow · commit cd466fddb95f · 2026-03-24T22:06:35.000Z
* feat: extract DigestReplayGuard from Vault RequestAuthorizer

Extract the request-digest deduplication logic from requestAuthorizer's
internal map into a standalone DigestReplayGuard component. The
requestAuthorizer now delegates to DigestReplayGuard internally — no
behavior change for the existing on-chain allowlist flow.

DigestReplayGuard will also be used by the upcoming JWT auth flow
(both at the gateway handler and capability gateway handler layers)
to reject replayed requests using the JWT's request_digest claim.
Having it as a standalone component allows both auth flows to share
the same dedup mechanism.

Includes 9 unit tests covering: first-call success, duplicate rejection,
expiry cleanup, mixed-expiry scenarios, concurrent access safety, and
empty-digest edge case. All existing RequestAuthorizer tests pass as-is.

Made-with: Cursor

* fix: preserve eager expired-entry cleanup in AuthorizeRequest

Before the DigestReplayGuard extraction, clearExpiredAuthorizedRequests
ran via defer on every AuthorizeRequest call, pruning stale entries even
when the request was rejected by the allowlist or expiry check. After
the refactor, cleanup only ran inside CheckAndRecord on the success
path, allowing unbounded accumulation of expired digests under sustained
rejected traffic.

Add a public ClearExpired() method to DigestReplayGuard and call it via
defer in AuthorizeRequest to restore the original eager-cleanup
behavior. Add a dedicated unit test for independent expiry pruning.

Made-with: Cursor
diff --git a/core/capabilities/vault/digest_replay_guard.go b/core/capabilities/vault/digest_replay_guard.go
@@ -0,0 +1,63 @@
+package vault
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+var ErrDigestAlreadySeen = errors.New("request already authorized previously")
+
+// DigestReplayGuard prevents replay of already-processed requests by tracking
+// request digests with expiry timestamps. It is safe for concurrent use.
+//
+// Used by both the on-chain allowlist flow and the JWT auth flow to ensure
+// that a given request digest is only accepted once.
+type DigestReplayGuard struct {
+	mu      sync.Mutex
+	seen    map[string]int64 // digest → unix expiry timestamp
+	nowFunc func() time.Time // injectable for testing
+}
+
+func NewDigestReplayGuard() *DigestReplayGuard {
+	return &DigestReplayGuard{
+		seen:    make(map[string]int64),
+		nowFunc: time.Now,
+	}
+}
+
+// CheckAndRecord returns ErrDigestAlreadySeen if the digest was previously
+// recorded and has not yet expired. Otherwise it records the digest with
+// the given expiry timestamp (unix seconds, UTC).
+//
+// Expired entries are cleaned up on every call.
+func (g *DigestReplayGuard) CheckAndRecord(digest string, expiresAtUnix int64) error {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+
+	g.clearExpiredLocked()
+
+	if _, exists := g.seen[digest]; exists {
+		return ErrDigestAlreadySeen
+	}
+
+	g.seen[digest] = expiresAtUnix
+	return nil
+}
+
+// ClearExpired removes all entries whose expiry timestamp is in the past.
+// Call this to eagerly reclaim memory even when CheckAndRecord is not invoked.
+func (g *DigestReplayGuard) ClearExpired() {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.clearExpiredLocked()
+}
+
+func (g *DigestReplayGuard) clearExpiredLocked() {
+	now := g.nowFunc().UTC().Unix()
+	for digest, expiry := range g.seen {
+		if now > expiry {
+			delete(g.seen, digest)
+		}
+	}
+}
diff --git a/core/capabilities/vault/digest_replay_guard_test.go b/core/capabilities/vault/digest_replay_guard_test.go
@@ -0,0 +1,183 @@
+package vault
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDigestReplayGuard_FirstCallSucceeds(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	err := guard.CheckAndRecord("digest-1", futureExpiry)
+	require.NoError(t, err)
+}
+
+func TestDigestReplayGuard_DuplicateRejected(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	err := guard.CheckAndRecord("digest-1", futureExpiry)
+	require.NoError(t, err)
+
+	err = guard.CheckAndRecord("digest-1", futureExpiry)
+	require.ErrorIs(t, err, ErrDigestAlreadySeen)
+}
+
+func TestDigestReplayGuard_DifferentDigestsIndependent(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	require.NoError(t, guard.CheckAndRecord("digest-1", futureExpiry))
+	require.NoError(t, guard.CheckAndRecord("digest-2", futureExpiry))
+	require.NoError(t, guard.CheckAndRecord("digest-3", futureExpiry))
+
+	require.ErrorIs(t, guard.CheckAndRecord("digest-1", futureExpiry), ErrDigestAlreadySeen)
+	require.ErrorIs(t, guard.CheckAndRecord("digest-2", futureExpiry), ErrDigestAlreadySeen)
+}
+
+func TestDigestReplayGuard_ExpiredEntryCleaned(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	now := time.Now()
+	guard.nowFunc = func() time.Time { return now }
+
+	pastExpiry := now.UTC().Unix() - 10
+	err := guard.CheckAndRecord("digest-1", pastExpiry)
+	require.NoError(t, err)
+
+	// Advance time past the expiry — next call should clean up the entry
+	guard.nowFunc = func() time.Time { return now.Add(20 * time.Second) }
+
+	// Same digest should succeed because the expired entry was cleaned up
+	err = guard.CheckAndRecord("digest-1", now.Add(20*time.Second).UTC().Unix()+100)
+	require.NoError(t, err)
+}
+
+func TestDigestReplayGuard_NonExpiredEntryRetained(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	now := time.Now()
+	guard.nowFunc = func() time.Time { return now }
+
+	futureExpiry := now.UTC().Unix() + 100
+	require.NoError(t, guard.CheckAndRecord("digest-1", futureExpiry))
+
+	// Advance time, but NOT past the expiry
+	guard.nowFunc = func() time.Time { return now.Add(50 * time.Second) }
+
+	err := guard.CheckAndRecord("digest-1", futureExpiry)
+	require.ErrorIs(t, err, ErrDigestAlreadySeen)
+}
+
+func TestDigestReplayGuard_MixedExpiryCleanup(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	now := time.Now()
+	guard.nowFunc = func() time.Time { return now }
+
+	shortExpiry := now.UTC().Unix() + 10
+	longExpiry := now.UTC().Unix() + 200
+
+	require.NoError(t, guard.CheckAndRecord("short-lived", shortExpiry))
+	require.NoError(t, guard.CheckAndRecord("long-lived", longExpiry))
+
+	// Advance past short expiry but before long expiry
+	guard.nowFunc = func() time.Time { return now.Add(50 * time.Second) }
+
+	// Short-lived should be re-recordable (cleaned up)
+	require.NoError(t, guard.CheckAndRecord("short-lived", now.Add(50*time.Second).UTC().Unix()+100))
+
+	// Long-lived should still be rejected
+	require.ErrorIs(t, guard.CheckAndRecord("long-lived", longExpiry), ErrDigestAlreadySeen)
+}
+
+func TestDigestReplayGuard_ConcurrentAccess(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	const goroutines = 100
+	results := make([]error, goroutines)
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+
+	for i := range goroutines {
+		go func(idx int) {
+			defer wg.Done()
+			results[idx] = guard.CheckAndRecord("same-digest", futureExpiry)
+		}(i)
+	}
+	wg.Wait()
+
+	successCount := 0
+	duplicateCount := 0
+	for _, err := range results {
+		if err == nil {
+			successCount++
+		} else {
+			require.ErrorIs(t, err, ErrDigestAlreadySeen)
+			duplicateCount++
+		}
+	}
+
+	assert.Equal(t, 1, successCount, "exactly one goroutine should succeed")
+	assert.Equal(t, goroutines-1, duplicateCount, "all others should be rejected as duplicates")
+}
+
+func TestDigestReplayGuard_ConcurrentDifferentDigests(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	const goroutines = 50
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+
+	errors := make([]error, goroutines)
+	for i := range goroutines {
+		go func(idx int) {
+			defer wg.Done()
+			digest := "digest-" + string(rune('A'+idx))
+			errors[idx] = guard.CheckAndRecord(digest, futureExpiry)
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errors {
+		assert.NoError(t, err, "goroutine %d should succeed for unique digest", i)
+	}
+}
+
+func TestDigestReplayGuard_ClearExpiredIndependently(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	now := time.Now()
+	guard.nowFunc = func() time.Time { return now }
+
+	shortExpiry := now.UTC().Unix() + 5
+	longExpiry := now.UTC().Unix() + 200
+
+	require.NoError(t, guard.CheckAndRecord("ephemeral", shortExpiry))
+	require.NoError(t, guard.CheckAndRecord("durable", longExpiry))
+
+	// Advance past the short expiry
+	guard.nowFunc = func() time.Time { return now.Add(30 * time.Second) }
+
+	// ClearExpired should prune without needing a CheckAndRecord call
+	guard.ClearExpired()
+
+	guard.mu.Lock()
+	_, ephemeralPresent := guard.seen["ephemeral"]
+	_, durablePresent := guard.seen["durable"]
+	guard.mu.Unlock()
+
+	assert.False(t, ephemeralPresent, "expired entry should have been pruned")
+	assert.True(t, durablePresent, "non-expired entry should remain")
+}
+
+func TestDigestReplayGuard_EmptyDigest(t *testing.T) {
+	guard := NewDigestReplayGuard()
+	futureExpiry := time.Now().UTC().Unix() + 100
+
+	require.NoError(t, guard.CheckAndRecord("", futureExpiry))
+	require.ErrorIs(t, guard.CheckAndRecord("", futureExpiry), ErrDigestAlreadySeen)
+}
diff --git a/core/capabilities/vault/request_authorizer.go b/core/capabilities/vault/request_authorizer.go
@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"sync"
 	"time"
 
 	jsonrpc "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2"
@@ -19,16 +18,15 @@ type RequestAuthorizer interface {
 	AuthorizeRequest(ctx context.Context, req jsonrpc.Request[json.RawMessage]) (isAuthorized bool, owner string, err error)
 }
 type requestAuthorizer struct {
-	workflowRegistrySyncer    workflowsyncerv2.WorkflowRegistrySyncer
-	alreadyAuthorizedRequests map[string]int64
-	alreadyAuthorizedMutex    sync.Mutex
-	lggr                      logger.Logger
+	workflowRegistrySyncer workflowsyncerv2.WorkflowRegistrySyncer
+	replayGuard            *DigestReplayGuard
+	lggr                   logger.Logger
 }
 
 // AuthorizeRequest authorizes a request based on the request digest and the allowlisted requests.
 // It does NOT check if the request method is allowed.
 func (r *requestAuthorizer) AuthorizeRequest(ctx context.Context, req jsonrpc.Request[json.RawMessage]) (isAuthorized bool, owner string, err error) {
-	defer r.clearExpiredAuthorizedRequests()
+	defer r.replayGuard.ClearExpired()
 	r.lggr.Infow("AuthorizeRequest", "method", req.Method, "requestID", req.ID)
 	requestDigest, err := req.Digest()
 	if err != nil {
@@ -61,31 +59,21 @@ func (r *requestAuthorizer) AuthorizeRequest(ctx context.Context, req jsonrpc.Re
 			"allowedRequestsStrs", allowedRequestsStrs)
 		return false, "", errors.New("request not allowlisted")
 	}
-	authorizedRequestStr := string(allowlistedRequest.RequestDigest[:])
 
-	r.alreadyAuthorizedMutex.Lock()
-	defer r.alreadyAuthorizedMutex.Unlock()
-	if r.alreadyAuthorizedRequests[authorizedRequestStr] > 0 {
-		r.lggr.Infow("AuthorizeRequest already authorized previously", "method", req.Method, "requestID", req.ID, "authorizedRequestStr", authorizedRequestStr)
-		return false, "", errors.New("request already authorized previously")
-	}
 	if time.Now().UTC().Unix() > int64(allowlistedRequest.ExpiryTimestamp) {
+		authorizedRequestStr := string(allowlistedRequest.RequestDigest[:])
 		r.lggr.Infow("AuthorizeRequest expired authorization", "method", req.Method, "requestID", req.ID, "authorizedRequestStr", authorizedRequestStr)
 		return false, "", errors.New("request authorization expired")
 	}
-	r.lggr.Infow("AuthorizeRequest success in auth", "method", req.Method, "requestID", req.ID, "authorizedRequestStr", authorizedRequestStr)
-	r.alreadyAuthorizedRequests[authorizedRequestStr] = int64(allowlistedRequest.ExpiryTimestamp)
-	return true, allowlistedRequest.Owner.Hex(), nil
-}
 
-func (r *requestAuthorizer) clearExpiredAuthorizedRequests() {
-	r.alreadyAuthorizedMutex.Lock()
-	defer r.alreadyAuthorizedMutex.Unlock()
-	for request, expiry := range r.alreadyAuthorizedRequests {
-		if time.Now().UTC().Unix() > expiry {
-			delete(r.alreadyAuthorizedRequests, request)
-		}
+	digestKey := string(allowlistedRequest.RequestDigest[:])
+	if err := r.replayGuard.CheckAndRecord(digestKey, int64(allowlistedRequest.ExpiryTimestamp)); err != nil {
+		r.lggr.Infow("AuthorizeRequest already authorized previously", "method", req.Method, "requestID", req.ID, "authorizedRequestStr", digestKey)
+		return false, "", err
 	}
+
+	r.lggr.Infow("AuthorizeRequest success in auth", "method", req.Method, "requestID", req.ID, "authorizedRequestStr", digestKey)
+	return true, allowlistedRequest.Owner.Hex(), nil
 }
 
 func (r *requestAuthorizer) fetchAllowlistedItem(allowListedRequests []workflow_registry_wrapper_v2.WorkflowRegistryOwnerAllowlistedRequest, digest [32]byte) *workflow_registry_wrapper_v2.WorkflowRegistryOwnerAllowlistedRequest {
@@ -99,8 +87,8 @@ func (r *requestAuthorizer) fetchAllowlistedItem(allowListedRequests []workflow_
 
 func NewRequestAuthorizer(lggr logger.Logger, workflowRegistrySyncer workflowsyncerv2.WorkflowRegistrySyncer) *requestAuthorizer {
 	return &requestAuthorizer{
-		workflowRegistrySyncer:    workflowRegistrySyncer,
-		lggr:                      logger.Named(lggr, "VaultRequestAuthorizer"),
-		alreadyAuthorizedRequests: make(map[string]int64),
+		workflowRegistrySyncer: workflowRegistrySyncer,
+		lggr:                   logger.Named(lggr, "VaultRequestAuthorizer"),
+		replayGuard:            NewDigestReplayGuard(),
 	}
 }
