opensource/dify
3
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-10-16 15:35:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): extract forgot password application service (#41111)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Byron.wang
2026-09-03 01:38:59 +00:00
committed by GitHub
co-authored by autofix-ci[bot]
parent 8129555642
commit be7d11ca1d
24 changed files with 1716 additions and 816 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/.importlinter
+1
View File
@@ -211,6 +211,7 @@ source_modules =
services.account_deletion_feedback_service
services.account_education_service
services.account_errors
services.account_forgot_password_service
services.account_initialization_service
services.account_integration_service
services.account_password_service
api/controllers/console/auth/forgot_password.py
+46 -101
View File
@@ -1,6 +1,3 @@
import base64
import secrets
from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field
@@ -13,25 +10,23 @@ from controllers.console.auth.error import (
InvalidEmailError,
InvalidTokenError,
PasswordMismatchError,
PasswordResetRateLimitExceededError,
)
from controllers.console.error import AccountNotFound, EmailSendIpLimitError
from controllers.console.wraps import email_password_login_enabled, model_validate, setup_required
from extensions.ext_database import db
from libs.helper import EmailStr, extract_remote_ip
from libs.password import hash_password
from services.account_service import AccountService, TenantService
from extensions.ext_application_services import application_services
from libs.helper import EmailStr, dump_response, extract_remote_ip
from services import account_errors
from services.entities.auth_entities import (
ForgotPasswordCheckPayload,
ForgotPasswordResetPayload,
ForgotPasswordSendPayload,
)
from services.system_feature_service import SystemFeatureService
class ForgotPasswordEmailResponse(BaseModel):
result: str = Field(description="Operation result")
data: str | None = Field(default=None, description="Reset token")
code: str | None = Field(default=None, description="Error code if account not found")
class ForgotPasswordCheckResponse(BaseModel):
@@ -70,27 +65,20 @@ class ForgotPasswordSendEmailApi(Resource):
@email_password_login_enabled
@model_validate(ForgotPasswordSendPayload)
def post(self, req_data: ForgotPasswordSendPayload):
normalized_email = req_data.email.lower()
ip_address = extract_remote_ip(request)
if AccountService.is_email_send_ip_limit(ip_address):
raise EmailSendIpLimitError()
language = "zh-Hans" if req_data.language == "zh-Hans" else "en-US"
try:
token = application_services().accounts.forgot_password.send_code(
email=req_data.email,
language=language,
ip_address=ip_address,
)
except account_errors.ForgotPasswordSendIPLimitedError:
raise EmailSendIpLimitError() from None
except account_errors.ForgotPasswordSendRateLimitError as error:
raise PasswordResetRateLimitExceededError(error.retry_after_minutes) from None
if req_data.language is not None and req_data.language == "zh-Hans":
language = "zh-Hans"
else:
language = "en-US"
account = AccountService.get_account_by_email_with_case_fallback(req_data.email, session=db.session())
token = AccountService.send_reset_password_email(
account=account,
email=normalized_email,
language=language,
is_allow_register=SystemFeatureService.is_registration_allowed(),
)
return {"result": "success", "data": token}
return dump_response(ForgotPasswordEmailResponse, {"result": "success", "data": token})
@console_ns.route("/forgot-password/validity")
@@ -108,40 +96,26 @@ class ForgotPasswordCheckApi(Resource):
@email_password_login_enabled
@model_validate(ForgotPasswordCheckPayload)
def post(self, req_data: ForgotPasswordCheckPayload):
try:
verification = application_services().accounts.forgot_password.verify_code(
email=req_data.email,
code=req_data.code,
token=req_data.token,
)
except account_errors.ForgotPasswordVerificationLimitError:
raise EmailPasswordResetLimitError() from None
except account_errors.InvalidForgotPasswordTokenError:
raise InvalidTokenError() from None
except account_errors.InvalidForgotPasswordEmailError:
raise InvalidEmailError() from None
except account_errors.InvalidForgotPasswordCodeError:
raise EmailCodeError() from None
user_email = req_data.email.lower()
is_forgot_password_error_rate_limit = AccountService.is_forgot_password_error_rate_limit(user_email)
if is_forgot_password_error_rate_limit:
raise EmailPasswordResetLimitError()
token_data = AccountService.get_reset_password_data(req_data.token)
if token_data is None:
raise InvalidTokenError()
token_email = token_data.get("email")
if not isinstance(token_email, str):
raise InvalidEmailError()
normalized_token_email = token_email.lower()
if user_email != normalized_token_email:
raise InvalidEmailError()
if req_data.code != token_data.get("code"):
AccountService.add_forgot_password_error_rate_limit(user_email)
raise EmailCodeError()
# Verified, revoke the first token
AccountService.revoke_reset_password_token(req_data.token)
# Refresh token data by generating a new token
_, new_token = AccountService.generate_reset_password_token(
token_email, code=req_data.code, additional_data={"phase": "reset"}
return dump_response(
ForgotPasswordCheckResponse,
{"is_valid": True, "email": verification.email, "token": verification.token},
)
AccountService.reset_forgot_password_error_rate_limit(user_email)
return {"is_valid": True, "email": normalized_token_email, "token": new_token}
@console_ns.route("/forgot-password/resets")
class ForgotPasswordResetApi(Resource):
@@ -158,46 +132,17 @@ class ForgotPasswordResetApi(Resource):
@email_password_login_enabled
@model_validate(ForgotPasswordResetPayload)
def post(self, req_data: ForgotPasswordResetPayload):
try:
application_services().accounts.forgot_password.reset(
token=req_data.token,
new_password=req_data.new_password,
password_confirm=req_data.password_confirm,
)
except account_errors.ForgotPasswordMismatchError:
raise PasswordMismatchError() from None
except account_errors.InvalidForgotPasswordTokenError:
raise InvalidTokenError() from None
except account_errors.AccountNotFoundError:
raise AccountNotFound() from None
# Validate passwords match
if req_data.new_password != req_data.password_confirm:
raise PasswordMismatchError()
# Validate token and get reset data
reset_data = AccountService.get_reset_password_data(req_data.token)
if not reset_data:
raise InvalidTokenError()
# Must use token in reset phase
if reset_data.get("phase", "") != "reset":
raise InvalidTokenError()
# Revoke token to prevent reuse
AccountService.revoke_reset_password_token(req_data.token)
# Generate secure salt and hash password
salt = secrets.token_bytes(16)
password_hashed = hash_password(req_data.new_password, salt)
email = reset_data.get("email", "")
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
if account:
account = db.session.merge(account)
self._update_existing_account(account, password_hashed, salt)
db.session.commit()
else:
raise AccountNotFound()
return {"result": "success"}
def _update_existing_account(self, account, password_hashed, salt):
# Update existing account credentials
account.password = base64.b64encode(password_hashed).decode()
account.password_salt = base64.b64encode(salt).decode()
# Create workspace if needed
if (
not TenantService.get_join_tenants(account, session=db.session())
and SystemFeatureService.is_workspace_creation_allowed()
):
TenantService.create_owner_tenant(account, session=db.session())
return dump_response(ForgotPasswordResetResponse, {"result": "success"})
api/extensions/ext_application_services.py
+30 -2
View File
@@ -85,9 +85,18 @@ from services.account_email_registration_adapters import (
TokenManagerEmailRegistrationTokenGateway,
)
from services.account_email_registration_service import AccountEmailRegistrationService
from services.account_forgot_password_adapters import (
CeleryForgotPasswordNotificationGateway,
RateLimiterForgotPasswordSendLimiter,
RedisForgotPasswordSecurityGateway,
RedisForgotPasswordTokenGateway,
SecureForgotPasswordCodeGenerator,
SystemFeatureServiceForgotPasswordRegistrationPolicy,
)
from services.account_forgot_password_service import AccountForgotPasswordService
from services.account_initialization_service import AccountInitializationService
from services.account_integration_service import AccountIntegrationService
from services.account_password_hasher import LegacyAccountPasswordHasher
from services.account_password_hasher import DefaultAccountPasswordHasher
from services.account_password_service import AccountPasswordService
from services.account_profile_service import AccountProfileService
from services.app_definition_query_service import AppDefinitionQueryService
@@ -179,6 +188,7 @@ class AccountServices:
deletion: AccountDeletionService
deletion_feedback: AccountDeletionFeedbackService
education: AccountEducationService
forgot_password: AccountForgotPasswordService
initialization: AccountInitializationService
integrations: AccountIntegrationService
password: AccountPasswordService
@@ -280,6 +290,7 @@ def build_application_services(
builtin=builtin_catalog,
)
workspace_query_repository = WorkspaceQueryRepository(session_factory=database_client)
passwords = DefaultAccountPasswordHasher()
return ApplicationServices(
accounts=AccountServices(
avatar=AccountAvatarService(
@@ -365,6 +376,23 @@ def build_application_services(
redis_client=redis,
),
),
forgot_password=AccountForgotPasswordService(
accounts=accounts,
passwords=passwords,
tokens=RedisForgotPasswordTokenGateway(
redis=redis,
expiry_seconds=int(dify_config.RESET_PASSWORD_TOKEN_EXPIRY_MINUTES * 60),
),
codes=SecureForgotPasswordCodeGenerator(),
notifications=CeleryForgotPasswordNotificationGateway(),
send_limits=RateLimiterForgotPasswordSendLimiter(redis=redis),
security=RedisForgotPasswordSecurityGateway(
redis=redis,
email_send_ip_limit_per_minute=dify_config.EMAIL_SEND_IP_LIMIT_PER_MINUTE,
verification_lockout_duration=dify_config.FORGOT_PASSWORD_LOCKOUT_DURATION,
),
registration=SystemFeatureServiceForgotPasswordRegistrationPolicy(),
),
initialization=AccountInitializationService(
accounts=accounts,
invitation_required=deployment_edition == DeploymentEdition.CLOUD,
@@ -373,7 +401,7 @@ def build_application_services(
integrations=AccountIntegrationService(integrations=integrations),
password=AccountPasswordService(
accounts=accounts,
passwords=LegacyAccountPasswordHasher(),
passwords=passwords,
),
profile=AccountProfileService(accounts=accounts),
),
api/openapi/markdown/console-openapi.md
-1
View File
@@ -18395,7 +18395,6 @@ Enum class for fetch from.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| code | string | Error code if account not found | No |
| data | string | Reset token | No |
| result | string | Operation result | Yes |
api/repositories/account_repository.py
+2 -2
View File
@@ -34,9 +34,9 @@ class SQLAlchemyAccountRepository(AccountRepository):
@override
def find_by_email(self, email: str) -> AccountSnapshot | None:
with self._session_factory() as session:
account = session.scalar(select(Account).where(Account.email == email).limit(1))
account = session.execute(select(Account).where(Account.email == email)).scalar_one_or_none()
if account is None and email != email.lower():
account = session.scalar(select(Account).where(Account.email == email.lower()).limit(1))
account = session.execute(select(Account).where(Account.email == email.lower())).scalar_one_or_none()
return self._to_snapshot(account) if account is not None else None
@override
api/services/account_change_email_adapters.py
+9 -59
View File
@@ -4,7 +4,6 @@ import secrets
from typing import override
from pydantic import TypeAdapter, ValidationError
from redis import RedisError
from extensions.ext_redis import RedisClientWrapper
from libs.helper import RateLimiter, TokenManager
@@ -16,6 +15,7 @@ from services.account_change_email_ports import (
ChangeEmailSendLimiter,
ChangeEmailTokenGateway,
)
from services.account_security_gateway import RedisAccountEmailSecurityGateway
from services.billing_service import BillingService
from services.entities.account_entities import (
AccountChangeEmailNewEmailToken,
@@ -115,7 +115,7 @@ class RateLimiterChangeEmailSendLimiter(ChangeEmailSendLimiter):
return int(self._rate_limiter.time_window / 60)
class RedisChangeEmailSecurityGateway(ChangeEmailSecurityGateway):
class RedisChangeEmailSecurityGateway(RedisAccountEmailSecurityGateway, ChangeEmailSecurityGateway):
def __init__(
self,
*,
@@ -124,63 +124,13 @@ class RedisChangeEmailSecurityGateway(ChangeEmailSecurityGateway):
verification_failure_limit: int,
verification_lockout_duration: int,
) -> None:
self._redis = redis
self._email_send_ip_limit_per_minute = email_send_ip_limit_per_minute
self._verification_failure_limit = verification_failure_limit
self._verification_lockout_duration = verification_lockout_duration
@override
def is_ip_limited(self, ip_address: str) -> bool:
minute_key = f"email_send_ip_limit_minute:{ip_address}"
freeze_key = f"email_send_ip_limit_freeze:{ip_address}"
hour_limit_key = f"email_send_ip_limit_hour:{ip_address}"
try:
if self._redis.get(freeze_key):
return True
current_minute_count = int(self._redis.get(minute_key) or 0)
if current_minute_count > self._email_send_ip_limit_per_minute:
hour_limit_count = int(self._redis.get(hour_limit_key) or 0)
if hour_limit_count >= 1:
self._redis.setex(freeze_key, 60 * 60, 1)
return True
if not self._redis.set(hour_limit_key, 1, ex=60 * 10, nx=True):
self._redis.setex(freeze_key, 60 * 60, 1)
return True
self._redis.setex(minute_key, 60, current_minute_count + 1)
self._redis.expire(minute_key, 60)
return False
except RedisError:
return False
@override
def is_verification_limited(self, email: str) -> bool:
try:
count = self._redis.get(self._verification_key(email))
return count is not None and int(count) > self._verification_failure_limit
except RedisError:
return False
@override
def record_verification_failure(self, email: str) -> None:
try:
key = self._verification_key(email)
count = int(self._redis.get(key) or 0) + 1
self._redis.setex(key, self._verification_lockout_duration, count)
except RedisError:
return None
@override
def reset_verification_failures(self, email: str) -> None:
try:
self._redis.delete(self._verification_key(email))
except RedisError:
return None
@staticmethod
def _verification_key(email: str) -> str:
return f"change_email_error_rate_limit:{email}"
super().__init__(
redis=redis,
email_send_ip_limit_per_minute=email_send_ip_limit_per_minute,
verification_failure_limit=verification_failure_limit,
verification_lockout_duration=verification_lockout_duration,
verification_key_prefix="change_email_error_rate_limit",
)
class BillingAccountEmailPolicyGateway(AccountEmailPolicyGateway):
api/services/account_errors.py
+32
View File
@@ -17,6 +17,38 @@ class InvalidAccountPasswordError(AccountApplicationError):
"""The requested password does not satisfy the account password policy."""
class ForgotPasswordSendIPLimitedError(AccountApplicationError):
"""The caller IP exceeded the email-send policy."""
class ForgotPasswordSendRateLimitError(AccountApplicationError):
"""Too many password-reset messages were requested for the address."""
def __init__(self, retry_after_minutes: int) -> None:
super().__init__(retry_after_minutes)
self.retry_after_minutes = retry_after_minutes
class ForgotPasswordVerificationLimitError(AccountApplicationError):
"""Too many invalid password-reset codes were submitted."""
class InvalidForgotPasswordTokenError(AccountApplicationError):
"""The password-reset token is absent, malformed, or in the wrong phase."""
class InvalidForgotPasswordEmailError(AccountApplicationError):
"""The request address does not match the password-reset token."""
class InvalidForgotPasswordCodeError(AccountApplicationError):
"""The verification code does not match the password-reset token."""
class ForgotPasswordMismatchError(AccountApplicationError):
"""The password and its confirmation do not match."""
class AvatarFileNotFoundError(AccountApplicationError):
"""The requested avatar file does not exist or is not owned by the account."""
api/services/account_forgot_password_adapters.py
+268
View File
@@ -0,0 +1,268 @@
"""Infrastructure adapters for the account forgot-password application service."""
import secrets
import uuid
from typing import override
from pydantic import TypeAdapter, ValidationError
from extensions.ext_redis import RedisClientWrapper
from extensions.redis_names import serialize_redis_name
from libs.helper import RateLimiter
from services.account_forgot_password_service import (
FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS,
FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX,
FORGOT_PASSWORD_SEND_RATE_LIMIT_WINDOW_SECONDS,
FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT,
FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX,
ForgotPasswordCodeGenerator,
ForgotPasswordNotificationGateway,
ForgotPasswordRegistrationPolicy,
ForgotPasswordSecurityGateway,
ForgotPasswordSendLimiter,
ForgotPasswordTokenGateway,
)
from services.account_security_gateway import RedisAccountEmailSecurityGateway
from services.entities.account_entities import (
ForgotPasswordResetToken,
ForgotPasswordToken,
ForgotPasswordVerificationToken,
)
from services.entities.auth_entities import (
ForgotPasswordResetTokenData,
ForgotPasswordTokenData,
ForgotPasswordVerificationTokenData,
)
from services.system_feature_service import SystemFeatureService
from tasks.mail_reset_password_task import (
send_reset_password_mail_task,
send_reset_password_mail_task_when_account_not_exist,
)
_RESET_PASSWORD_TOKEN_TYPE = "reset_password"
_token_data_adapter: TypeAdapter[ForgotPasswordTokenData] = TypeAdapter(ForgotPasswordTokenData)
_COMPARE_AND_SET_CURRENT_TOKEN_LUA = """
local current = redis.call('GET', KEYS[1])
if current ~= ARGV[1] then
return 0
end
redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
return 1
"""
class RedisForgotPasswordTokenGateway(ForgotPasswordTokenGateway):
def __init__(self, *, redis: RedisClientWrapper, expiry_seconds: int) -> None:
self._redis = redis
self._expiry_seconds = expiry_seconds
@override
def read_verification(self, token: str) -> ForgotPasswordVerificationToken | None:
key = f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{token}"
raw_payload = self._redis.get(key)
if raw_payload is None:
return None
token_data = self._deserialize(raw_payload) if isinstance(raw_payload, str | bytes | bytearray) else None
if not isinstance(token_data, ForgotPasswordVerificationToken):
return None
if token_data.account_id is not None:
if not self._is_current(token_data.account_id, token):
return None
return token_data
@override
def claim_reset(self, token: str) -> ForgotPasswordResetToken | None:
key = f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{token}"
raw_payload = self._redis.getdel(key)
if raw_payload is None:
return None
token_data = self._deserialize(raw_payload) if isinstance(raw_payload, str | bytes | bytearray) else None
if not isinstance(token_data, ForgotPasswordResetToken):
return None
if token_data.account_id is not None and not self._is_current(token_data.account_id, token):
return None
return token_data
@staticmethod
def _deserialize(raw_payload: str | bytes | bytearray) -> ForgotPasswordToken | None:
try:
payload = _token_data_adapter.validate_json(raw_payload)
except ValidationError:
return None
if isinstance(payload, ForgotPasswordResetTokenData):
return ForgotPasswordResetToken(
email=str(payload.email),
code=payload.code,
account_id=payload.account_id,
)
return ForgotPasswordVerificationToken(
email=str(payload.email),
code=payload.code,
account_id=payload.account_id,
)
@override
def issue(self, token_data: ForgotPasswordToken) -> str:
token = self._create_token(token_data)
account_id = token_data.account_id
if account_id is None:
return token
account_key = f"{_RESET_PASSWORD_TOKEN_TYPE}:account:{account_id}"
previous_token = self._redis.get(account_key)
self._redis.setex(account_key, self._expiry_seconds, token)
if isinstance(previous_token, bytes):
previous_token = previous_token.decode()
if isinstance(previous_token, str) and previous_token != token:
self._redis.delete(f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{previous_token}")
return token
@override
def promote(self, claimed_token: str, token_data: ForgotPasswordResetToken) -> str | None:
account_id = token_data.account_id
if account_id is None:
key = f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{claimed_token}"
raw_payload = self._redis.getdel(key)
claimed = self._deserialize(raw_payload) if isinstance(raw_payload, str | bytes | bytearray) else None
if not isinstance(claimed, ForgotPasswordVerificationToken) or claimed.promote() != token_data:
return None
return self.issue(token_data)
reset_token = self._create_token(token_data)
promoted = self._compare_and_set_current(
account_id=account_id,
expected=claimed_token,
replacement=reset_token,
expiry_seconds=self._expiry_seconds,
)
if promoted:
self._redis.delete(f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{claimed_token}")
return reset_token
self._redis.delete(f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{reset_token}")
return None
def _create_token(self, token_data: ForgotPasswordToken) -> str:
payload: ForgotPasswordTokenData
if isinstance(token_data, ForgotPasswordResetToken):
payload = ForgotPasswordResetTokenData(
account_id=token_data.account_id,
email=token_data.email,
code=token_data.code,
phase="reset",
)
else:
payload = ForgotPasswordVerificationTokenData(
account_id=token_data.account_id,
email=token_data.email,
code=token_data.code,
)
token = str(uuid.uuid4())
self._redis.setex(
f"{_RESET_PASSWORD_TOKEN_TYPE}:token:{token}",
self._expiry_seconds,
payload.model_dump_json(),
)
return token
def _compare_and_set_current(
self,
*,
account_id: str,
expected: str,
replacement: str,
expiry_seconds: int,
) -> bool:
account_key = f"{_RESET_PASSWORD_TOKEN_TYPE}:account:{account_id}"
return bool(
self._redis.eval(
_COMPARE_AND_SET_CURRENT_TOKEN_LUA,
1,
serialize_redis_name(account_key),
expected,
replacement,
expiry_seconds,
)
)
def _is_current(self, account_id: str, token: str) -> bool:
current = self._redis.get(f"{_RESET_PASSWORD_TOKEN_TYPE}:account:{account_id}")
if isinstance(current, bytes):
current = current.decode()
return current == token
class SecureForgotPasswordCodeGenerator(ForgotPasswordCodeGenerator):
@override
def generate(self) -> str:
return "".join(str(secrets.randbelow(exclusive_upper_bound=10)) for _ in range(6))
class CeleryForgotPasswordNotificationGateway(ForgotPasswordNotificationGateway):
@override
def send(
self,
*,
email: str,
code: str,
language: str,
account_exists: bool,
registration_allowed: bool,
) -> None:
if account_exists:
send_reset_password_mail_task.delay(language=language, to=email, code=code)
return
send_reset_password_mail_task_when_account_not_exist.delay(
language=language,
to=email,
is_allow_register=registration_allowed,
)
class RateLimiterForgotPasswordSendLimiter(ForgotPasswordSendLimiter):
def __init__(self, *, redis: RedisClientWrapper) -> None:
self._rate_limiter = RateLimiter(
prefix=FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX,
max_attempts=FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS,
time_window=FORGOT_PASSWORD_SEND_RATE_LIMIT_WINDOW_SECONDS,
redis_client=redis,
)
@override
def is_limited(self, email: str) -> bool:
return self._rate_limiter.is_rate_limited(email)
@override
def record(self, email: str) -> None:
self._rate_limiter.increment_rate_limit(email)
@property
@override
def retry_after_minutes(self) -> int:
return int(self._rate_limiter.time_window / 60)
class RedisForgotPasswordSecurityGateway(RedisAccountEmailSecurityGateway, ForgotPasswordSecurityGateway):
def __init__(
self,
*,
redis: RedisClientWrapper,
email_send_ip_limit_per_minute: int,
verification_lockout_duration: int,
) -> None:
super().__init__(
redis=redis,
email_send_ip_limit_per_minute=email_send_ip_limit_per_minute,
verification_failure_limit=FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT,
verification_lockout_duration=verification_lockout_duration,
verification_key_prefix=FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX,
)
class SystemFeatureServiceForgotPasswordRegistrationPolicy(ForgotPasswordRegistrationPolicy):
@override
def is_registration_allowed(self) -> bool:
return SystemFeatureService.is_registration_allowed()
api/services/account_forgot_password_service.py
+186
View File
@@ -0,0 +1,186 @@
"""Application service implementing the account forgot-password use case."""
from typing import Protocol
from services.account_errors import (
AccountNotFoundError,
ForgotPasswordMismatchError,
ForgotPasswordSendIPLimitedError,
ForgotPasswordSendRateLimitError,
ForgotPasswordVerificationLimitError,
InvalidForgotPasswordCodeError,
InvalidForgotPasswordEmailError,
InvalidForgotPasswordTokenError,
)
from services.account_ports import AccountPasswordHasher, AccountRepository
from services.entities.account_entities import (
ForgotPasswordResetToken,
ForgotPasswordToken,
ForgotPasswordVerification,
ForgotPasswordVerificationToken,
)
FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX = "reset_password_rate_limit"
FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS = 1
FORGOT_PASSWORD_SEND_RATE_LIMIT_WINDOW_SECONDS = 60
FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT = 5
FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX = "forgot_password_error_rate_limit"
class ForgotPasswordTokenGateway(Protocol):
def read_verification(self, token: str) -> ForgotPasswordVerificationToken | None: ...
def claim_reset(self, token: str) -> ForgotPasswordResetToken | None: ...
def issue(self, token_data: ForgotPasswordToken) -> str: ...
def promote(self, claimed_token: str, token_data: ForgotPasswordResetToken) -> str | None: ...
class ForgotPasswordCodeGenerator(Protocol):
def generate(self) -> str: ...
class ForgotPasswordNotificationGateway(Protocol):
def send(
self,
*,
email: str,
code: str,
language: str,
account_exists: bool,
registration_allowed: bool,
) -> None: ...
class ForgotPasswordSendLimiter(Protocol):
def is_limited(self, email: str) -> bool: ...
def record(self, email: str) -> None: ...
@property
def retry_after_minutes(self) -> int: ...
class ForgotPasswordSecurityGateway(Protocol):
def is_ip_limited(self, ip_address: str) -> bool: ...
def is_verification_limited(self, email: str) -> bool: ...
def record_verification_failure(self, email: str) -> None: ...
def reset_verification_failures(self, email: str) -> None: ...
class ForgotPasswordRegistrationPolicy(Protocol):
def is_registration_allowed(self) -> bool: ...
class AccountForgotPasswordService:
def __init__(
self,
*,
accounts: AccountRepository,
passwords: AccountPasswordHasher,
tokens: ForgotPasswordTokenGateway,
codes: ForgotPasswordCodeGenerator,
notifications: ForgotPasswordNotificationGateway,
send_limits: ForgotPasswordSendLimiter,
security: ForgotPasswordSecurityGateway,
registration: ForgotPasswordRegistrationPolicy,
) -> None:
self._accounts = accounts
self._passwords = passwords
self._tokens = tokens
self._codes = codes
self._notifications = notifications
self._send_limits = send_limits
self._security = security
self._registration = registration
def send_code(
self,
*,
email: str,
language: str,
ip_address: str,
) -> str:
if self._security.is_ip_limited(ip_address):
raise ForgotPasswordSendIPLimitedError
normalized_email = email.lower()
account = self._accounts.find_by_email(email)
destination = account.email if account is not None else normalized_email
registration_allowed = self._registration.is_registration_allowed()
if self._send_limits.is_limited(destination):
raise ForgotPasswordSendRateLimitError(self._send_limits.retry_after_minutes)
code = self._codes.generate()
token = self._tokens.issue(
ForgotPasswordVerificationToken(
email=destination,
code=code,
account_id=account.id if account is not None else None,
)
)
self._notifications.send(
email=destination,
code=code,
language=language,
account_exists=account is not None,
registration_allowed=registration_allowed,
)
self._send_limits.record(destination)
return token
def verify_code(
self,
*,
email: str,
code: str,
token: str,
) -> ForgotPasswordVerification:
normalized_email = email.lower()
if self._security.is_verification_limited(normalized_email):
raise ForgotPasswordVerificationLimitError
token_data = self._tokens.read_verification(token)
if token_data is None:
raise InvalidForgotPasswordTokenError
normalized_token_email = token_data.email.lower()
if normalized_email != normalized_token_email:
raise InvalidForgotPasswordEmailError
if code != token_data.code:
self._security.record_verification_failure(normalized_email)
raise InvalidForgotPasswordCodeError
reset_token = self._tokens.promote(token, token_data.promote())
if reset_token is None:
raise InvalidForgotPasswordTokenError
self._security.reset_verification_failures(normalized_email)
return ForgotPasswordVerification(email=normalized_token_email, token=reset_token)
def reset(
self,
*,
token: str,
new_password: str,
password_confirm: str,
) -> None:
if new_password != password_confirm:
raise ForgotPasswordMismatchError
token_data = self._tokens.claim_reset(token)
if token_data is None:
raise InvalidForgotPasswordTokenError
if token_data.account_id is None:
account = self._accounts.find_by_email(token_data.email)
else:
account = self._accounts.get(token_data.account_id)
if account is None or account.email.lower() != token_data.email.lower():
raise AccountNotFoundError
password = self._passwords.hash(new_password)
if self._accounts.update_password(account.id, password) is None:
raise AccountNotFoundError
api/services/account_password_hasher.py
+2 -2
View File
@@ -1,4 +1,4 @@
"""Legacy password implementation behind the account password port."""
"""Default password implementation behind the account password port."""
import base64
import secrets
@@ -10,7 +10,7 @@ from services.account_ports import AccountPasswordHasher
from services.entities.account_entities import AccountPasswordDigest
class LegacyAccountPasswordHasher(AccountPasswordHasher):
class DefaultAccountPasswordHasher(AccountPasswordHasher):
@override
def verify(self, password: str, *, password_hash: str, password_salt: str) -> bool:
return compare_password(password, password_hash, password_salt)
api/services/account_security_gateway.py
+78
View File
@@ -0,0 +1,78 @@
"""Shared Redis-backed security policies for account email flows."""
import logging
from redis import RedisError
from extensions.ext_redis import RedisClientWrapper
logger = logging.getLogger(__name__)
class RedisAccountEmailSecurityGateway:
def __init__(
self,
*,
redis: RedisClientWrapper,
email_send_ip_limit_per_minute: int,
verification_failure_limit: int,
verification_lockout_duration: int,
verification_key_prefix: str,
) -> None:
self._redis = redis
self._email_send_ip_limit_per_minute = email_send_ip_limit_per_minute
self._verification_failure_limit = verification_failure_limit
self._verification_lockout_duration = verification_lockout_duration
self._verification_key_prefix = verification_key_prefix
def is_ip_limited(self, ip_address: str) -> bool:
minute_key = f"email_send_ip_limit_minute:{ip_address}"
freeze_key = f"email_send_ip_limit_freeze:{ip_address}"
hour_limit_key = f"email_send_ip_limit_hour:{ip_address}"
try:
if self._redis.get(freeze_key):
return True
current_minute_count = int(self._redis.get(minute_key) or 0)
if current_minute_count > self._email_send_ip_limit_per_minute:
hour_limit_count = int(self._redis.get(hour_limit_key) or 0)
if hour_limit_count >= 1:
self._redis.setex(freeze_key, 60 * 60, 1)
return True
if not self._redis.set(hour_limit_key, 1, ex=60 * 10, nx=True):
self._redis.setex(freeze_key, 60 * 60, 1)
return True
self._redis.setex(minute_key, 60, current_minute_count + 1)
self._redis.expire(minute_key, 60)
return False
except RedisError:
logger.warning("Redis unavailable while checking the account email IP limit", exc_info=True)
return False
def is_verification_limited(self, email: str) -> bool:
try:
count = self._redis.get(self._verification_key(email))
return count is not None and int(count) > self._verification_failure_limit
except RedisError:
logger.warning("Redis unavailable while checking the account email verification limit", exc_info=True)
return False
def record_verification_failure(self, email: str) -> None:
try:
key = self._verification_key(email)
count = int(self._redis.get(key) or 0) + 1
self._redis.setex(key, self._verification_lockout_duration, count)
except RedisError:
logger.warning("Redis unavailable while recording an account email verification failure", exc_info=True)
return None
def reset_verification_failures(self, email: str) -> None:
try:
self._redis.delete(self._verification_key(email))
except RedisError:
logger.warning("Redis unavailable while resetting account email verification failures", exc_info=True)
return None
def _verification_key(self, email: str) -> str:
return f"{self._verification_key_prefix}:{email}"
api/services/account_service.py
+16 -6
View File
@@ -48,6 +48,13 @@ from models.account import (
from models.dataset import Dataset
from models.model import App, DifySetup
from services.account_email import normalize_email
from services.account_forgot_password_service import (
FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS,
FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX,
FORGOT_PASSWORD_SEND_RATE_LIMIT_WINDOW_SECONDS,
FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT,
FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX,
)
from services.billing_service import BillingService
from services.email_code_login_challenge import (
EmailCodeLoginChallengeResult,
@@ -155,7 +162,11 @@ class AccountService:
CHANGE_EMAIL_PHASE_OLD = ChangeEmailPhase.OLD_EMAIL
CHANGE_EMAIL_PHASE_NEW = ChangeEmailPhase.NEW_EMAIL
reset_password_rate_limiter = RateLimiter(prefix="reset_password_rate_limit", max_attempts=1, time_window=60 * 1)
reset_password_rate_limiter = RateLimiter(
prefix=FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX,
max_attempts=FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS,
time_window=FORGOT_PASSWORD_SEND_RATE_LIMIT_WINDOW_SECONDS,
)
email_code_login_rate_limiter = RateLimiter(
prefix="email_code_login_rate_limit", max_attempts=3, time_window=300 * 1
)
@@ -163,7 +174,6 @@ class AccountService:
owner_transfer_rate_limiter = RateLimiter(prefix="owner_transfer_rate_limit", max_attempts=1, time_window=60 * 1)
LOGIN_MAX_ERROR_LIMITS = 5
FORGOT_PASSWORD_MAX_ERROR_LIMITS = 5
CHANGE_EMAIL_MAX_ERROR_LIMITS = 5
OWNER_TRANSFER_MAX_ERROR_LIMITS = 5
@@ -1001,7 +1011,7 @@ class AccountService:
@staticmethod
@redis_fallback(default_return=None)
def add_forgot_password_error_rate_limit(email: str):
key = f"forgot_password_error_rate_limit:{email}"
key = f"{FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX}:{email}"
count = redis_client.get(key)
if count is None:
count = 0
@@ -1011,20 +1021,20 @@ class AccountService:
@staticmethod
@redis_fallback(default_return=False)
def is_forgot_password_error_rate_limit(email: str) -> bool:
key = f"forgot_password_error_rate_limit:{email}"
key = f"{FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX}:{email}"
count = redis_client.get(key)
if count is None:
return False
count = int(count)
if count > AccountService.FORGOT_PASSWORD_MAX_ERROR_LIMITS:
if count > FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT:
return True
return False
@staticmethod
@redis_fallback(default_return=None)
def reset_forgot_password_error_rate_limit(email: str):
key = f"forgot_password_error_rate_limit:{email}"
key = f"{FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX}:{email}"
redis_client.delete(key)
@staticmethod
api/services/entities/account_entities.py
+26
View File
@@ -57,6 +57,32 @@ class AccountPasswordDigest:
password_salt: str
@dataclass(frozen=True, slots=True)
class ForgotPasswordVerificationToken:
email: str
code: str
account_id: str | None = None
def promote(self) -> ForgotPasswordResetToken:
return ForgotPasswordResetToken(email=self.email, code=self.code, account_id=self.account_id)
@dataclass(frozen=True, slots=True)
class ForgotPasswordResetToken:
email: str
code: str
account_id: str | None = None
type ForgotPasswordToken = ForgotPasswordVerificationToken | ForgotPasswordResetToken
@dataclass(frozen=True, slots=True)
class ForgotPasswordVerification:
email: str
token: str
@dataclass(frozen=True, slots=True)
class AccountIntegrationSnapshot:
provider: str
api/services/entities/auth_entities.py
+20
View File
@@ -66,6 +66,26 @@ class ForgotPasswordResetPayload(BaseModel):
return valid_password(value)
class ForgotPasswordTokenBase(BaseModel):
token_type: Literal["reset_password"] = "reset_password"
account_id: str | None = None
email: EmailStr
code: str = Field(min_length=1)
model_config = ConfigDict(extra="forbid")
class ForgotPasswordVerificationTokenData(ForgotPasswordTokenBase):
pass
class ForgotPasswordResetTokenData(ForgotPasswordTokenBase):
phase: Literal["reset"]
type ForgotPasswordTokenData = ForgotPasswordResetTokenData | ForgotPasswordVerificationTokenData
class ChangeEmailTokenBase(BaseModel):
"""Stored change-email token payload.
api/tests/unit_tests/controllers/console/auth/test_forgot_password.py
+210 -159
View File
@@ -1,184 +1,235 @@
"""Unit tests for forgot password controller endpoints."""
"""Transport-boundary tests for Console forgot-password endpoints."""
from __future__ import annotations
from collections.abc import Iterator
from unittest.mock import patch
from types import SimpleNamespace
import pytest
from flask import Flask
from controllers.console.auth.forgot_password import (
ForgotPasswordCheckApi,
ForgotPasswordResetApi,
ForgotPasswordSendEmailApi,
from controllers.console.auth import forgot_password
from controllers.console.auth.error import (
EmailCodeError,
EmailPasswordResetLimitError,
InvalidEmailError,
InvalidTokenError,
PasswordMismatchError,
PasswordResetRateLimitExceededError,
)
from controllers.console.error import AccountNotFound, EmailSendIpLimitError
from enums import DeploymentEdition
from models.account import Account
from models.engine import db
from services.entities.feature_entities import SystemFeatureModel
from tests.unit_tests.config_override import config_overrides_context
from services import account_errors
from services.entities.account_entities import ForgotPasswordVerification
from tests.unit_tests.config_override import apply_config_overrides
class FakeForgotPasswordService:
def __init__(self) -> None:
self.error: Exception | None = None
self.send_arguments: dict[str, str] | None = None
self.verify_arguments: dict[str, str] | None = None
self.reset_arguments: dict[str, str] | None = None
def _raise_if_needed(self) -> None:
if self.error is not None:
raise self.error
def send_code(
self,
*,
email: str,
language: str,
ip_address: str,
) -> str:
self.send_arguments = {"email": email, "language": language, "ip_address": ip_address}
self._raise_if_needed()
return "reset-token"
def verify_code(
self,
*,
email: str,
code: str,
token: str,
) -> ForgotPasswordVerification:
self.verify_arguments = {"email": email, "code": code, "token": token}
self._raise_if_needed()
return ForgotPasswordVerification(email="user@example.com", token="promoted-token")
def reset(
self,
*,
token: str,
new_password: str,
password_confirm: str,
) -> None:
self.reset_arguments = {
"token": token,
"new_password": new_password,
"password_confirm": password_confirm,
}
self._raise_if_needed()
@pytest.fixture(autouse=True)
def admit_forgot_password_requests(monkeypatch: pytest.MonkeyPatch) -> None:
apply_config_overrides(
monkeypatch,
DEPLOYMENT_EDITION=DeploymentEdition.CLOUD,
ENABLE_EMAIL_PASSWORD_LOGIN=True,
)
@pytest.fixture
def database_app() -> Iterator[Flask]:
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db.init_app(app)
with app.app_context():
Account.__table__.create(db.engine)
yield app
def service(monkeypatch: pytest.MonkeyPatch) -> FakeForgotPasswordService:
service = FakeForgotPasswordService()
services = SimpleNamespace(accounts=SimpleNamespace(forgot_password=service))
monkeypatch.setattr(forgot_password, "application_services", lambda: services)
monkeypatch.setattr(forgot_password, "extract_remote_ip", lambda _request: "127.0.0.1")
return service
class TestForgotPasswordSendEmailApi:
@patch("controllers.console.auth.forgot_password.AccountService.get_account_by_email_with_case_fallback")
@patch("controllers.console.auth.forgot_password.AccountService.send_reset_password_email")
@patch("controllers.console.auth.forgot_password.AccountService.is_email_send_ip_limit", return_value=False)
@patch("controllers.console.auth.forgot_password.extract_remote_ip", return_value="127.0.0.1")
def test_send_normalizes_email(
self,
mock_extract_ip,
mock_is_ip_limit,
mock_send_email,
mock_get_account,
app: Flask,
def test_send_parses_input_and_serializes_service_result(app: Flask, service: FakeForgotPasswordService) -> None:
with app.test_request_context(
"/forgot-password",
method="POST",
json={"email": "User@Example.com", "language": "zh-Hans"},
):
account = Account(name="User", email="user@example.com")
mock_get_account.return_value = account
mock_send_email.return_value = "token-123"
response = forgot_password.ForgotPasswordSendEmailApi().post()
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
controller_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
is_allow_register=True,
)
with (
patch(
"controllers.console.auth.forgot_password.SystemFeatureService.is_registration_allowed",
return_value=controller_features.is_allow_register,
),
config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD),
patch(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
return_value=wraps_features.enable_email_password_login,
),
):
with app.test_request_context(
"/forgot-password",
method="POST",
json={"email": "User@Example.com", "language": "zh-Hans"},
):
response = ForgotPasswordSendEmailApi().post()
assert response == {"result": "success", "data": "token-123"}
mock_send_email.assert_called_once_with(
account=account,
email="user@example.com",
language="zh-Hans",
is_allow_register=True,
)
mock_is_ip_limit.assert_called_once_with("127.0.0.1")
mock_extract_ip.assert_called_once()
assert response == {"result": "success", "data": "reset-token"}
assert service.send_arguments == {
"email": "User@Example.com",
"language": "zh-Hans",
"ip_address": "127.0.0.1",
}
class TestForgotPasswordCheckApi:
@patch("controllers.console.auth.forgot_password.AccountService.reset_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.generate_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.add_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
def test_check_normalizes_email(
self,
mock_rate_limit_check,
mock_get_data,
mock_add_rate,
mock_revoke_token,
mock_generate_token,
mock_reset_rate,
app: Flask,
def test_send_defaults_unsupported_language(app: Flask, service: FakeForgotPasswordService) -> None:
with app.test_request_context(
"/forgot-password",
method="POST",
json={"email": "user@example.com", "language": "fr-FR"},
):
mock_rate_limit_check.return_value = False
mock_get_data.return_value = {"email": "Admin@Example.com", "code": "4321"}
mock_generate_token.return_value = (None, "new-token")
forgot_password.ForgotPasswordSendEmailApi().post()
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
)
with (
config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD),
patch(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
return_value=wraps_features.enable_email_password_login,
),
):
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "ADMIN@Example.com", "code": "4321", "token": "token-123"},
):
response = ForgotPasswordCheckApi().post()
assert response == {"is_valid": True, "email": "admin@example.com", "token": "new-token"}
mock_rate_limit_check.assert_called_once_with("admin@example.com")
mock_generate_token.assert_called_once_with(
"Admin@Example.com",
code="4321",
additional_data={"phase": "reset"},
)
mock_reset_rate.assert_called_once_with("admin@example.com")
mock_add_rate.assert_not_called()
mock_revoke_token.assert_called_once_with("token-123")
assert service.send_arguments is not None
assert service.send_arguments["language"] == "en-US"
class TestForgotPasswordResetApi:
@patch("controllers.console.auth.forgot_password.ForgotPasswordResetApi._update_existing_account")
@patch("controllers.console.auth.forgot_password.AccountService.get_account_by_email_with_case_fallback")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
def test_reset_fetches_account_with_original_email(
self,
mock_get_reset_data,
mock_revoke_token,
mock_get_account,
mock_update_account,
database_app: Flask,
@pytest.mark.parametrize(
("service_error", "request_error"),
[
(account_errors.ForgotPasswordSendIPLimitedError(), EmailSendIpLimitError),
(account_errors.ForgotPasswordSendRateLimitError(2), PasswordResetRateLimitExceededError),
],
)
def test_send_maps_application_errors(
app: Flask,
service: FakeForgotPasswordService,
service_error: Exception,
request_error: type[Exception],
) -> None:
service.error = service_error
with (
app.test_request_context(
"/forgot-password",
method="POST",
json={"email": "user@example.com"},
),
pytest.raises(request_error),
):
mock_get_reset_data.return_value = {"phase": "reset", "email": "User@Example.com"}
account = Account(name="User", email="user@example.com")
db.session.add(account)
db.session.commit()
mock_get_account.return_value = account
forgot_password.ForgotPasswordSendEmailApi().post()
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
)
with (
config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD),
patch(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
return_value=wraps_features.enable_email_password_login,
),
):
with database_app.test_request_context(
"/forgot-password/resets",
method="POST",
json={
"token": "token-123",
"new_password": "ValidPass123!",
"password_confirm": "ValidPass123!",
},
):
response = ForgotPasswordResetApi().post()
assert response == {"result": "success"}
mock_get_reset_data.assert_called_once_with("token-123")
mock_revoke_token.assert_called_once_with("token-123")
mock_update_account.assert_called_once()
def test_verify_serializes_promoted_token(app: Flask, service: FakeForgotPasswordService) -> None:
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "USER@example.com", "code": "123456", "token": "verification-token"},
):
response = forgot_password.ForgotPasswordCheckApi().post()
assert response == {"is_valid": True, "email": "user@example.com", "token": "promoted-token"}
assert service.verify_arguments == {
"email": "USER@example.com",
"code": "123456",
"token": "verification-token",
}
@pytest.mark.parametrize(
("service_error", "request_error"),
[
(account_errors.ForgotPasswordVerificationLimitError(), EmailPasswordResetLimitError),
(account_errors.InvalidForgotPasswordTokenError(), InvalidTokenError),
(account_errors.InvalidForgotPasswordEmailError(), InvalidEmailError),
(account_errors.InvalidForgotPasswordCodeError(), EmailCodeError),
],
)
def test_verify_maps_application_errors(
app: Flask,
service: FakeForgotPasswordService,
service_error: Exception,
request_error: type[Exception],
) -> None:
service.error = service_error
with (
app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "user@example.com", "code": "123456", "token": "token"},
),
pytest.raises(request_error),
):
forgot_password.ForgotPasswordCheckApi().post()
def test_reset_delegates_and_serializes_result(app: Flask, service: FakeForgotPasswordService) -> None:
with app.test_request_context(
"/forgot-password/resets",
method="POST",
json={
"token": "reset-token",
"new_password": "ValidPass123!",
"password_confirm": "ValidPass123!",
},
):
response = forgot_password.ForgotPasswordResetApi().post()
assert response == {"result": "success"}
assert service.reset_arguments == {
"token": "reset-token",
"new_password": "ValidPass123!",
"password_confirm": "ValidPass123!",
}
@pytest.mark.parametrize(
("service_error", "request_error"),
[
(account_errors.ForgotPasswordMismatchError(), PasswordMismatchError),
(account_errors.InvalidForgotPasswordTokenError(), InvalidTokenError),
(account_errors.AccountNotFoundError(), AccountNotFound),
],
)
def test_reset_maps_application_errors(
app: Flask,
service: FakeForgotPasswordService,
service_error: Exception,
request_error: type[Exception],
) -> None:
service.error = service_error
with (
app.test_request_context(
"/forgot-password/resets",
method="POST",
json={
"token": "reset-token",
"new_password": "ValidPass123!",
"password_confirm": "ValidPass123!",
},
),
pytest.raises(request_error),
):
forgot_password.ForgotPasswordResetApi().post()
api/tests/unit_tests/controllers/console/auth/test_password_reset.py
-478
View File
@@ -1,478 +0,0 @@
"""Unit tests for password reset controller flows."""
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from controllers.console.auth.error import (
EmailCodeError,
EmailPasswordResetLimitError,
InvalidEmailError,
InvalidTokenError,
PasswordMismatchError,
)
from controllers.console.auth.forgot_password import (
ForgotPasswordCheckApi,
ForgotPasswordResetApi,
ForgotPasswordSendEmailApi,
)
from controllers.console.error import AccountNotFound, EmailSendIpLimitError
from enums import DeploymentEdition
from models.account import Account, Tenant, TenantAccountJoin
from services.entities.feature_entities import SystemFeatureModel
from tests.unit_tests.config_override import apply_config_overrides
SQLITE_MODELS = (Account, Tenant, TenantAccountJoin)
@contextmanager
def _bind_database_session(session: Session) -> Generator[scoped_session[Session]]:
"""Bind the controller's session proxy to the SQLite test engine."""
database_session = scoped_session(sessionmaker(bind=session.get_bind(), expire_on_commit=False))
try:
with patch("controllers.console.auth.forgot_password.db.session", database_session):
yield database_session
finally:
database_session.remove()
@pytest.fixture(autouse=True)
def enable_password_login_wrappers(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep endpoint decorators deterministic without requiring the configured app database."""
apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.CLOUD)
monkeypatch.setattr(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
lambda: SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
),
)
class TestForgotPasswordSendEmailApi:
"""Test cases for sending password reset emails."""
@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
@patch("controllers.console.auth.forgot_password.AccountService.is_email_send_ip_limit")
@patch("controllers.console.auth.forgot_password.AccountService.send_reset_password_email")
def test_send_reset_email_success(
self,
mock_send_email,
mock_is_ip_limit,
app: Flask,
sqlite_session: Session,
):
# Arrange
mock_is_ip_limit.return_value = False
mock_send_email.return_value = "reset_token_123"
# Act
with (
_bind_database_session(sqlite_session),
app.test_request_context(
"/forgot-password", method="POST", json={"email": "test@example.com", "language": "en-US"}
),
):
api = ForgotPasswordSendEmailApi()
response = api.post()
# Assert
assert response["result"] == "success"
assert response["data"] == "reset_token_123"
mock_send_email.assert_called_once()
@patch("controllers.console.auth.forgot_password.AccountService.is_email_send_ip_limit")
def test_send_reset_email_ip_rate_limited(self, mock_is_ip_limit, app: Flask):
"""
Test password reset email blocked by IP rate limit.
Verifies that:
- EmailSendIpLimitError is raised when IP limit exceeded
- No email is sent when rate limited
"""
# Arrange
mock_is_ip_limit.return_value = True
# Act & Assert
with app.test_request_context("/forgot-password", method="POST", json={"email": "test@example.com"}):
api = ForgotPasswordSendEmailApi()
with pytest.raises(EmailSendIpLimitError):
api.post()
@pytest.mark.parametrize(
("language_input", "expected_language"),
[
("zh-Hans", "zh-Hans"),
("en-US", "en-US"),
("fr-FR", "en-US"), # Defaults to en-US for unsupported
(None, "en-US"), # Defaults to en-US when not provided
],
)
@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
@patch("controllers.console.auth.forgot_password.AccountService.is_email_send_ip_limit")
@patch("controllers.console.auth.forgot_password.AccountService.send_reset_password_email")
def test_send_reset_email_language_handling(
self,
mock_send_email,
mock_is_ip_limit,
language_input,
expected_language,
app: Flask,
sqlite_session: Session,
):
"""
Test password reset email with different language preferences.
Verifies that:
- Language parameter is correctly processed
- Unsupported languages default to en-US
"""
# Arrange
mock_is_ip_limit.return_value = False
mock_send_email.return_value = "token"
# Act
with (
_bind_database_session(sqlite_session),
app.test_request_context(
"/forgot-password", method="POST", json={"email": "test@example.com", "language": language_input}
),
):
api = ForgotPasswordSendEmailApi()
api.post()
# Assert
call_args = mock_send_email.call_args
assert call_args.kwargs["language"] == expected_language
class TestForgotPasswordCheckApi:
"""Test cases for verifying password reset codes."""
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.generate_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.reset_forgot_password_error_rate_limit")
def test_verify_code_success(
self,
mock_reset_rate_limit: MagicMock,
mock_generate_token: MagicMock,
mock_revoke_token: MagicMock,
mock_get_data: MagicMock,
mock_is_rate_limit,
app: Flask,
):
"""
Test successful verification code validation.
Verifies that:
- Valid code is accepted
- Old token is revoked
- New token is generated for reset phase
- Rate limit is reset on success
"""
# Arrange
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "test@example.com", "code": "123456"}
mock_generate_token.return_value = (None, "new_token")
# Act
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "test@example.com", "code": "123456", "token": "old_token"},
):
api = ForgotPasswordCheckApi()
response = api.post()
# Assert
assert response["is_valid"] is True
assert response["email"] == "test@example.com"
assert response["token"] == "new_token"
mock_revoke_token.assert_called_once_with("old_token")
mock_generate_token.assert_called_once_with(
"test@example.com", code="123456", additional_data={"phase": "reset"}
)
mock_reset_rate_limit.assert_called_once_with("test@example.com")
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.generate_reset_password_token")
@patch("controllers.console.auth.forgot_password.AccountService.reset_forgot_password_error_rate_limit")
def test_verify_code_preserves_token_email_case(
self,
mock_reset_rate_limit: MagicMock,
mock_generate_token: MagicMock,
mock_revoke_token: MagicMock,
mock_get_data: MagicMock,
mock_is_rate_limit,
app: Flask,
):
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "User@Example.com", "code": "999888"}
mock_generate_token.return_value = (None, "fresh-token")
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "user@example.com", "code": "999888", "token": "upper_token"},
):
response = ForgotPasswordCheckApi().post()
assert response == {"is_valid": True, "email": "user@example.com", "token": "fresh-token"}
mock_generate_token.assert_called_once_with(
"User@Example.com", code="999888", additional_data={"phase": "reset"}
)
mock_revoke_token.assert_called_once_with("upper_token")
mock_reset_rate_limit.assert_called_once_with("user@example.com")
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
def test_verify_code_rate_limited(self, mock_is_rate_limit, app: Flask):
"""
Test code verification blocked by rate limit.
Verifies that:
- EmailPasswordResetLimitError is raised when limit exceeded
- Prevents brute force attacks on verification codes
"""
# Arrange
mock_is_rate_limit.return_value = True
# Act & Assert
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "test@example.com", "code": "123456", "token": "token"},
):
api = ForgotPasswordCheckApi()
with pytest.raises(EmailPasswordResetLimitError):
api.post()
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
def test_verify_code_invalid_token(self, mock_get_data, mock_is_rate_limit, app: Flask):
"""
Test code verification with invalid token.
Verifies that:
- InvalidTokenError is raised for invalid/expired tokens
"""
# Arrange
mock_is_rate_limit.return_value = False
mock_get_data.return_value = None
# Act & Assert
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "test@example.com", "code": "123456", "token": "invalid_token"},
):
api = ForgotPasswordCheckApi()
with pytest.raises(InvalidTokenError):
api.post()
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
def test_verify_code_email_mismatch(self, mock_get_data, mock_is_rate_limit, app: Flask):
"""
Test code verification with mismatched email.
Verifies that:
- InvalidEmailError is raised when email doesn't match token
- Prevents token abuse
"""
# Arrange
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "original@example.com", "code": "123456"}
# Act & Assert
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "different@example.com", "code": "123456", "token": "token"},
):
api = ForgotPasswordCheckApi()
with pytest.raises(InvalidEmailError):
api.post()
@patch("controllers.console.auth.forgot_password.AccountService.is_forgot_password_error_rate_limit")
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.add_forgot_password_error_rate_limit")
def test_verify_code_wrong_code(self, mock_add_rate_limit, mock_get_data, mock_is_rate_limit, app: Flask):
"""
Test code verification with incorrect code.
Verifies that:
- EmailCodeError is raised for wrong code
- Rate limit counter is incremented
"""
# Arrange
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "test@example.com", "code": "123456"}
# Act & Assert
with app.test_request_context(
"/forgot-password/validity",
method="POST",
json={"email": "test@example.com", "code": "wrong_code", "token": "token"},
):
api = ForgotPasswordCheckApi()
with pytest.raises(EmailCodeError):
api.post()
mock_add_rate_limit.assert_called_once_with("test@example.com")
class TestForgotPasswordResetApi:
"""Test cases for resetting password with verified token."""
@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
def test_reset_password_success(
self,
mock_revoke_token: MagicMock,
mock_get_data: MagicMock,
app: Flask,
sqlite_session: Session,
):
"""
Test successful password reset.
Verifies that:
- Password is updated with new hashed value
- Token is revoked after use
- Success response is returned
"""
# Arrange
mock_get_data.return_value = {"email": "test@example.com", "phase": "reset"}
# Act
with _bind_database_session(sqlite_session) as database_session:
account = Account(name="Test User", email="test@example.com")
tenant = Tenant(name="Test Workspace")
database_session.add_all([account, tenant])
database_session.flush()
database_session.add(TenantAccountJoin(tenant_id=tenant.id, account_id=account.id))
database_session.commit()
account_id = account.id
with app.test_request_context(
"/forgot-password/resets",
method="POST",
json={
"token": "valid_token",
"new_password": "NewPass123!",
"password_confirm": "NewPass123!",
},
):
api = ForgotPasswordResetApi()
response = api.post()
updated_account = database_session.get(Account, account_id)
# Assert
assert response["result"] == "success"
mock_revoke_token.assert_called_once_with("valid_token")
assert updated_account is not None
assert updated_account.password is not None
assert updated_account.password_salt is not None
def test_reset_password_mismatch(self, app: Flask):
"""
Test password reset with mismatched passwords.
Verifies that:
- PasswordMismatchError is raised when passwords don't match
- No password update occurs
"""
# Act & Assert
with app.test_request_context(
"/forgot-password/resets",
method="POST",
json={"token": "token", "new_password": "NewPass123!", "password_confirm": "DifferentPass123!"},
):
api = ForgotPasswordResetApi()
with pytest.raises(PasswordMismatchError):
api.post()
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
def test_reset_password_invalid_token(self, mock_get_data, app: Flask):
"""
Test password reset with invalid token.
Verifies that:
- InvalidTokenError is raised for invalid/expired tokens
"""
# Arrange
mock_get_data.return_value = None
# Act & Assert
with app.test_request_context(
"/forgot-password/resets",
method="POST",
json={"token": "invalid_token", "new_password": "NewPass123!", "password_confirm": "NewPass123!"},
):
api = ForgotPasswordResetApi()
with pytest.raises(InvalidTokenError):
api.post()
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
def test_reset_password_wrong_phase(self, mock_get_data, app: Flask):
"""
Test password reset with token not in reset phase.
Verifies that:
- InvalidTokenError is raised when token is not in reset phase
- Prevents use of verification-phase tokens for reset
"""
# Arrange
mock_get_data.return_value = {"email": "test@example.com", "phase": "verify"}
# Act & Assert
with app.test_request_context(
"/forgot-password/resets",
method="POST",
json={"token": "token", "new_password": "NewPass123!", "password_confirm": "NewPass123!"},
):
api = ForgotPasswordResetApi()
with pytest.raises(InvalidTokenError):
api.post()
@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
@patch("controllers.console.auth.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.console.auth.forgot_password.AccountService.revoke_reset_password_token")
def test_reset_password_account_not_found(
self, mock_revoke_token, mock_get_data, app: Flask, sqlite_session: Session
):
"""
Test password reset for non-existent account.
Verifies that:
- AccountNotFound is raised when account doesn't exist
"""
# Arrange
mock_get_data.return_value = {"email": "nonexistent@example.com", "phase": "reset"}
# Act & Assert
with (
_bind_database_session(sqlite_session),
app.test_request_context(
"/forgot-password/resets",
method="POST",
json={"token": "token", "new_password": "NewPass123!", "password_confirm": "NewPass123!"},
),
):
api = ForgotPasswordResetApi()
with pytest.raises(AccountNotFound):
api.post()
api/tests/unit_tests/extensions/test_ext_application_services.py
+24 -1
View File
@@ -2,6 +2,7 @@
import json
from types import SimpleNamespace
from typing import cast
from unittest.mock import MagicMock, call, patch
from uuid import uuid4
@@ -23,7 +24,7 @@ from repositories.account_integration_repository import SQLAlchemyAccountIntegra
from repositories.account_repository import SQLAlchemyAccountRepository
from repositories.app_site_command_repository import AppSiteCommandRepository
from repositories.workflow_run_archive_repository import WorkflowRunArchiveBundleQueryRepository
from services import recommended_app_catalog_gateway
from services import account_forgot_password_service, recommended_app_catalog_gateway
from services.account_activation_adapters import (
BillingAccountActivationEligibility,
BillingWorkspaceMembershipCache,
@@ -38,6 +39,11 @@ from services.account_email_registration_adapters import (
RedisEmailRegistrationSecurityGateway,
TokenManagerEmailRegistrationTokenGateway,
)
from services.account_forgot_password_adapters import (
RateLimiterForgotPasswordSendLimiter,
RedisForgotPasswordSecurityGateway,
RedisForgotPasswordTokenGateway,
)
from services.app_site_service import AppSiteService
from services.auth.data_source_api_key_auth_service import DataSourceApiKeyAuthService
from services.billing_portal_service import BillingPortalService
@@ -371,6 +377,23 @@ def test_build_application_services_wires_account_profile_repository(
assert isinstance(accounts, SQLAlchemyAccountRepository)
assert accounts._session_factory is sqlite_session_factory
assert services.accounts.password._accounts is accounts
forgot_password = services.accounts.forgot_password
assert forgot_password._accounts is accounts
assert forgot_password._passwords is services.accounts.password._passwords
tokens = cast(RedisForgotPasswordTokenGateway, forgot_password._tokens)
send_limiter = cast(RateLimiterForgotPasswordSendLimiter, forgot_password._send_limits)
security = cast(RedisForgotPasswordSecurityGateway, forgot_password._security)
assert tokens._redis is send_limiter._rate_limiter._redis_client
assert send_limiter._rate_limiter.prefix == account_forgot_password_service.FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX
assert (
send_limiter._rate_limiter.max_attempts
== account_forgot_password_service.FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS
)
assert (
security._verification_failure_limit
== account_forgot_password_service.FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT
)
assert security._verification_key_prefix == account_forgot_password_service.FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX
assert services.accounts.initialization._accounts is accounts
assert not services.accounts.initialization._invitation_required
assert services.accounts.change_email._accounts is accounts
api/tests/unit_tests/repositories/test_account_repository.py
+18
View File
@@ -1,6 +1,7 @@
from datetime import datetime
import pytest
from sqlalchemy.exc import MultipleResultsFound
from sqlalchemy.orm import Session, sessionmaker
from models.account import Account, AccountIntegrate, AccountStatus, InvitationCode, InvitationCodeStatus
@@ -111,6 +112,23 @@ def test_account_repository_finds_email_with_lowercase_fallback(
assert account.email == "account@example.com"
def test_account_repository_fails_closed_for_duplicate_email(
sqlite_session: Session,
sqlite_session_factory: sessionmaker[Session],
) -> None:
sqlite_session.add_all(
[
Account(name="First", email="duplicate@example.com"),
Account(name="Second", email="duplicate@example.com"),
]
)
sqlite_session.commit()
repository = SQLAlchemyAccountRepository(sqlite_session_factory)
with pytest.raises(MultipleResultsFound):
repository.find_by_email("duplicate@example.com")
def test_account_integration_repository_lists_integrations(
sqlite_session: Session,
sqlite_session_factory: sessionmaker[Session],
api/tests/unit_tests/services/test_account_forgot_password_adapters.py
+188
View File
@@ -0,0 +1,188 @@
"""Tests for forgot-password infrastructure adapters."""
import json
from typing import cast
import pytest
from extensions.ext_redis import RedisClientWrapper
from services.account_forgot_password_adapters import RedisForgotPasswordTokenGateway
from services.entities.account_entities import ForgotPasswordResetToken, ForgotPasswordVerificationToken
class FakeRedis:
def __init__(self) -> None:
self.values: dict[str, object] = {}
def getdel(self, key: str) -> object | None:
return self.values.pop(key, None)
def get(self, key: str) -> object | None:
return self.values.get(key)
def setex(self, key: str, _seconds: int, value: object) -> bool:
self.values[key] = value
return True
def delete(self, key: str) -> int:
return int(self.values.pop(key, None) is not None)
def eval(
self,
_script: str,
_key_count: int,
key: str,
expected: str,
*args: object,
) -> int:
if self.values.get(key) != expected:
return 0
replacement, _expiry_seconds = args
self.values[key] = replacement
return 1
def _gateway(redis: FakeRedis | None = None) -> RedisForgotPasswordTokenGateway:
return RedisForgotPasswordTokenGateway(
redis=cast(RedisClientWrapper, redis or FakeRedis()),
expiry_seconds=300,
)
@pytest.mark.parametrize(
"payload",
[
{"email": "User@Example.com", "code": "123456", "phase": "reset"},
{"email": "user@example.com", "code": "123456", "phase": "unknown"},
{"email": None, "code": "123456"},
],
)
def test_token_gateway_read_verification_rejects_other_payloads_without_consuming(
payload: dict[str, object],
) -> None:
redis = FakeRedis()
redis.values["reset_password:token:token"] = json.dumps(payload)
assert _gateway(redis).read_verification("token") is None
assert "reset_password:token:token" in redis.values
def test_token_gateway_reads_verification_token_without_consuming() -> None:
redis = FakeRedis()
redis.values["reset_password:token:token"] = json.dumps({"email": "User@Example.com", "code": "123456"})
gateway = _gateway(redis)
expected = ForgotPasswordVerificationToken(email="User@Example.com", code="123456")
assert gateway.read_verification("token") == expected
assert gateway.read_verification("token") == expected
def test_token_gateway_allows_only_one_successful_reset_claim() -> None:
redis = FakeRedis()
redis.values["reset_password:token:token"] = json.dumps(
{"email": "user@example.com", "code": "123456", "phase": "reset"}
)
gateway = _gateway(redis)
assert isinstance(gateway.claim_reset("token"), ForgotPasswordResetToken)
assert gateway.claim_reset("token") is None
def test_token_gateway_rejects_superseded_account_token() -> None:
redis = FakeRedis()
redis.values["reset_password:token:v1"] = json.dumps(
{"account_id": "account-1", "email": "user@example.com", "code": "123456", "phase": "reset"}
)
redis.values["reset_password:account:account-1"] = "v2"
assert _gateway(redis).claim_reset("v1") is None
assert "reset_password:token:v1" not in redis.values
def test_token_gateway_serializes_reset_phase() -> None:
redis = FakeRedis()
token = _gateway(redis).issue(
ForgotPasswordResetToken(email="User@Example.com", code="123456", account_id="account-1")
)
assert json.loads(cast(str, redis.values[f"reset_password:token:{token}"])) == {
"token_type": "reset_password",
"account_id": "account-1",
"email": "User@Example.com",
"code": "123456",
"phase": "reset",
}
assert redis.values["reset_password:account:account-1"] == token
def test_token_gateway_binds_verification_token_to_existing_account() -> None:
redis = FakeRedis()
token = _gateway(redis).issue(
ForgotPasswordVerificationToken(
email="User@Example.com",
code="123456",
account_id="account-1",
)
)
assert redis.values["reset_password:account:account-1"] == token
def test_token_gateway_promotes_only_the_current_account_token() -> None:
redis = FakeRedis()
current_key = "reset_password:account:account-1"
redis.values["reset_password:token:v1"] = json.dumps(
{"account_id": "account-1", "email": "user@example.com", "code": "123456"}
)
redis.values[current_key] = "v1"
gateway = _gateway(redis)
claimed = gateway.read_verification("v1")
assert isinstance(claimed, ForgotPasswordVerificationToken)
reset_token = gateway.promote("v1", claimed.promote())
assert reset_token is not None
assert redis.values[current_key] == reset_token
assert "reset_password:token:v1" not in redis.values
assert json.loads(cast(str, redis.values[f"reset_password:token:{reset_token}"]))["phase"] == "reset"
def test_token_gateway_promotion_does_not_revoke_concurrent_new_token() -> None:
redis = FakeRedis()
old_key = "reset_password:token:v1"
current_key = "reset_password:account:account-1"
redis.values[old_key] = json.dumps({"account_id": "account-1", "email": "user@example.com", "code": "123456"})
redis.values[current_key] = "v1"
gateway = _gateway(redis)
claimed = gateway.read_verification("v1")
assert isinstance(claimed, ForgotPasswordVerificationToken)
new_token = gateway.issue(
ForgotPasswordVerificationToken(
email="user@example.com",
code="654321",
account_id="account-1",
)
)
assert gateway.promote("v1", claimed.promote()) is None
assert redis.values[current_key] == new_token
assert old_key not in redis.values
assert f"reset_password:token:{new_token}" in redis.values
def test_token_gateway_consumes_legacy_verification_token_when_promoting() -> None:
redis = FakeRedis()
old_key = "reset_password:token:legacy"
redis.values[old_key] = json.dumps({"email": "user@example.com", "code": "123456"})
gateway = _gateway(redis)
claimed = gateway.read_verification("legacy")
assert isinstance(claimed, ForgotPasswordVerificationToken)
reset_token = gateway.promote("legacy", claimed.promote())
assert reset_token is not None
assert old_key not in redis.values
assert f"reset_password:token:{reset_token}" in redis.values
api/tests/unit_tests/services/test_account_forgot_password_service.py
+510
View File
@@ -0,0 +1,510 @@
"""Unit tests for the framework-neutral forgot-password application service."""
from __future__ import annotations
from datetime import datetime
from typing import cast
import pytest
from services import account_errors
from services.account_forgot_password_service import (
AccountForgotPasswordService,
ForgotPasswordCodeGenerator,
ForgotPasswordNotificationGateway,
ForgotPasswordRegistrationPolicy,
ForgotPasswordSecurityGateway,
ForgotPasswordSendLimiter,
ForgotPasswordTokenGateway,
)
from services.account_ports import AccountPasswordHasher, AccountRepository
from services.entities.account_entities import (
AccountPasswordDigest,
AccountSnapshot,
ForgotPasswordResetToken,
ForgotPasswordToken,
ForgotPasswordVerificationToken,
)
def _account(*, email: str = "User@Example.com") -> AccountSnapshot:
return AccountSnapshot(
id="account-1",
name="User",
email=email,
avatar=None,
is_password_set=False,
interface_language="en-US",
interface_theme="light",
timezone="UTC",
last_login_at=None,
last_login_ip=None,
status="active",
initialized_at=None,
created_at=datetime(2026, 8, 1),
)
class FakeAccounts:
def __init__(self, account: AccountSnapshot | None = None) -> None:
self.account = account
self.requested_ids: list[str] = []
self.requested_emails: list[str] = []
self.updated: tuple[str, AccountPasswordDigest] | None = None
self.update_succeeds = True
self.update_error: Exception | None = None
def get(self, account_id: str) -> AccountSnapshot | None:
self.requested_ids.append(account_id)
return self.account
def find_by_email(self, email: str) -> AccountSnapshot | None:
self.requested_emails.append(email)
return self.account
def update_password(self, account_id: str, password: AccountPasswordDigest) -> AccountSnapshot | None:
self.updated = (account_id, password)
if self.update_error is not None:
raise self.update_error
return self.account if self.update_succeeds else None
class FakePasswords:
def __init__(self) -> None:
self.plaintext: str | None = None
self.digest = AccountPasswordDigest(password_hash="hash", password_salt="salt")
def hash(self, password: str) -> AccountPasswordDigest:
self.plaintext = password
return self.digest
class FakeTokens:
def __init__(self) -> None:
self.stored: dict[str, ForgotPasswordToken | None] = {}
self.issued: list[ForgotPasswordToken] = []
self.read_tokens: list[str] = []
self.claimed_reset_tokens: list[str] = []
self.promotions: list[tuple[str, ForgotPasswordResetToken]] = []
self.promotion_succeeds = True
def read_verification(self, token: str) -> ForgotPasswordVerificationToken | None:
self.read_tokens.append(token)
token_data = self.stored.get(token)
return token_data if isinstance(token_data, ForgotPasswordVerificationToken) else None
def claim_reset(self, token: str) -> ForgotPasswordResetToken | None:
self.claimed_reset_tokens.append(token)
token_data = self.stored.pop(token, None)
return token_data if isinstance(token_data, ForgotPasswordResetToken) else None
def issue(self, token_data: ForgotPasswordToken) -> str:
self.issued.append(token_data)
return f"issued-{len(self.issued)}"
def promote(self, claimed_token: str, token_data: ForgotPasswordResetToken) -> str | None:
self.promotions.append((claimed_token, token_data))
self.stored.pop(claimed_token, None)
if not self.promotion_succeeds:
return None
return self.issue(token_data)
class FakeCodeGenerator:
def generate(self) -> str:
return "123456"
class FakeNotifications:
def __init__(self) -> None:
self.sent: list[dict[str, object]] = []
def send(
self,
*,
email: str,
code: str,
language: str,
account_exists: bool,
registration_allowed: bool,
) -> None:
self.sent.append(
{
"email": email,
"code": code,
"language": language,
"account_exists": account_exists,
"registration_allowed": registration_allowed,
}
)
class FakeSendLimiter:
retry_after_minutes = 1
def __init__(self) -> None:
self.limited = False
self.checked: list[str] = []
self.recorded: list[str] = []
def is_limited(self, email: str) -> bool:
self.checked.append(email)
return self.limited
def record(self, email: str) -> None:
self.recorded.append(email)
class FakeSecurity:
def __init__(self) -> None:
self.ip_limited = False
self.verification_limited = False
self.checked_ips: list[str] = []
self.failed_emails: list[str] = []
self.reset_emails: list[str] = []
def is_ip_limited(self, ip_address: str) -> bool:
self.checked_ips.append(ip_address)
return self.ip_limited
def is_verification_limited(self, _email: str) -> bool:
return self.verification_limited
def record_verification_failure(self, email: str) -> None:
self.failed_emails.append(email)
def reset_verification_failures(self, email: str) -> None:
self.reset_emails.append(email)
class FakeRegistrationPolicy:
def __init__(self, allowed: bool = True) -> None:
self.allowed = allowed
def is_registration_allowed(self) -> bool:
return self.allowed
class Dependencies:
def __init__(self, account: AccountSnapshot | None = None) -> None:
self.accounts = FakeAccounts(account)
self.passwords = FakePasswords()
self.tokens = FakeTokens()
self.codes = FakeCodeGenerator()
self.notifications = FakeNotifications()
self.send_limits = FakeSendLimiter()
self.security = FakeSecurity()
self.registration = FakeRegistrationPolicy()
self.service = AccountForgotPasswordService(
accounts=cast(AccountRepository, self.accounts),
passwords=cast(AccountPasswordHasher, self.passwords),
tokens=cast(ForgotPasswordTokenGateway, self.tokens),
codes=cast(ForgotPasswordCodeGenerator, self.codes),
notifications=cast(ForgotPasswordNotificationGateway, self.notifications),
send_limits=cast(ForgotPasswordSendLimiter, self.send_limits),
security=cast(ForgotPasswordSecurityGateway, self.security),
registration=cast(ForgotPasswordRegistrationPolicy, self.registration),
)
def test_send_code_preserves_existing_account_email_case() -> None:
dependencies = Dependencies(_account())
token = dependencies.service.send_code(
email="USER@Example.com",
language="zh-Hans",
ip_address="127.0.0.1",
)
assert token == "issued-1"
assert dependencies.accounts.requested_emails == ["USER@Example.com"]
assert dependencies.tokens.issued == [
ForgotPasswordVerificationToken(email="User@Example.com", code="123456", account_id="account-1")
]
assert dependencies.notifications.sent == [
{
"email": "User@Example.com",
"code": "123456",
"language": "zh-Hans",
"account_exists": True,
"registration_allowed": True,
}
]
assert dependencies.send_limits.recorded == ["User@Example.com"]
def test_send_code_normalizes_unknown_account_email() -> None:
dependencies = Dependencies()
dependencies.registration.allowed = False
dependencies.service.send_code(
email="MISSING@Example.com",
language="en-US",
ip_address="127.0.0.1",
)
assert dependencies.tokens.issued == [ForgotPasswordVerificationToken(email="missing@example.com", code="123456")]
assert dependencies.notifications.sent[0]["account_exists"] is False
assert dependencies.notifications.sent[0]["registration_allowed"] is False
def test_send_code_stops_at_ip_limit() -> None:
dependencies = Dependencies(_account())
dependencies.security.ip_limited = True
with pytest.raises(account_errors.ForgotPasswordSendIPLimitedError):
dependencies.service.send_code(
email="user@example.com",
language="en-US",
ip_address="127.0.0.1",
)
assert dependencies.accounts.requested_emails == []
assert dependencies.tokens.issued == []
def test_send_code_stops_at_address_limit() -> None:
dependencies = Dependencies(_account())
dependencies.send_limits.limited = True
with pytest.raises(account_errors.ForgotPasswordSendRateLimitError) as exc_info:
dependencies.service.send_code(
email="user@example.com",
language="en-US",
ip_address="127.0.0.1",
)
assert exc_info.value.retry_after_minutes == 1
assert dependencies.tokens.issued == []
def test_verify_code_promotes_token_and_resets_failure_limit() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["verification-token"] = ForgotPasswordVerificationToken(
email="User@Example.com",
code="123456",
account_id="account-1",
)
result = dependencies.service.verify_code(
email="USER@example.com",
code="123456",
token="verification-token",
)
assert result.email == "user@example.com"
assert result.token == "issued-1"
assert dependencies.tokens.read_tokens == ["verification-token"]
assert "verification-token" not in dependencies.tokens.stored
assert dependencies.tokens.issued == [
ForgotPasswordResetToken(email="User@Example.com", code="123456", account_id="account-1")
]
assert dependencies.tokens.promotions == [
(
"verification-token",
ForgotPasswordResetToken(email="User@Example.com", code="123456", account_id="account-1"),
)
]
assert dependencies.security.reset_emails == ["user@example.com"]
def test_verify_code_rejects_promotion_after_token_is_superseded() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["verification-token"] = ForgotPasswordVerificationToken(
email="user@example.com",
code="123456",
account_id="account-1",
)
dependencies.tokens.promotion_succeeds = False
with pytest.raises(account_errors.InvalidForgotPasswordTokenError):
dependencies.service.verify_code(
email="user@example.com",
code="123456",
token="verification-token",
)
assert "verification-token" not in dependencies.tokens.stored
assert dependencies.security.reset_emails == []
def test_verify_code_rejects_reset_phase_token() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["reset-token"] = ForgotPasswordResetToken(
email="user@example.com",
code="123456",
)
with pytest.raises(account_errors.InvalidForgotPasswordTokenError):
dependencies.service.verify_code(
email="user@example.com",
code="123456",
token="reset-token",
)
assert "reset-token" in dependencies.tokens.stored
def test_verify_code_rejects_mismatched_email() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["token"] = ForgotPasswordVerificationToken(
email="original@example.com",
code="123456",
)
with pytest.raises(account_errors.InvalidForgotPasswordEmailError):
dependencies.service.verify_code(
email="different@example.com",
code="123456",
token="token",
)
assert "token" in dependencies.tokens.stored
def test_verify_code_keeps_token_after_wrong_code_and_allows_retry() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["token"] = ForgotPasswordVerificationToken(
email="user@example.com",
code="123456",
)
with pytest.raises(account_errors.InvalidForgotPasswordCodeError):
dependencies.service.verify_code(
email="USER@example.com",
code="wrong",
token="token",
)
assert dependencies.security.failed_emails == ["user@example.com"]
assert "token" in dependencies.tokens.stored
result = dependencies.service.verify_code(
email="USER@example.com",
code="123456",
token="token",
)
assert result.token == "issued-1"
assert "token" not in dependencies.tokens.stored
def test_verify_code_stops_at_failure_limit() -> None:
dependencies = Dependencies()
dependencies.security.verification_limited = True
with pytest.raises(account_errors.ForgotPasswordVerificationLimitError):
dependencies.service.verify_code(
email="user@example.com",
code="123456",
token="token",
)
def test_reset_claims_token_and_updates_password() -> None:
dependencies = Dependencies(_account(email="user@example.com"))
dependencies.tokens.stored["reset-token"] = ForgotPasswordResetToken(
email="User@Example.com",
code="123456",
account_id="account-1",
)
dependencies.service.reset(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
)
assert dependencies.tokens.claimed_reset_tokens == ["reset-token"]
assert "reset-token" not in dependencies.tokens.stored
assert dependencies.accounts.requested_ids == ["account-1"]
assert dependencies.accounts.requested_emails == []
assert dependencies.passwords.plaintext == "ValidPass123!"
assert dependencies.accounts.updated == ("account-1", dependencies.passwords.digest)
def test_reset_rejects_password_mismatch_before_reading_token() -> None:
dependencies = Dependencies(_account())
with pytest.raises(account_errors.ForgotPasswordMismatchError):
dependencies.service.reset(
token="reset-token",
new_password="ValidPass123!",
password_confirm="OtherPass123!",
)
assert dependencies.tokens.claimed_reset_tokens == []
def test_reset_consumes_claimed_token_when_account_is_missing() -> None:
dependencies = Dependencies()
dependencies.tokens.stored["reset-token"] = ForgotPasswordResetToken(
email="missing@example.com",
code="123456",
account_id="missing-account",
)
with pytest.raises(account_errors.AccountNotFoundError):
dependencies.service.reset(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
)
assert "reset-token" not in dependencies.tokens.stored
assert dependencies.accounts.updated is None
def test_reset_consumes_claimed_token_when_password_commit_fails() -> None:
dependencies = Dependencies(_account())
dependencies.tokens.stored["reset-token"] = ForgotPasswordResetToken(
email="user@example.com",
code="123456",
account_id="account-1",
)
dependencies.accounts.update_error = RuntimeError("database unavailable")
with pytest.raises(RuntimeError, match="database unavailable"):
dependencies.service.reset(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
)
assert dependencies.tokens.claimed_reset_tokens == ["reset-token"]
assert "reset-token" not in dependencies.tokens.stored
def test_reset_supports_legacy_token_without_account_id() -> None:
dependencies = Dependencies(_account(email="user@example.com"))
dependencies.tokens.stored["legacy-token"] = ForgotPasswordResetToken(
email="User@Example.com",
code="123456",
)
dependencies.service.reset(
token="legacy-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
)
assert dependencies.accounts.requested_ids == []
assert dependencies.accounts.requested_emails == ["User@Example.com"]
def test_reset_rejects_account_when_email_changed_after_token_issue() -> None:
dependencies = Dependencies(_account(email="changed@example.com"))
dependencies.tokens.stored["reset-token"] = ForgotPasswordResetToken(
email="original@example.com",
code="123456",
account_id="account-1",
)
with pytest.raises(account_errors.AccountNotFoundError):
dependencies.service.reset(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
)
assert dependencies.accounts.updated is None
api/tests/unit_tests/services/test_account_password_hasher.py
+3 -3
View File
@@ -1,11 +1,11 @@
import pytest
from services.account_errors import InvalidAccountPasswordError
from services.account_password_hasher import LegacyAccountPasswordHasher
from services.account_password_hasher import DefaultAccountPasswordHasher
def test_hash_produces_a_digest_that_can_be_verified() -> None:
passwords = LegacyAccountPasswordHasher()
passwords = DefaultAccountPasswordHasher()
digest = passwords.hash("password123")
@@ -22,7 +22,7 @@ def test_hash_produces_a_digest_that_can_be_verified() -> None:
def test_hash_maps_password_policy_validation() -> None:
passwords = LegacyAccountPasswordHasher()
passwords = DefaultAccountPasswordHasher()
with pytest.raises(InvalidAccountPasswordError, match="Password must contain letters and numbers"):
passwords.hash("letters-only")
api/tests/unit_tests/services/test_account_security_gateway.py
+47
View File
@@ -0,0 +1,47 @@
"""Tests for shared account email security infrastructure."""
import logging
from typing import cast
import pytest
from redis import RedisError
from extensions.ext_redis import RedisClientWrapper
from services.account_security_gateway import RedisAccountEmailSecurityGateway
class FailingRedis:
def get(self, _key: str) -> None:
raise RedisError("offline")
def delete(self, _key: str) -> None:
raise RedisError("offline")
@pytest.fixture
def gateway() -> RedisAccountEmailSecurityGateway:
return RedisAccountEmailSecurityGateway(
redis=cast(RedisClientWrapper, FailingRedis()),
email_send_ip_limit_per_minute=3,
verification_failure_limit=5,
verification_lockout_duration=3600,
verification_key_prefix="forgot_password_error_rate_limit",
)
def test_redis_failures_remain_fail_open_but_are_logged(
gateway: RedisAccountEmailSecurityGateway,
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level(logging.WARNING, logger="services.account_security_gateway"):
assert gateway.is_ip_limited("127.0.0.1") is False
assert gateway.is_verification_limited("user@example.com") is False
gateway.record_verification_failure("user@example.com")
gateway.reset_verification_failures("user@example.com")
assert [record.getMessage() for record in caplog.records] == [
"Redis unavailable while checking the account email IP limit",
"Redis unavailable while checking the account email verification limit",
"Redis unavailable while recording an account email verification failure",
"Redis unavailable while resetting account email verification failures",
]
packages/contracts/generated/api/console/forgot-password/types.gen.ts
-1
View File
@@ -10,7 +10,6 @@ export type ForgotPasswordSendPayload = {
}
export type ForgotPasswordEmailResponse = {
code?: string | null
data?: string | null
result: string
}
packages/contracts/generated/api/console/forgot-password/zod.gen.ts
-1
View File
@@ -14,7 +14,6 @@ export const zForgotPasswordSendPayload = z.object({
* ForgotPasswordEmailResponse
*/
export const zForgotPasswordEmailResponse = z.object({
code: z.string().nullish(),
data: z.string().nullish(),
result: z.string(),
})