Coverage for src/lexigram/web/di/middleware_setup.py: 21%
110 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Middleware registration helper for the Lexigram web layer."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any, cast
7from lexigram.contracts.core import HookRegistryProtocol
8from lexigram.contracts.exceptions.container import UnresolvableDependencyError
9from lexigram.contracts.exceptions.provider import ModuleVisibilityError
10from lexigram.contracts.infra.cache import CacheBackendProtocol
11from lexigram.logging import get_logger
12from lexigram.primitives.context import Context
13from lexigram.web.middleware.security import SecurityHeadersMiddleware
15if TYPE_CHECKING:
16 from starlette.applications import Starlette
18 from lexigram.contracts.core.di import ContainerResolverProtocol
19 from lexigram.web.config import WebConfig
21logger = get_logger(__name__)
23__all__ = ["MiddlewareSetup"]
26class MiddlewareSetup:
27 """Encapsulates Starlette middleware registration for the Lexigram web layer.
29 Extracted from ``WebProvider`` to keep the provider focused on DI lifecycle
30 (register/boot/shutdown) rather than application configuration details.
32 Args:
33 config: The web configuration driving middleware selection and ordering.
34 """
36 def __init__(
37 self,
38 config: WebConfig,
39 hooks: HookRegistryProtocol | None = None,
40 ) -> None:
41 self._config = config
42 self._hooks = hooks
44 async def configure(
45 self,
46 app: Starlette,
47 container: ContainerResolverProtocol,
48 ) -> None:
49 """Apply all configured middleware to the application.
51 Middleware is applied in security-first order:
53 1. Auto-configure CSP for API documentation (mutates ``self._config``).
54 2. Security headers.
55 3. CORS.
56 4. CSRF (opt-in).
57 5. Request context propagation.
58 6. DI scope (request-scoped container).
59 7. Request body size limit.
60 8. Web hook emission middleware (outermost for broad coverage).
62 Args:
63 app: The Starlette application to configure.
64 container: The resolved DI container; used for optional CSRF cache
65 resolution.
66 """
67 await self._configure_csp()
68 self._add_security_headers(app)
69 self._add_host_validation(app)
70 self._add_cors(app)
71 await self._add_csrf(app, container)
72 self._add_request_context(app, container)
73 self._add_di_scope(app, container)
74 self._add_body_limit(app)
75 self._add_hooks(app)
77 # ------------------------------------------------------------------
78 # Private helpers
79 # ------------------------------------------------------------------
81 async def _configure_csp(self) -> None:
82 """Auto-configure CSP directives for API documentation if enabled."""
83 api_docs = getattr(self._config, "api_docs", None)
84 if not api_docs or not api_docs.enabled:
85 return
87 required_domains = api_docs.get_required_domains()
88 if not required_domains:
89 return
91 if not hasattr(self._config, "security") or not self._config.security:
92 from lexigram.web.security.config import SecurityConfig
94 self._config.security = SecurityConfig()
96 csp = self._config.security.csp
97 directives = csp.directives if hasattr(csp, "directives") else {}
98 for directive, domains in required_domains.items():
99 if directive not in directives:
100 directives[directive] = set()
101 current = directives[directive]
102 if isinstance(current, str):
103 current_set = {current}
104 elif not isinstance(current, set):
105 current_set = set(current) if current else set()
106 else:
107 current_set = current
109 current_set.update(domains)
110 directives[directive] = current_set
112 logger.info(
113 "auto_configured_csp_for_api_docs",
114 directives={
115 k: list(v) if isinstance(v, set) else v for k, v in directives.items()
116 },
117 )
119 def _add_security_headers(self, app: Starlette) -> None:
120 """Add security headers middleware when a security config is present."""
121 if hasattr(self._config, "security") and self._config.security:
122 if not getattr(self._config.security, "enabled", True):
123 return
124 app.add_middleware(
125 SecurityHeadersMiddleware,
126 config=self._config.security,
127 enabled=True,
128 )
130 def _add_host_validation(self, app: Starlette) -> None:
131 """Add host-header validation when an allowed_hosts allowlist is set."""
132 if not hasattr(self._config, "security") or not self._config.security:
133 return
134 allowed_hosts = list(getattr(self._config.security, "allowed_hosts", []))
135 if not allowed_hosts:
136 return
138 from lexigram.web.middleware.host import HostValidationMiddleware
140 app.add_middleware(HostValidationMiddleware, allowed_hosts=allowed_hosts)
142 def _add_cors(self, app: Starlette) -> None:
143 """Add CORS middleware when CORS is enabled in config."""
144 if not self._config.cors.enabled:
145 return
147 from lexigram.logging.debug import is_debug_mode
148 from lexigram.web.security.config import CORSConfig
149 from lexigram.web.security.cors.middleware import (
150 CORSMiddleware as WebCORSMiddleware,
151 )
153 cors_cfg = self._config.cors
154 _debug_active = is_debug_mode() or self._config.server.debug
155 allowed_origins = cors_cfg.allow_origins
156 if cors_cfg.debug_permissive and _debug_active:
157 allowed_origins = ["*"]
158 logger.warning(
159 "cors.debug_permissive",
160 reason="debug_permissive=True and debug mode active — using wildcard CORS origins",
161 )
163 web_cors_config = CORSConfig(
164 allowed_origins=allowed_origins,
165 allow_credentials=cors_cfg.allow_credentials,
166 allow_methods=cors_cfg.allow_methods,
167 allow_headers=cors_cfg.allow_headers,
168 expose_headers=cors_cfg.expose_headers,
169 max_age=cors_cfg.max_age,
170 )
171 app.add_middleware(WebCORSMiddleware, config=web_cors_config)
173 async def _add_csrf(
174 self,
175 app: Starlette,
176 container: ContainerResolverProtocol,
177 ) -> None:
178 """Add CSRF protection middleware when opt-in config is present."""
179 security_cfg = getattr(self._config, "security", None)
180 if security_cfg is None:
181 return
182 if not getattr(security_cfg, "enabled", True):
183 return
184 if not getattr(security_cfg, "enable_csrf", True):
185 return
186 csrf_cfg = getattr(security_cfg, "csrf", None)
187 if csrf_cfg is None or not csrf_cfg.enabled:
188 return
190 from lexigram.web.security.csrf.middleware import CSRFProtectionMiddleware
192 csrf_cache = None
193 try:
194 csrf_cache = cast("Any", container).resolve_sync(CacheBackendProtocol)
195 except (
196 UnresolvableDependencyError,
197 AttributeError,
198 RuntimeError,
199 ModuleVisibilityError,
200 ):
201 pass
203 app.add_middleware(
204 CSRFProtectionMiddleware,
205 config=csrf_cfg,
206 cache=csrf_cache,
207 )
209 def _add_request_context(
210 self,
211 app: Starlette,
212 container: ContainerResolverProtocol,
213 ) -> None:
214 """Add request context middleware wired to the shared core context."""
215 from lexigram.web.middleware.request_context import RequestContextMiddleware
217 context = cast("Any", container).resolve_sync(Context)
218 app.add_middleware(RequestContextMiddleware, context=context)
220 def _add_di_scope(
221 self,
222 app: Starlette,
223 container: ContainerResolverProtocol,
224 ) -> None:
225 """Add DI scope middleware for request-scoped container resolution."""
226 from lexigram.web.middleware.di_scope import DIScopeMiddleware
228 app.add_middleware(
229 DIScopeMiddleware,
230 container=cast("Any", container),
231 )
233 def _add_hooks(self, app: Starlette) -> None:
234 """Add outermost web hook middleware for request/response events."""
235 from lexigram.web.middleware.hooks import WebHooksMiddleware
237 app.add_middleware(WebHooksMiddleware, hooks=self._hooks)
239 def _add_body_limit(self, app: Starlette) -> None:
240 """Add request body size limit middleware when configured.
242 Added before the hook middleware so oversized requests are still
243 rejected near the edge while hooks can observe the prepared response.
244 """
245 max_body_size = getattr(self._config, "max_body_size", None)
246 if max_body_size is None:
247 return
249 from lexigram.web.middleware.body_limit import RequestBodySizeLimitMiddleware
251 app.add_middleware(
252 RequestBodySizeLimitMiddleware,
253 max_body_size=max_body_size,
254 )