Coverage for src / lexigram / admin / di / sub_providers / auth.py: 88%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Admin auth sub-provider — authentication, guards, sessions, CSRF, sanitization."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.contracts.auth.repositories import SessionRepositoryProtocol
8from lexigram.contracts.core.health import HealthCheckResult, HealthStatus
9from lexigram.logging import get_logger
11if TYPE_CHECKING:
12 from lexigram.admin.config import AdminConfig
13 from lexigram.contracts.core.di import (
14 ContainerRegistrarProtocol,
15 ContainerResolverProtocol,
16 )
18logger = get_logger(__name__)
21class AdminAuthSubProvider:
22 """Manages admin authentication infrastructure: guards, sessions, CSRF, security.
24 Registers auth-related services: guard chain, session manager, CSRF protection,
25 input sanitization, and security headers. Auth primitives are expected to be
26 provided through contract bindings by the host application's auth provider.
27 """
29 def __init__(
30 self,
31 config: AdminConfig,
32 auth_provider: Any | None = None,
33 **kwargs: object,
34 ) -> None:
35 self._config = config
36 self._auth_provider = auth_provider
37 self._kwargs = kwargs
38 self._initialized = False
40 @property
41 def config(self) -> AdminConfig:
42 """Return current admin config."""
43 return self._config
45 async def register(self, container: ContainerRegistrarProtocol) -> None:
46 """Register auth services: guard chain, session manager, CSRF, sanitizer."""
47 from lexigram.admin.auth.guard_chain import AdminGuardChain
48 from lexigram.admin.auth.guards import GuardConfig
49 from lexigram.admin.auth.session_manager import AdminSessionManager
50 from lexigram.admin.auth.store.direct_sql import DirectSQLAdminUserStore
51 from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
52 from lexigram.admin.auth.store.session_sql import AdminSessionSqlRepository
53 from lexigram.admin.middleware.input_sanitizer import AdminInputSanitizer
54 from lexigram.admin.middleware.security_headers import AdminSecurityHeaders
55 from lexigram.admin.rbac.service import PermissionService
56 from lexigram.contracts.auth.repositories import SessionRepositoryProtocol
58 # Bind the admin user store protocol to the SQL implementation.
59 # SetupMiddleware and any other service that needs to manage admin-panel
60 # accounts depend on AdminUserStoreProtocol — never on the concrete class.
61 container.singleton(AdminUserStoreProtocol, DirectSQLAdminUserStore)
63 # Register guard chain (container will instantiate via DI)
64 container.singleton(AdminGuardChain, AdminGuardChain)
66 # Register the SQL repository as SessionRepositoryProtocol so the
67 # container can inject it into AdminSessionManager.
68 container.singleton(SessionRepositoryProtocol, AdminSessionSqlRepository)
69 container.singleton(AdminSessionManager, AdminSessionManager)
71 # Guard config — pre-constructed with defaults
72 container.singleton(GuardConfig, GuardConfig())
74 container.singleton(AdminInputSanitizer, AdminInputSanitizer)
75 container.singleton(AdminSecurityHeaders, AdminSecurityHeaders)
76 container.singleton(PermissionService, PermissionService)
78 # ── AdminAuthorizerProtocol — request-entry RBAC (AUTH-09, AUTH-18) ────
79 # Default deny-all policy (fail-closed). App authors override this
80 # binding with a concrete policy (e.g. PiccolinaAdminAuthPolicy) in
81 # their contributor's on_admin_boot hook.
82 from lexigram.contracts.admin.authorizer import AdminAuthorizerProtocol
84 class _DefaultAuthorizer:
85 """Default admin authorizer — allows all authenticated users."""
87 async def authorize_request(self, user: object, request: object) -> bool:
88 return getattr(user, "user_id", None) is not None
90 async def can_view(
91 self, user: object, resource: str, record: object = None
92 ) -> bool: # noqa: ARG002
93 return getattr(user, "user_id", None) is not None
95 async def can_create(self, user: object, resource: str) -> bool: # noqa: ARG002
96 return False
98 async def can_update(
99 self, user: object, resource: str, record: object = None
100 ) -> bool: # noqa: ARG002
101 return False
103 async def can_delete(
104 self, user: object, resource: str, record: object = None
105 ) -> bool: # noqa: ARG002
106 return False
108 async def can_execute_action(
109 self, user: object, resource: str, action: str, record: object = None
110 ) -> bool: # noqa: ARG002
111 return False
113 container.singleton(AdminAuthorizerProtocol, _DefaultAuthorizer)
115 # ------------------------------------------------------------------
116 # New admin auth services
117 # ------------------------------------------------------------------
118 self._register_new_auth_services(container)
120 # ------------------------------------------------------------------
121 # Contract-first auth integration.
122 # Concrete implementations for auth primitives are expected to be
123 # registered by the application's auth provider.
124 # ------------------------------------------------------------------
126 def _register_new_auth_services(
127 self, container: ContainerRegistrarProtocol
128 ) -> None:
129 """Register SQL stores and auth services introduced in the new auth layer.
131 Registration order:
132 1. SQL stores (protocol → concrete SQL class, DI-wired via ``@inject``).
133 2. ``AdminAuditLogService`` (DI-wired via ``@inject``).
134 3. ``AdminPasswordPolicyService`` — pre-built from config; no DI deps.
135 4. ``AdminCsrfService`` — pre-built from config; no DI deps.
136 5. ``AdminSessionService`` — thin ``@inject`` subclass to pass config
137 values alongside the DI-resolved ``SessionRepositoryProtocol``.
138 6. ``AdminLoginAttemptService`` — DI-wired; cache is optional and wired
139 in ``boot()`` if ``CacheBackendProtocol`` is available.
140 7. ``AdminAuthService`` — DI-wired orchestrator.
142 Args:
143 container: The container registrar for the current boot phase.
144 """
145 from lexigram.admin.auth.protocols import (
146 AdminAccountLockoutStoreProtocol,
147 AdminAuditLogServiceProtocol,
148 AdminAuditLogStoreProtocol,
149 AdminAuthServiceProtocol,
150 AdminCsrfServiceProtocol,
151 AdminLoginAttemptServiceProtocol,
152 AdminLoginAttemptStoreProtocol,
153 AdminPasswordPolicyServiceProtocol,
154 AdminSessionServiceProtocol,
155 )
156 from lexigram.admin.auth.services.audit_log_service import AdminAuditLogService
157 from lexigram.admin.auth.services.auth_service import AdminAuthService
158 from lexigram.admin.auth.services.csrf_service import AdminCsrfService
159 from lexigram.admin.auth.services.login_attempt_service import (
160 AdminLoginAttemptService,
161 )
162 from lexigram.admin.auth.services.password_policy_service import (
163 AdminPasswordPolicyService,
164 )
165 from lexigram.admin.auth.services.session_service import AdminSessionService
166 from lexigram.admin.auth.store.audit_log_sql import AdminAuditLogSqlStore
167 from lexigram.admin.auth.store.lockout_sql import AdminAccountLockoutSqlStore
168 from lexigram.admin.auth.store.login_attempt_sql import (
169 AdminLoginAttemptSqlStore,
170 )
171 from lexigram.di.decorators import inject
173 # ── SQL stores ────────────────────────────────────────────────────
174 container.singleton(AdminLoginAttemptStoreProtocol, AdminLoginAttemptSqlStore)
175 container.singleton(
176 AdminAccountLockoutStoreProtocol, AdminAccountLockoutSqlStore
177 )
178 container.singleton(AdminAuditLogStoreProtocol, AdminAuditLogSqlStore)
180 # ── Config extraction (safe getattr — works even when config is None) ──
181 _auth_cfg = getattr(self._config, "auth", None)
182 _pp_cfg = getattr(_auth_cfg, "password_policy", None)
183 _sec_cfg = getattr(_auth_cfg, "security", None)
185 # ── AdminAuditLogService — DI-wired via @inject ───────────────────
186 container.singleton(AdminAuditLogServiceProtocol, AdminAuditLogService)
188 # ── AdminPasswordPolicyService — pre-built from config ────────────
189 # No DI dependencies; all params come from config, so a pre-built
190 # instance is the cleanest registration approach.
191 container.singleton(
192 AdminPasswordPolicyServiceProtocol,
193 AdminPasswordPolicyService(
194 min_length=getattr(_pp_cfg, "min_length", 12),
195 max_length=getattr(_pp_cfg, "max_length", 128),
196 require_uppercase=getattr(_pp_cfg, "require_uppercase", True),
197 require_lowercase=getattr(_pp_cfg, "require_lowercase", True),
198 require_digit=getattr(_pp_cfg, "require_digit", True),
199 require_special=getattr(_pp_cfg, "require_special", True),
200 reject_common_passwords=getattr(
201 _pp_cfg, "reject_common_passwords", True
202 ),
203 reject_containing_email=getattr(
204 _pp_cfg, "reject_containing_email", True
205 ),
206 ),
207 )
209 # ── AdminCsrfService — pre-built from config ──────────────────────
210 _session_secret: str = getattr(
211 _auth_cfg, "session_secret", "change-me-in-production"
212 )
213 # Token lifetime aligns with session idle TTL (AUTH-08)
214 _csrf_lifetime: int = getattr(_auth_cfg, "idle_timeout", 3600)
215 container.singleton(
216 AdminCsrfServiceProtocol,
217 AdminCsrfService(secret=_session_secret, token_lifetime=_csrf_lifetime),
218 )
220 # ── AdminSessionService — @inject subclass to pass config lifetimes ──
221 # Follows the same pattern as _AdminSessionCookieBackend: a thin
222 # @inject-decorated inner class captures config values from the
223 # closure while letting the container inject SessionRepositoryProtocol.
224 _session_lifetime: int = getattr(_auth_cfg, "session_lifetime", 86400)
225 _idle_timeout: int = getattr(_auth_cfg, "idle_timeout", 3600)
226 _fingerprint_secret: str = getattr(
227 _auth_cfg, "session_secret", "change-me-in-production"
228 )
230 @inject
231 class _AdminSessionServiceConfigured(AdminSessionService):
232 """Admin-scoped SessionService with config-driven lifetimes."""
234 def __init__(
235 self,
236 session_repo: SessionRepositoryProtocol,
237 ) -> None:
238 super().__init__(
239 session_repo=session_repo,
240 session_lifetime=_session_lifetime,
241 idle_timeout=_idle_timeout,
242 fingerprint_secret=_fingerprint_secret,
243 )
245 container.singleton(AdminSessionServiceProtocol, _AdminSessionServiceConfigured)
247 # ── AdminLoginAttemptService — DI-wired; cache wired in boot() ────
248 # The @inject decorator resolves attempt_store and lockout_store from
249 # the container. CacheBackendProtocol is optional (defaults to None);
250 # it is wired post-registration in boot() when cache is available.
251 container.singleton(AdminLoginAttemptServiceProtocol, AdminLoginAttemptService)
253 # ── AdminAuthService — DI-wired orchestrator ──────────────────────
254 container.singleton(AdminAuthServiceProtocol, AdminAuthService)
256 logger.debug("admin_auth.new_services_registered")
258 async def boot(self, container: ContainerResolverProtocol) -> None:
259 """Boot auth services: initialize guard chain, session management.
261 Attempts to initialize the schema for every new SQL store. Failures
262 are logged at WARNING level and never re-raised so that a missing
263 table does not prevent the admin panel from starting.
265 Also wires the optional ``CacheBackendProtocol`` into
266 ``AdminLoginAttemptService`` when the cache provider is available.
268 Args:
269 container: The container resolver (container is frozen at this point).
270 """
271 self._initialized = True
273 # ── Schema initialization for new auth stores ─────────────────────
274 from lexigram.admin.auth.protocols import (
275 AdminAccountLockoutStoreProtocol,
276 AdminAuditLogStoreProtocol,
277 AdminLoginAttemptServiceProtocol,
278 AdminLoginAttemptStoreProtocol,
279 )
281 for _store_protocol in (
282 AdminLoginAttemptStoreProtocol,
283 AdminAccountLockoutStoreProtocol,
284 AdminAuditLogStoreProtocol,
285 ):
286 try:
287 _store = await container.resolve(
288 _store_protocol, bypass_visibility=True
289 )
290 await _store.ensure_schema() # type: ignore[attr-defined]
291 except Exception as e:
292 logger.exception(f"admin_auth.schema_init_failed: {e}") # noqa: BLE001
293 logger.warning(
294 "admin_auth.schema_init_failed",
295 protocol=str(_store_protocol),
296 )
298 # ── Wire cache into AdminLoginAttemptService (optional) ───────────
299 try:
300 from lexigram.contracts.infra.cache import CacheBackendProtocol
302 _cache = await container.resolve(CacheBackendProtocol)
303 _attempt_svc = await container.resolve(
304 AdminLoginAttemptServiceProtocol, bypass_visibility=True
305 )
306 if hasattr(_attempt_svc, "_cache"):
307 _attempt_svc._cache = _cache
308 logger.debug("admin_auth.cache_wired")
309 except Exception:
310 logger.debug("admin_auth.cache_not_available")
312 async def shutdown(self) -> None:
313 """Shut down auth services."""
314 self._initialized = False
316 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
317 """Return auth infrastructure health status."""
318 return HealthCheckResult(
319 component="admin_auth",
320 status=HealthStatus.HEALTHY if self._initialized else HealthStatus.UNKNOWN,
321 message="Admin auth operational"
322 if self._initialized
323 else "Not yet initialized",
324 )
327__all__ = ["AdminAuthSubProvider"]