vmis/backend/app/services/mail_client.py

"""
MailClient — 透過 SSH + docker exec mailserver 管理 docker-mailserver。
domain 在 docker-mailserver 中是隱式的（由 mailbox 決定），
所以 domain_exists 檢查是否有任何 @domain 的 mailbox。
"""
import logging
from typing import Optional
import dns.resolver

from app.core.config import settings

logger = logging.getLogger(__name__)

MAILSERVER_CONTAINER = "mailserver"


class MailClient:

    def _ssh_exec(self, cmd: str) -> tuple[int, str]:
        """SSH 到 10.1.0.254 執行指令，回傳 (exit_code, stdout)"""
        import paramiko
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(
            settings.DOCKER_SSH_HOST,
            username=settings.DOCKER_SSH_USER,
            timeout=15,
        )
        _, stdout, _ = client.exec_command(cmd)
        output = stdout.read().decode().strip()
        exit_code = stdout.channel.recv_exit_status()
        client.close()
        return exit_code, output

    def check_mx_dns(self, domain: str) -> bool:
        """驗證 domain 的 MX record 是否指向正確的 mail server"""
        try:
            answers = dns.resolver.resolve(domain, "MX")
            for rdata in answers:
                if settings.MAIL_MX_HOST in str(rdata.exchange).rstrip("."):
                    return True
            return False
        except Exception as e:
            logger.warning(f"MX DNS check failed for {domain}: {e}")
            return False

    def domain_exists(self, domain: str) -> bool:
        """檢查 mailserver 是否有任何 @domain 的 mailbox（docker-mailserver 的 domain 由 mailbox 決定）"""
        try:
            code, output = self._ssh_exec(
                f"docker exec {MAILSERVER_CONTAINER} setup email list 2>/dev/null | grep -q '@{domain}' && echo yes || echo no"
            )
            return output.strip() == "yes"
        except Exception as e:
            logger.warning(f"domain_exists({domain}) SSH failed: {e}")
            return False

    def create_domain(self, domain: str) -> bool:
        """
        docker-mailserver 不需要顯式建立 domain（mailbox 新增時自動處理）。
        新增一個 postmaster@ 系統帳號來確保 domain 被識別。
        """
        try:
            import secrets
            passwd = secrets.token_urlsafe(16)
            code, output = self._ssh_exec(
                f"docker exec {MAILSERVER_CONTAINER} setup email add postmaster@{domain} {passwd} 2>&1"
            )
            if code == 0:
                logger.info(f"create_domain({domain}): postmaster account created")
                return True
            # 若帳號已存在視為成功
            if "already exists" in output.lower():
                return True
            logger.error(f"create_domain({domain}) failed (exit {code}): {output}")
            return False
        except Exception as e:
            logger.error(f"create_domain({domain}) SSH failed: {e}")
            return False

    def mailbox_exists(self, email: str) -> bool:
        try:
            code, output = self._ssh_exec(
                f"docker exec {MAILSERVER_CONTAINER} setup email list 2>/dev/null | grep -q '{email}' && echo yes || echo no"
            )
            return output.strip() == "yes"
        except Exception as e:
            logger.warning(f"mailbox_exists({email}) SSH failed: {e}")
            return False

    def create_mailbox(self, email: str, password: Optional[str], quota_gb: int = 20) -> bool:
        try:
            import secrets
            passwd = password or secrets.token_urlsafe(16)
            quota_mb = quota_gb * 1024
            code, output = self._ssh_exec(
                f"docker exec {MAILSERVER_CONTAINER} setup email add {email} {passwd} 2>&1"
            )
            if code != 0 and "already exists" not in output.lower():
                logger.error(f"create_mailbox({email}) failed (exit {code}): {output}")
                return False
            # 設定配額
            self._ssh_exec(
                f"docker exec {MAILSERVER_CONTAINER} setup quota set {email} {quota_mb}M 2>/dev/null"
            )
            return True
        except Exception as e:
            logger.error(f"create_mailbox({email}) SSH failed: {e}")
            return False