Coverage for src/lexigram/web/security/cors/middleware.py: 20%

1"""CORS middleware for ASGI applications. 

2 

3Implements Cross-Origin Resource Sharing (CORS) per the WHATWG CORS spec, 

4using configuration from :class:`~lexigram.web.security.config.CORSConfig`. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.logging import get_logger 

12from lexigram.web.security.config import CORSConfig 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Callable 

16 

17logger = get_logger(__name__) 

18 

19_SIMPLE_METHODS = {"GET", "HEAD", "POST"} 

20_SIMPLE_HEADERS = { 

21 "accept", 

22 "accept-language", 

23 "content-language", 

24 "content-type", 

25} 

26 

27 

28class CORSMiddleware: 

29 """ASGI middleware for Cross-Origin Resource Sharing (CORS). 

30 

31 Handles preflight requests, adds CORS headers to responses, and enforces 

32 origin allowlist and credential policies per CORS specification. 

33 

34 Attributes: 

35 app: The wrapped ASGI application. 

36 config: CORS configuration. 

37 """ 

38 

39 def __init__(self, app: Callable[..., Any], config: CORSConfig) -> None: 

40 """Initialize middleware. 

41 

42 Args: 

43 app: ASGI application to wrap. 

44 config: CORS configuration. 

45 """ 

46 self.app = app 

47 self.config = config 

48 self._preflight_headers: dict[str, str] = {} 

49 self._build_preflight_headers() 

50 

51 def _build_preflight_headers(self) -> None: 

52 """Pre-compute preflight response headers.""" 

53 methods_str = ", ".join(sorted(self.config.allow_methods)) 

54 headers_str = ( 

55 ", ".join(sorted(self.config.allow_headers)) 

56 if self.config.allow_headers != ["*"] 

57 else "*" 

58 ) 

59 

60 self._preflight_headers = { 

61 "Access-Control-Allow-Methods": methods_str, 

62 "Access-Control-Allow-Headers": headers_str, 

63 } 

64 

65 if self.config.expose_headers: 

66 self._preflight_headers["Access-Control-Expose-Headers"] = ", ".join( 

67 self.config.expose_headers 

68 ) 

69 

70 if self.config.max_age: 

71 self._preflight_headers["Access-Control-Max-Age"] = str(self.config.max_age) 

72 

73 def _is_origin_allowed(self, origin: str) -> bool: 

74 """Check if origin is in allow-list (case-insensitive). 

75 

76 Args: 

77 origin: Request origin. 

78 

79 Returns: 

80 True if origin is allowed, False otherwise. 

81 """ 

82 if not origin: 

83 return False 

84 

85 if "*" in self.config.allowed_origins: 

86 return True 

87 

88 origin_lower = origin.lower() 

89 return any(o.lower() == origin_lower for o in self.config.allowed_origins) 

90 

91 def _get_cors_headers(self, origin: str | None) -> dict[str, str]: 

92 """Compute CORS response headers for a request. 

93 

94 Args: 

95 origin: Request origin. 

96 

97 Returns: 

98 Dictionary of CORS headers to add to response. 

99 """ 

100 headers: dict[str, str] = {} 

101 

102 if not origin or not self._is_origin_allowed(origin): 

103 return headers 

104 

105 if "*" in self.config.allowed_origins: 

106 if self.config.allow_credentials: 

107 headers["Access-Control-Allow-Origin"] = origin 

108 else: 

109 headers["Access-Control-Allow-Origin"] = "*" 

110 else: 

111 headers["Access-Control-Allow-Origin"] = origin 

112 

113 if self.config.allow_credentials: 

114 headers["Access-Control-Allow-Credentials"] = "true" 

115 

116 if self.config.expose_headers: 

117 headers["Access-Control-Expose-Headers"] = ", ".join( 

118 self.config.expose_headers 

119 ) 

120 

121 # Vary: Origin required when the response differs by origin (not wildcard). 

122 # Without it, shared caches may serve a CORS-enabled response to a different 

123 # origin that should not receive CORS headers — a cache-poisoning risk. 

124 if headers.get("Access-Control-Allow-Origin") != "*": 

125 headers["Vary"] = "Origin" 

126 

127 return headers 

128 

129 async def __call__( 

130 self, 

131 scope: dict[str, Any], 

132 receive: Callable[..., Any], 

133 send: Callable[..., Any], 

134 ) -> None: 

135 """ASGI middleware entry point. 

136 

137 Args: 

138 scope: ASGI scope. 

139 receive: ASGI receive callable. 

140 send: ASGI send callable. 

141 """ 

142 if scope["type"] != "http": 

143 await self.app(scope, receive, send) 

144 return 

145 

146 headers = {k.lower(): v for k, v in scope.get("headers", [])} 

147 origin_bytes = headers.get(b"origin") 

148 origin = origin_bytes.decode("latin-1") if origin_bytes else None 

149 method = scope.get("method", "GET") 

150 

151 # Handle preflight requests 

152 if method == "OPTIONS" and origin: 

153 cors_headers = self._get_cors_headers(origin) 

154 if cors_headers: 

155 # Validate Access-Control-Request-Headers against allowlist. 

156 requested_headers_raw = headers.get( 

157 b"access-control-request-headers", b"" 

158 ).decode() 

159 if requested_headers_raw: 

160 requested = { 

161 h.strip().lower() for h in requested_headers_raw.split(",") 

162 } 

163 if "*" not in self.config.allow_headers: 

164 allowed = {h.lower() for h in self.config.allow_headers} 

165 if not requested.issubset(allowed): 

166 logger.warning( 

167 "cors_preflight_rejected_headers", 

168 origin=origin, 

169 requested=sorted(requested), 

170 allowed=sorted(allowed), 

171 ) 

172 await send( 

173 { 

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

175 "status": 403, 

176 "headers": [], 

177 } 

178 ) 

179 await send( 

180 { 

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

182 "body": b"", 

183 "more_body": False, 

184 } 

185 ) 

186 return 

187 

188 all_headers = {**self._preflight_headers, **cors_headers} 

189 await send( 

190 { 

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

192 "status": 204, 

193 "headers": [ 

194 (k.encode(), v.encode()) for k, v in all_headers.items() 

195 ], 

196 } 

197 ) 

198 await send({"type": "http.response.body", "body": b""}) 

199 return 

200 

201 # Wrap send to add CORS headers to actual response 

202 async def send_with_cors(message: dict[str, Any]) -> None: 

203 if message["type"] == "http.response.start" and origin: 

204 cors_headers = self._get_cors_headers(origin) 

205 if cors_headers: 

206 headers_list = list(message.get("headers", [])) 

207 for k, v in cors_headers.items(): 

208 headers_list.append((k.encode(), v.encode())) 

209 message["headers"] = headers_list 

210 

211 await send(message) 

212 

213 await self.app(scope, receive, send_with_cors) 

214 

215 

216class CORSMiddlewareFactory: 

217 """Factory that builds configured :class:`CORSMiddleware` instances.""" 

218 

219 def __init__(self, config: CORSConfig | None = None) -> None: 

220 self._config = config or CORSConfig() 

221 

222 def __call__( 

223 self, 

224 app: Callable[..., Any], 

225 ) -> CORSMiddleware: 

226 """Return a middleware wrapping the provided ASGI app.""" 

227 return CORSMiddleware(app=app, config=self._config) 

228 

229 

230__all__ = ["CORSMiddleware", "CORSMiddlewareFactory"]