xiaoxia-saas/packages/application/auth/password_reset_use_case.py

"""
密码重置 Use Case
"""

import logging
import secrets
from datetime import datetime, timedelta, timezone
from typing import Optional

from packages.adapters.smtp import get_email_service
from packages.application.auth.password_hasher import password_hasher, password_validator

logger = logging.getLogger(__name__)


class RequestPasswordResetRequest:
    """请求密码重置"""

    def __init__(self, email: str):
        self.email = email.strip().lower()


class RequestPasswordResetUseCase:
    """请求密码重置用例"""

    def __init__(
        self,
        user_repository,
        base_url: str,
        token_expire_hours: int = 1,
        email_service=None,
    ):
        """
        初始化请求密码重置用例

        Args:
            user_repository: 用户仓储
            base_url: 应用基础 URL（用于生成重置链接），必须由调用方传入
            token_expire_hours: 令牌过期时间（小时）
            email_service: 邮件服务
        """
        self.user_repository = user_repository
        self.base_url = base_url
        self.token_expire_hours = token_expire_hours
        self.email_service = email_service or get_email_service()

    def execute(self, request: RequestPasswordResetRequest) -> tuple[bool, Optional[str]]:
        """
        执行密码重置请求

        Args:
            request: 重置请求

        Returns:
            (是否成功, 错误信息)
        """
        try:
            if not request.email:
                return False, "Email is required"

            # 查找用户
            user = self.user_repository.find_by_email(request.email)

            # 安全考虑：即使用户不存在，也返回成功（避免暴露用户存在性）
            if not user:
                return True, None

            # 生成重置令牌
            reset_token = secrets.token_urlsafe(32)
            reset_url = f"{self.base_url}/reset-password?token={reset_token}"

            # 设置令牌和过期时间
            user.password_reset_token = reset_token
            user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=self.token_expire_hours)

            # 保存用户
            self.user_repository.save(user)

            # 发送重置邮件
            try:
                success, error = self.email_service.send_password_reset_email(
                    to_email=user.email,
                    username=user.username or user.display_name,
                    reset_url=reset_url,
                )

                if not success:
                    logger.warning(f"Password reset email failed: {error}")
                    # 不返回错误，避免暴露用户存在性
            except Exception as e:
                logger.error(f"Email service error: {e}", exc_info=True)

            return True, None

        except Exception as e:
            return False, f"Password reset request failed: {str(e)}"


class ResetPasswordRequest:
    """重置密码请求"""

    def __init__(self, token: str, new_password: str):
        self.token = token
        self.new_password = new_password


class ResetPasswordUseCase:
    """重置密码用例"""

    def __init__(self, user_repository):
        self.user_repository = user_repository

    def execute(self, request: ResetPasswordRequest) -> tuple[bool, Optional[str]]:
        """
        执行密码重置

        Args:
            request: 重置请求

        Returns:
            (是否成功, 错误信息)
        """
        try:
            if not request.token:
                return False, "Reset token is required"

            if not request.new_password:
                return False, "New password is required"

            # 验证新密码强度
            valid, error = password_validator.validate(request.new_password)
            if not valid:
                return False, error

            # 查找用户
            user = self.user_repository.find_by_password_reset_token(request.token)
            if not user:
                return False, "Invalid or expired reset token"

            # 检查令牌是否过期
            if user.password_reset_expires_at:
                expires_at = user.password_reset_expires_at
                if expires_at.tzinfo is None:
                    expires_at = expires_at.replace(tzinfo=timezone.utc)
                if datetime.now(timezone.utc) > expires_at:
                    return False, "Reset token has expired"

            # 哈希新密码
            hashed_password = password_hasher.hash_password(request.new_password)

            # 更新用户密码
            user.password_hash = hashed_password
            user.password_reset_token = None
            user.password_reset_expires_at = None

            # 保存用户
            self.user_repository.save(user)

            return True, None

        except Exception as e:
            return False, f"Password reset failed: {str(e)}"