Coverage for src/lexigram/admin/middleware/csrf.py: 0%

1"""Admin CSRF protection middleware.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from starlette.requests import Request as StarletteRequest 

8from starlette.types import ASGIApp, Receive, Scope, Send 

9 

10from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol 

11from lexigram.admin.auth.types import AdminSecurityEventType 

12from lexigram.logging import get_logger 

13 

14logger = get_logger(__name__) 

15 

16# Paths that bypass CSRF validation (relative to admin mount point) 

17_CSRF_BYPASS_PATHS: frozenset[str] = frozenset( 

18 { 

19 "/login", 

20 "/setup", 

21 "/health", 

22 } 

23) 

24 

25# Methods that require CSRF validation 

26_CSRF_METHODS: frozenset[str] = frozenset({"POST", "PUT", "PATCH", "DELETE"}) 

27 

28 

29class AdminCsrfMiddleware: 

30 """CSRF protection middleware for admin panel routes. 

31 

32 Validates CSRF tokens on all state-mutating requests (POST/PUT/PATCH/DELETE). 

33 Tokens are session-scoped HMAC-SHA256 values generated by AdminCsrfService. 

34 

35 Bypasses validation for: 

36 - GET, HEAD, OPTIONS requests (safe methods) 

37 - Login and setup pages (pre-session forms) 

38 - Static file paths 

39 - Health check endpoint 

40 """ 

41 

42 def __init__( 

43 self, 

44 app: ASGIApp, 

45 csrf_service: AdminCsrfServiceProtocol, 

46 audit_service: Any = None, 

47 ) -> None: 

48 """Initialize with ASGI app and CSRF service. 

49 

50 Args: 

51 app: The next ASGI application. 

52 csrf_service: CSRF token validation service. 

53 audit_service: Optional audit service for CSRF violation events. 

54 """ 

55 self._app = app 

56 self._csrf_service = csrf_service 

57 self._audit_service = audit_service 

58 

59 async def _audit_violation(self, scope: Scope, reason: str) -> None: 

60 """Record a CSRF violation, best-effort.""" 

61 if not self._audit_service: 

62 return 

63 try: 

64 client = scope.get("client") 

65 await self._audit_service.log_event( 

66 event_type=AdminSecurityEventType.CSRF_VIOLATION, 

67 ip_address=client[0] if client else "unknown", 

68 user_agent="", 

69 success=False, 

70 metadata={"path": scope.get("path", ""), "reason": reason}, 

71 ) 

72 except Exception: # noqa: BLE001 — audit failures must not break requests 

73 logger.warning("csrf.audit_failed", reason=reason) 

74 

75 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

76 """ASGI entry point. 

77 

78 Args: 

79 scope: ASGI scope dict. 

80 receive: ASGI receive callable. 

81 send: ASGI send callable. 

82 """ 

83 if scope.get("type") != "http": 

84 await self._app(scope, receive, send) 

85 return 

86 

87 method = scope.get("method", "") 

88 path = scope.get("path", "") 

89 

90 # Skip non-mutating methods 

91 if method not in _CSRF_METHODS: 

92 await self._app(scope, receive, send) 

93 return 

94 

95 # Skip bypass paths 

96 if self._is_bypass_path(path): 

97 await self._app(scope, receive, send) 

98 return 

99 

100 request = StarletteRequest(scope, receive) 

101 

102 if not await self._validate_csrf(request): 

103 await self._send_403(send) 

104 return 

105 

106 await self._app(scope, receive, send) 

107 

108 def _is_bypass_path(self, path: str) -> bool: 

109 """Check if the path bypasses CSRF validation. 

110 

111 Args: 

112 path: Request path. 

113 

114 Returns: 

115 True if CSRF validation should be skipped. 

116 """ 

117 # Strip the admin prefix if present 

118 check_path = path 

119 if check_path.startswith("/admin"): 

120 check_path = check_path[len("/admin") :] 

121 

122 if check_path in _CSRF_BYPASS_PATHS: 

123 return True 

124 return bool(check_path.startswith("/static")) 

125 

126 async def _validate_csrf(self, request: StarletteRequest) -> bool: 

127 """Extract and validate CSRF token from request. 

128 

129 Content-Type dictates token location (AUTH-07): 

130 - ``application/x-www-form-urlencoded`` / ``multipart/form-data`` 

131 → token from form body. 

132 - Any other Content-Type (JSON, etc.) → token from ``X-CSRF-Token`` 

133 header. 

134 Mismatches are rejected with 403. 

135 

136 Args: 

137 request: Starlette Request object. 

138 

139 Returns: 

140 True if CSRF token is valid, False otherwise. 

141 """ 

142 try: 

143 session = getattr(request, "session", {}) 

144 # Pre-session forms (password reset, verify-email) bind their 

145 # token to ``csrf_session_id``; authenticated flows re-use the 

146 # ``admin_user_id`` scope. Validate against whichever exists. 

147 session_id: str = session.get("csrf_session_id") or session.get( 

148 "admin_user_id", "anonymous" 

149 ) 

150 

151 content_type = (request.headers.get("content-type") or "").lower() 

152 is_form = content_type.startswith( 

153 ("application/x-www-form-urlencoded", "multipart/form-data") 

154 ) 

155 

156 token: str | None = None 

157 

158 if is_form: 

159 try: 

160 form = await request.form() 

161 request.scope["admin_form_data"] = form 

162 raw = form.get("csrf_token") 

163 if isinstance(raw, str): 

164 token = raw 

165 except (RuntimeError, ValueError, OSError): # noqa: BLE001 

166 pass 

167 # Fall back to X-CSRF-Token header (HTMX injects this via 

168 # htmx:configRequest). Needed when hx-include triggers 

169 # form-encoding but the token lives in the header. 

170 if not token: 

171 token = request.headers.get("X-CSRF-Token") 

172 else: 

173 # JSON, fetch, etc.: token must come from X-CSRF-Token header 

174 token = request.headers.get("X-CSRF-Token") 

175 

176 if not token: 

177 logger.warning( 

178 "csrf.token_missing", 

179 path=str(request.url.path), 

180 content_type=content_type, 

181 ) 

182 await self._audit_violation(request.scope, "token_missing") 

183 return False 

184 

185 is_valid = self._csrf_service.validate_token(session_id, token) 

186 if not is_valid: 

187 logger.warning( 

188 "csrf.token_invalid", 

189 path=str(request.url.path), 

190 session_id=session_id, 

191 ) 

192 await self._audit_violation(request.scope, "token_invalid") 

193 return is_valid 

194 except Exception: # noqa: BLE001 

195 logger.warning("csrf.validation_error") 

196 return False 

197 

198 async def _send_403(self, send: Send) -> None: 

199 """Send a 403 Forbidden HTML response. 

200 

201 Args: 

202 send: ASGI send callable. 

203 """ 

204 body = ( 

205 b"<!DOCTYPE html>\n" 

206 b"<html><head><title>403 Forbidden</title></head>\n" 

207 b"<body><h1>403 Forbidden</h1>\n" 

208 b"<p>Invalid or missing CSRF token. " 

209 b"Please reload the page and try again.</p>\n" 

210 b'<a href="/admin/">Return to Admin</a>\n' 

211 b"</body></html>" 

212 ) 

213 await send( 

214 { 

215 "type": "http.response.start", 

216 "status": 403, 

217 "headers": [ 

218 [b"content-type", b"text/html; charset=utf-8"], 

219 [b"content-length", str(len(body)).encode()], 

220 ], 

221 } 

222 ) 

223 await send( 

224 { 

225 "type": "http.response.body", 

226 "body": body, 

227 } 

228 ) 

229 

230 

231__all__ = ["AdminCsrfMiddleware"]