import tkinter as tk from tkinter import ttk, messagebox import requests, json, os, threading, subprocess, shutil from datetime import datetime # ============ 基础配置 ============ requests.packages.urllib3.disable_warnings() CLOUDFLARE_API_BASE = "https://api.cloudflare.com/client/v4" CONFIG_FILE = "config.json" CLOUDFLARED_DIR = "cloudflared" # ============ Cloudflare API 封装 ============ class CloudflareAPI: def __init__(self, api_token): self.api_token = api_token self.headers = { "Authorization": f"Bearer {self.api_token}", "Content-Type": "application/json" } def _safe_request(self, method, url, **kwargs): try: r = requests.request(method, url, headers=self.headers, timeout=20, **kwargs) if r.status_code >= 400: # 返回统一结构，方便上层判断 return {"success": False, "status_code": r.status_code, "error_text": r.text.strip()[:500]} try: return r.json() except Exception: return {"success": False, "status_code": r.status_code, "error_text": r.text.strip()[:500]} except Exception as e: return {"success": False, "errors": [str(e)]} def create_tunnel(self, account_id, tunnel_name): url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/cfd_tunnel" data = {"name": tunnel_name} return self._safe_request("POST", url, json=data, verify=True) def list_tunnels(self, account_id): """ 列出账户下隧道 返回：list of tunnel dicts 或者空列表 """ url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/cfd_tunnel" res = self._safe_request("GET", url, verify=True) if res.get("success") and "result" in res: # result 通常是隧道列表 return res["result"] # 失败时返回空列表，并保留原始响应 return [] def get_routes(self, account_id, tunnel_id): url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/cfd_tunnel/{tunnel_id}/configurations" res = self._safe_request("GET", url, verify=True) if res.get("success") and "result" in res: ingress = res["result"].get("config", {}).get("ingress", []) routes = [r for r in ingress if "hostname" in r] return routes return [] def update_config(self, account_id, tunnel_id, ingress_rules): url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/cfd_tunnel/{tunnel_id}/configurations" data = {"config": {"ingress": ingress_rules}} try: r = requests.put(url, json=data, headers=self.headers, timeout=20, verify=True) return { "success": r.status_code == 200, "status_code": r.status_code, "result": r.json() if "application/json" in r.headers.get("Content-Type", "") else r.text } except Exception as e: return {"success": False, "errors": [str(e)]} # ============ 主界面类 ============ class TunnelApp: def __init__(self, root): self.root = root self.root.title("Cloudflare 隧道自动配置工具") self.root.geometry("980x720") self.routes = [] os.makedirs(CLOUDFLARED_DIR, exist_ok=True) self.cloudflared_process = None # 保存 cloudflared 进程句柄 self.tunnel_list = [] # 存放从 API 获取的隧道信息 self._create_widgets() self._load_config() self._check_ssl() def _create_widgets(self): frame = ttk.Frame(self.root, padding=12) frame.pack(fill=tk.BOTH, expand=True) row = 0 ttk.Label(frame, text="API Token:").grid(row=row, column=0, sticky="w") self.api_token = ttk.Entry(frame, width=60, show="*") self.api_token.grid(row=row, column=1, columnspan=4, sticky="w", pady=2) row += 1 ttk.Label(frame, text="Account ID:").grid(row=row, column=0, sticky="w") self.account_id = ttk.Entry(frame, width=60) self.account_id.grid(row=row, column=1, columnspan=2, pady=2, sticky="w") row += 1 ttk.Label(frame, text="Zone ID:").grid(row=row, column=0, sticky="w") self.zone_id = ttk.Entry(frame, width=60) self.zone_id.grid(row=row, column=1, columnspan=2, pady=2, sticky="w") row += 1 ttk.Label(frame, text="隧道 ID:").grid(row=row, column=0, sticky="w") self.tunnel_id = ttk.Entry(frame, width=60) self.tunnel_id.grid(row=row, column=1, columnspan=2, pady=2, sticky="w") row += 1 # 新增：已有隧道下拉选择与刷新按钮 ttk.Label(frame, text="已有隧道:").grid(row=row, column=0, sticky="w") self.tunnel_combo = ttk.Combobox(frame, width=60, state="readonly") self.tunnel_combo.grid(row=row, column=1, columnspan=2, sticky="w", pady=2) self.tunnel_combo.bind("<>", lambda e: threading.Thread(target=self.on_tunnel_selected, daemon=True).start()) ttk.Button(frame, text="🔁 刷新隧道列表", command=self._thread_refresh_tunnel_list).grid(row=row, column=3, padx=6, sticky="w") ttk.Button(frame, text="🔍 获取选中隧道路由", command=lambda: threading.Thread(target=self.refresh_routes_for_selected, daemon=True).start()).grid(row=row, column=4, padx=6, sticky="w") row += 1 ttk.Label(frame, text="主域名:").grid(row=row, column=0, sticky="w") self.domain = ttk.Entry(frame, width=60) self.domain.grid(row=row, column=1, columnspan=2, pady=2, sticky="w") row += 1 ttk.Label(frame, text="隧道名称:").grid(row=row, column=0, sticky="w") self.tunnel_name = ttk.Entry(frame, width=60) self.tunnel_name.grid(row=row, column=1, columnspan=2, pady=2, sticky="w") row += 1 ttk.Separator(frame, orient="horizontal").grid(row=row, column=0, columnspan=5, sticky="ew", pady=8) row += 1 ttk.Label(frame, text="子域名:").grid(row=row, column=0, sticky="w") self.subdomain_entry = ttk.Entry(frame, width=25) self.subdomain_entry.grid(row=row, column=1, sticky="w") ttk.Label(frame, text="端口号:").grid(row=row, column=2, sticky="w") self.port_entry = ttk.Entry(frame, width=15) self.port_entry.grid(row=row, column=3, sticky="w") ttk.Button(frame, text="➕ 添加路由", command=self.add_route).grid(row=row, column=4, padx=5) row += 1 self.tree = ttk.Treeview(frame, columns=("hostname", "port"), show="headings", height=12) self.tree.heading("hostname", text="主机名") self.tree.heading("port", text="端口号") self.tree.column("hostname", width=580) self.tree.column("port", width=100) self.tree.grid(row=row, column=0, columnspan=5, pady=5, sticky="nsew") row += 1 btn_frame = ttk.Frame(frame) btn_frame.grid(row=row, column=0, columnspan=5, pady=10, sticky="w") ttk.Button(btn_frame, text="🚀 创建新隧道", command=self._thread_create_tunnel).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="🔁 添加/更新路由", command=self._thread_update_routes).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="💾 保存配置", command=self._save_config).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="🛑 停止隧道", command=self._thread_stop_tunnel).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="🗑 一键删除当前隧道", command=self._thread_delete_tunnel).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="❌ 删除选中路由", command=self.delete_selected_route).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="✏️ 修改选中路由", command=self.modify_selected_route).pack(side=tk.LEFT, padx=6) ttk.Button(btn_frame, text="🔄 刷新路由列表(使用隧道ID)", command=self._thread_refresh_routes).pack(side=tk.LEFT, padx=6) row += 1 ttk.Label(frame, text="日志输出（可复制）:").grid(row=row, column=0, sticky="w") row += 1 self.log_text = tk.Text(frame, height=18, wrap="word") self.log_text.grid(row=row, column=0, columnspan=5, sticky="nsew", pady=(0, 5)) frame.rowconfigure(row, weight=1) frame.columnconfigure(1, weight=1) def log(self, msg): timestamp = datetime.now().strftime("[%Y-%m-%d %H:%M:%S]") text = f"{timestamp} {msg}\n" try: self.log_text.insert(tk.END, text) self.log_text.see(tk.END) self.root.update_idletasks() except Exception: # GUI 可能已关闭 pass def add_route(self): sub = self.subdomain_entry.get().strip() port = self.port_entry.get().strip() domain = self.domain.get().strip() if not sub or not port: self.log("❌ 子域名和端口号不能为空。") return if not port.isdigit(): self.log("❌ 端口号必须为数字。") return hostname = f"{sub}.{domain}" if domain else sub for r in self.routes: if r["hostname"] == hostname: self.log("⚠️ 该子域名已存在路由。") return self.routes.append({"hostname": hostname, "port": port}) self.tree.insert("", tk.END, values=(hostname, port)) self.subdomain_entry.delete(0, tk.END) self.port_entry.delete(0, tk.END) self.log(f"✅ 已添加路由：{hostname} -> localhost:{port}") def _thread_create_tunnel(self): threading.Thread(target=self.create_tunnel, daemon=True).start() def _thread_update_routes(self): threading.Thread(target=self.update_routes, daemon=True).start() def _thread_delete_tunnel(self): threading.Thread(target=self.delete_tunnel, daemon=True).start() def _thread_stop_tunnel(self): threading.Thread(target=self.stop_tunnel, daemon=True).start() def _thread_refresh_routes(self): threading.Thread(target=self.refresh_routes, daemon=True).start() def _thread_refresh_tunnel_list(self): threading.Thread(target=self.refresh_tunnel_list, daemon=True).start() def create_tunnel(self): api_token = self.api_token.get().strip() account_id = self.account_id.get().strip() tunnel_name = self.tunnel_name.get().strip() if not all([api_token, account_id, tunnel_name]): self.log("⚠️ 请填写完整 API Token、Account ID 和隧道名称。") return self._save_config() cf = CloudflareAPI(api_token) self.log("🔄 正在创建 Cloudflare 隧道...") result = cf.create_tunnel(account_id, tunnel_name) self.log(json.dumps(result, indent=2, ensure_ascii=False)) if result.get("success"): tunnel_id = result["result"]["id"] self.tunnel_id.delete(0, tk.END) self.tunnel_id.insert(0, tunnel_id) self.log(f"✅ 隧道创建成功，ID: {tunnel_id}") # 生成 cloudflared 文件并启动 self._generate_cloudflared_files(result["result"]) # 新建成功后刷新隧道列表（以便下拉中包含新建隧道） self._thread_refresh_tunnel_list() else: self.log("❌ 创建失败。") def _generate_cloudflared_files(self, tunnel_result): name = tunnel_result.get("name") tunnel_id = tunnel_result.get("id") creds = tunnel_result.get("credentials_file", {}) cred_path = os.path.join(CLOUDFLARED_DIR, f"{name}.json") config_path = os.path.join(CLOUDFLARED_DIR, "config.yaml") with open(cred_path, "w", encoding="utf-8") as f: json.dump(creds, f, indent=2) ingress = [] for r in self.routes: ingress.append({"hostname": r["hostname"], "service": f"http://localhost:{r['port']}"} ) ingress.append({"service": "http_status:404"}) # 将路径分隔符反斜杠用原始字符串写法替换，避免f-string反斜杠错误 cred_path_for_yaml = cred_path.replace("\\", "/") yaml_content = ( f"tunnel: {tunnel_id}\n" f"credentials-file: {cred_path_for_yaml}\n\n" "ingress:\n" ) for r in ingress: if "hostname" in r: yaml_content += f" - hostname: {r['hostname']}\n service: {r['service']}\n" else: yaml_content += f" - service: {r['service']}\n" with open(config_path, "w", encoding="utf-8") as f: f.write(yaml_content) self.log(f"✅ 已生成 cloudflared 配置文件: {config_path}") self._start_cloudflared(name) def _start_cloudflared(self, name): exe_path = "cloudflared.exe" if os.name == "nt" else "cloudflared" if not shutil.which(exe_path): self.log("⚠️ 未检测到 cloudflared，请确保已在系统路径或程序目录中。") return cmd = [exe_path, "tunnel", "--config", os.path.join(CLOUDFLARED_DIR, "config.yaml"), "run", name] self.log("🚀 正在启动 cloudflared 隧道...") if self.cloudflared_process and self.cloudflared_process.poll() is None: self.log("🛑 关闭已有 cloudflared 进程...") self.cloudflared_process.terminate() try: self.cloudflared_process.wait(timeout=10) self.log("✅ 旧进程已关闭。") except subprocess.TimeoutExpired: self.log("⚠️ 旧进程关闭超时，强制杀死。") self.cloudflared_process.kill() self.cloudflared_process.wait() self.log("✅ 旧进程已强制关闭。") try: self.cloudflared_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) self.log("✅ 隧道已启动，等待连接建立...") except Exception as e: self.log(f"❌ 启动 cloudflared 失败：{e}") def stop_tunnel(self): if self.cloudflared_process and self.cloudflared_process.poll() is None: self.log("🛑 正在停止本地 cloudflared 隧道进程...") self.cloudflared_process.terminate() try: self.cloudflared_process.wait(timeout=10) self.log("✅ 隧道进程已停止。") except subprocess.TimeoutExpired: self.log("⚠️ 进程停止超时，强制杀死进程。") self.cloudflared_process.kill() self.cloudflared_process.wait() self.log("✅ 隧道进程已强制停止。") self.cloudflared_process = None else: self.log("ℹ️ 当前无运行的 cloudflared 隧道进程。") def update_routes(self): api_token = self.api_token.get().strip() account_id = self.account_id.get().strip() tunnel_id = self.tunnel_id.get().strip() if not all([api_token, account_id, tunnel_id]): self.log("⚠️ 请填写完整 API Token、Account ID 和隧道 ID。") return cf = CloudflareAPI(api_token) self._save_config() self.log("🔄 正在更新路由配置...") ingress = [] for r in self.routes: ingress.append({"hostname": r["hostname"], "service": f"http://localhost:{r['port']}"} ) ingress.append({"service": "http_status:404"}) res = cf.update_config(account_id, tunnel_id, ingress) self.log(json.dumps(res, indent=2, ensure_ascii=False)) if res.get("success"): self.log("✅ 路由已成功更新。") else: self.log("❌ 路由更新失败。") def delete_tunnel(self): api_token = self.api_token.get().strip() account_id = self.account_id.get().strip() tunnel_id = self.tunnel_id.get().strip() if not all([api_token, account_id, tunnel_id]): return self.log("⚠️ 请填写完整 API Token、Account ID 和隧道 ID 后再删除隧道。") if not messagebox.askyesno("确认删除", f"确定要删除隧道ID: {tunnel_id} 吗？此操作不可恢复！"): self.log("⚠️ 用户取消了隧道删除操作。") return self.stop_tunnel() exe_path = "cloudflared.exe" if os.name == "nt" else "cloudflared" cleanup_cmd = [exe_path, "tunnel", "cleanup", tunnel_id] self.log(f"🧹 正在清理隧道 {tunnel_id} 的活动连接...") try: result = subprocess.run(cleanup_cmd, capture_output=True, text=True, timeout=15) self.log(f"🧹 cleanup 输出：{result.stdout.strip()}") if result.returncode != 0: self.log(f"⚠️ cleanup 命令执行失败：{result.stderr.strip()}") except Exception as e: self.log(f"❌ cleanup 命令异常：{e}") cf = CloudflareAPI(api_token) url = f"{CLOUDFLARE_API_BASE}/accounts/{account_id}/cfd_tunnel/{tunnel_id}" self.log(f"🗑 正在删除隧道 {tunnel_id} ...") try: r = requests.delete(url, headers=cf.headers, timeout=20) if r.status_code == 200: self.log(f"✅ 隧道删除成功！") self.tunnel_id.delete(0, tk.END) self.routes.clear() for item in self.tree.get_children(): self.tree.delete(item) self._save_config() # 删除成功后刷新隧道列表 self._thread_refresh_tunnel_list() else: self.log(f"❌ 删除失败，HTTP状态码: {r.status_code}，内容: {r.text}") except Exception as e: self.log(f"❌ 删除隧道异常: {e}") def refresh_routes(self): api_token = self.api_token.get().strip() account_id = self.account_id.get().strip() tunnel_id = self.tunnel_id.get().strip() if not all([api_token, account_id, tunnel_id]): self.log("⚠️ 请填写完整 API Token、Account ID 和隧道 ID。") return cf = CloudflareAPI(api_token) self.log("🔄 正在刷新路由列表...") routes = cf.get_routes(account_id, tunnel_id) if routes is None: self.log("❌ 获取路由时出现错误。") return if not routes: self.log("ℹ️ 无法获取路由，可能隧道不存在或无路由。") return # 将 service 字符串的端口解析出来（尽量兼容） parsed = [] for r in routes: service = r.get("service", "") port = "" # 可能形如 "http://localhost:8080" try: if ":" in service: port = service.split(":")[-1].strip() else: port = "" except Exception: port = "" parsed.append({"hostname": r.get("hostname", ""), "port": port}) self.routes = parsed self.tree.delete(*self.tree.get_children()) for r in self.routes: self.tree.insert("", tk.END, values=(r["hostname"], r["port"])) self.log(f"✅ 已刷新 {len(self.routes)} 条路由。") def refresh_routes_for_selected(self): # 根据下拉选中项获取 tunnel_id 并刷新其路由 sel = self.tunnel_combo.current() if sel < 0 or sel >= len(self.tunnel_list): self.log("⚠️ 请先在下拉中选择一个隧道。") return t = self.tunnel_list[sel] self.tunnel_id.delete(0, tk.END) self.tunnel_id.insert(0, t.get("id", "")) # 填写隧道名称到输入框（可选） if t.get("name"): self.tunnel_name.delete(0, tk.END) self.tunnel_name.insert(0, t.get("name")) # 刷新隧道路由 self._thread_refresh_routes() def delete_selected_route(self): selected = self.tree.selection() if not selected: self.log("⚠️ 请先选择要删除的路由。") return for item in selected: values = self.tree.item(item, "values") self.tree.delete(item) self.routes = [r for r in self.routes if r["hostname"] != values[0]] self.log(f"✅ 已删除路由：{values[0]}") # 异步调用更新远端路由配置 threading.Thread(target=self.update_routes, daemon=True).start() def modify_selected_route(self): selected = self.tree.selection() if not selected or len(selected) != 1: self.log("⚠️ 请选中且仅选中一个路由进行修改。") return item = selected[0] old_values = self.tree.item(item, "values") def save_modification(): new_sub = sub_entry.get().strip() new_port = port_entry.get().strip() domain = self.domain.get().strip() if not new_sub or not new_port or not new_port.isdigit(): messagebox.showerror("错误", "子域名不能为空，端口必须是数字。") return new_host = f"{new_sub}.{domain}" if domain else new_sub # 检查冲突 for r in self.routes: if r["hostname"] == new_host and new_host != old_values[0]: messagebox.showerror("错误", "该子域名已存在。") return # 更新数据 for r in self.routes: if r["hostname"] == old_values[0]: r["hostname"] = new_host r["port"] = new_port break self.tree.item(item, values=(new_host, new_port)) edit_win.destroy() self.log(f"✏️ 修改路由：{old_values[0]} -> {new_host}, 端口：{old_values[1]} -> {new_port}") edit_win = tk.Toplevel(self.root) edit_win.title("修改路由") tk.Label(edit_win, text="子域名:").grid(row=0, column=0, padx=5, pady=5) sub_entry = ttk.Entry(edit_win, width=25) sub_entry.grid(row=0, column=1, padx=5, pady=5) if old_values[0].endswith(self.domain.get()): sub_entry.insert(0, old_values[0].replace("." + self.domain.get(), "")) else: # 若 domain 未设置或不匹配，则插入整hostname sub_entry.insert(0, old_values[0]) tk.Label(edit_win, text="端口号:").grid(row=1, column=0, padx=5, pady=5) port_entry = ttk.Entry(edit_win, width=10) port_entry.grid(row=1, column=1, padx=5, pady=5) port_entry.insert(0, old_values[1]) ttk.Button(edit_win, text="保存修改", command=save_modification).grid(row=2, column=0, columnspan=2, pady=10) def _save_config(self): data = { "api_token": self.api_token.get().strip(), "account_id": self.account_id.get().strip(), "zone_id": self.zone_id.get().strip(), "tunnel_id": self.tunnel_id.get().strip(), "domain": self.domain.get().strip(), "tunnel_name": self.tunnel_name.get().strip(), "routes": self.routes } try: with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) self.log("💾 配置已保存。") except Exception as e: self.log(f"❌ 保存配置失败：{e}") def _load_config(self): if not os.path.exists(CONFIG_FILE): return try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: data = json.load(f) self.api_token.delete(0, tk.END) self.api_token.insert(0, data.get("api_token", "")) self.account_id.delete(0, tk.END) self.account_id.insert(0, data.get("account_id", "")) self.zone_id.delete(0, tk.END) self.zone_id.insert(0, data.get("zone_id", "")) self.tunnel_id.delete(0, tk.END) self.tunnel_id.insert(0, data.get("tunnel_id", "")) self.domain.delete(0, tk.END) self.domain.insert(0, data.get("domain", "")) self.tunnel_name.delete(0, tk.END) self.tunnel_name.insert(0, data.get("tunnel_name", "")) self.routes = data.get("routes", []) for r in self.routes: self.tree.insert("", tk.END, values=(r["hostname"], r["port"])) self.log("✅ 配置已加载。") # 加载后尝试刷新隧道列表（如果有账号和 token） if data.get("api_token") and data.get("account_id"): self._thread_refresh_tunnel_list() except Exception as e: self.log(f"❌ 加载配置失败：{e}") def _check_ssl(self): # 简单检查 SSL 支持，防止 requests 证书错误 try: r = requests.get("https://www.google.com", timeout=5) if r.status_code == 200: self.log("✅ SSL 支持正常。") else: self.log("⚠️ SSL 检查异常。") except Exception as e: self.log(f"⚠️ SSL 检查失败: {e}") # ============ 隧道列表与选择处理 ============ def refresh_tunnel_list(self): api_token = self.api_token.get().strip() account_id = self.account_id.get().strip() if not api_token or not account_id: self.log("⚠️ 刷新隧道前请填写 API Token 与 Account ID。") return cf = CloudflareAPI(api_token) self.log("🔄 正在获取隧道列表...") tunnels = cf.list_tunnels(account_id) if not isinstance(tunnels, list): self.log(f"❌ 获取隧道列表失败：{tunnels}") return self.tunnel_list = tunnels # 构造显示文本：名称 (ID) — 简要信息 combo_items = [] for t in tunnels: name = t.get("name", "") tid = t.get("id", "") created = t.get("created_at", "") combo_items.append(f"{name} ({tid})") # 更新 UI（主线程） def update_combo(): self.tunnel_combo['values'] = combo_items if combo_items: self.tunnel_combo.current(0) self.log(f"✅ 获取到 {len(combo_items)} 个隧道。") self.root.after(0, update_combo) def on_tunnel_selected(self): sel = self.tunnel_combo.current() if sel < 0 or sel >= len(self.tunnel_list): return t = self.tunnel_list[sel] tid = t.get("id", "") name = t.get("name", "") self.tunnel_id.delete(0, tk.END) self.tunnel_id.insert(0, tid) if name: self.tunnel_name.delete(0, tk.END) self.tunnel_name.insert(0, name) self.log(f"ℹ️ 已选隧道：{name} ({tid}) — 可点击'获取选中隧道路由'载入该隧道路由。") # ============ 入口 ============ if __name__ == "__main__": root = tk.Tk() app = TunnelApp(root) root.mainloop()