#!/usr/bin/env python3 """ 光湖语言系统 · 公钥守门人复活协议 v1.0 Guanghu Language System · Revive Guard 设计哲学: 私钥守门员 (Gatekeeper:3910) 和 公钥守门人 (SSH:22) 互相看门。 任一个倒下，另一个能通过本复活协议拉起来。 如果两个都倒了，本服务是最后防线（systemd Restart=always）。 协议: POST /revive/request → 速率限制 → 发邮件到冰朔主权邮箱 → 返回 challenge_id （⊢ 不需要密码 · 验证码只发到冰朔邮箱 · 只有冰朔能确认） POST /revive/confirm → 校验验证码 → systemctl restart gatekeeper + sshd + pm2 GET /health → 心跳检测 安全设计: ⊢ 不存密码 · 不进仓库 · 验证码用一次就扔 ⊢ SMTP 密码走环境变量 QQ_SMTP_AUTH_CODE（不进仓库） ⊢ 速率限制保护：每分钟最多 2 次复活请求 部署: 每台服务器 /opt/zhuyuan/revive-guard/revive-guard.py 监听: 0.0.0.0:8922 守护: systemd (Restart=always) 或 PM2 """ import os, json, time, hmac, hashlib, secrets, smtplib, subprocess, threading from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from http.server import HTTPServer, BaseHTTPRequestHandler # ═══════════════════════════════════════════ # 配置（所有服务器共享） # ═══════════════════════════════════════════ PORT = int(os.environ.get("REVIVE_GUARD_PORT", "8922")) SOVEREIGN_ID = "ICE-GL∞" SOVEREIGN_EMAIL = os.environ.get("SOVEREIGN_EMAIL", "565183519@qq.com") # QQ 邮箱 SMTP（小湖灯邮件通道） # ⊢ 密码不进仓库 · 走环境变量 QQ_SMTP_AUTH_CODE SMTP_HOST = "smtp.qq.com" SMTP_PORT = 465 SMTP_USER = os.environ.get("SMTP_USER", "565183519@qq.com") SMTP_PASS = os.environ.get("QQ_SMTP_AUTH_CODE", "") CODE_TTL = 300 # 验证码 5 分钟过期 RATE_LIMIT_WINDOW = 60 # 速率限制窗口 MAX_REQUESTS_PER_WINDOW = 2 # 无密码门槛后收紧：每分钟最多 2 次 # 运行时状态 pending_codes = {} # {challenge_id: {code, server_ip, expires_at}} rate_limit = [] # [timestamp, ...] lock = threading.Lock() def get_server_ip(): """获取本机公网 IP""" try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(2) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except: return "unknown" def send_email(code, server_ip): """小湖灯 · 发送复活验证码到冰朔主权邮箱 · HTML 美化版""" html_body = f"""
ICE-GL∞ 光湖语言系统
🛡️ 守门人复活 · 验证码
验 证 码
{code}
⏰ 5 分钟内有效 · 用完即废
服务器
{server_ip}
📋 使用方式
将此验证码发给 铸渊 → 铸渊调用
POST /revive/confirm → 守门人复活
⊢ 不进仓库 · 不存密码 · 一次即废
小湖灯 · 自动发送
国作登字-2026-A-00037559
ICE-GL∞ GUANGHU LANGUAGE SYSTEM
""" # 纯文本降级版（邮件客户端不支持 HTML 时显示） plain_body = f"""光湖语言系统 · 守门人复活 ══════════════════════ 验证码: {code} 服务器: {server_ip} 有效期: 5 分钟（用完即废） 将此验证码发给铸渊 → /revive/confirm ⊢ ICE-GL∞ 小湖灯自动发送 国作登字-2026-A-00037559""" msg = MIMEMultipart("alternative") msg.attach(MIMEText(plain_body, "plain", "utf-8")) msg.attach(MIMEText(html_body, "html", "utf-8")) msg["Subject"] = f"🔐 光湖 · 复活验证码 {code[:3]}***" msg["From"] = SMTP_USER msg["To"] = SOVEREIGN_EMAIL with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10) as s: s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) def revive_services(): """复活所有守门服务""" results = {} # 1. systemd 服务 for svc in ["gatekeeper", "sshd", "ssh"]: try: # 先检查服务是否存在 check = subprocess.run( ["systemctl", "is-enabled", svc], capture_output=True, text=True, timeout=5 ) if check.returncode != 0: continue r = subprocess.run( ["systemctl", "restart", svc], capture_output=True, text=True, timeout=15 ) results[svc] = "✅ restarted" if r.returncode == 0 else f"❌ {r.stderr.strip()[:80]}" except Exception as e: results[svc] = f"⚠️ {str(e)[:60]}" # 2. PM2 进程（gatekeeper 如果用 PM2 管） try: r = subprocess.run( ["pm2", "restart", "gatekeeper"], capture_output=True, text=True, timeout=15 ) results["gatekeeper(pm2)"] = "✅ restarted" if r.returncode == 0 else f"⚠️ {r.stderr.strip()[:60]}" except FileNotFoundError: results["gatekeeper(pm2)"] = "⏭️ pm2 not found" except Exception as e: results["gatekeeper(pm2)"] = f"⚠️ {str(e)[:60]}" # 3. 也尝试 restart api-proxy-gateway try: subprocess.run( ["pm2", "restart", "api-proxy-gateway"], capture_output=True, text=True, timeout=10 ) results["api-proxy(pm2)"] = "✅ restarted" except: pass return results def check_rate_limit(): """速率限制：每分钟最多 MAX_REQUESTS_PER_WINDOW 次""" now = time.time() with lock: rate_limit[:] = [t for t in rate_limit if now - t < RATE_LIMIT_WINDOW] if len(rate_limit) >= MAX_REQUESTS_PER_WINDOW: return False rate_limit.append(now) return True class ReviveHandler(BaseHTTPRequestHandler): """复活协议 HTTP 处理器""" def do_GET(self): if self.path == "/health": self.send_json(200, { "ok": True, "service": "revive-guard", "version": "1.0.0", "server": get_server_ip(), "sovereign": SOVEREIGN_ID }) else: self.send_json(404, {"error": "仅支持 POST /revive/request · /revive/confirm"}) def do_POST(self): try: length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) if length else {} except: self.send_json(400, {"error": "请求体需为 JSON"}) return if self.path == "/revive/request": self._handle_request(body) elif self.path == "/revive/confirm": self._handle_confirm(body) else: self.send_json(404, {"error": "未知端点 · 可用: /revive/request /revive/confirm /health"}) def _handle_request(self, body): # 速率限制（无密码门槛后加强：每分钟最多 2 次） if not check_rate_limit(): self.send_json(429, {"error": "请求过于频繁 · 请60秒后重试"}) return server_ip = body.get("server", get_server_ip()) # ⊢ 不需要密码 —— 验证码只发到冰朔主权邮箱 # ⊢ 只有冰朔（持有邮箱）能拿到验证码 → 只有冰朔能让铸渊确认复活 # ⊢ 验证码用一次就扔 · 推到仓库也没用 # SMTP 未配置 → 拒绝 if not SMTP_PASS: self.send_json(500, {"error": "SMTP 未配置 · 服务器管理员需设置 QQ_SMTP_AUTH_CODE 环境变量"}) return # 生成 6 位数字验证码 code = str(secrets.randbelow(900000) + 100000) cid = secrets.token_hex(16) # 清除过期 now = time.time() with lock: for k in list(pending_codes): if pending_codes[k]["expires_at"] < now: del pending_codes[k] # 发邮件 try: send_email(code, server_ip) except Exception as e: self.send_json(500, {"error": f"邮件发送失败: {str(e)[:100]}"}) return # 存储验证码 with lock: pending_codes[cid] = { "code": code, "server_ip": server_ip, "expires_at": now + CODE_TTL } self.send_json(200, { "ok": True, "challenge_id": cid, "message": f"验证码已发送至 {SOVEREIGN_EMAIL}", "expires_in": CODE_TTL, "server": server_ip }) def _handle_confirm(self, body): cid = body.get("challenge_id", "") code = body.get("code", "") with lock: entry = pending_codes.pop(cid, None) if not entry: self.send_json(403, {"error": "无效或过期的 challenge_id"}) return if entry["expires_at"] < time.time(): self.send_json(403, {"error": "验证码已过期"}) return if not hmac.compare_digest(entry["code"], code): self.send_json(403, {"error": "验证码错误"}) return # 🎯 复活！ results = revive_services() self.send_json(200, { "ok": True, "message": f"守门人复活完成 · {entry['server_ip']}", "revived": [k for k, v in results.items() if "✅" in v], "details": results }) def send_json(self, status, data): self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("X-Sovereign", "ICE-GL") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) def log_message(self, fmt, *args): """静默日志（生产环境不打印每条请求）""" pass if __name__ == "__main__": server = HTTPServer(("0.0.0.0", PORT), ReviveHandler) print(f"光湖·复活守门人 v1.0 · 监听 :{PORT}") print(f"主权者: {SOVEREIGN_ID} · 邮箱: {SOVEREIGN_EMAIL}") try: server.serve_forever() except KeyboardInterrupt: print("\n复活守门人已停止") server.shutdown()