Coverage for api/auth/controller.py: 96.89%
225 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-03 15:30 +0000
1import logging
2from datetime import datetime, timedelta
4from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
5from sqlalchemy.ext.asyncio import AsyncSession
7from core.config import settings
8from core.dependencies import get_db
9from core.redis import get_redis
10from core.security import verify_email_verification_token, verify_password_reset_token, verify_token
11from extensions.smtp import SMTPMailer, get_mailer
12from utils.custom_exception import (
13 AuthenticationException,
14 ConflictException,
15 EmailVerificationRequiredException,
16 NotFoundException,
17 PasswordResetRequiredException,
18 RegistrationDisabledException,
19 SMTPNotConfiguredException,
20 ValidationException,
21)
22from utils.get_real_ip import get_real_ip
23from utils.response import APIResponse, common_responses, make_error_examples, parse_responses
25from .schema import (
26 ActionRequiredResponse,
27 CsrfTokenResponse,
28 ForgotPasswordRequest,
29 LogoutRequest,
30 PasswordResetCooldownResponse,
31 ResendVerificationRequest,
32 ResetPasswordRequest,
33 TokenResponse,
34 TokenValidationResponse,
35 UserLogin,
36 UserLoginResponse,
37 UserRegister,
38 UserResponse,
39 action_required_response_examples,
40)
41from .services import (
42 forgot_password,
43 get_or_create_csrf_token,
44 get_password_reset_cooldown,
45 login,
46 logout,
47 logout_all_devices,
48 register,
49 resend_verification_email,
50 reset_password,
51 token,
52 validate_password_reset_token,
53 verify_csrf_token,
54 verify_email,
55)
57logger = logging.getLogger("auth")
58router = APIRouter(tags=["Auth"])
61@router.post(
62 "/register",
63 response_model=APIResponse[UserLoginResponse],
64 response_model_exclude_none=True,
65 summary="Register account",
66 responses=parse_responses(
67 {
68 200: ("User registered successfully", UserLoginResponse),
69 202: ("Email verification required", None),
70 409: ("Email already exists", None),
71 503: ("Registration is disabled", None),
72 },
73 common_responses,
74 ),
75)
76async def register_api(
77 user_data: UserRegister,
78 request: Request,
79 response: Response,
80 db: AsyncSession = Depends(get_db),
81 redis_client=Depends(get_redis),
82 mailer: SMTPMailer = Depends(get_mailer),
83):
84 try:
85 client_ip = get_real_ip(request)
86 user_agent = request.headers.get("user-agent", "Registration")
88 result = await register(db, redis_client, user_data, client_ip, user_agent, mailer)
90 user = result["user"]
91 session_id = result["session_id"]
92 access_token = result["access_token"]
93 csrf_token = result["csrf_token"]
95 _set_session_cookie(response, session_id)
96 _set_csrf_cookie(response, csrf_token)
98 user_response = UserResponse(
99 id=user.id,
100 first_name=user.first_name,
101 last_name=user.last_name,
102 email=user.email,
103 phone=user.phone,
104 )
106 response_data = UserLoginResponse(
107 access_token=access_token,
108 expires_at=datetime.now().astimezone()
109 + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
110 user=user_response,
111 )
112 return APIResponse(code=200, message="User registered successfully", data=response_data)
113 except EmailVerificationRequiredException:
114 resp = APIResponse(code=202, message="Email verification required")
115 raise HTTPException(status_code=202, detail=resp.dict(exclude_none=True))
116 except ConflictException:
117 raise HTTPException(status_code=409, detail="Email already exists")
118 except RegistrationDisabledException:
119 raise HTTPException(status_code=503, detail="Registration is disabled")
120 except Exception:
121 raise HTTPException(status_code=500)
124@router.post(
125 "/login",
126 response_model=APIResponse[UserLoginResponse],
127 response_model_exclude_none=True,
128 summary="Login account",
129 responses=parse_responses(
130 {
131 200: ("User logged in successfully", UserLoginResponse),
132 202: ("Action required", ActionRequiredResponse, action_required_response_examples),
133 401: ("Invalid email or password", None),
134 },
135 common_responses,
136 ),
137)
138async def login_api(
139 user_data: UserLogin,
140 request: Request,
141 response: Response,
142 db: AsyncSession = Depends(get_db),
143 redis_client=Depends(get_redis),
144 mailer: SMTPMailer = Depends(get_mailer),
145):
146 try:
147 client_ip = get_real_ip(request)
148 user_agent = request.headers.get("user-agent", "")
150 result = await login(db, redis_client, user_data, client_ip, user_agent, mailer)
152 user = result["user"]
153 session_id = result["session_id"]
154 access_token = result["access_token"]
155 csrf_token = result["csrf_token"]
157 user_response = UserResponse(
158 id=user.id,
159 first_name=user.first_name,
160 last_name=user.last_name,
161 email=user.email,
162 phone=user.phone,
163 )
165 _set_session_cookie(response, session_id)
166 _set_csrf_cookie(response, csrf_token)
168 response_data = UserLoginResponse(
169 access_token=access_token,
170 expires_at=datetime.now().astimezone()
171 + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
172 user=user_response,
173 )
174 return APIResponse(code=200, message="User logged in successfully", data=response_data)
175 except PasswordResetRequiredException as e:
176 resp = APIResponse(code=202, message="Password reset required", data=e.details)
177 raise HTTPException(status_code=202, detail=resp.dict(exclude_none=True))
178 except EmailVerificationRequiredException as e:
179 resp = APIResponse(code=202, message="Email verification required", data=e.details)
180 raise HTTPException(status_code=202, detail=resp.dict(exclude_none=True))
181 except AuthenticationException:
182 raise HTTPException(status_code=401, detail="Invalid email or password")
183 except Exception:
184 raise HTTPException(status_code=500)
187@router.post(
188 "/logout",
189 response_model=APIResponse[None],
190 response_model_exclude_none=True,
191 summary="Logout account",
192 responses=parse_responses(
193 {
194 200: ("User logged out successfully", None),
195 401: (
196 "Unauthorized",
197 None,
198 make_error_examples(
199 401,
200 {
201 "invalidSession": "Invalid or expired session",
202 "invalidToken": "Invalid or expired token",
203 },
204 ),
205 ),
206 },
207 common_responses,
208 ),
209)
210async def logout_api(
211 logout_data: LogoutRequest,
212 token: dict = Depends(verify_token),
213 response: Response = None,
214 db: AsyncSession = Depends(get_db),
215 redis_client=Depends(get_redis),
216):
217 """
218 Logout user from current device or all devices
220 Args:
221 logout_data: Contains logout_all flag to determine logout scope
222 """
223 try:
224 user_id = token.get("sub")
225 session_id = token.get("sid")
227 if logout_data.logout_all:
228 # Logout from all devices
229 if await logout_all_devices(db, redis_client, user_id):
230 if response:
231 _clear_auth_cookies(response)
232 return APIResponse(code=200, message="User logged out successfully")
233 else:
234 # Logout from current device only
235 if not session_id:
236 raise AuthenticationException("Invalid or expired session")
238 if await logout(db, redis_client, user_id, session_id):
239 if response:
240 _clear_auth_cookies(response)
241 return APIResponse(code=200, message="User logged out successfully")
242 except AuthenticationException:
243 raise HTTPException(status_code=401, detail="Invalid or expired session")
244 except Exception:
245 raise HTTPException(status_code=500)
248@router.post(
249 "/token",
250 response_model=APIResponse[TokenResponse],
251 response_model_exclude_unset=True,
252 summary="Refresh token",
253 responses=parse_responses(
254 {
255 200: ("Token refreshed successfully", TokenResponse),
256 401: (
257 "Unauthorized",
258 None,
259 make_error_examples(
260 401,
261 {
262 "invalidSession": "Invalid or expired session",
263 "invalidCsrf": "Invalid or expired CSRF token",
264 },
265 ),
266 ),
267 },
268 common_responses,
269 ),
270)
271async def token_api(
272 request: Request, db: AsyncSession = Depends(get_db), redis_client=Depends(get_redis)
273):
274 """Validate CSRF token then use session_id cookie to issue new access_token"""
275 try:
276 csrf_token = request.headers.get("X-CSRF-Token")
277 sid_from_csrf = await verify_csrf_token(redis_client, csrf_token)
279 session_id = request.cookies.get("session_id")
280 if not session_id or session_id != sid_from_csrf:
281 raise AuthenticationException("Invalid or expired session")
283 new_access_token = await token(db, redis_client, session_id)
284 response_data = TokenResponse(
285 access_token=new_access_token,
286 expires_at=datetime.now().astimezone()
287 + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
288 )
289 return APIResponse(code=200, message="Token refreshed successfully", data=response_data)
290 except AuthenticationException as e:
291 raise HTTPException(status_code=401, detail=e.message)
292 except NotFoundException:
293 raise HTTPException(status_code=401, detail="Invalid or expired session")
294 except Exception:
295 raise HTTPException(status_code=500)
298@router.post(
299 "/csrf-token",
300 response_model=APIResponse[CsrfTokenResponse],
301 response_model_exclude_none=True,
302 summary="Get CSRF token",
303 responses=parse_responses(
304 {
305 200: ("CSRF token retrieved successfully", CsrfTokenResponse),
306 401: ("Invalid or expired session", None),
307 },
308 common_responses,
309 ),
310)
311async def csrf_token_api(
312 request: Request,
313 response: Response,
314 redis_client=Depends(get_redis),
315):
316 """Issue CSRF token for the current session (cookie-based)."""
317 try:
318 session_id = request.cookies.get("session_id")
319 if not session_id:
320 raise AuthenticationException("Invalid or expired session")
322 csrf_token = await get_or_create_csrf_token(redis_client, session_id)
323 _set_csrf_cookie(response, csrf_token)
325 response_data = CsrfTokenResponse(
326 csrf_token=csrf_token,
327 expires_at=datetime.now().astimezone()
328 + timedelta(minutes=settings.CSRF_TOKEN_EXPIRE_MINUTES),
329 )
330 return APIResponse(
331 code=200, message="CSRF token retrieved successfully", data=response_data
332 )
333 except AuthenticationException:
334 raise HTTPException(status_code=401, detail="Invalid or expired session")
335 except Exception:
336 raise HTTPException(status_code=500)
339@router.post(
340 "/reset-password",
341 response_model=APIResponse[UserLoginResponse],
342 response_model_exclude_none=True,
343 summary="Reset password with token",
344 responses=parse_responses(
345 {
346 200: ("Password reset successfully", UserLoginResponse),
347 401: ("Invalid or expired token", None),
348 404: ("User not found", None),
349 },
350 common_responses,
351 ),
352)
353async def reset_password_api(
354 request: Request,
355 response: Response,
356 request_data: ResetPasswordRequest,
357 token: dict = Depends(verify_password_reset_token),
358 db: AsyncSession = Depends(get_db),
359 redis_client=Depends(get_redis),
360):
361 """Reset password using token"""
362 try:
363 client_ip = get_real_ip(request)
364 user_agent = request.headers.get("user-agent", "")
366 result = await reset_password(
367 db, redis_client, token, request_data.new_password, client_ip, user_agent
368 )
370 user = result["user"]
371 session_id = result["session_id"]
372 access_token = result["access_token"]
373 csrf_token = result["csrf_token"]
375 user_response = UserResponse(
376 id=user.id,
377 first_name=user.first_name,
378 last_name=user.last_name,
379 email=user.email,
380 phone=user.phone,
381 )
383 response_data = UserLoginResponse(
384 access_token=access_token,
385 expires_at=datetime.now().astimezone()
386 + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
387 user=user_response,
388 )
390 _set_session_cookie(response, session_id)
391 _set_csrf_cookie(response, csrf_token)
393 return APIResponse(code=200, message="Password reset successfully", data=response_data)
394 except AuthenticationException:
395 raise HTTPException(status_code=401, detail="Invalid or expired token")
396 except NotFoundException:
397 raise HTTPException(status_code=404, detail="User not found")
398 except Exception:
399 raise HTTPException(status_code=500)
402@router.get(
403 "/validate-reset-token",
404 response_model=APIResponse[TokenValidationResponse],
405 response_model_exclude_none=True,
406 summary="Validate password reset token",
407 responses=parse_responses({200: ("Token is valid", TokenValidationResponse)}, common_responses),
408)
409async def validate_reset_token_api(
410 token: dict = Depends(verify_password_reset_token), db: AsyncSession = Depends(get_db)
411):
412 """Validate password reset token without consuming it"""
413 try:
414 result = await validate_password_reset_token(db, token)
415 return APIResponse(code=200, message="Token is valid", data=result)
416 except AuthenticationException:
417 raise HTTPException(status_code=401, detail="Invalid or expired token")
418 except Exception:
419 raise HTTPException(status_code=500)
422@router.post(
423 "/forgot-password",
424 response_model=APIResponse[None],
425 response_model_exclude_none=True,
426 summary="Send reset password email",
427 responses=parse_responses(
428 {
429 200: ("Reset password email sent", None),
430 400: ("Please wait before requesting another password reset email", None),
431 403: ("Account is disabled", None),
432 404: ("User not registered", None),
433 503: ("SMTP is disabled", None),
434 },
435 common_responses,
436 ),
437)
438async def forgot_password_api(
439 request: Request,
440 request_data: ForgotPasswordRequest,
441 db: AsyncSession = Depends(get_db),
442 mailer: SMTPMailer = Depends(get_mailer),
443 redis_client=Depends(get_redis),
444):
445 """
446 Send password reset email based on input email.
447 """
448 try:
449 await forgot_password(db, request_data.email, mailer, redis_client)
450 return APIResponse(code=200, message="Reset password email sent")
451 except ValidationException:
452 raise HTTPException(
453 status_code=400, detail="Please wait before requesting another password reset email"
454 )
455 except AuthenticationException:
456 raise HTTPException(status_code=403, detail="Account is disabled")
457 except NotFoundException:
458 raise HTTPException(status_code=404, detail="User not registered")
459 except SMTPNotConfiguredException:
460 raise HTTPException(status_code=503, detail="SMTP is disabled")
461 except Exception:
462 raise HTTPException(status_code=500)
465@router.get(
466 "/forgot-password/cooldown",
467 response_model=APIResponse[PasswordResetCooldownResponse],
468 response_model_exclude_none=True,
469 summary="Get password reset email cooldown status",
470 responses=parse_responses(
471 {
472 200: ("Cooldown status retrieved", PasswordResetCooldownResponse),
473 },
474 common_responses,
475 ),
476)
477async def get_password_reset_cooldown_api(
478 email: str = Query(..., description="Email address to check cooldown for"),
479 redis_client=Depends(get_redis),
480):
481 """
482 Get remaining cooldown time for password reset email.
483 Returns 0 if no cooldown is active.
484 """
485 try:
486 result = await get_password_reset_cooldown(email, redis_client)
487 response_data = PasswordResetCooldownResponse(cooldown_seconds=result["cooldown_seconds"])
488 return APIResponse(code=200, message="Cooldown status retrieved", data=response_data)
489 except Exception:
490 raise HTTPException(status_code=500)
493@router.get(
494 "/verify-email",
495 response_model=APIResponse[UserLoginResponse],
496 response_model_exclude_none=True,
497 summary="Verify email address",
498 responses=parse_responses(
499 {
500 200: ("Email verified successfully", UserLoginResponse),
501 401: ("Invalid or expired token", None),
502 404: ("User not found", None),
503 409: ("Email already exists", None),
504 },
505 common_responses,
506 ),
507)
508async def verify_email_api(
509 request: Request,
510 response: Response,
511 token: dict = Depends(verify_email_verification_token),
512 db: AsyncSession = Depends(get_db),
513 redis_client=Depends(get_redis),
514):
515 """Verify email address using token and create session"""
516 try:
517 client_ip = get_real_ip(request)
518 user_agent = request.headers.get("user-agent", "")
520 result = await verify_email(db, redis_client, token, client_ip, user_agent)
522 user = result["user"]
523 session_id = result["session_id"]
524 access_token = result["access_token"]
525 csrf_token = result["csrf_token"]
527 user_response = UserResponse(
528 id=user.id,
529 first_name=user.first_name,
530 last_name=user.last_name,
531 email=user.email,
532 phone=user.phone,
533 )
535 response_data = UserLoginResponse(
536 access_token=access_token,
537 expires_at=datetime.now().astimezone()
538 + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
539 user=user_response,
540 )
542 _set_session_cookie(response, session_id)
543 _set_csrf_cookie(response, csrf_token)
545 return APIResponse(code=200, message="Email verified successfully", data=response_data)
546 except AuthenticationException:
547 raise HTTPException(status_code=401, detail="Invalid or expired token")
548 except NotFoundException:
549 raise HTTPException(status_code=404, detail="User not found")
550 except ConflictException:
551 raise HTTPException(status_code=409, detail="Email already exists")
552 except Exception:
553 raise HTTPException(status_code=500)
556@router.post(
557 "/resend-verification",
558 response_model=APIResponse[None],
559 response_model_exclude_none=True,
560 summary="Resend email verification",
561 responses=parse_responses(
562 {
563 200: ("Verification email sent", None),
564 400: ("Please wait before requesting another verification email", None),
565 403: ("Account is disabled", None),
566 404: ("User not registered", None),
567 503: ("SMTP is disabled", None),
568 },
569 common_responses,
570 ),
571)
572async def resend_verification_api(
573 request: Request,
574 request_data: ResendVerificationRequest,
575 db: AsyncSession = Depends(get_db),
576 mailer: SMTPMailer = Depends(get_mailer),
577 redis_client=Depends(get_redis),
578):
579 """Resend email verification email"""
580 try:
581 await resend_verification_email(db, request_data.email, mailer, redis_client)
582 return APIResponse(code=200, message="Verification email sent")
583 except ValidationException:
584 raise HTTPException(
585 status_code=400, detail="Please wait before requesting another verification email"
586 )
587 except AuthenticationException:
588 raise HTTPException(status_code=403, detail="Account is disabled")
589 except NotFoundException:
590 raise HTTPException(status_code=404, detail="User not registered")
591 except SMTPNotConfiguredException:
592 raise HTTPException(status_code=503, detail="SMTP is disabled")
593 except Exception:
594 raise HTTPException(status_code=500)
597@router.get(
598 "/resend-verification/cooldown",
599 response_model=APIResponse[PasswordResetCooldownResponse],
600 response_model_exclude_none=True,
601 summary="Get email verification cooldown status",
602 responses=parse_responses(
603 {
604 200: ("Cooldown status retrieved", PasswordResetCooldownResponse),
605 },
606 common_responses,
607 ),
608)
609async def get_email_verification_cooldown_api(
610 email: str = Query(..., description="Email address to check cooldown for"),
611 redis_client=Depends(get_redis),
612):
613 """
614 Get remaining cooldown time for email verification email.
615 Returns 0 if no cooldown is active.
616 """
617 try:
618 cooldown_key = f"email_verification_cooldown:{email}"
619 remaining_seconds = await redis_client.ttl(cooldown_key)
621 # TTL returns -1 if key exists but has no expiry, -2 if key doesn't exist
622 if remaining_seconds < 0:
623 remaining_seconds = 0
625 response_data = PasswordResetCooldownResponse(cooldown_seconds=remaining_seconds)
626 return APIResponse(code=200, message="Cooldown status retrieved", data=response_data)
627 except Exception:
628 raise HTTPException(status_code=500)
631def _set_session_cookie(response: Response, session_id: str) -> None:
632 response.set_cookie(
633 key="session_id",
634 value=session_id,
635 httponly=settings.COOKIE_HTTPONLY,
636 secure=settings.COOKIE_SECURE,
637 samesite=settings.COOKIE_SAMESITE,
638 max_age=settings.SESSION_EXPIRE_MINUTES * 60,
639 )
642def _set_csrf_cookie(response: Response, csrf_token: str) -> None:
643 response.set_cookie(
644 key="csrf_token",
645 value=csrf_token,
646 httponly=False,
647 secure=settings.COOKIE_SECURE,
648 samesite=settings.COOKIE_SAMESITE,
649 max_age=settings.CSRF_TOKEN_EXPIRE_MINUTES * 60,
650 )
653def _clear_auth_cookies(response: Response) -> None:
654 response.delete_cookie("session_id")
655 response.delete_cookie("csrf_token")