""" 代理直投群发器 - Web版 (单机直投 SMTP 群发) ============================================== 启动: python app.py 访问: http://127.0.0.1:5000 特性: - 代理IP直投SMTP (SOCKS隧道 -> 收件域 MX:25) - STARTTLS 自动协商 + EHLO扩展感知 - IP 复用长连接 + 连续 N 封拒才换 + IP 主动轮换 - 同 IP 同 MX 速率限制 + 多 MX 轮询 + EHLO 随机 + SMTP 拟人化节奏 - 邮件反指纹伪装 (8种 MUA 客户端池 Message-ID/X-Mailer/伪Received/Date抖动) - multipart 真实结构 + Header 顺序 + 追踪像素剥离 + DKIM relaxed/relaxed - HTML 深度变形 (句序隆替 + 同义词表 + CSS 重命名 + 客户端购脚) - 发件域 warm-up + 同收件人冷却 + 退信率 backoff - 中性 SPF (v=spf1 ?all) + Cloudflare 域名一键配DNS - 凌晨定时暂停 - ★ 本地 VPN (大陆机器必须) ★ """ import os import sys import json import time import hmac import subprocess import random import string # Windows GBK 控制台直接运行 python app.py 时避免 ✓ 等符号编码崩溃 if sys.platform == 'win32': try: sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') except Exception: pass def _install_package(pkg_name): print('[AUTO INSTALL] 安装缺失依赖: ' + pkg_name) try: subprocess.check_call( [sys.executable, '-m', 'pip', 'install', pkg_name, '--quiet'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) print('[AUTO INSTALL] ✓ 安装成功: ' + pkg_name) return True except subprocess.CalledProcessError: # Ubuntu 24.04 / Debian 12 PEP 668: 需要 --break-system-packages try: subprocess.check_call( [sys.executable, '-m', 'pip', 'install', pkg_name, '--quiet', '--break-system-packages'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) print('[AUTO INSTALL] ✓ 安装成功(兼容模式): ' + pkg_name) return True except subprocess.CalledProcessError: print('[AUTO INSTALL] ✗ 安装失败: ' + pkg_name) return False except Exception as e: print('[AUTO INSTALL] ✗ 安装异常: ' + pkg_name) return False def _ensure_package(import_name, pip_name): try: __import__(import_name) return True except ImportError: if _install_package(pip_name): try: __import__(import_name) return True except ImportError: return False return False def _preflight_check(): print('=' * 60) print('启动前环境自检...') print('=' * 60) print() v = sys.version_info print('Python版本: ' + str(v.major) + '.' + str(v.minor) + '.' + str(v.micro)) if not (v.major == 3 and v.minor >= 8): print(' ✗ Python版本过低') return False print(' ✓ Python版本兼容') print() required = [ ('flask', 'flask', 'Web服务框架'), ('socks', 'PySocks', 'SOCKS代理隧道'), ('dns', 'dnspython', 'MX记录查询'), ('dkim', 'dkimpy', 'DKIM签名'), ('requests', 'requests', 'Cloudflare API'), ('cryptography', 'cryptography', 'DKIM密钥生成'), ('aiosmtpd', 'aiosmtpd', 'SMTP中继接收'), ] all_ok = True for mod_name, pip_name, desc in required: try: __import__(mod_name) print(' ✓ ' + pip_name + ' 已安装 (' + desc + ')') except ImportError: print(' ✗ ' + pip_name + ' 缺失 - 正在自动安装 (' + desc + ')') if not _ensure_package(mod_name, pip_name): all_ok = False else: print(' ✓ ' + pip_name + ' 自动安装成功') print() if not all_ok: print('✗ 环境自检失败, 请手动执行:') print(' pip install flask PySocks dnspython dkimpy requests cryptography') return False print('✓ 环境自检通过') print() return True if not _preflight_check(): print() print('=' * 60) print('启动失败 — 请按上面提示修复后重新运行') print('=' * 60) if sys.platform == 'win32' and sys.stdin.isatty(): try: input('按回车键退出...') except (EOFError, KeyboardInterrupt): pass sys.exit(1) import dns.resolver # noqa: F401 from flask import Flask, request, jsonify, send_file, render_template, Response, make_response, session, redirect import base64 as _b64 import secrets as _secrets app = Flask(__name__) _THIS_DIR = os.path.dirname(os.path.abspath(__file__)) _TEMPLATE_DIR = os.path.join(_THIS_DIR, 'templates') app.template_folder = _TEMPLATE_DIR # ============================================================ # Web 访问配置 (web_config.json): 公网部署必须配 auth_user/auth_pwd # ============================================================ WEB_CONFIG_FILE = os.path.join(_THIS_DIR, 'web_config.json') def _load_web_config(): cfg = {'host': '127.0.0.1', 'port': 5000, 'auth_user': '', 'auth_pwd': '', 'enable_direct_send': True, 'enable_bridge': True} if os.path.exists(WEB_CONFIG_FILE): try: with open(WEB_CONFIG_FILE, 'r', encoding='utf-8-sig') as f: loaded = json.load(f) for k in ('host', 'port', 'auth_user', 'auth_pwd', 'enable_direct_send', 'enable_bridge'): if k in (loaded or {}): cfg[k] = loaded[k] except Exception: pass return cfg WEB_CONFIG = _load_web_config() # 纯管理端模式: 关闭直投引擎与SMTP中继, 主控只做 数据/分配/参数/内容 管理 DIRECT_SEND_ENABLED = bool(WEB_CONFIG.get('enable_direct_send', True)) BRIDGE_ENABLED = bool(WEB_CONFIG.get('enable_bridge', True)) and DIRECT_SEND_ENABLED # ============================================================ # 多账户认证 (users.json): admin/员工账户 + 一账户绑定一台电脑 # ============================================================ USERS_FILE = os.path.join(_THIS_DIR, 'users.json') def _hash_pwd(pwd, salt): """旧算法 (SHA-256+salt): 只用于校验历史存量口令, 新口令一律走 PBKDF2""" import hashlib return hashlib.sha256((salt + '::' + str(pwd)).encode('utf-8')).hexdigest() _PBKDF2_ITER = 120_000 def _hash_pwd_v2(pwd, salt): """PBKDF2-HMAC-SHA256 (12万轮): 抗离线爆破; 存储格式 pbkdf2$$$""" import hashlib dk = hashlib.pbkdf2_hmac('sha256', str(pwd).encode('utf-8'), salt.encode('utf-8'), _PBKDF2_ITER) return 'pbkdf2$%d$%s$%s' % (_PBKDF2_ITER, salt, dk.hex()) def _is_pwd_v2(stored): return isinstance(stored, str) and stored.startswith('pbkdf2$') def _verify_pwd_field(password, stored, salt): """校验口令: 兼容 v2(PBKDF2) 与 v1(SHA-256); 返回 (是否通过, 是否需要升级到v2)""" if _is_pwd_v2(stored): try: _, it, s, h = stored.split('$', 3) dk = __import__('hashlib').pbkdf2_hmac('sha256', str(password).encode('utf-8'), s.encode('utf-8'), int(it)) return hmac.compare_digest(dk.hex(), h), False except Exception: return False, False return hmac.compare_digest(_hash_pwd(password, salt), stored), True def _load_users(): if os.path.exists(USERS_FILE): try: with open(USERS_FILE, 'r', encoding='utf-8') as f: data = json.load(f) return data.get('users') or {} except Exception: pass return {} def _save_users(users): try: with open(USERS_FILE, 'w', encoding='utf-8') as f: json.dump({'users': users}, f, ensure_ascii=False, indent=2) return True except Exception: return False def _verify_user_pwd(username, password): users = _load_users() u = users.get(username) if not u: return None ok, need_upgrade = _verify_pwd_field(password, u.get('pwd', ''), u.get('salt', '')) if not ok: return None # 旧口令自动升级为 PBKDF2 (登录即迁移, 无需用户改密) if need_upgrade: try: u['pwd'] = _hash_pwd_v2(password, u.get('salt') or _secrets.token_hex(8)) _save_users(users) except Exception: pass return u def _init_users(): """首次启动: 从 web_config 迁移 admin 账号到 users.json""" users = _load_users() if users: return au = str(WEB_CONFIG.get('auth_user') or 'admin').strip() ap = str(WEB_CONFIG.get('auth_pwd') or '') if not ap: ap = _secrets.token_hex(8) # 无配置则生成随机密码(日志会打印) salt = _secrets.token_hex(8) users = { au: {'pwd': _hash_pwd(ap, salt), 'salt': salt, 'role': 'admin', 'created_at': time.strftime('%Y-%m-%d %H:%M:%S'), 'last_login': ''}, } _save_users(users) print('[AUTH] 初始化账号: ' + au + ' / ' + ap) _init_users() # Session 密钥: 首次生成并持久化 (重启后登录态不失效) if not WEB_CONFIG.get('session_secret'): WEB_CONFIG['session_secret'] = _secrets.token_hex(16) try: with open(WEB_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(WEB_CONFIG, f, ensure_ascii=False, indent=2) except Exception: pass app.secret_key = WEB_CONFIG.get('session_secret', 'jpmail_secret_fallback') def _current_auth_info(): """返回当前登录用户信息 (session 或 Basic), 未登录返回 None 单会话控制: 账户新登录会生成新 token, 旧会话 token 不匹配即失效(被踢下线) """ user = session.get('user') if user: u = _load_users().get(user) if u and session.get('token') and session.get('token') == u.get('session_token', ''): return {'username': user, 'role': u.get('role', 'staff')} session.clear() return None auth = request.headers.get('Authorization', '') if auth.startswith('Basic '): try: decoded = _b64.b64decode(auth[6:]).decode('utf-8') uname, _, pwd = decoded.partition(':') u = _verify_user_pwd(uname, pwd) if u: session['user'] = uname session['role'] = u.get('role', 'staff') session['token'] = u.get('session_token', '') return {'username': uname, 'role': u.get('role', 'staff')} except Exception: pass return None @app.before_request def _auth_check(): info = _current_auth_info() if info: return None path = request.path # 放行登录页/登录接口 if path == '/login' or path.startswith('/api/auth/login'): return None if path.startswith('/api/'): return jsonify({'ok': False, 'msg': '未登录或登录已过期'}), 401 return redirect('/login') @app.after_request def _no_cache(resp): """页面不缓存 (防止旧版JS缓存导致按钮点击无响应)""" resp.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' resp.headers['Pragma'] = 'no-cache' resp.headers['Expires'] = '0' return resp @app.route('/api/auth/login', methods=['POST']) def api_auth_login(): """登录 (单会话: 新登录生效, 旧会话被踢下线)""" data = request.json or {} username = (data.get('username') or '').strip() password = str(data.get('password') or '') u = _verify_user_pwd(username, password) if not u: return jsonify({'ok': False, 'msg': '账号或密码错误'}) users = _load_users() token = _secrets.token_hex(16) u['session_token'] = token u['last_login'] = time.strftime('%Y-%m-%d %H:%M:%S') users[username] = u _save_users(users) session['user'] = username session['role'] = u.get('role', 'staff') session['token'] = token return jsonify({'ok': True, 'msg': '登录成功', 'role': u.get('role', 'staff'), 'username': username}) @app.route('/api/auth/logout', methods=['POST']) def api_auth_logout(): session.clear() return jsonify({'ok': True, 'msg': '已退出登录'}) @app.route('/api/auth/info') def api_auth_info(): info = _current_auth_info() if not info: return jsonify({'ok': False, 'msg': '未登录'}) return jsonify({'ok': True, 'username': info['username'], 'role': info['role']}) @app.route('/api/change_auth', methods=['POST']) def api_change_auth(): """修改当前登录账号的密码 (旧密码验证, 立即生效, 其他会话被踢下线)""" info = _current_auth_info() if not info: return jsonify({'ok': False, 'msg': '未登录'}) data = request.json or {} old_pwd = str(data.get('old_pwd') or '') new_pwd = str(data.get('new_pwd') or '') if len(new_pwd) < 6: return jsonify({'ok': False, 'msg': '新密码至少6位'}) users = _load_users() u = users.get(info['username']) ok_old, _ = _verify_pwd_field(old_pwd, (u or {}).get('pwd', ''), (u or {}).get('salt', '')) if not u or not ok_old: return jsonify({'ok': False, 'msg': '旧密码不正确'}) salt = _secrets.token_hex(8) u['pwd'] = _hash_pwd_v2(new_pwd, salt) u['salt'] = salt u['session_token'] = session.get('token', '') # 保留本会话 users[info['username']] = u _save_users(users) # 同步 web_config 的 auth_user/auth_pwd (Basic Auth 兼容) try: WEB_CONFIG['auth_user'] = info['username'] WEB_CONFIG['auth_pwd'] = new_pwd with open(WEB_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(WEB_CONFIG, f, ensure_ascii=False, indent=2) except Exception: pass return jsonify({'ok': True, 'msg': '密码已修改 (立即生效)'}) # ============================================================ # 一键部署节点 (主控 SSH → 新服务器自动安装节点) # ============================================================ @app.route('/api/node_deploy', methods=['POST']) def api_node_deploy(): if not DEPLOY_OK: return jsonify({'ok': False, 'msg': '部署引擎未加载 (主控缺少 paramiko: pip install paramiko)'}) data = request.json or {} name = (data.get('name') or '').strip() ip = (data.get('ip') or '').strip() ssh_port = int(data.get('ssh_port') or 22) ssh_user = (data.get('ssh_user') or 'root').strip() ssh_pwd = str(data.get('ssh_pwd') or '') node_port = int(data.get('node_port') or 8898) if not ssh_pwd: return jsonify({'ok': False, 'msg': '请填SSH密码'}) ips_text = (data.get('ips') or '').strip() if ips_text: import re as _re ips = [i.strip() for i in _re.split(r'[,\n\s]+', ips_text) if i.strip()] ips = list(dict.fromkeys(ips)) if not ips: return jsonify({'ok': False, 'msg': '批量IP为空'}) prefix = name or '节点' ok, msg = deploy_node.start_deploy_batch(ips, ssh_port, ssh_user, ssh_pwd, node_port, prefix) return jsonify({'ok': ok, 'msg': msg}) if not ip: return jsonify({'ok': False, 'msg': '请填节点IP'}) ok, msg = deploy_node.start_deploy(name, ip, ssh_port, ssh_user, ssh_pwd, node_port) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/node_deploy_status') def api_node_deploy_status(): if not DEPLOY_OK: return jsonify({'ok': True, 'running': False, 'done': True, 'logs': [], 'msg': '部署引擎未加载'}) d = deploy_node.deploy_status() d['ok'] = True return jsonify(d) # ============================================================ # IMAP 自动进箱测试 (节点发测试信 → IMAP 检查落点) # ============================================================ @app.route('/api/inbox_test', methods=['POST']) def api_inbox_test(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} node_id = data.get('node_id') or '' to_email = (data.get('to_email') or '').strip() imap_cfg = data.get('imap') or {} proxy = data.get('proxy') or {} subject_prefix = (data.get('subject_prefix') or 'お知らせ').strip() wait_sec = int(data.get('wait_sec') or 20) if '@' not in to_email: return jsonify({'ok': False, 'msg': '请填测试收件邮箱'}) if not imap_cfg.get('host'): return jsonify({'ok': False, 'msg': '请填 IMAP 服务器'}) node = master._find_node(node_id) if not node: return jsonify({'ok': False, 'msg': '节点不存在'}) token = 'ibx' + ''.join(random.choices(string.ascii_lowercase + string.digits, k=8)) subject = subject_prefix + ' [' + token + ']' html = ('

平素よりお世話になっております。

' '

このメールは配信テストです。テストID: ' + token + '

') ok, resp = master._node_call(node, 'POST', '/send_test', {'to': to_email, 'subject': subject, 'html': html}, timeout=90) sent_ok = bool(ok and (resp or {}).get('ok')) send_msg = (resp or {}).get('msg', '') if resp else '' send_code = str((resp or {}).get('code', '')) if resp else '' time.sleep(min(max(wait_sec, 5), 60)) try: import imap_inbox_test chk = imap_inbox_test.imap_check_inbox(imap_cfg, proxy, token) except Exception as e: chk = {'ok': False, 'placement': 'not_found', 'msg': 'IMAP检查异常: ' + str(e)[:80]} placement = chk.get('placement', 'not_found') if placement == 'inbox': verdict = '✅ 进收件箱' elif placement == 'spam': verdict = '❌ 进垃圾箱' else: verdict = '❓ 未找到(可能延迟/其他文件夹)' return jsonify({ 'ok': True, 'token': token, 'subject': subject, 'sent_ok': sent_ok, 'send_msg': send_msg, 'send_code': send_code, 'placement': placement, 'verdict': verdict, 'imap_msg': chk.get('msg', ''), 'folders_checked': chk.get('folders_checked', [])[:20], }) # ============================================================ # 邮局部署 (📮 Tab: 批量部署邮件栈 + 自动DNS + 监控) # ============================================================ @app.route('/api/mail_deploy_config', methods=['GET', 'POST']) def api_mail_deploy_config(): if not MAILDEPLOY_OK: return jsonify({'ok': False, 'msg': '邮局部署引擎未加载'}) if request.method == 'POST': data = request.json or {} cfg = mail_deploy.get_config() for k in cfg: if k in data and data[k] is not None: cfg[k] = data[k] mail_deploy.save_config(cfg) return jsonify({'ok': True, 'msg': '已保存', 'config': cfg}) return jsonify({'ok': True, 'config': mail_deploy.get_config()}) @app.route('/api/mail_servers', methods=['GET']) def api_mail_servers(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db servers = master_db.mail_server_all() for s in servers: s['ssh_pwd'] = '********' if s.get('ssh_pwd') else '' return jsonify({'ok': True, 'servers': servers}) @app.route('/api/mail_servers_status', methods=['GET']) def api_mail_servers_status(): """邮局服务器列表 + 实时发送状态""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) servers = master.mail_servers_live() for s in servers: s['ssh_pwd'] = '********' if s.get('ssh_pwd') else '' return jsonify({'ok': True, 'servers': servers}) @app.route('/api/mail_server_add', methods=['POST']) def api_mail_server_add(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} name = (data.get('name') or '').strip() ip = (data.get('ip') or '').strip() hostname = (data.get('hostname') or '').strip() if not ip: return jsonify({'ok': False, 'msg': '请填服务器IP'}) if '.' not in hostname: return jsonify({'ok': False, 'msg': '请填发件域名 (如 mail1.weicai168.com)'}) ms = { 'id': 'ms' + str(int(time.time() * 1000)), 'name': name or hostname, 'ip': ip, 'ssh_port': int(data.get('ssh_port') or 22), 'ssh_user': (data.get('ssh_user') or 'root').strip(), 'ssh_pwd': str(data.get('ssh_pwd') or ''), 'hostname': hostname, 'mail_stack': (data.get('mail_stack') or 'haraka').strip(), 'smtp_port': int(data.get('smtp_port') or 10587), 'minute_limit': max(0, int(data.get('minute_limit') if data.get('minute_limit') is not None else 0)), 'day_limit': max(0, int(data.get('day_limit') if data.get('day_limit') is not None else 0)), 'status': '未部署', 'dns_status': '', 'created_at': time.strftime('%Y-%m-%d %H:%M:%S'), } master_db.mail_server_add(ms) # 响应脱敏: 不回传明文 ssh_pwd ms_resp = dict(ms) if ms_resp.get('ssh_pwd'): ms_resp['ssh_pwd'] = '********' return jsonify({'ok': True, 'msg': '已添加: ' + (name or hostname), 'server': ms_resp}) @app.route('/api/mail_server_del', methods=['POST']) def api_mail_server_del(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} master_db.mail_server_del(data.get('id') or '') return jsonify({'ok': True, 'msg': '已删除'}) @app.route('/api/mail_server_update', methods=['POST']) def api_mail_server_update(): """更新邮局服务器字段 (smtp_port等)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} ms_id = data.get('id') or '' if not master_db.mail_server_get(ms_id): return jsonify({'ok': False, 'msg': '服务器不存在'}) updates = {} if 'smtp_port' in data and data['smtp_port'] is not None: updates['smtp_port'] = int(data['smtp_port']) if 'minute_limit' in data and data['minute_limit'] is not None: updates['minute_limit'] = max(0, int(data['minute_limit'])) if 'day_limit' in data and data['day_limit'] is not None: updates['day_limit'] = max(0, int(data['day_limit'])) if 'name' in data and data['name'] is not None: updates['name'] = str(data['name']) # 换IP流程: 允许改 IP/SSH信息 (烧IP后换新服务器重部署) if 'ip' in data and data['ip']: updates['ip'] = str(data['ip']) if 'ssh_port' in data and data['ssh_port'] is not None: updates['ssh_port'] = int(data['ssh_port']) if 'ssh_user' in data and data['ssh_user']: updates['ssh_user'] = str(data['ssh_user']) if 'ssh_pwd' in data and data['ssh_pwd']: updates['ssh_pwd'] = str(data['ssh_pwd']) if 'hostname' in data and data['hostname']: updates['hostname'] = str(data['hostname']) master_db.mail_server_update(ms_id, **updates) return jsonify({'ok': True, 'msg': '已更新'}) @app.route('/api/mail_servers_batch_limits', methods=['POST']) def api_mail_servers_batch_limits(): """批量修改邮局限速 (0=无限制)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} ids = data.get('ids') or [] minute_limit = max(0, int(data.get('minute_limit') if data.get('minute_limit') is not None else 0)) day_limit = max(0, int(data.get('day_limit') if data.get('day_limit') is not None else 0)) if not ids: return jsonify({'ok': False, 'msg': '未选择邮局服务器'}) n = 0 for ms_id in ids: if master_db.mail_server_get(ms_id): master_db.mail_server_update(ms_id, minute_limit=minute_limit, day_limit=day_limit) n += 1 m = '已更新 %d 台限速: %s/分, %s/天' % (n, minute_limit if minute_limit > 0 else '不限', day_limit if day_limit > 0 else '不限') return jsonify({'ok': True, 'msg': m}) @app.route('/api/check_domain_reputation', methods=['POST']) def api_check_domain_reputation(): """检测链接域名信誉: 域名黑名单(Spamhaus DBL/SURBL/URIBL/Barracuda) + Google举报 + 服务器IP黑名单""" import re as _re data = request.json or {} domain = (data.get('domain') or '').strip().lower() domain = _re.sub(r'^https?://', '', domain) domain = domain.split('/')[0].split('?')[0].split('#')[0].strip() if '.' not in domain: return jsonify({'ok': False, 'msg': '域名无效 (例: example.com 或 https://example.com)'}) results = [] def dns_bl(name, bl_domain, label): """DNS黑名单查询: NXDOMAIN=未列入""" try: import dns.resolver as _dns ans = _dns.resolve(name + '.' + bl_domain, 'A', lifetime=6) code = str(ans[0]) results.append({'source': label, 'bad': True, 'msg': '❌ 已列入 (' + code + ')'}) except Exception as e: if 'NXDOMAIN' in str(e) or 'does not exist' in str(e): results.append({'source': label, 'bad': False, 'msg': '✅ 未列入'}) else: results.append({'source': label, 'bad': False, 'msg': '⚠ 查询异常: ' + str(e)[:50]}) # 0. 域名解析检查 (店铺域名必须能解析; 不能解析 = 异常信号) a_ips = [] nxdomain = False try: import dns.resolver as _dns0 a_ips = [str(a) for a in _dns0.resolve(domain, 'A', lifetime=6)] if a_ips: results.append({'source': '域名解析', 'bad': False, 'msg': '✅ 正常解析 (' + a_ips[0] + ')'}) else: results.append({'source': '域名解析', 'bad': True, 'msg': '❌ 无A记录 (域名没有绑定服务器)'}) except Exception as e: if 'NXDOMAIN' in str(e) or 'does not exist' in str(e): nxdomain = True results.append({'source': '域名解析', 'bad': True, 'msg': '❌ 域名不存在/无法解析 (NXDOMAIN)'}) else: results.append({'source': '域名解析', 'bad': True, 'msg': '❌ 解析异常: ' + str(e)[:50]}) # 1-3. 域名黑名单 (Spamhaus DBL / SURBL / Barracuda BRBL; URIBL已弃用通配移除) dns_bl(domain, 'dbl.spamhaus.org', 'Spamhaus DBL') dns_bl(domain, 'multi.surbl.org', 'SURBL') dns_bl(domain, 'b.barracudacentral.org', 'Barracuda BRBL') # 4. Sucuri SiteCheck (免key, 覆盖恶意域名/无法访问) try: import requests as _rq r = _rq.get('https://sitecheck.sucuri.net/api/v3/', params={'scan': domain}, timeout=30, headers={'User-Agent': 'Mozilla/5.0'}) if r.status_code == 200: sj = r.json() bl = sj.get('blacklists') or [] warn = sj.get('warnings') or {} if bl: vendors = '、'.join([str(b.get('vendor', '')) for b in bl if b.get('vendor')]) results.append({'source': 'Sucuri黑名单', 'bad': True, 'msg': '❌ 已列入 (' + (vendors or '有记录') + ')'}) else: results.append({'source': 'Sucuri黑名单', 'bad': False, 'msg': '✅ 未列入'}) if warn: wmsgs = [] for k, items in warn.items(): if isinstance(items, list): for it in items: if isinstance(it, dict) and it.get('msg'): wmsgs.append(str(it.get('msg'))[:60]) if wmsgs: results.append({'source': 'Sucuri扫描', 'bad': True, 'msg': '❌ ' + '; '.join(wmsgs[:2])}) except Exception as e: results.append({'source': 'Sucuri', 'bad': False, 'msg': '⚠ 查询异常: ' + str(e)[:50]}) # 5. Google 安全浏览 (浏览器/Gmail 举报标记) try: import requests as _rq2 r = _rq2.get('https://transparencyreport.google.com/transparencyreport/api/v3/safebrowsing/status', params={'site': domain}, timeout=12, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}) if '"unsafe"' in r.text: results.append({'source': 'Google安全浏览', 'bad': True, 'msg': '❌ 已被标记不安全/被举报'}) else: results.append({'source': 'Google安全浏览', 'bad': False, 'msg': '✅ 未被标记'}) except Exception as e: results.append({'source': 'Google安全浏览', 'bad': False, 'msg': '⚠ 查询异常: ' + str(e)[:50]}) # 6. VirusTotal (可选: 配置项里填免费API Key后生效, 汇总70+杀软引擎) vt_key = str(WEB_CONFIG.get('vt_api_key') or '').strip() if vt_key: try: import requests as _rq3 r = _rq3.get('https://www.virustotal.com/api/v3/domains/' + domain, headers={'x-apikey': vt_key}, timeout=20) if r.status_code == 200: vj = r.json() stats = (vj.get('data', {}).get('attributes', {}).get('last_analysis_stats') or {}) mal = int(stats.get('malicious') or 0) susp = int(stats.get('suspicious') or 0) total = int(stats.get('harmless') or 0) + mal + susp + int(stats.get('undetected') or 0) if mal + susp > 0: results.append({'source': 'VirusTotal', 'bad': True, 'msg': '❌ 引擎报毒: 恶意' + str(mal) + ' 可疑' + str(susp) + ' (共' + str(total) + '引擎)'}) else: results.append({'source': 'VirusTotal', 'bad': False, 'msg': '✅ 未检出 (' + str(total) + '引擎)'}) elif r.status_code == 401 or r.status_code == 403: results.append({'source': 'VirusTotal', 'bad': False, 'msg': '⚠ API Key 无效'}) else: results.append({'source': 'VirusTotal', 'bad': False, 'msg': '⚠ 查询失败 HTTP' + str(r.status_code)}) except Exception as e: results.append({'source': 'VirusTotal', 'bad': False, 'msg': '⚠ 查询异常: ' + str(e)[:50]}) else: results.append({'source': 'VirusTotal', 'bad': False, 'msg': 'ℹ 未配置API Key (virustotal.com免费注册, 填入下方输入框)'}) # 7. 服务器IP黑名单 (域名A记录 → Spamhaus ZEN + Barracuda + SpamCop) try: import dns.resolver as _dnsr ips = a_ips or [str(a) for a in _dnsr.resolve(domain, 'A', lifetime=6)] for ip in ips[:2]: rev = '.'.join(reversed(ip.split('.'))) z_bad = False try: _dnsr.resolve(rev + '.zen.spamhaus.org', 'A', lifetime=5) z_bad = True except Exception: pass b_bad = False try: _dnsr.resolve(rev + '.b.barracudacentral.org', 'A', lifetime=5) b_bad = True except Exception: pass s_bad = False try: _dnsr.resolve(rev + '.bl.spamcop.net', 'A', lifetime=5) s_bad = True except Exception: pass if z_bad or b_bad or s_bad: parts = [] if z_bad: parts.append('Spamhaus ZEN') if b_bad: parts.append('Barracuda') if s_bad: parts.append('SpamCop') results.append({'source': '服务器IP ' + ip, 'bad': True, 'msg': '❌ 已列入: ' + '、'.join(parts)}) else: results.append({'source': '服务器IP ' + ip, 'bad': False, 'msg': '✅ 未列入 (Spamhaus ZEN/Barracuda/SpamCop)'}) except Exception as e: results.append({'source': '服务器IP', 'bad': False, 'msg': '⚠ A记录解析失败: ' + str(e)[:50]}) bad_count = sum(1 for r in results if r.get('bad')) if nxdomain: verdict = '❌ 域名不存在/无法解析 — 正常店铺域名必须有DNS解析, 强烈建议不要使用' elif bad_count: verdict = '❌ 域名或IP有黑名单记录, 建议不要在邮件里放这个链接' else: verdict = '✅ 域名+IP全部干净, 可以放心放在邮件里' return jsonify({'ok': True, 'domain': domain, 'results': results, 'bad_count': bad_count, 'verdict': verdict}) @app.route('/api/save_vt_key', methods=['POST']) def api_save_vt_key(): """保存 VirusTotal API Key (免费, 用于链接域名信誉检测)""" data = request.json or {} k = str(data.get('vt_api_key') or '').strip() WEB_CONFIG['vt_api_key'] = k try: with open(WEB_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(WEB_CONFIG, f, ensure_ascii=False, indent=2) return jsonify({'ok': True, 'msg': '已保存' + (' (留空则跳过VirusTotal检测)' if not k else '')}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)[:80]}) @app.route('/api/mail_server_test', methods=['POST']) def api_mail_server_test(): """测试邮局 SMTP 连接+认证""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db import ms_sender data = request.json or {} ms = master_db.mail_server_get(data.get('id') or '') if not ms: return jsonify({'ok': False, 'msg': '服务器不存在'}) if not ms.get('smtp_user') or not ms.get('smtp_pwd'): return jsonify({'ok': False, 'msg': '未部署 (无SMTP账号)'}) ok, msg = ms_sender.test_connection(ms) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/mail_server_import', methods=['POST']) def api_mail_server_import(): """批量导入邮局服务器 (原部署面板快速粘贴逻辑: IP多行 + 域名按序配对, 同IP覆盖更新)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db import re as _re import random as _rnd import string as _str data = request.json or {} ip_text = data.get('ip_text') or '' domain_text = data.get('domain_text') or '' prefix = (data.get('prefix') or '').strip() start_number = int(data.get('start_number') or 1) ssh_port = int(data.get('ssh_port') or 22) ssh_user = (data.get('ssh_user') or 'root').strip() ssh_pwd = str(data.get('ssh_pwd') or '') mail_stack = (data.get('mail_stack') or 'haraka').strip() smtp_port = int(data.get('smtp_port') or 10587) minute_limit = max(0, int(data.get('minute_limit') if data.get('minute_limit') is not None else 0)) day_limit = max(0, int(data.get('day_limit') if data.get('day_limit') is not None else 0)) preview_only = bool(data.get('preview', False)) ips = [i.strip() for i in _re.split(r'[,\n\s]+', ip_text) if i.strip()] domains = [] for line in (domain_text or '').split('\n'): parts = [p.strip() for p in line.split(',') if p.strip()] if parts: domains.append(parts[0]) if not ips: return jsonify({'ok': False, 'msg': '缺少IP'}) if not domains: return jsonify({'ok': False, 'msg': '缺少域名'}) def gen_hostname(idx): domain = domains[idx % len(domains)] if prefix: return prefix + str(start_number + idx) + '.' + domain return ''.join(_rnd.choices(_str.ascii_lowercase, k=_rnd.randint(4, 8))) + '.' + domain if preview_only: preview = [{'ip': ip, 'hostname': gen_hostname(idx), 'domain': domains[idx % len(domains)]} for idx, ip in enumerate(ips)] return jsonify({'ok': True, 'preview': preview, 'total': len(preview)}) existing = {s['ip']: s for s in master_db.mail_server_all()} added = updated = 0 for idx, ip in enumerate(ips): hostname = gen_hostname(idx) if ip in existing: old = existing[ip] ms = dict(old) ms['name'] = old.get('name') or hostname ms['ssh_port'] = ssh_port ms['ssh_user'] = ssh_user ms['ssh_pwd'] = ssh_pwd ms['hostname'] = hostname ms['mail_stack'] = mail_stack ms['smtp_port'] = smtp_port ms['minute_limit'] = minute_limit ms['day_limit'] = day_limit updated += 1 else: ms = { 'id': 'ms' + str(int(time.time() * 1000)) + str(idx), 'name': hostname, 'ip': ip, 'ssh_port': ssh_port, 'ssh_user': ssh_user, 'ssh_pwd': ssh_pwd, 'hostname': hostname, 'mail_stack': mail_stack, 'smtp_port': smtp_port, 'minute_limit': minute_limit, 'day_limit': day_limit, 'status': '未部署', 'dns_status': '', 'created_at': time.strftime('%Y-%m-%d %H:%M:%S'), } added += 1 master_db.mail_server_add(ms) return jsonify({'ok': True, 'msg': '导入完成: 新增 ' + str(added) + ' 台, 更新 ' + str(updated) + ' 台', 'added': added, 'updated': updated}) @app.route('/api/mail_deploy_start', methods=['POST']) def api_mail_deploy_start(): if not MAILDEPLOY_OK: return jsonify({'ok': False, 'msg': '邮局部署引擎未加载'}) data = request.json or {} ids = data.get('ids') or [] action = data.get('action') or 'deploy' if not ids: return jsonify({'ok': False, 'msg': '请先勾选邮局服务器'}) # 重复ID拦截: 传重复勾选直接拒绝 (前端bug兜底; worker侧亦有二次去重) if len(ids) != len(set(ids)): return jsonify({'ok': False, 'msg': '勾选异常: 存在重复服务器ID (%d个 → %d个唯一), 请刷新页面重试' % (len(ids), len(set(ids)))}) ok, msg = mail_deploy.start(ids, action) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/mail_deploy_status') def api_mail_deploy_status(): if not MAILDEPLOY_OK: return jsonify({'ok': False, 'msg': '邮局部署引擎未加载'}) return jsonify({'ok': True, **mail_deploy.status()}) @app.route('/api/mail_monitor', methods=['POST']) def api_mail_monitor(): if not MAILDEPLOY_OK: return jsonify({'ok': False, 'msg': '邮局部署引擎未加载'}) data = request.json or {} ids = data.get('ids') or [] if not ids: return jsonify({'ok': False, 'msg': '请先勾选邮局服务器'}) return jsonify({'ok': True, 'stats': mail_deploy.monitor(ids)}) _PROJECT_ROOT = os.path.dirname(_THIS_DIR) if _PROJECT_ROOT not in sys.path: sys.path.insert(0, _PROJECT_ROOT) # 集成群发引擎 (纯管理端模式下不加载直投引擎) if DIRECT_SEND_ENABLED: try: import web_send WEB_SEND_OK = True except Exception as _e: WEB_SEND_OK = False import traceback as _tr print('[WARN] web_send 导入失败, 群发Tab不可用:') print(_tr.format_exc()) else: WEB_SEND_OK = False print('[INFO] 纯管理端模式: 直投引擎已关闭 (只做数据/分配/参数/内容管理)') # 集成 DNS 管理 (CF / DKIM) try: from core.dns_manager import CloudFlareDNS from core.dkim_manager import DKIMManager from core.models import DNSAccount, ProxyIP, MailBox DNS_MANAGER_OK = True except Exception as _e: DNS_MANAGER_OK = False import traceback as _tr print('[WARN] core.* 导入失败, 域名管理Tab不可用:') print(_tr.format_exc()) # 一键部署节点引擎 (主控 SSH → 新服务器自动安装) try: import deploy_node DEPLOY_OK = True except Exception as _e: DEPLOY_OK = False import traceback as _tr print('[WARN] deploy_node 导入失败, 一键部署不可用:') print(_tr.format_exc()) # 邮局部署引擎 (主控 SSH → 批量安装邮件栈 + 自动写DNS) try: import mail_deploy MAILDEPLOY_OK = True except Exception as _e: MAILDEPLOY_OK = False import traceback as _tr print('[WARN] mail_deploy 导入失败, 邮局部署不可用:') print(_tr.format_exc()) # 集成 SMTP 中继接收 (纯管理端模式下不加载) if BRIDGE_ENABLED: try: import smtp_receive BRIDGE_OK = True except Exception as _e: BRIDGE_OK = False import traceback as _tr print('[WARN] smtp_receive 导入失败, SMTP中继不可用:') print(_tr.format_exc()) else: BRIDGE_OK = False print('[INFO] 纯管理端模式: SMTP中继已关闭') # 集成集群主控 (多节点控制) try: import master MASTER_OK = True except Exception as _e: MASTER_OK = False import traceback as _tr print('[WARN] master 导入失败, 集群控制不可用:') print(_tr.format_exc()) # 敏感配置文件路径 CF_CONFIG_FILE = os.path.join(_THIS_DIR, 'cf_config.json') DKIM_KEY_DIR = os.path.join(_THIS_DIR, 'dkim_keys') if not os.path.exists(DKIM_KEY_DIR): try: os.makedirs(DKIM_KEY_DIR, exist_ok=True) except Exception: pass # ============================================================ # Flask 路由 # ============================================================ @app.route('/') def index(): info = _current_auth_info() if not info: return redirect('/login') return render_template('base.html', direct_send_enabled=DIRECT_SEND_ENABLED, current_user_name=info['username'], current_user_role=info['role']) @app.route('/login') def login_page(): if _current_auth_info(): return redirect('/') return render_template('login.html') # ============================================================ # 群发 API # ============================================================ def _proxy_to_model(p): p = p or {} return ProxyIP( ip=p.get('host', ''), port=int(p.get('port') or 0), proxy_type=p.get('proxy_type', 'socks5'), username=p.get('user', ''), pwd=p.get('pwd', ''), enable=True, ) def _box_dict_to_model(b): b = b or {} return MailBox( enable=True, email=b.get('email', ''), user=b.get('email', ''), pwd=b.get('pwd', ''), sender_name=b.get('sender_name', ''), random_user=bool(b.get('random_user', False)), ) def _get_dkim_keys_for_boxes(boxes): keys = {} if not DNS_MANAGER_OK or not os.path.isdir(DKIM_KEY_DIR): return keys for b in boxes: email = b.get('email', '') if isinstance(b, dict) else getattr(b, 'email', '') if '@' in email: domain = email.split('@')[-1].lower() sel = (b.get('dkim_selector') if isinstance(b, dict) else getattr(b, 'dkim_selector', None)) or 'default' pem = os.path.join(DKIM_KEY_DIR, domain + '_' + sel + '.pem') if os.path.exists(pem) and domain not in keys: try: with open(pem, 'r', encoding='utf-8') as f: keys[domain] = {'selector': sel, 'private_key': f.read()} except Exception: pass return keys @app.route('/api/list_valid_emails') def api_list_valid_emails(): """列出 数据/valid_emails.txt 的邮箱 (供群发Tab读取)""" valid_file = os.path.join(_THIS_DIR, '数据', 'valid_emails.txt') if not os.path.exists(valid_file): return jsonify({'ok': True, 'count': 0, 'emails': []}) emails = [] try: with open(valid_file, 'r', encoding='utf-8') as f: for line in f: e = line.strip() if e and '@' in e: emails.append(e) except Exception: pass return jsonify({'ok': True, 'count': len(emails), 'emails': emails}) @app.route('/api/send_start', methods=['POST']) def api_send_start(): if not WEB_SEND_OK: return jsonify({'ok': False, 'msg': 'web_send 模块未加载'}) data = request.json or {} emails = data.get('emails') or [] if not emails: return jsonify({'ok': False, 'msg': '收件邮箱列表为空'}) boxes_raw = data.get('boxes') or [] boxes = [_box_dict_to_model(b) for b in boxes_raw] if not boxes: return jsonify({'ok': False, 'msg': '未配置发件箱'}) proxies_raw = data.get('proxies') or [] if not proxies_raw: return jsonify({'ok': False, 'msg': '未配置代理'}) proxies = [_proxy_to_model(p) for p in proxies_raw] dkim_keys = _get_dkim_keys_for_boxes(boxes_raw) cfg = { 'emails': emails, 'subjects': data.get('subjects') or ['重要なお知らせ'], 'html_parts': data.get('html_parts') or [''], 'text_parts': data.get('text_parts') or [], 'boxes': boxes, 'proxies': proxies, 'thread_count': int(data.get('thread_count') or 10), 'send_interval': float(data.get('send_interval') or 0), 'send_timeout': int(data.get('send_timeout') or 60), 'helo_domain': data.get('helo_domain') or '', 'reply_to': data.get('reply_to') or '', 'custom_headers': data.get('custom_headers') or {}, 'magic_config': data.get('magic_config') or {'enabled': False}, 'dkim_keys': dkim_keys, 'use_cache': bool(data.get('use_cache', True)), 'ip_reject_threshold': int(data.get('ip_reject_threshold') or 3), 'dns_servers': data.get('dns_servers') or None, # 反识别新参数 'disguise_enabled': bool(data.get('disguise_enabled', True)), 'ip_proactive_threshold': int(data.get('ip_proactive_threshold') or 30), 'mx_rate_per_minute': int(data.get('mx_rate_per_minute') or 0), 'schedule_pause': bool(data.get('schedule_pause', False)), 'schedule_pause_start': int(data.get('schedule_pause_start') or 0), 'schedule_pause_end': int(data.get('schedule_pause_end') or 6), 'smtp_human_pace': bool(data.get('smtp_human_pace', True)), # M+N+O warm-up 等 'warmup_enabled': bool(data.get('warmup_enabled', True)), 'warmup_cooldown_days': int(data.get('warmup_cooldown_days', 7)), # === 工业级进箱改进 === 'verp_enabled': bool(data.get('verp_enabled', True)), 'disguise_mode': data.get('disguise_mode', 'mua'), 'dkim_extended': bool(data.get('dkim_extended', True)), 'unsub_filter_enabled': bool(data.get('unsub_filter_enabled', True)), 'smtp_pace_mode': data.get('smtp_pace_mode', 'medium'), # === 本地 VPN (大陆机器必须) === 'use_local_vpn': bool(data.get('use_local_vpn', False)), 'local_vpn_port': int(data.get('local_vpn_port', 7890)), 'local_vpn_type': data.get('local_vpn_type', 'http'), 'always_new_ip': bool(data.get('always_new_ip', True)), # === 发件人显示控制 === 'from_display_name': data.get('from_display_name') or '', 'from_domain_mode': data.get('from_domain_mode') or 'auto', 'from_custom_domain': data.get('from_custom_domain') or '', 'personalize_subject': bool(data.get('personalize_subject', False)), 'avoid_spam_words': bool(data.get('avoid_spam_words', False)), 'invisible_segments': bool(data.get('invisible_segments', False)), 'embed_images': bool(data.get('embed_images', False)), 'timestamp_personalize': bool(data.get('timestamp_personalize', False)), 'send_mode': data.get('send_mode', 'auth'), 'raw_sender_domain': data.get('raw_sender_domain', ''), } try: web_send.reset_state() ok = web_send.start_send(cfg) if not ok: return jsonify({'ok': False, 'msg': '启动失败(看日志)'}) return jsonify({'ok': True, 'msg': '群发任务已启动'}) except Exception as e: return jsonify({'ok': False, 'msg': '启动异常: ' + str(e)}) @app.route('/api/send_stop', methods=['POST']) def api_send_stop(): if not WEB_SEND_OK: return jsonify({'ok': False}) web_send.stop_send() return jsonify({'ok': True}) @app.route('/api/send_progress') def api_send_progress(): if not WEB_SEND_OK: return jsonify({'status': 'error', 'msg': 'web_send 未加载'}) return jsonify(web_send.snapshot()) @app.route('/api/send_download') def api_send_download(): f = web_send.SEND_RESULT_CSV if WEB_SEND_OK else None if not f or not os.path.exists(f): return '暂无发送结果', 404 return send_file(f, as_attachment=True, download_name='send_result.csv') @app.route('/api/clean_send_result', methods=['POST']) def api_clean_send_result(): try: if WEB_SEND_OK and os.path.exists(web_send.SEND_RESULT_CSV): os.remove(web_send.SEND_RESULT_CSV) if WEB_SEND_OK and os.path.exists(web_send.SENT_CACHE_FILE): os.remove(web_send.SENT_CACHE_FILE) warmup_data = os.path.join(_THIS_DIR, 'warmup_data.json') if os.path.exists(warmup_data): os.remove(warmup_data) web_send.reset_state() return jsonify({'ok': True, 'msg': '已清空'}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)}) # ============================================================ # DNS 管理 (Cloudflare, 支持多账号) # ============================================================ def _load_cf_accounts(): """返回账号列表 [{api_email, api_key, label}]; 兼容旧版单账号格式""" if not os.path.exists(CF_CONFIG_FILE): return [] try: with open(CF_CONFIG_FILE, 'r', encoding='utf-8') as f: cfg = json.load(f) except Exception: return [] if isinstance(cfg, list): return [c for c in cfg if isinstance(c, dict) and c.get('api_key')] if isinstance(cfg, dict) and cfg.get('api_key'): return [{'api_email': cfg.get('api_email', ''), 'api_key': cfg.get('api_key', ''), 'label': cfg.get('label', '') or '默认账号'}] return [] def _save_cf_accounts(accounts): try: with open(CF_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(accounts, f, ensure_ascii=False, indent=2) return True except Exception: return False def _load_cf_account(): """兼容旧调用: 返回第一个账号""" accounts = _load_cf_accounts() return accounts[0] if accounts else None def _account_from_dict(cfg): if not cfg or not cfg.get('api_key'): return None _key = (cfg.get('api_key') or '').strip() is_token = (_key.startswith('cfat_') or _key.startswith('cf_') or _key.startswith('cfut_') or len(_key) >= 40) if not is_token and not cfg.get('api_email'): return None return DNSAccount(provider='cloudflare', api_email=cfg.get('api_email', ''), api_key=cfg.get('api_key', '')) def _account_zones(accounts): """{acct_idx: set(domain_lower)} 缓存每个账号的 zone 列表 (用于域名归属匹配)""" zones_cache = {} for i, a in enumerate(accounts): acc = _account_from_dict(a) if acc is None: zones_cache[i] = set() continue try: provider = CloudFlareDNS(acc) zones_cache[i] = set((d.domain or '').lower() for d in provider.get_domains()) except Exception: zones_cache[i] = set() return zones_cache def _make_dns_account(): cfg = _load_cf_account() if not cfg or not cfg.get('api_key'): return None return _account_from_dict(cfg) @app.route('/api/save_dns_account', methods=['POST']) def api_save_dns_account(): data = request.json or {} email = (data.get('api_email') or '').strip() key = (data.get('api_key') or '').strip() label = (data.get('label') or '').strip() accounts = _load_cf_accounts() idx = data.get('idx') if idx is not None and isinstance(idx, int) and 0 <= idx < len(accounts): # 更新已有账号: Key 留空则保留原Key accounts[idx]['api_email'] = email if key: accounts[idx]['api_key'] = key if label: accounts[idx]['label'] = label msg = '已更新账号: ' + (accounts[idx].get('label') or ('账号' + str(idx + 1))) else: if not key: return jsonify({'ok': False, 'msg': 'API Key 不能为空'}) label = label or ('账号' + str(len(accounts) + 1)) accounts.append({'api_email': email, 'api_key': key, 'label': label}) msg = '已添加账号: ' + label ok = _save_cf_accounts(accounts) return jsonify({'ok': ok, 'msg': msg if ok else '保存失败'}) @app.route('/api/get_dns_accounts') def api_get_dns_accounts(): accounts = _load_cf_accounts() result = [] for i, a in enumerate(accounts): k = a.get('api_key', '') masked = (k[:6] + '...' + k[-4:]) if len(k) > 12 else (k[:3] + '***' if k else '') result.append({'idx': i, 'api_email': a.get('api_email', ''), 'api_key': masked, 'label': a.get('label', ''), 'has_key': bool(k)}) return jsonify({'ok': True, 'accounts': result}) @app.route('/api/del_dns_account', methods=['POST']) def api_del_dns_account(): data = request.json or {} idx = data.get('idx') accounts = _load_cf_accounts() if not isinstance(idx, int) or not (0 <= idx < len(accounts)): return jsonify({'ok': False, 'msg': '账号不存在'}) removed = accounts.pop(idx) _save_cf_accounts(accounts) return jsonify({'ok': True, 'msg': '已删除账号: ' + (removed.get('label') or ('账号' + str(idx + 1)))}) @app.route('/api/get_dns_account') def api_get_dns_account(): cfg = _load_cf_account() k = (cfg or {}).get('api_key', '') masked = (k[:6] + '...' + k[-4:]) if len(k) > 12 else (('***' if k else '')) return jsonify({'ok': bool(cfg), 'api_email': (cfg or {}).get('api_email', ''), 'api_key': masked}) @app.route('/api/dns_test_auth') def api_dns_test_auth(): """测试 CF 认证 (可指定 ?idx= 测单账号, 缺省测全部) + 返回域名数 两种模式自动识别: - API Token (cfat_ 开头) → Bearer 认证, 不用 email - Global API Key (37 位 hex) → X-Auth-Email + X-Auth-Key, 必须有 email """ if not DNS_MANAGER_OK: return jsonify({'ok': False, 'msg': 'core 模块未加载 (请看启动日志找具体原因)'}) accounts = _load_cf_accounts() if not accounts: return jsonify({'ok': False, 'msg': '请先保存 CF 账号'}) idx = request.args.get('idx') try: idx = int(idx) if idx is not None else None except Exception: idx = None targets = [accounts[idx]] if idx is not None and 0 <= idx < len(accounts) else accounts import requests as _req ok_all = True total = 0 msgs = [] for a in targets: label = a.get('label') or '账号' acc = _account_from_dict(a) if not acc: ok_all = False msgs.append(label + ': 配置无效') continue try: provider = CloudFlareDNS(acc) headers = provider._auth_headers() resp = _req.get('https://api.cloudflare.com/client/v4/zones', headers=headers, params={'per_page': 5}, timeout=20) data = resp.json() if data.get('success'): n = len(data.get('result', [])) total += n msgs.append(label + ': 认证成功 (' + str(n) + '域)') else: ok_all = False errors = data.get('errors', []) msgs.append(label + ': ' + (errors[0].get('message') if errors else '未知错误')) except Exception as e: ok_all = False msgs.append(label + ': 异常 ' + str(e)[:60]) return jsonify({'ok': ok_all, 'msg': '; '.join(msgs), 'domain_count': total}) @app.route('/api/dns_pull_domains') def api_dns_pull_domains(): if not DNS_MANAGER_OK: return jsonify({'ok': False, 'msg': 'core 模块未加载 (请看启动日志找具体原因)'}) accounts = _load_cf_accounts() if not accounts: return jsonify({'ok': False, 'msg': '请先保存CF账号'}) result = [] errors = [] for i, a in enumerate(accounts): label = a.get('label') or ('账号' + str(i + 1)) acc = _account_from_dict(a) if acc is None: errors.append(label + ': 配置无效') continue try: provider = CloudFlareDNS(acc) domains = provider.get_domains() except Exception as e: errors.append(label + ': ' + str(e)[:60]) continue for d in domains: r = {'domain': d.domain, 'status': d.status, 'enable': True, 'acct_idx': i, 'acct_label': label} try: records = provider.get_records(d.domain) r['has_spf'] = any(x.get('type') == 'TXT' and 'spf1' in (x.get('content', '') or '').lower() for x in records) r['has_mx'] = any(x.get('type') == 'MX' for x in records) r['has_dkim'] = any(x.get('type') == 'TXT' and 'dkim' in (x.get('name', '') + x.get('content', '') or '').lower() for x in records) r['has_dmarc'] = any(x.get('type') == 'TXT' and 'dmarc' in (x.get('name', '') + x.get('content', '') or '').lower() for x in records) except Exception as e: r['error'] = str(e)[:80] result.append(r) return jsonify({'ok': True, 'domains': result, 'msg': ('部分账号失败: ' + '; '.join(errors)) if errors else ''}) @app.route('/api/dns_batch_configure', methods=['POST']) def api_dns_batch_configure(): if not DNS_MANAGER_OK: return jsonify({'ok': False, 'msg': 'core 模块未加载'}) data = request.json or {} domains = data.get('domains') or [] domain_accts = data.get('domain_accts') or [] # [{domain, acct_idx}] spf_policy = data.get('spf_policy', '?all') dmarc_policy = data.get('dmarc_policy', 'none') selector = data.get('selector', 'default') dns_mode = data.get('dns_mode', 'apex') # apex=主域名直接用 / sub=子域名使用 sub_prefix = str(data.get('sub_prefix') or '').strip().strip('.').lower() if dns_mode == 'sub' and not sub_prefix: return jsonify({'ok': False, 'msg': '子域名使用模式需要填子域名前缀'}) if not domains: return jsonify({'ok': False, 'msg': '未指定域名'}) accounts = _load_cf_accounts() if not accounts: return jsonify({'ok': False, 'msg': '先配置CF账号'}) acct_map = {x.get('domain'): x.get('acct_idx') for x in domain_accts} zones_cache = None def get_provider(domain): nonlocal zones_cache idx = acct_map.get(domain) if isinstance(idx, int) and 0 <= idx < len(accounts): acc = _account_from_dict(accounts[idx]) return CloudFlareDNS(acc) if acc else None # 未指定账号 → 按 zone 归属自动匹配 if zones_cache is None: zones_cache = _account_zones(accounts) for i in range(len(accounts)): if (domain or '').lower() in zones_cache.get(i, set()): acc = _account_from_dict(accounts[i]) return CloudFlareDNS(acc) if acc else None return None success, fail = 0, 0 err_msgs = [] for d in domains: try: provider = get_provider(d) if not provider: fail += 1 err_msgs.append(d + '(账号未匹配)') continue ok_all = True send_domain = (sub_prefix + '.' + d) if (dns_mode == 'sub' and sub_prefix) else d is_sub = send_domain != d spf_content = 'v=spf1 ' + spf_policy.strip() spf_name = sub_prefix if is_sub else '@' if provider.upsert_record(d, 'TXT', spf_name, spf_content, log_cb=lambda lvl, m: None) == 'failed': ok_all = False dmarc = 'v=DMARC1; p=' + dmarc_policy + '; sp=' + dmarc_policy + ';' dmarc_name = '_dmarc.' + sub_prefix if is_sub else '_dmarc' if provider.upsert_record(d, 'TXT', dmarc_name, dmarc, log_cb=lambda lvl, m: None) == 'failed': ok_all = False dns_txt = os.path.join(DKIM_KEY_DIR, send_domain + '_' + selector + '_dns.txt') if os.path.exists(dns_txt): try: with open(dns_txt, 'r', encoding='utf-8') as f: dkim_record = f.read().strip() dkim_name = selector + '._domainkey.' + sub_prefix if is_sub else selector + '._domainkey' if provider.upsert_record(d, 'TXT', dkim_name, dkim_record, log_cb=lambda lvl, m: None) == 'failed': ok_all = False except Exception: pass if ok_all: success += 1 else: fail += 1 err_msgs.append(d) except Exception: fail += 1 err_msgs.append(d) msg = '配置完成: 成功' + str(success) + ' 失败' + str(fail) if err_msgs: msg += ' | 失败域名: ' + ', '.join(err_msgs[:10]) return jsonify({'ok': True, 'success': success, 'fail': fail, 'msg': msg}) @app.route('/api/dkim_generate', methods=['POST']) def api_dkim_generate(): if not DNS_MANAGER_OK: return jsonify({'ok': False, 'msg': 'core 模块未加载'}) data = request.json or {} domains = data.get('domains') or [] domain_accts = data.get('domain_accts') or [] # [{domain, acct_idx}] selector = data.get('selector', 'default') dns_mode = data.get('dns_mode', 'apex') # apex=主域名直接用 / sub=子域名使用 sub_prefix = str(data.get('sub_prefix') or '').strip().strip('.').lower() if dns_mode == 'sub' and not sub_prefix: return jsonify({'ok': False, 'msg': '子域名使用模式需要填子域名前缀'}) if not domains: return jsonify({'ok': False, 'msg': '未指定域名'}) accounts = _load_cf_accounts() acct_map = {x.get('domain'): x.get('acct_idx') for x in domain_accts} zones_cache = None def get_provider(domain): nonlocal zones_cache idx = acct_map.get(domain) if isinstance(idx, int) and 0 <= idx < len(accounts): acc = _account_from_dict(accounts[idx]) return CloudFlareDNS(acc) if acc else None if zones_cache is None: zones_cache = _account_zones(accounts) for i in range(len(accounts)): if (domain or '').lower() in zones_cache.get(i, set()): acc = _account_from_dict(accounts[i]) return CloudFlareDNS(acc) if acc else None return None success, fail = 0, 0 for d in domains: try: send_domain = (sub_prefix + '.' + d) if (dns_mode == 'sub' and sub_prefix) else d kp = DKIMManager.generate_keypair(send_domain, selector=selector) DKIMManager.save_keypair(kp, DKIM_KEY_DIR) provider = get_provider(d) if provider: try: dkim_name = selector + '._domainkey' if send_domain == d else selector + '._domainkey.' + sub_prefix provider.upsert_record(d, 'TXT', dkim_name, kp['dns_record'], log_cb=lambda lvl, m: None) except Exception: pass success += 1 except Exception: fail += 1 return jsonify({'ok': True, 'success': success, 'fail': fail}) @app.route('/api/subdomain_batch_create', methods=['POST']) def api_subdomain_batch_create(): """批量创建子域名发件套件: 对每个 .<主域> 生成DKIM密钥 + 写 SPF/DKIM/DMARC 入参: domain(主域, zone必须在CF账号里), count(子域名数), prefix(默认s), spf_policy(默认?all 直投中性), dmarc_policy(默认none), selector(默认default) """ if not DNS_MANAGER_OK: return jsonify({'ok': False, 'msg': 'core 模块未加载'}) data = request.json or {} domain = str(data.get('domain') or '').strip().lower().strip('.') count = int(data.get('count') or 0) prefix = str(data.get('prefix') or 's').strip().strip('.').lower() spf_policy = data.get('spf_policy', '?all') dmarc_policy = data.get('dmarc_policy', 'none') selector = data.get('selector', 'default') if not domain: return jsonify({'ok': False, 'msg': '未指定主域名'}) if count < 1 or count > 200: return jsonify({'ok': False, 'msg': '子域名数量需在 1-200 之间'}) if not prefix: return jsonify({'ok': False, 'msg': '前缀不能为空'}) accounts = _load_cf_accounts() if not accounts: return jsonify({'ok': False, 'msg': '先配置CF账号'}) zones_cache = None def get_provider(dm): nonlocal zones_cache if zones_cache is None: zones_cache = _account_zones(accounts) for i in range(len(accounts)): if (dm or '').lower() in zones_cache.get(i, set()): acc = _account_from_dict(accounts[i]) return CloudFlareDNS(acc) if acc else None return None provider = get_provider(domain) if not provider: return jsonify({'ok': False, 'msg': domain + ' 不在CF账号的zone里'}) created = [] failed = [] for i in range(1, count + 1): sub = prefix + str(i) send_domain = sub + '.' + domain try: kp = DKIMManager.generate_keypair(send_domain, selector=selector) DKIMManager.save_keypair(kp, DKIM_KEY_DIR) ok_all = True spf_content = 'v=spf1 ' + str(spf_policy).strip() if provider.upsert_record(domain, 'TXT', sub, spf_content, log_cb=lambda lvl, m: None) == 'failed': ok_all = False dmarc = 'v=DMARC1; p=' + dmarc_policy + '; sp=' + dmarc_policy + ';' if provider.upsert_record(domain, 'TXT', '_dmarc.' + sub, dmarc, log_cb=lambda lvl, m: None) == 'failed': ok_all = False if provider.upsert_record(domain, 'TXT', selector + '._domainkey.' + sub, kp['dns_record'], log_cb=lambda lvl, m: None) == 'failed': ok_all = False if ok_all: created.append(send_domain) else: failed.append(send_domain) except Exception: failed.append(sub + '.' + domain) return jsonify({'ok': True, 'created': created, 'failed': failed, 'msg': '创建完成: 成功 ' + str(len(created)) + ' 个, 失败 ' + str(len(failed)) + ' 个'}) def api_preview_disguise(): """把模板 HTML 做反检测变形, 返回收件人真实收到的内容""" data = request.json or {} html = data.get('html', '') mode = data.get('mode', 'mua') if not html: return jsonify({'ok': True, 'html': ''}) try: from disguise_preview import disguise_for_preview result = disguise_for_preview(html, mode=mode) return jsonify({'ok': True, 'html': result}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)[:80], 'html': html}) @app.route('/api/get_settings') def api_get_settings(): info = { 'cf_configured': bool(_load_cf_accounts()), 'web_send_available': WEB_SEND_OK, 'dns_manager_available': DNS_MANAGER_OK, 'bridge_available': BRIDGE_OK, } return jsonify({'ok': True, 'settings': info}) # ============================================================ # SMTP 中继接收控制 (PHPMailer Pro / 伯勒 → 代理直投) # ============================================================ @app.route('/api/bridge_status') def api_bridge_status(): if not BRIDGE_OK: return jsonify({'ok': False, 'msg': 'smtp_receive 未加载'}) return jsonify({'ok': True, 'status': smtp_receive.bridge_status()}) @app.route('/api/bridge_start', methods=['POST']) def api_bridge_start(): if not BRIDGE_OK: return jsonify({'ok': False, 'msg': 'smtp_receive 未加载'}) ok, msg = smtp_receive.start_bridge_server() return jsonify({'ok': ok, 'msg': msg if ok else msg}) @app.route('/api/bridge_stop', methods=['POST']) def api_bridge_stop(): if not BRIDGE_OK: return jsonify({'ok': False, 'msg': 'smtp_receive 未加载'}) smtp_receive.stop_bridge_server() return jsonify({'ok': True}) @app.route('/api/bridge_save_config', methods=['POST']) def api_bridge_save_config(): if not BRIDGE_OK: return jsonify({'ok': False, 'msg': 'smtp_receive 未加载'}) data = request.json or {} cfg = dict(smtp_receive.DEFAULT_CONFIG) for k, v in data.items(): if k in cfg: cfg[k] = v saved = smtp_receive.save_config(cfg) return jsonify({'ok': saved, 'msg': '已保存到 ' + smtp_receive.CONFIG_FILE if saved else '保存失败'}) # ============================================================ # 集群控制 (主控 → 多节点直投) # ============================================================ @app.route('/api/nodes', methods=['GET']) def api_nodes_list(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) # token 脱敏: 列表接口不回传完整token (防泄露后直连节点) nodes = master.load_nodes() for n in nodes: t = n.get('token') or '' n['token'] = (t[:6] + '...' + t[-4:]) if len(t) > 12 else ('***' if t else '') return jsonify({'ok': True, 'nodes': nodes}) @app.route('/api/node_add', methods=['POST']) def api_node_add(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} node = master.node_add(data.get('name', ''), data.get('ip', ''), int(data.get('port') or 8898), data.get('token', '')) return jsonify({'ok': True, 'node': node}) @app.route('/api/node_del', methods=['POST']) def api_node_del(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ok = master.node_remove(data.get('node_id', '')) return jsonify({'ok': ok}) @app.route('/api/nodes_del', methods=['POST']) def api_nodes_del(): """批量删除节点 (含任务-节点关联)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) n = 0 for nid in ids: if master.node_remove(str(nid)): n += 1 return jsonify({'ok': n > 0, 'msg': '已删除 %d 个节点' % n}) @app.route('/api/node_status', methods=['POST']) def api_node_status(): """主控代理转发节点状态 (浏览器只连主控, 单通道稳定; 节点只接受主控IP)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} node = master._find_node(data.get('node_id', '')) if not node: return jsonify({'ok': False, 'online': False, 'msg': '节点不存在'}) ok, resp = master._node_call(node, 'GET', '/status', timeout=8) if not ok: return jsonify({'ok': False, 'online': False, 'msg': (resp or {}).get('msg', '离线')}) return jsonify({'ok': True, 'online': True, **resp}) @app.route('/api/proxies', methods=['GET']) def api_proxies(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db return jsonify({'ok': True, 'proxies': master_db.proxy_all()}) @app.route('/api/proxy_add', methods=['POST']) def api_proxy_add(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} if not (data.get('host') or '').strip(): return jsonify({'ok': False, 'msg': '代理地址不能为空'}) master_db.proxy_add(data) return jsonify({'ok': True, 'proxies': master_db.proxy_all()}) @app.route('/api/proxy_del', methods=['POST']) def api_proxy_del(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} master_db.proxy_remove(int(data.get('id') or 0)) return jsonify({'ok': True, 'proxies': master_db.proxy_all()}) @app.route('/api/proxy_clear', methods=['POST']) def api_proxy_clear(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db master_db.proxy_clear() return jsonify({'ok': True, 'proxies': master_db.proxy_all()}) @app.route('/api/node_ping', methods=['POST']) def api_node_ping(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} return jsonify(master.node_ping(data.get('node_id', ''))) @app.route('/api/uploads', methods=['GET']) def api_uploads(): """已上传的收件人文件列表""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) try: import master_db return jsonify({'ok': True, 'uploads': master_db.upload_all()}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)[:100]}) # 任务内容持久化 (旧单份配置, 兼容保留) + 内容库 (多版本) CONTENT_CONFIG_FILE = os.path.join(_THIS_DIR, 'content_config.json') @app.route('/api/content_save', methods=['POST']) def api_content_save(): """保存任务内容 (旧单份接口, 兼容保留)""" data = request.json or {} try: with open(CONTENT_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return jsonify({'ok': True, 'msg': '内容已保存'}) except Exception as e: return jsonify({'ok': False, 'msg': '保存失败: ' + str(e)[:100]}) @app.route('/api/content_load', methods=['GET']) def api_content_load(): """读取已保存的任务内容 (旧单份接口, 兼容保留)""" try: if os.path.exists(CONTENT_CONFIG_FILE): with open(CONTENT_CONFIG_FILE, 'r', encoding='utf-8') as f: return jsonify({'ok': True, 'content': json.load(f)}) except Exception: pass return jsonify({'ok': True, 'content': {}}) # ===== 内容库 (多版本: 保存/列表/读取/更新/删除) ===== @app.route('/api/contents', methods=['GET']) def api_contents_list(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db return jsonify({'ok': True, 'contents': master_db.content_all()}) @app.route('/api/content_lib_save', methods=['POST']) def api_content_lib_save(): """内容库保存: 带id=更新, 不带id=新增; 返回列表""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} name = (data.get('name') or '').strip() or ('内容-' + time.strftime('%m%d%H%M')) content = data.get('content') or {} try: cid = data.get('id') if cid: master_db.content_update(int(cid), name, content) return jsonify({'ok': True, 'msg': '已更新: ' + name, 'id': int(cid)}) new_id = master_db.content_add(name, content) return jsonify({'ok': True, 'msg': '已保存: ' + name, 'id': new_id}) except Exception as e: return jsonify({'ok': False, 'msg': '保存失败: ' + str(e)[:100]}) @app.route('/api/content_lib_get', methods=['POST']) def api_content_lib_get(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} item = master_db.content_get(data.get('id') or 0) if not item: return jsonify({'ok': False, 'msg': '内容不存在'}) return jsonify({'ok': True, 'item': item}) @app.route('/api/content_lib_del', methods=['POST']) def api_content_lib_del(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} master_db.content_del(data.get('id') or 0) return jsonify({'ok': True, 'contents': master_db.content_all()}) # ===== 魔改配置库 (多版本) ===== @app.route('/api/mk_libs', methods=['GET']) def api_mk_libs_list(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db return jsonify({'ok': True, 'mks': master_db.mk_all()}) @app.route('/api/mk_lib_save', methods=['POST']) def api_mk_lib_save(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} name = (data.get('name') or '').strip() or ('魔改-' + time.strftime('%m%d%H%M')) cfg = data.get('cfg') or {} try: mid = data.get('id') if mid: master_db.mk_update(int(mid), name, cfg) return jsonify({'ok': True, 'msg': '已更新: ' + name, 'id': int(mid)}) new_id = master_db.mk_add(name, cfg) return jsonify({'ok': True, 'msg': '已保存: ' + name, 'id': new_id}) except Exception as e: return jsonify({'ok': False, 'msg': '保存失败: ' + str(e)[:100]}) @app.route('/api/mk_lib_get', methods=['POST']) def api_mk_lib_get(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} item = master_db.mk_get(data.get('id') or 0) if not item: return jsonify({'ok': False, 'msg': '配置不存在'}) return jsonify({'ok': True, 'item': item}) @app.route('/api/mk_lib_del', methods=['POST']) def api_mk_lib_del(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} master_db.mk_del(data.get('id') or 0) return jsonify({'ok': True, 'mks': master_db.mk_all()}) @app.route('/api/upload_delete', methods=['POST']) def api_upload_delete(): """删除已上传的收件人文件""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) try: import master_db data = request.json or {} uid = int(data.get('id') or 0) for u in master_db.upload_all(): if u['id'] == uid: try: if os.path.exists(u['path']): os.remove(u['path']) except Exception: pass break master_db._exec('DELETE FROM uploads WHERE id=?', (uid,)) master_db._commit() return jsonify({'ok': True}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)[:100]}) @app.route('/api/upload_recipients', methods=['POST']) def api_upload_recipients(): """主控上传收件人文件 (存 master_data/upload_<时间戳>.txt)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) f = request.files.get('file') if not f: return jsonify({'ok': False, 'msg': '未收到文件'}) master._ensure_dirs() fname = 'upload_' + str(int(time.time())) + '.txt' path = os.path.join(master.MASTER_DATA_DIR, fname) try: f.save(path) count = 0 with open(path, 'r', encoding='utf-8') as fh: for line in fh: e = line.strip() if e and '@' in e: count += 1 try: import master_db master_db.upload_add(fname, path, count) except Exception: pass return jsonify({'ok': True, 'path': path, 'filename': fname, 'count': count}) except Exception as e: return jsonify({'ok': False, 'msg': str(e)[:100]}) @app.route('/api/node_test_proxy', methods=['POST']) def api_node_test_proxy(): """让指定节点测试一条代理是否可用""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} node = master._find_node(data.get('node_id', '')) if not node: return jsonify({'ok': False, 'msg': '节点不存在'}) ok, resp = master._node_call(node, 'POST', '/test_proxy', {'proxy': data.get('proxy') or {}}, timeout=25) return jsonify({'ok': ok and (resp or {}).get('ok', False), 'msg': (resp or {}).get('msg', '连接失败')}) @app.route('/api/task_create', methods=['POST']) def api_task_create(): """建任务 + 切分收件人 + 分发节点 + 远程启动""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ok, msg, task = master.task_create( name=data.get('name', ''), recipients_source=data.get('recipients_source', ''), content=data.get('content') or {}, send_cfg=data.get('send_cfg') or {}, boxes=data.get('boxes') or [], node_ids=data.get('node_ids') or [], node_proxies=data.get('node_proxies') or {}, node_threads=data.get('node_threads') or {}, node_quotas=data.get('node_quotas') or {}, boxes_per_node=int(data.get('boxes_per_node') or 0), dedupe=bool(data.get('dedupe', True)), auto_start=bool(data.get('auto_start', True)), recipients_sources=data.get('recipients_sources') or [], shuffle_merge=bool(data.get('shuffle_merge', True)), data_only=bool(data.get('data_only', False)), ) return jsonify({'ok': ok, 'msg': msg, 'task': task}) @app.route('/api/task_start', methods=['POST']) def api_task_start(): """启动已分配任务 (可选携带内容下发)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ok, msg = master.task_start(data.get('task_id', ''), data.get('content') or None, data.get('node_ids') or None) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/node_start', methods=['POST']) def api_node_start(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] content = data.get('content') or None if ids: ok, msg = master.nodes_start(ids, content=content) else: ok, msg = master.node_start(data.get('node_id', '')) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/nodes_clear', methods=['POST']) def api_nodes_clear(): """多选节点批量清数据 (任务+收件人+结果+缓存)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) results = master.nodes_clear(ids, clear_proxy_only=False) ok_count = sum(1 for r in results if r.get('ok')) return jsonify({'ok': True, 'msg': '已清数据 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点', 'results': results}) @app.route('/api/nodes_clear_proxy', methods=['POST']) def api_nodes_clear_proxy(): """多选节点批量清代理 (任务保留, 只清代理配置)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) results = master.nodes_clear(ids, clear_proxy_only=True) ok_count = sum(1 for r in results if r.get('ok')) return jsonify({'ok': True, 'msg': '已清代理 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点', 'results': results}) @app.route('/api/tasks_clear', methods=['POST']) def api_tasks_clear(): """批量清空任务记录(多选)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('task_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择任务'}) n = master.tasks_clear(ids) return jsonify({'ok': True, 'msg': '已清空 ' + str(n) + ' 个任务记录'}) @app.route('/api/tasks_stop', methods=['POST']) def api_tasks_stop(): """批量停止任务(多选)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('task_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择任务'}) ok_n, fail_n = master.tasks_stop(ids) return jsonify({'ok': True, 'msg': '已停止 ' + str(ok_n) + ' 个, 失败 ' + str(fail_n) + ' 个'}) @app.route('/api/tasks_pull', methods=['POST']) def api_tasks_pull(): """批量拉取结果(多选)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('task_ids') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择任务'}) total = master.tasks_pull(ids) return jsonify({'ok': True, 'msg': '已入库 ' + str(total) + ' 条结果'}) @app.route('/api/nodes_set_boxes', methods=['POST']) def api_nodes_set_boxes(): """多选域名+多选节点: 自动生成发件箱并连同DKIM密钥下发 (域名管理分配)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] domains = data.get('domains') or [] sender_name = data.get('sender_name') or 'お知らせ' if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) if not domains: return jsonify({'ok': False, 'msg': '未选择域名'}) boxes = [{'email': 'noreply@' + d, 'sender_name': sender_name, 'random_user': True, 'dkim_selector': 'default'} for d in domains] dkim_keys = master._load_master_dkim_keys(boxes) results = master.node_set_boxes(ids, boxes, dkim_keys) ok_count = sum(1 for r in results if r.get('ok')) fails = [r.get('name', '') + ': ' + r.get('msg', '') for r in results if not r.get('ok')] m = '已下发发件域名到 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点' if fails: m += ' | 失败: ' + '; '.join(fails[:5]) if not dkim_keys: m += ' | ⚠️ 未找到DKIM密钥(请先在域名管理生成DKIM)' return jsonify({'ok': True, 'msg': m, 'results': results}) @app.route('/api/nodes_set_content', methods=['POST']) def api_nodes_set_content(): """多选节点批量下发内容 (独立分配内容)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] content = data.get('content') or {} if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) results = master.node_set_content(ids, content) ok_count = sum(1 for r in results if r.get('ok')) fails = [r.get('name', '') + ': ' + r.get('msg', '') for r in results if not r.get('ok')] m = '已下发内容到 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点' if fails: m += ' | 失败: ' + '; '.join(fails[:5]) return jsonify({'ok': True, 'msg': m, 'results': results}) @app.route('/api/nodes_set_proxy', methods=['POST']) def api_nodes_set_proxy(): """多选节点批量下发代理池 (独立分配代理)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] proxies = data.get('proxies') or [] if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) results = master.node_set_proxy(ids, proxies) ok_count = sum(1 for r in results if r.get('ok')) return jsonify({'ok': True, 'msg': '已下发代理到 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点', 'results': results}) @app.route('/api/nodes_set_config', methods=['POST']) def api_nodes_set_config(): """多选节点批量下发魔改配置 (独立分配魔改)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] send_cfg = data.get('send_cfg') or {} if not ids: return jsonify({'ok': False, 'msg': '未选择节点'}) results = master.node_set_config(ids, send_cfg) ok_count = sum(1 for r in results if r.get('ok')) return jsonify({'ok': True, 'msg': '已下发魔改配置到 ' + str(ok_count) + '/' + str(len(ids)) + ' 个节点', 'results': results}) @app.route('/api/node_stop', methods=['POST']) def api_node_stop(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ids = data.get('node_ids') or [] if ids: ok, msg = master.nodes_stop(ids) else: ok, msg = master.node_stop(data.get('node_id', '')) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/task_list', methods=['GET']) def api_task_list(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) return jsonify({'ok': True, 'tasks': master.task_list()}) @app.route('/api/task_status', methods=['POST']) def api_task_status(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} st = master.task_status(data.get('task_id', '')) if st is None: return jsonify({'ok': False, 'msg': '任务不存在'}) return jsonify({'ok': True, 'status': st}) @app.route('/api/task_stop', methods=['POST']) def api_task_stop(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ok, msg = master.task_stop(data.get('task_id', '')) return jsonify({'ok': ok, 'msg': msg}) @app.route('/api/task_pull_results', methods=['POST']) def api_task_pull_results(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} r = master.task_pull_results(data.get('task_id', '')) if r is None: return jsonify({'ok': False, 'msg': '任务不存在'}) return jsonify({'ok': True, 'result': r}) @app.route('/api/task_results', methods=['POST']) def api_task_results(): """从 SQLite 查询发送结果明细 (键集分页, 亿级快速)""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db data = request.json or {} task_id = data.get('task_id') or None status = data.get('status') or None domain = data.get('domain') or None before_id = data.get('before_id') or None limit = int(data.get('limit') or 500) rows = master_db.result_query(task_id=task_id, status=status, domain=domain, before_id=before_id, limit=limit) stats = master_db.result_stats(task_id=task_id) has_more = len(rows) >= limit return jsonify({'ok': True, 'rows': rows, 'stats': stats, 'has_more': has_more}) @app.route('/api/stats', methods=['GET']) def api_stats(): """总览统计: 任务数/节点数/发送总量/成功率/域TOP""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db tasks = master.task_list() nodes = master.load_nodes() overall = master_db.result_stats() task_stats = [] for t in tasks: task_stats.append({'task_id': t['task_id'], 'name': t['name'], 'recipients_total': t['recipients_total'], **master_db.result_stats(t['task_id'])}) return jsonify({'ok': True, 'overall': overall, 'tasks': task_stats, 'node_count': len(nodes), 'task_count': len(tasks)}) @app.route('/api/results_export', methods=['POST']) def api_results_export(): """导出任务结果为 CSV""" if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) import master_db import csv as _csv import io as _io data = request.json or {} task_id = data.get('task_id') or None rows = master_db.result_query(task_id=task_id, limit=1000000) buf = _io.StringIO() w = _csv.writer(buf) w.writerow(['task_id', 'node_id', 'email', 'status', 'code', 'msg', 'timestamp', 'proxy_id']) for r in rows: w.writerow([r['task_id'], r['node_id'], r['email'], r['status'], r['code'], r['msg'], r['timestamp'], r['proxy_id']]) resp = make_response(buf.getvalue()) resp.headers['Content-Type'] = 'text/csv; charset=utf-8' resp.headers['Content-Disposition'] = 'attachment; filename=results_' + str(task_id or 'all') + '.csv' return resp @app.route('/api/task_remove', methods=['POST']) def api_task_remove(): if not MASTER_OK: return jsonify({'ok': False, 'msg': 'master 未加载'}) data = request.json or {} ok = master.task_remove(data.get('task_id', '')) return jsonify({'ok': ok}) if __name__ == '__main__': print('=' * 60) print('代理直投群发器 Web服务 (极致反检测版)') print('=' * 60) print() print('访问: http://127.0.0.1:5000') print() print('反检测特性:') print(' ✓ STARTTLS自动协商 + EHLO扩展感知') print(' ✓ 8种MUA客户端伪装 (Outlook/Gmail/Apple/Thunderbird/iPhone/Yahoo/Becky)') print(' ✓ Message-ID/伪Received/Feedback-ID/Date抖动/Header顺序') print(' ✓ multipart结构 + DKIM relaxed/relaxed canonicalization (12头)') print(' ✓ HTML追踪像素剥离 + 句序隆替 + 同义词表 + CSS重命名') print(' ✓ IP复用+连续N拒才换+主动轮换+同IP同MX速率限制+多MX轮询') print(' ✓ EHLO域随机子域 + 发件人邮箱每次随机') print(' ✓ 发件域warm-up曲线 + 收件人N天冷却 + 退信率backoff') print(' ✓ 凌晨定时暂停 + SMTP拟人化节奏 (fast/medium/slow)') print(' ✓ 中性SPF (v=spf1 ?all) + DKIM + DMARC自动配置') print(' ✓ RFC8058一键退订 + RFC2047编码 + VERP + ESP/MUA双模式') print(' ✓ 本地VPN支持 (大陆机器必须, 链路: 本机->VPN->代理->MX)') print() if not WEB_SEND_OK: print(' ⚠ web_send 未导入, 群发Tab将不可用') if not DNS_MANAGER_OK: print(' ⚠ core.dns_manager 未导入, 域名管理Tab将不可用') print() _wc = _load_web_config() print('访问: http://' + str(_wc.get('host', '127.0.0.1')) + ':' + str(_wc.get('port', 5000))) if _wc.get('auth_user'): print('认证: ' + str(_wc['auth_user']) + ' / ' + str(_wc['auth_pwd'])) print() app.run(host=str(_wc.get('host', '127.0.0.1')), port=int(_wc.get('port', 5000)), debug=False, threaded=True)