Coverage for src/lexigram/admin/middleware/security_headers.py: 94%

1"""Security headers middleware for lexigram-admin. 

2 

3Applies OWASP-recommended HTTP security headers to every response. 

4Implements SecurityHeadersProtocol from lexigram-contracts. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any 

10 

11from lexigram.admin.settings.panel.models import DEFAULT_CSP 

12from lexigram.contracts.security import SecurityHeadersProtocol 

13from lexigram.di.decorators import inject 

14from lexigram.logging import get_logger 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Callable 

18 

19logger = get_logger(__name__) 

20 

21 

22@inject 

23class AdminSecurityHeaders: 

24 """Concrete implementation of SecurityHeadersProtocol for lexigram-admin. 

25 

26 Merges OWASP-recommended headers into an existing header mapping without 

27 overwriting values that have been set explicitly by the application. 

28 """ 

29 

30 def __init__( 

31 self, 

32 csp: str = DEFAULT_CSP, 

33 hsts_max_age: int = 63072000, # 2 years 

34 ) -> None: 

35 self._headers: dict[str, str] = { 

36 "Strict-Transport-Security": f"max-age={hsts_max_age}; includeSubDomains", 

37 "X-Frame-Options": "DENY", 

38 "X-Content-Type-Options": "nosniff", 

39 "Referrer-Policy": "strict-origin-when-cross-origin", 

40 "Permissions-Policy": ( 

41 "geolocation=(), microphone=(), camera=(), " 

42 "payment=(), usb=(), fullscreen=(self)" 

43 ), 

44 "Content-Security-Policy": csp, 

45 } 

46 

47 def apply(self, headers: dict[str, str]) -> dict[str, str]: 

48 """Apply security headers to an existing headers mapping. 

49 

50 Only adds headers that are not already present so that route-level 

51 overrides are preserved. 

52 

53 Args: 

54 headers: Mutable mapping of header names to values. 

55 

56 Returns: 

57 The headers mapping with security headers merged in. 

58 """ 

59 for name, value in self._headers.items(): 

60 headers.setdefault(name, value) 

61 return headers 

62 

63 

64# Verify structural compliance at import time (zero-cost check via isinstance) 

65assert isinstance(AdminSecurityHeaders(), SecurityHeadersProtocol) # noqa: S101 # import-time protocol conformance 

66 

67 

68@inject 

69class SecurityHeadersMiddleware: 

70 """ASGI middleware that injects security headers on every HTTP response. 

71 

72 Wraps the inner application and calls AdminSecurityHeaders.apply on the 

73 response headers before they are sent to the client. When a 

74 ``settings_store`` is provided, CSP and HSTS values are read from it on 

75 first use (once per process) and cached. 

76 """ 

77 

78 def __init__( 

79 self, 

80 app: Callable, 

81 headers_service: SecurityHeadersProtocol | None = None, 

82 settings_store: Any = None, 

83 ) -> None: 

84 self._app = app 

85 self._service: SecurityHeadersProtocol = ( 

86 headers_service if headers_service is not None else AdminSecurityHeaders() 

87 ) 

88 self._settings_store = settings_store 

89 self._resolved: SecurityHeadersProtocol | None = None 

90 

91 async def _resolve_headers(self) -> SecurityHeadersProtocol: 

92 """Return the headers service, applying settings overrides once.""" 

93 if self._resolved is not None: 

94 return self._resolved 

95 

96 service = self._service 

97 if self._settings_store is not None: 

98 try: 

99 csp = await self._settings_store.get("admin.security.csp") 

100 hsts = await self._settings_store.get("admin.security.hsts_max_age") 

101 if csp or hsts: 

102 service = AdminSecurityHeaders( 

103 csp=str(csp) if csp else DEFAULT_CSP, 

104 hsts_max_age=int(hsts) if hsts else 63072000, 

105 ) 

106 except (RuntimeError, ValueError, TypeError) as exc: 

107 logger.warning("admin.security_headers.settings_error", error=str(exc)) 

108 

109 self._resolved = service 

110 return service 

111 

112 async def __call__( 

113 self, 

114 scope: dict[str, Any], 

115 receive: Callable, 

116 send: Callable, 

117 ) -> None: 

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

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

120 return 

121 

122 service = await self._resolve_headers() 

123 

124 async def send_with_headers(message: dict[str, Any]) -> None: 

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

126 raw_headers: list[tuple[bytes, bytes]] = list( 

127 message.get("headers", []) 

128 ) 

129 # Build a mutable dict from existing headers (preserve case) 

130 existing: dict[str, str] = { 

131 k.decode(): v.decode() for k, v in raw_headers 

132 } 

133 updated = service.apply(existing) 

134 message = { 

135 **message, 

136 "headers": [(k.encode(), v.encode()) for k, v in updated.items()], 

137 } 

138 await send(message) 

139 

140 await self._app(scope, receive, send_with_headers) 

141 

142 

143__all__ = ["AdminSecurityHeaders", "SecurityHeadersMiddleware"]