Coverage for src / lexigram / admin / auth / services / delegating_auth_adapter.py: 0%
82 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +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.integration import AdminUser
24from lexigram.contracts import AuthenticatedUserProtocol
25from lexigram.logging import get_logger
26from lexigram.result import Err, Ok, Result
28logger = get_logger(__name__)
31# ---------------------------------------------------------------------------
32# Protocols for capabilities we delegate INTO admin's existing services
33# ---------------------------------------------------------------------------
36class AdminRateLimiterProtocol(Protocol):
37 """Enforces IP-based rate limiting for admin logins."""
39 async def check_ip_rate_limit(self, ip_address: str) -> None: ...
40 async def record_attempt(
41 self, email: str, ip_address: str, success: bool, **metadata: Any
42 ) -> None: ...
45class AdminLockoutProtocol(Protocol):
46 """Enforces per-account lockout for admin logins."""
48 async def check_account_lockout(self, email: str) -> None: ...
49 async def clear_lockout(self, email: str) -> None: ...
52class AdminAuditProtocol(Protocol):
53 """Records admin security events (fire-and-forget)."""
55 async def log_login_success(
56 self, user_id: str, email: str, ip_address: str, **metadata: Any
57 ) -> None: ...
58 async def log_login_failure(
59 self, email: str, ip_address: str, reason: str, **metadata: Any
60 ) -> None: ...
61 async def log_logout(self, session_id: str, user_id: str | None = None) -> None: ...
64# ---------------------------------------------------------------------------
65# Wrapper types for auth results
66# ---------------------------------------------------------------------------
69@dataclass
70class AdminAuthResult:
71 """Result of a successful admin authentication."""
73 user: AdminUser
74 session_id: str
75 expires_at: datetime
78# ---------------------------------------------------------------------------
79# The adapter
80# ---------------------------------------------------------------------------
83class DelegatingAuthAdapter:
84 """Wraps lexigram-auth services for admin authentication and session mgmt.
86 This adapter is the bridge between admin's security pipeline and the
87 framework's canonical auth services. Admin-internal concerns (rate
88 limiting, lockout, auditing) are handled through injected admin
89 protocol implementations, while credential verification and session
90 lifecycle are delegated to lexigram-auth.
92 The adapter is designed for construction via DI::
94 adapter = DelegatingAuthAdapter(
95 auth_service=authentication_service,
96 cookie_backend=session_cookie_backend,
97 rate_limiter=admin_rate_limiter,
98 lockout=admin_lockout_service,
99 audit=admin_audit_service,
100 )
101 result = await adapter.authenticate_admin("admin@x.com", "pw", "127.0.0.1")
102 """
104 def __init__(
105 self,
106 auth_service: object | None = None,
107 cookie_backend: object | None = None,
108 rate_limiter: AdminRateLimiterProtocol | None = None,
109 lockout: AdminLockoutProtocol | None = None,
110 audit: AdminAuditProtocol | None = None,
111 session_lifetime: int = 86400,
112 ) -> None:
113 self._auth_service = auth_service
114 self._cookie_backend = cookie_backend
115 self._rate_limiter = rate_limiter
116 self._lockout = lockout
117 self._audit = audit
118 self._session_lifetime = session_lifetime
120 async def authenticate_admin(
121 self,
122 email: str,
123 password: str,
124 ip_address: str,
125 user_agent: str = "",
126 ) -> Result[AdminAuthResult, str]:
127 """Authenticate an admin user via lexigram-auth.
129 Pipeline:
130 1. IP rate-limit check (admin concern).
131 2. Account lockout check (admin concern).
132 3. Credential verification via ``AuthenticationService.authenticate_user``.
133 4. Clear lockout on success.
134 5. Create session via ``SessionCookieBackend.login``.
135 6. Audit logging (admin concern).
137 Returns:
138 Ok(AdminAuthResult) on success, Err(str) on failure.
139 """
140 # Step 1 — IP rate limit
141 if self._rate_limiter is not None:
142 try:
143 await self._rate_limiter.check_ip_rate_limit(ip_address)
144 except Exception as exc:
145 if self._audit is not None:
146 await self._audit.log_login_failure(
147 email=email,
148 ip_address=ip_address,
149 reason="ip_rate_limited",
150 )
151 logger.warning(
152 "admin_auth_ip_rate_limited",
153 ip_address=ip_address,
154 email=email,
155 )
156 return Err(str(exc))
158 # Step 2 — Account lockout
159 if self._lockout is not None:
160 try:
161 await self._lockout.check_account_lockout(email)
162 except Exception as exc:
163 if self._audit is not None:
164 await self._audit.log_login_failure(
165 email=email,
166 ip_address=ip_address,
167 reason="account_locked",
168 )
169 logger.warning(
170 "admin_auth_account_locked",
171 email=email,
172 ip_address=ip_address,
173 )
174 return Err(str(exc))
176 # Step 3 — Credential verification via lexigram-auth
177 if self._auth_service is None:
178 return Err("Authentication service unavailable")
179 result = await self._auth_service.authenticate_user(email, password) # type: ignore[attr-defined]
180 if result.is_err():
181 if self._audit is not None:
182 await self._audit.log_login_failure(
183 email=email,
184 ip_address=ip_address,
185 reason="invalid_credentials",
186 )
187 logger.info("admin_auth_failure", email=email, ip_address=ip_address)
188 return Err("Invalid email or password.")
190 framework_user: AuthenticatedUserProtocol = result.unwrap()
192 # Step 4 — Clear lockout on success
193 if self._lockout is not None:
194 await self._lockout.clear_lockout(email)
196 # Step 5 — Wrap framework user in AdminUser (AUTH-11: delegate, not copy)
197 admin_user = AdminUser(
198 id=framework_user.user_id,
199 email=framework_user.email,
200 name=framework_user.name,
201 framework_user=framework_user,
202 )
204 # Step 6 — Audit success
205 if self._audit is not None:
206 await self._audit.log_login_success(
207 user_id=framework_user.user_id,
208 email=framework_user.email,
209 ip_address=ip_address,
210 )
212 logger.info(
213 "admin_auth_success",
214 user_id=framework_user.user_id,
215 email=framework_user.email,
216 )
218 expires_at = datetime.now(UTC) + timedelta(seconds=self._session_lifetime)
220 return Ok(
221 AdminAuthResult(
222 user=admin_user,
223 session_id=framework_user.user_id,
224 expires_at=expires_at,
225 )
226 )
228 async def login(self, response: Any, user_id: str, expires_in: int = 86400) -> str:
229 """Set a session cookie on the HTTP response.
231 Delegates to ``SessionCookieBackend.login``.
233 Args:
234 response: Starlette ``Response`` to set cookie on.
235 user_id: User identifier to associate with the session.
236 expires_in: Session TTL in seconds.
238 Returns:
239 The created ``session_id``.
240 """
241 return await self._cookie_backend.login(response, user_id, expires_in) # type: ignore[union-attr]
243 async def logout(self, request: Any, response: Any) -> None:
244 """Revoke the session and clear the cookie.
246 Delegates to ``SessionCookieBackend.logout``.
248 Args:
249 request: Incoming ``Request`` containing the session cookie.
250 response: Outgoing ``Response`` to clear the cookie on.
251 """
252 await self._cookie_backend.logout(request, response) # type: ignore[union-attr]
254 if self._audit is not None:
255 session_id = request.cookies.get(
256 getattr(self._cookie_backend, "cookie_name", "session_id")
257 )
258 await self._audit.log_logout(
259 session_id=session_id or "unknown",
260 )
262 async def resolve_admin_user(self, request: Any) -> AdminUser | None:
263 """Resolve the currently authenticated admin user from a request.
265 Reads the session cookie, validates the session via
266 ``SessionCookieBackend.authenticate``, and wraps the framework user
267 in an ``AdminUser`` with delegation enabled.
269 Args:
270 request: The incoming Starlette ``Request``.
272 Returns:
273 An ``AdminUser`` with ``framework_user`` set, or ``None``.
274 """
275 framework_user = await self._cookie_backend.authenticate(request) # type: ignore[union-attr]
276 if framework_user is None:
277 return None
279 return self._wrap_framework_user(framework_user)
281 async def resolve_admin_user_from_scope(self, scope: Scope) -> AdminUser | None:
282 """Resolve the authenticated admin user from an ASGI scope.
284 Lightweight variant of :meth:`resolve_admin_user` that builds a
285 minimal Starlette ``Request`` from the raw ASGI scope so that
286 ``AdminAuthGuardMiddleware`` can call it without the overhead of
287 constructing a full Request object.
289 Args:
290 scope: ASGI connection scope (``type == "http"``).
292 Returns:
293 An ``AdminUser`` with ``framework_user`` set, or ``None``.
294 """
295 if scope.get("type") != "http":
296 return None
298 request = StarletteRequest(scope)
299 return await self.resolve_admin_user(request)
301 # -----------------------------------------------------------------------
302 # Internal helpers
303 # -----------------------------------------------------------------------
305 @staticmethod
306 def _wrap_framework_user(user: AuthenticatedUserProtocol) -> AdminUser:
307 """Wrap a framework ``AuthenticatedUserProtocol`` in an ``AdminUser``.
309 Properties are delegated through ``framework_user`` (AUTH-11) instead
310 of being eagerly copied from the source.
311 """
312 return AdminUser(
313 id=user.user_id,
314 email=user.email,
315 name=user.name,
316 framework_user=user,
317 )
320__all__ = [
321 "AdminAuditProtocol",
322 "AdminAuthResult",
323 "AdminLockoutProtocol",
324 "AdminRateLimiterProtocol",
325 "DelegatingAuthAdapter",
326]