Coverage for src/lexigram/admin/auth/services/delegating_auth_adapter.py: 95%
83 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Delegating auth adapter — wraps lexigram-auth services for admin use.
3This adapter implements the same auth patterns admin currently uses
4(``authenticate_admin``, ``resolve_admin_user``, session lifecycle) but
5delegates the heavy lifting to ``lexigram.auth``'s ``AuthenticationService``
6and ``SessionCookieBackend``.
8Admin-specific concerns (rate limiting, lockout, audit logging) that live
9outside lexigram-auth are handled through injected protocol implementations
10passed to this adapter, ensuring the adapter itself has no direct dependency
11on admin-internal services.
12"""
14from __future__ import annotations
16from dataclasses import dataclass
17from datetime import UTC, datetime, timedelta
18from typing import Any, Protocol
20from starlette.requests import Request as StarletteRequest
21from starlette.types import Scope
23from lexigram.admin.auth.errors import (
24 AccountLockedError,
25 AdminAuthError,
26 InvalidCredentialsError,
27 RateLimitExceededError,
28)
29from lexigram.admin.auth.integration import AdminUser
30from lexigram.contracts import AuthenticatedUserProtocol
31from lexigram.logging import get_logger
32from lexigram.result import Err, Ok, Result
34logger = get_logger(__name__)
37# ---------------------------------------------------------------------------
38# Protocols for capabilities we delegate INTO admin's existing services
39# ---------------------------------------------------------------------------
42class AdminRateLimiterProtocol(Protocol):
43 """Enforces IP-based rate limiting for admin logins."""
45 async def check_ip_rate_limit(self, ip_address: str) -> None: ...
46 async def record_attempt(
47 self, email: str, ip_address: str, success: bool, **metadata: Any
48 ) -> None: ...
51class AdminLockoutProtocol(Protocol):
52 """Enforces per-account lockout for admin logins."""
54 async def check_account_lockout(self, email: str) -> None: ...
55 async def clear_lockout(self, email: str) -> None: ...
58class AdminAuditProtocol(Protocol):
59 """Records admin security events (fire-and-forget)."""
61 async def log_login_success(
62 self, user_id: str, email: str, ip_address: str, **metadata: Any
63 ) -> None: ...
64 async def log_login_failure(
65 self, email: str, ip_address: str, reason: str, **metadata: Any
66 ) -> None: ...
67 async def log_logout(self, session_id: str, user_id: str | None = None) -> None: ...
70# ---------------------------------------------------------------------------
71# Wrapper types for auth results
72# ---------------------------------------------------------------------------
75@dataclass
76class AdminAuthResult:
77 """Result of a successful admin authentication."""
79 user: AdminUser
80 session_id: str
81 expires_at: datetime
84# ---------------------------------------------------------------------------
85# The adapter
86# ---------------------------------------------------------------------------
89class DelegatingAuthAdapter:
90 """Wraps lexigram-auth services for admin authentication and session mgmt.
92 This adapter is the bridge between admin's security pipeline and the
93 framework's canonical auth services. Admin-internal concerns (rate
94 limiting, lockout, auditing) are handled through injected admin
95 protocol implementations, while credential verification and session
96 lifecycle are delegated to lexigram-auth.
98 The adapter is designed for construction via DI::
100 adapter = DelegatingAuthAdapter(
101 auth_service=authentication_service,
102 cookie_backend=session_cookie_backend,
103 rate_limiter=admin_rate_limiter,
104 lockout=admin_lockout_service,
105 audit=admin_audit_service,
106 )
107 result = await adapter.authenticate_admin("admin@x.com", "pw", "127.0.0.1")
108 """
110 def __init__(
111 self,
112 auth_service: object | None = None,
113 cookie_backend: object | None = None,
114 rate_limiter: AdminRateLimiterProtocol | None = None,
115 lockout: AdminLockoutProtocol | None = None,
116 audit: AdminAuditProtocol | None = None,
117 session_lifetime: int = 86400,
118 ) -> None:
119 self._auth_service = auth_service
120 self._cookie_backend = cookie_backend
121 self._rate_limiter = rate_limiter
122 self._lockout = lockout
123 self._audit = audit
124 self._session_lifetime = session_lifetime
126 async def authenticate_admin(
127 self,
128 email: str,
129 password: str,
130 ip_address: str,
131 user_agent: str = "",
132 ) -> Result[AdminAuthResult, AdminAuthError]:
133 """Authenticate an admin user via lexigram-auth.
135 Pipeline:
136 1. IP rate-limit check (admin concern).
137 2. Account lockout check (admin concern).
138 3. Credential verification via ``AuthenticationService.authenticate_user``.
139 4. Clear lockout on success.
140 5. Create session via ``SessionCookieBackend.login``.
141 6. Audit logging (admin concern).
143 Returns:
144 ``Ok(AdminAuthResult)`` on success, or ``Err(AdminAuthError)``
145 when rate-limited, locked out, or credentials are invalid.
146 """
147 # Step 1 — IP rate limit
148 if self._rate_limiter is not None:
149 try:
150 await self._rate_limiter.check_ip_rate_limit(ip_address)
151 except RateLimitExceededError as exc:
152 if self._audit is not None:
153 await self._audit.log_login_failure(
154 email=email,
155 ip_address=ip_address,
156 reason="ip_rate_limited",
157 )
158 logger.warning(
159 "admin_auth_ip_rate_limited",
160 ip_address=ip_address,
161 email=email,
162 )
163 return Err(exc)
165 # Step 2 — Account lockout
166 if self._lockout is not None:
167 try:
168 await self._lockout.check_account_lockout(email)
169 except AccountLockedError as exc:
170 if self._audit is not None:
171 await self._audit.log_login_failure(
172 email=email,
173 ip_address=ip_address,
174 reason="account_locked",
175 )
176 logger.warning(
177 "admin_auth_account_locked",
178 email=email,
179 ip_address=ip_address,
180 )
181 return Err(exc)
183 # Step 3 — Credential verification via lexigram-auth
184 if self._auth_service is None:
185 return Err(AdminAuthError("Authentication service unavailable"))
186 result = await self._auth_service.authenticate_user(email, password) # type: ignore[attr-defined]
187 if result.is_err():
188 if self._audit is not None:
189 await self._audit.log_login_failure(
190 email=email,
191 ip_address=ip_address,
192 reason="invalid_credentials",
193 )
194 logger.info("admin_auth_failure", email=email, ip_address=ip_address)
195 return Err(InvalidCredentialsError("Invalid email or password."))
197 framework_user: AuthenticatedUserProtocol = result.unwrap()
199 # Step 4 — Clear lockout on success
200 if self._lockout is not None:
201 await self._lockout.clear_lockout(email)
203 # Step 5 — Wrap framework user in AdminUser (AUTH-11: delegate, not copy)
204 admin_user = AdminUser(
205 id=framework_user.user_id,
206 email=framework_user.email,
207 name=framework_user.name,
208 framework_user=framework_user,
209 )
211 # Step 6 — Audit success
212 if self._audit is not None:
213 await self._audit.log_login_success(
214 user_id=framework_user.user_id,
215 email=framework_user.email,
216 ip_address=ip_address,
217 )
219 logger.info(
220 "admin_auth_success",
221 user_id=framework_user.user_id,
222 email=framework_user.email,
223 )
225 expires_at = datetime.now(UTC) + timedelta(seconds=self._session_lifetime)
227 return Ok(
228 AdminAuthResult(
229 user=admin_user,
230 session_id=framework_user.user_id,
231 expires_at=expires_at,
232 )
233 )
235 async def login(self, response: Any, user_id: str, expires_in: int = 86400) -> str:
236 """Set a session cookie on the HTTP response.
238 Delegates to ``SessionCookieBackend.login``.
240 Args:
241 response: Starlette ``Response`` to set cookie on.
242 user_id: User identifier to associate with the session.
243 expires_in: Session TTL in seconds.
245 Returns:
246 The created ``session_id``.
247 """
248 return await self._cookie_backend.login(response, user_id, expires_in) # type: ignore[union-attr]
250 async def logout(self, request: Any, response: Any) -> None:
251 """Revoke the session and clear the cookie.
253 Delegates to ``SessionCookieBackend.logout``.
255 Args:
256 request: Incoming ``Request`` containing the session cookie.
257 response: Outgoing ``Response`` to clear the cookie on.
258 """
259 await self._cookie_backend.logout(request, response) # type: ignore[union-attr]
261 if self._audit is not None:
262 session_id = request.cookies.get(
263 getattr(self._cookie_backend, "cookie_name", "session_id")
264 )
265 await self._audit.log_logout(
266 session_id=session_id or "unknown",
267 )
269 async def resolve_admin_user(self, request: Any) -> AdminUser | None:
270 """Resolve the currently authenticated admin user from a request.
272 Reads the session cookie, validates the session via
273 ``SessionCookieBackend.authenticate``, and wraps the framework user
274 in an ``AdminUser`` with delegation enabled.
276 Args:
277 request: The incoming Starlette ``Request``.
279 Returns:
280 An ``AdminUser`` with ``framework_user`` set, or ``None``.
281 """
282 framework_user = await self._cookie_backend.authenticate(request) # type: ignore[union-attr]
283 if framework_user is None:
284 return None
286 return self._wrap_framework_user(framework_user)
288 async def resolve_admin_user_from_scope(self, scope: Scope) -> AdminUser | None:
289 """Resolve the authenticated admin user from an ASGI scope.
291 Lightweight variant of :meth:`resolve_admin_user` that builds a
292 minimal Starlette ``Request`` from the raw ASGI scope so that
293 ``AdminAuthGuardMiddleware`` can call it without the overhead of
294 constructing a full Request object.
296 Args:
297 scope: ASGI connection scope (``type == "http"``).
299 Returns:
300 An ``AdminUser`` with ``framework_user`` set, or ``None``.
301 """
302 if scope.get("type") != "http":
303 return None
305 request = StarletteRequest(scope)
306 return await self.resolve_admin_user(request)
308 # -----------------------------------------------------------------------
309 # Internal helpers
310 # -----------------------------------------------------------------------
312 @staticmethod
313 def _wrap_framework_user(user: AuthenticatedUserProtocol) -> AdminUser:
314 """Wrap a framework ``AuthenticatedUserProtocol`` in an ``AdminUser``.
316 Properties are delegated through ``framework_user`` (AUTH-11) instead
317 of being eagerly copied from the source.
318 """
319 return AdminUser(
320 id=user.user_id,
321 email=user.email,
322 name=user.name,
323 framework_user=user,
324 )
327__all__ = [
328 "AdminAuditProtocol",
329 "AdminAuthResult",
330 "AdminLockoutProtocol",
331 "AdminRateLimiterProtocol",
332 "DelegatingAuthAdapter",
333]