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

1"""Security headers middleware for web applications. 

2 

3Implements OWASP recommended security headers to protect against common vulnerabilities. 

4Consumes SecurityConfig directly from application.yaml structure. 

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 ( 

13 CrossOriginConfig, 

14 CSPConfig, 

15 CSRFConfig, 

16 HSTSConfig, 

17 SecurityConfig, 

18) 

19from lexigram.web.security.sanitization import HeaderSanitizer 

20 

21logger = get_logger(__name__) 

22 

23_header_sanitizer = HeaderSanitizer() 

24 

25 

26if TYPE_CHECKING: 

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

28 

29 

30class SecurityHeadersMiddleware: 

31 """Middleware to add security headers to all responses. 

32 

33 Implements OWASP security best practices for HTTP headers. 

34 Consumes SecurityConfig directly from application.yaml. 

35 

36 Usage: 

37 from lexigram.web.middleware.security import SecurityHeadersMiddleware 

38 from lexigram.web.security.config import SecurityConfig 

39 

40 config = SecurityConfig(...) 

41 app.add_middleware(SecurityHeadersMiddleware, config=config) 

42 """ 

43 

44 def __init__( 

45 self, 

46 app: ASGIApp, 

47 config: SecurityConfig | None = None, 

48 enabled: bool = True, 

49 ): 

50 """Initialize security headers middleware. 

51 

52 Args: 

53 app: ASGI application 

54 config: SecurityConfig from application.yaml 

55 enabled: Whether middleware is enabled 

56 """ 

57 self.app = app 

58 self.enabled = enabled 

59 

60 self.config = config or SecurityConfig() 

61 

62 if enabled: 

63 logger.info("Security headers middleware enabled") 

64 if self.config.hsts.enabled: 

65 logger.info("HSTS enabled - ensure HTTPS is configured") 

66 else: 

67 logger.warning("Security headers middleware disabled") 

68 

69 def _build_csp_header(self) -> str: 

70 """Build Content-Security-Policy header value from directives.""" 

71 if not self.config.csp.enabled: 

72 return "" 

73 return self.config.csp.build_header() 

74 

75 def _build_hsts_header(self) -> str: 

76 """Build Strict-Transport-Security header value.""" 

77 parts = [f"max-age={self.config.hsts.max_age}"] 

78 if self.config.hsts.include_subdomains: 

79 parts.append("includeSubDomains") 

80 if self.config.hsts.preload: 

81 parts.append("preload") 

82 return "; ".join(parts) 

83 

84 def _build_permissions_policy_header(self) -> str: 

85 """Build Permissions-Policy header value.""" 

86 policies = [] 

87 for feature, allowlist in self.config.permissions_policy.items(): 

88 policies.append(f"{feature}={allowlist}") 

89 return ", ".join(policies) 

90 

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

92 """Add security headers to response.""" 

93 if scope["type"] != "http" or not self.enabled: 

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

95 return 

96 

97 async def send_wrapper(message: Any) -> None: 

98 if message["type"] == "http.response.start": 

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

100 

101 # Content-Security-Policy 

102 if self.config.csp.enabled and self.config.csp.directives: 

103 csp = self._build_csp_header() 

104 if csp: 

105 headers.append((b"content-security-policy", csp.encode())) 

106 

107 # Custom headers from config dict — sanitize values before encoding 

108 # to prevent CRLF injection / response-splitting attacks. 

109 for header_name, header_value in self.config.custom_headers.items(): 

110 name = header_name.lower().encode() 

111 safe_value = _header_sanitizer.sanitize_header_value( 

112 str(header_value) 

113 ) 

114 headers.append((name, safe_value.encode())) 

115 

116 # Strict-Transport-Security (only in production with HTTPS) 

117 if self.config.hsts.enabled: 

118 hsts = self._build_hsts_header() 

119 headers.append((b"strict-transport-security", hsts.encode())) 

120 

121 # Referrer-Policy (use referrer_policy field if not already in custom_headers) 

122 if ( 

123 self.config.referrer_policy 

124 and "referrer-policy" not in self.config.custom_headers 

125 ): 

126 headers.append( 

127 (b"referrer-policy", self.config.referrer_policy.encode()), 

128 ) 

129 

130 # Permissions-Policy 

131 if self.config.permissions_policy: 

132 policy = self._build_permissions_policy_header() 

133 headers.append((b"permissions-policy", policy.encode())) 

134 

135 # Cross-Origin policies 

136 if self.config.cross_origin.enabled: 

137 headers.append( 

138 ( 

139 b"cross-origin-embedder-policy", 

140 self.config.cross_origin.embedder_policy.encode(), 

141 ), 

142 ) 

143 headers.append( 

144 ( 

145 b"cross-origin-opener-policy", 

146 self.config.cross_origin.opener_policy.encode(), 

147 ), 

148 ) 

149 headers.append( 

150 ( 

151 b"cross-origin-resource-policy", 

152 self.config.cross_origin.resource_policy.encode(), 

153 ), 

154 ) 

155 

156 message["headers"] = headers 

157 

158 await send(message) 

159 

160 await self.app(scope, receive, send_wrapper) 

161 

162 

163def create_security_middleware( 

164 config: SecurityConfig, 

165 enabled: bool = True, 

166) -> dict[str, Any]: 

167 """Factory to create SecurityHeadersMiddleware kwargs from SecurityConfig. 

168 

169 Returns kwargs dict suitable for Starlette's Middleware class: 

170 Middleware(SecurityHeadersMiddleware, **kwargs) 

171 """ 

172 return { 

173 "config": config, 

174 "enabled": enabled, 

175 } 

176 

177 

178def create_development_config() -> SecurityConfig: 

179 """Create security headers config for development. 

180 

181 More permissive CSP and disables HSTS. 

182 """ 

183 return SecurityConfig( 

184 hsts=HSTSConfig(enabled=False), 

185 csp=CSPConfig( 

186 enabled=True, 

187 directives={ 

188 "default-src": "'self'", 

189 "script-src": "'self' 'unsafe-inline' 'unsafe-eval' http://localhost:* https://unpkg.com https://cdn.jsdelivr.net", 

190 "script-src-elem": "'self' 'unsafe-inline' 'unsafe-eval' http://localhost:* https://unpkg.com https://cdn.jsdelivr.net", 

191 "style-src": "'self' 'unsafe-inline' http://localhost:* https://unpkg.com https://cdn.jsdelivr.net", 

192 "img-src": "'self' data: http: https: blob:", 

193 "font-src": "'self' data: http://localhost:* https://unpkg.com", 

194 "connect-src": "'self' http://localhost:* ws://localhost:* wss://localhost:* https: wss:", 

195 "frame-ancestors": "'self' http://localhost:*", 

196 }, 

197 ), 

198 cross_origin=CrossOriginConfig(enabled=False), 

199 ) 

200 

201 

202def create_production_config() -> SecurityConfig: 

203 """Create security headers config optimized for production. 

204 

205 More restrictive CSP, enables HSTS and CSRF. 

206 """ 

207 return SecurityConfig( 

208 hsts=HSTSConfig( 

209 enabled=True, 

210 max_age=31536000, 

211 include_subdomains=True, 

212 preload=False, 

213 ), 

214 csp=CSPConfig( 

215 enabled=True, 

216 directives={ 

217 "default-src": "'self'", 

218 "script-src": "'self'", 

219 "script-src-elem": "'self'", 

220 "style-src": "'self'", 

221 "img-src": "'self' https: data: blob:", 

222 "font-src": "'self' data:", 

223 "connect-src": "'self' wss: https:", 

224 "frame-ancestors": "'none'", 

225 "base-uri": "'self'", 

226 "form-action": "'self'", 

227 "upgrade-insecure-requests": "", 

228 }, 

229 ), 

230 cross_origin=CrossOriginConfig(enabled=True), 

231 csrf=CSRFConfig(enabled=True), 

232 ) 

233 

234 

235__all__ = [ 

236 "SecurityHeadersMiddleware", 

237 "create_development_config", 

238 "create_production_config", 

239 "create_security_middleware", 

240]