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

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 session_id: str = session.get("admin_user_id", "anonymous") 

145 

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

147 is_form = content_type.startswith( 

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

149 ) 

150 

151 token: str | None = None 

152 

153 if is_form: 

154 try: 

155 form = await request.form() 

156 request.scope["admin_form_data"] = form 

157 raw = form.get("csrf_token") 

158 if isinstance(raw, str): 

159 token = raw 

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

161 pass 

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

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

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

165 if not token: 

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

167 else: 

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

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

170 

171 if not token: 

172 logger.warning( 

173 "csrf.token_missing", 

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

175 content_type=content_type, 

176 ) 

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

178 return False 

179 

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

181 if not is_valid: 

182 logger.warning( 

183 "csrf.token_invalid", 

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

185 session_id=session_id, 

186 ) 

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

188 return is_valid 

189 except Exception: # noqa: BLE001 

190 logger.warning("csrf.validation_error") 

191 return False 

192 

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

194 """Send a 403 Forbidden HTML response. 

195 

196 Args: 

197 send: ASGI send callable. 

198 """ 

199 body = ( 

200 b"<!DOCTYPE html>\n" 

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

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

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

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

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

206 b"</body></html>" 

207 ) 

208 await send( 

209 { 

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

211 "status": 403, 

212 "headers": [ 

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

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

215 ], 

216 } 

217 ) 

218 await send( 

219 { 

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

221 "body": body, 

222 } 

223 ) 

224 

225 

226__all__ = ["AdminCsrfMiddleware"]