fix(tokenstore): serialize pass revocation refreshes with revoke
Concurrent IsRevoked calls after cache expiry each spawned a pass read and could publish an older snapshot over a newer one. A dedicated mutex now runs one refresh at a time, rechecks freshness after acquiring it, and orders refreshes with Revoke.
2 files · +85 -8
M internal/adapters/out/tokenstore/pass.go +29 -8
| @@ -76,6 +76,11 @@ type PassStore struct { | |||
| 76 | 76 | revokedList []string // cached revocation list | |
| 77 | 77 | revokedSet map[string]struct{} // for O(1) lookup | |
| 78 | 78 | ||
| 79 | + | // revokedMu serializes revocation list reads from pass with Revoke, so only | |
| 80 | + | // one refresh runs at a time and an older snapshot never overwrites a newer | |
| 81 | + | // one. Lock order: revokedMu before cacheMu. | |
| 82 | + | revokedMu sync.Mutex | |
| 83 | + | ||
| 79 | 84 | // revokedLoadedAt is when the revocation cache was last refreshed. | |
| 80 | 85 | // The list is re-read from pass once revokedTTL has elapsed, so revocations | |
| 81 | 86 | // made by other processes (e.g. the CLI) are picked up. | |
| @@ -384,8 +389,8 @@ func (s *PassStore) ListTokens(ctx context.Context) ([]domain.Token, error) { | |||
| 384 | 389 | ||
| 385 | 390 | // Revoke adds token ID to revocation list in pass. | |
| 386 | 391 | func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | |
| 387 | - | s.cacheMu.Lock() | |
| 388 | - | defer s.cacheMu.Unlock() | |
| 392 | + | s.revokedMu.Lock() | |
| 393 | + | defer s.revokedMu.Unlock() | |
| 389 | 394 | ||
| 390 | 395 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | |
| 391 | 396 | defer cancel() | |
| @@ -415,7 +420,9 @@ func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | |||
| 415 | 420 | } | |
| 416 | 421 | ||
| 417 | 422 | // Update cache | |
| 423 | + | s.cacheMu.Lock() | |
| 418 | 424 | s.setRevokedCacheLocked(revokedList) | |
| 425 | + | s.cacheMu.Unlock() | |
| 419 | 426 | ||
| 420 | 427 | s.log.Info(). | |
| 421 | 428 | Str(zerowrap.FieldLayer, "adapter"). | |
| @@ -426,6 +433,18 @@ func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | |||
| 426 | 433 | return nil | |
| 427 | 434 | } | |
| 428 | 435 | ||
| 436 | + | // cachedRevoked reports whether tokenID is in the cached revocation list and | |
| 437 | + | // whether that cache is still fresh. | |
| 438 | + | func (s *PassStore) cachedRevoked(tokenID string) (revoked, fresh bool) { | |
| 439 | + | s.cacheMu.RLock() | |
| 440 | + | defer s.cacheMu.RUnlock() | |
| 441 | + | if s.revokedList == nil || time.Since(s.revokedLoadedAt) >= s.revokedTTL { | |
| 442 | + | return false, false | |
| 443 | + | } | |
| 444 | + | _, revoked = s.revokedSet[tokenID] | |
| 445 | + | return revoked, true | |
| 446 | + | } | |
| 447 | + | ||
| 429 | 448 | // setRevokedCacheLocked replaces the revocation cache. Caller must hold cacheMu for writing. | |
| 430 | 449 | func (s *PassStore) setRevokedCacheLocked(revokedList []string) { | |
| 431 | 450 | s.revokedList = revokedList | |
| @@ -439,14 +458,16 @@ func (s *PassStore) setRevokedCacheLocked(revokedList []string) { | |||
| 439 | 458 | // IsRevoked checks if token ID is in revocation list. | |
| 440 | 459 | // The list is cached for revokedTTL; read errors are returned and never cached. | |
| 441 | 460 | func (s *PassStore) IsRevoked(ctx context.Context, tokenID string) (bool, error) { | |
| 442 | - | // Check cache first | |
| 443 | - | s.cacheMu.RLock() | |
| 444 | - | if s.revokedList != nil && time.Since(s.revokedLoadedAt) < s.revokedTTL { | |
| 445 | - | _, revoked := s.revokedSet[tokenID] | |
| 446 | - | s.cacheMu.RUnlock() | |
| 461 | + | if revoked, fresh := s.cachedRevoked(tokenID); fresh { | |
| 462 | + | return revoked, nil | |
| 463 | + | } | |
| 464 | + | ||
| 465 | + | // Single refresh at a time; recheck in case another caller just refreshed. | |
| 466 | + | s.revokedMu.Lock() | |
| 467 | + | defer s.revokedMu.Unlock() | |
| 468 | + | if revoked, fresh := s.cachedRevoked(tokenID); fresh { | |
| 447 | 469 | return revoked, nil | |
| 448 | 470 | } | |
| 449 | - | s.cacheMu.RUnlock() | |
| 450 | 471 | ||
| 451 | 472 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | |
| 452 | 473 | defer cancel() | |
| @@ -76,6 +76,11 @@ type PassStore struct { | |||||
| 76 | revokedList []string // cached revocation list | 76 | revokedList []string // cached revocation list | ||
| 77 | revokedSet map[string]struct{} // for O(1) lookup | 77 | revokedSet map[string]struct{} // for O(1) lookup | ||
| 78 | 78 | ||||
| 79 | + | // revokedMu serializes revocation list reads from pass with Revoke, so only | |||
| 80 | + | // one refresh runs at a time and an older snapshot never overwrites a newer | |||
| 81 | + | // one. Lock order: revokedMu before cacheMu. | |||
| 82 | + | revokedMu sync.Mutex | |||
| 83 | + | ||||
| 79 | // revokedLoadedAt is when the revocation cache was last refreshed. | 84 | // revokedLoadedAt is when the revocation cache was last refreshed. | ||
| 80 | // The list is re-read from pass once revokedTTL has elapsed, so revocations | 85 | // The list is re-read from pass once revokedTTL has elapsed, so revocations | ||
| 81 | // made by other processes (e.g. the CLI) are picked up. | 86 | // made by other processes (e.g. the CLI) are picked up. | ||
| @@ -384,8 +389,8 @@ func (s *PassStore) ListTokens(ctx context.Context) ([]domain.Token, error) { | |||||
| 384 | 389 | ||||
| 385 | // Revoke adds token ID to revocation list in pass. | 390 | // Revoke adds token ID to revocation list in pass. | ||
| 386 | func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | 391 | func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | ||
| 387 | - | s.cacheMu.Lock() | 392 | + | s.revokedMu.Lock() |
| 388 | - | defer s.cacheMu.Unlock() | 393 | + | defer s.revokedMu.Unlock() |
| 389 | 394 | ||||
| 390 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | 395 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | ||
| 391 | defer cancel() | 396 | defer cancel() | ||
| @@ -415,7 +420,9 @@ func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | |||||
| 415 | } | 420 | } | ||
| 416 | 421 | ||||
| 417 | // Update cache | 422 | // Update cache | ||
| 423 | + | s.cacheMu.Lock() | |||
| 418 | s.setRevokedCacheLocked(revokedList) | 424 | s.setRevokedCacheLocked(revokedList) | ||
| 425 | + | s.cacheMu.Unlock() | |||
| 419 | 426 | ||||
| 420 | s.log.Info(). | 427 | s.log.Info(). | ||
| 421 | Str(zerowrap.FieldLayer, "adapter"). | 428 | Str(zerowrap.FieldLayer, "adapter"). | ||
| @@ -426,6 +433,18 @@ func (s *PassStore) Revoke(ctx context.Context, tokenID string) error { | |||||
| 426 | return nil | 433 | return nil | ||
| 427 | } | 434 | } | ||
| 428 | 435 | ||||
| 436 | + | // cachedRevoked reports whether tokenID is in the cached revocation list and | |||
| 437 | + | // whether that cache is still fresh. | |||
| 438 | + | func (s *PassStore) cachedRevoked(tokenID string) (revoked, fresh bool) { | |||
| 439 | + | s.cacheMu.RLock() | |||
| 440 | + | defer s.cacheMu.RUnlock() | |||
| 441 | + | if s.revokedList == nil || time.Since(s.revokedLoadedAt) >= s.revokedTTL { | |||
| 442 | + | return false, false | |||
| 443 | + | } | |||
| 444 | + | _, revoked = s.revokedSet[tokenID] | |||
| 445 | + | return revoked, true | |||
| 446 | + | } | |||
| 447 | + | ||||
| 429 | // setRevokedCacheLocked replaces the revocation cache. Caller must hold cacheMu for writing. | 448 | // setRevokedCacheLocked replaces the revocation cache. Caller must hold cacheMu for writing. | ||
| 430 | func (s *PassStore) setRevokedCacheLocked(revokedList []string) { | 449 | func (s *PassStore) setRevokedCacheLocked(revokedList []string) { | ||
| 431 | s.revokedList = revokedList | 450 | s.revokedList = revokedList | ||
| @@ -439,14 +458,16 @@ func (s *PassStore) setRevokedCacheLocked(revokedList []string) { | |||||
| 439 | // IsRevoked checks if token ID is in revocation list. | 458 | // IsRevoked checks if token ID is in revocation list. | ||
| 440 | // The list is cached for revokedTTL; read errors are returned and never cached. | 459 | // The list is cached for revokedTTL; read errors are returned and never cached. | ||
| 441 | func (s *PassStore) IsRevoked(ctx context.Context, tokenID string) (bool, error) { | 460 | func (s *PassStore) IsRevoked(ctx context.Context, tokenID string) (bool, error) { | ||
| 442 | - | // Check cache first | 461 | + | if revoked, fresh := s.cachedRevoked(tokenID); fresh { |
| 443 | - | s.cacheMu.RLock() | 462 | + | return revoked, nil |
| 444 | - | if s.revokedList != nil && time.Since(s.revokedLoadedAt) < s.revokedTTL { | 463 | + | } |
| 445 | - | _, revoked := s.revokedSet[tokenID] | 464 | + | |
| 446 | - | s.cacheMu.RUnlock() | 465 | + | // Single refresh at a time; recheck in case another caller just refreshed. |
| 466 | + | s.revokedMu.Lock() | |||
| 467 | + | defer s.revokedMu.Unlock() | |||
| 468 | + | if revoked, fresh := s.cachedRevoked(tokenID); fresh { | |||
| 447 | return revoked, nil | 469 | return revoked, nil | ||
| 448 | } | 470 | } | ||
| 449 | - | s.cacheMu.RUnlock() | |||
| 450 | 471 | ||||
| 451 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | 472 | ctx, cancel := context.WithTimeout(ctx, s.timeout) | ||
| 452 | defer cancel() | 473 | defer cancel() | ||
M internal/adapters/out/tokenstore/pass_test.go +56 -0
| @@ -440,6 +440,7 @@ cmd="$1"; shift | |||
| 440 | 440 | case "$cmd" in | |
| 441 | 441 | show) | |
| 442 | 442 | path="$1" | |
| 443 | + | echo x >> "$DATA/.show_calls" | |
| 443 | 444 | if [ -e "$DATA/FAIL" ]; then | |
| 444 | 445 | echo "gpg: decryption failed: No secret key" >&2 | |
| 445 | 446 | exit 2 | |
| @@ -519,6 +520,61 @@ func TestPassStore_IsRevoked_ReadFailureNotCached(t *testing.T) { | |||
| 519 | 520 | assert.True(t, revoked) | |
| 520 | 521 | } | |
| 521 | 522 | ||
| 523 | + | // expireRevokedCache marks the revocation cache as stale. | |
| 524 | + | func expireRevokedCache(store *PassStore) { | |
| 525 | + | store.cacheMu.Lock() | |
| 526 | + | store.revokedLoadedAt = time.Time{} | |
| 527 | + | store.cacheMu.Unlock() | |
| 528 | + | } | |
| 529 | + | ||
| 530 | + | func showCalls(t *testing.T, dataDir string) int { | |
| 531 | + | t.Helper() | |
| 532 | + | data, err := os.ReadFile(filepath.Join(dataDir, ".show_calls")) | |
| 533 | + | require.NoError(t, err) | |
| 534 | + | return len(data) / len("x\n") | |
| 535 | + | } | |
| 536 | + | ||
| 537 | + | func TestPassStore_IsRevoked_ConcurrentExpiryRefreshesOnce(t *testing.T) { | |
| 538 | + | dataDir := setupFakePass(t) | |
| 539 | + | ctx := context.Background() | |
| 540 | + | require.NoError(t, os.WriteFile(revokedFile(dataDir), []byte(`["tok-1"]`), 0o600)) | |
| 541 | + | ||
| 542 | + | store := newTestPassStore() | |
| 543 | + | _, err := store.IsRevoked(ctx, "tok-1") | |
| 544 | + | require.NoError(t, err) | |
| 545 | + | expireRevokedCache(store) | |
| 546 | + | ||
| 547 | + | var wg sync.WaitGroup | |
| 548 | + | for i := 0; i < 20; i++ { | |
| 549 | + | wg.Add(1) | |
| 550 | + | go func() { | |
| 551 | + | defer wg.Done() | |
| 552 | + | revoked, err := store.IsRevoked(ctx, "tok-1") | |
| 553 | + | assert.NoError(t, err) | |
| 554 | + | assert.True(t, revoked) | |
| 555 | + | }() | |
| 556 | + | } | |
| 557 | + | wg.Wait() | |
| 558 | + | ||
| 559 | + | assert.Equal(t, 2, showCalls(t, dataDir), "one initial load plus a single refresh") | |
| 560 | + | } | |
| 561 | + | ||
| 562 | + | func TestPassStore_IsRevoked_ReadFailureAfterExpiryFailsClosed(t *testing.T) { | |
| 563 | + | dataDir := setupFakePass(t) | |
| 564 | + | ctx := context.Background() | |
| 565 | + | require.NoError(t, os.WriteFile(revokedFile(dataDir), []byte(`["tok-1"]`), 0o600)) | |
| 566 | + | ||
| 567 | + | store := newTestPassStore() | |
| 568 | + | _, err := store.IsRevoked(ctx, "tok-1") | |
| 569 | + | require.NoError(t, err) | |
| 570 | + | ||
| 571 | + | require.NoError(t, os.WriteFile(filepath.Join(dataDir, "FAIL"), nil, 0o600)) | |
| 572 | + | expireRevokedCache(store) | |
| 573 | + | ||
| 574 | + | _, err = store.IsRevoked(ctx, "tok-2") | |
| 575 | + | require.Error(t, err) | |
| 576 | + | } | |
| 577 | + | ||
| 522 | 578 | func TestPassStore_Revoke_ReadFailureKeepsStoredList(t *testing.T) { | |
| 523 | 579 | dataDir := setupFakePass(t) | |
| 524 | 580 | original := `["tok-old"]` | |
| @@ -440,6 +440,7 @@ cmd="$1"; shift | |||||
| 440 | case "$cmd" in | 440 | case "$cmd" in | ||
| 441 | show) | 441 | show) | ||
| 442 | path="$1" | 442 | path="$1" | ||
| 443 | + | echo x >> "$DATA/.show_calls" | |||
| 443 | if [ -e "$DATA/FAIL" ]; then | 444 | if [ -e "$DATA/FAIL" ]; then | ||
| 444 | echo "gpg: decryption failed: No secret key" >&2 | 445 | echo "gpg: decryption failed: No secret key" >&2 | ||
| 445 | exit 2 | 446 | exit 2 | ||
| @@ -519,6 +520,61 @@ func TestPassStore_IsRevoked_ReadFailureNotCached(t *testing.T) { | |||||
| 519 | assert.True(t, revoked) | 520 | assert.True(t, revoked) | ||
| 520 | } | 521 | } | ||
| 521 | 522 | ||||
| 523 | + | // expireRevokedCache marks the revocation cache as stale. | |||
| 524 | + | func expireRevokedCache(store *PassStore) { | |||
| 525 | + | store.cacheMu.Lock() | |||
| 526 | + | store.revokedLoadedAt = time.Time{} | |||
| 527 | + | store.cacheMu.Unlock() | |||
| 528 | + | } | |||
| 529 | + | ||||
| 530 | + | func showCalls(t *testing.T, dataDir string) int { | |||
| 531 | + | t.Helper() | |||
| 532 | + | data, err := os.ReadFile(filepath.Join(dataDir, ".show_calls")) | |||
| 533 | + | require.NoError(t, err) | |||
| 534 | + | return len(data) / len("x\n") | |||
| 535 | + | } | |||
| 536 | + | ||||
| 537 | + | func TestPassStore_IsRevoked_ConcurrentExpiryRefreshesOnce(t *testing.T) { | |||
| 538 | + | dataDir := setupFakePass(t) | |||
| 539 | + | ctx := context.Background() | |||
| 540 | + | require.NoError(t, os.WriteFile(revokedFile(dataDir), []byte(`["tok-1"]`), 0o600)) | |||
| 541 | + | ||||
| 542 | + | store := newTestPassStore() | |||
| 543 | + | _, err := store.IsRevoked(ctx, "tok-1") | |||
| 544 | + | require.NoError(t, err) | |||
| 545 | + | expireRevokedCache(store) | |||
| 546 | + | ||||
| 547 | + | var wg sync.WaitGroup | |||
| 548 | + | for i := 0; i < 20; i++ { | |||
| 549 | + | wg.Add(1) | |||
| 550 | + | go func() { | |||
| 551 | + | defer wg.Done() | |||
| 552 | + | revoked, err := store.IsRevoked(ctx, "tok-1") | |||
| 553 | + | assert.NoError(t, err) | |||
| 554 | + | assert.True(t, revoked) | |||
| 555 | + | }() | |||
| 556 | + | } | |||
| 557 | + | wg.Wait() | |||
| 558 | + | ||||
| 559 | + | assert.Equal(t, 2, showCalls(t, dataDir), "one initial load plus a single refresh") | |||
| 560 | + | } | |||
| 561 | + | ||||
| 562 | + | func TestPassStore_IsRevoked_ReadFailureAfterExpiryFailsClosed(t *testing.T) { | |||
| 563 | + | dataDir := setupFakePass(t) | |||
| 564 | + | ctx := context.Background() | |||
| 565 | + | require.NoError(t, os.WriteFile(revokedFile(dataDir), []byte(`["tok-1"]`), 0o600)) | |||
| 566 | + | ||||
| 567 | + | store := newTestPassStore() | |||
| 568 | + | _, err := store.IsRevoked(ctx, "tok-1") | |||
| 569 | + | require.NoError(t, err) | |||
| 570 | + | ||||
| 571 | + | require.NoError(t, os.WriteFile(filepath.Join(dataDir, "FAIL"), nil, 0o600)) | |||
| 572 | + | expireRevokedCache(store) | |||
| 573 | + | ||||
| 574 | + | _, err = store.IsRevoked(ctx, "tok-2") | |||
| 575 | + | require.Error(t, err) | |||
| 576 | + | } | |||
| 577 | + | ||||
| 522 | func TestPassStore_Revoke_ReadFailureKeepsStoredList(t *testing.T) { | 578 | func TestPassStore_Revoke_ReadFailureKeepsStoredList(t *testing.T) { | ||
| 523 | dataDir := setupFakePass(t) | 579 | dataDir := setupFakePass(t) | ||
| 524 | original := `["tok-old"]` | 580 | original := `["tok-old"]` | ||