RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
971 lines
34 KiB
Python
971 lines
34 KiB
Python
import os
|
||
import secrets
|
||
import imaplib
|
||
import smtplib
|
||
import uuid
|
||
import time
|
||
import email
|
||
import re
|
||
import ssl # 新增：用于创建忽略证书的SSL上下文
|
||
import base64
|
||
import json
|
||
from email.header import decode_header, Header
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.mime.base import MIMEBase
|
||
from email import encoders
|
||
from flask import Flask, request, jsonify, Response # 增加 Response
|
||
from flask_cors import CORS
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = os.environ.get('MAIL_FLASK_SECRET_KEY', '') or secrets.token_hex(32)
|
||
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 # 上传上限 100MB
|
||
CORS(app)
|
||
|
||
# ========== 修改点1：改为本地地址 ==========
|
||
IMAP_HOST = '127.0.0.1'
|
||
IMAP_PORT = 993
|
||
SMTP_HOST = '127.0.0.1'
|
||
SMTP_PORT = 465
|
||
|
||
TOKEN_MAP = {}
|
||
TOKEN_EXPIRE = 30 * 24 * 3600 # 30 天；滑动过期：每次使用自动续期，闲置超 30 天才失效
|
||
|
||
# ========== 新增：注册配置 ==========
|
||
REGISTER_AUTH_CODE = os.environ.get('MAIL_REGISTER_AUTH_CODE', '') # 注册认证码（由环境注入）
|
||
HMAIL_ADMIN_PASSWORD = os.environ.get('HMAIL_ADMIN_PASSWORD', '') # hMailServer 管理员密码（由环境注入）
|
||
|
||
# ========== 新增：Autodiscover 配置 ==========
|
||
AUTODISCOVER_DOMAINS = ['rrrosmontis.icu', 'aeolianchi.me'] # 允许 Autodiscover 的邮箱域名列表
|
||
MAIL_SERVER = 'mail.rrrosmontis.icu' # 真实的邮件服务器地址（公网可解析）
|
||
IMAP_SSL_PORT = 993
|
||
POP_SSL_PORT = 995
|
||
SMTP_SSL_PORT = 465
|
||
|
||
def decode_mime_str(s):
|
||
"""通用解码邮件头字段（如 Subject, From 等）"""
|
||
if not s:
|
||
return ''
|
||
parts = decode_header(s)
|
||
decoded_parts = []
|
||
for part, charset in parts:
|
||
if isinstance(part, bytes):
|
||
try:
|
||
if charset:
|
||
decoded_parts.append(part.decode(charset, errors='replace'))
|
||
else:
|
||
decoded_parts.append(part.decode('utf-8', errors='replace'))
|
||
except Exception:
|
||
decoded_parts.append(part.decode('utf-8', errors='replace'))
|
||
else:
|
||
decoded_parts.append(part)
|
||
return ''.join(decoded_parts)
|
||
|
||
def decode_attachment_names(raw):
|
||
"""
|
||
解码 X-Attachments 头部中的附件名列表。
|
||
原始字符串可能包含多个 MIME 编码片段，用逗号分隔。
|
||
例如： "=?utf-8?b?5Lit5paH?=,=?utf-8?b?dGVzdC50eHQ=?="
|
||
返回解码后的普通文件名列表。
|
||
"""
|
||
if not raw:
|
||
return []
|
||
parts = decode_header(raw)
|
||
decoded_parts = []
|
||
for part, charset in parts:
|
||
if isinstance(part, bytes):
|
||
try:
|
||
if charset:
|
||
decoded_parts.append(part.decode(charset, errors='replace'))
|
||
else:
|
||
decoded_parts.append(part.decode('utf-8', errors='replace'))
|
||
except Exception:
|
||
decoded_parts.append(part.decode('utf-8', errors='replace'))
|
||
else:
|
||
decoded_parts.append(part)
|
||
combined = ''.join(decoded_parts)
|
||
return [name.strip() for name in combined.split(',') if name.strip()]
|
||
|
||
def generate_token(email, password):
|
||
token = uuid.uuid4().hex
|
||
TOKEN_MAP[token] = {'email': email, 'password': password, 'created': time.time()}
|
||
return token
|
||
|
||
def validate_token(token):
|
||
data = TOKEN_MAP.get(token)
|
||
if not data:
|
||
return None
|
||
if time.time() - data['created'] > TOKEN_EXPIRE:
|
||
del TOKEN_MAP[token]
|
||
return None
|
||
# 滑动续期：一旦使用即自动续 30 天
|
||
data['created'] = time.time()
|
||
return data['email'], data['password']
|
||
|
||
# ========== 修改点2：忽略SSL证书验证 ==========
|
||
def imap_login(email, password):
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email, password)
|
||
return conn
|
||
except Exception as e:
|
||
print(f"IMAP login error: {e}")
|
||
return None
|
||
|
||
# ========== 修改点3：SMTP同样忽略证书 ==========
|
||
def smtp_login(email, password):
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
smtp = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context)
|
||
smtp.ehlo()
|
||
smtp.login(email, password)
|
||
return smtp
|
||
except Exception as e:
|
||
print(f"SMTP login error: {e}")
|
||
return None
|
||
|
||
def extract_email_body(msg):
|
||
body = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/html":
|
||
payload = part.get_payload(decode=True)
|
||
charset = part.get_content_charset() or 'utf-8'
|
||
try:
|
||
body = payload.decode(charset, errors='replace')
|
||
except:
|
||
body = payload.decode('utf-8', errors='replace')
|
||
return body
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
charset = part.get_content_charset() or 'utf-8'
|
||
try:
|
||
body = payload.decode(charset, errors='replace')
|
||
except:
|
||
body = payload.decode('utf-8', errors='replace')
|
||
return body
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
charset = msg.get_content_charset() or 'utf-8'
|
||
try:
|
||
body = payload.decode(charset, errors='replace')
|
||
except:
|
||
body = payload.decode('utf-8', errors='replace')
|
||
return body
|
||
|
||
|
||
def build_cid_map(msg):
|
||
"""遍历 MIME，构建 {Content-ID: data URL}，用于内嵌图片(cid:)的渲染。"""
|
||
cid_map = {}
|
||
for part in msg.walk():
|
||
cid = part.get("Content-ID")
|
||
if not cid:
|
||
continue
|
||
cid = cid.strip().strip("<>").strip()
|
||
if cid.lower().startswith("cid:"):
|
||
cid = cid[4:].strip()
|
||
if not cid:
|
||
continue
|
||
payload = part.get_payload(decode=True)
|
||
if not payload:
|
||
continue
|
||
ctype = part.get_content_type() or "application/octet-stream"
|
||
b64 = base64.b64encode(payload).decode("ascii")
|
||
cid_map[cid] = f"data:{ctype};base64,{b64}"
|
||
return cid_map
|
||
|
||
|
||
def replace_cid_with_data(body, cid_map):
|
||
"""把 HTML 正文里的 cid:xxx 替换为 data URL。"""
|
||
def _repl(m):
|
||
return cid_map.get(m.group(1), m.group(0))
|
||
return re.sub(r"cid:([A-Za-z0-9._\-@]+)", _repl, body)
|
||
|
||
|
||
# ========== 新增：Autodiscover 辅助函数 ==========
|
||
def extract_email_from_autodiscover_request(request_data: str) -> str | None:
|
||
"""从 Autodiscover 请求的 XML 中提取邮箱地址"""
|
||
match = re.search(r'<EMailAddress>(.*?)</EMailAddress>', request_data, re.IGNORECASE)
|
||
if match:
|
||
return match.group(1).strip()
|
||
return None
|
||
|
||
def generate_autodiscover_xml(email: str) -> str:
|
||
"""生成符合 Microsoft 规范的 Autodiscover 响应 XML"""
|
||
username = email
|
||
display_name = email.split('@')[0]
|
||
return f'''<?xml version="1.0" encoding="utf-8"?>
|
||
<Autodiscover xmlns="http://schemas.microsoft.com/exchange/autodiscover/responseschema/2006">
|
||
<Response xmlns="http://schemas.microsoft.com/exchange/autodiscover/outlook/responseschema/2006a">
|
||
<User>
|
||
<DisplayName>{display_name}</DisplayName>
|
||
</User>
|
||
<Account>
|
||
<AccountType>email</AccountType>
|
||
<Action>settings</Action>
|
||
<Protocol>
|
||
<Type>IMAP</Type>
|
||
<Server>{MAIL_SERVER}</Server>
|
||
<Port>{IMAP_SSL_PORT}</Port>
|
||
<LoginName>{username}</LoginName>
|
||
<SSL>on</SSL>
|
||
<Authentication>password</Authentication>
|
||
</Protocol>
|
||
<Protocol>
|
||
<Type>POP3</Type>
|
||
<Server>{MAIL_SERVER}</Server>
|
||
<Port>{POP_SSL_PORT}</Port>
|
||
<LoginName>{username}</LoginName>
|
||
<SSL>on</SSL>
|
||
<Authentication>password</Authentication>
|
||
</Protocol>
|
||
<Protocol>
|
||
<Type>SMTP</Type>
|
||
<Server>{MAIL_SERVER}</Server>
|
||
<Port>{SMTP_SSL_PORT}</Port>
|
||
<LoginName>{username}</LoginName>
|
||
<SSL>on</SSL>
|
||
<Authentication>password</Authentication>
|
||
<UsePOPAuth>off</UsePOPAuth>
|
||
<UseSPA>off</UseSPA>
|
||
</Protocol>
|
||
</Account>
|
||
</Response>
|
||
</Autodiscover>'''
|
||
|
||
# ========== 新增：Autodiscover 端点 ==========
|
||
@app.route('/autodiscover/autodiscover.xml', methods=['GET', 'POST'])
|
||
def autodiscover():
|
||
"""Microsoft Outlook / Exchange 自动发现端点"""
|
||
email = None
|
||
if request.method == 'GET':
|
||
email = request.args.get('Email')
|
||
else: # POST
|
||
body = request.get_data(as_text=True)
|
||
email = extract_email_from_autodiscover_request(body)
|
||
|
||
if not email:
|
||
return Response("Missing or invalid email address", status=400, mimetype='text/plain')
|
||
|
||
# 仅允许自己的域名
|
||
if not any(email.lower().endswith('@' + d) for d in AUTODISCOVER_DOMAINS):
|
||
return Response("Domain not allowed", status=403, mimetype='text/plain')
|
||
|
||
xml_response = generate_autodiscover_xml(email)
|
||
return Response(xml_response, mimetype='text/xml', status=200)
|
||
|
||
# ========== 原有路由 ==========
|
||
@app.route('/')
|
||
def index():
|
||
return jsonify({
|
||
'name': 'Royirin Rosmontis Mail API',
|
||
'endpoints': [
|
||
'/api/login [POST]',
|
||
'/api/register [POST]',
|
||
'/api/delete_account [POST]',
|
||
'/api/mailboxes [GET]',
|
||
'/api/messages [GET]',
|
||
'/api/message/<uid> [GET]',
|
||
'/api/send [POST]',
|
||
'/api/logout [POST]',
|
||
'/api/sent [GET]',
|
||
'/api/delete/<uid> [POST]'
|
||
]
|
||
})
|
||
|
||
@app.route('/v1/mail/api/login', methods=['POST'])
|
||
def api_login():
|
||
data = request.get_json()
|
||
email = data.get('email', '').strip()
|
||
password = data.get('password', '')
|
||
if not email or not password:
|
||
return jsonify({'error': '邮箱和密码不能为空'}), 400
|
||
imap = imap_login(email, password)
|
||
if imap:
|
||
imap.logout()
|
||
token = generate_token(email, password)
|
||
return jsonify({'token': token, 'email': email})
|
||
else:
|
||
return jsonify({'error': '邮箱或密码错误'}), 401
|
||
|
||
|
||
# ========== ROSMANKA ID SSO 集成 ==========
|
||
SSO_API_BASE = 'http://127.0.0.1:4550/v1/sso'
|
||
SSO_SECRETS_PATH = 'C:/call/backends/sso/secrets.json'
|
||
|
||
def _sso_service_token():
|
||
"""读取本机 SSO 的服务共享密钥（单一密钥源，不落代码）。"""
|
||
try:
|
||
with open(SSO_SECRETS_PATH, 'r', encoding='utf-8') as f:
|
||
return json.load(f).get('service_tokens', {}).get('mail', '')
|
||
except Exception:
|
||
return ''
|
||
|
||
@app.route('/v1/mail/api/sso_exchange', methods=['POST'])
|
||
def api_sso_exchange():
|
||
"""SSO 会话 token → mail token。
|
||
流程：SSO 验会话 + 检查 mail 服务 tier>=1 → 取回托管的邮箱凭据 → IMAP 复核 → 发 mail token。
|
||
"""
|
||
data = request.get_json(silent=True) or {}
|
||
sso_token = (data.get('sso_token') or '').strip()
|
||
if not sso_token:
|
||
return jsonify({'error': '缺少 sso_token'}), 400
|
||
try:
|
||
import requests as _rq
|
||
r = _rq.post(SSO_API_BASE + '/service/exchange',
|
||
json={'service': 'mail', 'token': sso_token},
|
||
headers={'X-SSO-Service-Token': _sso_service_token()}, timeout=10)
|
||
try:
|
||
d = r.json()
|
||
except Exception:
|
||
d = {}
|
||
except Exception:
|
||
return jsonify({'error': 'SSO 服务不可达'}), 502
|
||
if r.status_code != 200 or not d.get('ok'):
|
||
return jsonify({'error': d.get('error') or 'SSO 认证失败'}), (r.status_code if r.status_code != 200 else 401)
|
||
email_addr = d['external_id']
|
||
password = d['external_secret']
|
||
imap = imap_login(email_addr, password)
|
||
if not imap:
|
||
return jsonify({'error': '邮箱凭据校验失败，请联系管理员'}), 502
|
||
imap.logout()
|
||
token = generate_token(email_addr, password)
|
||
return jsonify({'token': token, 'email': email_addr, 'sso': True})
|
||
|
||
|
||
def create_hmail_account(email, password):
|
||
"""通过 hMailServer COM API 创建邮箱账户"""
|
||
try:
|
||
import win32com.client
|
||
import pythoncom
|
||
except ImportError:
|
||
return False, '服务器缺少 pywin32，无法创建账户'
|
||
|
||
if not HMAIL_ADMIN_PASSWORD or HMAIL_ADMIN_PASSWORD == 'CHANGE_ME':
|
||
return False, 'HMAIL_ADMIN_PASSWORD 未配置，请联系管理员'
|
||
|
||
pythoncom.CoInitialize()
|
||
try:
|
||
app = win32com.client.Dispatch("hMailServer.Application")
|
||
if not app.Authenticate("Administrator", HMAIL_ADMIN_PASSWORD):
|
||
return False, 'hMailServer 管理员认证失败'
|
||
|
||
domain_name = email.split('@')[1]
|
||
try:
|
||
domain = app.Domains.ItemByName(domain_name)
|
||
except Exception:
|
||
return False, f'域名 {domain_name} 不存在'
|
||
|
||
try:
|
||
existing = domain.Accounts.ItemByAddress(email)
|
||
if existing:
|
||
return False, '该邮箱已存在'
|
||
except Exception:
|
||
pass
|
||
|
||
account = domain.Accounts.Add()
|
||
account.Address = email
|
||
account.Password = password
|
||
account.Active = True
|
||
account.Save()
|
||
return True, None
|
||
except Exception as e:
|
||
return False, f'创建账户失败: {str(e)}'
|
||
finally:
|
||
pythoncom.CoUninitialize()
|
||
|
||
def delete_hmail_account(email):
|
||
"""通过 hMailServer COM API 删除邮箱账户"""
|
||
try:
|
||
import win32com.client
|
||
import pythoncom
|
||
except ImportError:
|
||
return False, '服务器缺少 pywin32，无法删除账户'
|
||
|
||
if not HMAIL_ADMIN_PASSWORD or HMAIL_ADMIN_PASSWORD == 'CHANGE_ME':
|
||
return False, 'HMAIL_ADMIN_PASSWORD 未配置，请联系管理员'
|
||
|
||
pythoncom.CoInitialize()
|
||
try:
|
||
app = win32com.client.Dispatch("hMailServer.Application")
|
||
if not app.Authenticate("Administrator", HMAIL_ADMIN_PASSWORD):
|
||
return False, 'hMailServer 管理员认证失败'
|
||
|
||
domain_name = email.split('@')[1]
|
||
try:
|
||
domain = app.Domains.ItemByName(domain_name)
|
||
except Exception:
|
||
return False, f'域名 {domain_name} 不存在'
|
||
|
||
try:
|
||
account = domain.Accounts.ItemByAddress(email)
|
||
except Exception:
|
||
return False, '该邮箱账户不存在'
|
||
|
||
account.Delete()
|
||
return True, None
|
||
except Exception as e:
|
||
return False, f'删除账户失败: {str(e)}'
|
||
finally:
|
||
pythoncom.CoUninitialize()
|
||
|
||
@app.route('/v1/mail/api/register', methods=['POST'])
|
||
def api_register():
|
||
# 注册已收归 ROSMANKA ID SSO（邀请码制），此端点停用
|
||
return jsonify({'error': '注册已迁移至统一认证中心', 'sso': 'https://sso.rrrosmontis.icu/register.html'}), 410
|
||
|
||
@app.route('/v1/mail/api/delete_account', methods=['POST'])
|
||
def api_delete_account():
|
||
# 注销已收归 ROSMANKA ID SSO（一处注销处处死完），此端点停用
|
||
return jsonify({'error': '注销已迁移至统一认证中心（账户中心 → 注销账户）',
|
||
'sso': 'https://sso.rrrosmontis.icu/profile/'}), 410
|
||
|
||
def get_hmail_forward(email):
|
||
"""通过 hMailServer COM API 获取账户的邮件转发配置"""
|
||
try:
|
||
import win32com.client
|
||
import pythoncom
|
||
except ImportError:
|
||
return None, '服务器缺少 pywin32，无法获取转发配置'
|
||
|
||
if not HMAIL_ADMIN_PASSWORD or HMAIL_ADMIN_PASSWORD == 'CHANGE_ME':
|
||
return None, 'HMAIL_ADMIN_PASSWORD 未配置，请联系管理员'
|
||
|
||
pythoncom.CoInitialize()
|
||
try:
|
||
app = win32com.client.Dispatch("hMailServer.Application")
|
||
if not app.Authenticate("Administrator", HMAIL_ADMIN_PASSWORD):
|
||
return None, 'hMailServer 管理员认证失败'
|
||
|
||
domain_name = email.split('@')[1]
|
||
try:
|
||
domain = app.Domains.ItemByName(domain_name)
|
||
except Exception:
|
||
return None, f'域名 {domain_name} 不存在'
|
||
|
||
try:
|
||
account = domain.Accounts.ItemByAddress(email)
|
||
except Exception:
|
||
return None, '该邮箱账户不存在'
|
||
|
||
return {'enabled': bool(account.ForwardEnabled), 'address': account.ForwardAddress or '',
|
||
'keep_original': bool(account.ForwardKeepOriginal)}, None
|
||
except Exception as e:
|
||
return None, f'获取转发配置失败: {str(e)}'
|
||
finally:
|
||
pythoncom.CoUninitialize()
|
||
|
||
def set_hmail_forward(email, enabled, address):
|
||
"""通过 hMailServer COM API 设置账户的邮件转发"""
|
||
try:
|
||
import win32com.client
|
||
import pythoncom
|
||
except ImportError:
|
||
return False, '服务器缺少 pywin32，无法设置转发配置'
|
||
|
||
if not HMAIL_ADMIN_PASSWORD or HMAIL_ADMIN_PASSWORD == 'CHANGE_ME':
|
||
return False, 'HMAIL_ADMIN_PASSWORD 未配置，请联系管理员'
|
||
|
||
pythoncom.CoInitialize()
|
||
try:
|
||
app = win32com.client.Dispatch("hMailServer.Application")
|
||
if not app.Authenticate("Administrator", HMAIL_ADMIN_PASSWORD):
|
||
return False, 'hMailServer 管理员认证失败'
|
||
|
||
domain_name = email.split('@')[1]
|
||
try:
|
||
domain = app.Domains.ItemByName(domain_name)
|
||
except Exception:
|
||
return False, f'域名 {domain_name} 不存在'
|
||
|
||
try:
|
||
account = domain.Accounts.ItemByAddress(email)
|
||
except Exception:
|
||
return False, '该邮箱账户不存在'
|
||
|
||
account.ForwardEnabled = bool(enabled)
|
||
account.ForwardKeepOriginal = True # 转发同时保留本地原件
|
||
if enabled and address:
|
||
account.ForwardAddress = address
|
||
account.Save()
|
||
return True, None
|
||
except Exception as e:
|
||
return False, f'设置转发配置失败: {str(e)}'
|
||
finally:
|
||
pythoncom.CoUninitialize()
|
||
|
||
@app.route('/v1/mail/api/forward', methods=['GET', 'POST'])
|
||
def api_forward():
|
||
"""查询/设置账户邮件转发"""
|
||
if request.method == 'GET':
|
||
token = request.args.get('token', '')
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, _ = creds
|
||
|
||
data, error = get_hmail_forward(email_addr)
|
||
if error:
|
||
return jsonify({'error': error}), 500
|
||
return jsonify(data)
|
||
|
||
data = request.get_json(silent=True) or {}
|
||
token = data.get('token', '')
|
||
enabled = bool(data.get('enabled', False))
|
||
address = (data.get('address') or '').strip()
|
||
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, _ = creds
|
||
|
||
# 地址暂允许为空，后续补齐转发地址后再开启必填校验
|
||
success, error = set_hmail_forward(email_addr, enabled, address)
|
||
if not success:
|
||
return jsonify({'error': error}), 500
|
||
return jsonify({'success': True})
|
||
|
||
@app.route('/v1/mail/api/change_password', methods=['POST'])
|
||
def api_change_password():
|
||
# 改密已收归 ROSMANKA ID SSO（一处更改处处更改），此端点停用
|
||
return jsonify({'error': '修改密码已迁移至统一认证中心（账户中心 → 安全）',
|
||
'sso': 'https://sso.rrrosmontis.icu/profile/'}), 410
|
||
|
||
@app.route('/v1/mail/api/mailboxes', methods=['GET'])
|
||
def api_mailboxes():
|
||
token = request.args.get('token')
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
if not validate_token(token):
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
return jsonify({'mailboxes': ['INBOX']})
|
||
|
||
@app.route('/v1/mail/api/messages', methods=['GET'])
|
||
def api_messages():
|
||
token = request.args.get('token')
|
||
folder = request.args.get('folder', 'INBOX')
|
||
page = int(request.args.get('page', 1))
|
||
per_page = int(request.args.get('per_page', 20))
|
||
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, password = creds
|
||
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email_addr, password)
|
||
conn.select(folder)
|
||
|
||
result, data = conn.uid('search', None, "ALL")
|
||
if result != 'OK':
|
||
return jsonify({'messages': [], 'has_more': False})
|
||
|
||
uid_list = data[0].split()
|
||
total = len(uid_list)
|
||
|
||
start = max(0, total - page * per_page)
|
||
end = total - (page - 1) * per_page
|
||
if start < 0:
|
||
start = 0
|
||
page_uids = uid_list[start:end] if start < end else []
|
||
|
||
messages = []
|
||
for uid in page_uids:
|
||
result, msg_data = conn.uid('fetch', uid, '(BODY.PEEK[HEADER])')
|
||
if result != 'OK':
|
||
continue
|
||
raw_header = msg_data[0][1]
|
||
msg = email.message_from_bytes(raw_header)
|
||
from_ = decode_mime_str(msg.get('From', '未知'))
|
||
subject = decode_mime_str(msg.get('Subject', '无主题'))
|
||
date = msg.get('Date', '')
|
||
|
||
attachments_raw = msg.get('X-Attachments', '')
|
||
attachments = decode_attachment_names(attachments_raw)
|
||
|
||
messages.append({
|
||
'uid': uid.decode() if isinstance(uid, bytes) else uid,
|
||
'from': from_,
|
||
'subject': subject,
|
||
'date': date,
|
||
'attachments': attachments
|
||
})
|
||
|
||
conn.close()
|
||
conn.logout()
|
||
|
||
messages.reverse()
|
||
return jsonify({
|
||
'messages': messages,
|
||
'has_more': page * per_page < total
|
||
})
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
|
||
@app.route('/v1/mail/api/sent', methods=['GET'])
|
||
def api_sent():
|
||
token = request.args.get('token')
|
||
page = int(request.args.get('page', 1))
|
||
per_page = int(request.args.get('per_page', 20))
|
||
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, password = creds
|
||
|
||
conn = None
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email_addr, password)
|
||
|
||
typ, folder_data = conn.list()
|
||
if typ != 'OK':
|
||
return jsonify({'error': '无法获取文件夹列表'}), 500
|
||
|
||
sent_candidates = ['Sent', 'Sent Messages', 'Sent Items', '已发送', '已发送邮件']
|
||
sent_folder = None
|
||
for line in folder_data:
|
||
line_str = line.decode() if isinstance(line, bytes) else line
|
||
folder_name = None
|
||
match = re.search(r'"([^"]+)"$', line_str)
|
||
if match:
|
||
folder_name = match.group(1)
|
||
else:
|
||
parts = line_str.split()
|
||
if parts:
|
||
folder_name = parts[-1].strip('"')
|
||
if folder_name:
|
||
if any(folder_name.lower() == cand.lower() for cand in sent_candidates):
|
||
sent_folder = folder_name
|
||
break
|
||
if 'sent' in folder_name.lower() or '已发送' in folder_name:
|
||
sent_folder = folder_name
|
||
break
|
||
|
||
if not sent_folder:
|
||
sent_folder = 'Sent'
|
||
|
||
try:
|
||
conn.select(sent_folder)
|
||
except Exception:
|
||
return jsonify({'messages': [], 'has_more': False})
|
||
|
||
result, data = conn.uid('search', None, "ALL")
|
||
if result != 'OK':
|
||
return jsonify({'messages': [], 'has_more': False})
|
||
|
||
uid_list = data[0].split()
|
||
total = len(uid_list)
|
||
|
||
start = max(0, total - page * per_page)
|
||
end = total - (page - 1) * per_page
|
||
if start < 0:
|
||
start = 0
|
||
page_uids = uid_list[start:end] if start < end else []
|
||
|
||
messages = []
|
||
for uid in page_uids:
|
||
result, msg_data = conn.uid('fetch', uid, '(BODY.PEEK[HEADER])')
|
||
if result != 'OK':
|
||
continue
|
||
raw_header = msg_data[0][1]
|
||
msg = email.message_from_bytes(raw_header)
|
||
from_ = decode_mime_str(msg.get('From', '未知'))
|
||
to = decode_mime_str(msg.get('To', '未知'))
|
||
subject = decode_mime_str(msg.get('Subject', '无主题'))
|
||
date = msg.get('Date', '')
|
||
|
||
attachments_raw = msg.get('X-Attachments', '')
|
||
attachments = decode_attachment_names(attachments_raw)
|
||
|
||
messages.append({
|
||
'uid': uid.decode() if isinstance(uid, bytes) else uid,
|
||
'from': from_,
|
||
'to': to,
|
||
'subject': subject,
|
||
'date': date,
|
||
'attachments': attachments
|
||
})
|
||
|
||
messages.reverse()
|
||
return jsonify({
|
||
'messages': messages,
|
||
'has_more': page * per_page < total
|
||
})
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
finally:
|
||
if conn:
|
||
try:
|
||
conn.close()
|
||
except:
|
||
pass
|
||
conn.logout()
|
||
|
||
@app.route('/v1/mail/api/message/<uid>', methods=['GET'])
|
||
def api_message(uid):
|
||
token = request.args.get('token')
|
||
folder = request.args.get('folder', 'INBOX')
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, password = creds
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email_addr, password)
|
||
conn.select(folder)
|
||
result, data = conn.uid('fetch', uid, '(RFC822)')
|
||
if result != 'OK':
|
||
return jsonify({'error': '邮件不存在'}), 404
|
||
raw_email = data[0][1]
|
||
msg = email.message_from_bytes(raw_email)
|
||
body = extract_email_body(msg)
|
||
cid_map = build_cid_map(msg)
|
||
if cid_map:
|
||
body = replace_cid_with_data(body, cid_map)
|
||
subject = decode_mime_str(msg.get('Subject', ''))
|
||
from_ = decode_mime_str(msg.get('From', ''))
|
||
to = decode_mime_str(msg.get('To', ''))
|
||
date = msg.get('Date', '')
|
||
|
||
attachments_raw = msg.get('X-Attachments', '')
|
||
attachments = decode_attachment_names(attachments_raw)
|
||
|
||
conn.close()
|
||
conn.logout()
|
||
return jsonify({
|
||
'body': body,
|
||
'subject': subject,
|
||
'from': from_,
|
||
'to': to,
|
||
'date': date,
|
||
'attachments': attachments
|
||
})
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
|
||
@app.route('/v1/mail/api/attachment/<uid>', methods=['GET'])
|
||
def api_attachment(uid):
|
||
token = request.args.get('token')
|
||
folder = request.args.get('folder', 'INBOX')
|
||
filename_b64 = request.args.get('filename_b64', '')
|
||
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
if not filename_b64:
|
||
return jsonify({'error': '缺少filename_b64'}), 400
|
||
|
||
try:
|
||
filename = base64.b64decode(filename_b64).decode('utf-8')
|
||
except Exception:
|
||
return jsonify({'error': 'filename_b64解码失败'}), 400
|
||
|
||
email_addr, password = creds
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email_addr, password)
|
||
conn.select(folder)
|
||
|
||
result, data = conn.uid('fetch', uid, '(RFC822)')
|
||
if result != 'OK':
|
||
return jsonify({'error': '邮件不存在'}), 404
|
||
|
||
raw_email = data[0][1]
|
||
msg = email.message_from_bytes(raw_email)
|
||
|
||
for part in msg.walk():
|
||
disp = part.get('Content-Disposition', '')
|
||
if not disp or 'attachment' not in disp:
|
||
continue
|
||
|
||
raw_name = part.get_filename()
|
||
if not raw_name:
|
||
continue
|
||
|
||
decoded_name = decode_mime_str(raw_name)
|
||
if decoded_name == filename:
|
||
payload = part.get_payload(decode=True)
|
||
if payload is None:
|
||
continue
|
||
|
||
content_type = part.get_content_type() or 'application/octet-stream'
|
||
# 对中文文件名按 UTF-8 进行 RFC 5987 编码，确保下载名不乱码
|
||
from urllib.parse import quote
|
||
ascii_name = quote(filename, safe='')
|
||
headers = {
|
||
'Content-Disposition': f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{ascii_name}",
|
||
'Content-Type': content_type
|
||
}
|
||
conn.close()
|
||
conn.logout()
|
||
return Response(payload, headers=headers)
|
||
|
||
conn.close()
|
||
conn.logout()
|
||
return jsonify({'error': '附件不存在'}), 404
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
|
||
@app.route('/v1/mail/api/delete/<uid>', methods=['POST'])
|
||
def api_delete(uid):
|
||
data = request.get_json()
|
||
token = data.get('token')
|
||
folder = data.get('folder', 'INBOX')
|
||
if not token:
|
||
return jsonify({'error': '缺少token'}), 401
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, password = creds
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
conn = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
||
conn.login(email_addr, password)
|
||
conn.select(folder)
|
||
result, _ = conn.uid('store', uid, '+FLAGS', '\\Deleted')
|
||
if result != 'OK':
|
||
return jsonify({'error': '删除失败'}), 500
|
||
conn.expunge()
|
||
conn.close()
|
||
conn.logout()
|
||
return jsonify({'success': True})
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
|
||
@app.route('/v1/mail/api/send', methods=['POST'])
|
||
def api_send():
|
||
if 'token' not in request.form:
|
||
return jsonify({'error': '缺少token'}), 400
|
||
token = request.form['token']
|
||
to_addr = request.form.get('to')
|
||
subject = request.form.get('subject')
|
||
body = request.form.get('body')
|
||
files = request.files.getlist('attachments')
|
||
|
||
if not all([token, to_addr, subject]):
|
||
return jsonify({'error': '参数不全'}), 400
|
||
|
||
body = body or ''
|
||
|
||
creds = validate_token(token)
|
||
if not creds:
|
||
return jsonify({'error': 'token无效或已过期'}), 401
|
||
email_addr, password = creds
|
||
|
||
try:
|
||
context = ssl._create_unverified_context()
|
||
smtp = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context)
|
||
smtp.ehlo()
|
||
smtp.login(email_addr, password)
|
||
|
||
msg = MIMEMultipart()
|
||
msg['From'] = email_addr
|
||
msg['To'] = to_addr
|
||
msg['Subject'] = subject
|
||
msg.attach(MIMEText(body, 'html', 'utf-8'))
|
||
|
||
attachment_names = []
|
||
for file in files:
|
||
if file.filename:
|
||
part = MIMEBase('application', 'octet-stream')
|
||
part.set_payload(file.read())
|
||
encoders.encode_base64(part)
|
||
|
||
part.add_header('Content-Disposition', 'attachment', filename=file.filename)
|
||
msg.attach(part)
|
||
attachment_names.append(file.filename)
|
||
|
||
if attachment_names:
|
||
encoded_names = [str(Header(name, 'utf-8')) for name in attachment_names]
|
||
msg['X-Attachments'] = ','.join(encoded_names)
|
||
|
||
smtp.send_message(msg)
|
||
smtp.quit()
|
||
|
||
# 保存到已发送文件夹
|
||
try:
|
||
imap_context = ssl._create_unverified_context()
|
||
imap = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=imap_context)
|
||
imap.login(email_addr, password)
|
||
|
||
typ, folder_data = imap.list()
|
||
if typ == 'OK':
|
||
sent_candidates = ['Sent', 'Sent Messages', 'Sent Items', '已发送', '已发送邮件']
|
||
sent_folder = None
|
||
for line in folder_data:
|
||
line_str = line.decode() if isinstance(line, bytes) else line
|
||
folder_name = None
|
||
match = re.search(r'"([^"]+)"$', line_str)
|
||
if match:
|
||
folder_name = match.group(1)
|
||
else:
|
||
parts = line_str.split()
|
||
if parts:
|
||
folder_name = parts[-1].strip('"')
|
||
|
||
if folder_name:
|
||
if any(folder_name.lower() == cand.lower() for cand in sent_candidates):
|
||
sent_folder = folder_name
|
||
break
|
||
if 'sent' in folder_name.lower() or '已发送' in folder_name:
|
||
sent_folder = folder_name
|
||
break
|
||
|
||
if not sent_folder:
|
||
sent_folder = 'Sent'
|
||
|
||
try:
|
||
imap.select(sent_folder)
|
||
except Exception:
|
||
try:
|
||
imap.create(sent_folder)
|
||
imap.select(sent_folder)
|
||
except Exception:
|
||
imap.close()
|
||
imap.logout()
|
||
return jsonify({'success': True})
|
||
|
||
msg_bytes = msg.as_bytes()
|
||
imap.append(sent_folder, '\\Seen', imaplib.Time2Internaldate(time.time()), msg_bytes)
|
||
imap.close()
|
||
imap.logout()
|
||
else:
|
||
imap.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
return jsonify({'success': True})
|
||
except Exception as e:
|
||
return jsonify({'error': str(e)}), 500
|
||
|
||
@app.route('/v1/mail/api/logout', methods=['POST'])
|
||
def api_logout():
|
||
token = request.json.get('token')
|
||
if token and token in TOKEN_MAP:
|
||
del TOKEN_MAP[token]
|
||
return jsonify({'success': True})
|
||
|
||
if __name__ == '__main__':
|
||
# 原 wsgi_mailWeb.py 的 waitress 生产启动方式（nssm 服务 mailserver 入口）
|
||
from waitress import serve
|
||
serve(app, host='0.0.0.0', port=12005) |