Coverage for src/lexigram/auth/web/middleware/token_cache.py: 66%

1"""Token caching utilities for authentication middleware.""" 

2 

3from __future__ import annotations 

4 

5from collections import OrderedDict 

6import hashlib 

7import threading 

8from typing import TYPE_CHECKING, Any 

9 

10from lexigram.primitives import clock as ambient_clock 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.infra.cache import CacheBackendProtocol 

14 

15 

16class TokenCache: 

17 """Cache for JWT tokens with LRU eviction and TTL support. 

18 

19 Implements: 

20 - TTL (time-to-live) for entries 

21 - LRU (least recently used) eviction when max_size is reached 

22 - Automatic cleanup of expired entries 

23 - Optional backend cache integration via CacheBackendProtocol protocol 

24 

25 When no CacheBackendProtocol is provided, this falls back to an in-memory LRU 

26 cache (L1 layer using :class:`collections.OrderedDict`). For production 

27 deployments—especially multi-worker or multi-host setups—inject a 

28 distributed backend. The recommended options are: 

29 

30 * ``lexigram.cache.backends.memory.MemoryCacheBackend`` (from 

31 *lexigram-cache*) for a process-level shared cache with richer 

32 configuration. 

33 * A Redis-backed ``CacheBackendProtocol`` implementation for cross-process token 

34 invalidation. 

35 

36 Note: ``lexigram-auth`` does not import from ``lexigram-cache`` directly 

37 (that would violate the extension-package boundary); pass the backend 

38 via constructor injection instead. 

39 """ 

40 

41 def __init__( 

42 self, 

43 max_size: int = 10_000, 

44 ttl_seconds: float = 300.0, 

45 cache_backend: CacheBackendProtocol | None = None, 

46 ): 

47 """Initialize token cache. 

48 

49 Args: 

50 max_size: Maximum number of tokens to cache in L1 (local) layer. 

51 ttl_seconds: Time-to-live for cached entries in seconds. 

52 cache_backend: Optional L2 cache backend for distributed caching. 

53 """ 

54 self._lock = threading.Lock() 

55 self._cache: OrderedDict[str, dict[str, Any]] = OrderedDict() 

56 self._max_size = max_size 

57 self._ttl = ttl_seconds 

58 self._backend = cache_backend 

59 self._hits = 0 

60 self._misses = 0 

61 

62 def _cleanup_expired(self) -> None: 

63 """Remove expired entries from cache.""" 

64 now = ambient_clock.monotonic() 

65 expired_keys = [k for k, v in self._cache.items() if v["expires_at"] <= now] 

66 for k in expired_keys: 

67 del self._cache[k] 

68 

69 def _evict_if_needed(self) -> None: 

70 """Evict oldest entry if cache is full.""" 

71 while len(self._cache) >= self._max_size: 

72 self._cache.popitem(last=False) 

73 

74 async def get(self, token: str) -> Any | None: 

75 """Get cached user for token, or None if not cached/expired. 

76 

77 Checks L1 (local) first, then L2 (backend) if available. 

78 """ 

79 now = ambient_clock.monotonic() 

80 token_hash = hashlib.sha256(token.encode()).hexdigest() 

81 

82 with self._lock: 

83 cached = self._cache.get(token_hash) 

84 

85 if cached is not None and cached["expires_at"] > now: 

86 self._hits += 1 

87 self._cache.move_to_end(token_hash) 

88 return cached["user"] 

89 

90 if cached is not None: 

91 del self._cache[token_hash] 

92 

93 if self._backend is not None: 

94 cached = None 

95 get_result = await self._backend.get(f"token:{token_hash}") 

96 if get_result.is_ok(): 

97 cached = get_result.unwrap() 

98 if cached is not None: 

99 self._hits += 1 

100 await self.set(token, cached) 

101 return cached 

102 

103 self._misses += 1 

104 return None 

105 

106 async def set(self, token: str, user: Any) -> None: 

107 """Cache user for token. 

108 

109 Stores in both L1 (local) and L2 (backend) if available. 

110 """ 

111 token_hash = hashlib.sha256(token.encode()).hexdigest() 

112 expires_at = ambient_clock.monotonic() + self._ttl 

113 

114 with self._lock: 

115 self._cleanup_expired() 

116 self._evict_if_needed() 

117 

118 self._cache[token_hash] = { 

119 "user": user, 

120 "expires_at": expires_at, 

121 } 

122 self._cache.move_to_end(token_hash) 

123 

124 if self._backend is not None: 

125 await self._backend.set(f"token:{token_hash}", user, ttl=int(self._ttl)) 

126 

127 async def invalidate(self, token: str) -> None: 

128 """Invalidate a specific token from cache.""" 

129 token_hash = hashlib.sha256(token.encode()).hexdigest() 

130 

131 with self._lock: 

132 self._cache.pop(token_hash, None) 

133 

134 if self._backend is not None: 

135 await self._backend.delete(f"token:{token_hash}") 

136 

137 def clear(self) -> None: 

138 """Clear all cached tokens.""" 

139 with self._lock: 

140 self._cache.clear() 

141 self._hits = 0 

142 self._misses = 0 

143 

144 if self._backend is not None: 

145 pass 

146 

147 def stats(self) -> dict[str, Any]: 

148 """Get cache statistics. 

149 

150 Returns: 

151 Dict with hits, misses, size, max_size, and hit_rate. 

152 """ 

153 with self._lock: 

154 total = self._hits + self._misses 

155 hit_rate = (self._hits / total * 100) if total > 0 else 0.0 

156 return { 

157 "hits": self._hits, 

158 "misses": self._misses, 

159 "size": len(self._cache), 

160 "max_size": self._max_size, 

161 "hit_rate_percent": round(hit_rate, 2), 

162 } 

163 

164 

165__all__ = ["TokenCache"]