Coverage for src/lexigram/auth/authn/blacklist.py: 87%
84 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""JWT token blacklist management.
3Provides :class:`JWTBlacklist` for revoking individual tokens and all tokens
4for a user. Supports both in-process (fallback) and cache-backed storage.
6Key schema:
7- Individual token: ``jwt:blacklist:{sha256_hex}``
8- User-level block: ``jwt:blacklist:user:{user_id}``
9"""
11from __future__ import annotations
13import hashlib
14from typing import TYPE_CHECKING, Any
16import jwt
18from lexigram.logging import get_logger
19from lexigram.primitives import clock as ambient_clock
21if TYPE_CHECKING:
22 from lexigram.contracts.audit import AuditLoggerProtocol
23 from lexigram.contracts.auth.exceptions import TokenError as ContractsTokenError
24 from lexigram.contracts.infra.cache import CacheBackendProtocol
25 from lexigram.result import Result
27__all__ = ["JWTBlacklist"]
29logger = get_logger(__name__)
32class JWTBlacklist:
33 """Hash-based JWT blacklist supporting both cache and in-process storage.
35 When a :class:`~lexigram.contracts.cache.CacheBackendProtocol` is provided,
36 revocation entries are persisted with a TTL derived from the token's own
37 expiry time. Without a cache the in-process ``_blacklist`` set is used as
38 a fallback (entries are lost on process restart).
40 Args:
41 cache: Optional cache backend for distributed blacklist storage.
42 algorithm: JWT signing algorithm string (used in audit entries).
43 current_key_id: Callable that returns the active signing key ID.
44 access_expiration_hours: Access token lifetime in hours.
45 refresh_expiration_days: Refresh token lifetime in days.
46 audit_logger: Optional audit logger for revocation events.
47 """
49 def __init__(
50 self,
51 cache: CacheBackendProtocol | None,
52 algorithm: str,
53 current_key_id_fn: Any,
54 access_expiration_hours: int,
55 refresh_expiration_days: int,
56 audit_logger: AuditLoggerProtocol | None = None,
57 cache_resolver: Any = None,
58 ) -> None:
59 self._cache = cache
60 self._algorithm = algorithm
61 self._current_key_id_fn = current_key_id_fn
62 self._access_expiration_hours = access_expiration_hours
63 self._refresh_expiration_days = refresh_expiration_days
64 self._audit_logger = audit_logger
65 self._fallback: set[str] = set()
66 # Optional late-binding source for the cache when none was supplied
67 # at construction (DI ordering: auth boots before cache providers).
68 self._cache_resolver: Any = cache_resolver
70 def attach_cache(self, cache: Any) -> None:
71 """Attach an explicit cache backend immediately."""
72 """Attach a cache backend after construction.
74 Lets DI providers supply the application cache during ``boot()``
75 when it was not available at construction time, enabling
76 cache-backed revocation without rebuilding the token manager.
78 Args:
79 cache: Cache backend implementing the blacklist operations used
80 here (``exists``/``set`` semantics of CacheBackendProtocol).
81 """
82 self._cache = cache
84 def attach_cache_resolver(self, resolver: Any) -> None:
85 """Attach a deferred cache source invoked on first revocation use.
87 Ordering-proof alternative to :meth:`attach_cache`: providers that
88 boot before the cache layer can hand over a zero-arg callable
89 returning the backend (or ``None``) once it exists.
91 Args:
92 resolver: Zero-argument callable returning a cache backend or
93 ``None``.
94 """
95 self._cache_resolver = resolver
97 def _effective_cache(self) -> Any:
98 """Return the cache backend, resolving lazily when configured."""
99 if self._cache is None and self._cache_resolver is not None:
100 try:
101 self._cache = self._cache_resolver()
102 except Exception as exc: # noqa: BLE001 — fail closed below
103 logger.warning("blacklist_cache_resolve_failed", error=str(exc))
104 return None
105 return self._cache
107 # ── Public API ────────────────────────────────────────────────────────
109 async def revoke(self, token: str) -> Result[None, ContractsTokenError]:
110 """Add *token* to the blacklist.
112 Args:
113 token: The raw JWT string to revoke.
115 Returns:
116 ``Ok(None)`` on success or if the token is already expired.
117 ``Err(TokenError)`` if the cache write fails.
119 Raises:
120 RuntimeError: On cache-level infrastructure failures.
121 OSError: On network-level failures.
122 ConnectionError: On connection failures.
123 """
124 from lexigram.contracts.auth.exceptions import TokenError as ContractsTokenError
125 from lexigram.result import Err, Ok
127 token_hash = hashlib.sha256(token.encode()).hexdigest()
128 cache = self._effective_cache()
130 if not cache:
131 self._fallback.add(token_hash)
132 return Ok(None)
134 try:
135 unverified_payload = jwt.decode(token, options={"verify_signature": False})
136 exp = unverified_payload.get("exp", 0)
137 now = int(ambient_clock.timestamp())
138 ttl = max(0, exp - now)
140 if ttl <= 0:
141 return Ok(None)
143 blacklist_key = f"jwt:blacklist:{token_hash}"
144 success = bool(await cache.set(key=blacklist_key, value="1", ttl=ttl))
146 if success and self._audit_logger is not None:
147 from lexigram.contracts.audit import AuditEntry
149 actor_id = str(unverified_payload.get("sub", "unknown"))
150 await self._audit_logger.log(
151 AuditEntry(
152 action="token.revoked",
153 actor_id=actor_id,
154 resource_type="Token",
155 resource_id=token_hash[:16],
156 outcome="success",
157 metadata={
158 "algorithm": self._algorithm,
159 "key_id": self._current_key_id_fn(),
160 },
161 )
162 )
164 return (
165 Ok(None)
166 if success
167 else Err(ContractsTokenError("Token revocation failed"))
168 )
170 except (RuntimeError, OSError, ConnectionError):
171 logger.exception("token_revoke_failed")
172 raise
174 async def revoke_all_for_user(
175 self, user_id: str
176 ) -> Result[None, ContractsTokenError]:
177 """Blacklist all tokens for *user_id* by writing a user-level sentinel.
179 Args:
180 user_id: The subject (``sub``) claim of the tokens to revoke.
182 Returns:
183 ``Ok(None)`` on success. ``Err(TokenError)`` if the write fails.
185 Raises:
186 RuntimeError: If no cache backend is configured.
187 """
188 from lexigram.contracts.auth.exceptions import TokenError as ContractsTokenError
189 from lexigram.result import Err, Ok
191 cache = self._effective_cache()
192 if not cache:
193 raise RuntimeError(
194 "Cannot blacklist token: no cache backend configured. "
195 "Configure Redis via AuthConfig.cache",
196 )
198 try:
199 blacklist_key = f"jwt:blacklist:user:{user_id}"
200 max_token_ttl = max(
201 self._access_expiration_hours * 3600,
202 self._refresh_expiration_days * 86400,
203 )
204 success = bool(
205 await cache.set(
206 key=blacklist_key,
207 value=int(ambient_clock.timestamp()),
208 ttl=max_token_ttl,
209 )
210 )
211 return (
212 Ok(None)
213 if success
214 else Err(ContractsTokenError("Failed to revoke all user tokens"))
215 )
216 except (RuntimeError, OSError, ConnectionError):
217 logger.exception("token_revoke_all_failed", user_id=user_id)
218 raise
220 async def is_blacklisted(self, token: str) -> bool:
221 """Return ``True`` if *token* has been revoked.
223 Checks both the individual token hash and the user-level sentinel.
224 Fails closed on cache errors (returns ``True``).
226 Args:
227 token: The raw JWT string to check.
229 Returns:
230 ``True`` if the token should be rejected, ``False`` otherwise.
231 """
232 token_hash = hashlib.sha256(token.encode()).hexdigest()
233 cache = self._effective_cache()
235 if not cache:
236 return token_hash in self._fallback
238 try:
239 if await cache.exists(f"jwt:blacklist:{token_hash}"):
240 return True
242 unverified_payload = jwt.decode(token, options={"verify_signature": False})
243 user_id = unverified_payload.get("sub")
244 return bool(user_id and await cache.exists(f"jwt:blacklist:user:{user_id}"))
246 except (RuntimeError, OSError, ConnectionError, jwt.InvalidTokenError):
247 logger.exception("token_blacklist_check_failed")
248 return True # Fail closed