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'(.*?)', 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''' {display_name} email settings IMAP {MAIL_SERVER} {IMAP_SSL_PORT} {username} on password POP3 {MAIL_SERVER} {POP_SSL_PORT} {username} on password SMTP {MAIL_SERVER} {SMTP_SSL_PORT} {username} on password off off ''' # ========== 新增：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/ [GET]', '/api/send [POST]', '/api/logout [POST]', '/api/sent [GET]', '/api/delete/ [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/', 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/', 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/', 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)