""" Authentication business logic service. Handles user credentials verification, JWT token issuance, PostgreSQL session tracking, token rotation, and secure revocation on logout. """ import asyncio import datetime import logging import re import secrets from uuid import UUID from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.core.exceptions import ( BadRequestException, ForbiddenException, NotFoundException, RateLimitException, ServiceUnavailableException, UnauthorizedException, ValidationException, ) from app.core.security import ( create_access_token, create_refresh_token, decode_access_token, decode_refresh_token, hash_password, verify_password, ) from app.core.state_store import ( blacklist_access_token, check_forgot_password_rate_limit, delete_password_reset_otp, is_refresh_token_valid, revoke_refresh_token, store_password_reset_otp, store_refresh_token, verify_password_reset_otp, ) from app.models.lab import Lab from app.models.user import User, UserRole from app.repositories.user_repository import UserRepository from app.schemas.auth import ( BootstrapRequest, BootstrapResponse, ForgotPasswordRequest, ForgotPasswordResponse, LoginRequest, MessageResponse, RefreshRequest, ResetPasswordRequest, TokenResponse, UserMeResponse, ) from app.services.email_service import EmailService _logger = logging.getLogger(__name__) def _mask_email_for_display(email: str) -> str: """Return a privacy-safe hint, e.g. m***@yopmail.com.""" local, _, domain = email.partition("@") if not domain: return email visible = local[:1] if local else "" return f"{visible}***@{domain}" class AuthService: """Service handling authentication logic and token lifecycle.""" def __init__( self, user_repository: UserRepository | None = None, email_service: EmailService | None = None, ) -> None: self.user_repo = user_repository or UserRepository() self.email_service = email_service or EmailService() async def login(self, db: AsyncSession, data: LoginRequest) -> TokenResponse: """ Authenticate a staff member by phone or email, and password. Validates credentials, active state, and saves refresh session into PostgreSQL. """ identifier = data.identifier if not identifier: raise UnauthorizedException(message="Please provide a phone number or email to log in.") user = await self.user_repo.get_by_identifier(db, identifier=identifier) if not user or not await asyncio.to_thread(verify_password, data.password, user.hashed_password): raise UnauthorizedException(message="Invalid phone number, email, or password.") if not user.is_active: raise UnauthorizedException( message="Your account is deactivated. Please contact your administrator." ) # Issue tokens access_token = create_access_token( subject=str(user.id), extra_claims={ "lab_id": str(user.lab_id), "role": str(user.role.value if hasattr(user.role, "value") else user.role), "name": user.name, }, ) refresh_token = create_refresh_token(subject=str(user.id)) # Store refresh token in PostgreSQL for stateful revocation ttl = settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS * 86400 await store_refresh_token(str(user.id), refresh_token, expires_in_seconds=ttl) return TokenResponse( access_token=access_token, refresh_token=refresh_token, token_type="bearer", expires_in=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60, ) async def refresh(self, db: AsyncSession, data: RefreshRequest) -> TokenResponse: """ Issue a new access token and rotate refresh token. Verifies the refresh token against both JWT claims and PostgreSQL state. """ payload = decode_refresh_token(data.refresh_token) user_id_str = str(payload.get("sub", "")) if not user_id_str: raise UnauthorizedException(message="Invalid refresh token payload.") # Assert token is active in PostgreSQL (not logged out or replaced) valid = await is_refresh_token_valid(user_id_str, data.refresh_token) if not valid: raise UnauthorizedException( message="Refresh token has expired or been revoked. Please log in again." ) try: user_uuid = UUID(user_id_str) except ValueError as exc: raise UnauthorizedException(message="Invalid user identifier in token.") from exc user = await self.user_repo.get_by_id(db, user_uuid) if not user or not user.is_active: raise UnauthorizedException(message="User account no longer exists or has been deactivated.") # Token rotation: create both a new access token and a new refresh token new_access_token = create_access_token( subject=str(user.id), extra_claims={ "lab_id": str(user.lab_id), "role": str(user.role.value if hasattr(user.role, "value") else user.role), "name": user.name, }, ) new_refresh_token = create_refresh_token(subject=str(user.id)) ttl = settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS * 86400 await store_refresh_token(str(user.id), new_refresh_token, expires_in_seconds=ttl) return TokenResponse( access_token=new_access_token, refresh_token=new_refresh_token, token_type="bearer", expires_in=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60, ) async def logout(self, user_id: UUID, token: str | None = None) -> None: """Invalidate user's active refresh session and blacklist access token in PostgreSQL.""" await revoke_refresh_token(str(user_id)) if token: try: payload = decode_access_token(token) exp = payload.get("exp") ttl = settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60 if exp and isinstance(exp, int | float): remaining = int(float(exp) - datetime.datetime.now(datetime.UTC).timestamp()) if remaining > 0: ttl = remaining await blacklist_access_token(token, expires_in_seconds=ttl) except Exception: await blacklist_access_token( token, expires_in_seconds=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60 ) async def bootstrap(self, db: AsyncSession, data: BootstrapRequest) -> BootstrapResponse: """ Bootstrap the first laboratory branch and initial Superadmin user. Security: Strictly one-time only. Once any user exists in the database, all subsequent bootstrap attempts are rejected with 403 Forbidden. """ count_stmt = select(func.count()).select_from(User) count_res = await db.execute(count_stmt) total_users = int(count_res.scalar_one() or 0) if total_users > 0: raise ForbiddenException( message=( "Initial setup has already been completed. " "Account creation via bootstrap is permanently disabled for security. " "Please log in or contact an existing administrator." ) ) # 1. Ensure lab exists or create one lab_stmt = select(Lab).where(Lab.deleted_at.is_(None)).limit(1) lab_res = await db.execute(lab_stmt) lab = lab_res.scalars().first() if not lab: lab = Lab( name=data.lab_name, address="Kathmandu, Nepal", ) db.add(lab) await db.flush() await db.refresh(lab) # 2. Create the superadmin user hashed_password = await asyncio.to_thread(hash_password, data.password) superadmin = User( lab_id=lab.id, name=data.admin_name, phone=data.admin_phone, email=data.admin_email, role=UserRole.SUPERADMIN, hashed_password=hashed_password, is_active=True, ) db.add(superadmin) await db.flush() await db.refresh(superadmin) # 3. Generate tokens access_token = create_access_token( subject=str(superadmin.id), extra_claims={ "lab_id": str(superadmin.lab_id), "role": str(superadmin.role.value if hasattr(superadmin.role, "value") else superadmin.role), "name": superadmin.name, }, ) refresh_token = create_refresh_token(subject=str(superadmin.id)) ttl = settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS * 86400 await store_refresh_token(str(superadmin.id), refresh_token, expires_in_seconds=ttl) return BootstrapResponse( user=UserMeResponse( id=superadmin.id, name=superadmin.name, phone=superadmin.phone, email=superadmin.email, role=str(superadmin.role.value if hasattr(superadmin.role, "value") else superadmin.role), lab_id=superadmin.lab_id, is_active=superadmin.is_active, ), tokens=TokenResponse( access_token=access_token, refresh_token=refresh_token, token_type="bearer", expires_in=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60, ), ) async def forgot_password(self, db: AsyncSession, data: ForgotPasswordRequest) -> ForgotPasswordResponse: """ Initiate a password reset flow by sending a 6-digit OTP to the registered email. Returns explicit outcomes for unknown email, inactive accounts, and delivery failures. """ email_str = str(data.email).strip().lower() is_allowed = await check_forgot_password_rate_limit(email_str, max_requests=5, window_seconds=900) if not is_allowed: raise RateLimitException( message="Too many password reset requests. Please wait 15 minutes before trying again." ) user = await self.user_repo.get_by_email_ci(db, email=email_str) if user is None: raise NotFoundException( message=( "No staff account is registered with this email address. " "Check the spelling or contact your lab administrator." ) ) if not user.is_active: raise ForbiddenException( message="This staff account is deactivated. Contact your lab administrator to reactivate it." ) otp_int = secrets.randbelow(1_000_000) otp_str = f"{otp_int:06d}" await store_password_reset_otp( email=email_str, otp=otp_str, user_id=str(user.id), expires_in_seconds=300, ) html_body = self.email_service.render_template( "password_reset_otp.html", {"user_name": user.name, "otp": otp_str}, ) success, error_msg = await self.email_service.send_email( recipient=email_str, subject="Security Verification Code — Munal LMS", html_content=html_body, dev_otp=otp_str, ) if not success: await delete_password_reset_otp(email_str) raise ServiceUnavailableException( message=error_msg or ( "We could not send the verification email. Configure outbound SMTP on the server " "(required for inboxes such as yopmail.com) or try again shortly." ) ) masked = _mask_email_for_display(email_str) message = ( f"A 6-digit verification code was sent to {masked}. " "Check your inbox and spam folder. The code expires in 5 minutes." ) if settings.smtp_credentials_are_placeholder: message += ( " This server is not using live outbound SMTP: open http://localhost:8025 (Mailpit) " "or the API server logs for the code—it will not arrive at yopmail.com " "until SMTP is configured." ) return ForgotPasswordResponse(message=message, sent=True) async def reset_password(self, db: AsyncSession, data: ResetPasswordRequest) -> MessageResponse: """ Verify the OTP and update the user's password. Validates password policy, checks OTP expiration and attempts, updates password hash, and invalidates active sessions. """ email_str = str(data.email).strip().lower() # 1. Validate password policy: min 8 chars, mixed case, numbers, special characters password = data.new_password if len(password) < 8: raise ValidationException(message="Password must be at least 8 characters long.") if not re.search(r"[A-Z]", password): raise ValidationException(message="Password must contain at least one uppercase letter.") if not re.search(r"[a-z]", password): raise ValidationException(message="Password must contain at least one lowercase letter.") if not re.search(r"\d", password): raise ValidationException(message="Password must contain at least one number.") if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", password): raise ValidationException(message="Password must contain at least one special character.") # Lock and consume the code atomically across all API workers. verification = await verify_password_reset_otp(email_str, data.otp) if verification.status == "expired": raise BadRequestException(message="Invalid or expired OTP passcode. Please request a new code.") if verification.status == "exhausted": raise BadRequestException( message="Maximum verification attempts exceeded. Please request a new code." ) if verification.status == "invalid": raise BadRequestException( message=f"Invalid OTP passcode. {verification.remaining_attempts} attempt(s) remaining." ) # 5. Look up user user = None user_id_str = verification.user_id if user_id_str: try: user = await self.user_repo.get_by_id(db, UUID(user_id_str)) except (ValueError, TypeError): pass if not user: user = await self.user_repo.get_by_email_ci(db, email=email_str) if not user or not user.is_active: await delete_password_reset_otp(email_str) raise BadRequestException(message="User account no longer exists or has been deactivated.") # 6. Hash new password and persist new_hashed = await asyncio.to_thread(hash_password, password) await self.user_repo.update_password(db, user, new_hashed) await db.commit() # 7. Cleanup OTP and invalidate active user sessions await delete_password_reset_otp(email_str) await revoke_refresh_token(str(user.id)) return MessageResponse( message="Password updated successfully. Please sign in with your new credentials." )