Coverage for src/lexigram/web/security/csrf/middleware.py: 16%
187 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""ASGI middleware for CSRF protection with dual-mode support.
3Supports two patterns:
41. Double-Submit Cookie (stateless, default) — the cookie carries an
5 HMAC-signed, expiring token (``base64url("{iss}:{ts}:{nonce}") +
6 "." + base64url(hmac)``) that must be echoed in the header.
72. Synchronizer Token (stateful, requires CacheBackendProtocol) — validates
8 against a server-side token stored in cache.
10Also issues (or rotates when stale) the CSRF cookie on safe methods
11(GET/HEAD/OPTIONS). Paths in ``CSRFConfig.excluded_paths`` are skipped for
12cookie-less requests; cookie-bearing requests on those paths are still
13validated.
15Fail-closed behavior: when ``CSRFConfig.secret_key`` is ``None`` in cookie
16(double-submit) mode, unsafe requests are always rejected — the token cannot
17be verified — while safe-method issuance still works (development UX).
18Configure ``LEX_WEB__SECURITY__CSRF__SECRET_KEY`` (required in production).
19"""
21from __future__ import annotations
23import base64
24import hashlib
25import hmac
26import secrets
27import time
28from typing import TYPE_CHECKING, Any, cast
30from lexigram.logging import get_logger
31from lexigram.validation import SecretStr
32from lexigram.web.security.config import CSRFConfig
34if TYPE_CHECKING:
35 from collections.abc import Awaitable, Callable
37 from lexigram.contracts.infra.cache import CacheBackendProtocol
39logger = get_logger(__name__)
41_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
43#: Issuer label embedded in every token (fixed per process/secret).
44_TOKEN_ISSUER = "web" # noqa: S105 # issuer label constant, not a credential
47def _b64decode(data: str) -> bytes:
48 """Decode unpadded URL-safe base64 (tokens are minted without padding)."""
49 return base64.urlsafe_b64decode(data.encode() + b"=" * (-len(data) % 4))
52class CSRFProtectionMiddleware:
53 """ASGI middleware for CSRF protection.
55 Supports both double-submit cookie and synchronizer token patterns.
56 If a ``cache`` is provided, the synchronizer token pattern is used (more
57 secure). Otherwise, the double-submit cookie pattern is used (stateless)
58 with HMAC-signed, expiring tokens.
60 Cookie issuance happens automatically on safe methods (GET/HEAD/OPTIONS)
61 so that clients receive a token before submitting state-changing requests.
62 Stale cookies (older than ``token_ttl``) are rotated on safe methods.
63 Paths listed in ``CSRFConfig.excluded_paths`` are skipped entirely for
64 cookie-less requests; cookie-bearing requests on those paths are still
65 validated, so cookie-authenticated form posts cannot bypass CSRF.
67 Example::
69 app = ASGIApp()
70 config = CSRFConfig(enabled=True, cookie_name="csrf_token",
71 secret_key="secret")
72 csrf = CSRFProtectionMiddleware(app, config=config)
74 # With cache for synchronizer pattern:
75 csrf = CSRFProtectionMiddleware(app, config=config, cache=cache_backend)
76 """
78 def __init__(
79 self,
80 app: Callable[..., Awaitable[None]],
81 config: CSRFConfig | None = None,
82 cache: CacheBackendProtocol | None = None,
83 ) -> None:
84 """Initialize CSRF middleware.
86 Args:
87 app: The ASGI application to wrap.
88 config: CSRF configuration. Uses defaults if not provided.
89 cache: Optional cache backend for synchronizer token pattern.
90 """
91 self._app = app
92 self._config = config or CSRFConfig()
93 self._cache = cache
94 self._exclude_content_types = [
95 ct.lower() for ct in self._config.exclude_content_types
96 ]
97 self._exclude_auth_schemes = [
98 s.lower() for s in self._config.exclude_auth_schemes
99 ]
101 # ------------------------------------------------------------------
102 # Token encoding / signing
103 # ------------------------------------------------------------------
105 def _build_signed_token(self, timestamp: int) -> str | None:
106 """Build a signed, expiring CSRF token.
108 Returns:
109 ``base64url("{iss}:{ts}:{nonce}") + "." + base64url(hmac)``,
110 or ``None`` when no ``secret_key`` is configured (verification
111 would be impossible).
112 """
113 raw_secret: SecretStr | str | None = self._config.secret_key
114 secret: str | None = None
115 if isinstance(raw_secret, SecretStr):
116 secret = raw_secret.get_secret_value()
117 elif raw_secret is not None:
118 secret = raw_secret
119 if not secret:
120 return None
121 nonce = secrets.token_hex(16)
122 payload = f"{_TOKEN_ISSUER}:{timestamp}:{nonce}"
123 encoded_payload = base64.urlsafe_b64encode(payload.encode()).rstrip(b"=")
124 signature = hmac.new(secret.encode(), encoded_payload, hashlib.sha256).digest()
125 encoded_sig = base64.urlsafe_b64encode(signature).rstrip(b"=")
126 return f"{encoded_payload.decode()}.{encoded_sig.decode()}"
128 def _parse_token(self, token: str) -> tuple[int, bytes] | None:
129 """Split a token into (timestamp, encoded payload) or ``None``.
131 Returns ``None`` for malformed tokens so callers can fail closed.
132 """
133 try:
134 encoded_payload, _ = token.split(".", 1)
135 payload = _b64decode(encoded_payload).decode()
136 issuer, ts_part, _nonce = payload.split(":", 2)
137 if issuer != _TOKEN_ISSUER:
138 return None
139 return int(ts_part), encoded_payload.encode()
140 except (ValueError, TypeError, UnicodeDecodeError):
141 return None
143 def _is_stale(self, token: str, now: int) -> bool:
144 """Return True when the token is missing, unparseable, or expired."""
145 parsed = self._parse_token(token)
146 if parsed is None:
147 return True
148 return now - parsed[0] > self._config.token_ttl
150 def _expected_signature(self, encoded_payload: bytes) -> bytes:
151 """Recompute the HMAC-SHA256 signature over the encoded payload."""
152 raw_sig = self._config.secret_key
153 secret: str = (
154 raw_sig.get_secret_value()
155 if isinstance(raw_sig, SecretStr)
156 else (raw_sig or "")
157 )
158 assert secret is not None # noqa: S101 # validated at boot
159 return hmac.new(secret.encode(), encoded_payload, hashlib.sha256).digest()
161 # ------------------------------------------------------------------
162 # ASGI plumbing
163 # ------------------------------------------------------------------
165 def _cache_key(self, session_id: str) -> str:
166 return f"csrf:sync:{session_id}"
168 def _is_excluded(self, path: str) -> bool:
169 return any(path.startswith(p) for p in self._config.excluded_paths)
171 def _parse_cookies(self, scope: dict[str, Any]) -> dict[str, str]:
172 cookies: dict[str, str] = {}
173 for name, value in scope.get("headers", []):
174 if name.lower() == b"cookie":
175 for part in value.decode().split(";"):
176 part = part.strip()
177 if "=" in part:
178 k, v = part.split("=", 1)
179 cookies[k.strip()] = v.strip()
180 return cookies
182 def _get_header(self, scope: dict[str, Any], header_name: str) -> str | None:
183 target = header_name.lower().encode()
184 for name, value in scope.get("headers", []):
185 if name.lower() == target:
186 return cast("str", value.decode())
187 return None
189 async def __call__(
190 self,
191 scope: dict[str, Any],
192 receive: Callable[[], Awaitable[dict[str, Any]]],
193 send: Callable[[dict[str, Any]], Awaitable[None]],
194 ) -> None:
195 """Process the request through CSRF validation.
197 Args:
198 scope: The ASGI scope dictionary.
199 receive: The ASGI receive callable.
200 send: The ASGI send callable.
201 """
202 if scope.get("type") != "http":
203 await self._app(scope, receive, send)
204 return
206 path = scope.get("path", "")
207 method = scope.get("method", "").upper()
209 if self._is_excluded(path):
210 if method in _SAFE_METHODS or not self._parse_cookies(scope):
211 await self._app(scope, receive, send)
212 return
214 if method in _SAFE_METHODS:
215 await self._handle_safe_method(scope, receive, send)
216 else:
217 await self._handle_unsafe_method(scope, receive, send)
219 async def _handle_safe_method(
220 self,
221 scope: dict[str, Any],
222 receive: Callable[[], Awaitable[dict[str, Any]]],
223 send: Callable[[dict[str, Any]], Awaitable[None]],
224 ) -> None:
225 """Issue or rotate a CSRF token on safe methods."""
226 cookies = self._parse_cookies(scope)
227 token_source = cookies.get(self._config.cookie_name)
228 now = int(time.time())
229 synchronizer = self._cache is not None
231 issue_new = token_source is None
232 if (
233 not synchronizer
234 and token_source is not None
235 and self._is_stale(token_source, now)
236 ):
237 issue_new = True
239 if issue_new:
240 if synchronizer:
241 token_source = secrets.token_urlsafe(32)
242 else:
243 token_source = self._build_signed_token(now) or secrets.token_urlsafe(
244 32
245 )
247 token = token_source
248 if self._cache:
249 cache_key = self._cache_key(cast("str", token_source))
250 result = await self._cache.get(cache_key)
251 cached = result.unwrap_or(None)
252 if cached is not None:
253 token = cached
254 if token is None:
255 token = secrets.token_urlsafe(32)
256 await self._cache.set(cache_key, token, ttl=self._config.token_ttl)
258 async def send_with_cookie(message: dict[str, Any]) -> None:
259 if message["type"] == "http.response.start":
260 headers = list(message.get("headers", []))
261 if issue_new:
262 cookie_parts = [
263 f"{self._config.cookie_name}={token_source}",
264 f"Path={self._config.cookie_path}",
265 f"SameSite={self._config.cookie_samesite.capitalize()}",
266 ]
267 if not synchronizer:
268 cookie_parts.append(f"Max-Age={self._config.token_ttl}")
269 if self._config.cookie_domain:
270 cookie_parts.append(f"Domain={self._config.cookie_domain}")
271 if self._config.cookie_secure:
272 cookie_parts.append("Secure")
273 if self._config.cookie_httponly:
274 cookie_parts.append("HttpOnly")
275 headers.append((b"set-cookie", "; ".join(cookie_parts).encode()))
276 # Expose the token in the response header so SPAs can read it
277 headers.append(
278 (
279 self._config.header_name.lower().encode(),
280 cast("str", token).encode(),
281 )
282 )
283 message = {**message, "headers": headers}
284 await send(message)
286 await self._app(scope, receive, send_with_cookie)
288 async def _handle_unsafe_method(
289 self,
290 scope: dict[str, Any],
291 receive: Callable[[], Awaitable[dict[str, Any]]],
292 send: Callable[[dict[str, Any]], Awaitable[None]],
293 ) -> None:
294 """Validate CSRF token on unsafe methods."""
295 # Programmatic API clients (JSON) — explicit opt-in only
296 content_type = (
297 (self._get_header(scope, "content-type") or "")
298 .split(";")[0]
299 .strip()
300 .lower()
301 )
302 if content_type and content_type in self._exclude_content_types:
303 await self._app(scope, receive, send)
304 return
306 # Token-authenticated clients don't need CSRF protection — explicit opt-in
307 auth_header = self._get_header(scope, "authorization") or ""
308 if auth_header:
309 scheme = auth_header.split(" ", 1)[0].lower()
310 if scheme in self._exclude_auth_schemes:
311 await self._app(scope, receive, send)
312 return
314 cookies = self._parse_cookies(scope)
315 token_source = cookies.get(self._config.cookie_name)
316 header_token = self._get_header(scope, self._config.header_name)
318 if not token_source or not header_token:
319 await self._reject(scope, receive, send, "missing_token")
320 return
322 if self._cache:
323 # Synchronizer mode — server-side comparison via the cache.
324 stored_result = await self._cache.get(self._cache_key(token_source))
325 stored = stored_result.unwrap_or(None)
326 if not stored or not hmac.compare_digest(stored, header_token):
327 await self._reject(scope, receive, send, "token_mismatch")
328 return
329 await self._app(scope, receive, send)
330 return
332 # Cookie (double-submit) mode — token must be signed, fresh, and
333 # echoed verbatim.
334 raw_verify = self._config.secret_key
335 verify_secret: str | None = (
336 raw_verify.get_secret_value()
337 if isinstance(raw_verify, SecretStr)
338 else raw_verify
339 )
340 if not verify_secret:
341 # Without a signing secret the token cannot be verified — fail closed.
342 await self._reject(scope, receive, send, "csrf_unverifiable")
343 return
345 if not hmac.compare_digest(token_source, header_token):
346 await self._reject(scope, receive, send, "token_mismatch")
347 return
349 parsed = self._parse_token(token_source)
350 if parsed is None:
351 await self._reject(scope, receive, send, "token_invalid")
352 return
354 ts, encoded_payload = parsed
355 if int(time.time()) - ts > self._config.token_ttl:
356 await self._reject(scope, receive, send, "token_expired")
357 return
359 try:
360 _, provided_sig = token_source.rsplit(".", 1)
361 provided = _b64decode(provided_sig)
362 except (ValueError, TypeError):
363 await self._reject(scope, receive, send, "token_invalid")
364 return
366 if not hmac.compare_digest(provided, self._expected_signature(encoded_payload)):
367 await self._reject(scope, receive, send, "token_invalid")
368 return
370 await self._app(scope, receive, send)
372 async def _reject(
373 self,
374 scope: dict[str, Any],
375 receive: Callable[[], Awaitable[dict[str, Any]]],
376 send: Callable[[dict[str, Any]], Awaitable[None]],
377 reason: str,
378 ) -> None:
379 logger.warning("security.csrf_violation", reason=reason)
380 body = (
381 f'{{"success": false, "error": {{"type": "csrf_error",'
382 f' "message": "CSRF validation failed: {reason.replace("_", " ")}"}}}}'
383 ).encode()
384 await send(
385 {
386 "type": "http.response.start",
387 "status": 403,
388 "headers": [(b"content-type", b"application/json")],
389 }
390 )
391 await send({"type": "http.response.body", "body": body})
394__all__ = ["CSRFProtectionMiddleware"]