Coverage for src/lexigram/admin/di/sub_providers/auth.py: 98%

1"""Admin auth sub-provider — authentication, guards, sessions, CSRF, sanitization.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7# Runtime imports (not TYPE_CHECKING): the @inject local subclasses below 

8# annotate their constructors with these protocol names. Under 

9# ``from __future__ import annotations`` those annotations are strings that 

10# get_type_hints resolves against the module globals — function-local imports 

11# are invisible to it, which would silently drop the injectable parameters. 

12from lexigram.admin.auth.protocols import ( 

13 AdminAuditLogServiceProtocol, 

14 AdminEmailOtpServiceProtocol, 

15 AdminEmailOtpStoreProtocol, 

16 AdminEmailVerificationServiceProtocol, 

17 AdminEmailVerificationStoreProtocol, 

18 AdminLoginAttemptServiceProtocol, 

19 AdminMfaServiceProtocol, 

20 AdminMfaStoreProtocol, 

21 AdminSessionServiceProtocol, 

22) 

23from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol 

24from lexigram.admin.services.notifications import AdminNotificationService 

25from lexigram.contracts.auth.repositories import SessionRepositoryProtocol 

26from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

27from lexigram.contracts.data import DatabaseProviderProtocol 

28from lexigram.logging import get_logger 

29from lexigram.validation import SecretStr 

30 

31if TYPE_CHECKING: 

32 from lexigram.admin.config import AdminConfig 

33 from lexigram.contracts.core.di import ( 

34 ContainerRegistrarProtocol, 

35 ContainerResolverProtocol, 

36 ) 

37 

38 

39async def _resolve_admin_authorizer(container: Any) -> Any: 

40 """Return the lexigram-auth AuthorizationService, constructing on demand.""" 

41 from lexigram.auth.authz.service import AuthorizationService 

42 

43 try: 

44 return await container.resolve(AuthorizationService, bypass_visibility=True) 

45 except Exception: # noqa: BLE001 — auth provider not bound: default engine 

46 return AuthorizationService() 

47 

48 

49logger = get_logger(__name__) 

50 

51 

52class AdminAuthSubProvider: 

53 """Manages admin authentication infrastructure: guards, sessions, CSRF, security. 

54 

55 Registers auth-related services: guard chain, session manager, CSRF protection, 

56 input sanitization, and security headers. Auth primitives are expected to be 

57 provided through contract bindings by the host application's auth provider. 

58 """ 

59 

60 def __init__( 

61 self, 

62 config: AdminConfig, 

63 auth_provider: Any | None = None, 

64 **kwargs: object, 

65 ) -> None: 

66 self._config = config 

67 self._auth_provider = auth_provider 

68 self._kwargs = kwargs 

69 self._initialized = False 

70 

71 @property 

72 def config(self) -> AdminConfig: 

73 """Return current admin config.""" 

74 return self._config 

75 

76 async def register(self, container: ContainerRegistrarProtocol) -> None: 

77 """Register auth services: guard chain, session manager, CSRF, sanitizer.""" 

78 from lexigram.admin.auth.guard_chain import AdminGuardChain 

79 from lexigram.admin.auth.guards import GuardConfig 

80 from lexigram.admin.auth.session_manager import AdminSessionManager 

81 from lexigram.admin.auth.store.direct_sql import DirectSQLAdminUserStore 

82 from lexigram.admin.auth.store.session_sql import AdminSessionSqlRepository 

83 from lexigram.admin.middleware.input_sanitizer import AdminInputSanitizer 

84 from lexigram.admin.middleware.security_headers import AdminSecurityHeaders 

85 from lexigram.admin.rbac.service import PermissionService 

86 from lexigram.contracts.auth.repositories import SessionRepositoryProtocol 

87 

88 # Bind the admin user store protocol to the SQL implementation, or to 

89 # the app-principal adapter when configured (spec D3). SetupMiddleware 

90 # and any other service that needs to manage admin-panel accounts 

91 # depend on AdminUserStoreProtocol — never on the concrete class. 

92 if getattr(self._config.auth, "principal_source", "internal") == "app": 

93 from lexigram.admin.auth.store.app_principal import ( 

94 AppPrincipalUserStoreAdapter, 

95 ) 

96 

97 container.singleton(AdminUserStoreProtocol, AppPrincipalUserStoreAdapter) 

98 else: 

99 container.singleton(AdminUserStoreProtocol, DirectSQLAdminUserStore) 

100 

101 # Register guard chain (container will instantiate via DI) 

102 container.singleton(AdminGuardChain, AdminGuardChain) 

103 

104 # Register the SQL repository as SessionRepositoryProtocol so the 

105 # container can inject it into AdminSessionManager. 

106 container.singleton(SessionRepositoryProtocol, AdminSessionSqlRepository) 

107 container.singleton(AdminSessionManager, AdminSessionManager) 

108 

109 # Guard config — pre-constructed with defaults 

110 container.singleton(GuardConfig, GuardConfig()) 

111 

112 container.singleton(AdminInputSanitizer, AdminInputSanitizer) 

113 container.singleton(AdminSecurityHeaders, AdminSecurityHeaders) 

114 container.singleton(PermissionService, PermissionService) 

115 

116 # ── AuthorizerProtocol — single PDP instance (spec §2.2) ──────────── 

117 # The one authorization engine: lexigram-auth's AuthorizationService. 

118 # Every consumer (ResourceManager, ActionExecutor, PermissionService, 

119 # AdminRoleService) resolves the SAME object from the container. 

120 # ── RequestAuthorizerProtocol — request-entry RBAC (AUTH-09, AUTH-18) ─ 

121 # Default: authenticated users pass (fail-closed on identity). 

122 # App authors override either binding in their on_admin_boot hook. 

123 from lexigram.admin.middleware.authorization import ( 

124 DefaultRequestAuthorizer, 

125 RequestAuthorizerProtocol, 

126 ) 

127 from lexigram.contracts.auth import AuthorizerProtocol 

128 

129 container.singleton(AuthorizerProtocol, factory=_resolve_admin_authorizer) 

130 container.singleton(RequestAuthorizerProtocol, DefaultRequestAuthorizer) 

131 

132 # ------------------------------------------------------------------ 

133 # New admin auth services 

134 # ------------------------------------------------------------------ 

135 self._register_new_auth_services(container) 

136 

137 # ------------------------------------------------------------------ 

138 # Contract-first auth integration. 

139 # Concrete implementations for auth primitives are expected to be 

140 # registered by the application's auth provider. 

141 # ------------------------------------------------------------------ 

142 

143 def _register_new_auth_services( 

144 self, container: ContainerRegistrarProtocol 

145 ) -> None: 

146 """Register SQL stores and auth services introduced in the new auth layer. 

147 

148 Registration order: 

149 1. SQL stores (protocol → concrete SQL class, DI-wired via ``@inject``). 

150 2. ``AdminAuditLogService`` (DI-wired via ``@inject``). 

151 3. ``AdminPasswordPolicyService`` — pre-built from config; no DI deps. 

152 4. ``AdminCsrfService`` — pre-built from config; no DI deps. 

153 5. ``AdminSessionService`` — thin ``@inject`` subclass to pass config 

154 values alongside the DI-resolved ``SessionRepositoryProtocol``. 

155 6. ``AdminLoginAttemptService`` — DI-wired; cache is optional and wired 

156 in ``boot()`` if ``CacheBackendProtocol`` is available. 

157 7. ``AdminAuthService`` — DI-wired orchestrator. 

158 

159 Args: 

160 container: The container registrar for the current boot phase. 

161 """ 

162 from lexigram.admin.auth.protocols import ( 

163 AdminAccountLockoutStoreProtocol, 

164 AdminAuditLogServiceProtocol, 

165 AdminAuditLogStoreProtocol, 

166 AdminAuthServiceProtocol, 

167 AdminCsrfServiceProtocol, 

168 AdminLoginAttemptServiceProtocol, 

169 AdminLoginAttemptStoreProtocol, 

170 AdminPasswordPolicyServiceProtocol, 

171 AdminPasswordResetServiceProtocol, 

172 AdminPasswordResetTokenStoreProtocol, 

173 AdminSessionServiceProtocol, 

174 ) 

175 from lexigram.admin.auth.services.audit_log_service import AdminAuditLogService 

176 from lexigram.admin.auth.services.auth_service import AdminAuthService 

177 from lexigram.admin.auth.services.csrf_service import AdminCsrfService 

178 from lexigram.admin.auth.services.login_attempt_service import ( 

179 AdminLoginAttemptService, 

180 ) 

181 from lexigram.admin.auth.services.password_policy_service import ( 

182 AdminPasswordPolicyService, 

183 ) 

184 from lexigram.admin.auth.services.password_reset_service import ( 

185 AdminPasswordResetService, 

186 ) 

187 from lexigram.admin.auth.services.session_service import AdminSessionService 

188 from lexigram.admin.auth.store.audit_log_sql import AdminAuditLogSqlStore 

189 from lexigram.admin.auth.store.lockout_sql import AdminAccountLockoutSqlStore 

190 from lexigram.admin.auth.store.login_attempt_sql import ( 

191 AdminLoginAttemptSqlStore, 

192 ) 

193 from lexigram.admin.auth.store.password_reset_token_sql import ( 

194 AdminPasswordResetTokenSqlStore, 

195 ) 

196 from lexigram.di.decorators import inject 

197 

198 # ── SQL stores ──────────────────────────────────────────────────── 

199 container.singleton(AdminLoginAttemptStoreProtocol, AdminLoginAttemptSqlStore) 

200 container.singleton( 

201 AdminAccountLockoutStoreProtocol, AdminAccountLockoutSqlStore 

202 ) 

203 container.singleton(AdminAuditLogStoreProtocol, AdminAuditLogSqlStore) 

204 container.singleton( 

205 AdminPasswordResetTokenStoreProtocol, AdminPasswordResetTokenSqlStore 

206 ) 

207 

208 # ── Config extraction (safe getattr — works even when config is None) ── 

209 _auth_cfg = getattr(self._config, "auth", None) 

210 _pp_cfg = getattr(_auth_cfg, "password_policy", None) 

211 _sec_cfg = getattr(_auth_cfg, "security", None) 

212 _mfa_cfg = getattr(_auth_cfg, "mfa", None) 

213 _email_verif_cfg = getattr(_auth_cfg, "email_verification", None) 

214 _email_otp_cfg = getattr(_auth_cfg, "email_otp", None) 

215 

216 # ── AdminAuditLogService — DI-wired via @inject ─────────────────── 

217 container.singleton(AdminAuditLogServiceProtocol, AdminAuditLogService) 

218 

219 # ── AdminPasswordPolicyService — pre-built from config ──────────── 

220 # Rule evaluation delegates to lexigram-auth's PasswordPolicy; 

221 # the admin service keeps only the email-containment rule + the 

222 # admin violation/message contract. 

223 from lexigram.auth import PasswordPolicy 

224 

225 container.singleton( 

226 AdminPasswordPolicyServiceProtocol, 

227 AdminPasswordPolicyService( 

228 policy=PasswordPolicy( 

229 min_length=getattr(_pp_cfg, "min_length", 12), 

230 max_length=getattr(_pp_cfg, "max_length", 128), 

231 require_uppercase=getattr(_pp_cfg, "require_uppercase", True), 

232 require_lowercase=getattr(_pp_cfg, "require_lowercase", True), 

233 require_digits=getattr(_pp_cfg, "require_digit", True), 

234 require_special=getattr(_pp_cfg, "require_special", True), 

235 prevent_common=getattr(_pp_cfg, "reject_common_passwords", True), 

236 ), 

237 reject_containing_email=getattr( 

238 _pp_cfg, "reject_containing_email", True 

239 ), 

240 ), 

241 ) 

242 

243 # ── AdminCsrfService — pre-built from config ────────────────────── 

244 _session_secret: str = getattr( 

245 _auth_cfg, "session_secret", "change-me-in-production" 

246 ) 

247 if isinstance(_session_secret, SecretStr): 

248 _session_secret = _session_secret.get_secret_value() 

249 # Token lifetime follows the dedicated CSRF expiry setting (AUTH-07); 

250 # the csrf_token_lifetime <= idle_timeout validator governs it. 

251 _csrf_lifetime: int = getattr(_auth_cfg, "csrf_token_lifetime", 3600) 

252 container.singleton( 

253 AdminCsrfServiceProtocol, 

254 AdminCsrfService(secret=_session_secret, token_lifetime=_csrf_lifetime), 

255 ) 

256 

257 # ── AdminSessionService — @inject subclass to pass config lifetimes ── 

258 # Follows the same pattern as _AdminSessionCookieBackend: a thin 

259 # @inject-decorated inner class captures config values from the 

260 # closure while letting the container inject SessionRepositoryProtocol. 

261 _session_lifetime: int = getattr(_auth_cfg, "session_lifetime", 86400) 

262 _idle_timeout: int = getattr(_auth_cfg, "idle_timeout", 3600) 

263 _fingerprint_secret: str = getattr( 

264 _auth_cfg, "session_secret", "change-me-in-production" 

265 ) 

266 if isinstance(_fingerprint_secret, SecretStr): 

267 _fingerprint_secret = _fingerprint_secret.get_secret_value() 

268 

269 @inject 

270 class _AdminSessionServiceConfigured(AdminSessionService): 

271 """Admin-scoped SessionService with config-driven lifetimes.""" 

272 

273 def __init__( 

274 self, 

275 session_repo: SessionRepositoryProtocol, 

276 ) -> None: 

277 super().__init__( 

278 session_repo=session_repo, 

279 session_lifetime=_session_lifetime, 

280 idle_timeout=_idle_timeout, 

281 fingerprint_secret=_fingerprint_secret, 

282 ) 

283 

284 container.singleton(AdminSessionServiceProtocol, _AdminSessionServiceConfigured) 

285 

286 # ── AdminLoginAttemptService — DI-wired; cache wired in boot() ──── 

287 # The @inject decorator resolves attempt_store and lockout_store from 

288 # the container. CacheBackendProtocol is optional (defaults to None); 

289 # it is wired post-registration in boot() when cache is available. 

290 container.singleton(AdminLoginAttemptServiceProtocol, AdminLoginAttemptService) 

291 

292 # ── AdminAuthService — DI-wired orchestrator ────────────────────── 

293 # The @inject decorator resolves all store/service deps from the 

294 # container; mfa_factor comes from config via the closure. 

295 

296 @inject 

297 class _AdminAuthServiceConfigured(AdminAuthService): 

298 """Admin-scoped auth service with config-driven factor selection.""" 

299 

300 def __init__( 

301 self, 

302 user_store: AdminUserStoreProtocol, 

303 attempt_service: AdminLoginAttemptServiceProtocol, 

304 audit_service: AdminAuditLogServiceProtocol, 

305 session_service: AdminSessionServiceProtocol, 

306 mfa_service: AdminMfaServiceProtocol | None = None, 

307 email_verification_service: AdminEmailVerificationServiceProtocol 

308 | None = None, 

309 email_otp_service: AdminEmailOtpServiceProtocol | None = None, 

310 ) -> None: 

311 super().__init__( 

312 user_store=user_store, 

313 attempt_service=attempt_service, 

314 audit_service=audit_service, 

315 session_service=session_service, 

316 mfa_service=mfa_service, 

317 email_verification_service=email_verification_service, 

318 email_otp_service=email_otp_service, 

319 mfa_factor=getattr(_mfa_cfg, "factor", "totp"), 

320 ) 

321 

322 container.singleton(AdminAuthServiceProtocol, _AdminAuthServiceConfigured) 

323 

324 # ── AdminPasswordResetService — DI-wired orchestrator ───────────── 

325 # Resolves user_store, token_store, audit, auth, and policy services 

326 # from the container. hasher and notification_service stay optional: 

327 # the hasher falls back to lexigram-auth's PasswordHasher at runtime, 

328 # and notifications are skipped when no service is bound. 

329 container.singleton( 

330 AdminPasswordResetServiceProtocol, AdminPasswordResetService 

331 ) 

332 

333 # ── AdminMfaService — @inject subclass captures config values ──── 

334 # Follows the _AdminSessionServiceConfigured pattern: the inner class 

335 # passes MFA config from the closure while the container injects the 

336 # store and audit service. 

337 from lexigram.admin.auth.services.mfa_service import AdminMfaService 

338 from lexigram.admin.auth.store.mfa_sql import AdminMfaSqlStore 

339 from lexigram.admin.config import AdminMfaConfig 

340 from lexigram.security.encryption import EncryptionService 

341 

342 _mfa_encryption = EncryptionService(secret_key=_session_secret) 

343 

344 @inject 

345 class _AdminMfaSqlStoreConfigured(AdminMfaSqlStore): 

346 """Admin-scoped MFA store with config-derived secret encryption.""" 

347 

348 def __init__(self, db: DatabaseProviderProtocol) -> None: 

349 super().__init__(db=db, encryption_service=_mfa_encryption) 

350 

351 container.singleton(AdminMfaStoreProtocol, _AdminMfaSqlStoreConfigured) 

352 

353 @inject 

354 class _AdminMfaServiceConfigured(AdminMfaService): 

355 """Admin-scoped MFA service with config-driven settings.""" 

356 

357 def __init__( 

358 self, 

359 store: AdminMfaStoreProtocol, 

360 audit_service: AdminAuditLogServiceProtocol, 

361 ) -> None: 

362 super().__init__( 

363 config=_mfa_cfg or AdminMfaConfig(), 

364 store=store, 

365 audit_service=audit_service, 

366 ) 

367 

368 container.singleton(AdminMfaServiceProtocol, _AdminMfaServiceConfigured) 

369 

370 # ── Email verification + email OTP — configured subclasses ──────── 

371 # Notification service is registered as a concrete singleton so the 

372 # email services can inject it; it no-ops when no mailer is bound 

373 # (fail-open for verification, Err for OTP delivery). 

374 from lexigram.admin.auth.services.email_otp_service import AdminEmailOtpService 

375 from lexigram.admin.auth.services.email_verification_service import ( 

376 AdminEmailVerificationService, 

377 ) 

378 from lexigram.admin.auth.store.email_otp_sql import AdminEmailOtpSqlStore 

379 from lexigram.admin.auth.store.email_verification_sql import ( 

380 AdminEmailVerificationSqlStore, 

381 ) 

382 from lexigram.admin.config import ( 

383 AdminEmailOtpConfig, 

384 AdminEmailVerificationConfig, 

385 ) 

386 

387 container.singleton(AdminNotificationService, AdminNotificationService) 

388 

389 container.singleton( 

390 AdminEmailVerificationStoreProtocol, AdminEmailVerificationSqlStore 

391 ) 

392 container.singleton(AdminEmailOtpStoreProtocol, AdminEmailOtpSqlStore) 

393 

394 @inject 

395 class _AdminEmailVerificationServiceConfigured(AdminEmailVerificationService): 

396 """Admin-scoped email verification service with config-driven settings.""" 

397 

398 def __init__( 

399 self, 

400 store: AdminEmailVerificationStoreProtocol, 

401 notification_service: AdminNotificationService | None = None, 

402 audit_service: AdminAuditLogServiceProtocol | None = None, 

403 ) -> None: 

404 super().__init__( 

405 config=_email_verif_cfg or AdminEmailVerificationConfig(), 

406 store=store, 

407 notification_service=notification_service, 

408 audit_service=audit_service, 

409 ) 

410 

411 container.singleton( 

412 AdminEmailVerificationServiceProtocol, 

413 _AdminEmailVerificationServiceConfigured, 

414 ) 

415 

416 @inject 

417 class _AdminEmailOtpServiceConfigured(AdminEmailOtpService): 

418 """Admin-scoped email OTP service with config-driven settings.""" 

419 

420 def __init__( 

421 self, 

422 store: AdminEmailOtpStoreProtocol, 

423 notification_service: AdminNotificationService | None = None, 

424 audit_service: AdminAuditLogServiceProtocol | None = None, 

425 ) -> None: 

426 super().__init__( 

427 config=_email_otp_cfg or AdminEmailOtpConfig(), 

428 store=store, 

429 notification_service=notification_service, 

430 audit_service=audit_service, 

431 ) 

432 

433 container.singleton( 

434 AdminEmailOtpServiceProtocol, _AdminEmailOtpServiceConfigured 

435 ) 

436 

437 # ── AdminRoleService — DI-wired RBAC orchestrator ──────────────── 

438 # The @inject decorator resolves role_store from the container. 

439 # authorization_service and audit_service are optional (None when 

440 # unbound — the service skips mirror/audit, fail open). 

441 from lexigram.admin.rbac.protocols import ( 

442 AdminRoleServiceProtocol, 

443 AdminRoleStoreProtocol, 

444 ) 

445 from lexigram.admin.rbac.role_service import AdminRoleService 

446 from lexigram.admin.rbac.roles_sql import AdminRoleSqlStore 

447 

448 container.singleton(AdminRoleStoreProtocol, AdminRoleSqlStore) 

449 container.singleton(AdminRoleServiceProtocol, AdminRoleService) 

450 

451 logger.debug("admin_auth.new_services_registered") 

452 

453 async def boot(self, container: ContainerResolverProtocol) -> None: 

454 """Boot auth services: initialize guard chain, session management. 

455 

456 Attempts to initialize the schema for every new SQL store. Failures 

457 are logged at WARNING level and never re-raised so that a missing 

458 table does not prevent the admin panel from starting. 

459 

460 Also wires the optional ``CacheBackendProtocol`` into 

461 ``AdminLoginAttemptService`` when the cache provider is available. 

462 

463 Args: 

464 container: The container resolver (container is frozen at this point). 

465 """ 

466 self._initialized = True 

467 

468 # ── Schema initialization for new auth stores ───────────────────── 

469 from lexigram.admin.auth.protocols import ( 

470 AdminAccountLockoutStoreProtocol, 

471 AdminAuditLogStoreProtocol, 

472 AdminEmailOtpStoreProtocol, 

473 AdminEmailVerificationServiceProtocol, 

474 AdminEmailVerificationStoreProtocol, 

475 AdminLoginAttemptServiceProtocol, 

476 AdminLoginAttemptStoreProtocol, 

477 AdminMfaStoreProtocol, 

478 AdminPasswordResetServiceProtocol, 

479 AdminPasswordResetTokenStoreProtocol, 

480 ) 

481 from lexigram.admin.rbac.protocols import AdminRoleStoreProtocol 

482 

483 for _store_protocol in ( 

484 AdminLoginAttemptStoreProtocol, 

485 AdminAccountLockoutStoreProtocol, 

486 AdminAuditLogStoreProtocol, 

487 AdminPasswordResetTokenStoreProtocol, 

488 AdminMfaStoreProtocol, 

489 AdminRoleStoreProtocol, 

490 AdminEmailVerificationStoreProtocol, 

491 AdminEmailOtpStoreProtocol, 

492 ): 

493 try: 

494 _store = await container.resolve( 

495 _store_protocol, bypass_visibility=True 

496 ) 

497 await _store.ensure_schema() # type: ignore[attr-defined] 

498 except Exception as e: 

499 logger.exception(f"admin_auth.schema_init_failed: {e}") # noqa: BLE001 

500 logger.warning( 

501 "admin_auth.schema_init_failed", 

502 protocol=str(_store_protocol), 

503 ) 

504 

505 # ── Wire cache into AdminLoginAttemptService (optional) ─────────── 

506 try: 

507 from lexigram.contracts.infra.cache import CacheBackendProtocol 

508 

509 _cache = await container.resolve(CacheBackendProtocol) 

510 _attempt_svc = await container.resolve( 

511 AdminLoginAttemptServiceProtocol, bypass_visibility=True 

512 ) 

513 if hasattr(_attempt_svc, "_cache"): 

514 _attempt_svc._cache = _cache 

515 logger.debug("admin_auth.cache_wired") 

516 except Exception: 

517 logger.debug("admin_auth.cache_not_available") 

518 

519 # ── Wire cache into AdminEmailVerificationService (optional) ───── 

520 try: 

521 from lexigram.contracts.infra.cache import CacheBackendProtocol 

522 

523 _cache = await container.resolve(CacheBackendProtocol) 

524 _verif_svc = await container.resolve( 

525 AdminEmailVerificationServiceProtocol, bypass_visibility=True 

526 ) 

527 if hasattr(_verif_svc, "_cache"): 

528 _verif_svc._cache = _cache 

529 logger.debug("admin_auth.verification_cache_wired") 

530 except Exception: 

531 logger.debug("admin_auth.verification_cache_not_available") 

532 

533 # ── Wire cache into AdminPasswordResetService (optional) ────────── 

534 try: 

535 from lexigram.contracts.infra.cache import CacheBackendProtocol 

536 

537 _cache = await container.resolve(CacheBackendProtocol) 

538 _reset_svc = await container.resolve( 

539 AdminPasswordResetServiceProtocol, bypass_visibility=True 

540 ) 

541 if hasattr(_reset_svc, "_cache"): 

542 _reset_svc._cache = _cache 

543 logger.debug("admin_auth.reset_cache_wired") 

544 except Exception: 

545 logger.debug("admin_auth.reset_cache_not_available") 

546 

547 async def shutdown(self) -> None: 

548 """Shut down auth services.""" 

549 self._initialized = False 

550 

551 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

552 """Return auth infrastructure health status.""" 

553 return HealthCheckResult( 

554 component="admin_auth", 

555 status=HealthStatus.HEALTHY if self._initialized else HealthStatus.UNKNOWN, 

556 message="Admin auth operational" 

557 if self._initialized 

558 else "Not yet initialized", 

559 ) 

560 

561 

562__all__ = ["AdminAuthSubProvider"]