""" RFC 8058 一键退订微服务 (端口 5500) ================================================= 独立跑在 127.0.0.1:5500, 接收两个端点: GET /unsubscribe?email=xxx@yyy.com&t= -> 浏览器点击场景 (带防伪签名) POST /unsubscribe -> RFC 8058 List-Unsubscribe-Post: List-Unsubscribe=One-Click body: email=xxx@yyy.com&t=&... 邮件正文里退订头格式为: List-Unsubscribe: List-Unsubscribe-Post: List-Unsubscribe=One-Click ★ 安全机制 (防投毒/防伪造): 1. 每个退订链接带 HMAC 签名 (t= 参数), 与 email 绑定, 服务端验签后才写入 2. RFC 8058 一键退订场景豁免验签 (ESP 的 POST 无法预知签名), 但只接受 POST 且严格限速 3. email 永不原样回显进 HTML (转义 + 格式校验) 收件人邮箱客户端点"退订"按钮 -> ESP 自动 POST 到这个端点 本服务把邮箱写入 unsubscribe.txt -> 群发器启动时自动过滤 生成签名 (主控/群发器侧): from unsub_server import make_token t = make_token('xxx@yyy.com') 启动方式 (跟主程序并行跑): python unsub_server.py 或: start /B python unsub_server.py """ import os import sys import hmac import hashlib import threading import time import json from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse, parse_qs _THIS_DIR = os.path.dirname(os.path.abspath(__file__)) UNSUB_FILE = os.path.join(_THIS_DIR, 'unsubscribe.txt') UNSUB_LOCK = threading.Lock() LAST_LOG_TIME = [0] # ★ 签名密钥: 优先环境变量, 否则用本机文件密钥 (自动生成一次, 持久保存) _SECRET_FILE = os.path.join(_THIS_DIR, 'unsub_secret.key') def _load_secret(): env = os.environ.get('UNSUB_SECRET', '') if env: return env.encode('utf-8') try: if os.path.exists(_SECRET_FILE): with open(_SECRET_FILE, 'r', encoding='utf-8') as f: s = f.read().strip() if s: return s.encode('utf-8') s = hmac.new(b'unsub-kdf', os.urandom(32), hashlib.sha256).hexdigest() with open(_SECRET_FILE, 'w', encoding='utf-8') as f: f.write(s) try: os.chmod(_SECRET_FILE, 0o600) except Exception: pass return s.encode('utf-8') except Exception: # 文件不可写时退化为进程内随机 (重启后签名变化, 旧链接失效属可接受降级) return os.urandom(32) _SECRET = _load_secret() # ★ 限速: RFC 8058 豁免验签的 POST 通道, 按 IP 限速防批量投毒 _RATE = {} _RATE_LOCK = threading.Lock() _RATE_WINDOW = 60 # 秒窗口 _RATE_MAX = 20 # 每 IP 每窗口最大成功退订数 (正常 ESP 触发远低于此) def make_token(email: str) -> str: """生成与 email 绑定的防伪签名 (退订链接用)""" return hmac.new(_SECRET, ('unsub:' + (email or '').strip().lower()).encode('utf-8'), hashlib.sha256).hexdigest()[:32] def _verify_token(email: str, token: str) -> bool: if not token or not email: return False expect = make_token(email) return hmac.compare_digest(expect, str(token)) def _valid_email(email: str) -> bool: """严格格式校验: 只允许 [a-z0-9._%+-]@[a-z0-9.-] 且无换行/控制符""" email = (email or '').strip().lower() if not email or email.count('@') != 1: return False if any(c in email for c in '\r\n\t <>"\''): return False local, _, domain = email.partition('@') if not local or not domain or '.' not in domain: return False allowed = set('abcdefghijklmnopqrstuvwxyz0123456789._%+-') return all(c in allowed for c in local + domain) def _rate_ok(ip: str) -> bool: now = time.time() with _RATE_LOCK: lst = _RATE.setdefault(ip, []) # 清窗口外 while lst and now - lst[0] > _RATE_WINDOW: lst.pop(0) if len(lst) >= _RATE_MAX: return False lst.append(now) return True def _append_unsub(email: str): email = (email or '').strip().lower() if not _valid_email(email): return False with UNSUB_LOCK: # 去重 (读全部 + 追加) existing = set() if os.path.exists(UNSUB_FILE): try: with open(UNSUB_FILE, 'r', encoding='utf-8') as f: existing = set(line.strip().lower() for line in f if line.strip()) except Exception: pass if email in existing: return False # 已退订 with open(UNSUB_FILE, 'a', encoding='utf-8') as f: f.write(email + '\n') # 每30秒打一次日志(避免刷屏) now = time.time() if now - LAST_LOG_TIME[0] > 30: LAST_LOG_TIME[0] = now print(f'[UNSUB] {email} 已加入退订清单(总 {len(existing) + 1})') return True def _html_escape(s: str) -> str: return (s.replace('&', '&').replace('<', '<').replace('>', '>') .replace('"', '"').replace("'", ''')) class UnsubHandler(BaseHTTPRequestHandler): def _send_response(self, code=200, body=b'OK', content_type='text/plain; charset=utf-8'): self.send_response(code) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(len(body))) # 跨域头 (允许 ESP 从任意 Web 客户端调用; 只读退订操作无敏感数据) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() try: self.wfile.write(body) except Exception: pass def _client_ip(self): return self.client_address[0] if self.client_address else 'unknown' def do_OPTIONS(self): self._send_response(204, b'') def do_GET(self): parsed = urlparse(self.path) if parsed.path == '/unsubscribe': qs = parse_qs(parsed.query) email = (qs.get('email', [''])[0] or '').strip() token = (qs.get('t', [''])[0] or '').strip() if email and '@' in email and _valid_email(email) and _verify_token(email, token): if not _rate_ok(self._client_ip()): self._send_response(429, b'too many requests') return _append_unsub(email) # 返回浏览器友好页面 (日文) — email 已转义, 杜绝 XSS safe = _html_escape(email) body = ( '' '配信停止' '' '

配信を停止しました

' '

' + safe + ' への配信を停止しました。

' '

ご対応ありがとうございました。

' '' ).encode('utf-8') self._send_response(200, body, 'text/html; charset=utf-8') elif email and not _verify_token(email, token): # 有 email 但签名缺失/错误: 防投毒, 返回中性提示 (不确认该邮箱是否存在) body = b'

Invalid Request

link expired or invalid

' self._send_response(403, body, 'text/html; charset=utf-8') else: body = b'

Bad Request

email parameter missing

' self._send_response(400, body, 'text/html; charset=utf-8') elif parsed.path == '/health': self._send_response(200, b'{"ok":true}', 'application/json') else: self._send_response(404, b'Not Found') def do_POST(self): parsed = urlparse(self.path) if parsed.path != '/unsubscribe': self._send_response(404, b'Not Found') return # RFC 8058: body 形如 email=xxx@yyy.com&... 一般 form-urlencoded try: content_length = int(self.headers.get('Content-Length', 0) or 0) except ValueError: self._send_response(400, b'bad content-length') return if content_length > 10240: # 10KB 上限 self._send_response(413, b'payload too large') return body = self.rfile.read(content_length) if content_length else b'' email = '' token = '' try: qs = parse_qs(body.decode('utf-8', errors='replace')) email = (qs.get('email', [''])[0] or qs.get('Recipient', [''])[0] or '').strip() token = (qs.get('t', [''])[0] or '').strip() except Exception: pass if not email: # 也支持 Query qs_get = parse_qs(parsed.query) email = (qs_get.get('email', [''])[0] or '').strip() token = (qs_get.get('t', [''])[0] or '').strip() if not (email and _valid_email(email)): self._send_response(400, b'missing or invalid email') return # 验签: 带 t 参数必须验签通过; RFC 8058 一键退订(无 t)走限速豁免通道 if token: if not _verify_token(email, token): self._send_response(403, b'invalid token') return else: # ESP 一键退订无法带签名 → 限速防批量投毒 if not _rate_ok(self._client_ip()): self._send_response(429, b'too many requests') return _append_unsub(email) # RFC 8058 要求: 成功后返回 200 OK, body 可空 self._send_response(200, b'OK') def log_message(self, format, *args): pass # 静默日志,避免污染命令行 def start_unsub_server(port=5500, host='0.0.0.0'): """启动退订微服务 host='0.0.0.0' -> 接收外部 POST (如果 cloudflare proxy 给域名后) 若纯本地, 用 127.0.0.1 """ # 确保文件存在 if not os.path.exists(UNSUB_FILE): try: open(UNSUB_FILE, 'w').close() except Exception: pass server = HTTPServer((host, port), UnsubHandler) print('=' * 60) print(f' RFC 8058 一键退订微服务已启动 (HMAC验签+限速)') print(f' GET/POST http://{"127.0.0.1" if host=="0.0.0.0" else host}:{port}/unsubscribe') print(f' 退订清单: {UNSUB_FILE}') print('=' * 60) try: server.serve_forever() except KeyboardInterrupt: print('[UNSUB] 已停止') if __name__ == '__main__': # 默认 0.0.0.0:5500 (可被外网 Cloudflare proxy 反代为 https://unsubscribe.yourdomain.com) # 若仅本地用: python unsub_server.py --local HOST = '0.0.0.0' PORT = 5500 if '--local' in sys.argv: HOST = '127.0.0.1' if '--port' in sys.argv: idx = sys.argv.index('--port') if idx + 1 < len(sys.argv): PORT = int(sys.argv[idx + 1]) start_unsub_server(port=PORT, host=HOST)