Coverage for src/lexigram/admin/controllers/auth/registration.py: 18%

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING 

4from urllib.parse import quote_plus 

5 

6from starlette.requests import Request 

7from starlette.responses import HTMLResponse, RedirectResponse 

8 

9from lexigram.admin.controllers.auth.core import ( 

10 _CACHE_CONTROL_NO_STORE, 

11 AuthCoreMixin, 

12 _humanize_error, 

13 logger, 

14) 

15from lexigram.admin.lib.template import ( 

16 render_email_verified_page, 

17 render_register_page, 

18 render_verify_email_page, 

19) 

20from lexigram.contracts.web import get, post 

21 

22if TYPE_CHECKING: 

23 from lexigram.admin.auth.protocols import ( 

24 AdminCsrfServiceProtocol, 

25 AdminEmailVerificationServiceProtocol, 

26 ) 

27 from lexigram.admin.auth.store import AdminUserStoreProtocol 

28 

29 

30class AuthRegistrationMixin(AuthCoreMixin): 

31 """AuthController registration endpoints.""" 

32 

33 _csrf_service: AdminCsrfServiceProtocol 

34 _email_verification_service: AdminEmailVerificationServiceProtocol | None 

35 _user_store: AdminUserStoreProtocol | None 

36 _registration_default_role: str 

37 _registration_domains: list[str] 

38 _registration_enabled: bool 

39 

40 @get("/register") 

41 async def register_form(self, request: Request) -> HTMLResponse | RedirectResponse: 

42 """Display the standalone registration page. 

43 

44 Only reachable when self-service registration is enabled in 

45 configuration; already-authenticated users are redirected home. 

46 

47 Args: 

48 request: Incoming HTTP request. 

49 

50 Returns: 

51 HTMLResponse with the rendered registration page, or a 

52 RedirectResponse when registration is disabled or the user is 

53 already signed in. 

54 """ 

55 if not self._registration_enabled or self._user_store is None: 

56 return RedirectResponse( 

57 url="/admin/login?error=" 

58 + quote_plus("Registration is not available."), 

59 status_code=302, 

60 ) 

61 error = _humanize_error(request.query_params.get("error", "")) 

62 notice = request.query_params.get("notice", "") 

63 user = getattr(request.state, "user", None) 

64 if user and user.user_id != "guest" and not error and not notice: 

65 return RedirectResponse( 

66 url="/admin/", status_code=302, headers=_CACHE_CONTROL_NO_STORE 

67 ) 

68 

69 name = request.query_params.get("name", "") 

70 email = request.query_params.get("email", "") 

71 csrf_token = self._fresh_csrf(request) 

72 

73 html = render_register_page( 

74 error=error, 

75 notice=notice, 

76 csrf_token=csrf_token, 

77 name=name, 

78 email=email, 

79 name_err=request.query_params.get("name_err", ""), 

80 email_err=request.query_params.get("email_err", ""), 

81 password_err=request.query_params.get("password_err", ""), 

82 confirmation_err=request.query_params.get("confirmation_err", ""), 

83 ) 

84 return HTMLResponse(content=html, headers=_CACHE_CONTROL_NO_STORE) 

85 

86 @post("/register") 

87 async def register_submit(self, request: Request) -> RedirectResponse: 

88 """Process the registration form. 

89 

90 Validates CSRF, required fields, password confirmation, and the 

91 configured email-domain allowlist, then persists the new account 

92 via the admin user store and signs the user in directly. 

93 

94 Args: 

95 request: Incoming HTTP request carrying form data. 

96 

97 Returns: 

98 RedirectResponse to /admin/ on success, or back to the 

99 registration page with an error query parameter. 

100 """ 

101 if not self._registration_enabled or self._user_store is None: 

102 return RedirectResponse( 

103 url="/admin/login?error=" 

104 + quote_plus("Registration is not available."), 

105 status_code=302, 

106 ) 

107 

108 form_data = request.scope.get("admin_form_data") or await request.form() 

109 name = str(form_data.get("name", "")).strip() 

110 email = str(form_data.get("email", "")).strip().lower() 

111 password = str(form_data.get("password", "")) 

112 password_confirmation = str(form_data.get("password_confirmation", "")) 

113 csrf_token = str(form_data.get("csrf_token", "")) 

114 

115 csrf_session_id = request.session.get("csrf_session_id", "") 

116 if not csrf_session_id or not self._csrf_service.validate_token( 

117 csrf_session_id, csrf_token 

118 ): 

119 logger.warning( 

120 "auth.csrf_validation_failed", ip=self._get_client_ip(request) 

121 ) 

122 return RedirectResponse( 

123 url=f"/admin/register?error={quote_plus('Invalid or expired security token. Please try again.')}", 

124 status_code=302, 

125 ) 

126 

127 if not name or not email or not password: 

128 name_err = "Name is required." if not name else "" 

129 email_err = "Email is required." if not email else "" 

130 password_err = "Password is required." if not password else "" 

131 params = [f"name={quote_plus(name)}", f"email={quote_plus(email)}"] 

132 if name_err: 

133 params.append(f"name_err={quote_plus(name_err)}") 

134 if email_err: 

135 params.append(f"email_err={quote_plus(email_err)}") 

136 if password_err: 

137 params.append(f"password_err={quote_plus(password_err)}") 

138 return RedirectResponse( 

139 url=f"/admin/register?{'&'.join(params)}", 

140 status_code=302, 

141 ) 

142 

143 if len(password) < 8: 

144 return RedirectResponse( 

145 url=f"/admin/register?error={quote_plus('Password must be at least 8 characters.')}&name={quote_plus(name)}&email={quote_plus(email)}&password_err={quote_plus('Password must be at least 8 characters.')}", 

146 status_code=302, 

147 ) 

148 

149 if password != password_confirmation: 

150 return RedirectResponse( 

151 url=f"/admin/register?error={quote_plus('Passwords do not match.')}&name={quote_plus(name)}&email={quote_plus(email)}&confirmation_err={quote_plus('Passwords do not match.')}", 

152 status_code=302, 

153 ) 

154 

155 if self._registration_domains and "@" in email: 

156 domain = email.rsplit("@", 1)[1] 

157 if domain not in self._registration_domains: 

158 return RedirectResponse( 

159 url=f"/admin/register?error={quote_plus('Registration is restricted to allowed email domains.')}&name={quote_plus(name)}&email={quote_plus(email)}", 

160 status_code=302, 

161 ) 

162 

163 existing = await self._user_store.get_user_by_email(email) 

164 if existing is not None: 

165 logger.info( 

166 "auth.register_duplicate_email", 

167 email=email, 

168 ip=self._get_client_ip(request), 

169 ) 

170 return RedirectResponse( 

171 url=f"/admin/register?error={quote_plus('An account with this email already exists. Please log in instead.')}&name={quote_plus(name)}&email={quote_plus(email)}", 

172 status_code=302, 

173 ) 

174 

175 ip = self._get_client_ip(request) 

176 user_agent = request.headers.get("user-agent", "") 

177 from lexigram.admin.lib.password import hash_password 

178 

179 hashed = hash_password(password) 

180 roles = ( 

181 [self._registration_default_role] 

182 if self._registration_default_role 

183 else None 

184 ) 

185 try: 

186 created = await self._user_store.create_user( 

187 name=name, 

188 email=email, 

189 hashed_password=hashed, 

190 roles=roles, 

191 ) 

192 except Exception as exc: # noqa: BLE001 — persistence failures surface to the user 

193 logger.warning( 

194 "auth.register_create_failed", 

195 email=email, 

196 ip=ip, 

197 error=str(exc), 

198 ) 

199 return RedirectResponse( 

200 url=f"/admin/register?error={quote_plus('Could not create the account: email may already be in use.')}&name={quote_plus(name)}&email={quote_plus(email)}", 

201 status_code=302, 

202 ) 

203 

204 user_id = str(getattr(created, "user_id", "") or getattr(created, "id", "")) 

205 logger.info("auth.register_success", email=email, user_id=user_id) 

206 

207 await self._audit_registration(request, ip, user_agent, email) 

208 

209 notice = "Account created successfully — please sign in." 

210 if ( 

211 self._email_verification_service is not None 

212 and user_id 

213 and await self._email_verification_service.is_required(user_id) 

214 ): 

215 verify_result = await self._email_verification_service.send_verification( 

216 user_id=user_id, 

217 email=email, 

218 user_name=name, 

219 base_url=str(request.base_url), 

220 ip_address=ip, 

221 ) 

222 if verify_result.is_ok(): 

223 notice = ( 

224 "Account created successfully — a verification email was " 

225 f"sent to {email}. Please verify your email before " 

226 "signing in." 

227 ) 

228 else: 

229 logger.warning( 

230 "auth.register_verification_send_failed", 

231 email=email, 

232 error=str(verify_result.unwrap_err()), 

233 ) 

234 notice = ( 

235 "Account created successfully — email verification is " 

236 "enabled, and you will be asked to verify your email " 

237 "before signing in." 

238 ) 

239 return RedirectResponse( 

240 url="/admin/login?notice=" + quote_plus(notice), 

241 status_code=302, 

242 ) 

243 

244 async def _audit_registration( 

245 self, request: Request, ip_address: str, user_agent: str, email: str 

246 ) -> None: 

247 """Record the registration audit event (best-effort). 

248 

249 Resolves the audit service from the request DI container, mirroring 

250 the theme-overrides pattern; failures are logged, never raised. 

251 

252 Args: 

253 request: The current request (carries the DI container). 

254 ip_address: Client IP for the audit record. 

255 user_agent: Client user agent for the audit record. 

256 email: Registered email address. 

257 """ 

258 try: 

259 from lexigram.admin.auth.protocols import AdminAuditLogServiceProtocol 

260 from lexigram.admin.auth.types import AdminSecurityEventType 

261 

262 container = getattr(request.state, "container", None) 

263 if container is None: 

264 return 

265 audit_service = await container.resolve( 

266 AdminAuditLogServiceProtocol, 

267 bypass_visibility=True, 

268 ) 

269 await audit_service.log_event( 

270 event_type=AdminSecurityEventType.USER_REGISTERED, 

271 ip_address=ip_address, 

272 user_agent=user_agent, 

273 success=True, 

274 metadata={"email": email}, 

275 ) 

276 except Exception as exc: # noqa: BLE001 — auditing is best-effort 

277 logger.warning("auth.register_audit_failed", error=str(exc)) 

278 

279 @get("/verify-email") 

280 async def verify_email_form( 

281 self, request: Request 

282 ) -> HTMLResponse | RedirectResponse: 

283 """Display the standalone email verification landing page. 

284 

285 Shown after a login attempt was gated on an unverified email. Lets 

286 the admin request a fresh verification link. 

287 

288 Args: 

289 request: Incoming HTTP request. 

290 

291 Returns: 

292 HTMLResponse with the rendered page, or a RedirectResponse when 

293 already authenticated. 

294 """ 

295 user = getattr(request.state, "user", None) 

296 if user and user.user_id != "guest": 

297 return RedirectResponse(url="/admin/", status_code=302) 

298 

299 email = request.session.get("verify_pending_email", "") 

300 next_url = request.session.get("verify_pending_next", "/admin/") 

301 error = _humanize_error(request.query_params.get("error", "")) 

302 notice = request.query_params.get("notice", "") 

303 csrf_token = self._fresh_csrf(request) 

304 

305 html = render_verify_email_page( 

306 email=email, 

307 error=error, 

308 notice=notice, 

309 csrf_token=csrf_token, 

310 next_url=next_url, 

311 ) 

312 return HTMLResponse(content=html, headers=_CACHE_CONTROL_NO_STORE) 

313 

314 @post("/verify-email/resend") 

315 async def verify_email_resend(self, request: Request) -> RedirectResponse: 

316 """Re-issue the verification email for a pending verification. 

317 

318 Rate limited per IP (5/hour, fail open) by the verification service. 

319 

320 Args: 

321 request: Incoming HTTP request carrying form data. 

322 

323 Returns: 

324 RedirectResponse to /admin/verify-email with a notice on success 

325 or an error query parameter on failure. 

326 """ 

327 form_data = request.scope.get("admin_form_data") or await request.form() 

328 csrf_token = str(form_data.get("csrf_token", "")) 

329 csrf_session_id = request.session.get("csrf_session_id", "") 

330 if not csrf_session_id or not self._csrf_service.validate_token( 

331 csrf_session_id, csrf_token 

332 ): 

333 logger.warning( 

334 "auth.csrf_validation_failed", ip=self._get_client_ip(request) 

335 ) 

336 return RedirectResponse( 

337 url=f"/admin/verify-email?error={quote_plus('Invalid or expired security token. Please try again.')}", 

338 status_code=302, 

339 ) 

340 

341 user_id = request.session.get("verify_pending_user_id", "") 

342 if not user_id: 

343 return RedirectResponse( 

344 url=f"/admin/login?error={quote_plus('Please sign in to request a new link.')}", 

345 status_code=302, 

346 ) 

347 if self._email_verification_service is None: 

348 return RedirectResponse( 

349 url=f"/admin/verify-email?error={quote_plus('Email verification is not available.')}", 

350 status_code=302, 

351 ) 

352 

353 email = str(form_data.get("email", "")) or request.session.get( 

354 "verify_pending_email", "" 

355 ) 

356 result = await self._email_verification_service.send_verification( 

357 user_id=user_id, 

358 email=email, 

359 user_name=email, 

360 base_url=str(request.base_url), 

361 ip_address=self._get_client_ip(request), 

362 ) 

363 if result.is_err(): 

364 return RedirectResponse( 

365 url=f"/admin/verify-email?error={quote_plus(_humanize_error(str(result.unwrap_err())))}", 

366 status_code=302, 

367 ) 

368 return RedirectResponse( 

369 url="/admin/verify-email?notice=" 

370 + quote_plus("A new verification link has been sent."), 

371 status_code=302, 

372 ) 

373 

374 @get("/verify-email/{token}") 

375 async def verify_email_token(self, request: Request) -> HTMLResponse: 

376 """Consume a verification token from an emailed link. 

377 

378 Renders a confirmation page on success and a failure page (invalid, 

379 used, or expired token) otherwise. No session is required. 

380 

381 Args: 

382 request: Incoming HTTP request (``token`` from path params). 

383 

384 Returns: 

385 HTMLResponse with the confirmation or failure page. 

386 """ 

387 token = request.path_params.get("token", "") 

388 if self._email_verification_service is None: 

389 return HTMLResponse( 

390 content=render_email_verified_page( 

391 error="Email verification is not available." 

392 ) 

393 ) 

394 

395 result = await self._email_verification_service.verify_token(token) 

396 if result.is_err(): 

397 logger.warning( 

398 "auth.verify_email_token_failed", 

399 token_prefix=token[:8], 

400 reason=str(result.unwrap_err()), 

401 ) 

402 return HTMLResponse( 

403 content=render_email_verified_page( 

404 error=_humanize_error(str(result.unwrap_err())) 

405 ) 

406 ) 

407 

408 for key in ( 

409 "verify_pending_user_id", 

410 "verify_pending_email", 

411 "verify_pending_next", 

412 ): 

413 request.session.pop(key, None) 

414 logger.info("auth.verify_email_success", token_prefix=token[:8]) 

415 return HTMLResponse(content=render_email_verified_page())