Coverage for src/lexigram/auth/authn/revocation.py: 0%

1"""Persistent token revocation store backed by a CacheBackendProtocol. 

2 

3Revocation entries are stored in the cache with a configurable TTL so that 

4revocations survive process restarts and are visible across all nodes in a 

5distributed deployment. 

6 

7Key format: ``lexigram:auth:revoked:{token_id}`` 

8 

9Example:: 

10 

11 from lexigram.auth.authn.revocation import PersistentTokenRevocationStore 

12 

13 store = PersistentTokenRevocationStore(cache=my_cache_backend, ttl=3600) 

14 await store.revoke("tok_abc123", expires_at=token.expires_at) 

15 revoked = await store.is_revoked("tok_abc123") 

16""" 

17 

18from __future__ import annotations 

19 

20from datetime import UTC, datetime 

21from typing import TYPE_CHECKING 

22 

23from lexigram.logging import get_logger 

24from lexigram.primitives import clock as ambient_clock 

25 

26if TYPE_CHECKING: 

27 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol 

28 

29__all__ = ["PersistentTokenRevocationStore"] 

30 

31logger = get_logger(__name__) 

32 

33_KEY_PREFIX = "lexigram:auth:revoked:" 

34_DEFAULT_TTL = 86400 # 24 hours 

35 

36 

37class PersistentTokenRevocationStore: 

38 """Persistent token revocation list backed by a :class:`CacheBackendProtocol`. 

39 

40 Stores revoked token IDs in the cache so that revocations survive process 

41 restarts and are consistent across all nodes in a distributed deployment. 

42 

43 A sentinel value (``"1"``) is written under 

44 ``lexigram:auth:revoked:{token_id}`` with the configured TTL. When 

45 ``expires_at`` is provided to :meth:`revoke`, the TTL is capped to the 

46 token's remaining lifetime so stale entries are not kept longer than needed. 

47 

48 Args: 

49 cache: Cache backend used to persist revoked token identifiers. 

50 ttl: Default expiry in seconds for revocation records. 

51 Defaults to ``86400`` (24 h). 

52 """ 

53 

54 def __init__( 

55 self, 

56 cache: CacheBackendProtocol, 

57 ttl: int = _DEFAULT_TTL, 

58 ) -> None: 

59 self._cache = cache 

60 self._ttl = ttl 

61 

62 # ------------------------------------------------------------------ 

63 # Internal helpers 

64 # ------------------------------------------------------------------ 

65 

66 def _make_key(self, token_id: str) -> str: 

67 """Return the cache key for a token ID.""" 

68 return f"{_KEY_PREFIX}{token_id}" 

69 

70 # ------------------------------------------------------------------ 

71 # Public API 

72 # ------------------------------------------------------------------ 

73 

74 async def revoke( 

75 self, 

76 token_id: str, 

77 expires_at: datetime | None = None, 

78 ) -> None: 

79 """Mark *token_id* as revoked. 

80 

81 If the token is already expired (``expires_at`` is in the past) the 

82 call is a no-op — there is no point persisting an entry for a token 

83 that the signature validation layer will reject on its own. 

84 

85 Args: 

86 token_id: Unique identifier of the JWT/opaque token to revoke. 

87 expires_at: Optional token expiry time. Used to compute a shorter 

88 TTL so the cache entry is not held longer than the 

89 token's own lifetime. 

90 """ 

91 ttl = self._ttl 

92 

93 if expires_at is not None: 

94 # Normalise to UTC-aware datetime for safe arithmetic. 

95 if expires_at.tzinfo is None: 

96 expires_at = expires_at.replace(tzinfo=UTC) 

97 now = ambient_clock.now() 

98 remaining = int((expires_at - now).total_seconds()) 

99 if remaining <= 0: 

100 logger.debug( 

101 "token_revocation_skipped_already_expired", 

102 token_id=token_id, 

103 ) 

104 return 

105 ttl = min(ttl, remaining) 

106 

107 key = self._make_key(token_id) 

108 await self._cache.set(key, "1", ttl=ttl) 

109 logger.info("token_revoked", token_id=token_id, ttl=ttl) 

110 

111 async def is_revoked(self, token_id: str) -> bool: 

112 """Return ``True`` if *token_id* is present in the revocation list. 

113 

114 Args: 

115 token_id: The token identifier to check. 

116 

117 Returns: 

118 ``True`` when the token has been explicitly revoked and the 

119 revocation entry has not yet expired, ``False`` otherwise. 

120 """ 

121 value = await self._cache.get(self._make_key(token_id)) 

122 return value is not None