Coverage for src/lexigram/admin/di/bundle_provider.py: 0%
241 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""Admin bundle provider — orchestrates focused sub-providers.
3Follows the AuthBundleProvider pattern from lexigram-auth.
5Mount-time phases (controllers, contributors, router) live in the
6``lexigram.admin.di.mount`` mixins; this module keeps the provider lifecycle
7(register/boot/shutdown/health) and the middleware wiring that must stay
8bound to this module's logger.
9"""
11from __future__ import annotations
13from typing import TYPE_CHECKING, Any, Self
15from lexigram.admin.config import AdminRbacConfig
16from lexigram.admin.di.mount import (
17 AdminMountContributorsMixin,
18 AdminMountControllersMixin,
19 AdminMountCoreMixin,
20 MountContext,
21)
22from lexigram.contracts.core.health import HealthCheckResult, HealthStatus
23from lexigram.contracts.core.provider import ProviderPriority
24from lexigram.di.provider import Provider
25from lexigram.logging import get_logger
27_log = get_logger(__name__)
29if TYPE_CHECKING:
30 from lexigram.admin.config import AdminConfig
31 from lexigram.contracts.core.di import (
32 ContainerRegistrarProtocol,
33 ContainerResolverProtocol,
34 )
37class AdminProvider(
38 AdminMountCoreMixin,
39 AdminMountControllersMixin,
40 AdminMountContributorsMixin,
41 Provider,
42):
43 """Orchestrates admin sub-providers for the full admin panel.
45 Sub-providers are focused helper classes (not Provider subclasses).
46 This follows the EventsProvider/AuthBundleProvider pattern.
48 Config is accepted only in __init__ and never mutated after construction.
49 Sub-providers are instantiated in register() — not in __init__ — so that
50 no DI work happens before the container is ready.
51 """
53 name = "admin"
54 priority = ProviderPriority.APPLICATION
55 config_key: str | None = None
56 # Do NOT declare config_key/config_model — admin config is always set
57 # programmatically via AdminModule.configure(), not from YAML auto-injection.
59 def __init__(
60 self,
61 config: AdminConfig | None = None,
62 auth_provider: Any | None = None,
63 resources: list[type] | None = None,
64 controllers: list[type] | None = None,
65 extra_providers: list[Any] | None = None,
66 **kwargs: Any,
67 ) -> None:
68 super().__init__(name="admin", priority=ProviderPriority.APPLICATION)
69 from lexigram.admin.config import AdminConfig as AdminConfigCls
71 self._config = config or AdminConfigCls()
72 self._auth_provider = auth_provider
73 self._resources = resources or []
74 self._controllers = controllers or []
75 self._extra_providers: list[Any] = extra_providers or []
76 self._kwargs = kwargs
78 if (
79 self._config.auth.env in {"production", "staging"}
80 and not self._config.strict_resource_resolution
81 ):
82 _log.warning(
83 "admin.strict_resource_resolution_disabled_in_production",
84 message="strict_resource_resolution=False in production/staging can hide missing admin routes",
85 )
87 # Sub-providers are populated in register() — empty until then.
88 self._sub_providers: list[Any] = []
89 self._resolved_resources: dict[str, Any] = {}
90 self._nav_item_builder: Any | None = None
91 self._admin_resolver: Any | None = None
92 self._mount_failures: dict[str, str] = {}
93 self._csrf_service: Any | None = None
94 # Middleware dependencies resolved in boot(); always assigned before
95 # mount_to_app() runs (boot failures fail application startup).
96 self._user_store: Any | None = None
97 self._session_service: Any | None = None
98 self._authorizer: Any | None = None
99 self._authorizer_service: Any | None = None
101 @property # type: ignore[misc]
102 def config(self) -> AdminConfig:
103 """Return current admin config."""
104 return self._config
106 @classmethod
107 def from_config(cls, config: AdminConfig, **context: Any) -> Self:
108 """Create provider from typed config."""
109 return cls(config=config, **context)
111 async def register(self, container: ContainerRegistrarProtocol) -> None:
112 """Register admin and all sub-providers.
114 Sub-providers are instantiated here (not in __init__) so that no DI
115 resolution or heavyweight initialisation happens before the container
116 lifecycle has started. No resolution is performed in this method —
117 only bindings are registered.
118 """
119 from lexigram.admin.controllers.dashboard import DashboardController
120 from lexigram.admin.controllers.impersonation import ImpersonationController
121 from lexigram.admin.controllers.tenancy import TenancyController
122 from lexigram.admin.controllers.widgets import WidgetController
123 from lexigram.admin.di.sub_providers.auth import AdminAuthSubProvider
124 from lexigram.admin.di.sub_providers.contributor import (
125 AdminContributorSubProvider,
126 )
127 from lexigram.admin.di.sub_providers.core import AdminCoreSubProvider
128 from lexigram.admin.di.sub_providers.dashboard import AdminDashboardSubProvider
129 from lexigram.admin.di.sub_providers.integrations import (
130 AdminIntegrationsSubProvider,
131 )
132 from lexigram.admin.di.sub_providers.realtime import AdminRealtimeSubProvider
133 from lexigram.admin.di.sub_providers.resource import AdminResourceSubProvider
134 from lexigram.admin.di.sub_providers.tenancy import AdminTenancySubProvider
135 from lexigram.admin.di.sub_providers.ui import AdminUISubProvider
136 from lexigram.admin.navigation.nav_item_builder import NavItemBuilder
138 contributor = AdminContributorSubProvider(
139 config=self._config,
140 contributors=self._kwargs.get("contributors", []),
141 )
142 tenancy = AdminTenancySubProvider(config=self._config)
143 # Boot order is intentional:
144 # 1. Core binds admin primitives used by later providers.
145 # 2. Auth binds identity/session services before resources/actions are wired.
146 # 3. Resource/UI/realtime bind user-facing admin surfaces.
147 # 4. Tenancy decorates data access before dashboard aggregation.
148 # 5. Dashboard consumes contributor registry state.
149 # 6. Contributor boots last because extension packages may depend on every earlier seat.
150 sub_providers: list[Any] = [
151 AdminCoreSubProvider(config=self._config, **self._kwargs),
152 AdminAuthSubProvider(
153 config=self._config, auth_provider=self._auth_provider
154 ),
155 AdminResourceSubProvider(config=self._config, resources=self._resources),
156 AdminUISubProvider(config=self._config),
157 AdminRealtimeSubProvider(config=self._config),
158 tenancy,
159 AdminDashboardSubProvider(
160 config=self._config,
161 contributor_registry=contributor.registry,
162 ),
163 contributor,
164 AdminIntegrationsSubProvider(config=self._config.integrations),
165 ]
166 self._sub_providers = sub_providers
168 nav_item_builder = NavItemBuilder(config=self._config)
169 self._nav_item_builder = nav_item_builder
171 container.singleton(AdminProvider, self)
172 # Register with string key so web provider can discover without importing admin
173 container.singleton("admin_bundle", self)
174 # Register NavItemBuilder as a pre-built instance (config is not in container)
175 container.singleton(NavItemBuilder, nav_item_builder)
176 # Register the resolved RBAC config so @inject consumers read the
177 # configured super-admin role, not a fresh default instance.
178 container.singleton(AdminRbacConfig, self._config.rbac)
179 # Register built-in controllers
180 container.singleton(WidgetController, WidgetController)
181 container.singleton(TenancyController, TenancyController)
182 container.singleton(DashboardController, DashboardController)
183 from lexigram.admin.services.impersonation import ImpersonationService
185 container.singleton(ImpersonationService, ImpersonationService)
186 container.singleton(ImpersonationController, ImpersonationController)
187 # Register the RBAC permission inventory (populated at mount time)
188 from lexigram.admin.rbac.inventory import PermissionInventoryService
190 container.singleton(PermissionInventoryService, PermissionInventoryService)
191 # Register controller classes for DI resolution
192 for controller_cls in self._controllers:
193 try:
194 container.singleton(controller_cls, controller_cls)
195 except Exception: # noqa: BLE001 — re-registration is expected; continue loop
196 _log.debug(
197 "admin.controller_already_registered",
198 controller=controller_cls.__name__,
199 )
200 # Register resource classes so the container can inject their service dependencies
201 for resource_cls in self._resources:
202 try:
203 container.singleton(resource_cls, resource_cls)
204 except Exception: # noqa: BLE001 — re-registration is expected; continue loop
205 _log.debug(
206 "admin.resource_already_registered", resource=resource_cls.__name__
207 )
208 for ep in self._extra_providers:
209 await ep.register(container)
210 for sp in self._sub_providers:
211 await sp.register(container)
213 # NOTE: TenantConfigProviderProtocol is not registered here — it is
214 # constructed directly in mount_to_app() via AdminSettingsDbProvider.
216 async def mount_to_app(
217 self,
218 app: Any,
219 container: ContainerResolverProtocol,
220 ) -> None:
221 """Build and mount the admin panel onto a Starlette application.
223 Called by the web provider during route setup, after the Starlette
224 app is created and all providers have booted.
226 Args:
227 app: The Starlette application to mount the admin panel on.
228 container: The DI resolver for resolving controller dependencies.
229 """
230 admin_resolver = self._admin_resolver or container
231 ctx = MountContext()
232 await self._mount_resources(admin_resolver, ctx)
233 await self._mount_settings_service(admin_resolver, ctx)
234 await self._mount_controllers(admin_resolver, ctx)
235 self._mount_nav_builder(ctx)
236 await self._mount_middleware(admin_resolver, ctx)
237 self._mount_tenant_scoping(ctx)
238 await self._mount_contributors(admin_resolver, ctx)
239 await self._mount_integration(container, ctx)
240 await self._mount_sse_widgets(container, ctx)
241 await self._mount_app_state(app, ctx)
242 _log.info("admin.mounted", prefix=self._config.prefix)
244 def _mount_nav_builder(self, ctx: MountContext) -> None:
245 """Populate the nav item builder with resolved resource instances."""
246 from lexigram.admin.navigation.nav_item_builder import NavItemBuilder
248 self._resolved_resources = ctx.resources
249 nav_builder = self._nav_item_builder
250 if nav_builder is None:
251 nav_builder = NavItemBuilder(config=self._config)
252 self._nav_item_builder = nav_builder
253 nav_builder.set_resources(ctx.resources)
254 ctx.nav_builder = nav_builder
256 async def _mount_middleware(self, admin_resolver: Any, ctx: MountContext) -> None:
257 """Resolve and stack admin middleware layers in mount order.
259 Args:
260 admin_resolver: The DI resolver for middleware dependencies.
261 ctx: Mount pipeline state (``middlewares`` populated in place).
262 """
263 # Resolve SetupMiddleware's dependency: the admin user store.
264 # Best-effort — if the store is not registered (e.g. custom auth setups)
265 # SetupMiddleware is simply not added and no setup redirect occurs.
266 middleware_stack: list[tuple[type, dict[str, Any]]] = []
267 try:
268 from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
269 from lexigram.admin.middleware.setup import SetupMiddleware
271 admin_user_store = await admin_resolver.resolve(
272 AdminUserStoreProtocol,
273 bypass_visibility=True,
274 )
275 middleware_stack.append(
276 (SetupMiddleware, {"admin_user_store": admin_user_store})
277 )
278 _log.debug("admin.setup_middleware_wired")
279 except Exception as exc: # noqa: BLE001 — SetupMiddleware is optional
280 _log.warning(
281 "admin.setup_middleware_skipped",
282 reason=str(exc),
283 )
285 # Wire AdminCsrfMiddleware — REQUIRED. The service is resolved at
286 # boot() (see boot()), so a missing binding fails startup, never
287 # silently disables CSRF.
288 from lexigram.admin.middleware.csrf import AdminCsrfMiddleware
290 csrf_service = await self._get_csrf_service(admin_resolver)
291 csrf_audit_service = None
292 try:
293 from lexigram.admin.auth.protocols import (
294 AdminAuditLogServiceProtocol,
295 )
297 csrf_audit_service = await admin_resolver.resolve(
298 AdminAuditLogServiceProtocol,
299 bypass_visibility=True,
300 )
301 except Exception as exc: # noqa: BLE001 — CSRF audit is optional
302 _log.warning("admin.csrf_audit_service_unavailable", reason=str(exc))
303 middleware_stack.append(
304 (
305 AdminCsrfMiddleware,
306 {
307 "csrf_service": csrf_service,
308 "audit_service": csrf_audit_service,
309 },
310 )
311 )
312 _log.debug("admin.csrf_middleware_wired")
314 # Wire session-based auth guard — redirects unauthenticated requests to login.
315 if self._config.require_auth:
316 try:
317 from lexigram.admin.middleware.auth_guard import (
318 AdminAuthGuardMiddleware,
319 )
321 middleware_stack.append((AdminAuthGuardMiddleware, {}))
322 _log.debug("admin.auth_guard_middleware_wired")
323 except Exception as exc: # noqa: BLE001 — guard middleware is optional
324 # Degrade (non-fatal), but log at error: the operator
325 # explicitly required auth and the guard could not be added.
326 if self._config.require_auth:
327 _log.error("admin.auth_guard_middleware_skipped", reason=str(exc))
328 else:
329 _log.warning("admin.auth_guard_middleware_skipped", reason=str(exc))
330 else:
331 _log.debug("admin.auth_guard_middleware_skipped_require_auth_unset")
333 # Wire auth middleware — loads user from session into request.state.user
334 # so that downstream authorization middleware can enforce RBAC.
335 # Dependencies resolved in boot(); a missing binding fails startup.
336 from lexigram.admin.middleware.auth import AdminAuthMiddleware
338 middleware_stack.append(
339 (
340 AdminAuthMiddleware,
341 {
342 "user_store": self._user_store,
343 "session_service": self._session_service,
344 "require_auth": False,
345 },
346 )
347 )
348 _log.debug("admin.auth_middleware_wired")
350 # Wire request-entry RBAC middleware — checks authorization before
351 # dispatching to handlers (AUTH-09, AUTH-18). Placed after the auth
352 # guard so request.state.user is populated. The authorizer is
353 # resolved in boot(); a missing binding fails startup.
354 from lexigram.admin.middleware.authorization import (
355 AdminAuthorizationMiddleware,
356 )
358 middleware_stack.append(
359 (AdminAuthorizationMiddleware, {"authorizer": self._authorizer})
360 )
361 _log.debug("admin.authorization_middleware_wired")
363 # Wire tenant middleware when tenancy is enabled (before auth guard
364 # so request.state.tenant_id is available during auth checks).
365 if self._config.tenancy.enabled:
366 from lexigram.admin.middleware.tenant import AdminTenantMiddleware
368 middleware_stack.insert(
369 0, (AdminTenantMiddleware, {"config": self._config.tenancy})
370 )
371 _log.debug("admin.tenant_middleware_wired")
373 # Wire AdminErrorMiddleware innermost so it catches exceptions from
374 # all other middleware and controllers. Best-effort — if the
375 # middleware fails to resolve, admin runs without custom error pages.
376 try:
377 from lexigram.admin.middleware.error import AdminErrorMiddleware
379 middleware_stack.append(
380 (
381 AdminErrorMiddleware,
382 {
383 "debug": self._config.debug,
384 "login_url": f"{self._config.prefix}/login",
385 },
386 )
387 )
388 _log.debug("admin.error_middleware_wired")
389 except Exception as exc: # noqa: BLE001 — error middleware is optional
390 _log.warning("admin.error_middleware_skipped", reason=str(exc))
392 # Wire HX-Push-Url for body-targeted htmx GETs so client-side
393 # navigation (htmx.ajax with target "body") keeps the address bar
394 # in sync and htmx history (back/forward) works.
395 from lexigram.admin.middleware.nav_push import AdminNavPushMiddleware
397 middleware_stack.append((AdminNavPushMiddleware, {}))
398 _log.debug("admin.nav_push_middleware_wired")
400 ctx.middlewares = middleware_stack
402 def _mount_tenant_scoping(self, ctx: MountContext) -> None:
403 """Wrap resource data sources with tenant scoping when enabled."""
404 if not self._config.tenancy.enabled:
405 return
406 from lexigram.admin.multitenancy.data_source import TenantScopedDataSource
408 tenant_id = self._config.tenancy.default_tenant_id
409 tenant_field = self._config.tenancy.tenant_field
410 for resource in ctx.resources.values():
411 ds = getattr(resource, "data_source", None) or getattr(
412 resource, "_data_source", None
413 )
414 if ds is not None:
415 scoped = TenantScopedDataSource(
416 data_source=ds,
417 tenant_id=tenant_id,
418 tenant_field=tenant_field,
419 )
420 resource.data_source = scoped
421 _log.debug("admin.resource_data_sources_tenant_scoped")
423 async def boot(self, container: ContainerResolverProtocol) -> None:
424 """Boot all sub-providers in order.
426 Raises:
427 RuntimeError: If the CSRF service cannot be resolved — admin
428 must not boot without CSRF enforcement (fail-closed).
429 """
430 self._admin_resolver = container
431 for sp in self._sub_providers:
432 await sp.boot(container)
434 # CSRF is mandatory. Resolve here, not in mount_to_app(): boot
435 # failures propagate through the orchestrator and fail application
436 # startup, whereas mount_to_app() exceptions are caught by the web
437 # provider's RouteSetup and logged, silently skipping the admin mount.
438 from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol
440 try:
441 self._csrf_service = await container.resolve(
442 AdminCsrfServiceProtocol,
443 bypass_visibility=True,
444 )
445 except Exception as exc: # noqa: BLE001 — re-raised as fatal below
446 _log.error(
447 "admin.csrf_service_resolution_failed",
448 error=str(exc),
449 error_type=type(exc).__name__,
450 )
451 raise RuntimeError(
452 "CSRF service could not be resolved; refusing to boot admin "
453 "without CSRF enforcement"
454 ) from exc
456 # The first-run wizard must be gated: boot refuses to start without
457 # a setup token (admin.auth.security.setup_token, legacy env var
458 # ADMIN_SETUP_TOKEN, or nested LEX_ADMIN_AUTH__SECURITY__SETUP_TOKEN)
459 # unless the operator explicitly opts out with
460 # admin.auth.security.setup_token_optin_unsafe=true for local/
461 # ephemeral environments only.
462 if (
463 not self._config.auth.security.setup_token
464 and not self._config.auth.security.setup_token_optin_unsafe
465 ):
466 raise RuntimeError(
467 "Refusing to boot admin without a setup token: set "
468 "ADMIN_SETUP_TOKEN (or config admin.auth.security.setup_token, "
469 "env LEX_ADMIN_AUTH__SECURITY__SETUP_TOKEN), or explicitly opt "
470 "out for local/ephemeral environments with "
471 "admin.auth.security.setup_token_optin_unsafe=true"
472 )
474 # AdminAuthMiddleware's dependencies are mandatory — the middleware
475 # that actually enforces identity must not be silently dropped by a
476 # mount-time resolution failure (RouteSetup swallows mount exception
477 # and skips the admin mount entirely). Resolve at boot with the same
478 # fail-loud shape as the CSRF block above.
479 from lexigram.admin.auth.protocols import AdminSessionServiceProtocol
480 from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
482 try:
483 self._user_store = await container.resolve(
484 AdminUserStoreProtocol,
485 bypass_visibility=True,
486 )
487 self._session_service = await container.resolve(
488 AdminSessionServiceProtocol,
489 bypass_visibility=True,
490 )
491 except Exception as exc: # noqa: BLE001 — re-raised as fatal below
492 _log.error(
493 "admin.auth_middleware_dependencies_resolution_failed",
494 error=str(exc),
495 error_type=type(exc).__name__,
496 )
497 raise RuntimeError(
498 "AdminAuthMiddleware dependencies could not be resolved; "
499 "refusing to boot admin without session validation"
500 ) from exc
502 # AdminAuthorizationMiddleware's authorizer is mandatory — RBAC
503 # enforcement must never silently degrade at startup.
504 from lexigram.admin.middleware.authorization import (
505 RequestAuthorizerProtocol,
506 )
508 try:
509 self._authorizer = await container.resolve(
510 RequestAuthorizerProtocol,
511 bypass_visibility=True,
512 )
513 except Exception as exc: # noqa: BLE001 — re-raised as fatal below
514 _log.error(
515 "admin.authorization_middleware_dependency_resolution_failed",
516 error=str(exc),
517 error_type=type(exc).__name__,
518 )
519 raise RuntimeError(
520 "AdminAuthorizationMiddleware's authorizer could not be "
521 "resolved; refusing to boot admin without RBAC enforcement"
522 ) from exc
524 from lexigram.contracts.auth import AuthorizerProtocol
526 try:
527 self._authorizer_service = await container.resolve(
528 AuthorizerProtocol,
529 bypass_visibility=True,
530 )
531 except Exception as exc: # noqa: BLE001 — re-raised as fatal below
532 _log.error(
533 "admin.authorizer_service_resolution_failed",
534 error=str(exc),
535 error_type=type(exc).__name__,
536 )
537 raise RuntimeError(
538 "AuthorizerProtocol could not be resolved; refusing to boot "
539 "admin without a per-resource permission source for search"
540 ) from exc
542 # Wire the container resolver into WidgetController so contributor
543 # render_widget() implementations can resolve their service dependencies.
544 try:
545 from lexigram.admin.controllers.widgets import WidgetController
547 wc = await container.resolve(WidgetController, bypass_visibility=True)
548 wc._resolver = container
549 except Exception:
550 _log.warning("admin.widget_controller_resolver_wire_failed", exc_info=True)
552 async def _get_csrf_service(self, admin_resolver: Any) -> Any:
553 """Return the boot-resolved CSRF service, resolving lazily if needed.
555 Mount can be invoked directly (tests / factory paths) without a prior
556 boot(); in that case resolve here. Failures are never swallowed.
558 Raises:
559 RuntimeError: If the CSRF service cannot be resolved.
560 """
561 if self._csrf_service is not None:
562 return self._csrf_service
563 from lexigram.admin.auth.protocols import AdminCsrfServiceProtocol
565 try:
566 self._csrf_service = await admin_resolver.resolve(
567 AdminCsrfServiceProtocol,
568 bypass_visibility=True,
569 )
570 except Exception as exc: # noqa: BLE001 — re-raised as fatal below
571 raise RuntimeError(
572 "CSRF service could not be resolved during admin mount"
573 ) from exc
574 return self._csrf_service
576 async def shutdown(self) -> None:
577 """Shut down sub-providers in reverse order."""
578 for sp in reversed(self._sub_providers):
579 await sp.shutdown()
581 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
582 """Aggregate health from all sub-providers."""
583 import asyncio
585 worst = HealthStatus.HEALTHY
586 details: dict[str, Any] = {}
587 for sp in self._sub_providers:
588 maybe = sp.health_check(timeout)
589 result: HealthCheckResult = (
590 await maybe if asyncio.iscoroutine(maybe) else maybe
591 )
592 details[result.component] = result.status.value
593 if result.status == HealthStatus.UNHEALTHY:
594 worst = HealthStatus.UNHEALTHY
595 elif (
596 result.status == HealthStatus.DEGRADED
597 and worst != HealthStatus.UNHEALTHY
598 ):
599 worst = HealthStatus.DEGRADED
600 elif (
601 result.status == HealthStatus.UNKNOWN and worst == HealthStatus.HEALTHY
602 ):
603 worst = HealthStatus.UNKNOWN
605 if self._mount_failures and worst != HealthStatus.UNHEALTHY:
606 worst = HealthStatus.DEGRADED
607 details["mount_failures"] = dict(self._mount_failures)
609 return HealthCheckResult(
610 component="admin",
611 status=worst,
612 message=f"Admin bundle: {worst.value}",
613 details=details,
614 )
617__all__ = ["AdminProvider"]