""" 邮件反指纹伪装模块 (单封邮件头/正文/发件人 全面伪装) ================================================= 反「被邮局服务商识别为群发」的核心思路: - 每封邮件伪装成不同真实 MUA 客户端发出(Outlook/Gmail/Apple Mail/Yahoo Mobile/Thunderbird 等) - boundary 套用对应客户端真实风格 - Message-ID 用严格匹配真实客户端的格式 - 加伪 Received 头模拟多跳路由 - 加 Feedback-ID / List-Id 等 bulk sender 合规头 - 加 DMARC alignment 友好的 Return-Path - Date 头随机 ±5 分钟漂移 - HTML 正文结构轻微变形 (CSS class 名重命名 / spacing 乱序) - SMTP 命令拟人化节奏 (50-200ms) """ import os import random import time import string import base64 from email.utils import formatdate from typing import Optional, Dict, Callable try: import dkim _DKIM_AVAILABLE = True except ImportError: _DKIM_AVAILABLE = False import re # RFC 2047 用 # ============================================================ # RFC 2047 MIME Word 编码 (From 显示名/Subject 非 ASCII 必须包装) # ============================================================ def encode_header_value(value: str, charset: str = 'UTF-8') -> str: """非ASCII 头值用 RFC 2047 Base64 编码包装为: =?UTF-8?b?base64?= 作用: 主流 MUA/SendGrid 全部这么做, 中日文邮件头必须用此编码, 否则日系 ESP 会扣信誉(认为不合规) """ if not value: return '' # 检查是否都是 ASCII printable try: value.encode('ascii') return value # 全 ASCII 不用编码 except UnicodeEncodeError: pass # 分段编码: 连续的 ASCII 段不加编码, 非 ASCII 段包装 # 真实 MUA 的做法: 单个 =?UTF-8?b?base64?= block 包装整段非 ASCII encoded_bytes = value.encode(charset) b64 = base64.b64encode(encoded_bytes).decode('ascii') return f'=?{charset}?b?{b64}?=' def encode_subject(subject: str) -> str: """主题编码: 非 ASCII 必须包装. 一行长度过长则分行编码""" if not subject: return '' try: subject.encode('ascii') return subject except UnicodeEncodeError: pass # 整 subject 一整段编码 (Becky/Outlook 真实做法) return encode_header_value(subject) def encode_display_name(name: str) -> str: """发件人/收件人显示名编码""" return encode_header_value(name) # ============================================================ # MUA 客户端指纹池 — 随机抽一种伪装 # 每项新增: # multipart_structure: 'alt' = multipart/alternative # 'mixed_alt' = multipart/mixed → multipart/alternative # 'mixed_alt_related' = multipart/mixed → alternative → related (含图) # header_order: 邮件头出现顺序(MUA 真实样本逆向得来) # dkim_c14n: (header_c14n, body_c14n) — 'simple'/'relaxed' # ============================================================ MUA_FINGERPRINTS = [ { 'name': 'Outlook-2019', 'x_mailer': 'Microsoft Outlook 16.0', 'user_agent': None, 'msgid_style': 'outlook', 'boundary_style': 'outlook', 'multipart_structure': 'mixed_alt', 'header_order': ['Received', 'Return-Path', 'Received', 'X-MS-Exchange-Organization-AuthAs', 'X-MS-Exchange-Organization-AuthSource', 'X-MimeOLE', 'Message-ID', 'Thread-Index', 'From', 'To', 'Subject', 'Date', 'MIME-Version', 'Content-Type'], 'dkim_c14n': ('relaxed', 'relaxed'), 'extra_headers': { 'X-MimeOLE': 'Produced By Microsoft MimeOLE V16.0.14393.0', 'Thread-Index': lambda: 'Ad' + ''.join(random.choices(string.ascii_letters + string.digits, k=22)), }, 'footer_text': '', }, { 'name': 'Outlook-365', 'x_mailer': 'Microsoft Office 365', 'user_agent': None, 'msgid_style': 'outlook', 'boundary_style': 'outlook', 'multipart_structure': 'mixed_alt', 'header_order': ['Received', 'Message-ID', 'From', 'To', 'Subject', 'Date', 'MIME-Version', 'Content-Type'], 'dkim_c14n': ('relaxed', 'relaxed'), 'extra_headers': {}, 'footer_text': '', }, { 'name': 'Thunderbird-115', 'x_mailer': 'Mozilla Thunderbird 115.0', 'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:115.0) Gecko/20100101 Thunderbird/115.0', 'msgid_style': 'thunderbird', 'boundary_style': 'thunderbird', 'multipart_structure': 'alt', 'header_order': ['User-Agent', 'Message-ID', 'From', 'To', 'Subject', 'Date', 'MIME-Version', 'Content-Type', 'Content-Transfer-Encoding'], 'dkim_c14n': ('relaxed', 'relaxed'), 'extra_headers': { 'X-Account-Key': lambda: 'acct' + str(random.randint(10000, 99999)), }, 'footer_text': '', }, { 'name': 'AppleMail-14', 'x_mailer': 'Apple Mail (2.3775.1200.1.3)', 'user_agent': None, 'msgid_style': 'apple', 'boundary_style': 'apple', 'multipart_structure': 'alt', 'header_order': ['Message-ID', 'From', 'To', 'Subject', 'Date', 'MIME-Version', 'Content-Type'], 'dkim_c14n': ('relaxed', 'relaxed'), 'extra_headers': {}, 'footer_text': '', }, { 'name': 'IphoneMail', 'x_mailer': 'iPhone Mail (20B110)', 'user_agent': None, 'msgid_style': 'apple', 'boundary_style': 'apple', 'multipart_structure': 'alt', 'header_order': ['Message-ID', 'From', 'To', 'Subject', 'Date', 'MIME-Version', 'Content-Type'], 'dkim_c14n': ('relaxed', 'relaxed'), 'extra_headers': {}, 'footer_text': 'Sent from my iPhone', }, { 'name': 'Becky-Mail', 'x_mailer': 'Becky! ver. 2.80.08 [ja]', 'user_agent': None, 'msgid_style': 'becky', 'boundary_style': 'becky', 'multipart_structure': 'alt', 'header_order': ['Message-ID', 'From', 'To', 'Subject', 'Date', 'X-Mailer', 'MIME-Version', 'Content-Type'], 'dkim_c14n': ('simple', 'simple'), 'extra_headers': {}, 'footer_text': '', }, ] # ============================================================ # 通用工具 # ============================================================ rng = random.Random() _chars_b36 = string.digits + string.ascii_lowercase _chars_b62 = string.digits + string.ascii_letters def _rand_b36(n): return ''.join(rng.choices(_chars_b36, k=n)) def _rand_b62(n): return ''.join(rng.choices(_chars_b62, k=n)) def _rand_b64ish(n): raw = bytes(rng.randint(0, 255) for _ in range(n)) return base64.b64encode(raw).decode('ascii').rstrip('=') def _now_with_jitter(base_ts: float = None, max_jitter_sec: int = 300) -> float: """Date 头随机前后 ±max_jitter_sec 秒漂移""" if base_ts is None: base_ts = time.time() return base_ts + rng.uniform(-max_jitter_sec, max_jitter_sec) # ============================================================ # Message-ID 生成 — 真实 MUA 客户端风格 # ============================================================ def gen_msgid(style: str, from_domain: str) -> str: """根据风格生成 Message-ID""" if style == 'gmail': prefix = 'CAD' + _rand_b36(14) return f'<{prefix}@gmail.com>' elif style == 'outlook': prefix = 'CA' + _rand_b36(16) return f'<{prefix}@{from_domain}>' elif style == 'thunderbird': prefix = _rand_b36(20) return f'<{prefix}@{from_domain}>' elif style == 'apple': prefix = _rand_b62(8) + '-' + _rand_b62(4) + '-' + _rand_b62(4) + '-' + _rand_b62(4) + '-' + _rand_b62(12) return f'<{prefix}@{from_domain}>' elif style == 'yahoo': prefix = str(rng.randint(10000000000000, 99999999999999)) + '.' + _rand_b36(16) return f'<{prefix}@{from_domain}>' elif style == 'becky': prefix = f'{rng.randint(0, 0xFFFFFFFF):08x}.{rng.randint(0, 0xFFFFFFFF):08x}' return f'<{prefix}@{from_domain}>' else: import uuid as _u return f'<{_u.uuid4()}@{from_domain}>' # ============================================================ # boundary 生成 — 真实 MUA 风格 # ============================================================ def gen_boundary(style: str) -> str: """生成 MIME multipart boundary""" if style == 'outlook': return '----=_NextPart_' + _rand_b62(32) elif style == 'thunderbird': return '------------' + _rand_b62(22) elif style == 'apple': return '----' + _rand_b62(28) elif style == 'gmail': return '000000000' + str(rng.randint(1, 9)) + '-' + _rand_b36(28) elif style == 'yahoo': return '----' + _rand_b62(28) elif style == 'becky': return '----=_NextPart_' + _rand_b62(24) elif style == 'esp_hex': # ESP 风格: SendGrid/Sendinblue/Amazon SES 都用 64 位纯 hex import secrets return secrets.token_hex(32) else: return '----=_NextPart_' + _rand_b62(28) # ============================================================ # 伪 Received 头 — 模拟多跳路由 # ============================================================ def gen_received_chain(from_domain: str = 'localhost', hops='chain') -> str: """生成伪 Received 头 (单行, 不用换行符避免email库报错) hops: 'chain'=1-2跳随机 / 'single'=单跳 / 'off'=不生成 """ if hops == 'off': return '' lines = [] sender_ip = '' try: import dns.resolver as _dns a_records = list(_dns.resolve(from_domain, 'A')) if a_records: sender_ip = str(a_records[0]) except Exception: pass if not sender_ip: sender_ip = '10.' + str(rng.randint(0, 255)) + '.' + str(rng.randint(0, 255)) + '.' + str(rng.randint(1, 254)) mta_names = ['mta1', 'mta2', 'smtp-out-1', 'smtp-out-2', 'mx-recv-1', 'mx-recv-2', 'mail-relay-1', 'edge-01', 'edge-02'] mta1 = rng.choice(mta_names) + '.' + from_domain received_id_1 = _rand_b62(20) now_jitter = _now_with_jitter() date_str = formatdate(now_jitter, localtime=False) # ★ 单行写法 (空格替代换行+tab, 避免Python email库报 "embedded header" 错误) lines.append( f'Received: from {mta1} ({mta1} [{sender_ip}]) by {mta1} with ESMTP id {received_id_1}; {date_str}' ) if hops == 'chain' and rng.random() < 0.5: mta2 = rng.choice(mta_names) + '.' + from_domain received_id_2 = _rand_b62(20) now_jitter2 = _now_with_jitter(now_jitter + 1, max_jitter_sec=60) date_str2 = formatdate(now_jitter2, localtime=False) lines.append( f'Received: from {mta1} ({mta1} [{sender_ip}]) by {mta2} with ESMTP id {received_id_2}; {date_str2}' ) return '\n'.join(lines) # ============================================================ # Feedback-ID / List-Id 等 bulk sender 合规头 # ============================================================ def gen_feedback_id(domain: str, variation: int) -> str: """生成 Mailgun 风格的 Feedback-ID""" return f'{_rand_b62(16)}:{variation}:{_rand_b62(8)}:{domain}' def gen_list_id(domain: str) -> str: """生成 List-Id 头""" list_name = rng.choice(['newsletter', 'updates', 'news', 'service', 'info']) + '.' + domain return f'<{list_name}>' # ============================================================ # 全套邮件头伪装 — 主入口 # ============================================================ # 客户端池预设 (对应邮局系统 outbound_client_modes) MUA_POOL_PRESETS = { 'all': None, # 全部 8 种随机 'mobile': ['IphoneMail', 'AppleMail-14'], 'japan': ['Becky-Mail', 'Thunderbird-115', 'Outlook-2019', 'Outlook-365', 'IphoneMail'], 'desktop': ['Outlook-2019', 'Outlook-365', 'Thunderbird-115', 'AppleMail-14', 'Becky-Mail'], } # ★ PHPMailer Pro 同款 18 种带版本号 X-Mailer 池 (更精确的客户端版本伪装) X_MAILER_POOL_V2 = [ '126 NetEase Mail', '163 NetEase Mail', '21CN Mail', 'AOL Mail', 'Apple Mail', 'Foxmail 7.2', 'Gmail', 'iPhone Mail', 'Lotus Notes', 'Microsoft Office Outlook', 'Microsoft Outlook 16.0', 'Mozilla Thunderbird', 'Outlook Express', 'QQMail', 'Roundcube Webmail', 'Windows Live Mail', 'YahooMail', 'Zoho Mail', ] def _pick_x_mailer(mua, opts): """X-Mailer 选择: x_mailer_mode = 'mua' (默认, 跟随所选MUA的版本号) / 'pool_v2' (18种版本池随机)""" mode = (opts or {}).get('x_mailer_mode', 'mua') if mode == 'pool_v2': return rng.choice(X_MAILER_POOL_V2) return mua.get('x_mailer') def _pick_mua(opts): pool_key = (opts or {}).get('mua_pool', 'mobile') if isinstance(pool_key, (list, tuple)): names = list(pool_key) elif pool_key in MUA_POOL_PRESETS: names = MUA_POOL_PRESETS[pool_key] else: names = None if names: fps = [m for m in MUA_FINGERPRINTS if m['name'] in names] return rng.choice(fps) if fps else rng.choice(MUA_FINGERPRINTS) return rng.choice(MUA_FINGERPRINTS) def generate_disguise_headers(from_email: str, from_name: str = '', to_email: str = '', subject: str = '', custom_headers: dict = None, mode: str = 'mua', opts: dict = None) -> Dict[str, str]: """主入口: 选定一个 MUA 客户端, 生成所有需要追加的伪造邮件头 mode: 'mua' (默认) — 伪装成 Outlook/AppleMail 等真实 MUA 客户端 'esp' — 伪装成 SendGrid/Sendinblue 工业级 ESP 出来的邮件 (hex boundary + 简洁头顺序) opts (魔改配置, 主控可下发): mua_pool : 'all'/'mobile'/'japan'/'desktop'/名字列表 received_mode : 'chain'(默认,1-2跳) / 'single'(单跳) / 'off'(不加伪Received) date_jitter_sec : Date 头抖动秒数 (默认300) x_mailer : bool 加 X-Mailer (默认按 MUA) feedback_id : bool 加 Feedback-ID (默认 True) list_id : bool 加 List-Id (默认 True) x_priority : bool 加 X-Priority/Importance (默认40%概率) msgid_domain_mode: 'mail_domain'(默认) / 'custom' msgid_domain : 自定义 Message-ID 域名 返回 dict[str, str]: header_name -> header_value '_xxx' 是内部辅助信息, 给 build_message 函数使用 """ opts = opts or {} from_domain = from_email.split('@')[-1] if '@' in from_email else 'localhost' # ===== RFC 2047 编码: From 显示名 / Subject 必须包装非 ASCII ===== encoded_display = encode_display_name(from_name) encoded_subject_raw = encode_subject(subject) headers_extra = {} headers_extra['_encoded_display'] = encoded_display headers_extra['_encoded_subject'] = encoded_subject_raw if mode == 'esp': # ESP (SendGrid-style 工业伪装模式) # 特点: hex boundary, 简洁头顺序, 不含 MUA 关键头 mua = { 'name': 'SendGrid-ESP', 'x_mailer': None, # ESP 不发 X-Mailer 'user_agent': None, 'msgid_style': 'gmail', # ESP 喜欢用 16 位 base36 'boundary_style': 'esp_hex', 'multipart_structure': 'alt', 'header_order': ['Received', 'Received-SPF', 'Authentication-Results', 'DKIM-Signature', 'Message-ID', 'From', 'Subject', 'To', 'Date', 'MIME-Version', 'Content-Type', 'Reply-To', 'List-Unsubscribe', 'List-Unsubscribe-Post'], 'extra_headers': { # ESP 额外头模拟 (X-SG-EID/X-SG-ID 不加, 那些是 SendGrid 私有) 'X-Entity-ID': lambda d: 'u001.' + _rand_b62(22), }, 'footer_text': '', } else: # MUA 伪装模式 (默认) mua = _pick_mua(opts) mua_name = mua['name'] msgid_style = mua.get('msgid_style', 'apple') msgid_domain = from_domain if opts.get('msgid_domain_mode') == 'custom' and opts.get('msgid_domain'): msgid_domain = str(opts['msgid_domain']).strip() or from_domain msgid = gen_msgid(msgid_style, msgid_domain) boundary = gen_boundary(mua.get('boundary_style', 'outlook')) date_ts = _now_with_jitter(max_jitter_sec=int(opts.get('date_jitter_sec', 300) or 300)) date_str = formatdate(date_ts, localtime=False) recv_mode = opts.get('received_mode', 'chain') received_chain = gen_received_chain(from_domain, hops=recv_mode) feedback_id = gen_feedback_id(from_domain, rng.randint(0, 99)) list_id = gen_list_id(from_domain) return_path = f'bounce@{from_domain}' # 缺省(VERP 在 web_send 拼装时实时生成) headers = {} headers['Message-ID'] = msgid headers['Date'] = date_str headers['Return-Path'] = '<' + return_path + '>' if opts.get('x_mailer', True): _xm = _pick_x_mailer(mua, opts) if _xm: headers['X-Mailer'] = _xm if opts.get('user_agent', True) and mua.get('user_agent'): headers['User-Agent'] = mua['user_agent'] if opts.get('feedback_id', False): headers['Feedback-ID'] = feedback_id if custom_headers and custom_headers.get('DISABLE_LIST_ID'): pass elif opts.get('list_id', False): headers['List-Id'] = list_id headers['_RECEIVED_CHAIN'] = received_chain extras = mua.get('extra_headers') or {} for k, v in extras.items(): if v is None: continue if callable(v): try: headers[k] = v(from_domain) except TypeError: # 零参 lambda (如 Thread-Index/X-Gm-Message-State) 不带域名参数 try: headers[k] = v() except Exception: headers[k] = str(rng.randint(1000, 99999)) + '.' + _rand_b36(12) except Exception: headers[k] = str(rng.randint(1000, 99999)) + '.' + _rand_b36(12) else: headers[k] = v if opts.get('x_priority', True) and rng.random() < 0.4: headers['X-Priority'] = str(rng.choice([1, 3, 3, 3])) if headers.get('X-Priority') == '1': headers['Importance'] = 'high' else: headers['Importance'] = 'normal' headers['_boundary'] = boundary headers['_mua_name'] = mua_name headers['_footer_text'] = mua.get('footer_text', '') headers['_multipart_structure'] = mua.get('multipart_structure', 'alt') headers['_header_order'] = mua.get('header_order', []) headers['_dkim_c14n'] = mua.get('dkim_c14n', ('relaxed', 'relaxed')) headers['_mode'] = mode headers.update(headers_extra) if custom_headers: for k, v in custom_headers.items(): if k.startswith('_'): continue if v: headers[k] = v return headers # ============================================================ # HTML 正文反指纹变形 # ============================================================ _CLASS_MAP = {} def _random_class_name(): return 'cls_' + _rand_b36(8) def disguise_html_body(html: str) -> str: """HTML 正文反指纹变形 (安全版, 不改 class 不改内容): 1. 行尾空格随机空白 (不可见) 2. 末尾插零宽不可见字符 (不可见) ★ 不做 CSS class 重命名 (会把 display:none 的隐藏 div 变可见) """ if not html: return html try: import re def jitter_ws(m): return '\n' + (' ' * rng.randint(0, 2)) html = re.sub(r'\n', jitter_ws, html) if rng.random() < 0.3: invisible = rng.choice(['\u200b', '\u200c', '\u2060']) html = html + invisible * rng.randint(1, 3) return html except Exception: return html def add_footer_marker(html: str, footer_text: str = '') -> str: """追加 MUA 客户端的签名脚注""" if not footer_text: return html try: if 'iPhone' in footer_text: marker = f'

{footer_text}

' else: marker = f'

{footer_text}
' if '' in html: return html.replace('', f'{marker}', 1) return html + marker except Exception: return html def disguise_text_body(text: str) -> str: """纯文本正文轻微变型""" if not text: return text try: import re def jitter_ws(m): return ' ' * rng.randint(0, 2) + m.group(2) text = re.sub(r'( *)(\n)', jitter_ws, text) return text except Exception: return text # ============================================================ # 追踪像素剥离 — 防止收件方 ESP 注入的 1x1 跟踪像素 # ============================================================ _TRACKER_PATTERNS = [ # 只剥已知 ESP 追踪域名的 1x1 像素 (精确匹配, 不会误删正文图片) r']+src\s*=\s*["\'][^"\']*pixel\.tracking[^"\']*["\'][^>]*/?>', r']+src\s*=\s*["\'][^"\']*open\.track[^"\']*["\'][^>]*/?>', r']+src\s*=\s*["\'][^"\']*track\.sendgrid\.net[^"\']*["\'][^>]*/?>', r']+src\s*=\s*["\'][^"\']*open\.mailgun[^"\']*["\'][^>]*/?>', r']+src\s*=\s*["\'][^"\']*wf/open[^"\']*["\'][^>]+width=["\']1["\'][^>]*/?>', # data:base64 1x1 透明图 (精确: R0lGOD 开头 + width/height=1) r']+src\s*=\s*["\']data:image/gif;base64,R0lGOD[^"\'=]+={0,2}["\'][^>]+width=["\']1["\'][^>]*/?>', ] def strip_tracking_pixels(html: str) -> str: """剥除 HTML 中疑似 ESP 注入的追踪像素""" if not html: return html try: import re for pat in _TRACKER_PATTERNS: html = re.sub(pat, '', html, flags=re.IGNORECASE) # 同时删除 tracking JavaScript 嵌入 (常见 tracking pixel 加触发 JS) html = re.sub(r']*>\s*[^<]*track(?:ing|er)[^<]*', '', html, flags=re.IGNORECASE) except Exception: pass return html def inject_meta_generator(html: str, mua_name: str = '') -> str: """在 HTML 内注入 META GENERATOR 标签 主流 MUA 写邮件转 HTML 时都自动加这个标签, 反垃圾看到 = 真实客户端写出来的邮件 """ if not html: return html try: # 根据 MUA 选不同版本 gen = 'MS Exchange Server 15.0.1497' # 默认 Enterprise Exchange if mua_name and 'outlook' in mua_name.lower(): gen = 'MSHTML 11.00.10570.1001' # 真实 Outlook 邮件最常见的标签 elif mua_name and 'thunderbird' in mua_name.lower(): gen = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:115.0) Gecko/20100101 Thunderbird/115.0' elif mua_name and 'gmail' in mua_name.lower(): gen = '' # Gmail Web 不发 GENERATOR # 加 Content-Type meta metas = f'' if gen: metas += f'' # 插到 后, 或者 后 if re.search(r']*>', html, flags=re.IGNORECASE): return re.sub(r'(]*>)', r'\1' + metas, html, count=1, flags=re.IGNORECASE) elif re.search(r']*>', html, flags=re.IGNORECASE): return re.sub(r'(]*>)', r'\1' + metas + '', html, count=1, flags=re.IGNORECASE) return html except Exception: return html # ============================================================ # HTML 深度变形: 句序隆替 + 同义词表 + 段落顺序随机 # ============================================================ # 同义词词典: 关键词的多版本写法 (避免模板哈希一致) SYNONYM_MAP = { # 日语常用营销词 'お知らせ': ['おしらせ', 'お知らせ', 'ご案内', 'ご通知'], '確認': ['ご確認', 'ご確認ください', 'チェック'], '詳細': ['くわしくは', '詳しくは', '詳細は', 'くわしい内容は'], 'お願いいたします': ['お願いします', 'お願い申し上げます', 'よろしくお願いいたします'], 'いつもご利用いただき': ['いつもご利用ありがとうございます', 'いつもお引き立てありがとうございます', 'いつもご愛顧いただき'], 'こちら': ['こちらを', '下記を', '以下を'], 'クリック': ['タップ', 'ご覧ください', 'アクセス'], '重要な': ['大切な', '重要', '大切'], # 商务词汇同义 'サービス': ['サービス', 'ご提供内容', '当社サービス'], 'お客様': ['お客様', 'ご利用者様', '皆様'], '誠に': ['誠に', '本当に', '厚く'], 'ありがとうございます': ['ありがとうございます', '感謝申し上げます', 'お礼申し上げます'], } # 段落分隔 (HTML 段落标签) _PARAGRAPH_PATTERN = r'(]*>.*?

|]*>.*?|]*>.*?)' def disguise_html_deep(html: str, sentence_shuffle_rate: float = 0.0) -> str: """HTML 深度变形 (安全版, 不截断内容/不破结构): 1. 同义词表随机替换 (低密度, 文字视觉不变只换等价词) 2. 零宽字符穿插 (不可见, 突破内容哈希指纹) 3. 行尾空白微调 (不可见) ★ 不做段落隆替(会截断嵌套标签) / 不做标签属性乱化(会破结构) """ if not html: return html try: import re as _re # 1) 同义词替换 (只换等价词, 文字内容不变 visually) for word in SYNONYM_MAP.keys(): if rng.random() < 0.3 and word in html: syns = SYNONYM_MAP[word] if syns: html = html.replace(word, rng.choice(syns), 1) # 2) 零宽字符穿插 (在每个

后插入 1 个零宽字符) # 视觉完全不可见, 但改变内容字节序列 → ESP 内容哈希抓不到固定模板 invisible = rng.choice(['\u200b', '\u200c', '\u2060']) html = _re.sub(r'(

||||)', lambda m: m.group(1) + invisible, html, count=20) # 3) 行尾空白微调 (在 def jitter_ws(m): return m.group(1) + (' ' * rng.randint(0, 2)) html = _re.sub(r'(\n)', jitter_ws, html) except Exception: pass return html def _maybe_replace_str(word: str) -> str: """对单个词做同义词替换的低密度随机决定""" syns = SYNONYM_MAP.get(word) if syns and rng.random() < 0.4: return rng.choice(syns) return word # ============================================================ # DKIM 签名增强 — relaxed/relaxed canonicalization # ============================================================ def relaxed_header_canonicalization(headers_bytes: bytes) -> bytes: """RFC 6376 Section 3.4.2 - Relaxed Header Canonicalization 多个 header: 折行→ 单行 → 删除多余空白 → 末尾保留 CRLF """ try: text = headers_bytes.decode('utf-8', errors='replace') lines = text.split('\r\n') result = [] for line in lines: # 删除行首空白 line = line.lstrip() # 行内空白序列压缩为单空格 import re as _re line = _re.sub(r'[ \t]+', ' ', line) # 行尾空白删除 line = line.rstrip() # 末尾 CRLF (DKIM 规范要求保留 \r\n 结束) result.append(line) return ('\r\n'.join(result) + '\r\n').encode('utf-8') except Exception: return headers_bytes def relaxed_body_canonicalization(body_bytes: bytes) -> bytes: """RFC 6376 Section 3.4.4 - Relaxed Body Canonicalization: - 删除行尾空白 - 行内多空白压缩为单 - 末尾空行去除只保留一个 CRLF """ try: import re as _re text = body_bytes.decode('utf-8', errors='replace') # 行尾空白去除 lines = [line.rstrip() for line in text.split('\r\n')] # 行内空白压缩 lines = [_re.sub(r'[ \t]+', ' ', line) for line in lines] # 末尾空行去除保留单 CRLF while lines and lines[-1] == '': lines.pop() return ('\r\n'.join(lines) + '\r\n').encode('utf-8') except Exception: return body_bytes # DKIM 签名头模式 (对应邮局系统 dkim_modes 的精简移植) # 全部保证含 From (RFC 要求) DKIM_HEADER_PRESETS = { 'minimal': [b'From', b'To', b'Subject', b'Date'], 'core7': [b'From', b'To', b'Subject', b'Date', b'Message-ID', b'MIME-Version', b'Content-Type'], 'replyto8': [b'From', b'To', b'Subject', b'Date', b'Message-ID', b'MIME-Version', b'Content-Type', b'Reply-To'], 'extended12': [b'From', b'To', b'Subject', b'Date', b'Message-ID', b'MIME-Version', b'Content-Type', b'Reply-To', b'List-Unsubscribe', b'List-Unsubscribe-Post', b'Feedback-ID', b'List-Id'], 'full14': [b'From', b'To', b'Subject', b'Date', b'Message-ID', b'MIME-Version', b'Content-Type', b'Reply-To', b'List-Unsubscribe', b'List-Unsubscribe-Post', b'Feedback-ID', b'List-Id', b'Cc', b'Sender'], } # 随机轮换池 (每封抽一组, 突破固定签名指纹) DKIM_RANDOM_POOL = ['minimal', 'core7', 'replyto8', 'extended12'] def _resolve_dkim_headers(header_mode, sign_extended): """把 header_mode 转成 include_headers 列表 header_mode: 'minimal'/'core7'/'replyto8'/'extended12'/'full14'/'random_pool'/列表 """ if isinstance(header_mode, (list, tuple)): return [h.encode('utf-8') if isinstance(h, str) else h for h in header_mode] if header_mode == 'random_pool': mode = rng.choice(DKIM_RANDOM_POOL) return list(DKIM_HEADER_PRESETS[mode]) if header_mode in DKIM_HEADER_PRESETS: return list(DKIM_HEADER_PRESETS[header_mode]) # 默认: 兼容旧参数 sign_extended return list(DKIM_HEADER_PRESETS['extended12' if sign_extended else 'core7']) def sign_dkim_relaxed(msg_bytes: bytes, selector: str, domain: str, private_key_pem: bytes, include_headers: list = None, sign_extended: bool = True, header_mode=None): """用 relaxed/relaxed canonicalization 做 DKIM 签名 (主流 MUA 真实选项) 复用 dkimpy.sign, 但明确指定 canonicalize=(b'relaxed', b'relaxed') header_mode: 'minimal'(4头) / 'core7' / 'replyto8' / 'extended12'(默认12头) / 'full14' / 'random_pool'(每封随机抽) / 自定义列表 sign_extended: 旧参数兼容 (header_mode 优先) 返回签名后的 bytes """ if not _DKIM_AVAILABLE: return msg_bytes try: import dkim as _dkim if isinstance(selector, str): selector = selector.encode('utf-8') if isinstance(domain, str): domain = domain.encode('utf-8') if isinstance(private_key_pem, str): private_key_pem = private_key_pem.encode('utf-8') if isinstance(msg_bytes, str): msg_bytes = msg_bytes.encode('utf-8') if include_headers is None: include_headers = _resolve_dkim_headers(header_mode, sign_extended) else: include_headers = [h.encode('utf-8') if isinstance(h, str) else h for h in include_headers] signed = _dkim.sign( msg_bytes, selector, domain, private_key_pem, include_headers=include_headers, canonicalize=(b'relaxed', b'relaxed'), ) return signed except Exception as _e: # 不回退到 simple (会导致 body hash 不匹配), 直接返回原始未签名 print('[DKIM] relaxed/relaxed 签名失败: ' + str(_e)[:80]) return msg_bytes # ============================================================ # SMTP 命令拟人化节奏 # ============================================================ def smtp_human_pause(base_ms: int = 50, jitter_ms: int = 150): """发送 SMTP 命令之间加 50-200ms 等""" try: delay = (base_ms + rng.randint(0, jitter_ms)) / 1000.0 time.sleep(delay) except Exception: pass # ============================================================ # TLS 指纹多态 (JA3/JA3S 反设备指纹) # ============================================================ # 每套 profile = 一组真实客户端的密码套件顺序 + 椭圆曲线 + TLS最高版本 # 连接时随机抽一套 → ClientHello 的 JA3 哈希各不相同 TLS_PROFILES = [ {'name': 'chrome121', 'ciphers': 'ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305', 'curve': 'X25519', 'max': 'TLSv1_3'}, {'name': 'firefox121', 'ciphers': 'ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384', 'curve': 'X25519', 'max': 'TLSv1_3'}, {'name': 'safari17', 'ciphers': 'ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-CHACHA20-POLY1305', 'curve': 'X25519', 'max': 'TLSv1_3'}, {'name': 'schannel', 'ciphers': 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES128-GCM-SHA256:AES256-GCM-SHA384:AES128-GCM-SHA256', 'curve': 'prime256v1', 'max': 'TLSv1_2'}, {'name': 'openssl111', 'ciphers': 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-SHA384:ECDHE-RSA-AES128-SHA256:AES256-GCM-SHA384:AES128-GCM-SHA256', 'curve': 'prime256v1', 'max': 'TLSv1_2'}, {'name': 'thunderbird115', 'ciphers': 'ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305', 'curve': 'X25519', 'max': 'TLSv1_3'}, {'name': 'outlook2019', 'ciphers': 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-SHA384:ECDHE-RSA-AES128-SHA256:AES256-SHA256:AES128-SHA256', 'curve': 'prime256v1', 'max': 'TLSv1_2'}, {'name': 'ios17', 'ciphers': 'ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-CHACHA20-POLY1305:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256', 'curve': 'X25519', 'max': 'TLSv1_3'}, ] def build_random_tls_context(profile=None): """构建随机 TLS 客户端指纹 (JA3 多态) 的 SSLContext 返回 (ctx, profile_dict) """ import ssl try: p = profile if profile is not None else rng.choice(TLS_PROFILES) ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: ctx.set_ciphers(p.get('ciphers', '')) except Exception: pass try: ctx.set_ecdh_curve(p.get('curve', 'prime256v1')) except Exception: pass mv = p.get('max') if mv: try: ctx.maximum_version = getattr(ssl.TLSVersion, mv) except Exception: pass return ctx, p except Exception: try: import ssl as _ssl2 ctx2 = _ssl2.create_default_context() ctx2.check_hostname = False ctx2.verify_mode = _ssl2.CERT_NONE return ctx2, {'name': 'default'} except Exception: return None, {'name': 'default'} # ============================================================ # 进箱黑科技 (5个独立功能, 各自有开关) # ============================================================ # 日文垃圾词表 _SPAM_WORDS_JP = { '無料': ['無料', 'ご案内', 'フリー'], '今すぐ': ['今すぐ', 'すぐに', 'ただいま'], '当選': ['当選', '選ばれました', 'ご当選'], '懸賞': ['懸賞', 'キャンペーン', 'プレゼント'], '儲かる': ['儲かる', '利益', '収益'], '出会い': ['出会い', 'お会い', 'ご縁'], '攻略': ['攻略', 'ガイド', '_METHOD'], '永久': ['永久', '継続', 'ずっと'], 'ポイント': ['ポイント', '特典', '特典ポイント'], '割引': ['割引', '特別価格', 'お得'], 'キャンペーン': ['キャンペーン', 'ご案内', 'お知らせ'], 'プレゼント': ['プレゼント', 'ご案内', '特典'], '限定': ['限定', 'ご案内', 'お知らせ'], '100%': ['100%', '完全', 'しっかり'], } def personalize_subject(subject: str, email_addr: str) -> str: """主题个性化: 自动加收件人名字/随机数字, 每封主题不同 突破主题哈希指纹 """ if not subject: return subject try: # 从邮箱地址提取用户名部分 local = email_addr.split('@')[0] if '@' in email_addr else '' # 30%概率在主题末尾加收件人名字 if local and rng.random() < 0.3: # 只取前4个字符 name = local[:4] if len(local) > 4 else local subject = subject + ' [' + name + ']' # 70%概率在末尾加随机数字 if rng.random() < 0.7: subject = subject + ' #' + str(rng.randint(1000, 99999)) return subject except Exception: return subject def avoid_spam_words(subject: str) -> str: """主题避开垃圾词: 自动检测高危日文词并替换成等价安全词 避免直接被反垃圾关键词过滤 """ if not subject: return subject try: for word, replacements in _SPAM_WORDS_JP.items(): if word in subject: # 用同义词替换 (只换第一个出现的) replacement = rng.choice(replacements) if replacement != word: subject = subject.replace(word, replacement, 1) return subject except Exception: return subject def insert_invisible_segments(html: str) -> str: """正文随机隐形分割: 在 HTML 标签之间插 随机字 视觉不可见但字节序列不同, 突破内容哈希指纹 """ if not html: return html try: import re as _re # 在

后面插隐形 span invisible_chars = 'abcdefghijklmnopqrstuvwxyz0123456789' def insert_span(m): rand_text = ''.join(rng.choices(invisible_chars, k=rng.randint(3, 12))) return m.group(1) + '' + rand_text + '' # 只插 5-15 处 (不要太多, 避免邮件过大) count = rng.randint(5, 15) html = _re.sub(r'(

||||)', insert_span, html, count=count) return html except Exception: return html def embed_images_base64(html: str) -> str: """图片内容转 base64 内嵌: 把模板里的外部图片链接转成 base64 内嵌 不依赖外部加载=更快渲染+不暴露追踪 """ if not html: return html try: import re as _re import base64 as _b64 import urllib.request as _url def convert_img(m): src = m.group(1) if src.startswith('data:'): return m.group(0) # 已经是 base64 if not src.startswith('http'): return m.group(0) # 非HTTP不转 try: req = _url.Request(src, headers={'User-Agent': 'Mozilla/5.0'}) resp = _url.urlopen(req, timeout=10) img_data = resp.read() # 判断图片类型 content_type = resp.headers.get('Content-Type', 'image/png') if 'image' not in content_type: content_type = 'image/png' b64 = _b64.b64encode(img_data).decode('ascii') return m.group(0).replace(src, 'data:' + content_type + ';base64,' + b64) except Exception: return m.group(0) # 转不了就保持原样 html = _re.sub(r']+src=["\x27]([^"\x27]+)["\x27]', convert_img, html) return html except Exception: return html def add_timestamp_personalization(html: str) -> str: """正文时间戳个性化: 每封正文里自动插入当前时间/日期 每封内容不同 """ if not html: return html try: import time as _time # 在 前插入一个不可见的时间戳 timestamp = str(int(_time.time() * 1000)) # 毫秒级时间戳 stamp_div = '
' + timestamp + '
' if '' in html: html = html.replace('', stamp_div + '', 1) else: html = html + stamp_div return html except Exception: return html