diff --git a/packages/application/auth/__init__.py b/packages/application/auth/__init__.py index 6debd1bdc..ea8e45585 100644 --- a/packages/application/auth/__init__.py +++ b/packages/application/auth/__init__.py @@ -15,6 +15,12 @@ from packages.application.auth.login_use_case import ( LogoutUseCase, LogoutRequest, ) +from packages.application.auth.password_reset_use_case import ( + RequestPasswordResetUseCase, + RequestPasswordResetRequest, + ResetPasswordUseCase, + ResetPasswordRequest, +) __all__ = [ "RegisterUserUseCase", @@ -29,4 +35,8 @@ __all__ = [ "RefreshTokenRequest", "LogoutUseCase", "LogoutRequest", + "RequestPasswordResetUseCase", + "RequestPasswordResetRequest", + "ResetPasswordUseCase", + "ResetPasswordRequest", ] diff --git a/packages/application/auth/password_reset_use_case.py b/packages/application/auth/password_reset_use_case.py new file mode 100644 index 000000000..31942d71d --- /dev/null +++ b/packages/application/auth/password_reset_use_case.py @@ -0,0 +1,145 @@ +""" +密码重置 Use Case +""" +import secrets +from datetime import datetime, timedelta, timezone +from typing import Optional + +from packages.domain.auth import password_hasher, password_validator, email_service + + +class RequestPasswordResetRequest: + """请求密码重置""" + + def __init__(self, email: str): + self.email = email.strip().lower() + + +class RequestPasswordResetUseCase: + """请求密码重置用例""" + + def __init__( + self, + user_repository, + base_url: str = "http://localhost:3000", + token_expire_hours: int = 1, + ): + self.user_repository = user_repository + self.base_url = base_url + self.token_expire_hours = token_expire_hours + + 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 = email_service.send_password_reset_email( + to_email=user.email, + username=user.username or user.display_name, + reset_url=reset_url, + ) + + if not success: + print(f"Failed to send password reset email: {error}") + # 不返回错误,避免暴露用户存在性 + except Exception as e: + print(f"Email service error: {e}") + + 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: + if datetime.now(timezone.utc) > user.password_reset_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)}" diff --git a/tests/unit/test_password_reset_use_case.py b/tests/unit/test_password_reset_use_case.py new file mode 100644 index 000000000..426900b43 --- /dev/null +++ b/tests/unit/test_password_reset_use_case.py @@ -0,0 +1,197 @@ +""" +密码重置 Use Case 测试 +""" +import pytest +from unittest.mock import Mock, patch +from datetime import datetime, timedelta, timezone +from packages.application.auth.password_reset_use_case import ( + RequestPasswordResetUseCase, + RequestPasswordResetRequest, + ResetPasswordUseCase, + ResetPasswordRequest, +) +from packages.domain.entities import User + + +class TestRequestPasswordResetUseCase: + """请求密码重置测试""" + + @pytest.fixture + def mock_user_repo(self): + repo = Mock() + repo.find_by_email = Mock(return_value=None) + repo.save = Mock() + return repo + + @pytest.fixture + def use_case(self, mock_user_repo): + return RequestPasswordResetUseCase( + user_repository=mock_user_repo, + base_url="https://test.com", + token_expire_hours=1, + ) + + @pytest.fixture + def test_user(self): + return User( + id="user-123", + email="test@example.com", + username="testuser", + display_name="Test User", + password_hash="hash", + ) + + @patch('packages.application.auth.password_reset_use_case.email_service') + def test_request_reset_success(self, mock_email_service, use_case, mock_user_repo, test_user): + """测试请求重置成功""" + mock_user_repo.find_by_email.return_value = test_user + mock_email_service.send_password_reset_email.return_value = (True, None) + + request = RequestPasswordResetRequest(email="test@example.com") + success, error = use_case.execute(request) + + assert success is True + assert error is None + + # 验证保存了用户 + mock_user_repo.save.assert_called_once() + saved_user = mock_user_repo.save.call_args[0][0] + assert saved_user.password_reset_token is not None + assert saved_user.password_reset_expires_at is not None + + # 验证发送了邮件 + mock_email_service.send_password_reset_email.assert_called_once() + + @patch('packages.application.auth.password_reset_use_case.email_service') + def test_request_reset_user_not_exists(self, mock_email_service, use_case, mock_user_repo): + """测试用户不存在(仍返回成功,避免暴露)""" + mock_user_repo.find_by_email.return_value = None + + request = RequestPasswordResetRequest(email="nonexistent@example.com") + success, error = use_case.execute(request) + + assert success is True # 安全考虑,仍返回成功 + assert error is None + + # 不发送邮件 + mock_email_service.send_password_reset_email.assert_not_called() + + def test_request_reset_missing_email(self, use_case): + """测试缺少邮箱""" + request = RequestPasswordResetRequest(email="") + success, error = use_case.execute(request) + + assert success is False + assert error == "Email is required" + + +class TestResetPasswordUseCase: + """重置密码测试""" + + @pytest.fixture + def mock_user_repo(self): + repo = Mock() + repo.find_by_password_reset_token = Mock(return_value=None) + repo.save = Mock() + return repo + + @pytest.fixture + def use_case(self, mock_user_repo): + return ResetPasswordUseCase(user_repository=mock_user_repo) + + @pytest.fixture + def test_user(self): + return User( + id="user-123", + email="test@example.com", + username="testuser", + display_name="Test User", + password_hash="old-hash", + password_reset_token="valid-token", + password_reset_expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + ) + + def test_reset_password_success(self, use_case, mock_user_repo, test_user): + """测试重置密码成功""" + mock_user_repo.find_by_password_reset_token.return_value = test_user + + request = ResetPasswordRequest( + token="valid-token", + new_password="NewSecurePass123", + ) + success, error = use_case.execute(request) + + assert success is True + assert error is None + + # 验证密码已更新 + assert test_user.password_hash != "old-hash" + assert test_user.password_reset_token is None + assert test_user.password_reset_expires_at is None + + # 验证保存了用户 + mock_user_repo.save.assert_called_once() + + def test_reset_password_weak_password(self, use_case, mock_user_repo, test_user): + """测试弱密码""" + mock_user_repo.find_by_password_reset_token.return_value = test_user + + request = ResetPasswordRequest( + token="valid-token", + new_password="weak", + ) + success, error = use_case.execute(request) + + assert success is False + assert "at least 8 characters" in error + + def test_reset_password_invalid_token(self, use_case, mock_user_repo): + """测试无效令牌""" + mock_user_repo.find_by_password_reset_token.return_value = None + + request = ResetPasswordRequest( + token="invalid-token", + new_password="NewSecurePass123", + ) + success, error = use_case.execute(request) + + assert success is False + assert error == "Invalid or expired reset token" + + def test_reset_password_expired_token(self, use_case, mock_user_repo, test_user): + """测试过期令牌""" + test_user.password_reset_expires_at = datetime.now(timezone.utc) - timedelta(hours=1) + mock_user_repo.find_by_password_reset_token.return_value = test_user + + request = ResetPasswordRequest( + token="valid-token", + new_password="NewSecurePass123", + ) + success, error = use_case.execute(request) + + assert success is False + assert error == "Reset token has expired" + + def test_reset_password_missing_token(self, use_case): + """测试缺少令牌""" + request = ResetPasswordRequest( + token="", + new_password="NewSecurePass123", + ) + success, error = use_case.execute(request) + + assert success is False + assert error == "Reset token is required" + + def test_reset_password_missing_password(self, use_case, mock_user_repo, test_user): + """测试缺少新密码""" + mock_user_repo.find_by_password_reset_token.return_value = test_user + + request = ResetPasswordRequest( + token="valid-token", + new_password="", + ) + success, error = use_case.execute(request) + + assert success is False + assert error == "New password is required"