Coverage for src/lexigram/admin/controllers/profile.py: 27%
106 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1"""User profile controller for the admin interface.
3Renders the authenticated user's profile page inside the admin shell and
4handles password changes. Two-factor authentication management lives on
5the dedicated ``/admin/profile/mfa`` screen (AuthController); this page
6surfaces its status and links to it.
7"""
9from __future__ import annotations
11from secrets import token_hex
12from typing import Any
13from urllib.parse import quote_plus
15from starlette.requests import Request
16from starlette.responses import RedirectResponse, Response
18from lexigram.admin.auth.models import AdminUser
19from lexigram.admin.auth.protocols import (
20 AdminAuditLogServiceProtocol,
21 AdminCsrfServiceProtocol,
22 AdminMfaServiceProtocol,
23)
24from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
25from lexigram.admin.auth.types import AdminSecurityEventType
26from lexigram.admin.controllers.base import AdminController
27from lexigram.admin.engine.renderer import AdminRenderer
28from lexigram.admin.lib.template import render_profile_page
29from lexigram.contracts.web import get, post
30from lexigram.logging import get_logger
32logger = get_logger(__name__)
34__all__ = ["ProfileController"]
37class ProfileController(AdminController):
38 """Profile management controller.
40 Routes:
41 GET /admin/profile - Profile page (admin shell)
42 POST /admin/profile/password - Change the current user's password
43 """
45 prefix = "/profile"
47 def __init__(
48 self,
49 renderer: AdminRenderer,
50 csrf_service: AdminCsrfServiceProtocol | None = None,
51 mfa_service: AdminMfaServiceProtocol | None = None,
52 ) -> None:
53 """Initialise the profile controller.
55 Args:
56 renderer: AdminRenderer for shell page rendering.
57 csrf_service: Optional CSRF token service.
58 mfa_service: Optional 2FA service; ``None`` hides MFA state.
59 """
60 super().__init__(renderer=renderer)
61 self._csrf_service = csrf_service
62 self._mfa_service = mfa_service
63 self._user_store: AdminUserStoreProtocol | None = None
65 # -- helpers --
67 def _csrf_token(self, request: Request) -> str:
68 """Return a CSRF token, creating and persisting the session id."""
69 csrf_session_id = str(request.session.get("csrf_session_id", ""))
70 if not csrf_session_id:
71 csrf_session_id = token_hex(16)
72 request.session["csrf_session_id"] = csrf_session_id
73 if self._csrf_service is None:
74 return ""
75 return self._csrf_service.generate_token(csrf_session_id)
77 def _csrf_ok(self, request: Request, csrf_token: str) -> bool:
78 """Validate the submitted CSRF token against the session id.
80 Args:
81 request: Incoming HTTP request (session already loaded).
82 csrf_token: Token submitted with the form.
84 Returns:
85 ``True`` when the token matches the session.
86 """
87 if self._csrf_service is None:
88 return True
89 csrf_session_id = str(request.session.get("csrf_session_id", ""))
90 return bool(
91 csrf_session_id
92 and self._csrf_service.validate_token(csrf_session_id, csrf_token)
93 )
95 @staticmethod
96 def _redirect(url: str, message: str, is_error: bool = False) -> Response:
97 """Return a 302 redirect carrying an error or notice flash message."""
98 key = "error" if is_error else "notice"
99 return RedirectResponse(
100 url=f"{url}?{key}={quote_plus(message)}",
101 status_code=302,
102 )
104 async def _audit(
105 self,
106 request: Request,
107 event_type: AdminSecurityEventType,
108 success: bool,
109 **metadata: Any,
110 ) -> None:
111 """Append a security audit event, best-effort."""
112 try:
113 container = getattr(request.state, "container", None)
114 if container is None:
115 return
116 audit_service = await container.resolve(
117 AdminAuditLogServiceProtocol,
118 )
119 client = getattr(request, "client", None)
120 await audit_service.log_event(
121 event_type=event_type,
122 ip_address=getattr(client, "host", "unknown"),
123 user_agent=request.headers.get("user-agent", "") or "",
124 success=success,
125 metadata=metadata,
126 )
127 except Exception: # noqa: BLE001 — audit failures must not break requests
128 logger.warning("profile.audit_failed", event_type=event_type.value)
130 # -- routes --
132 @get("/")
133 async def profile_page(self, request: Request) -> Response:
134 """Render the current user's profile page inside the admin shell.
136 Args:
137 request: Incoming HTTP request (authenticated).
139 Returns:
140 HTMLResponse with the profile page, or a redirect to login
141 when unauthenticated.
142 """
143 user: AdminUser = self.current_user(request)
144 if not user or user.user_id == "guest":
145 return RedirectResponse(
146 url="/admin/login?next=/admin/profile", status_code=302
147 )
149 mfa_enabled = False
150 if self._mfa_service is not None:
151 try:
152 mfa_enabled = await self._mfa_service.is_enabled(str(user.user_id))
153 except Exception: # noqa: BLE001 — non-fatal for page rendering
154 logger.warning("profile.mfa_status_failed")
156 html = render_profile_page(
157 name=user.name or user.email,
158 email=user.email,
159 roles=list(getattr(user, "roles", None) or ()),
160 user_id=str(user.user_id),
161 mfa_enabled=mfa_enabled,
162 csrf_token=self._csrf_token(request),
163 current_password_err=str(
164 request.query_params.get("current_password_err", "")
165 ),
166 new_password_err=str(request.query_params.get("new_password_err", "")),
167 confirmation_err=str(request.query_params.get("confirmation_err", "")),
168 )
169 from lexigram.admin.state.context import AdminContextManager
171 async with AdminContextManager(request) as ctx:
172 error = request.query_params.get("error", "")
173 notice = request.query_params.get("notice", "")
174 if error:
175 ctx.add_flash(error, "error")
176 if notice:
177 ctx.add_flash(notice, "success")
178 return await self.render_admin(
179 request,
180 html,
181 title="Profile",
182 breadcrumbs=self.generate_breadcrumbs(
183 ("Home", "/admin/"),
184 current="Profile",
185 ),
186 )
188 @post("/password")
189 async def password_submit(self, request: Request) -> Response:
190 """Change the current user's password.
192 Validates CSRF, verifies the current password, and persists the new
193 hash through the admin user store. The session is left intact.
195 Args:
196 request: Incoming HTTP request carrying form data.
198 Returns:
199 RedirectResponse back to the profile page with a notice or
200 error flash message.
201 """
202 user: AdminUser = self.current_user(request)
203 if not user or user.user_id == "guest":
204 return RedirectResponse(
205 url="/admin/login?next=/admin/profile", status_code=302
206 )
207 if self._user_store is None:
208 return self._redirect(
209 "/admin/profile", "Password change is unavailable.", True
210 )
212 form = request.scope.get("admin_form_data") or await request.form()
213 csrf_token = str(form.get("csrf_token", ""))
214 current = str(form.get("current_password", ""))
215 new_password = str(form.get("new_password", ""))
216 confirmation = str(form.get("new_password_confirmation", ""))
218 if not self._csrf_ok(request, csrf_token):
219 return self._redirect(
220 "/admin/profile",
221 "Invalid or expired security token. Please try again.",
222 True,
223 )
224 if not current or not new_password or not confirmation:
225 params = [
226 p
227 for p in (
228 "current_password_err=Current%20password%20is%20required."
229 if not current
230 else "",
231 "new_password_err=New%20password%20is%20required."
232 if not new_password
233 else "",
234 "confirmation_err=Please%20confirm%20the%20new%20password."
235 if not confirmation
236 else "",
237 )
238 if p
239 ]
240 return RedirectResponse(
241 url=f"/admin/profile?{'&'.join(params)}",
242 status_code=302,
243 )
244 if len(new_password) < 8:
245 return RedirectResponse(
246 url="/admin/profile?new_password_err="
247 + quote_plus("New password must be at least 8 characters."),
248 status_code=302,
249 )
250 if new_password != confirmation:
251 return RedirectResponse(
252 url="/admin/profile?confirmation_err="
253 + quote_plus("New passwords do not match."),
254 status_code=302,
255 )
257 authenticated = await self._user_store.authenticate(user.email, current)
258 if authenticated is None:
259 logger.warning("profile.password_verify_failed", email=user.email)
260 return self._redirect(
261 "/admin/profile", "Current password is incorrect.", True
262 )
264 from lexigram.admin.lib.password import hash_password
266 record = await self._user_store.get_user_by_email(user.email)
267 if record is None:
268 return self._redirect(
269 "/admin/profile", "Account not found. Please sign in again.", True
270 )
271 record.hashed_password = hash_password(new_password)
272 await self._user_store.update_user(record)
274 await self._audit(
275 request,
276 AdminSecurityEventType.PASSWORD_CHANGED,
277 success=True,
278 email=user.email,
279 )
280 logger.info("profile.password_changed", email=user.email)
281 return self._redirect("/admin/profile", "Password updated successfully.")