Coverage for api/auth/services.py: 98.64%

1import ast 

2from datetime import datetime, timedelta 

3from urllib.parse import quote 

4 

5import redis 

6from jose import JWTError, jwt 

7from sqlalchemy import func, or_, select, update 

8from sqlalchemy.ext.asyncio import AsyncSession 

9 

10from core.config import settings 

11from core.security import ( 

12 clear_user_all_sessions, 

13 create_access_token, 

14 create_csrf_token, 

15 create_email_verification_token, 

16 create_password_reset_token, 

17 extend_session_ttl, 

18 hash_password, 

19 verify_password, 

20) 

21from extensions.smtp import SMTPMailer 

22from models.email_verification_tokens import EmailVerificationTokens 

23from models.login_logs import LoginLogs 

24from models.password_reset_tokens import PasswordResetTokens 

25from models.user_sessions import UserSessions 

26from models.users import Users 

27from utils.custom_exception import ( 

28 AuthenticationException, 

29 ConflictException, 

30 EmailVerificationRequiredException, 

31 NotFoundException, 

32 PasswordResetRequiredException, 

33 RegistrationDisabledException, 

34 ServerException, 

35 SMTPNotConfiguredException, 

36 ValidationException, 

37) 

38from utils.email_templates import ( 

39 EMAIL_VERIFICATION_TEMPLATE, 

40 PASSWORD_RESET_TEMPLATE, 

41) 

42 

43from .schema import ( 

44 ActionRequiredResponse, 

45 LoginResult, 

46 SessionResult, 

47 TokenValidationResponse, 

48 UserLogin, 

49 UserRegister, 

50) 

51 

52 

53async def register( 

54 db: AsyncSession, 

55 redis_client: redis.Redis, 

56 user_data: UserRegister, 

57 ip_address: str, 

58 user_agent: str, 

59 mailer: SMTPMailer | None = None, 

60) -> LoginResult: 

61 """User register""" 

62 if not settings.REGISTRATION_ENABLE: 

63 raise RegistrationDisabledException("Registration is disabled") 

64 

65 user = await _create_user(db, user_data) 

66 

67 # Check if email verification is required 

68 if settings.EMAIL_VERIFICATION_ENABLE and settings.SMTP_ENABLE and mailer: 

69 # Check cooldown 

70 cooldown_key = f"email_verification_cooldown:{user.email}" 

71 remaining_seconds = await redis_client.ttl(cooldown_key) 

72 

73 if remaining_seconds > 0: 

74 # In cooldown, return 202 without data 

75 await _log_login_attempt( 

76 db, 

77 email=user.email, 

78 ip_address=ip_address, 

79 user_agent=user_agent, 

80 is_success=True, 

81 user_id=user.id, 

82 ) 

83 raise EmailVerificationRequiredException( 

84 message="Email verification required", details=None 

85 ) 

86 else: 

87 # Not in cooldown, send verification email 

88 await _send_registration_verification_email(db, mailer, user) 

89 

90 # Set cooldown 

91 await redis_client.setex( 

92 cooldown_key, settings.EMAIL_VERIFICATION_COOLDOWN_SECONDS, "1" 

93 ) 

94 

95 await _log_login_attempt( 

96 db, 

97 email=user.email, 

98 ip_address=ip_address, 

99 user_agent=user_agent, 

100 is_success=True, 

101 user_id=user.id, 

102 ) 

103 raise EmailVerificationRequiredException( 

104 message="Email verification required", details=None 

105 ) 

106 

107 session_result = await _create_user_session(db, redis_client, user, ip_address, user_agent) 

108 await _log_login_attempt( 

109 db, 

110 email=user.email, 

111 ip_address=ip_address, 

112 user_agent=user_agent, 

113 is_success=True, 

114 user_id=user.id, 

115 ) 

116 return { 

117 "user": user, 

118 "session_id": session_result["session_id"], 

119 "access_token": session_result["access_token"], 

120 "csrf_token": session_result["csrf_token"], 

121 } 

122 

123 

124async def login( 

125 db: AsyncSession, 

126 redis_client: redis.Redis, 

127 login_data: UserLogin, 

128 ip_address: str, 

129 user_agent: str, 

130 mailer: SMTPMailer | None = None, 

131) -> LoginResult: 

132 """User login""" 

133 result = await db.execute(select(Users).where(Users.email == login_data.email)) 

134 user = result.scalar_one_or_none() 

135 

136 if not user: 

137 await _log_login_attempt( 

138 db, 

139 email=login_data.email, 

140 ip_address=ip_address, 

141 user_agent=user_agent, 

142 is_success=False, 

143 failure_reason="User not found", 

144 ) 

145 raise AuthenticationException("Invalid email or password") 

146 

147 # Check if user account is disabled 

148 if not user.status: 

149 await _log_login_attempt( 

150 db, 

151 email=login_data.email, 

152 ip_address=ip_address, 

153 user_agent=user_agent, 

154 is_success=False, 

155 failure_reason="Account disabled", 

156 ) 

157 raise AuthenticationException("Account is disabled") 

158 

159 # Now verify password 

160 if not await verify_password(login_data.password, user.hash_password): 

161 await _log_login_attempt( 

162 db, 

163 email=login_data.email, 

164 ip_address=ip_address, 

165 user_agent=user_agent, 

166 is_success=False, 

167 failure_reason="Invalid password", 

168 ) 

169 raise AuthenticationException("Invalid email or password") 

170 

171 # Check if password reset is required 

172 if user.password_reset_required: 

173 reset_token = await create_password_reset_token(user.id, user.email) 

174 

175 reset_token_record = PasswordResetTokens( 

176 user_id=user.id, 

177 token=reset_token, 

178 expires_at=datetime.now().astimezone() 

179 + timedelta(minutes=settings.PASSWORD_RESET_TOKEN_EXPIRE_MINUTES), 

180 ) 

181 db.add(reset_token_record) 

182 await db.commit() 

183 

184 await _log_login_attempt( 

185 db, 

186 email=user.email, 

187 ip_address=ip_address, 

188 user_agent=user_agent, 

189 is_success=True, 

190 user_id=user.id, 

191 ) 

192 

193 raise PasswordResetRequiredException( 

194 message="Password reset required", 

195 details=ActionRequiredResponse( 

196 action_type="password_reset", 

197 token=reset_token, 

198 expires_at=reset_token_record.expires_at.isoformat() 

199 if reset_token_record.expires_at 

200 else None, 

201 ), 

202 ) 

203 

204 # Check if email verification is required 

205 if settings.EMAIL_VERIFICATION_ENABLE and settings.SMTP_ENABLE and mailer: 

206 if not user.email_verified: 

207 # Check cooldown 

208 cooldown_key = f"email_verification_cooldown:{user.email}" 

209 remaining_seconds = await redis_client.ttl(cooldown_key) 

210 

211 if remaining_seconds > 0: 

212 # In cooldown, return 202 with cooldown time 

213 await _log_login_attempt( 

214 db, 

215 email=user.email, 

216 ip_address=ip_address, 

217 user_agent=user_agent, 

218 is_success=True, 

219 user_id=user.id, 

220 ) 

221 # Calculate expires_at from cooldown 

222 expires_at = ( 

223 datetime.now().astimezone() + timedelta(seconds=remaining_seconds) 

224 ).isoformat() 

225 raise EmailVerificationRequiredException( 

226 message="Email verification required", 

227 details=ActionRequiredResponse( 

228 action_type="email_verification", token=None, expires_at=expires_at 

229 ), 

230 ) 

231 else: 

232 # Not in cooldown, send verification email 

233 await _send_registration_verification_email(db, mailer, user) 

234 

235 # Set cooldown 

236 await redis_client.setex( 

237 cooldown_key, settings.EMAIL_VERIFICATION_COOLDOWN_SECONDS, "1" 

238 ) 

239 

240 await _log_login_attempt( 

241 db, 

242 email=user.email, 

243 ip_address=ip_address, 

244 user_agent=user_agent, 

245 is_success=True, 

246 user_id=user.id, 

247 ) 

248 # Calculate expires_at from cooldown 

249 expires_at = ( 

250 datetime.now().astimezone() 

251 + timedelta(seconds=settings.EMAIL_VERIFICATION_COOLDOWN_SECONDS) 

252 ).isoformat() 

253 raise EmailVerificationRequiredException( 

254 message="Email verification required", 

255 details=ActionRequiredResponse( 

256 action_type="email_verification", token=None, expires_at=expires_at 

257 ), 

258 ) 

259 

260 session_result = await _create_user_session(db, redis_client, user, ip_address, user_agent) 

261 await _log_login_attempt( 

262 db, 

263 email=user.email, 

264 ip_address=ip_address, 

265 user_agent=user_agent, 

266 is_success=True, 

267 user_id=user.id, 

268 ) 

269 return { 

270 "user": user, 

271 "session_id": session_result["session_id"], 

272 "access_token": session_result["access_token"], 

273 "csrf_token": session_result["csrf_token"], 

274 } 

275 

276 

277async def logout( 

278 db: AsyncSession, redis_client: redis.Redis, user_id: str, session_id: str 

279) -> bool: 

280 """User logout""" 

281 try: 

282 redis_key = f"session:{session_id}" 

283 await redis_client.delete(redis_key, f"csrf:{session_id}") 

284 

285 result = await db.execute( 

286 select(UserSessions).where( 

287 UserSessions.user_id == user_id, UserSessions.id == session_id 

288 ) 

289 ) 

290 session = result.scalar_one_or_none() 

291 if session: 

292 session.is_active = False 

293 await db.commit() 

294 

295 return True 

296 except Exception as e: 

297 raise ServerException(f"Logout failed: {e}") 

298 

299 

300async def logout_all_devices(db: AsyncSession, redis_client: redis.Redis, user_id: str) -> bool: 

301 """Logout user from all devices""" 

302 try: 

303 return await clear_user_all_sessions(db, redis_client, user_id) 

304 except Exception as e: 

305 raise ServerException(f"Failed to logout all devices: {e}") 

306 

307 

308async def get_or_create_csrf_token( 

309 redis_client: redis.Redis, 

310 session_id: str, 

311) -> str: 

312 """Return existing CSRF token or create a new one (does not extend TTL).""" 

313 session_raw = await redis_client.get(f"session:{session_id}") 

314 if not session_raw: 

315 raise AuthenticationException("Invalid or expired session") 

316 

317 existing = await redis_client.get(_csrf_redis_key(session_id)) 

318 if existing: 

319 return existing.decode() if isinstance(existing, bytes) else existing 

320 

321 return await _create_csrf_token_for_session(redis_client, session_id) 

322 

323 

324async def verify_csrf_token( 

325 redis_client: redis.Redis, 

326 csrf_token: str | None, 

327) -> str: 

328 """Validate CSRF token and return session_id.""" 

329 if not csrf_token: 

330 raise AuthenticationException("Invalid or expired CSRF token") 

331 

332 try: 

333 payload = jwt.decode( 

334 csrf_token, 

335 settings.SECRET_KEY, 

336 algorithms=[settings.ALGORITHM], 

337 ) 

338 except JWTError: 

339 raise AuthenticationException("Invalid or expired CSRF token") 

340 

341 if payload.get("token_type") != "csrf": 

342 raise AuthenticationException("Invalid or expired CSRF token") 

343 

344 session_id = payload.get("sid") 

345 if not session_id: 

346 raise AuthenticationException("Invalid or expired CSRF token") 

347 

348 stored = await redis_client.get(_csrf_redis_key(session_id)) 

349 if not stored: 

350 raise AuthenticationException("Invalid or expired CSRF token") 

351 

352 stored_token = stored.decode() if isinstance(stored, bytes) else stored 

353 if stored_token != csrf_token: 

354 raise AuthenticationException("Invalid or expired CSRF token") 

355 

356 return session_id 

357 

358 

359async def token(db: AsyncSession, redis_client: redis.Redis, session_id: str) -> str: 

360 """Use session_id (Cookie) to issue new access_token and refresh session""" 

361 raw = await redis_client.get(f"session:{session_id}") 

362 if not raw: 

363 raise AuthenticationException("Invalid or expired session") 

364 try: 

365 data = ast.literal_eval(raw) 

366 except Exception: 

367 raise AuthenticationException("Invalid or expired session") 

368 

369 user_id = data.get("user_id") 

370 if not user_id: 

371 raise AuthenticationException("Invalid or expired session") 

372 

373 # Verify user exists 

374 user = await _get_user_by_id(db, user_id) 

375 if not user: 

376 raise NotFoundException("User not found") 

377 

378 if not user.status: 

379 raise AuthenticationException("Account is disabled") 

380 

381 new_access_token = await create_access_token( 

382 data={"sub": user_id, "email": user.email, "sid": session_id} 

383 ) 

384 

385 data["access_token"] = new_access_token 

386 await extend_session_ttl(redis_client, session_id, data) 

387 await _update_session_expiry(db, session_id) 

388 

389 return new_access_token 

390 

391 

392async def reset_password( 

393 db: AsyncSession, 

394 redis_client: redis.Redis, 

395 token: dict, 

396 new_password: str, 

397 ip_address: str, 

398 user_agent: str, 

399) -> LoginResult: 

400 """Reset password using token""" 

401 try: 

402 user_id = token.get("sub") 

403 token_string = token.get("token") 

404 

405 result = await db.execute( 

406 select(PasswordResetTokens).where( 

407 PasswordResetTokens.token == token_string, 

408 PasswordResetTokens.user_id == user_id, 

409 PasswordResetTokens.is_used.is_(False), 

410 PasswordResetTokens.expires_at > func.now(), 

411 ) 

412 ) 

413 token_record = result.scalar_one_or_none() 

414 

415 if not token_record: 

416 raise AuthenticationException("Invalid or expired token") 

417 

418 result = await db.execute(select(Users).where(Users.id == user_id)) 

419 user = result.scalar_one_or_none() 

420 

421 if not user: 

422 raise NotFoundException("User not found") 

423 

424 user.hash_password = await hash_password(new_password) 

425 user.password_reset_required = False 

426 

427 token_record.is_used = True 

428 

429 # Force logout all devices 

430 await clear_user_all_sessions(db, redis_client, user_id) 

431 

432 # Create new session 

433 session_result = await _create_user_session(db, redis_client, user, ip_address, user_agent) 

434 

435 await db.commit() 

436 

437 return { 

438 "user": user, 

439 "session_id": session_result["session_id"], 

440 "access_token": session_result["access_token"], 

441 "csrf_token": session_result["csrf_token"], 

442 } 

443 

444 except AuthenticationException, NotFoundException: 

445 raise 

446 except Exception as e: 

447 raise ServerException(f"Failed to reset password: {str(e)}") 

448 

449 

450async def validate_password_reset_token(db: AsyncSession, token: dict) -> TokenValidationResponse: 

451 """Validate password reset token without consuming it""" 

452 try: 

453 user_id = token.get("sub") 

454 token_string = token.get("token") 

455 

456 result = await db.execute( 

457 select(PasswordResetTokens).where( 

458 PasswordResetTokens.token == token_string, 

459 PasswordResetTokens.user_id == user_id, 

460 PasswordResetTokens.is_used.is_(False), 

461 PasswordResetTokens.expires_at > func.now(), 

462 ) 

463 ) 

464 token_record = result.scalar_one_or_none() 

465 

466 if not token_record: 

467 raise AuthenticationException("Invalid or expired token") 

468 

469 result = await db.execute(select(Users).where(Users.id == user_id)) 

470 user = result.scalar_one_or_none() 

471 

472 if not user or not user.status: 

473 raise AuthenticationException("User not found or account disabled") 

474 

475 return TokenValidationResponse(is_valid=True) 

476 

477 except AuthenticationException: 

478 raise 

479 except Exception as e: 

480 raise ServerException(f"Token validation failed: {str(e)}") 

481 

482 

483async def forgot_password( 

484 db: AsyncSession, 

485 email: str, 

486 mailer: SMTPMailer, 

487 redis_client: redis.Redis, 

488) -> dict: 

489 """ 

490 Forgot password and send reset password email 

491 """ 

492 try: 

493 user = await _get_user_by_email_for_password_reset(db, email) 

494 

495 if not getattr(mailer, "enabled", False): 

496 raise SMTPNotConfiguredException("SMTP is disabled") 

497 

498 cooldown_key = f"password_reset_cooldown:{email}" 

499 remaining_seconds = await redis_client.ttl(cooldown_key) 

500 

501 if remaining_seconds > 0: 

502 raise ValidationException( 

503 ( 

504 f"Please wait {remaining_seconds} seconds before requesting " 

505 "another password reset email" 

506 ), 

507 details={"cooldown_seconds": remaining_seconds}, 

508 ) 

509 

510 token_meta = await _request_password_reset_email(db, user) 

511 reset_token = token_meta["reset_token"] 

512 reset_url = ( 

513 f"http{'s' if settings.SSL_ENABLE else ''}://" 

514 f"{settings.HOSTNAME}:{settings.FRONTEND_PORT}" 

515 f"/auth/reset-password?token={quote(reset_token, safe='')}" 

516 ) 

517 

518 # Render email template with user name and app name 

519 user_name = f"{user.first_name} {user.last_name}".strip() 

520 app_name = settings.PROJECT_NAME 

521 

522 email_content = PASSWORD_RESET_TEMPLATE.render( 

523 reset_url=reset_url, 

524 user_name=user_name, 

525 app_name=app_name, 

526 ) 

527 

528 mailer.send_text( 

529 to_emails=[email], 

530 subject=email_content["subject"], 

531 body=email_content["body"], 

532 html_body=email_content.get("html_body"), 

533 ) 

534 

535 # Set cooldown period in Redis 

536 await redis_client.setex(cooldown_key, settings.PASSWORD_RESET_EMAIL_COOLDOWN_SECONDS, "1") 

537 

538 return {**token_meta, "reset_url": reset_url} 

539 

540 except ( 

541 NotFoundException, 

542 AuthenticationException, 

543 SMTPNotConfiguredException, 

544 ValidationException, 

545 ): 

546 raise 

547 except Exception as e: 

548 raise ServerException(f"Failed to send password reset email: {str(e)}") 

549 

550 

551async def get_password_reset_cooldown( 

552 email: str, 

553 redis_client: redis.Redis, 

554) -> dict: 

555 """ 

556 Get remaining cooldown time for password reset email. 

557 

558 Returns: 

559 Dict with 'cooldown_seconds' (0 if no cooldown active) 

560 """ 

561 cooldown_key = f"password_reset_cooldown:{email}" 

562 remaining_seconds = await redis_client.ttl(cooldown_key) 

563 

564 # TTL returns -1 if key exists but has no expiry, -2 if key doesn't exist 

565 if remaining_seconds < 0: 

566 remaining_seconds = 0 

567 

568 return {"cooldown_seconds": remaining_seconds} 

569 

570 

571async def _update_session_expiry(db: AsyncSession, session_id: str) -> None: 

572 """Update session expiry time in database""" 

573 try: 

574 result = await db.execute(select(UserSessions).where(UserSessions.id == session_id)) 

575 session = result.scalar_one_or_none() 

576 if session: 

577 session.expires_at = datetime.now().astimezone() + timedelta( 

578 minutes=settings.SESSION_EXPIRE_MINUTES 

579 ) 

580 await db.commit() 

581 except Exception as e: 

582 raise ServerException(f"Failed to update session expiry in database: {e}") 

583 

584 

585async def _create_user(db: AsyncSession, user_data: UserRegister) -> Users: 

586 try: 

587 result = await db.execute( 

588 select(Users).where( 

589 or_(Users.email == user_data.email, Users.pending_email == user_data.email) 

590 ) 

591 ) 

592 existing_user = result.scalar_one_or_none() 

593 if existing_user: 

594 raise ConflictException("Email already exists") 

595 

596 user = Users( 

597 first_name=user_data.first_name, 

598 last_name=user_data.last_name, 

599 email=user_data.email, 

600 phone=user_data.phone, 

601 hash_password=await hash_password(user_data.password), 

602 ) 

603 db.add(user) 

604 await db.commit() 

605 await db.refresh(user) 

606 return user 

607 except ConflictException: 

608 raise 

609 except Exception as e: 

610 raise ServerException(f"Failed to create user: {str(e)}") 

611 

612 

613async def _get_user_by_id(db: AsyncSession, user_id: str) -> Users | None: 

614 result = await db.execute(select(Users).where(Users.id == user_id)) 

615 return result.scalar_one_or_none() 

616 

617 

618async def _create_user_session( 

619 db: AsyncSession, redis_client: redis.Redis, user: Users, ip_address: str, user_agent: str 

620) -> SessionResult: 

621 try: 

622 session = UserSessions( 

623 user_id=user.id, 

624 jwt_access_token="", 

625 ip_address=ip_address, 

626 user_agent=user_agent, 

627 expires_at=datetime.now().astimezone() 

628 + timedelta(minutes=settings.SESSION_EXPIRE_MINUTES), 

629 ) 

630 db.add(session) 

631 await db.commit() 

632 await db.refresh(session) 

633 

634 session_id = session.id 

635 access_token = await create_access_token( 

636 data={"sub": user.id, "email": user.email, "sid": session_id} 

637 ) 

638 

639 session.jwt_access_token = access_token 

640 await db.commit() 

641 

642 redis_key = f"session:{session_id}" 

643 session_data = { 

644 "user_id": user.id, 

645 "email": user.email, 

646 "ip_address": ip_address, 

647 "user_agent": user_agent, 

648 "access_token": access_token, 

649 "created_at": datetime.now().astimezone().isoformat(), 

650 "last_activity": datetime.now().astimezone().isoformat(), 

651 } 

652 

653 await redis_client.setex(redis_key, settings.SESSION_EXPIRE_MINUTES * 60, str(session_data)) 

654 

655 csrf_token = await _create_csrf_token_for_session(redis_client, session_id) 

656 

657 return { 

658 "session_id": session_id, 

659 "access_token": access_token, 

660 "csrf_token": csrf_token, 

661 } 

662 except Exception as e: 

663 raise ServerException(f"Failed to create user session: {str(e)}") 

664 

665 

666async def _log_login_attempt( 

667 db: AsyncSession, 

668 email: str, 

669 ip_address: str, 

670 user_agent: str, 

671 is_success: bool, 

672 user_id: str | None = None, 

673 failure_reason: str | None = None, 

674) -> None: 

675 log = LoginLogs( 

676 user_id=user_id, 

677 email=email, 

678 ip_address=ip_address, 

679 user_agent=user_agent, 

680 is_success=is_success, 

681 failure_reason=failure_reason, 

682 ) 

683 db.add(log) 

684 await db.commit() 

685 

686 

687async def _get_user_by_email_for_password_reset(db: AsyncSession, email: str) -> Users: 

688 result = await db.execute(select(Users).where(Users.email == email)) 

689 user = result.scalar_one_or_none() 

690 if not user: 

691 raise NotFoundException("User not registered") 

692 if not user.status: 

693 raise AuthenticationException("Account is disabled") 

694 return user 

695 

696 

697async def _request_password_reset_email( 

698 db: AsyncSession, 

699 user: Users, 

700) -> dict: 

701 """ 

702 Create a password reset token record for the user and return token metadata. 

703 Invalidates all previous unused tokens for this user before creating a new one. 

704 """ 

705 now = datetime.now().astimezone() 

706 expires_at = now + timedelta(minutes=settings.PASSWORD_RESET_TOKEN_EXPIRE_MINUTES) 

707 

708 # Invalidate all previous unused tokens for this user 

709 await db.execute( 

710 update(PasswordResetTokens) 

711 .where(PasswordResetTokens.user_id == user.id, PasswordResetTokens.is_used.is_(False)) 

712 .values(is_used=True) 

713 ) 

714 

715 reset_token = await create_password_reset_token(user.id, user.email) 

716 reset_token_record = PasswordResetTokens( 

717 user_id=user.id, 

718 token=reset_token, 

719 expires_at=expires_at, 

720 ) 

721 db.add(reset_token_record) 

722 await db.commit() 

723 

724 return { 

725 "reset_token": reset_token, 

726 "expires_at": expires_at, 

727 "user_id": user.id, 

728 } 

729 

730 

731async def verify_email( 

732 db: AsyncSession, redis_client: redis.Redis, token: dict, ip_address: str, user_agent: str 

733) -> LoginResult: 

734 """Verify email using token and create session""" 

735 try: 

736 user_id = token.get("sub") 

737 email = token.get("email") 

738 verification_type = token.get("verification_type") 

739 token_string = token.get("token") 

740 

741 # Verify token record exists and is valid 

742 result = await db.execute( 

743 select(EmailVerificationTokens).where( 

744 EmailVerificationTokens.token == token_string, 

745 EmailVerificationTokens.user_id == user_id, 

746 EmailVerificationTokens.email == email, 

747 EmailVerificationTokens.token_type == verification_type, 

748 EmailVerificationTokens.is_used.is_(False), 

749 EmailVerificationTokens.expires_at > func.now(), 

750 ) 

751 ) 

752 token_record = result.scalar_one_or_none() 

753 

754 if not token_record: 

755 raise AuthenticationException("Invalid or expired token") 

756 

757 # Get user 

758 result = await db.execute(select(Users).where(Users.id == user_id)) 

759 user = result.scalar_one_or_none() 

760 

761 if not user: 

762 raise NotFoundException("User not found") 

763 

764 if not user.status: 

765 raise AuthenticationException("Account is disabled") 

766 

767 # Mark token as used 

768 token_record.is_used = True 

769 

770 if verification_type == "registration": 

771 # Mark email as verified 

772 user.email_verified = True 

773 elif verification_type == "email_change": 

774 # Update email from pending_email to email 

775 if user.pending_email != email: 

776 raise AuthenticationException("Email mismatch") 

777 

778 # Check if new email already exists 

779 result = await db.execute( 

780 select(Users).where(Users.email == email, Users.id != user_id) 

781 ) 

782 if result.scalar_one_or_none(): 

783 raise ConflictException("Email already exists") 

784 

785 user.email = email 

786 user.pending_email = None 

787 user.email_verified = True 

788 

789 # Create session for verified user 

790 session_result = await _create_user_session(db, redis_client, user, ip_address, user_agent) 

791 

792 await db.commit() 

793 

794 return { 

795 "user": user, 

796 "session_id": session_result["session_id"], 

797 "access_token": session_result["access_token"], 

798 "csrf_token": session_result["csrf_token"], 

799 } 

800 

801 except AuthenticationException, NotFoundException, ConflictException: 

802 raise 

803 except Exception as e: 

804 raise ServerException(f"Failed to verify email: {str(e)}") 

805 

806 

807async def resend_verification_email( 

808 db: AsyncSession, 

809 email: str, 

810 mailer: SMTPMailer, 

811 redis_client: redis.Redis, 

812) -> dict: 

813 """Resend email verification""" 

814 try: 

815 user = await _get_user_by_email_for_password_reset(db, email) 

816 

817 if not settings.SMTP_ENABLE or not getattr(mailer, "enabled", False): 

818 raise SMTPNotConfiguredException("SMTP is disabled") 

819 

820 # Check cooldown 

821 cooldown_key = f"email_verification_cooldown:{email}" 

822 remaining_seconds = await redis_client.ttl(cooldown_key) 

823 

824 if remaining_seconds > 0: 

825 raise ValidationException( 

826 ( 

827 f"Please wait {remaining_seconds} seconds before requesting " 

828 "another verification email" 

829 ), 

830 details={"cooldown_seconds": remaining_seconds}, 

831 ) 

832 

833 # Determine verification type 

834 if user.email_verified: 

835 # If email is already verified but there's a pending email, resend email change 

836 # verification 

837 if user.pending_email: 

838 token_meta = await _request_email_change_verification_email( 

839 db, user, user.pending_email 

840 ) 

841 verification_url = ( 

842 f"http{'s' if settings.SSL_ENABLE else ''}://" 

843 f"{settings.HOSTNAME}:{settings.FRONTEND_PORT}" 

844 f"/auth/verify-email?token={quote(token_meta['verification_token'], safe='')}" 

845 ) 

846 

847 user_name = f"{user.first_name} {user.last_name}".strip()