Coverage for src/lexigram/admin/controllers/setup.py: 95%
124 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""First-run setup controller for Lexigram Admin.
3Provides the initial account creation wizard shown when no admin users exist.
4The SetupMiddleware redirects all admin requests here until at least one
5admin account has been created.
6"""
8from __future__ import annotations
10import secrets
11from urllib.parse import quote_plus
13from starlette.requests import Request
14from starlette.responses import HTMLResponse, RedirectResponse
16from lexigram.admin.auth.protocols import (
17 AdminAuditLogServiceProtocol,
18 AdminCsrfServiceProtocol,
19 AdminEmailVerificationServiceProtocol,
20 AdminPasswordPolicyServiceProtocol,
21)
22from lexigram.admin.auth.store import AdminUserStoreProtocol
23from lexigram.admin.auth.types import AdminSecurityEventType
24from lexigram.admin.config import AdminConfig, AdminRbacConfig
25from lexigram.admin.controllers.base import AdminController
26from lexigram.admin.engine.renderer import AdminRenderer
27from lexigram.admin.lib.template import render_setup_page
28from lexigram.contracts.core import TaskManagerProtocol
29from lexigram.contracts.web import get, post
30from lexigram.di.decorators import inject
31from lexigram.logging import get_logger
33logger = get_logger(__name__)
36@inject
37class SetupController(AdminController):
38 """First-run setup wizard controller.
40 Provides:
41 - GET /setup — Display account creation form
42 - POST /setup — Create the first admin account
43 """
45 prefix = ""
47 def __init__(
48 self,
49 config: AdminConfig,
50 user_store: AdminUserStoreProtocol,
51 password_policy_service: AdminPasswordPolicyServiceProtocol,
52 audit_service: AdminAuditLogServiceProtocol,
53 csrf_service: AdminCsrfServiceProtocol,
54 renderer: AdminRenderer,
55 task_manager: TaskManagerProtocol | None = None,
56 email_verification_service: AdminEmailVerificationServiceProtocol | None = None,
57 rbac_config: AdminRbacConfig | None = None,
58 ) -> None:
59 """Initialise setup controller.
61 Args:
62 config: Admin configuration; the optional setup token is read
63 from ``config.auth.security.setup_token`` (single enforcement
64 source, env var ``ADMIN_SETUP_TOKEN`` honored via alias).
65 user_store: Store used to check and create admin accounts.
66 password_policy_service: Validates passwords against all configured
67 policy rules; returns every violation, not just the first.
68 audit_service: Records security events; guaranteed never to raise.
69 csrf_service: Generates and validates CSRF tokens for the
70 pre-session setup form (bypassed by the CSRF middleware).
71 renderer: AdminRenderer required by AdminController base.
72 task_manager: Optional; injected by container in production.
73 email_verification_service: Optional email verification
74 orchestrator; when present and the gate applies to the new
75 account, a verification email is sent and the user is
76 informed after creation.
77 rbac_config: Optional; the resolved RBAC config whose
78 ``super_admin_role`` names the role granted to the first
79 admin account.
80 """
81 super().__init__(renderer, task_manager)
82 self._config = config
83 self._user_store = user_store
84 self._password_policy_service = password_policy_service
85 self._audit_service = audit_service
86 self._csrf_service = csrf_service
87 self._email_verification_service = email_verification_service
88 self._rbac_config = rbac_config
90 def _fresh_csrf(self, request: Request) -> str:
91 """Generate a fresh session-scoped CSRF token for the setup form."""
92 csrf_session_id = secrets.token_urlsafe(16)
93 request.session["csrf_session_id"] = csrf_session_id
94 return self._csrf_service.generate_token(csrf_session_id)
96 # ------------------------------------------------------------------
97 # GET /setup
98 # ------------------------------------------------------------------
100 @get("/setup")
101 async def setup_form(self, request: Request) -> HTMLResponse | RedirectResponse:
102 """Display the first-run setup form.
104 If at least one admin account already exists, a locked message is shown
105 so the user knows to log in with their existing credentials.
107 Args:
108 request: Incoming HTTP request.
110 Returns:
111 HTMLResponse with the rendered setup page.
112 """
113 required_token = self._config.auth.security.setup_token
114 try:
115 count = await self._user_store.get_admin_count()
116 except (RuntimeError, ValueError, OSError) as e:
117 logger.warning("setup.count_failed error=%s", e)
118 html = render_setup_page(
119 error="Unable to verify setup status. Database may be unavailable.",
120 csrf_token=self._fresh_csrf(request),
121 setup_token_required=bool(required_token),
122 )
123 return HTMLResponse(content=html, status_code=503)
124 if count > 0:
125 html = render_setup_page(
126 locked=True,
127 error="Setup is already complete. Please log in with your existing account.",
128 )
129 return HTMLResponse(content=html, status_code=200)
131 error = request.query_params.get("error", "")
132 html = render_setup_page(
133 error=error,
134 csrf_token=self._fresh_csrf(request),
135 setup_token_required=bool(required_token),
136 )
137 return HTMLResponse(content=html)
139 # ------------------------------------------------------------------
140 # POST /setup
141 # ------------------------------------------------------------------
143 @post("/setup")
144 async def setup_submit(self, request: Request) -> HTMLResponse | RedirectResponse:
145 """Process the first-run setup form and create the initial admin account.
147 Validates the optional setup token, enforces the full password policy
148 (all violations reported simultaneously), hashes the password with
149 bcrypt, persists the user, audits the outcome, then redirects to the
150 login page.
152 Args:
153 request: Incoming HTTP request carrying form data.
155 Returns:
156 RedirectResponse to ``/admin/login?next=/admin/`` on success, or
157 an HTMLResponse re-rendering the setup form with error details on
158 any validation or persistence failure.
159 """
160 required_token = self._config.auth.security.setup_token
161 try:
162 count = await self._user_store.get_admin_count()
163 except (RuntimeError, ValueError, OSError) as e:
164 logger.warning("setup.count_failed error=%s", e)
165 html = render_setup_page(
166 error="Unable to verify setup status. Database may be unavailable.",
167 csrf_token=self._fresh_csrf(request),
168 setup_token_required=bool(required_token),
169 )
170 return HTMLResponse(content=html, status_code=503)
171 if count > 0:
172 html = render_setup_page(
173 locked=True,
174 error="Setup is already complete. Please log in with your existing account.",
175 )
176 return HTMLResponse(content=html, status_code=200)
178 form_data = await request.form()
179 name = str(form_data.get("name", "")).strip()
180 email = str(form_data.get("email", "")).strip()
181 password = str(form_data.get("password", "")).strip()
182 confirm = str(form_data.get("confirm_password", "")).strip()
183 setup_token_input = str(form_data.get("setup_token", "")).strip()
184 csrf_token = str(form_data.get("csrf_token", ""))
186 ip = self._get_client_ip(request)
187 user_agent = request.headers.get("user-agent", "")
189 # ── CSRF validation ────────────────────────────────────────────
190 csrf_session_id = request.session.get("csrf_session_id", "")
191 if not csrf_session_id or not self._csrf_service.validate_token(
192 csrf_session_id, csrf_token
193 ):
194 logger.warning("setup.csrf_validation_failed", ip=ip)
195 html = render_setup_page(
196 error="Invalid or expired security token. Please reload the page and try again.",
197 csrf_token=self._fresh_csrf(request),
198 setup_token_required=bool(required_token),
199 )
200 return HTMLResponse(content=html, status_code=422)
202 # ── Optional setup-token guard ─────────────────────────────────
203 if required_token and setup_token_input != required_token:
204 logger.warning("setup.token_mismatch", ip=ip)
205 await self._audit_service.log_event(
206 event_type=AdminSecurityEventType.SETUP_BLOCKED,
207 ip_address=ip,
208 user_agent=user_agent,
209 success=False,
210 metadata={"reason": "invalid_setup_token"},
211 )
212 html = render_setup_page(
213 error="Invalid setup token.",
214 csrf_token=self._fresh_csrf(request),
215 setup_token_required=bool(required_token),
216 )
217 return HTMLResponse(content=html, status_code=403)
219 # ── Basic field presence ───────────────────────────────────────
220 if not name or not email or not password:
221 html = render_setup_page(
222 error="All fields are required.",
223 csrf_token=self._fresh_csrf(request),
224 setup_token_required=bool(required_token),
225 )
226 return HTMLResponse(content=html, status_code=422)
228 if password != confirm:
229 html = render_setup_page(
230 error="Passwords do not match.",
231 csrf_token=self._fresh_csrf(request),
232 setup_token_required=bool(required_token),
233 )
234 return HTMLResponse(content=html, status_code=422)
236 # ── Full password policy validation (all violations) ───────────
237 policy_result = self._password_policy_service.validate(password, email=email)
238 if not policy_result.is_valid:
239 violation_lines = "\n".join(
240 f"• {v.message}" for v in policy_result.violations
241 )
242 html = render_setup_page(
243 error=violation_lines,
244 csrf_token=self._fresh_csrf(request),
245 setup_token_required=bool(required_token),
246 )
247 return HTMLResponse(content=html, status_code=422)
249 # ── Hash and persist ───────────────────────────────────────────
250 hashed_password = _hash_password(password)
252 try:
253 created_result = await self._user_store.claim_first_admin(
254 name=name,
255 email=email,
256 hashed_password=hashed_password,
257 roles=[(self._rbac_config or AdminRbacConfig()).super_admin_role],
258 )
259 except Exception as exc:
260 # Treat any persistence failure (duplicate email, DB error, etc.)
261 # as a non-fatal setup error that is shown back to the user.
262 logger.error("setup.create_user_failed", email=email, error=str(exc))
263 html = render_setup_page(
264 error=f"Failed to create account: {exc}",
265 csrf_token=self._fresh_csrf(request),
266 setup_token_required=bool(required_token),
267 )
268 return HTMLResponse(content=html, status_code=422)
270 if created_result.is_err():
271 # Another submission created the first admin between the
272 # pre-flight count check and this insert — lock the wizard.
273 html = render_setup_page(
274 locked=True,
275 error="Setup is already complete. Please log in with your existing account.",
276 )
277 return HTMLResponse(content=html, status_code=200)
278 created = created_result.unwrap()
280 logger.info("setup.first_admin_created", email=email)
282 await self._audit_service.log_event(
283 event_type=AdminSecurityEventType.SETUP_COMPLETED,
284 ip_address=ip,
285 user_agent=user_agent,
286 success=True,
287 metadata={"email": email},
288 )
290 notice = ""
291 user_id = str(getattr(created, "user_id", "") or getattr(created, "id", ""))
292 if (
293 self._email_verification_service is not None
294 and user_id
295 and await self._email_verification_service.is_required(user_id)
296 ):
297 send_result = await self._email_verification_service.send_verification(
298 user_id=user_id,
299 email=email,
300 user_name=name,
301 base_url=str(request.base_url),
302 ip_address=ip,
303 )
304 if send_result.is_ok():
305 notice = (
306 f"Account created successfully — a verification email was "
307 f"sent to {email}. Please verify your email before signing in."
308 )
309 else:
310 logger.error(
311 "setup.verification_send_failed",
312 email=email,
313 error=str(send_result.unwrap_err()),
314 )
315 notice = (
316 "Account created successfully — email verification is "
317 "enabled, and you will be asked to verify your email "
318 "before signing in."
319 )
321 url = "/admin/login?next=/admin/"
322 if notice:
323 url += "¬ice=" + quote_plus(notice)
324 return RedirectResponse(url=url, status_code=302)
326 # ------------------------------------------------------------------
327 # Helpers
328 # ------------------------------------------------------------------
330 def _get_client_ip(self, request: Request) -> str:
331 """Extract the real client IP from the request.
333 Prefers the first value of the ``X-Forwarded-For`` header when present
334 (set by reverse proxies), falling back to the direct TCP peer address.
336 Args:
337 request: Incoming HTTP request.
339 Returns:
340 IP address string, or ``"unknown"`` when unavailable.
341 """
342 forwarded = request.headers.get("x-forwarded-for", "")
343 if forwarded:
344 return forwarded.split(",")[0].strip()
345 return request.client.host if request.client else "unknown"
348def _hash_password(plain: str) -> str:
349 """Hash a plain-text password using bcrypt.
351 Bcrypt with 12 rounds is used. A missing ``bcrypt`` package raises
352 ``RuntimeError`` (fail-closed) instead of degrading to SHA-256.
354 Args:
355 plain: Plain-text password string.
357 Returns:
358 Hashed password string suitable for storage.
360 Raises:
361 RuntimeError: When the ``bcrypt`` package is not installed.
362 """
363 from lexigram.admin.lib.password import hash_password
365 return hash_password(plain)
368__all__ = ["SetupController"]