""" First GUI — PySide6 版 Author: Spade-sec | https://github.com/Spade-sec/First """ import asyncio import json import multiprocessing import os import queue import sys import threading from datetime import datetime from PySide6.QtCore import ( Qt, QTimer, QPropertyAnimation, QEasingCurve, Property, QRect, Signal, QPoint, QUrl, ) from PySide6.QtGui import QPainter, QColor, QFont, QIcon, QPixmap, QDesktopServices from PySide6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QFrame, QPushButton, QScrollArea, QTextEdit, QTreeWidget, QTreeWidgetItem, QProgressBar, QStackedWidget, QMenu, QHeaderView, QAbstractItemView, ) from src.cli import CliOptions, DEBUG_PORT, CDP_PORT from src.logger import Logger from src.engine import DebugEngine from src.navigator import MiniProgramNavigator from src.cloud_audit import CloudAuditor from src.userscript import load_userscripts_by_files from src.js_analyzer import analyze_js, merge_results, CATEGORY_INFO, save_report, load_reports, delete_report # ══════════════════════════════════════════ # 配色 # ══════════════════════════════════════════ _D = dict( bg="#1c1c24", card="#262632", input="#181820", sidebar="#111118", sb_hover="#1c1c28", sb_active="#222232", border="#303040", border2="#3a3a4c", text1="#e8e8f0", text2="#8888a0", text3="#5c5c6c", text4="#3c3c4c", accent="#4ade80", accent2="#22c55e", success="#4ade80", error="#f87171", warning="#fbbf24", ) _L = dict( bg="#f2f2f6", card="#ffffff", input="#eeeef2", sidebar="#ffffff", sb_hover="#f2f2f6", sb_active="#e6e6ea", border="#d8d8dc", border2="#c8c8cc", text1="#1a1a22", text2="#6e6e78", text3="#9e9ea8", text4="#c0c0c8", accent="#16a34a", accent2="#15803d", success="#16a34a", error="#dc2626", warning="#ca8a04", ) _TH = {"dark": _D, "light": _L} _FN = "Microsoft YaHei UI" _FM = "Consolas" _MENU = [ ("control", "\u25c9", "控制台"), ("navigator", "\u2b21", "路由导航"), ("hook", "\u25c8", "Hook"), ("cloud", "\u2601", "云扫描"), ("security", "\u2623", "敏感提取"), ("logs", "\u2261", "运行日志"), ] # ══════════════════════════════════════════ # 配置持久化 # ══════════════════════════════════════════ _BASE_DIR = os.path.dirname(os.path.abspath(__file__)) if getattr(sys, 'frozen', False): _BASE_DIR = os.path.dirname(sys.executable) _CFG_FILE = os.path.join(_BASE_DIR, "gui_config.json") os.makedirs(os.path.join(_BASE_DIR, "userscripts"), exist_ok=True) os.makedirs(os.path.join(_BASE_DIR, "hook_scripts"), exist_ok=True) os.makedirs(os.path.join(_BASE_DIR, "scan_reports"), exist_ok=True) def _load_cfg(): try: with open(_CFG_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: return {} def _save_cfg(data): try: with open(_CFG_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception: pass # ══════════════════════════════════════════ # QSS 主题 # ══════════════════════════════════════════ def build_qss(tn): c = _TH[tn] sel_bg = "#1e3a2a" if tn == "dark" else "#d4edda" sel_fg = "#a0f0c0" if tn == "dark" else "#155724" hdr_bg = "#222230" if tn == "dark" else "#e8e8ec" row_bg = c["input"] return f""" /* ── 全局 ── */ QMainWindow, QWidget#central {{ background: {c['bg']}; }} /* ── 侧栏 ── */ QFrame#sidebar {{ background: {c['sidebar']}; }} QFrame#sidebar QLabel {{ background: transparent; }} QFrame#sb_head {{ background: {c['sidebar']}; }} QLabel#sb_logo {{ color: {c['text1']}; font-size: 13px; font-weight: bold; background: transparent; }} QFrame#sb_hline {{ background: {c['border']}; max-height: 1px; min-height: 1px; }} QLabel#sb_theme {{ color: {c['text3']}; background: transparent; padding: 4px 12px; }} QLabel#sb_theme:hover {{ color: {c['text1']}; }} /* ── 菜单项 ── */ QFrame.sb_item {{ background: {c['sidebar']}; border-radius: 8px; padding: 8px 10px; }} QFrame.sb_item:hover {{ background: {c['sb_hover']}; }} QFrame.sb_item_active {{ background: {c['sb_active']}; border-radius: 8px; padding: 8px 10px; }} QFrame.sb_item QLabel.sb_icon {{ color: {c['text3']}; background: transparent; }} QFrame.sb_item QLabel.sb_name {{ color: {c['text2']}; background: transparent; }} QFrame.sb_item_active QLabel.sb_icon {{ color: {c['accent']}; background: transparent; }} QFrame.sb_item_active QLabel.sb_name {{ color: {c['text1']}; background: transparent; }} /* ── 分割线 ── */ QFrame#vline {{ background: {c['border']}; max-width: 1px; min-width: 1px; }} QFrame#hdr_line {{ background: {c['border']}; max-height: 1px; min-height: 1px; }} /* ── 标题 ── */ QLabel#page_title {{ color: {c['text1']}; font-size: 17px; font-weight: bold; padding-left: 24px; background: transparent; }} /* ── 圆角卡片 ── */ QFrame.card {{ background: {c['card']}; border-radius: 12px; border: none; }} QFrame.card QLabel {{ background: transparent; }} QFrame.card QLabel.title {{ color: {c['text1']}; font-weight: bold; font-size: 11px; }} QFrame.card QLabel.subtitle {{ color: {c['text2']}; font-size: 9px; }} /* ── 通用 Label ── */ QLabel {{ color: {c['text2']}; background: transparent; }} QLabel.bold {{ color: {c['text1']}; font-weight: bold; }} QLabel.muted {{ color: {c['text3']}; }} QLabel.accent {{ color: {c['accent']}; }} /* ── 按钮 ── */ QPushButton {{ background: {c['accent']}; color: #111118; border: none; border-radius: 8px; padding: 5px 16px; font-size: 10px; }} QPushButton:hover {{ background: {c['accent2']}; }} QPushButton:disabled {{ background: {"#1a3a2a" if tn == "dark" else "#b0dfc0"}; color: {"#3a6a4a" if tn == "dark" else "#5a8a6a"}; }} /* ── 输入框 ── */ QLineEdit {{ background: {c['input']}; color: {c['text1']}; border: none; border-radius: 10px; padding: 6px 12px; font-size: 10px; selection-background-color: {c['accent']}; selection-color: #111118; }} QLineEdit:focus {{ border: 1px solid {c['accent']}; }} /* ── 文本框 ── */ QTextEdit {{ background: {c['input']}; color: {c['accent']}; border: none; border-radius: 8px; padding: 10px 14px; font-family: {_FM}; font-size: 10px; selection-background-color: {c['accent']}; selection-color: #111118; }} /* ── 树形控件 ── */ QTreeWidget {{ background: {c['card']}; color: {c['text2']}; border: none; font-size: 10px; outline: 0; }} QTreeWidget::item {{ padding: 4px 8px; border: none; text-align: left; }} QTreeWidget::item:selected {{ background: {sel_bg}; color: {sel_fg}; }} QTreeWidget::item:hover {{ background: {c['sb_hover']}; }} QHeaderView::section {{ background: {hdr_bg}; color: {c['text1']}; border: none; padding: 4px 8px; font-weight: bold; font-size: 10px; text-align: left; }} /* ── 进度条 ── */ QProgressBar {{ background: {c['border']}; border: none; border-radius: 4px; height: 6px; text-align: center; }} QProgressBar::chunk {{ background: {c['accent']}; border-radius: 4px; }} /* ── 滚动条 ── */ QScrollBar:vertical {{ background: transparent; width: 6px; margin: 0; }} QScrollBar::handle:vertical {{ background: {"#3a6a4a" if tn == "dark" else "#8fc4a0"}; border-radius: 3px; min-height: 20px; }} QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical {{ height: 0; }} QScrollBar::add-page:vertical, QScrollBar::sub-page:vertical {{ background: transparent; }} QScrollBar:horizontal {{ background: transparent; height: 6px; }} QScrollBar::handle:horizontal {{ background: {"#3a6a4a" if tn == "dark" else "#8fc4a0"}; border-radius: 3px; min-width: 20px; }} QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal {{ width: 0; }} QScrollBar::add-page:horizontal, QScrollBar::sub-page:horizontal {{ background: transparent; }} /* ── 滚动区域 ── */ QScrollArea {{ background: transparent; border: none; }} QScrollArea > QWidget > QWidget {{ background: transparent; }} /* ── 右键菜单 ── */ QMenu {{ background: {c['card']}; color: {c['text1']}; border: 1px solid {c['border']}; border-radius: 6px; padding: 4px; }} QMenu::item {{ padding: 6px 20px; border-radius: 4px; }} QMenu::item:selected {{ background: {c['accent']}; color: #ffffff; }} QMenu::separator {{ height: 1px; background: {c['border']}; margin: 4px 8px; }} /* ── Hook 行 ── */ QFrame.hook_row {{ background: {row_bg}; border-radius: 8px; }} QFrame.hook_row QLabel {{ background: transparent; }} QLabel.js_badge {{ background: {c['accent']}; color: {"#ffffff" if tn == "dark" else "#111118"}; font-weight: bold; font-size: 9px; padding: 2px 6px; border-radius: 4px; }} /* ── Completer popup ── */ QListView {{ background: {c['input']}; color: {c['text1']}; border: 1px solid {c['border']}; border-radius: 6px; outline: 0; }} QListView::item:selected {{ background: {c['accent']}; color: #111118; }} """ # ══════════════════════════════════════════ # 自定义控件 # ══════════════════════════════════════════ class ToggleSwitch(QWidget): toggled = Signal(bool) def __init__(self, checked=False, parent=None): super().__init__(parent) self._checked = checked self._thumb_pos = 1.0 if checked else 0.0 self._on_color = QColor("#4ade80") self._off_color = QColor("#3c3c4c") self._thumb_color = QColor("#ffffff") self.setFixedSize(44, 24) self.setCursor(Qt.PointingHandCursor) self._anim = QPropertyAnimation(self, b"thumbPos") self._anim.setDuration(150) self._anim.setEasingCurve(QEasingCurve.OutCubic) def isChecked(self): return self._checked def setChecked(self, v): if self._checked == v: return self._checked = v self._anim.stop() self._anim.setStartValue(self._thumb_pos) self._anim.setEndValue(1.0 if v else 0.0) self._anim.start() self.toggled.emit(v) def _get_thumb_pos(self): return self._thumb_pos def _set_thumb_pos(self, v): self._thumb_pos = v self.update() thumbPos = Property(float, _get_thumb_pos, _set_thumb_pos) def mousePressEvent(self, e): self.setChecked(not self._checked) def paintEvent(self, e): p = QPainter(self) p.setRenderHint(QPainter.Antialiasing) w, h = self.width(), self.height() r = h / 2 # track track_color = QColor(self._on_color) if self._checked else QColor(self._off_color) p.setPen(Qt.NoPen) p.setBrush(track_color) p.drawRoundedRect(0, 0, w, h, r, r) # thumb tr = r - 3 cx = r + self._thumb_pos * (w - 2 * r) p.setBrush(self._thumb_color) p.drawEllipse(QPoint(int(cx), int(r)), int(tr), int(tr)) def set_colors(self, on, off): self._on_color = QColor(on) self._off_color = QColor(off) self.update() class AnimatedStackedWidget(QStackedWidget): """Page switch with a lightweight vertical slide animation. Uses QPropertyAnimation on widget geometry instead of QGraphicsOpacityEffect, which forces expensive off-screen compositing of the entire subtree (causing visible lag on heavy pages like the cloud-scan QTreeWidget). """ def __init__(self, parent=None): super().__init__(parent) self._anim = None def setCurrentIndexAnimated(self, idx): if idx == self.currentIndex(): return old_idx = self.currentIndex() old_widget = self.currentWidget() new_widget = self.widget(idx) if new_widget is None: self.setCurrentIndex(idx) return # Determine slide direction: down when going forward, up when back h = self.height() offset = h // 4 # slide only a quarter of the height for subtlety start_y = offset if idx > old_idx else -offset # Immediately switch the page (no off-screen compositing) self.setCurrentIndex(idx) # Animate just the position of the new page final_rect = new_widget.geometry() start_rect = QRect(final_rect) start_rect.moveTop(final_rect.top() + start_y) anim = QPropertyAnimation(new_widget, b"geometry") anim.setDuration(150) anim.setStartValue(start_rect) anim.setEndValue(final_rect) anim.setEasingCurve(QEasingCurve.OutCubic) self._anim = anim # prevent GC anim.start() class StatusDot(QWidget): def __init__(self, parent=None): super().__init__(parent) self.setFixedSize(10, 10) self._color = QColor("#3c3c4c") def set_color(self, color): self._color = QColor(color) self.update() def paintEvent(self, e): p = QPainter(self) p.setRenderHint(QPainter.Antialiasing) p.setPen(Qt.NoPen) p.setBrush(self._color) p.drawEllipse(1, 1, 8, 8) # ══════════════════════════════════════════ # 辅助函数 # ══════════════════════════════════════════ def _make_card(): f = QFrame() f.setProperty("class", "card") return f def _make_label(text, bold=False, muted=False, mono=False): l = QLabel(text) if bold: l.setProperty("class", "bold") elif muted: l.setProperty("class", "muted") if mono: l.setFont(QFont(_FM, 10)) return l def _make_btn(text, callback=None): b = QPushButton(text) if callback: b.clicked.connect(callback) return b def _make_entry(placeholder="", width=None): e = QLineEdit() e.setPlaceholderText(placeholder) if width: e.setFixedWidth(width) return e # ══════════════════════════════════════════ # 敏感提取 — 独立进程 worker # ══════════════════════════════════════════ def _sec_analyze_one(src): """独立进程 Pool 的 worker 函数：分析单个 JS 源码。""" try: return analyze_js(src) except Exception: return None def _sec_worker_proc(js_sources, appid, base_dir, result_q, name=""): """在独立进程中并行分析 JS，通过 multiprocessing.Queue 回传进度和结果。""" try: total = len(js_sources) total_size = sum(len(s) for s in js_sources) if total == 0: result_q.put(("done", {}, 0, 0)) return import os workers = min(total, max(4, os.cpu_count() or 4)) result_q.put(("progress", 35, f"并行分析 ({workers} 线程) ...")) from concurrent.futures import ThreadPoolExecutor, as_completed results = [] done_count = 0 with ThreadPoolExecutor(max_workers=workers) as pool: futs = {pool.submit(_sec_analyze_one, src): i for i, src in enumerate(js_sources)} for fut in as_completed(futs): r = fut.result() if r: results.append(r) done_count += 1 if done_count % 5 == 0 or done_count == total: pct = 30 + int(done_count / total * 60) result_q.put(("progress", pct, f"分析中 {done_count}/{total} ...")) merged = merge_results(results) if results else {} result_q.put(("progress", 95, "生成报告...")) try: save_report(base_dir, appid, merged, total, total_size, name=name) except Exception: pass result_q.put(("done", merged, total, total_size)) except Exception as e: result_q.put(("error", f"分析进程异常: {e}")) result_q.put(("done", None, 0, 0)) def _sec_run_worker(js_sources, appid, base_dir, sec_q, name=""): """启动独立分析进程，用后台线程转发 multiprocessing.Queue 到 UI 的 queue.Queue。""" mp_q = multiprocessing.Queue() proc = multiprocessing.Process( target=_sec_worker_proc, args=(js_sources, appid, base_dir, mp_q, name), daemon=True, ) proc.start() # 后台线程：等待进程完成，转发消息到 UI 队列 def _relay(): while proc.is_alive() or not mp_q.empty(): try: item = mp_q.get(timeout=0.2) sec_q.put(item) except Exception: pass # 确保队列清空 while not mp_q.empty(): try: sec_q.put(mp_q.get_nowait()) except Exception: break t = threading.Thread(target=_relay, daemon=True) t.start() # ══════════════════════════════════════════ # 主窗口 # ══════════════════════════════════════════ class App(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("First") _ico = os.path.join(_BASE_DIR, "icon.ico") if os.path.exists(_ico): self.setWindowIcon(QIcon(_ico)) self.resize(960, 620) self.setMinimumSize(780, 500) self._cfg = _load_cfg() self._tn = self._cfg.get("theme", "dark") self._pg = "control" self._running = False self._loop = self._loop_th = self._engine = self._navigator = self._auditor = None self._cloud_call_history = {} self._cloud_all_items = [] self._cloud_row_results = {} self._cancel_ev = None self._route_poll_id = None self._all_routes = [] self._cloud_scan_active = False self._cloud_scan_poll_timer = None self._redirect_guard_on = False self._hook_injected = set() self._blocked_seen = 0 self._sec_scanning = False self._sec_result = None self._sec_view = "scan" # "scan" | "report" | "history" self._log_q = queue.Queue() self._sts_q = queue.Queue() self._rte_q = queue.Queue() self._cld_q = queue.Queue() self._sec_q = queue.Queue() self._selected_preload = list(self._cfg.get("selected_preload_scripts", [])) self._nav_route_idx = -1 self._sb_items = {} self._page_map = {} self._build() self.setStyleSheet(build_qss(self._tn)) self._show("control") self._tick_timer = QTimer() self._tick_timer.timeout.connect(self._tick) self._tick_timer.start(80) # ────────────────────────────────── # 布局 # ────────────────────────────────── def _build(self): central = QWidget() central.setObjectName("central") self.setCentralWidget(central) root_h = QHBoxLayout(central) root_h.setContentsMargins(0, 0, 0, 0) root_h.setSpacing(0) # ── 侧栏 ── self._sb = QFrame() self._sb.setObjectName("sidebar") self._sb.setFixedWidth(180) sb_lay = QVBoxLayout(self._sb) sb_lay.setContentsMargins(0, 0, 0, 0) sb_lay.setSpacing(0) sb_head = QFrame() sb_head.setObjectName("sb_head") sb_head.setFixedHeight(76) sb_head_lay = QVBoxLayout(sb_head) sb_head_lay.addStretch() logo_row = QHBoxLayout() logo_row.setContentsMargins(0, 0, 0, 0) logo_row.setSpacing(6) logo_row.addStretch() _ico_path = os.path.join(_BASE_DIR, "icon.ico") if os.path.exists(_ico_path): _ico_lbl = QLabel() _pix = QPixmap(_ico_path).scaled(20, 20, Qt.KeepAspectRatio, Qt.SmoothTransformation) _ico_lbl.setPixmap(_pix) _ico_lbl.setFixedSize(20, 20) logo_row.addWidget(_ico_lbl) self._sb_logo = QLabel("First") self._sb_logo.setObjectName("sb_logo") logo_row.addWidget(self._sb_logo) logo_row.addStretch() sb_head_lay.addLayout(logo_row) sb_head_lay.addStretch() sb_lay.addWidget(sb_head) hline = QFrame() hline.setObjectName("sb_hline") hline.setFixedHeight(1) sb_lay.addWidget(hline, 0, Qt.AlignTop) sb_nav = QWidget() sb_nav_lay = QVBoxLayout(sb_nav) sb_nav_lay.setContentsMargins(8, 10, 8, 10) sb_nav_lay.setSpacing(2) for pid, icon, name in _MENU: row = QFrame() row.setCursor(Qt.PointingHandCursor) row.setProperty("class", "sb_item") row_lay = QHBoxLayout(row) row_lay.setContentsMargins(10, 0, 8, 0) row_lay.setSpacing(6) ic = QLabel(icon) ic.setProperty("class", "sb_icon") ic.setFont(QFont(_FN, 13)) nm = QLabel(name) nm.setProperty("class", "sb_name") nm.setFont(QFont(_FN, 10)) row_lay.addWidget(ic) row_lay.addWidget(nm, 1) sb_nav_lay.addWidget(row) row.mousePressEvent = lambda e, p=pid: self._show(p) self._sb_items[pid] = (row, ic, nm) sb_nav_lay.addStretch() sb_lay.addWidget(sb_nav, 1) self._sb_theme = QLabel() self._sb_theme.setObjectName("sb_theme") self._sb_theme.setAlignment(Qt.AlignCenter) self._sb_theme.setCursor(Qt.PointingHandCursor) self._sb_theme.setFont(QFont(_FN, 9)) self._sb_theme.mousePressEvent = lambda e: self._toggle_theme() sb_lay.addWidget(self._sb_theme) sb_author = QLabel("by vs-olitus") sb_author.setObjectName("sb_theme") sb_author.setAlignment(Qt.AlignCenter) sb_author.setFont(QFont(_FN, 8)) sb_lay.addWidget(sb_author) sb_gh = QLabel("github.com/Spade-sec/First") sb_gh.setObjectName("sb_theme") sb_gh.setAlignment(Qt.AlignCenter) sb_gh.setFont(QFont(_FN, 7)) sb_gh.setCursor(Qt.PointingHandCursor) sb_gh.mousePressEvent = lambda e: ( QDesktopServices.openUrl(QUrl("https://github.com/Spade-sec/First")), self._log_add("info", "[gui] 已打开 GitHub 页面")) sb_lay.addWidget(sb_gh) sb_lay.addSpacing(12) self._update_theme_label() root_h.addWidget(self._sb) vline = QFrame() vline.setObjectName("vline") vline.setFixedWidth(1) root_h.addWidget(vline) # ── 右侧 ── right = QWidget() right_lay = QVBoxLayout(right) right_lay.setContentsMargins(0, 0, 0, 0) right_lay.setSpacing(0) hdr_frame = QWidget() hdr_frame.setFixedHeight(60) hdr_lay = QHBoxLayout(hdr_frame) hdr_lay.setContentsMargins(0, 0, 0, 0) self._hdr_title = QLabel("") self._hdr_title.setObjectName("page_title") hdr_lay.addWidget(self._hdr_title) hdr_lay.addStretch() right_lay.addWidget(hdr_frame) hdr_line = QFrame() hdr_line.setObjectName("hdr_line") hdr_line.setFixedHeight(1) right_lay.addWidget(hdr_line) self._stack = AnimatedStackedWidget() right_lay.addWidget(self._stack, 1) root_h.addWidget(right, 1) self._build_control() self._build_navigator() self._build_hook() self._build_cloud() self._build_security() self._build_logs() # ── 控制台 ── def _build_control(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 8, 24, 8) lay.setSpacing(6) lay.setAlignment(Qt.AlignTop) # Card 1: 连接设置 c1 = _make_card() c1_lay = QVBoxLayout(c1) c1_lay.setContentsMargins(16, 10, 16, 10) c1_lay.setSpacing(6) c1_lay.addWidget(_make_label("连接设置", bold=True)) row1 = QHBoxLayout() row1.addWidget(QLabel("调试端口")) self._dp_ent = _make_entry(width=100) self._dp_ent.setText(str(self._cfg.get("debug_port", DEBUG_PORT))) self._dp_ent.textChanged.connect(lambda: self._auto_save()) row1.addWidget(self._dp_ent) row1.addSpacing(20) row1.addWidget(QLabel("CDP 端口")) self._cp_ent = _make_entry(width=100) self._cp_ent.setText(str(self._cfg.get("cdp_port", CDP_PORT))) self._cp_ent.textChanged.connect(lambda: self._auto_save()) row1.addWidget(self._cp_ent) row1.addStretch() c1_lay.addLayout(row1) c1_lay.addWidget(_make_label("调试选项", bold=True)) chkr = QHBoxLayout() self._tog_dm = ToggleSwitch(self._cfg.get("debug_main", False)) self._tog_dm.toggled.connect(lambda v: self._auto_save()) chkr.addWidget(self._tog_dm) chkr.addWidget(QLabel("调试主包")) chkr.addSpacing(24) self._tog_df = ToggleSwitch(self._cfg.get("debug_frida", False)) self._tog_df.toggled.connect(lambda v: self._auto_save()) chkr.addWidget(self._tog_df) chkr.addWidget(QLabel("调试 Frida")) chkr.addStretch() c1_lay.addLayout(chkr) lay.addWidget(c1) # Card 2: 前加载脚本 c2 = _make_card() c2_lay = QVBoxLayout(c2) c2_lay.setContentsMargins(16, 10, 16, 10) c2_lay.setSpacing(4) hdr_row = QHBoxLayout() hdr_row.addWidget(_make_label("前加载脚本", bold=True)) hdr_row.addWidget(_make_label("(启动调试前可用)", muted=True)) hdr_row.addStretch() self._btn_preload_refresh = _make_btn("刷新", self._preload_refresh) hdr_row.addWidget(self._btn_preload_refresh) c2_lay.addLayout(hdr_row) self._preload_container = QVBoxLayout() self._preload_container.setSpacing(2) c2_lay.addLayout(self._preload_container) lay.addWidget(c2) self._preload_refresh() # Action row ar = QHBoxLayout() self._btn_start = _make_btn("▶ 启动调试", self._do_start) self._btn_start.setFont(QFont(_FN, 10, QFont.Bold)) ar.addWidget(self._btn_start) self._btn_stop = _make_btn("■ 停止", self._do_stop) self._btn_stop.setFont(QFont(_FN, 10, QFont.Bold)) self._btn_stop.setEnabled(False) ar.addWidget(self._btn_stop) ar.addStretch() lay.addLayout(ar) # DevTools URL dt_row = QHBoxLayout() self._devtools_lbl = QLabel("") self._devtools_lbl.setProperty("class", "accent") self._devtools_lbl.setFont(QFont(_FM, 8)) self._devtools_lbl.setCursor(Qt.PointingHandCursor) self._devtools_lbl.mousePressEvent = lambda e: self._copy_devtools_url() dt_row.addWidget(self._devtools_lbl) self._devtools_copy_hint = QLabel("") self._devtools_copy_hint.setProperty("class", "muted") self._devtools_copy_hint.setFont(QFont(_FN, 8)) dt_row.addWidget(self._devtools_copy_hint) dt_row.addStretch() lay.addLayout(dt_row) # Card 3: 运行状态 c3 = _make_card() c3_lay = QVBoxLayout(c3) c3_lay.setContentsMargins(16, 10, 16, 10) c3_lay.setSpacing(2) c3_lay.addWidget(_make_label("运行状态", bold=True)) self._dots = {} for key, name in [("frida", "Frida"), ("miniapp", "小程序"), ("devtools", "DevTools")]: dr = QHBoxLayout() dot = StatusDot() dr.addWidget(dot) lb = QLabel(f"{name}: 未连接") dr.addWidget(lb) dr.addStretch() c3_lay.addLayout(dr) self._dots[key] = (dot, lb, name) self._app_lbl = QLabel("应用: --") self._app_lbl.setProperty("class", "muted") c3_lay.addWidget(self._app_lbl) lay.addWidget(c3) self._stack.addWidget(page) self._page_map["control"] = self._stack.count() - 1 # ── 路由导航 ── def _build_navigator(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 12, 24, 16) lay.setSpacing(10) # 搜索栏 sf = QHBoxLayout() sf.addWidget(QLabel("搜索")) self._srch_ent = _make_entry("输入路由关键字搜索...") self._srch_ent.textChanged.connect(self._do_filter) sf.addWidget(self._srch_ent, 1) lay.addLayout(sf) # 路由树 tc = _make_card() tc_lay = QVBoxLayout(tc) tc_lay.setContentsMargins(0, 0, 0, 0) self._tree = QTreeWidget() self._tree.setHeaderHidden(True) self._tree.setSelectionMode(QAbstractItemView.SingleSelection) self._tree.setContextMenuPolicy(Qt.CustomContextMenu) self._tree.customContextMenuRequested.connect(self._nav_context_menu) tc_lay.addWidget(self._tree) lay.addWidget(tc, 1) # 手动输入跳转 mi = QHBoxLayout() mi.addWidget(QLabel("手动跳转")) self._nav_input = _make_entry("输入路由路径，回车跳转...") self._nav_input.returnPressed.connect(self._do_manual_go) mi.addWidget(self._nav_input, 1) self._btn_manual_go = _make_btn("跳转", self._do_manual_go) mi.addWidget(self._btn_manual_go) self._btn_copy_route = _make_btn("复制路由", self._do_copy_route) self._btn_copy_route.setEnabled(False) mi.addWidget(self._btn_copy_route) lay.addLayout(mi) # 导航按钮行 1 b1 = QHBoxLayout() self._btn_go = _make_btn("跳转", self._do_go) self._btn_go.setEnabled(False) b1.addWidget(self._btn_go) self._btn_redir = _make_btn("重定向", self._do_redir) self._btn_redir.setEnabled(False) b1.addWidget(self._btn_redir) self._btn_relaunch = _make_btn("重启到页面", self._do_relaunch) self._btn_relaunch.setEnabled(False) b1.addWidget(self._btn_relaunch) self._btn_back = _make_btn("返回上页", self._do_back) self._btn_back.setEnabled(False) b1.addWidget(self._btn_back) b1.addStretch() self._btn_fetch = _make_btn("获取路由", self._do_fetch) self._btn_fetch.setEnabled(False) b1.addWidget(self._btn_fetch) lay.addLayout(b1) # 导航按钮行 2: 上一个/下一个 + 遍历 + 防跳转 b2 = QHBoxLayout() self._btn_prev = _make_btn("◀ 上一个", self._do_prev) self._btn_prev.setEnabled(False) b2.addWidget(self._btn_prev) self._btn_next = _make_btn("下一个 ▶", self._do_next) self._btn_next.setEnabled(False) b2.addWidget(self._btn_next) b2.addSpacing(12) self._btn_auto = _make_btn("自动遍历", self._do_autovis) self._btn_auto.setEnabled(False) b2.addWidget(self._btn_auto) self._btn_autostop = _make_btn("停止遍历", self._do_autostop) self._btn_autostop.setEnabled(False) b2.addWidget(self._btn_autostop) b2.addSpacing(12) self._guard_switch = ToggleSwitch(False) self._guard_switch.setFixedSize(36, 18) self._guard_switch.setEnabled(False) self._guard_switch.toggled.connect(self._do_toggle_guard_switch) b2.addWidget(self._guard_switch) self._guard_label = QLabel("防跳转: 关闭") b2.addWidget(self._guard_label) b2.addStretch() lay.addLayout(b2) self._prog = QProgressBar() self._prog.setMaximum(100) self._prog.setValue(0) self._prog.setTextVisible(False) self._prog.setFixedHeight(6) lay.addWidget(self._prog) self._route_lbl = QLabel("当前路由: --") lay.addWidget(self._route_lbl) self._stack.addWidget(page) self._page_map["navigator"] = self._stack.count() - 1 # ── Hook 页面 ── def _build_hook(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 12, 24, 16) lay.setSpacing(10) tip_row = QHBoxLayout() self._hook_tip = QLabel("将 .js 文件放入 hook_scripts/ 目录，点击「注入」即时执行") self._hook_tip.setProperty("class", "muted") tip_row.addWidget(self._hook_tip) tip_row.addStretch() self._btn_hook_refresh = _make_btn("刷新列表", self._hook_refresh) tip_row.addWidget(self._btn_hook_refresh) lay.addLayout(tip_row) c1 = _make_card() c1_lay = QVBoxLayout(c1) c1_lay.setContentsMargins(12, 12, 12, 12) c1_lay.setSpacing(6) scroll = QScrollArea() scroll.setWidgetResizable(True) self._hook_inner = QWidget() self._hook_inner_lay = QVBoxLayout(self._hook_inner) self._hook_inner_lay.setContentsMargins(0, 0, 0, 0) self._hook_inner_lay.setSpacing(6) self._hook_inner_lay.addStretch() scroll.setWidget(self._hook_inner) c1_lay.addWidget(scroll) lay.addWidget(c1, 1) self._hook_status_lbls = {} self._hook_refresh() self._stack.addWidget(page) self._page_map["hook"] = self._stack.count() - 1 def _hook_refresh(self): while self._hook_inner_lay.count() > 1: item = self._hook_inner_lay.takeAt(0) w = item.widget() if w: w.deleteLater() self._hook_status_lbls = {} hook_dir = os.path.join(_BASE_DIR, "hook_scripts") js_files = sorted(f for f in os.listdir(hook_dir) if f.endswith(".js")) if os.path.isdir(hook_dir) else [] if not js_files: lbl = QLabel("hook_scripts/ 目录下无 .js 文件") lbl.setAlignment(Qt.AlignCenter) self._hook_inner_lay.insertWidget(0, lbl) return for fn in js_files: row = QFrame() row.setProperty("class", "hook_row") row.setFixedHeight(52) row_lay = QHBoxLayout(row) row_lay.setContentsMargins(12, 0, 12, 0) row_lay.setSpacing(8) icon_lbl = QLabel("JS") icon_lbl.setProperty("class", "js_badge") icon_lbl.setFont(QFont(_FM, 8, QFont.Bold)) icon_lbl.setFixedWidth(30) icon_lbl.setAlignment(Qt.AlignCenter) row_lay.addWidget(icon_lbl) name_lbl = QLabel(fn) name_lbl.setFont(QFont(_FN, 10)) row_lay.addWidget(name_lbl, 1) injected = fn in self._hook_injected status_lbl = QLabel("● 已注入" if injected else "○ 未注入") c = _TH[self._tn] status_lbl.setStyleSheet(f"color: {c['success'] if injected else c['text3']};") row_lay.addWidget(status_lbl) self._hook_status_lbls[fn] = status_lbl inject_btn = _make_btn("注入", lambda checked=False, f=fn: self._hook_inject(f)) row_lay.addWidget(inject_btn) clear_btn = _make_btn("清除", lambda checked=False, f=fn: self._hook_clear(f)) row_lay.addWidget(clear_btn) self._hook_inner_lay.insertWidget(self._hook_inner_lay.count() - 1, row) def _hook_inject(self, filename): if not self._engine or not self._loop or not self._loop.is_running(): self._log_add("error", "[Hook] 请先启动调试") return hook_dir = os.path.join(_BASE_DIR, "hook_scripts") filepath = os.path.join(hook_dir, filename) try: with open(filepath, "r", encoding="utf-8") as f: source = f.read() except Exception as e: self._log_add("error", f"[Hook] 读取文件失败: {e}") return asyncio.run_coroutine_threadsafe( self._ahook_inject(filename, source), self._loop) async def _ahook_inject(self, filename, source): try: await self._engine.evaluate_js(source, timeout=5.0) self._hook_injected.add(filename) self._log_q.put(("info", f"[Hook] 已注入: {filename}")) self._log_q.put(("__hook_status__", filename, True)) except Exception as e: self._log_q.put(("error", f"[Hook] 注入失败 {filename}: {e}")) def _hook_clear(self, filename): self._hook_injected.discard(filename) self._hook_update_status(filename, False) self._log_add("info", f"[Hook] 已清除标记: {filename}（注意: JS 注入后无法真正撤销，需刷新页面）") def _hook_update_status(self, filename, injected): c = _TH[self._tn] lbl = self._hook_status_lbls.get(filename) if lbl: lbl.setText("● 已注入" if injected else "○ 未注入") lbl.setStyleSheet(f"color: {c['success'] if injected else c['text3']};") # ── 云扫描 ── def _build_cloud(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 12, 24, 16) lay.setSpacing(10) ctrl = QHBoxLayout() self._btn_cloud_toggle = _make_btn("停止捕获", self._cloud_do_toggle) ctrl.addWidget(self._btn_cloud_toggle) self._btn_cloud_static = _make_btn("静态扫描", self._cloud_do_static_scan) ctrl.addWidget(self._btn_cloud_static) self._btn_cloud_clear = _make_btn("清空记录", self._cloud_do_clear) ctrl.addWidget(self._btn_cloud_clear) self._cloud_scan_lbl = QLabel("") ctrl.addWidget(self._cloud_scan_lbl) ctrl.addStretch() self._btn_cloud_export = _make_btn("导出报告", self._cloud_do_export) ctrl.addWidget(self._btn_cloud_export) lay.addLayout(ctrl) tc = _make_card() tc_lay = QVBoxLayout(tc) tc_lay.setContentsMargins(12, 8, 12, 8) tc_lay.setSpacing(4) title_row = QHBoxLayout() title_row.addWidget(_make_label("云函数捕获记录", bold=True)) self._cloud_env_lbl = QLabel("全局捕获（默认开启）") title_row.addWidget(self._cloud_env_lbl) title_row.addStretch() title_row.addWidget(QLabel("搜索")) self._cloud_search_ent = _make_entry(width=180) self._cloud_search_ent.textChanged.connect(self._cloud_filter) title_row.addWidget(self._cloud_search_ent) tc_lay.addLayout(title_row) self._cloud_tree = QTreeWidget() self._cloud_tree.setRootIsDecorated(False) self._cloud_tree.setIndentation(0) self._cloud_tree.setHeaderLabels(["AppID", "类型", "名称", "参数", "状态", "时间"]) header = self._cloud_tree.header() header.setDefaultAlignment(Qt.AlignLeft | Qt.AlignVCenter) header.setStretchLastSection(False) header.setSectionResizeMode(0, QHeaderView.Interactive) header.setSectionResizeMode(1, QHeaderView.Interactive) header.setSectionResizeMode(2, QHeaderView.Interactive) header.setSectionResizeMode(3, QHeaderView.Stretch) header.setSectionResizeMode(4, QHeaderView.Interactive) header.setSectionResizeMode(5, QHeaderView.Interactive) self._cloud_tree.setColumnWidth(0, 100) self._cloud_tree.setColumnWidth(1, 70) self._cloud_tree.setColumnWidth(2, 140) self._cloud_tree.setColumnWidth(4, 50) self._cloud_tree.setColumnWidth(5, 70) self._cloud_tree.setSelectionMode(QAbstractItemView.SingleSelection) self._cloud_tree.itemClicked.connect(self._cloud_on_select) self._cloud_tree.setContextMenuPolicy(Qt.CustomContextMenu) self._cloud_tree.customContextMenuRequested.connect(self._cloud_tree_context_menu) tc_lay.addWidget(self._cloud_tree) lay.addWidget(tc, 1) call_row = QHBoxLayout() call_row.addWidget(QLabel("手动调用")) self._cloud_name_ent = _make_entry(width=140) call_row.addWidget(self._cloud_name_ent) call_row.addWidget(QLabel("参数")) self._cloud_data_ent = _make_entry() self._cloud_data_ent.setText("{}") call_row.addWidget(self._cloud_data_ent, 1) self._btn_cloud_call = _make_btn("调用", self._cloud_do_call) call_row.addWidget(self._btn_cloud_call) lay.addLayout(call_row) self._cloud_result = QTextEdit() self._cloud_result.setReadOnly(True) self._cloud_result.setFixedHeight(120) self._cloud_result.setFont(QFont(_FM, 9)) lay.addWidget(self._cloud_result) bot = QHBoxLayout() self._cloud_status_lbl = QLabel("捕获: 0 条") bot.addWidget(self._cloud_status_lbl) bot.addStretch() lay.addLayout(bot) self._stack.addWidget(page) self._page_map["cloud"] = self._stack.count() - 1 # ── 敏感提取 ── def _build_security(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 8, 24, 8) lay.setSpacing(6) # 顶部操作栏 bar = QHBoxLayout() self._btn_sec_scan = _make_btn("扫描当前小程序", self._sec_do_scan) bar.addWidget(self._btn_sec_scan) self._btn_sec_history = _make_btn("历史记录", self._sec_show_history) bar.addWidget(self._btn_sec_history) self._btn_sec_back = _make_btn("返回", self._sec_back_to_scan) self._btn_sec_back.setVisible(False) bar.addWidget(self._btn_sec_back) bar.addStretch() self._sec_status_lbl = QLabel("") self._sec_status_lbl.setProperty("class", "muted") bar.addWidget(self._sec_status_lbl) lay.addLayout(bar) # 进度条 self._sec_prog = QProgressBar() self._sec_prog.setFixedHeight(6) self._sec_prog.setRange(0, 100) self._sec_prog.setValue(0) self._sec_prog.setTextVisible(False) self._sec_prog.setVisible(False) lay.addWidget(self._sec_prog) # 主内容区 — 内含 QStackedWidget 切换 scan/report/history self._sec_stack = QStackedWidget() # --- 扫描提示页 (index 0) --- scan_hint = QWidget() sh_lay = QVBoxLayout(scan_hint) sh_lay.setAlignment(Qt.AlignCenter) hint_lbl = QLabel("连接小程序后，点击「扫描当前小程序」提取 JS 中的敏感信息") hint_lbl.setProperty("class", "muted") hint_lbl.setAlignment(Qt.AlignCenter) hint_lbl.setWordWrap(True) sh_lay.addWidget(hint_lbl) self._sec_stack.addWidget(scan_hint) # --- 报告页 (index 1) --- report_page = QWidget() rp_lay = QVBoxLayout(report_page) rp_lay.setContentsMargins(0, 0, 0, 0) rp_lay.setSpacing(6) # 报告头部: appid + 概要 rp_hdr = QHBoxLayout() self._sec_rpt_header = QLabel("") self._sec_rpt_header.setProperty("class", "bold") rp_hdr.addWidget(self._sec_rpt_header) rp_hdr.addStretch() self._sec_rpt_summary = QLabel("") self._sec_rpt_summary.setProperty("class", "muted") rp_hdr.addWidget(self._sec_rpt_summary) rp_lay.addLayout(rp_hdr) # 左右分栏 rp_split = QHBoxLayout() rp_split.setSpacing(8) # 左栏: 类别列表 (滚动) left_scroll = QScrollArea() left_scroll.setWidgetResizable(True) left_scroll.setFrameShape(QFrame.NoFrame) left_scroll.setFixedWidth(220) left_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) left_inner = QWidget() self._sec_cat_lay = QVBoxLayout(left_inner) self._sec_cat_lay.setContentsMargins(0, 0, 0, 0) self._sec_cat_lay.setSpacing(2) self._sec_cat_lay.addStretch() left_scroll.setWidget(left_inner) rp_split.addWidget(left_scroll) # 右栏: 详情列表 (滚动) right_card = _make_card() right_card_lay = QVBoxLayout(right_card) right_card_lay.setContentsMargins(0, 0, 0, 0) right_card_lay.setSpacing(0) # 右栏顶部: 类别名 + 复制全部 self._sec_detail_bar = QHBoxLayout() self._sec_detail_bar.setContentsMargins(14, 8, 14, 4) self._sec_detail_title = QLabel("选择左侧类别查看详情") self._sec_detail_title.setProperty("class", "bold") self._sec_detail_bar.addWidget(self._sec_detail_title) self._sec_detail_bar.addStretch() self._btn_sec_copy_all = _make_btn("复制全部", self._sec_copy_all) self._btn_sec_copy_all.setVisible(False) self._sec_detail_bar.addWidget(self._btn_sec_copy_all) right_card_lay.addLayout(self._sec_detail_bar) # 右栏内容: QTextEdit (只读，自带滚动条，填满剩余空间) self._sec_detail_text = QTextEdit() self._sec_detail_text.setReadOnly(True) self._sec_detail_text.setFont(QFont(_FM, 9)) self._sec_detail_text.setTextInteractionFlags( Qt.TextSelectableByMouse | Qt.TextSelectableByKeyboard) self._sec_detail_text.setFrameShape(QFrame.NoFrame) right_card_lay.addWidget(self._sec_detail_text, 1) rp_split.addWidget(right_card, 1) rp_lay.addLayout(rp_split, 1) self._sec_cur_cat = None # 当前选中类别 key self._sec_cat_widgets = {} # key -> row widget self._sec_stack.addWidget(report_page) # --- 历史页 (index 2) --- history_page = QWidget() hp_lay = QVBoxLayout(history_page) hp_lay.setContentsMargins(0, 0, 0, 0) hp_lay.setSpacing(6) self._sec_hist_tree = QTreeWidget() self._sec_hist_tree.setHeaderLabels(["时间", "AppID", "名称", "JS数量", "发现项"]) self._sec_hist_tree.setColumnCount(5) self._sec_hist_tree.header().setStretchLastSection(True) self._sec_hist_tree.setRootIsDecorated(False) self._sec_hist_tree.setAlternatingRowColors(False) self._sec_hist_tree.header().setDefaultAlignment(Qt.AlignLeft | Qt.AlignVCenter) self._sec_hist_tree.itemDoubleClicked.connect(self._sec_hist_open) self._sec_hist_tree.setContextMenuPolicy(Qt.CustomContextMenu) self._sec_hist_tree.customContextMenuRequested.connect(self._sec_hist_menu) hp_lay.addWidget(self._sec_hist_tree, 1) self._sec_stack.addWidget(history_page) lay.addWidget(self._sec_stack, 1) self._stack.addWidget(page) self._page_map["security"] = self._stack.count() - 1 # ── 日志 ── def _build_logs(self): page = QWidget() lay = QVBoxLayout(page) lay.setContentsMargins(24, 12, 24, 16) lay.setSpacing(10) hdr = QHBoxLayout() hdr.addWidget(_make_label("日志输出", bold=True)) hdr.addStretch() self._btn_clear = _make_btn("清空", self._do_clear) hdr.addWidget(self._btn_clear) lay.addLayout(hdr) lc = _make_card() lc_lay = QVBoxLayout(lc) lc_lay.setContentsMargins(0, 0, 0, 0) self._logbox = QTextEdit() self._logbox.setReadOnly(True) self._logbox.setFont(QFont(_FM, 9)) lc_lay.addWidget(self._logbox) lay.addWidget(lc, 1) self._stack.addWidget(page) self._page_map["logs"] = self._stack.count() - 1 # ────────────────────────────────── # 前加载脚本 # ────────────────────────────────── def _preload_refresh(self): while self._preload_container.count(): item = self._preload_container.takeAt(0) w = item.widget() if w: w.deleteLater() scripts_dir = os.path.join(_BASE_DIR, "userscripts") js_files = sorted(f for f in os.listdir(scripts_dir) if f.endswith(".js")) if os.path.isdir(scripts_dir) else [] if not js_files: lbl = QLabel("userscripts/ 目录下无 .js 文件") lbl.setProperty("class", "muted") self._preload_container.addWidget(lbl) return for fn in js_files: row = QHBoxLayout() tog = ToggleSwitch(fn in self._selected_preload) tog.setFixedSize(36, 18) tog.toggled.connect(lambda v, f=fn: self._preload_toggle(f, v)) row.addWidget(tog) row.addWidget(QLabel(fn)) row.addStretch() w = QWidget() w.setLayout(row) self._preload_container.addWidget(w) def _preload_toggle(self, filename, val): if val: if filename not in self._selected_preload: self._selected_preload.append(filename) else: if filename in self._selected_preload: self._selected_preload.remove(filename) self._auto_save() # ────────────────────────────────── # 页面切换 # ────────────────────────────────── def _show(self, pid): self._pg = pid idx = self._page_map.get(pid, 0) self._stack.setCurrentIndexAnimated(idx) titles = {k: n for k, _, n in _MENU} self._hdr_title.setText(titles.get(pid, "")) self._hl_sb() def _hl_sb(self): for pid, (fr, ic, nm) in self._sb_items.items(): if pid == self._pg: fr.setProperty("class", "sb_item_active") else: fr.setProperty("class", "sb_item") fr.style().unpolish(fr) fr.style().polish(fr) ic.style().unpolish(ic) ic.style().polish(ic) nm.style().unpolish(nm) nm.style().polish(nm) # ────────────────────────────────── # 主题 # ────────────────────────────────── def _toggle_theme(self): self._tn = "light" if self._tn == "dark" else "dark" self.setStyleSheet(build_qss(self._tn)) self._update_theme_label() self._update_toggle_colors() self._hl_sb() self._auto_save() def _update_theme_label(self): txt = "\u2600 浅色模式" if self._tn == "dark" else "\u263d 深色模式" self._sb_theme.setText(txt) def _update_toggle_colors(self): c = _TH[self._tn] for tog in (self._tog_dm, self._tog_df): tog.set_colors(c["accent"], c["text4"]) def _auto_save(self): data = { "theme": self._tn, "debug_port": self._dp_ent.text(), "cdp_port": self._cp_ent.text(), "debug_main": self._tog_dm.isChecked(), "debug_frida": self._tog_df.isChecked(), "selected_preload_scripts": list(self._selected_preload), } _save_cfg(data) # ────────────────────────────────── # 业务 # ────────────────────────────────── def _copy_devtools_url(self): url = self._devtools_lbl.text() if url: QApplication.clipboard().setText(url) c = _TH[self._tn] self._devtools_copy_hint.setText("已复制!") self._devtools_copy_hint.setStyleSheet(f"color: {c['success']};") QTimer.singleShot(1500, lambda: ( self._devtools_copy_hint.setText("点击复制"), self._devtools_copy_hint.setStyleSheet(f"color: {c['text3']};") )) self._log_add("info", "[gui] DevTools 链接已复制到剪贴板") def _do_clear(self): self._logbox.clear() _LOG_MAX_BLOCKS = 500 # 最多保留的日志行数 def _log_add(self, lv, txt): c = _TH[self._tn] color_map = { "info": c["text2"], "error": c["error"], "debug": c["text3"], "frida": c["accent"], "warn": c["warning"], } color = color_map.get(lv, c["text2"]) self._logbox.append(f'{txt}') # 限制日志行数，防止 QTextEdit 内容过多导致 UI 卡顿 doc = self._logbox.document() overflow = doc.blockCount() - self._LOG_MAX_BLOCKS if overflow > 50: # 攒够 50 行再批量删，减少操作频率 cursor = self._logbox.textCursor() cursor.movePosition(cursor.MoveOperation.Start) for _ in range(overflow): cursor.movePosition(cursor.MoveOperation.Down, cursor.MoveMode.KeepAnchor) cursor.removeSelectedText() cursor.deleteChar() # 删掉残留空行 sb = self._logbox.verticalScrollBar() sb.setValue(sb.maximum()) def _do_start(self): if self._running: return try: dp = int(self._dp_ent.text()) cp = int(self._cp_ent.text()) except ValueError: self._log_add("error", "[gui] 端口号无效") return scripts_dir = os.path.join(_BASE_DIR, "userscripts") selected_files = [os.path.join(scripts_dir, fn) for fn in self._selected_preload] opts = CliOptions( debug_port=dp, cdp_port=cp, debug_main=self._tog_dm.isChecked(), debug_frida=self._tog_df.isChecked(), scripts_dir=scripts_dir, script_files=selected_files) logger = Logger(opts) logger.set_output_callback(lambda lv, tx: self._log_q.put((lv, tx))) us = load_userscripts_by_files(selected_files) if selected_files else [] if us: logger.info(f"[脚本] 已加载 {len(us)} 个") else: logger.info("[脚本] 无脚本") self._engine = DebugEngine(opts, logger, us) self._navigator = MiniProgramNavigator(self._engine) self._auditor = CloudAuditor(self._engine) self._engine.on_status_change(lambda s: self._sts_q.put(s)) self._loop = asyncio.new_event_loop() self._loop_th = threading.Thread( target=lambda: (asyncio.set_event_loop(self._loop), self._loop.run_forever()), daemon=True) self._loop_th.start() asyncio.run_coroutine_threadsafe(self._astart(), self._loop) self._running = True self._btn_start.setEnabled(False) self._btn_stop.setEnabled(True) self._btn_fetch.setEnabled(True) url = f"devtools://devtools/bundled/inspector.html?ws=127.0.0.1:{cp}" self._devtools_lbl.setText(url) c = _TH[self._tn] self._devtools_copy_hint.setText("点击复制") self._devtools_copy_hint.setStyleSheet(f"color: {c['text3']};") self._log_add("info", f"[gui] 浏览器访问: {url}") async def _astart(self): try: await self._engine.start() except Exception as e: self._log_q.put(("error", f"[gui] 启动失败: {e}")) QTimer.singleShot(0, self._on_fail) def _on_fail(self): self._running = False self._btn_start.setEnabled(True) self._btn_stop.setEnabled(False) self._btn_fetch.setEnabled(False) self._nav_btns(False) if self._loop and self._loop.is_running(): self._loop.call_soon_threadsafe(self._loop.stop) def _do_stop(self): if not self._running: return self._running = False self._poll_route_stop() if self._cloud_scan_active: self._cloud_scan_active = False if self._cloud_scan_poll_timer: self._cloud_scan_poll_timer.stop() self._cloud_scan_poll_timer = None if self._cancel_ev: self._cancel_ev.set() if self._engine and self._loop and self._loop.is_running(): fut = asyncio.run_coroutine_threadsafe(self._engine.stop(), self._loop) fut.add_done_callback(lambda _: self._loop.call_soon_threadsafe(self._loop.stop)) self._btn_start.setEnabled(True) self._btn_stop.setEnabled(False) self._btn_fetch.setEnabled(False) self._nav_btns(False) self._btn_autostop.setEnabled(False) self._redirect_guard_on = False self._guard_switch.setChecked(False) self._guard_label.setText("防跳转: 关闭") self._devtools_lbl.setText("") self._devtools_copy_hint.setText("") if self._sec_scanning: self._sec_scanning = False self._btn_sec_scan.setEnabled(True) self._sec_prog.setVisible(False) def _nav_btns(self, on): for b in (self._btn_go, self._btn_redir, self._btn_relaunch, self._btn_back, self._btn_auto, self._btn_prev, self._btn_next, self._btn_copy_route): b.setEnabled(on) self._guard_switch.setEnabled(on) def _do_fetch(self): if self._engine and self._loop: asyncio.run_coroutine_threadsafe(self._afetch(), self._loop) async def _afetch(self): try: await self._navigator.fetch_config() self._rte_q.put(("routes", self._navigator.pages, self._navigator.tab_bar_pages)) self._rte_q.put(("app_info", self._navigator.app_info)) QTimer.singleShot(0, self._poll_route_start) except Exception as e: self._log_q.put(("error", f"[导航] 获取失败: {e}")) def _poll_route_start(self): if not self._running: return if self._engine and self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._apoll_route(), self._loop) self._route_poll_id = QTimer.singleShot(2000, self._poll_route_start) def _poll_route_stop(self): self._route_poll_id = None async def _apoll_route(self): try: r = await self._navigator.get_current_route() self._rte_q.put(("current", r)) if self._redirect_guard_on: blocked = await self._navigator.get_blocked_redirects() if blocked: self._rte_q.put(("blocked", blocked)) except Exception: pass def _sel_route(self): items = self._tree.selectedItems() if not items: self._log_add("error", "[导航] 请先选择路由") return None item = items[0] return item.data(0, Qt.UserRole) def _do_go(self): r = self._sel_route() if r and self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("navigate_to", r, "跳转"), self._loop) def _do_redir(self): r = self._sel_route() if r and self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("redirect_to", r, "重定向"), self._loop) def _do_relaunch(self): r = self._sel_route() if r and self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("relaunch_to", r, "重启"), self._loop) def _do_back(self): if self._engine and self._loop: asyncio.run_coroutine_threadsafe(self._aback(), self._loop) async def _anav(self, method, route, desc): try: await getattr(self._navigator, method)(route) self._log_q.put(("info", f"[导航] 已{desc}到: {route}")) except Exception as e: self._log_q.put(("error", f"[导航] {desc}失败: {e}")) async def _aback(self): try: await self._navigator.navigate_back() self._log_q.put(("info", "[导航] 已返回")) except Exception as e: self._log_q.put(("error", f"[导航] 返回失败: {e}")) def _do_autovis(self): if not self._navigator or not self._navigator.pages: self._log_add("error", "[导航] 请先获取路由") return self._cancel_ev = asyncio.Event() self._btn_auto.setEnabled(False) self._btn_autostop.setEnabled(True) asyncio.run_coroutine_threadsafe( self._aauto(list(self._navigator.pages)), self._loop) async def _aauto(self, pages): def prog(i, total, route): self._rte_q.put(("progress", i, total, route)) try: await self._navigator.auto_visit( pages, delay=2.0, on_progress=prog, cancel_event=self._cancel_ev) except Exception as e: self._log_q.put(("error", f"[导航] 遍历出错: {e}")) finally: self._rte_q.put(("auto_done",)) def _do_autostop(self): if self._cancel_ev: self._cancel_ev.set() self._btn_autostop.setEnabled(False) self._btn_auto.setEnabled(True) def _do_prev(self): if not self._all_routes: self._log_add("error", "[导航] 请先获取路由") return if self._nav_route_idx <= 0: self._nav_route_idx = len(self._all_routes) - 1 else: self._nav_route_idx -= 1 route = self._all_routes[self._nav_route_idx] self._select_tree_route(route) self._log_add("info", f"[导航] 上一个: {route} ({self._nav_route_idx + 1}/{len(self._all_routes)})") if self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("navigate_to", route, "跳转"), self._loop) def _do_next(self): if not self._all_routes: self._log_add("error", "[导航] 请先获取路由") return if self._nav_route_idx >= len(self._all_routes) - 1: self._nav_route_idx = 0 else: self._nav_route_idx += 1 route = self._all_routes[self._nav_route_idx] self._select_tree_route(route) self._log_add("info", f"[导航] 下一个: {route} ({self._nav_route_idx + 1}/{len(self._all_routes)})") if self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("navigate_to", route, "跳转"), self._loop) def _do_manual_go(self): route = self._nav_input.text().strip().lstrip("/") if not route: self._log_add("error", "[导航] 请输入路由路径") return if self._engine and self._loop: asyncio.run_coroutine_threadsafe( self._anav("navigate_to", route, "跳转"), self._loop) def _do_copy_route(self): items = self._tree.selectedItems() if not items: self._log_add("error", "[导航] 请先选择路由") return route = items[0].data(0, Qt.UserRole) if route: QApplication.clipboard().setText(route) self._log_add("info", f"[导航] 已复制路由: {route}") def _nav_context_menu(self, pos): item = self._tree.itemAt(pos) if not item: return route = item.data(0, Qt.UserRole) if not route: return self._tree.setCurrentItem(item) menu = QMenu(self) menu.addAction("复制路由", lambda: ( QApplication.clipboard().setText(route), self._log_add("info", f"[导航] 已复制: {route}"))) menu.addSeparator() menu.addAction("跳转", lambda: asyncio.run_coroutine_threadsafe( self._anav("navigate_to", route, "跳转"), self._loop) if self._engine and self._loop else None) menu.addAction("重定向", lambda: asyncio.run_coroutine_threadsafe( self._anav("redirect_to", route, "重定向"), self._loop) if self._engine and self._loop else None) menu.addAction("重启到页面", lambda: asyncio.run_coroutine_threadsafe( self._anav("relaunch_to", route, "重启"), self._loop) if self._engine and self._loop else None) menu.exec(self._tree.viewport().mapToGlobal(pos)) def _do_toggle_guard_switch(self, checked): if not self._engine or not self._loop: self._guard_switch.blockSignals(True) self._guard_switch.setChecked(not checked) self._guard_switch.blockSignals(False) return asyncio.run_coroutine_threadsafe(self._atoggle_guard(checked), self._loop) async def _atoggle_guard(self, enable): try: if enable: r = await self._navigator.enable_redirect_guard() if r.get("ok"): self._redirect_guard_on = True self._blocked_seen = 0 self._log_q.put(("info", "[导航] 防跳转已开启，将拦截 redirectTo/reLaunch")) QTimer.singleShot(0, lambda: self._guard_label.setText("防跳转: 开启")) else: self._redirect_guard_on = False self._log_q.put(("error", "[导航] 开启防跳转失败")) QTimer.singleShot(0, self._guard_reset_switch) else: await self._navigator.disable_redirect_guard() self._redirect_guard_on = False self._log_q.put(("info", "[导航] 防跳转已关闭")) QTimer.singleShot(0, lambda: self._guard_label.setText("防跳转: 关闭")) except Exception as e: self._log_q.put(("error", f"[导航] 防跳转切换失败: {e}")) QTimer.singleShot(0, self._guard_reset_switch) def _guard_reset_switch(self): self._guard_switch.blockSignals(True) self._guard_switch.setChecked(self._redirect_guard_on) self._guard_switch.blockSignals(False) self._guard_label.setText("防跳转: 开启" if self._redirect_guard_on else "防跳转: 关闭") def _do_filter(self): q = self._srch_ent.text().strip().lower() if not q: if self._navigator: self._fill_tree(self._all_routes, self._navigator.tab_bar_pages) return flt = [p for p in self._all_routes if q in p.lower()] self._tree.clear() for p in flt: item = QTreeWidgetItem([p]) item.setData(0, Qt.UserRole, p) self._tree.addTopLevelItem(item) def _fill_tree(self, pages, tab_bar): self._tree.clear() tabs = set(tab_bar) groups = {} for p in pages: parts = p.split("/") g = parts[0] if len(parts) > 1 else "(root)" groups.setdefault(g, []).append(p) tl = [p for p in pages if p in tabs] if tl: nd = QTreeWidgetItem(["TabBar"]) nd.setExpanded(True) self._tree.addTopLevelItem(nd) for p in tl: d = p.split("/")[-1] if "/" in p else p child = QTreeWidgetItem([d]) child.setData(0, Qt.UserRole, p) nd.addChild(child) for g in sorted(groups): nd = QTreeWidgetItem([g]) self._tree.addTopLevelItem(nd) for p in groups[g]: if p in tabs: continue d = p[len(g) + 1:] if p.startswith(g + "/") else p child = QTreeWidgetItem([d]) child.setData(0, Qt.UserRole, p) nd.addChild(child) def _select_tree_route(self, route): """Select the tree item matching the given route path.""" for i in range(self._tree.topLevelItemCount()): top = self._tree.topLevelItem(i) if top.data(0, Qt.UserRole) == route: self._tree.setCurrentItem(top) self._tree.scrollToItem(top) return for j in range(top.childCount()): child = top.child(j) if child.data(0, Qt.UserRole) == route: self._tree.setCurrentItem(child) self._tree.scrollToItem(child) return # ────────────────────────────────── # 云扫描业务 # ────────────────────────────────── def _cloud_tree_context_menu(self, pos): item = self._cloud_tree.itemAt(pos) if not item: return self._cloud_tree.setCurrentItem(item) vals = [item.text(i) for i in range(6)] menu = QMenu(self) full_text = " | ".join(vals) menu.addAction("复制整行", lambda: QApplication.clipboard().setText(full_text)) name_str = vals[2] if len(vals) > 2 else "" if name_str: menu.addAction(f"复制名称: {name_str[:30]}", lambda: QApplication.clipboard().setText(name_str)) menu.addSeparator() row_id = id(item) if row_id in self._cloud_row_results: res = self._cloud_row_results[row_id] menu.addAction("查看返回结果", lambda: self._cloud_show_result(name_str, res)) menu.addSeparator() menu.addAction("删除此项", lambda: self._cloud_delete_item(item)) menu.exec(self._cloud_tree.viewport().mapToGlobal(pos)) def _cloud_delete_item(self, item): vals = tuple(item.text(i) for i in range(6)) idx = self._cloud_tree.indexOfTopLevelItem(item) if idx >= 0: self._cloud_tree.takeTopLevelItem(idx) self._cloud_all_items = [v for v in self._cloud_all_items if tuple(str(x) for x in v) != vals] self._cloud_row_results.pop(id(item), None) self._cloud_update_status() def _cloud_show_result(self, name, result): detail = json.dumps(result, ensure_ascii=False, indent=2, default=str) c = _TH[self._tn] self._cloud_result.setHtml(f'「{name}」返回结果:\n{detail}') def _cloud_update_status(self): count = self._cloud_tree.topLevelItemCount() total = len(self._cloud_all_items) if count < total: self._cloud_status_lbl.setText(f"显示: {count} / {total} 条") else: self._cloud_status_lbl.setText(f"捕获: {count} 条") def _cloud_filter(self): kw = self._cloud_search_ent.text().strip().lower() self._cloud_tree.clear() for vals in self._cloud_all_items: if kw and not any(kw in str(v).lower() for v in vals): continue item = QTreeWidgetItem([str(v) for v in vals]) self._cloud_tree.addTopLevelItem(item) self._cloud_update_status() def _cloud_on_select(self, item): if item and item.columnCount() >= 4: self._cloud_name_ent.setText(item.text(2)) data_str = item.text(3).strip() try: json.loads(data_str) self._cloud_data_ent.setText(data_str) except Exception: self._cloud_data_ent.setText("{}") def _cloud_ensure_auditor(self): if not self._engine or not self._loop or not self._loop.is_running(): self._log_add("error", "[云扫描] 请先启动调试") return False if not self._auditor: self._auditor = CloudAuditor(self._engine) return True def _cloud_do_toggle(self): if not self._cloud_ensure_auditor(): return if self._cloud_scan_active: self._cloud_stop_scan() else: self._cloud_start_scan() def _cloud_start_scan(self): if not self._cloud_ensure_auditor(): return self._cloud_scan_active = True c = _TH[self._tn] self._btn_cloud_toggle.setText("停止捕获") self._cloud_scan_lbl.setText("捕获中...") self._cloud_scan_lbl.setStyleSheet(f"color: {c['success']};") self._log_add("info", "[云扫描] 全局捕获已启动") asyncio.run_coroutine_threadsafe(self._acloud_start(), self._loop) self._cloud_scan_poll() async def _acloud_start(self): try: await self._auditor.start() except Exception as e: self._log_q.put(("error", f"[云扫描] Hook 启动异常: {e}")) def _cloud_stop_scan(self): self._cloud_scan_active = False c = _TH[self._tn] self._btn_cloud_toggle.setText("开启捕获") self._cloud_scan_lbl.setText("已停止") self._cloud_scan_lbl.setStyleSheet(f"color: {c['text3']};") if self._cloud_scan_poll_timer: self._cloud_scan_poll_timer.stop() self._cloud_scan_poll_timer = None if self._auditor and self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._auditor.stop(), self._loop) self._log_add("info", "[云扫描] 全局捕获已停止") def _cloud_scan_poll(self): if not self._cloud_scan_active or not self._auditor: return if self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._acloud_poll(), self._loop) self._cloud_scan_poll_timer = QTimer() self._cloud_scan_poll_timer.setSingleShot(True) self._cloud_scan_poll_timer.timeout.connect(self._cloud_scan_poll) self._cloud_scan_poll_timer.start(2000) async def _acloud_poll(self): try: new_calls = await self._auditor.poll() if new_calls: self._cld_q.put(("new_calls", new_calls)) except Exception: pass def _cloud_do_static_scan(self): if not self._cloud_ensure_auditor(): return self._btn_cloud_static.setEnabled(False) self._log_add("info", "[云扫描] 开始静态扫描 JS 源码...") asyncio.run_coroutine_threadsafe(self._acloud_static_scan(), self._loop) async def _acloud_static_scan(self): try: def progress(msg): self._log_q.put(("info", f"[云扫描] {msg}")) results = await self._auditor.static_scan(on_progress=progress) self._cld_q.put(("static_results", results)) except Exception as e: self._log_q.put(("error", f"[云扫描] 静态扫描异常: {e}")) finally: self._cld_q.put(("static_done",)) def _cloud_do_clear(self): self._cloud_tree.clear() self._cloud_all_items.clear() self._cloud_row_results.clear() if self._auditor and self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._auditor.clear(), self._loop) self._cloud_status_lbl.setText("捕获: 0 条") def _cloud_do_call(self): if not self._cloud_ensure_auditor(): return name = self._cloud_name_ent.text().strip() if not name: self._cloud_result.setPlainText("请输入函数名") return try: data = json.loads(self._cloud_data_ent.text()) except (json.JSONDecodeError, TypeError): self._cloud_result.setPlainText("参数 JSON 格式错误") return self._btn_cloud_call.setEnabled(False) self._cloud_result.setPlainText(f"正在调用 {name} ...") asyncio.run_coroutine_threadsafe(self._acloud_call(name, data), self._loop) async def _acloud_call(self, name, data): try: res = await self._auditor.call_function(name, data) self._cld_q.put(("call_result", name, res)) except Exception as e: self._cld_q.put(("call_result", name, {"ok": False, "status": "fail", "error": str(e)})) def _cloud_do_export(self): if not self._auditor: self._log_add("error", "[云扫描] 无数据") return report = self._auditor.export_report(self._cloud_all_items, self._cloud_call_history) path = os.path.join(_BASE_DIR, "cloud_audit_report.json") try: with open(path, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, ensure_ascii=False, default=str) self._log_add("info", f"[云扫描] 报告已导出: {path}") except Exception as e: self._log_add("error", f"[云扫描] 导出失败: {e}") # ────────────────────────────────── # 敏感提取业务 # ────────────────────────────────── def _sec_do_scan(self): if self._sec_scanning: self._log_add("info", "[敏感提取] 扫描正在进行中，请稍候...") return if not self._engine or not self._loop or not self._loop.is_running(): self._log_add("error", "[敏感提取] 请先启动调试并连接小程序") return if not self._engine.miniapp_clients: self._log_add("error", "[敏感提取] 小程序未连接，请先打开小程序") return self._sec_scanning = True self._btn_sec_scan.setEnabled(False) self._sec_prog.setVisible(True) self._sec_prog.setValue(0) self._sec_status_lbl.setText("正在提取 JS 源码...") self._sec_stack.setCurrentIndex(0) self._log_add("info", "[敏感提取] 开始扫描...") fut = asyncio.run_coroutine_threadsafe(self._asec_scan(), self._loop) # 捕获 future 异常，防止静默失败 def _on_scan_done(f): try: f.result() except Exception as e: self._sec_q.put(("error", f"扫描协程异常: {e}")) self._sec_q.put(("done", None, 0, 0)) fut.add_done_callback(_on_scan_done) async def _asec_scan(self): try: # Step 1: 通过 CDP Debugger 获取所有 JS 脚本源码 self._sec_q.put(("progress", 5, "正在收集脚本列表...")) script_ids = [] def _on_parsed(data): p = data.get("params", {}) sid = p.get("scriptId") if sid: script_ids.append((sid, p.get("url", ""))) self._engine.on_cdp_event("Debugger.scriptParsed", _on_parsed) try: try: await self._engine.send_cdp_command("Debugger.disable", timeout=3.0) except Exception: pass await asyncio.sleep(0.3) try: await self._engine.send_cdp_command("Debugger.enable", timeout=5.0) except asyncio.TimeoutError: self._sec_q.put(("log", "Debugger.enable 超时，尝试重新启用...")) await asyncio.sleep(0.5) try: await self._engine.send_cdp_command("Debugger.enable", timeout=5.0) except Exception as e2: self._sec_q.put(("log", f"Debugger.enable 重试失败: {e2}")) try: await self._engine.send_cdp_command( "Debugger.setSkipAllPauses", {"skip": True}, timeout=3.0) except Exception: pass # 等待 scriptParsed 事件全部到达： # 先等 2 秒让大部分事件到达，然后连续 3 次(每次 0.4s)数量不变才认为稳定 await asyncio.sleep(2.0) stable_count = 0 prev = 0 for _ in range(15): cur = len(script_ids) if cur == prev and cur > 0: stable_count += 1 if stable_count >= 3: break else: stable_count = 0 prev = cur await asyncio.sleep(0.4) self._sec_q.put(("log", f"脚本收集完成: {len(script_ids)} 个")) except Exception as e: self._sec_q.put(("log", f"Debugger 启动异常: {e}")) finally: self._engine.off_cdp_event("Debugger.scriptParsed", _on_parsed) if not script_ids: self._sec_q.put(("error", "未能获取到脚本列表，请确保小程序已打开")) self._sec_q.put(("done", None, 0, 0)) return self._sec_q.put(("progress", 15, f"发现 {len(script_ids)} 个脚本，正在提取源码...")) # Step 2: 逐个获取脚本源码 js_sources = [] for i, (sid, url) in enumerate(script_ids): try: resp = await self._engine.send_cdp_command( "Debugger.getScriptSource", {"scriptId": sid}, timeout=8.0) source = resp.get("result", {}).get("scriptSource", "") if source and len(source) > 20: js_sources.append(source) except Exception: pass if i % 10 == 0: pct = 15 + int((i + 1) / len(script_ids) * 15) self._sec_q.put(("progress", pct, f"提取源码 {i+1}/{len(script_ids)} ...")) try: await self._engine.send_cdp_command("Debugger.disable", timeout=3.0) except Exception: pass if not js_sources: self._sec_q.put(("error", "脚本源码提取为空")) self._sec_q.put(("done", None, 0, 0)) return total_size = sum(len(s) for s in js_sources) self._sec_q.put(("progress", 30, f"已提取 {len(js_sources)} 个 JS 源码，启动独立分析进程...")) # Step 3: 获取小程序信息 (若还没获取过，自动 fetch_config) appid = "" app_name = "" if self._navigator: if not self._navigator.app_info or not self._navigator.app_info.get("appid"): try: await self._navigator.fetch_config() except Exception: pass appid = self._navigator.app_info.get("appid", "") app_name = self._navigator.app_info.get("name", "") if not appid: appid = "unknown" # Step 4: 在独立进程中分析 (不阻塞主进程/UI) _sec_run_worker(js_sources, appid, _BASE_DIR, self._sec_q, name=app_name) except Exception as e: self._sec_q.put(("error", f"分析失败: {e}")) self._sec_q.put(("done", None, 0, 0)) def _sec_extract_val(self, result): if not result: return None r = result.get("result", {}) inner = r.get("result", {}) return inner.get("value") def _sec_show_report(self, result, js_count=0, total_size=0, appid="", scan_time="", name=""): """在报告页渲染分析结果 (左类别列表 + 右详情)""" self._sec_result = result c = _TH[self._tn] if not scan_time: scan_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") header_parts = [] if name: header_parts.append(name) if appid: header_parts.append(f"AppID: {appid}") self._sec_rpt_header.setText(" | ".join(header_parts) if header_parts else "分析报告") total_findings = sum(len(v) for v in result.values() if isinstance(v, list)) summary = f"{js_count} 个 JS ({self._fmt_size(total_size)}) | " \ f"{total_findings} 项 | {scan_time}" if js_count else \ f"{total_findings} 项 | {scan_time}" self._sec_rpt_summary.setText(summary) # 清除旧的左栏类别行 while self._sec_cat_lay.count() > 1: item = self._sec_cat_lay.takeAt(0) w = item.widget() if w: w.deleteLater() self._sec_cat_widgets.clear() self._sec_cur_cat = None # 生成左栏类别行 for key in CATEGORY_INFO: items = result.get(key, []) cn, _ = CATEGORY_INFO[key] count = len(items) row = QFrame() row.setFixedHeight(38) row.setCursor(Qt.PointingHandCursor) row_lay = QHBoxLayout(row) row_lay.setContentsMargins(0, 0, 6, 0) row_lay.setSpacing(6) # 绿色竖条 bar = QFrame() bar.setFixedSize(4, 24) bar.setStyleSheet(f"background: {c['accent']}; border-radius: 2px;") row_lay.addWidget(bar) name_lbl = QLabel(cn) name_lbl.setProperty("class", "bold") name_lbl.setFont(QFont(_FN, 10)) row_lay.addWidget(name_lbl, 1) count_lbl = QLabel(str(count)) count_lbl.setFont(QFont(_FM, 9)) if count > 0: count_lbl.setStyleSheet(f"color: {c['accent']};") else: count_lbl.setStyleSheet(f"color: {c['text3']};") row_lay.addWidget(count_lbl) copy_btn = QPushButton("复制") copy_btn.setFixedSize(52, 26) copy_btn.setFont(QFont(_FN, 9)) copy_btn.clicked.connect(lambda _, k=key, b=copy_btn: self._sec_copy_cat(k, b)) row_lay.addWidget(copy_btn) row.mousePressEvent = lambda e, k=key: self._sec_select_cat(k) self._sec_cat_widgets[key] = row self._sec_cat_lay.insertWidget(self._sec_cat_lay.count() - 1, row) # 清空右侧 self._sec_clear_detail() self._sec_detail_title.setText("选择左侧类别查看详情") self._btn_sec_copy_all.setVisible(False) # 自动选中第一个有内容的类别 for key in CATEGORY_INFO: if result.get(key): self._sec_select_cat(key) break self._sec_stack.setCurrentIndex(1) self._btn_sec_back.setVisible(True) def _sec_select_cat(self, key): """选中左栏类别，在右栏显示详情列表""" c = _TH[self._tn] self._sec_cur_cat = key # 高亮左栏选中项 for k, row in self._sec_cat_widgets.items(): if k == key: row.setStyleSheet( f"QFrame {{ background: {c['sb_active']}; border-radius: 6px; }}" f" QLabel {{ color: {c['text1']}; }}" f" QPushButton {{ color: {c['text1']}; }}" ) else: row.setStyleSheet("") items = self._sec_result.get(key, []) if self._sec_result else [] cn, _ = CATEGORY_INFO.get(key, (key, "")) self._sec_detail_title.setText(f"{cn} ({len(items)})") self._btn_sec_copy_all.setVisible(len(items) > 0) # 直接填充 QTextEdit，每行一条 if items: self._sec_detail_text.setStyleSheet( f"QTextEdit {{ color: {c['text1']}; background: transparent;" f" selection-background-color: {c['accent']}; }}" ) self._sec_detail_text.setPlainText("\n".join(str(v) for v in items)) else: self._sec_detail_text.clear() def _sec_clear_detail(self): """清空右栏详情""" self._sec_detail_text.clear() def _sec_copy_cat(self, key, btn=None): """复制某个类别的全部内容""" items = self._sec_result.get(key, []) if self._sec_result else [] if items: QApplication.clipboard().setText("\n".join(str(v) for v in items)) cn, _ = CATEGORY_INFO.get(key, (key, "")) self._sec_status_lbl.setText(f"已复制 {cn} ({len(items)} 项)") if btn and isinstance(btn, QPushButton): old_text = btn.text() btn.setText("已复制") QTimer.singleShot(1200, lambda: btn.setText(old_text) if btn else None) def _sec_copy_all(self): """复制当前选中类别的全部内容""" if self._sec_cur_cat: self._sec_copy_cat(self._sec_cur_cat, self._btn_sec_copy_all) def _sec_show_history(self): self._sec_hist_tree.clear() reports = load_reports(_BASE_DIR) for r in reports: summary = r.get("summary", {}) total = sum(summary.values()) item = QTreeWidgetItem([ r.get("time", ""), r.get("appid", ""), r.get("name", ""), str(r.get("js_count", 0)), str(total), ]) item.setData(0, Qt.UserRole, r) self._sec_hist_tree.addTopLevelItem(item) self._sec_stack.setCurrentIndex(2) self._btn_sec_back.setVisible(True) self._sec_status_lbl.setText(f"共 {len(reports)} 条记录") def _sec_hist_open(self, item): r = item.data(0, Qt.UserRole) if not r: return result = r.get("result", {}) self._sec_show_report( result, js_count=r.get("js_count", 0), total_size=r.get("total_size", 0), appid=r.get("appid", ""), scan_time=r.get("time", ""), name=r.get("name", ""), ) def _sec_hist_menu(self, pos): item = self._sec_hist_tree.itemAt(pos) if not item: return r = item.data(0, Qt.UserRole) menu = QMenu(self) menu.addAction("查看报告", lambda: self._sec_hist_open(item)) if r: menu.addAction("删除", lambda: self._sec_hist_delete(item, r)) menu.exec(self._sec_hist_tree.viewport().mapToGlobal(pos)) def _sec_hist_delete(self, item, r): fn = r.get("_filename", "") if fn: delete_report(_BASE_DIR, fn) idx = self._sec_hist_tree.indexOfTopLevelItem(item) if idx >= 0: self._sec_hist_tree.takeTopLevelItem(idx) def _sec_back_to_scan(self): if self._sec_result: self._sec_stack.setCurrentIndex(1) else: self._sec_stack.setCurrentIndex(0) self._btn_sec_back.setVisible(False) self._sec_status_lbl.setText("") def _handle_sec(self, item): kind = item[0] if kind == "progress": _, pct, msg = item self._sec_prog.setValue(pct) self._sec_status_lbl.setText(msg) elif kind == "log": self._log_add("info", f"[敏感提取] {item[1]}") elif kind == "error": self._log_add("error", f"[敏感提取] {item[1]}") self._sec_status_lbl.setText(item[1]) elif kind == "done": _, result, js_count, total_size = item self._sec_scanning = False self._btn_sec_scan.setEnabled(True) self._sec_prog.setValue(100) QTimer.singleShot(500, lambda: self._sec_prog.setVisible(False)) if result: total_findings = sum(len(v) for v in result.values() if isinstance(v, list)) self._sec_status_lbl.setText(f"分析完成，发现 {total_findings} 项") self._log_add("info", f"[敏感提取] 完成: {js_count} 个 JS, 发现 {total_findings} 项") appid = "" app_name = "" if self._navigator and self._navigator.app_info: appid = self._navigator.app_info.get("appid", "") app_name = self._navigator.app_info.get("name", "") self._sec_show_report(result, js_count, total_size, appid, name=app_name) else: self._sec_status_lbl.setText("分析完成，无结果") @staticmethod def _fmt_size(n): if n < 1024: return f"{n} B" elif n < 1024 * 1024: return f"{n/1024:.1f} KB" else: return f"{n/1024/1024:.1f} MB" @staticmethod def _html_esc(s): return s.replace("&", "&").replace("<", "<").replace(">", ">") # ────────────────────────────────── # 轮询 # ────────────────────────────────── def _tick(self): for _ in range(60): # 每轮最多处理60条日志，防止阻塞UI try: msg = self._log_q.get_nowait() except queue.Empty: break if isinstance(msg, tuple) and len(msg) == 3 and msg[0] == "__hook_status__": _, fn, ok = msg self._hook_update_status(fn, ok) else: lv, tx = msg self._log_add(lv, tx) for _ in range(50): try: s = self._sts_q.get_nowait() except queue.Empty: break self._apply_sts(s) for _ in range(50): try: item = self._rte_q.get_nowait() except queue.Empty: break self._handle_rte(item) for _ in range(50): try: item = self._cld_q.get_nowait() except queue.Empty: break self._handle_cld(item) for _ in range(50): try: item = self._sec_q.get_nowait() except queue.Empty: break self._handle_sec(item) def _apply_sts(self, sts): c = _TH[self._tn] for key, (dot, lb, name) in self._dots.items(): on = sts.get(key, False) dot.set_color(c["success"] if on else c["text4"]) lb.setText(f"{name}: {'已连接' if on else '未连接'}") lb.setStyleSheet(f"color: {c['success'] if on else c['text2']};") self._nav_btns(sts.get("miniapp", False)) if sts.get("miniapp") and not self._cloud_scan_active and self._auditor: self._cloud_start_scan() def _handle_rte(self, item): kind = item[0] if kind == "routes": _, pages, tab = item self._all_routes = list(pages) self._fill_tree(pages, tab) elif kind == "app_info": info = item[1] aid = info.get("appid", "") ent = info.get("entry", "") txt = f"应用: {aid}" if aid else "应用: --" if ent: txt += f" | 入口: {ent}" self._app_lbl.setText(txt) elif kind == "current": r = item[1] self._route_lbl.setText(f"当前路由: /{r}" if r else "当前路由: --") elif kind == "progress": _, i, total, route = item if total > 0: self._prog.setValue(int((i / total) * 100)) if route != "done": self._select_tree_route(route) self._route_lbl.setText( f"正在访问: /{route}" if route != "done" else "遍历完成") elif kind == "blocked": blocked = item[1] for b in blocked[self._blocked_seen:]: self._log_add("warn", f"[防跳转] 拦截 {b.get('type','')} → {b.get('url','')} ({b.get('time','')})") self._blocked_seen = len(blocked) elif kind == "auto_done": self._prog.setValue(100) self._btn_auto.setEnabled(True) self._btn_autostop.setEnabled(False) self._log_add("info", "[导航] 遍历完成") def _handle_cld(self, item): kind = item[0] c = _TH[self._tn] _type_cn = {"function": "云函数", "storage": "存储", "container": "容器"} if kind == "new_calls": calls = item[1] if calls: kw = self._cloud_search_ent.text().strip().lower() for call in calls: data_str = json.dumps(call.get("data", {}), ensure_ascii=False) if len(data_str) > 80: data_str = data_str[:77] + "..." ctype = call.get("type", "function") type_label = _type_cn.get(ctype, ctype) if ctype.startswith("db"): type_label = "数据库" status = call.get("status", "") vals = (call.get("appId", ""), type_label, call.get("name", ""), data_str, status, call.get("timestamp", "")) self._cloud_all_items.append(vals) if kw and not any(kw in str(v).lower() for v in vals): continue tree_item = QTreeWidgetItem([str(v) for v in vals]) self._cloud_tree.addTopLevelItem(tree_item) result_data = call.get("result") or call.get("error") if result_data is not None: self._cloud_row_results[id(tree_item)] = { "status": status, "result": call.get("result"), "error": call.get("error"), "data": call.get("data"), } self._cloud_tree.scrollToBottom() self._cloud_update_status() self._cloud_scan_lbl.setText(f"捕获中... {len(self._cloud_all_items)} 条") self._cloud_scan_lbl.setStyleSheet(f"color: {c['success']};") elif kind == "static_results": funcs = item[1] if funcs: kw = self._cloud_search_ent.text().strip().lower() for f in funcs: params = ", ".join(f.get("params", [])) or "--" if len(params) > 80: params = params[:77] + "..." ftype = f.get("type", "function") type_label = {"function": "云函数", "storage": "存储", "database": "数据库"}.get(ftype, ftype) vals = (f.get("appId", ""), f"[静态]{type_label}", f["name"], params, f"x{f.get('count',1)}", "") self._cloud_all_items.append(vals) if kw and not any(kw in str(v).lower() for v in vals): continue tree_item = QTreeWidgetItem([str(v) for v in vals]) self._cloud_tree.addTopLevelItem(tree_item) self._cloud_tree.scrollToBottom() self._cloud_update_status() self._log_add("info", f"[云扫描] 静态扫描发现 {len(funcs)} 个云函数引用") elif kind == "static_done": self._btn_cloud_static.setEnabled(True) elif kind == "call_result": _, name, res = item self._btn_cloud_call.setEnabled(True) status = res.get("status", "unknown") if status == "success": detail = json.dumps(res.get("result", {}), ensure_ascii=False, default=str) self._cloud_result.setHtml( f'{name} -> 成功:\n{detail}') elif status == "fail": err = res.get("error", "") or res.get("reason", "未知错误") self._cloud_result.setHtml( f'{name} -> 失败: {err}') else: detail = json.dumps(res, ensure_ascii=False, default=str) self._cloud_result.setHtml( f'{name} -> {detail}') # ────────────────────────────────── # 退出 # ────────────────────────────────── def closeEvent(self, event): if self._running: self._do_stop() QTimer.singleShot(400, lambda: QApplication.quit()) event.ignore() else: event.accept() if __name__ == "__main__": multiprocessing.freeze_support() # PyInstaller 打包需要 import signal signal.signal(signal.SIGINT, signal.SIG_DFL) # Ctrl+C 直接退出 app = QApplication(sys.argv) app.setFont(QFont(_FN, 9)) window = App() window.show() sys.exit(app.exec())