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