feat: add password reset functionality with email templates

services/auth.py

@@ -1,5 +1,8 @@
+import asyncio
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Annotated
+import secrets
 
 import jwt
 from fastapi import Depends, HTTPException, status
@@ -10,9 +13,78 @@ from pydantic import BaseModel
 
 from models.database import UserAccount
 from services.config import ConfigCenter
+from services.logging import LogService
 
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 365
+PASSWORD_RESET_TOKEN_EXPIRE_MINUTES = 10
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+@dataclass
+class PasswordResetEntry:
+    user_id: int
+    email: str
+    username: str
+    expires_at: datetime
+    used: bool = False
+
+
+class PasswordResetStore:
+    _tokens: dict[str, PasswordResetEntry] = {}
+    _lock = asyncio.Lock()
+
+    @classmethod
+    def _cleanup(cls):
+        now = _now()
+        for token, record in list(cls._tokens.items()):
+            if record.used or record.expires_at < now:
+                cls._tokens.pop(token, None)
+
+    @classmethod
+    async def create(cls, user: UserAccount) -> str:
+        async with cls._lock:
+            cls._cleanup()
+            for key, record in list(cls._tokens.items()):
+                if record.user_id == user.id:
+                    cls._tokens.pop(key, None)
+            token = secrets.token_urlsafe(32)
+            expires_at = _now() + timedelta(minutes=PASSWORD_RESET_TOKEN_EXPIRE_MINUTES)
+            cls._tokens[token] = PasswordResetEntry(
+                user_id=user.id,
+                email=user.email or "",
+                username=user.username,
+                expires_at=expires_at,
+            )
+            return token
+
+    @classmethod
+    async def get(cls, token: str) -> PasswordResetEntry | None:
+        async with cls._lock:
+            cls._cleanup()
+            record = cls._tokens.get(token)
+            if not record or record.used:
+                return None
+            return record
+
+    @classmethod
+    async def mark_used(cls, token: str) -> None:
+        async with cls._lock:
+            record = cls._tokens.get(token)
+            if record:
+                record.used = True
+            cls._cleanup()
+
+    @classmethod
+    async def invalidate_user(cls, user_id: int, except_token: str | None = None) -> None:
+        async with cls._lock:
+            for key, record in list(cls._tokens.items()):
+                if record.user_id == user_id and key != except_token:
+                    cls._tokens.pop(key, None)
+            cls._cleanup()
 
 
 async def get_secret_key():
@@ -132,6 +204,94 @@ async def create_access_token(data: dict, expires_delta: timedelta | None = None
     return encoded_jwt
 
 
+def _normalize_email(email: str | None) -> str:
+    return (email or "").strip().lower()
+
+
+async def _send_password_reset_email(user: UserAccount, token: str) -> None:
+    from services.email import EmailService
+
+    app_domain = await ConfigCenter.get("APP_DOMAIN", None)
+    base_url = (app_domain or "http://localhost:5173").rstrip("/")
+    reset_link = f"{base_url}/reset-password?token={token}"
+    await EmailService.enqueue_email(
+        recipients=[user.email],
+        subject="Foxel 密码重置",
+        template="password_reset",
+        context={
+            "username": user.username,
+            "reset_link": reset_link,
+            "expire_minutes": PASSWORD_RESET_TOKEN_EXPIRE_MINUTES,
+        },
+    )
+
+
+async def request_password_reset(email: str) -> bool:
+    normalized = _normalize_email(email)
+    if not normalized:
+        return False
+    user = await UserAccount.get_or_none(email=normalized)
+    if not user or not user.email:
+        return False
+
+    token = await PasswordResetStore.create(user)
+    try:
+        await _send_password_reset_email(user, token)
+    except Exception as exc:  # noqa: BLE001
+        await PasswordResetStore.mark_used(token)
+        await PasswordResetStore.invalidate_user(user.id)
+        await LogService.error(
+            "auth",
+            f"Failed to enqueue password reset email: {exc}",
+            details={"user_id": user.id},
+            user_id=user.id,
+        )
+        raise HTTPException(status_code=500, detail="邮件发送失败") from exc
+    await LogService.action(
+        "auth",
+        "Password reset requested",
+        details={"user_id": user.id},
+        user_id=user.id,
+    )
+    return True
+
+
+async def verify_password_reset_token(token: str) -> UserAccount:
+    record = await PasswordResetStore.get(token)
+    if not record:
+        raise HTTPException(status_code=400, detail="重置链接无效")
+    user = await UserAccount.get_or_none(id=record.user_id)
+    if not user:
+        raise HTTPException(status_code=400, detail="重置链接无效")
+    if record.expires_at < _now():
+        await PasswordResetStore.mark_used(token)
+        raise HTTPException(status_code=400, detail="重置链接已过期")
+    return user
+
+
+async def reset_password_with_token(token: str, new_password: str) -> None:
+    record = await PasswordResetStore.get(token)
+    if not record:
+        raise HTTPException(status_code=400, detail="重置链接无效")
+    if record.expires_at < _now():
+        await PasswordResetStore.mark_used(token)
+        raise HTTPException(status_code=400, detail="重置链接已过期")
+
+    user = await UserAccount.get_or_none(id=record.user_id)
+    if not user:
+        raise HTTPException(status_code=400, detail="重置链接无效")
+    user.hashed_password = get_password_hash(new_password)
+    await user.save(update_fields=["hashed_password"])
+    await PasswordResetStore.mark_used(token)
+    await PasswordResetStore.invalidate_user(user.id)
+    await LogService.action(
+        "auth",
+        "Password reset via email",
+        details={"user_id": user.id},
+        user_id=user.id,
+    )
+
+
 async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,

services/email.py

@@ -0,0 +1,201 @@
+import asyncio
+import json
+import re
+import smtplib
+from email.message import EmailMessage
+from email.utils import formataddr
+from enum import Enum
+from pathlib import Path
+from string import Template
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, EmailStr, Field, ValidationError
+
+from services.config import ConfigCenter
+from services.logging import LogService
+
+
+class EmailSecurity(str, Enum):
+    NONE = "none"
+    SSL = "ssl"
+    STARTTLS = "starttls"
+
+
+class EmailConfig(BaseModel):
+    host: str
+    port: int = Field(..., gt=0)
+    username: Optional[str] = None
+    password: Optional[str] = None
+    sender_email: EmailStr
+    sender_name: Optional[str] = None
+    security: EmailSecurity = EmailSecurity.NONE
+    timeout: float = Field(default=30.0, gt=0.0)
+
+
+class EmailSendPayload(BaseModel):
+    recipients: List[EmailStr] = Field(..., min_length=1)
+    subject: str = Field(..., min_length=1)
+    template: str = Field(..., min_length=1)
+    context: Dict[str, Any] = Field(default_factory=dict)
+
+
+class EmailTemplateRenderer:
+    ROOT = Path("templates/email")
+
+    @classmethod
+    def _resolve_path(cls, template_name: str) -> Path:
+        if not re.fullmatch(r"[A-Za-z0-9_\-]+", template_name):
+            raise ValueError("Invalid template name")
+        return cls.ROOT / f"{template_name}.html"
+
+    @classmethod
+    async def list_templates(cls) -> list[str]:
+        cls.ROOT.mkdir(parents=True, exist_ok=True)
+        return sorted(
+            path.stem
+            for path in cls.ROOT.glob("*.html")
+            if path.is_file()
+        )
+
+    @classmethod
+    async def load(cls, template_name: str) -> str:
+        path = cls._resolve_path(template_name)
+        if not path.is_file():
+            raise FileNotFoundError(f"Email template '{template_name}' not found")
+        return await asyncio.to_thread(path.read_text, encoding="utf-8")
+
+    @classmethod
+    async def save(cls, template_name: str, content: str) -> None:
+        path = cls._resolve_path(template_name)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        await asyncio.to_thread(path.write_text, content, encoding="utf-8")
+
+    @classmethod
+    async def render(cls, template_name: str, context: Dict[str, Any]) -> str:
+        raw = await cls.load(template_name)
+        context = {k: str(v) for k, v in (context or {}).items()}
+        return Template(raw).safe_substitute(context)
+
+
+class EmailService:
+    CONFIG_KEY = "EMAIL_CONFIG"
+
+    @classmethod
+    async def _load_config(cls) -> EmailConfig:
+        raw_config = await ConfigCenter.get(cls.CONFIG_KEY)
+        if raw_config is None:
+            raise ValueError("Email configuration not found")
+
+        if isinstance(raw_config, str):
+            raw_config = raw_config.strip()
+            data: Any = json.loads(raw_config) if raw_config else {}
+        elif isinstance(raw_config, dict):
+            data = raw_config
+        else:
+            raise ValueError("Invalid email configuration format")
+
+        try:
+            return EmailConfig(**data)
+        except ValidationError as exc:
+            raise ValueError(f"Invalid email configuration: {exc}") from exc
+
+    @staticmethod
+    def _html_to_text(html: str) -> str:
+        stripped = re.sub(r"<[^>]+>", " ", html)
+        return " ".join(stripped.split())
+
+    @classmethod
+    async def _deliver(cls, config: EmailConfig, payload: EmailSendPayload, html_body: str):
+        message = EmailMessage()
+        message["Subject"] = payload.subject
+        message["From"] = formataddr((config.sender_name or str(config.sender_email), str(config.sender_email)))
+        message["To"] = ", ".join([str(addr) for addr in payload.recipients])
+
+        plain_body = cls._html_to_text(html_body)
+        message.set_content(plain_body or html_body)
+        message.add_alternative(html_body, subtype="html")
+
+        await asyncio.to_thread(cls._deliver_sync, config, message)
+
+    @staticmethod
+    def _deliver_sync(config: EmailConfig, message: EmailMessage):
+        if config.security == EmailSecurity.SSL:
+            smtp: smtplib.SMTP = smtplib.SMTP_SSL(config.host, config.port, timeout=config.timeout)
+        else:
+            smtp = smtplib.SMTP(config.host, config.port, timeout=config.timeout)
+
+        try:
+            if config.security == EmailSecurity.STARTTLS:
+                smtp.starttls()
+            if config.username and config.password:
+                smtp.login(config.username, config.password)
+            smtp.send_message(message)
+        finally:
+            try:
+                smtp.quit()
+            except Exception:
+                pass
+
+    @classmethod
+    async def enqueue_email(
+        cls,
+        recipients: List[str],
+        subject: str,
+        template: str,
+        context: Optional[Dict[str, Any]] = None,
+    ):
+        from services.task_queue import TaskProgress, task_queue_service
+
+        payload = EmailSendPayload(
+            recipients=recipients,
+            subject=subject,
+            template=template,
+            context=context or {},
+        )
+
+        task = await task_queue_service.add_task(
+            "send_email",
+            payload.model_dump(mode="json"),
+        )
+
+        await task_queue_service.update_progress(
+            task.id,
+            TaskProgress(stage="queued", percent=0.0, detail="Waiting to send"),
+        )
+        await LogService.action(
+            "email_service",
+            "Email task enqueued",
+            details={"task_id": task.id, "subject": subject, "template": template},
+        )
+        return task
+
+    @classmethod
+    async def send_from_task(cls, task_id: str, data: Dict[str, Any]):
+        from services.task_queue import TaskProgress, task_queue_service
+
+        payload = EmailSendPayload(**data)
+
+        await task_queue_service.update_progress(
+            task_id,
+            TaskProgress(stage="preparing", percent=10.0, detail="Rendering template"),
+        )
+
+        config = await cls._load_config()
+        html_body = await EmailTemplateRenderer.render(payload.template, payload.context)
+
+        await task_queue_service.update_progress(
+            task_id,
+            TaskProgress(stage="sending", percent=60.0, detail="Sending message"),
+        )
+
+        await cls._deliver(config, payload, html_body)
+
+        await task_queue_service.update_progress(
+            task_id,
+            TaskProgress(stage="completed", percent=100.0, detail="Email sent"),
+        )
+        await LogService.info(
+            "email_service",
+            "Email sent",
+            details={"task_id": task_id, "subject": payload.subject},
+        )

services/task_queue.py

@@ -130,6 +130,10 @@ class TaskQueueService:
 
                 result = await run_cross_mount_transfer_task(task)
                 task.result = result
+            elif task.name == "send_email":
+                from services.email import EmailService
+                await EmailService.send_from_task(task.id, task.task_info)
+                task.result = "Email sent"
             else:
                 raise ValueError(f"Unknown task name: {task.name}")