邮件服务器搭建教程(只收不发)--二

Administrator 4 2025-06-01

最终版:轻量级邮件接收与API系统搭建完整教程

本教程将指导您在一台全新的 Debian 12 服务器上,搭建一个私有的、仅用于接收邮件的系统。该系统能够将收到的邮件存入数据库,并通过两种方式访问:

  1. 一个供自动化脚本调用的 API 接口,可根据收件人和主题筛选邮件并返回其内容。

  2. 一个简单的网页界面,用于人工浏览和查看所有收到的邮件。

一、前期准备 (Prerequisites)

  1. 一台服务器:拥有一个固定的公网 IP 地址。本教程使用 Debian 12 系统和 IP 89.138.1.152

  2. 一个域名:您拥有该域名的 DNS 管理权限。本教程使用 mail.sijuly.nyc.com 作为邮件域名。

  3. SSH 访问权限:能够以 root 用户身份登录服务器。


二、DNS 配置 (DNS Configuration)

登录您的域名 sijuly.nyc.mn 的 DNS 管理后台,添加以下两条记录,将邮件路由指向您的服务器。

  1. A 记录

    • 类型: A

    • 主机/名称: mail

    • 值/指向: 89.158.16.152

    • TTL: 自动

  2. MX 记录

    • 类型: MX

    • 主机/名称: mail

    • 值/服务器: mail.sijuly.nyc.com

    • 优先级: 10

    • TTL: 自动


三、服务器环境配置 (Server Environment Configuration)

⚠️ 重要提示:本教程所有命令均以 root 用户身份执行,全程无需使用 sudo 命令。

  1. 更新系统

    Bash

    apt-get update && apt-get upgrade -y
    
  2. 安装必要的软件包

    Bash

    apt-get install python3-pip python3-venv ufw sqlite3 -y
    
  3. 配置防火墙

    Bash

    # 允许 SSH (保证您的连接不会中断)
    ufw allow ssh
    
    # 允许 SMTP (用于接收邮件)
    ufw allow 25/tcp
    
    # 允许我们的 API 和 Web 界面
    ufw allow 2099/tcp
    
    # 启用防火墙
    ufw enable 
    # 出现提示时,输入 y 并按回车
    

四、部署应用程序 (Deploying the Application)

1. 创建项目结构和虚拟环境

Bash

# 创建并进入项目目录
mkdir -p /opt/mail_api
cd /opt/mail_api

# 创建 Python 虚拟环境
python3 -m venv venv

# 激活虚拟环境
source venv/bin/activate

2. 安装 Python 依赖库

确保您已激活虚拟环境(命令提示符前有 (venv) 字样)。

Bash

pip install aiosmtpd flask gunicorn

3. 创建 app.py (API、数据库和Web界面)

使用 nano /opt/mail_api/app.py 命令创建文件,并将以下全部代码粘贴进去。

Python

import sqlite3
import re
from flask import Flask, request, Response
from email import message_from_bytes
from email.header import decode_header
from markupsafe import escape

# --- 配置 ---
DB_FILE = 'emails.db'
# !!! 强烈建议:为了安全,请将下面的 Token 修改为一个长而复杂的随机字符串
YOUR_API_TOKEN = "2088"

# --- 数据库操作 ---
def get_db_conn():
    conn = sqlite3.connect(DB_FILE, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    conn = get_db_conn()
    c = conn.cursor()
    # 最终的表结构
    c.execute('''
        CREATE TABLE IF NOT EXISTS received_emails (
            id INTEGER PRIMARY KEY AUTOINCREMENT, recipient TEXT, sender TEXT,
            subject TEXT, body TEXT, body_type TEXT, 
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    conn.close()

# --- 核心邮件处理逻辑 ---
def process_email_data(to_address, raw_email_data):
    msg = message_from_bytes(raw_email_data)
    subject_raw, encoding = decode_header(msg['Subject'])[0]
    if isinstance(subject_raw, bytes):
        subject = subject_raw.decode(encoding or 'utf-8', errors='ignore')
    else:
        subject = str(subject_raw)
    sender = msg.get('From')
    body, body_type = "", "text/plain"
    if msg.is_multipart():
        html_part, text_part = None, None
        for part in msg.walk():
            if "text/html" in part.get_content_type(): html_part = part
            elif "text/plain" in part.get_content_type(): text_part = part
        if html_part:
            body = html_part.get_payload(decode=True).decode(html_part.get_content_charset() or 'utf-8', errors='ignore')
            body_type = "text/html"
        elif text_part:
            body = text_part.get_payload(decode=True).decode(text_part.get_content_charset() or 'utf-8', errors='ignore')
            body_type = "text/plain"
    else:
        body = msg.get_payload(decode=True).decode(msg.get_content_charset() or 'utf-8', errors='ignore')
        body_type = msg.get_content_type()
    try:
        conn = get_db_conn()
        cursor = conn.cursor()
        cursor.execute("INSERT INTO received_emails (recipient, sender, subject, body, body_type) VALUES (?, ?, ?, ?, ?)",
                       (to_address, sender, subject, body, body_type))
        conn.commit()
        print(f"邮件已存入数据库: To='{to_address}', Subject='{subject}', Type='{body_type}'")
    finally:
        if conn: conn.close()

# --- Flask 应用 ---
app = Flask(__name__)

# 接口1:给脚本调用的API(已设置为不删除邮件)
@app.route('/Mail', methods=['GET'])
def get_mail_content():
    token = request.args.get('token')
    mail_address_to_find = request.args.get('mail')
    if not token or token != YOUR_API_TOKEN:
        return Response("❌ 无效的 token!", status=401, mimetype="text/plain; charset=utf-8")
    if not mail_address_to_find:
        return Response("❌ 参数错误:请提供 mail 地址。", status=400, mimetype="text/plain; charset=utf-8")
    subject_expected_1, subject_expected_2 = "Verify your email address", "验证您的电子邮件地址"
    try:
        conn = get_db_conn()
        cursor = conn.cursor()
        cursor.execute("SELECT id, subject, body, body_type FROM received_emails WHERE recipient = ? ORDER BY timestamp DESC LIMIT 50", (mail_address_to_find,))
        messages = cursor.fetchall()
        for msg in messages:
            if subject_expected_1.lower() in msg['subject'].lower() or subject_expected_2 in msg['subject']:
                if re.search(r"\b(\d{6})\b", msg['body']):
                    # cursor.execute("DELETE FROM received_emails WHERE id = ?", (msg['id'],)) # 已注释,不执行删除
                    # conn.commit()
                    print(f"成功匹配并返回邮件 (ID: {msg['id']}) for {mail_address_to_find}")
                    return Response(msg['body'], mimetype=f"{msg['body_type']}; charset=utf-8")
        conn.close()
        return Response(f"❌ 未找到 <{mail_address_to_find}> 符合条件的邮件。", status=404, mimetype="text/plain; charset=utf-8")
    finally:
        if 'conn' in locals() and conn: conn.close()

# 接口2:【新增】邮件列表查看页面
@app.route('/view_emails')
def view_emails():
    html = """
    <!DOCTYPE html><html><head><title>收件箱</title>
    <style>body{font-family: sans-serif; margin: 2em;} table{border-collapse: collapse; width: 100%;}
    th, td{border: 1px solid #ddd; padding: 8px; text-align: left;}
    tr:nth-child(even){background-color: #f2f2f2;} th{background-color: #4CAF50; color: white;}</style>
    </head><body><h2>收件箱 (最新 100 封)</h2>
    <table><tr><th>时间</th><th>发件人</th><th>收件人</th><th>主题</th><th>操作</th></tr>
    """
    conn = get_db_conn()
    cursor = conn.cursor()
    cursor.execute("SELECT id, timestamp, sender, recipient, subject FROM received_emails ORDER BY timestamp DESC LIMIT 100")
    emails = cursor.fetchall()
    conn.close()
    for email in emails:
        html += f"<tr><td>{escape(email['timestamp'])}</td><td>{escape(email['sender'])}</td><td>{escape(email['recipient'])}</td><td>{escape(email['subject'])}</td>"
        html += f'<td><a href="/view_email/{email["id"]}" target="_blank">查看</a></td></tr>'
    html += "</table></body></html>"
    return Response(html, mimetype="text/html; charset=utf-8")

# 接口3:【新增】单封邮件详情页面
@app.route('/view_email/<int:email_id>')
def view_email_detail(email_id):
    conn = get_db_conn()
    cursor = conn.cursor()
    cursor.execute("SELECT body, body_type FROM received_emails WHERE id = ?", (email_id,))
    email = cursor.fetchone()
    conn.close()
    if email:
        return Response(email['body'], mimetype=f"{email['body_type']}; charset=utf-8")
    return "邮件未找到", 404

init_db()

4. 创建 smtp_server.py (邮件接收服务)

使用 nano /opt/mail_api/smtp_server.py 命令创建文件,并将以下全部代码粘贴进去。

Python

import asyncio
from aiosmtpd.controller import Controller
from app import process_email_data

class CustomSMTPHandler:
    async def handle_DATA(self, server, session, envelope):
        print(f'收到邮件 from <{envelope.mail_from}> to <{envelope.rcpt_tos}>')
        for recipient in envelope.rcpt_tos:
            # 关键修复:增加了 await
            await asyncio.to_thread(process_email_data, recipient, envelope.content)
        return '250 OK'

if __name__ == '__main__':
    controller = Controller(CustomSMTPHandler(), hostname='0.0.0.0', port=25)
    print("SMTP 服务器正在启动,监听 0.0.0.0:25...")
    controller.start()
    print("SMTP 服务器已启动。按 Ctrl+C 关闭。")
    try:
        asyncio.get_event_loop().run_forever()
    except KeyboardInterrupt:
        print("正在关闭 SMTP 服务器...")
    finally:
        controller.stop()

五、设置为系统服务 (持久化运行)

为了让程序在后台稳定运行并开机自启,我们创建 systemd 服务。

1. 创建 API 服务 (mail-api.service)

使用 nano /etc/systemd/system/mail-api.service 创建文件,并粘贴以下内容:

Ini, TOML

[Unit]
Description=Gunicorn instance to serve the Mail API
After=network.target

[Service]
User=root
Group=www-data
WorkingDirectory=/opt/mail_api
Environment="PATH=/opt/mail_api/venv/bin"
ExecStart=/opt/mail_api/venv/bin/gunicorn --workers 3 --bind 0.0.0.0:2099 app:app
Restart=always

[Install]
WantedBy=multi-user.target

2. 创建 SMTP 服务 (mail-smtp.service)

使用 nano /etc/systemd/system/mail-smtp.service 创建文件,并粘贴以下内容:

Ini, TOML

[Unit]
Description=Custom Python SMTP Server for Mail Receiving
After=network.target

[Service]
User=root
WorkingDirectory=/opt/mail_api
ExecStart=/opt/mail_api/venv/bin/python3 smtp_server.py
AmbientCapabilities=CAP_NET_BIND_SERVICE
Restart=always

[Install]
WantedBy=multi-user.target

3. 启动并验证服务

Bash

# 重新加载配置
systemctl daemon-reload

# 同时重启并设置开机自启
systemctl restart mail-api.service mail-smtp.service
systemctl enable mail-api.service mail-smtp.service

# 分别检查两个服务的状态,确保都是绿色的 active (running)
systemctl status mail-api.service
systemctl status mail-smtp.service

六、使用与测试 (Usage and Testing)

1. 发送测试邮件

从您的个人邮箱向 test@mail.sijuly.nyc.com 发送一封邮件。

  • 主题: Verify your email address

  • 正文: This is a test, my code is 123456

2. 通过网页界面查看邮件

在您的浏览器中打开: http://89.13.126.152:2099/view_emails 您应该能看到刚刚收到的邮件列表。

3. 通过 API 接口提取邮件

在您本地电脑的终端运行:

Bash

curl "http://89.138.116.122:2099/Mail?token=2088&mail=test@mail.sijuly.nyc.com"

您应该能看到返回的邮件正文(HTML 或纯文本)。


七、常见问题与维护 (FAQ and Maintenance)

  • 如何查看实时日志?

    • API 服务日志: journalctl -u mail-api.service -f

    • SMTP 服务日志: journalctl -u mail-smtp.service -f

  • 如何更新代码?

    1. nano /opt/mail_api/app.py (或 smtp_server.py) 编辑文件并保存。

    2. systemctl restart mail-api.service (或 mail-smtp.service) 重启对应的服务。

  • 数据库文件在哪里?

    • 位于 /opt/mail_api/emails.db。您可以用 sqlite3 /opt/mail_api/emails.db 命令进入并进行 SQL 查询。