Coverage for src/lexigram/admin/controllers/auth/login.py: 21%
112 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1from __future__ import annotations
3import secrets
4from typing import TYPE_CHECKING
5from urllib.parse import quote_plus
7from starlette.requests import Request
8from starlette.responses import HTMLResponse, RedirectResponse
10from lexigram.admin.controllers.auth.core import (
11 _CACHE_CONTROL_NO_STORE,
12 AuthCoreMixin,
13 _humanize_error,
14 logger,
15)
16from lexigram.admin.lib.template import render_login_page
17from lexigram.contracts.web import get, post
19if TYPE_CHECKING:
20 from lexigram.admin.auth.protocols import (
21 AdminAuthServiceProtocol,
22 AdminCsrfServiceProtocol,
23 AdminEmailOtpServiceProtocol,
24 AdminEmailVerificationServiceProtocol,
25 AdminMfaServiceProtocol,
26 )
27 from lexigram.admin.observability.admin_metrics import AdminMetrics
30class AuthLoginMixin(AuthCoreMixin):
31 """AuthController login endpoints."""
33 _auth_service: AdminAuthServiceProtocol
34 _csrf_service: AdminCsrfServiceProtocol
35 _email_otp_service: AdminEmailOtpServiceProtocol | None
36 _email_verification_service: AdminEmailVerificationServiceProtocol | None
37 _metrics: AdminMetrics
38 _mfa_service: AdminMfaServiceProtocol | None
39 _registration_enabled: bool
41 @get("/login")
42 async def login_form(self, request: Request) -> HTMLResponse | RedirectResponse:
43 """Display the standalone login form.
45 Redirects authenticated users to *next_url*. For unauthenticated
46 visitors a fresh CSRF token is embedded in the rendered form.
48 Args:
49 request: Incoming HTTP request.
51 Returns:
52 HTMLResponse with the rendered login page, or a RedirectResponse
53 when the user is already authenticated.
54 """
55 next_url = self._safe_next_url(request.query_params.get("next", "/admin/"))
57 error = _humanize_error(request.query_params.get("error", ""))
58 notice = request.query_params.get("notice", "")
59 email_err = request.query_params.get("email_err", "")
60 password_err = request.query_params.get("password_err", "")
62 user = getattr(request.state, "user", None)
63 if user and user.user_id != "guest" and not error and not notice:
64 return RedirectResponse(
65 url=self._safe_next_url(next_url),
66 status_code=302,
67 headers=_CACHE_CONTROL_NO_STORE,
68 )
70 csrf_session_id = secrets.token_urlsafe(16)
71 request.session["csrf_session_id"] = csrf_session_id
72 csrf_token = self._csrf_service.generate_token(csrf_session_id)
74 html = render_login_page(
75 next_url=next_url,
76 error=error,
77 csrf_token=csrf_token,
78 notice=notice,
79 registration_enabled=self._registration_enabled,
80 email_err=email_err,
81 password_err=password_err,
82 )
83 return HTMLResponse(content=html, headers=_CACHE_CONTROL_NO_STORE)
85 @post("/login")
86 async def login_submit(self, request: Request) -> RedirectResponse:
87 """Process login form submission through the full security pipeline.
89 Validates the CSRF token, resolves the client IP, and delegates
90 credential verification to ``AdminAuthServiceProtocol.authenticate``.
91 On success the session is populated and the user is redirected to
92 *next_url*. On failure the login page is re-shown with a descriptive
93 error message.
95 Args:
96 request: Incoming HTTP request carrying form data.
98 Returns:
99 RedirectResponse to *next_url* on success, or back to the login
100 page with an error query parameter on failure.
101 """
102 form_data = request.scope.get("admin_form_data") or await request.form()
103 email = str(form_data.get("email", ""))
104 password = str(form_data.get("password", ""))
105 next_url = self._safe_next_url(str(form_data.get("next", "/admin/")))
106 csrf_token = str(form_data.get("csrf_token", ""))
108 # ── CSRF validation ────────────────────────────────────────────
109 csrf_session_id = request.session.get("csrf_session_id", "")
110 if not csrf_session_id or not self._csrf_service.validate_token(
111 csrf_session_id, csrf_token
112 ):
113 logger.warning(
114 "auth.csrf_validation_failed", ip=self._get_client_ip(request)
115 )
116 return RedirectResponse(
117 url=f"/admin/login?error={quote_plus('Invalid or expired security token. Please try again.')}&next={quote_plus(next_url)}",
118 status_code=302,
119 )
121 # ── Basic input guard ──────────────────────────────────────────
122 if not email or not password:
123 email_err = "Email is required." if not email else ""
124 password_err = "Password is required." if not password else ""
125 params = [f"next={quote_plus(next_url)}"]
126 if email_err:
127 params.append(f"email_err={quote_plus(email_err)}")
128 if password_err:
129 params.append(f"password_err={quote_plus(password_err)}")
130 return RedirectResponse(
131 url=f"/admin/login?{'&'.join(params)}",
132 status_code=302,
133 )
135 # ── Resolve client context ─────────────────────────────────────
136 ip = self._get_client_ip(request)
137 user_agent = request.headers.get("user-agent", "")
139 logger.info("auth.login_attempt", email=email, ip=ip)
141 result = await self._auth_service.authenticate(email, password, ip, user_agent)
143 if result.is_ok():
144 auth_result = result.unwrap()
145 if auth_result.mfa_required:
146 # 2FA challenge — park the identity in the session; the
147 # /login/2fa flow completes the login with a code. For the
148 # email factor the code is emailed right away.
149 factor = "totp"
150 if self._mfa_service is not None:
151 factor = self._mfa_service.get_factor()
152 request.session["mfa_pending_user_id"] = auth_result.user_id
153 request.session["mfa_pending_email"] = auth_result.email
154 request.session["mfa_pending_roles"] = auth_result.roles
155 request.session["mfa_pending_next"] = next_url
156 request.session["mfa_pending_factor"] = factor
157 if factor == "email" and self._email_otp_service is not None:
158 otp_result = await self._email_otp_service.send_otp(
159 user_id=auth_result.user_id,
160 email=auth_result.email,
161 user_name=auth_result.email,
162 )
163 if otp_result.is_err():
164 logger.warning(
165 "auth.login_email_otp_send_failed",
166 user_id=auth_result.user_id,
167 reason=str(otp_result.unwrap_err()),
168 )
169 logger.info(
170 "auth.login_mfa_challenge",
171 user_id=auth_result.user_id,
172 email=auth_result.email,
173 factor=factor,
174 )
175 return RedirectResponse(url="/admin/login/2fa", status_code=302)
177 if auth_result.email_verification_required:
178 # Verification gate — park the identity so the /verify-email
179 # flow can re-send links, then issue one now.
180 request.session["verify_pending_user_id"] = auth_result.user_id
181 request.session["verify_pending_email"] = auth_result.email
182 request.session["verify_pending_next"] = next_url
183 error_msg = ""
184 if self._email_verification_service is not None:
185 verify_result = (
186 await self._email_verification_service.send_verification(
187 user_id=auth_result.user_id,
188 email=auth_result.email,
189 user_name=auth_result.email,
190 base_url=str(request.base_url),
191 ip_address=ip,
192 )
193 )
194 if verify_result.is_err():
195 error_msg = str(verify_result.unwrap_err())
196 logger.error(
197 "auth.login_verification_send_failed",
198 user_id=auth_result.user_id,
199 reason=error_msg,
200 )
201 logger.info(
202 "auth.login_verification_required",
203 user_id=auth_result.user_id,
204 email=auth_result.email,
205 )
206 target = "/admin/verify-email"
207 if error_msg:
208 target += f"?error={quote_plus(error_msg)}"
209 return RedirectResponse(url=target, status_code=302)
211 request.session["admin_user_id"] = auth_result.user_id
212 request.session["admin_user_email"] = auth_result.email
213 request.session["admin_session_expires_at"] = (
214 auth_result.expires_at.isoformat()
215 )
216 request.session.pop("csrf_session_id", None)
217 if hasattr(auth_result, "session_id"):
218 request.session["session_id"] = auth_result.session_id
219 self._metrics.record_login(status="success")
220 logger.info(
221 "auth.login_success",
222 user_id=auth_result.user_id,
223 email=auth_result.email,
224 redirect=next_url,
225 )
226 from lexigram.admin.state.context import AdminContextManager
228 async with AdminContextManager(request) as ctx:
229 ctx.add_flash("Signed in successfully.", "success")
230 return RedirectResponse(url=self._safe_next_url(next_url), status_code=302)
232 error_msg = _humanize_error(str(result.unwrap_err()))
233 self._metrics.record_login(status="failure")
234 logger.warning("auth.login_failed", email=email, ip=ip, reason=error_msg)
235 return RedirectResponse(
236 url=f"/admin/login?error={quote_plus(error_msg)}&next={quote_plus(next_url)}",
237 status_code=302,
238 )
240 @get("/logout")
241 async def logout(self, request: Request) -> RedirectResponse:
242 """Invalidate the current session and redirect to the login page.
244 Calls ``AdminAuthServiceProtocol.invalidate_session`` before clearing
245 the Starlette session cookie so that any server-side session record is
246 also revoked.
248 Args:
249 request: Incoming HTTP request.
251 Returns:
252 RedirectResponse to /admin/login.
253 """
254 session_id = request.session.get("session_id")
255 if session_id:
256 await self._auth_service.invalidate_session(session_id)
257 logger.info("auth.logout", session_id=session_id)
259 request.session.clear()
260 return RedirectResponse(
261 url="/admin/login?notice=" + quote_plus("You have been signed out."),
262 status_code=302,
263 )