379 lines
14 KiB
Python
379 lines
14 KiB
Python
"""发送邮箱验证邮件（标准库 SMTP）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import smtplib
|
||
import ssl
|
||
from dataclasses import asdict
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.mime.text import MIMEText
|
||
from typing import Any
|
||
|
||
from .email_templates import build_password_reset_content, build_test_email_content, build_verify_email_content
|
||
from .system_settings import SystemSettingsData
|
||
|
||
logger = logging.getLogger("auth.email")
|
||
|
||
SMTP_TIMEOUT_SECONDS = 45
|
||
# 阿里企业邮自定义域名 SMTP（如 smtp.xxiaw.com）常出现 535，官方地址可正常认证
|
||
ALIBABA_SMTP_OFFICIAL = "smtp.mxhichina.com"
|
||
_SSL_CONTEXT = ssl.create_default_context()
|
||
|
||
|
||
def _ssl_context(*, insecure: bool = False) -> ssl.SSLContext:
|
||
if not insecure:
|
||
return _SSL_CONTEXT
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
return ctx
|
||
|
||
|
||
def _connect_ssl(host: str, port: int) -> smtplib.SMTP:
|
||
try:
|
||
return smtplib.SMTP_SSL(
|
||
host,
|
||
port,
|
||
timeout=SMTP_TIMEOUT_SECONDS,
|
||
context=_ssl_context(insecure=False),
|
||
)
|
||
except ssl.SSLCertVerificationError:
|
||
logger.warning(
|
||
"SMTP SSL certificate hostname mismatch for %s, retry without verify",
|
||
host,
|
||
)
|
||
return smtplib.SMTP_SSL(
|
||
host,
|
||
port,
|
||
timeout=SMTP_TIMEOUT_SECONDS,
|
||
context=_ssl_context(insecure=True),
|
||
)
|
||
|
||
|
||
def _starttls(server: smtplib.SMTP, host: str) -> None:
|
||
try:
|
||
server.starttls(context=_ssl_context(insecure=False))
|
||
except ssl.SSLCertVerificationError:
|
||
logger.warning(
|
||
"SMTP STARTTLS certificate hostname mismatch for %s, retry without verify",
|
||
host,
|
||
)
|
||
server.starttls(context=_ssl_context(insecure=True))
|
||
|
||
|
||
def build_verification_link(token: str, settings: SystemSettingsData) -> str:
|
||
return f"{settings.app_url_normalized()}/#/login?verify_token={token}"
|
||
|
||
|
||
def build_password_reset_link(token: str, settings: SystemSettingsData) -> str:
|
||
return f"{settings.app_url_normalized()}/#/login?reset_token={token}"
|
||
|
||
|
||
def resolve_smtp_mode(settings: SystemSettingsData) -> tuple[int, bool, bool]:
|
||
"""返回 (port, use_ssl, use_starttls)。"""
|
||
port = int(settings.smtp_port or 587)
|
||
use_ssl = bool(settings.smtp_use_ssl) or port == 465
|
||
# 465 只能 implicit SSL，绝不能 plain + STARTTLS
|
||
if port == 465:
|
||
use_ssl = True
|
||
use_starttls = False
|
||
elif port == 587:
|
||
use_ssl = False
|
||
use_starttls = bool(settings.smtp_use_tls)
|
||
else:
|
||
use_ssl = bool(settings.smtp_use_ssl)
|
||
use_starttls = bool(settings.smtp_use_tls) and not use_ssl
|
||
return port, use_ssl, use_starttls
|
||
|
||
|
||
def _open_smtp(settings: SystemSettingsData) -> smtplib.SMTP:
|
||
host = settings.smtp_host.strip()
|
||
port, use_ssl, use_starttls = resolve_smtp_mode(settings)
|
||
logger.info(
|
||
"SMTP connect host=%s port=%s ssl=%s starttls=%s user=%s",
|
||
host,
|
||
port,
|
||
use_ssl,
|
||
use_starttls,
|
||
settings.smtp_user or "",
|
||
)
|
||
|
||
if use_ssl:
|
||
server = _connect_ssl(host, port)
|
||
server.ehlo()
|
||
return server
|
||
|
||
server = smtplib.SMTP(host, port, timeout=SMTP_TIMEOUT_SECONDS)
|
||
server.ehlo()
|
||
if use_starttls:
|
||
_starttls(server, host)
|
||
server.ehlo()
|
||
return server
|
||
|
||
|
||
def _close_smtp(server: smtplib.SMTP) -> None:
|
||
try:
|
||
server.quit()
|
||
except Exception:
|
||
try:
|
||
server.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _login_smtp(server: smtplib.SMTP, settings: SystemSettingsData) -> tuple[smtplib.SMTP, bool]:
|
||
"""登录 SMTP，返回 (server, 是否使用了阿里官方地址回退)。"""
|
||
user = (settings.smtp_user or "").strip()
|
||
pwd = settings.smtp_password
|
||
if not user:
|
||
return server, False
|
||
if not pwd:
|
||
raise RuntimeError("未填写 SMTP 密码，请在设置中填写登录密码或客户端安全密码后重试")
|
||
try:
|
||
server.login(user, pwd)
|
||
return server, False
|
||
except smtplib.SMTPAuthenticationError as exc:
|
||
host = (settings.smtp_host or "").strip().lower()
|
||
if host == ALIBABA_SMTP_OFFICIAL:
|
||
raise RuntimeError(_format_login_error(exc, settings)) from exc
|
||
logger.warning(
|
||
"SMTP auth failed on %s (%s), retrying via %s",
|
||
host,
|
||
exc,
|
||
ALIBABA_SMTP_OFFICIAL,
|
||
)
|
||
_close_smtp(server)
|
||
fallback = SystemSettingsData(**{**asdict(settings), "smtp_host": ALIBABA_SMTP_OFFICIAL})
|
||
fallback_server = _open_smtp(fallback)
|
||
try:
|
||
fallback_server.login(user, pwd)
|
||
except Exception as retry_exc:
|
||
_close_smtp(fallback_server)
|
||
raise RuntimeError(_format_login_error(exc, settings)) from retry_exc
|
||
return fallback_server, True
|
||
|
||
|
||
def _format_login_error(exc: Exception, settings: SystemSettingsData) -> str:
|
||
if isinstance(exc, smtplib.SMTPAuthenticationError):
|
||
return _format_smtp_error(exc, settings)
|
||
if isinstance(exc, smtplib.SMTPServerDisconnected):
|
||
return (
|
||
"SMTP 服务器在登录时断开连接，通常不是加密方式问题，而是密码错误或账号未授权 SMTP。"
|
||
f"请确认用户名 {settings.smtp_user or ''} 与密码/客户端安全密码正确，"
|
||
"并在阿里企业邮后台开启「允许使用第三方客户端」。"
|
||
"若使用自定义域名 SMTP（如 smtp.xxiaw.com）仍失败，可改用 smtp.mxhichina.com 后重试。"
|
||
)
|
||
return _format_smtp_error(exc, settings)
|
||
|
||
|
||
def _format_smtp_error(exc: Exception, settings: SystemSettingsData) -> str:
|
||
msg = str(exc).strip() or exc.__class__.__name__
|
||
port, use_ssl, use_starttls = resolve_smtp_mode(settings)
|
||
hints: list[str] = []
|
||
|
||
if isinstance(exc, ssl.SSLCertVerificationError):
|
||
hints.append(
|
||
f"SSL 证书域名与 {settings.smtp_host} 不匹配（阿里企业邮自定义域名常见），"
|
||
"系统已尝试跳过证书校验；若仍失败请改用 smtp.mxhichina.com"
|
||
)
|
||
return f"{msg}。{' '.join(hints)}"
|
||
|
||
if isinstance(exc, smtplib.SMTPAuthenticationError):
|
||
host = (settings.smtp_host or "").strip().lower()
|
||
if host != ALIBABA_SMTP_OFFICIAL and host.startswith("smtp."):
|
||
hints.append(
|
||
f"若使用自定义域名 SMTP（{settings.smtp_host}），阿里企业邮常返回 535；"
|
||
f"请将 SMTP 服务器改为 {ALIBABA_SMTP_OFFICIAL} 后重试"
|
||
)
|
||
hints.append(
|
||
"SMTP 登录失败：请确认登录用户名为完整邮箱地址，"
|
||
"密码为邮箱登录密码或阿里企业邮「客户端安全密码」（非网页登录密码时需在邮箱设置中单独生成）"
|
||
)
|
||
hints.append("请在阿里企业邮管理后台确认已开启「允许使用第三方客户端」")
|
||
return f"{msg}。{' '.join(hints)}"
|
||
|
||
lower = msg.lower()
|
||
if "connection unexpectedly closed" in lower and port == 465 and not use_ssl:
|
||
hints.append("465 端口必须使用 SSL 加密，不能勾选 STARTTLS")
|
||
elif "connection unexpectedly closed" in lower and port == 465:
|
||
if use_ssl:
|
||
hints.append(
|
||
"连接已建立但在后续步骤失败：请重点检查 SMTP 密码/客户端安全密码，"
|
||
"或尝试将服务器改为 smtp.mxhichina.com"
|
||
)
|
||
else:
|
||
hints.append("465 端口必须使用 SSL 加密，请选择「SSL / TLS（端口 465）」")
|
||
if "timed out" in lower or "timeout" in lower or "10060" in lower:
|
||
if port == 587:
|
||
hints.append(
|
||
f"587 端口 STARTTLS 连接 {settings.smtp_host} 超时："
|
||
"该服务器可能仅支持 465 SSL，请将端口改为 465 并选择 SSL 加密"
|
||
)
|
||
else:
|
||
hints.append(
|
||
f"无法在 {SMTP_TIMEOUT_SECONDS}s 内连接 {settings.smtp_host}:{port}，"
|
||
"请检查服务器地址、端口、防火墙/安全组是否放行出站 SMTP"
|
||
)
|
||
if "authentication" in lower or "535" in lower:
|
||
hints.append("认证失败：请检查 SMTP 密码/客户端安全密码是否正确")
|
||
|
||
if hints:
|
||
return f"{msg}。{' '.join(hints)}"
|
||
return msg
|
||
|
||
|
||
def diagnose_smtp(settings: SystemSettingsData) -> dict[str, Any]:
|
||
"""分步检测 SMTP 连接，便于定位配置问题。"""
|
||
host = (settings.smtp_host or "").strip()
|
||
port, use_ssl, use_starttls = resolve_smtp_mode(settings)
|
||
result: dict[str, Any] = {
|
||
"host": host,
|
||
"port": port,
|
||
"mode": "ssl" if use_ssl else ("starttls" if use_starttls else "plain"),
|
||
"steps": [],
|
||
"ok": False,
|
||
}
|
||
if not host:
|
||
result["steps"].append({"step": "config", "ok": False, "message": "未填写 SMTP 服务器"})
|
||
return result
|
||
|
||
server: smtplib.SMTP | None = None
|
||
try:
|
||
server = _open_smtp(settings)
|
||
result["steps"].append({"step": "connect", "ok": True, "message": "连接成功"})
|
||
except Exception as exc:
|
||
result["steps"].append(
|
||
{"step": "connect", "ok": False, "message": _format_smtp_error(exc, settings)}
|
||
)
|
||
return result
|
||
|
||
try:
|
||
if settings.smtp_user:
|
||
if not settings.smtp_password:
|
||
raise RuntimeError("未填写 SMTP 密码，请在设置中填写登录密码或客户端安全密码后重试")
|
||
host_before = (settings.smtp_host or "").strip().lower()
|
||
server, used_fallback = _login_smtp(server, settings)
|
||
login_msg = "登录成功"
|
||
if used_fallback and host_before != ALIBABA_SMTP_OFFICIAL:
|
||
login_msg = (
|
||
f"登录成功（已通过 {ALIBABA_SMTP_OFFICIAL} 认证，"
|
||
f"建议将 SMTP 服务器改为 {ALIBABA_SMTP_OFFICIAL}）"
|
||
)
|
||
result["steps"].append({"step": "login", "ok": True, "message": login_msg})
|
||
else:
|
||
result["steps"].append({"step": "login", "ok": True, "message": "未配置用户名，跳过登录"})
|
||
result["ok"] = True
|
||
except Exception as exc:
|
||
result["steps"].append(
|
||
{"step": "login", "ok": False, "message": _format_login_error(exc, settings)}
|
||
)
|
||
finally:
|
||
if server:
|
||
try:
|
||
server.quit()
|
||
except Exception:
|
||
server.close()
|
||
return result
|
||
|
||
|
||
def _send_sync(
|
||
settings: SystemSettingsData,
|
||
to_email: str,
|
||
subject: str,
|
||
html_body: str,
|
||
text_body: str,
|
||
) -> None:
|
||
if not settings.smtp_configured():
|
||
raise RuntimeError("SMTP 未配置，无法发送邮件")
|
||
|
||
smtp_from = (settings.smtp_from or settings.smtp_user or "").strip()
|
||
msg = MIMEMultipart("alternative")
|
||
msg["Subject"] = subject
|
||
msg["From"] = smtp_from
|
||
msg["To"] = to_email
|
||
msg.attach(MIMEText(text_body, "plain", "utf-8"))
|
||
msg.attach(MIMEText(html_body, "html", "utf-8"))
|
||
|
||
server = _open_smtp(settings)
|
||
try:
|
||
if settings.smtp_user:
|
||
try:
|
||
server, _ = _login_smtp(server, settings)
|
||
except RuntimeError:
|
||
raise
|
||
except Exception as exc:
|
||
raise RuntimeError(_format_login_error(exc, settings)) from exc
|
||
server.sendmail(smtp_from, [to_email], msg.as_string())
|
||
except RuntimeError:
|
||
raise
|
||
except Exception as exc:
|
||
raise RuntimeError(_format_smtp_error(exc, settings)) from exc
|
||
finally:
|
||
try:
|
||
server.quit()
|
||
except Exception:
|
||
server.close()
|
||
|
||
|
||
async def send_password_reset_email(
|
||
to_email: str,
|
||
username: str,
|
||
token: str,
|
||
settings: SystemSettingsData,
|
||
) -> str:
|
||
"""发送密码重置邮件，返回重置链接。"""
|
||
link = build_password_reset_link(token, settings)
|
||
expire_hours = int(settings.email_verify_token_hours or 24)
|
||
subject, text_body, html_body = build_password_reset_content(
|
||
settings, username, to_email, link, expire_hours
|
||
)
|
||
|
||
if settings.smtp_configured():
|
||
await asyncio.to_thread(_send_sync, settings, to_email, subject, html_body, text_body)
|
||
logger.info("Password reset email sent to %s", to_email)
|
||
else:
|
||
logger.warning(
|
||
"SMTP 未配置，密码重置链接: user=%s email=%s link=%s",
|
||
username,
|
||
to_email,
|
||
link,
|
||
)
|
||
return link
|
||
|
||
|
||
async def send_verification_email(
|
||
to_email: str,
|
||
username: str,
|
||
token: str,
|
||
settings: SystemSettingsData,
|
||
) -> str:
|
||
"""发送验证邮件，返回验证链接。"""
|
||
link = build_verification_link(token, settings)
|
||
subject, text_body, html_body = build_verify_email_content(
|
||
settings, username, to_email, link
|
||
)
|
||
|
||
if settings.smtp_configured():
|
||
await asyncio.to_thread(_send_sync, settings, to_email, subject, html_body, text_body)
|
||
logger.info("Verification email sent to %s", to_email)
|
||
else:
|
||
logger.warning(
|
||
"SMTP 未配置，验证链接: user=%s email=%s link=%s",
|
||
username,
|
||
to_email,
|
||
link,
|
||
)
|
||
return link
|
||
|
||
|
||
async def send_test_email(to_email: str, settings: SystemSettingsData) -> None:
|
||
diagnosis = await asyncio.to_thread(diagnose_smtp, settings)
|
||
if not diagnosis.get("ok"):
|
||
failed = next((s for s in diagnosis.get("steps", []) if not s.get("ok")), None)
|
||
raise RuntimeError(failed["message"] if failed else "SMTP 连接检测失败")
|
||
|
||
subject, text_body, html_body = build_test_email_content(settings, to_email)
|
||
await asyncio.to_thread(_send_sync, settings, to_email, subject, html_body, text_body)
|