94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
"""Password reset service for VoIdeaAI.
|
|
|
|
Generates and validates reset tokens (JWT, type=reset, exp=1h).
|
|
Sends reset link via email_service when SMTP is configured.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.security import create_reset_token, decode_reset_token, get_password_hash
|
|
from app.models.user import User
|
|
from app.services.email_service import send_email
|
|
|
|
RESET_TOKEN_EXPIRE_HOURS = 1
|
|
|
|
|
|
async def send_reset_email(db: AsyncSession, email: str) -> tuple[bool, str]:
|
|
"""Send password reset email.
|
|
|
|
Args:
|
|
db: Database session
|
|
email: User's email address
|
|
|
|
Returns:
|
|
Tuple of (success, message)
|
|
"""
|
|
result = await db.execute(select(User).where(User.email == email))
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
return True, "If the email exists, a reset link has been sent."
|
|
|
|
settings = get_settings()
|
|
if not settings.smtp_host:
|
|
return False, "SMTP not configured. Password reset is unavailable."
|
|
|
|
token = create_reset_token({"sub": str(user.id)})
|
|
|
|
reset_link = f"{settings.server_external_url}/reset-password?token={token}"
|
|
html = _render_reset_email(user.display_name, reset_link)
|
|
|
|
sent = await send_email(user.email, "Сброс пароля — VoIdeaAI", html)
|
|
return sent, "If the email exists, a reset link has been sent."
|
|
|
|
|
|
async def reset_password(db: AsyncSession, token: str, new_password: str) -> tuple[bool, str]:
|
|
"""Reset password using a valid reset token.
|
|
|
|
Args:
|
|
db: Database session
|
|
token: JWT reset token
|
|
new_password: New password
|
|
|
|
Returns:
|
|
Tuple of (success, message)
|
|
"""
|
|
payload = decode_reset_token(token)
|
|
if not payload or payload.get("type") != "reset":
|
|
return False, "Invalid or expired reset token."
|
|
|
|
user_id = payload.get("sub")
|
|
if not user_id:
|
|
return False, "Invalid token payload."
|
|
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
return False, "User not found."
|
|
|
|
if not user.password_hash:
|
|
return False, "Cannot reset password for OAuth-only account."
|
|
|
|
user.password_hash = get_password_hash(new_password)
|
|
await db.commit()
|
|
return True, "Password has been reset successfully."
|
|
|
|
|
|
def _render_reset_email(username: str, reset_link: str) -> str:
|
|
"""Render password reset email HTML."""
|
|
return f"""<!DOCTYPE html>
|
|
<html><head><meta charset="utf-8"></head>
|
|
<body style="font-family:sans-serif;max-width:600px;margin:0 auto;">
|
|
<h2>Сброс пароля — VoIdeaAI</h2>
|
|
<p>Привет, {username}!</p>
|
|
<p>Вы запросили сброс пароля. Нажмите на ссылку ниже:</p>
|
|
<p><a href="{reset_link}" style="display:inline-block;padding:12px 24px;background:#4f46e5;color:#fff;text-decoration:none;border-radius:8px;">Сбросить пароль</a></p>
|
|
<p>Ссылка действует 1 час.</p>
|
|
<p>Если вы не запрашивали сброс — проигнорируйте это письмо.</p>
|
|
<hr><p style="color:#666;font-size:12px;">VoIdeaAI — идеи рождаются вслух, решения приходят мгновенно!</p>
|
|
</body></html>"""
|