+
+
---
@@ -60,7 +65,7 @@ pip install -r requirements.txt
16815, 16965, 17037, 17071, 17127, 18055, 18151, 18787, 18891, 18955, 19027, 19201
- 推荐微信版本为4.1.0.30
- 下载地址[weixin/4.1.0.30](https://github.com/vs-olitus/wx-version/releases/tag/4.1.0.30)
-
+
## 使用方法
### GUI 模式(推荐)
@@ -82,7 +87,7 @@ python gui.py
python main.py
# 自定义端口
-python main.py --debug-port 9421 --cdp-port 62000
+python main.py --cdp-port 62000
# 开启详细日志
python main.py --debug-main --debug-frida
diff --git a/frida/hook.js b/frida/hook.js
index f098450..c37d6e1 100644
--- a/frida/hook.js
+++ b/frida/hook.js
@@ -72,11 +72,6 @@ const hookOnLoadScene = (a1, sceneOffsets) => {
}
send("[hook] hook scene condition -> 1101");
miniappScenePtr.writeInt(1101);
-
- // TODO: customize debugging endpoint
- // const websocketServerStringPtr = passArgs.add(8).readPointer().add(520);
- // VERBOSE && console.log("[hook] hook websocket server, original: ", websocketServerStringPtr.readUtf8String());
- // websocketServerStringPtr.writeUtf8String("ws://127.0.0.1:8189/");
};
const patchOnLoadStart = (base, config) => {
diff --git a/gui.py b/gui.py
index 39a526e..d6e54b3 100644
--- a/gui.py
+++ b/gui.py
@@ -7,6 +7,7 @@ import json
import multiprocessing
import os
import queue
+import subprocess
import sys
import threading
@@ -19,10 +20,11 @@ from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QFrame, QPushButton, QScrollArea, QTextEdit,
QTreeWidget, QTreeWidgetItem, QProgressBar, QStackedWidget,
- QMenu, QHeaderView, QAbstractItemView,
+ QMenu, QHeaderView, QAbstractItemView, QFileDialog, QInputDialog,
+ QTabWidget, QTableWidget, QTableWidgetItem, QDialog,
)
-from src.cli import CliOptions, DEBUG_PORT, CDP_PORT
+from src.cli import CliOptions, CDP_PORT
from src.logger import Logger
from src.engine import DebugEngine
from src.navigator import MiniProgramNavigator
@@ -56,6 +58,7 @@ _MENU = [
("navigator", "⬡", "路由导航"),
("hook", "◈", "Hook"),
("cloud", "☁", "云扫描"),
+ ("extract", "◆", "敏感信息提取"),
("vconsole", "◇", "调试开关"),
("logs", "≡", "运行日志"),
]
@@ -232,6 +235,18 @@ def build_qss(tn):
background: {"#1a3a2a" if tn == "dark" else "#b0dfc0"};
color: {"#3a6a4a" if tn == "dark" else "#5a8a6a"};
}}
+ /* 表格内按钮 — 清除全局样式,由 inline setStyleSheet 控制 */
+ QTableWidget QPushButton {{
+ background: transparent;
+ color: {c['text2']};
+ border: none;
+ border-radius: 6px;
+ padding: 4px 12px;
+ font-size: 12px;
+ }}
+ QTableWidget QPushButton:hover {{
+ background: transparent;
+ }}
/* ── 输入框 ── */
QLineEdit {{
@@ -337,6 +352,34 @@ def build_qss(tn):
background: transparent;
}}
+ /* ── 表格控件 ── */
+ QTableWidget {{
+ background: {c['card']};
+ color: {c['text2']};
+ border: none;
+ font-size: 12px;
+ outline: 0;
+ gridline-color: {c['border']};
+ }}
+ QTableWidget::item {{
+ padding: 6px 10px;
+ border: none;
+ background: {c['card']};
+ color: {c['text2']};
+ }}
+ QTableWidget::item:selected {{
+ background: {sel_bg};
+ color: {sel_fg};
+ }}
+ QTableWidget QHeaderView::section {{
+ background: {hdr_bg};
+ color: {c['text1']};
+ border: none;
+ padding: 6px 10px;
+ font-weight: bold;
+ font-size: 12px;
+ }}
+
/* ── 滚动区域 ── */
QScrollArea {{
background: transparent;
@@ -614,6 +657,15 @@ class App(QMainWindow):
self._sb_items = {}
self._page_map = {}
+ # 敏感信息提取 状态
+ self._ext_proc = None
+ self._ext_thread = None
+ self._ext_q = queue.Queue()
+ self._ext_custom_patterns = dict(self._cfg.get("extract_custom_patterns", {}))
+ self._ext_app_states = {} # {appid: {"decompiled": bool, "scanned": bool, ...}}
+ self._ext_app_widgets = {} # {appid: row widget ref}
+ self._ext_current_op = None # ("decompile"/"scan", appid) or None
+
self._build()
self.setStyleSheet(build_qss(self._tn))
self._show("control")
@@ -783,6 +835,7 @@ class App(QMainWindow):
self._build_navigator()
self._build_hook()
self._build_cloud()
+ self._build_extract()
self._build_vconsole()
self._build_logs()
@@ -803,12 +856,6 @@ class App(QMainWindow):
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)))
@@ -972,6 +1019,9 @@ class App(QMainWindow):
self._btn_back = _make_btn("返回上页", self._do_back)
self._btn_back.setEnabled(False)
b1.addWidget(self._btn_back)
+ self._btn_refresh = _make_btn("刷新页面", self._do_refresh)
+ self._btn_refresh.setEnabled(False)
+ b1.addWidget(self._btn_refresh)
b1.addStretch()
self._btn_fetch = _make_btn("获取路由", self._do_fetch)
self._btn_fetch.setEnabled(False)
@@ -1233,6 +1283,1219 @@ class App(QMainWindow):
self._stack.addWidget(page)
self._page_map["cloud"] = self._stack.count() - 1
+ # ── 敏感信息提取 ──
+
+ def _build_extract(self):
+ # 外层使用 QStackedWidget 实现子页面切换
+ self._ext_stack = QStackedWidget()
+
+ # =================== 主页面 ===================
+ main_page = QWidget()
+ main_lay = QVBoxLayout(main_page)
+ main_lay.setContentsMargins(24, 8, 24, 8)
+ main_lay.setSpacing(8)
+
+ # --- Row 1: Applet目录 ---
+ c1 = _make_card()
+ c1_lay = QVBoxLayout(c1)
+ c1_lay.setContentsMargins(16, 10, 16, 10)
+ c1_lay.setSpacing(6)
+
+ path_row = QHBoxLayout()
+ path_row.addWidget(_make_label("Applet目录", bold=True))
+ self._ext_path_ent = _make_entry("wxapkg 包目录路径...")
+ # 自动检测默认路径
+ default_pkg = ""
+ if os.name == "nt":
+ user_home = os.path.expanduser("~")
+ default_pkg = os.path.join(
+ user_home, "AppData", "Roaming", "Tencent",
+ "xwechat", "radium", "Applet", "packages"
+ )
+ if not os.path.isdir(default_pkg):
+ default_pkg = ""
+ saved_path = self._cfg.get("extract_packages_dir", "")
+ if saved_path:
+ self._ext_path_ent.setText(saved_path)
+ elif default_pkg:
+ self._ext_path_ent.setText(default_pkg)
+ # 先连接信号,然后手动刷新一次(setText 不会重复触发因为信号在之后连接)
+ # 注意: setText 在信号连接前调用,所以不会触发重复刷新
+ self._ext_path_ent.textChanged.connect(self._ext_on_path_changed)
+ path_row.addWidget(self._ext_path_ent, 1)
+ btn_auto = _make_btn("自动选择", self._ext_auto_detect)
+ path_row.addWidget(btn_auto)
+ btn_browse = _make_btn("选择", self._ext_browse)
+ path_row.addWidget(btn_browse)
+ c1_lay.addLayout(path_row)
+ main_lay.addWidget(c1)
+
+ # --- Row 2: 功能区 ---
+ func_row = QHBoxLayout()
+ self._btn_ext_regex = _make_btn("正则配置", self._ext_goto_regex)
+ func_row.addWidget(self._btn_ext_regex)
+ self._btn_ext_clear_decompiled = _make_btn("清空解包文件", self._ext_clear_decompiled)
+ func_row.addWidget(self._btn_ext_clear_decompiled)
+ self._btn_ext_clear_applet = _make_btn("清空Applet目录", self._ext_clear_applet)
+ func_row.addWidget(self._btn_ext_clear_applet)
+ func_row.addStretch()
+ # 自动反编译 & 自动提取 开关
+ func_row.addWidget(_make_label("自动反编译"))
+ self._tog_auto_dec = ToggleSwitch(self._cfg.get("auto_decompile", False))
+ self._tog_auto_dec.toggled.connect(lambda v: (self._auto_save(), self._ext_auto_process_pending() if v else None))
+ func_row.addWidget(self._tog_auto_dec)
+ func_row.addWidget(_make_label("自动提取"))
+ self._tog_auto_scan = ToggleSwitch(self._cfg.get("auto_scan", False))
+ self._tog_auto_scan.toggled.connect(lambda v: (self._auto_save(), self._ext_auto_process_pending() if v else None))
+ func_row.addWidget(self._tog_auto_scan)
+ main_lay.addLayout(func_row)
+
+ # --- 进度 + 状态 ---
+ self._ext_prog = QProgressBar()
+ self._ext_prog.setMaximum(100)
+ self._ext_prog.setValue(0)
+ self._ext_prog.setTextVisible(False)
+ self._ext_prog.setFixedHeight(6)
+ main_lay.addWidget(self._ext_prog)
+ self._ext_status_lbl = QLabel("就绪")
+ self._ext_status_lbl.setProperty("class", "muted")
+ main_lay.addWidget(self._ext_status_lbl)
+
+ # --- Row 3: 小程序列表 ---
+ list_card = _make_card()
+ list_lay = QVBoxLayout(list_card)
+ list_lay.setContentsMargins(0, 6, 0, 6)
+ list_lay.setSpacing(0)
+
+ # 表头
+ hdr = QHBoxLayout()
+ hdr.setContentsMargins(16, 4, 16, 4)
+ hdr_appid = _make_label("AppID", bold=True)
+ hdr_appid.setFixedWidth(180)
+ hdr.addWidget(hdr_appid)
+ hdr_name = _make_label("名称", bold=True)
+ hdr_name.setMinimumWidth(100)
+ hdr.addWidget(hdr_name, 1)
+ hdr_ops = _make_label("操作", bold=True)
+ hdr.addWidget(hdr_ops)
+ list_lay.addLayout(hdr)
+
+ # 分割线
+ sep = QFrame()
+ sep.setFixedHeight(1)
+ sep.setStyleSheet("background: rgba(128,128,128,0.2);")
+ list_lay.addWidget(sep)
+
+ # 滚动区域
+ scroll = QScrollArea()
+ scroll.setWidgetResizable(True)
+ self._ext_list_inner = QWidget()
+ self._ext_list_layout = QVBoxLayout(self._ext_list_inner)
+ self._ext_list_layout.setContentsMargins(8, 4, 8, 4)
+ self._ext_list_layout.setSpacing(4)
+ self._ext_list_layout.addStretch()
+ scroll.setWidget(self._ext_list_inner)
+ list_lay.addWidget(scroll, 1)
+
+ main_lay.addWidget(list_card, 1)
+
+ # 日志区
+ self._ext_logbox = QTextEdit()
+ self._ext_logbox.setReadOnly(True)
+ self._ext_logbox.setFont(QFont(_FM, 9))
+ self._ext_logbox.setMaximumHeight(120)
+ main_lay.addWidget(self._ext_logbox)
+
+ self._ext_stack.addWidget(main_page) # index 0
+
+ # =================== 正则配置子页面 ===================
+ regex_page = QWidget()
+ regex_lay = QVBoxLayout(regex_page)
+ regex_lay.setContentsMargins(24, 8, 24, 8)
+ regex_lay.setSpacing(8)
+
+ # 返回按钮行
+ regex_top = QHBoxLayout()
+ btn_back_regex = _make_btn("← 返回", lambda: self._ext_stack.setCurrentIndex(0))
+ regex_top.addWidget(btn_back_regex)
+ regex_top.addWidget(_make_label("正则配置", bold=True))
+ regex_top.addStretch()
+ regex_lay.addLayout(regex_top)
+
+ # 自定义正则卡片
+ custom_card = _make_card()
+ cc_lay = QVBoxLayout(custom_card)
+ cc_lay.setContentsMargins(16, 10, 16, 10)
+ cc_lay.setSpacing(6)
+ cc_hdr = QHBoxLayout()
+ cc_hdr.addWidget(_make_label("自定义正则", bold=True))
+ cc_hdr.addWidget(_make_label("(提取时会与内置规则合并使用)", muted=True))
+ cc_hdr.addStretch()
+ btn_add = _make_btn("新建", self._ext_add_pattern)
+ cc_hdr.addWidget(btn_add)
+ cc_lay.addLayout(cc_hdr)
+
+ # 表头行
+ cc_hdr_row = QHBoxLayout()
+ cc_hdr_row.setContentsMargins(12, 4, 12, 4)
+ h1 = _make_label("栏目", bold=True); h1.setFixedWidth(120)
+ cc_hdr_row.addWidget(h1)
+ h2 = _make_label("正则表达式", bold=True)
+ cc_hdr_row.addWidget(h2, 1)
+ h3 = _make_label("状态", bold=True); h3.setFixedWidth(50)
+ cc_hdr_row.addWidget(h3)
+ h4 = _make_label("操作", bold=True); h4.setFixedWidth(180)
+ cc_hdr_row.addWidget(h4)
+ cc_lay.addLayout(cc_hdr_row)
+ cc_sep = QFrame(); cc_sep.setFixedHeight(1)
+ cc_sep.setStyleSheet("background: rgba(128,128,128,0.2);")
+ cc_lay.addWidget(cc_sep)
+ # 滚动区域放行列表
+ self._ext_pat_scroll = QScrollArea()
+ self._ext_pat_scroll.setWidgetResizable(True)
+ self._ext_pat_scroll.setStyleSheet("QScrollArea { border: none; }")
+ self._ext_pat_inner = QWidget()
+ self._ext_pat_layout = QVBoxLayout(self._ext_pat_inner)
+ self._ext_pat_layout.setContentsMargins(0, 0, 0, 0)
+ self._ext_pat_layout.setSpacing(4)
+ self._ext_pat_layout.addStretch()
+ self._ext_pat_scroll.setWidget(self._ext_pat_inner)
+ cc_lay.addWidget(self._ext_pat_scroll)
+ regex_lay.addWidget(custom_card, 1) # stretch=1,自定义区域占大空间
+
+ # 内置正则卡片(紧凑)
+ builtin_card = _make_card()
+ bc_lay = QVBoxLayout(builtin_card)
+ bc_lay.setContentsMargins(16, 10, 16, 10)
+ bc_lay.setSpacing(6)
+ bc_lay.addWidget(_make_label("内置正则规则 (只读)", bold=True))
+
+ self._ext_builtin_tree = QTreeWidget()
+ self._ext_builtin_tree.setHeaderLabels(["分类", "正则/说明"])
+ bh = self._ext_builtin_tree.header()
+ bh.setStretchLastSection(True)
+ bh.setSectionResizeMode(0, QHeaderView.Interactive)
+ self._ext_builtin_tree.setColumnWidth(0, 200)
+ self._ext_builtin_tree.setRootIsDecorated(False)
+ self._ext_builtin_tree.setMaximumHeight(200)
+ bc_lay.addWidget(self._ext_builtin_tree)
+ regex_lay.addWidget(builtin_card) # 无stretch,内置区域紧凑
+
+ self._ext_stack.addWidget(regex_page) # index 1
+
+ # =================== 查看敏感信息子页面 ===================
+ view_page = QWidget()
+ view_lay = QVBoxLayout(view_page)
+ view_lay.setContentsMargins(16, 8, 16, 8)
+ view_lay.setSpacing(6)
+
+ view_top = QHBoxLayout()
+ btn_back_view = _make_btn("← 返回", lambda: self._ext_stack.setCurrentIndex(0))
+ view_top.addWidget(btn_back_view)
+ self._ext_view_title = _make_label("查看敏感信息", bold=True)
+ view_top.addWidget(self._ext_view_title)
+ view_top.addStretch()
+ self._btn_ext_open_html = _make_btn("网页访问", self._ext_open_html)
+ view_top.addWidget(self._btn_ext_open_html)
+ view_lay.addLayout(view_top)
+
+ # 结果展示区 (滚动)
+ self._ext_view_scroll = QScrollArea()
+ self._ext_view_scroll.setWidgetResizable(True)
+ self._ext_view_scroll.setStyleSheet("QScrollArea { border: none; }")
+ self._ext_view_inner = QWidget()
+ self._ext_view_top_layout = QVBoxLayout(self._ext_view_inner)
+ self._ext_view_top_layout.setContentsMargins(0, 0, 0, 0)
+ self._ext_view_top_layout.setSpacing(0)
+ self._ext_view_scroll.setWidget(self._ext_view_inner)
+ view_lay.addWidget(self._ext_view_scroll, 1)
+
+ self._ext_stack.addWidget(view_page) # index 2
+
+ # 注册到主 stack
+ self._stack.addWidget(self._ext_stack)
+ self._page_map["extract"] = self._stack.count() - 1
+
+ # 填充内置正则
+ self._ext_fill_builtin_patterns()
+ self._ext_refresh_custom_patterns()
+
+ # 延迟加载小程序列表
+ QTimer.singleShot(500, self._ext_refresh_apps)
+
+ # 定时监控目录变化 (每5秒)
+ self._ext_watch_timer = QTimer()
+ self._ext_watch_timer.timeout.connect(self._ext_check_dir_changes)
+ self._ext_watch_timer.start(5000)
+ self._ext_last_appids = set() # 上一次扫描到的 appids
+
+ # 存储当前查看的 html path
+ self._ext_current_html = ""
+ self._ext_current_json = ""
+
+ # ============================================
+ # 提取页 - 辅助方法
+ # ============================================
+
+ def _ext_browse(self):
+ d = QFileDialog.getExistingDirectory(self, "选择小程序包目录", self._ext_path_ent.text())
+ if d:
+ self._ext_path_ent.setText(d)
+
+ def _ext_on_path_changed(self):
+ """路径变化时自动保存和刷新列表"""
+ self._auto_save()
+ self._ext_refresh_apps()
+
+ def _ext_check_dir_changes(self):
+ """定时检查目录是否有新的小程序包"""
+ pkg_dir = self._ext_path_ent.text().strip()
+ if not pkg_dir or not os.path.isdir(pkg_dir):
+ return
+
+ # 快速扫描子目录名称
+ try:
+ current_dirs = set()
+ for entry in os.listdir(pkg_dir):
+ if os.path.isdir(os.path.join(pkg_dir, entry)):
+ current_dirs.add(entry)
+ except Exception:
+ return
+
+ if current_dirs != self._ext_last_appids:
+ new_ids = current_dirs - self._ext_last_appids
+ self._ext_last_appids = current_dirs
+ if new_ids:
+ self._ext_refresh_apps()
+ for nid in new_ids:
+ self._ext_log(f"检测到新小程序: {nid}")
+ # 自动处理新增的小程序
+ self._ext_auto_process_new(new_ids)
+
+ def _ext_auto_detect(self):
+ """自动检测默认路径"""
+ if os.name == "nt":
+ user_home = os.path.expanduser("~")
+ default_pkg = os.path.join(
+ user_home, "AppData", "Roaming", "Tencent",
+ "xwechat", "radium", "Applet", "packages"
+ )
+ if os.path.isdir(default_pkg):
+ if self._ext_path_ent.text().strip() == default_pkg:
+ # 路径相同但强制刷新
+ self._ext_refresh_apps()
+ else:
+ self._ext_path_ent.setText(default_pkg)
+ self._ext_log("已自动检测到 Applet 目录")
+ else:
+ self._ext_log("未找到默认 Applet 目录,请手动选择")
+ else:
+ self._ext_log("非 Windows 系统,请手动选择目录")
+
+ def _ext_auto_process_new(self, new_ids):
+ """对新增的小程序自动执行反编译/扫描(按开关状态)"""
+ if not self._tog_auto_dec.isChecked() and not self._tog_auto_scan.isChecked():
+ return
+ # 如果当前已有任务在运行,排队延迟处理
+ if self._ext_current_op is not None:
+ QTimer.singleShot(3000, lambda ids=set(new_ids): self._ext_auto_process_new(ids))
+ return
+ for appid in sorted(new_ids):
+ state = self._ext_app_states.get(appid, {})
+ if self._tog_auto_dec.isChecked() and not state.get("decompiled", False):
+ self._ext_log(f"[自动] 开始反编译: {appid}")
+ self._ext_do_decompile(appid)
+ return # 一次处理一个,完成后 _handle_ext 会触发下一步
+ if self._tog_auto_scan.isChecked() and state.get("decompiled", False) and not state.get("scanned", False):
+ self._ext_log(f"[自动] 开始提取: {appid}")
+ self._ext_do_scan(appid)
+ return
+
+ def _ext_auto_process_pending(self):
+ """遍历所有已知app,对未处理的自动执行反编译/扫描"""
+ if self._ext_current_op is not None:
+ return
+ for appid, state in self._ext_app_states.items():
+ if self._tog_auto_dec.isChecked() and not state.get("decompiled", False):
+ self._ext_log(f"[自动] 开始反编译: {appid}")
+ self._ext_do_decompile(appid)
+ return
+ if self._tog_auto_scan.isChecked() and state.get("decompiled", False) and not state.get("scanned", False):
+ self._ext_log(f"[自动] 开始提取: {appid}")
+ self._ext_do_scan(appid)
+ return
+
+ def _ext_log(self, msg):
+ c = _TH[self._tn]
+ self._ext_logbox.append(f'{msg}')
+ sb = self._ext_logbox.verticalScrollBar()
+ sb.setValue(sb.maximum())
+
+ def _ext_get_app_name(self, appid, pkgs, output_base):
+ """尝试从已解包文件中读取小程序名称"""
+ app_output = os.path.join(output_base, appid)
+ decompile_dir = os.path.join(app_output, "decompiled")
+
+ # 尝试从 app-config.json 读取
+ for fname in ("app-config.json", "app.json"):
+ cfg_path = os.path.join(decompile_dir, fname)
+ if os.path.isfile(cfg_path):
+ try:
+ with open(cfg_path, "r", encoding="utf-8") as f:
+ cfg = json.load(f)
+ # 多种可能的字段
+ name = (cfg.get("window", {}).get("navigationBarTitleText", "")
+ or cfg.get("appname", "")
+ or cfg.get("entryPagePath", ""))
+ if name:
+ return f"{name} ({len(pkgs)}pkg)"
+ except Exception:
+ pass
+
+ return f"{len(pkgs)} pkg"
+
+ def _ext_refresh_apps(self):
+ """刷新小程序列表"""
+ pkg_dir = self._ext_path_ent.text().strip()
+ if not pkg_dir or not os.path.isdir(pkg_dir):
+ return
+
+ # 清除旧列表
+ while self._ext_list_layout.count() > 1: # 保留 stretch
+ item = self._ext_list_layout.takeAt(0)
+ w = item.widget()
+ if w:
+ w.deleteLater()
+ self._ext_app_widgets.clear()
+
+ # 扫描目录
+ try:
+ from src.wxapkg import find_wxapkg_files
+ all_pkgs = find_wxapkg_files(pkg_dir)
+ except Exception as e:
+ self._ext_log(f"扫描目录失败: {e}")
+ return
+
+ # 按 appid 分组
+ appid_groups = {}
+ for pkg in all_pkgs:
+ appid_groups.setdefault(pkg["appid"], []).append(pkg)
+
+ if not appid_groups:
+ self._ext_status_lbl.setText("未找到小程序")
+ return
+
+ c = _TH[self._tn]
+ output_base = os.path.join(_BASE_DIR, "output")
+
+ for appid, pkgs in sorted(appid_groups.items()):
+ row = QFrame()
+ row.setStyleSheet(
+ f"QFrame {{ background: {c['input']}; border-radius: 8px; }}"
+ f"QFrame QLabel {{ background: transparent; }}")
+ row_lay = QHBoxLayout(row)
+ row_lay.setContentsMargins(12, 6, 12, 6)
+ row_lay.setSpacing(6)
+
+ lbl_id = QLabel(appid)
+ lbl_id.setFixedWidth(180)
+ lbl_id.setFont(QFont(_FM, 9))
+ lbl_id.setStyleSheet(f"color: {c['text1']};")
+ row_lay.addWidget(lbl_id)
+
+ # 检查是否已解包
+ app_output = os.path.join(output_base, appid)
+ decompile_dir = os.path.join(app_output, "decompiled")
+ result_dir = os.path.join(app_output, "result")
+ is_decompiled = os.path.isdir(decompile_dir) and any(
+ f.endswith(('.js', '.html', '.htm'))
+ for _, _, files in os.walk(decompile_dir) for f in files
+ ) if os.path.isdir(decompile_dir) else False
+ is_scanned = os.path.isfile(os.path.join(result_dir, "report.json"))
+
+ # 尝试读取小程序名称
+ app_name = self._ext_get_app_name(appid, pkgs, output_base) if is_decompiled else f"{len(pkgs)} pkg (未反编译)"
+ lbl_name = QLabel(app_name)
+ lbl_name.setMinimumWidth(100)
+ lbl_name.setFont(QFont(_FN, 9))
+ lbl_name.setStyleSheet(f"color: {c['text2']};")
+ row_lay.addWidget(lbl_name, 1)
+
+ self._ext_app_states[appid] = {
+ "decompiled": is_decompiled,
+ "scanned": is_scanned,
+ "packages": pkgs,
+ "decompile_dir": decompile_dir,
+ "result_dir": result_dir,
+ }
+
+ # 按钮
+ btn_dec = QPushButton("反编译")
+ btn_dec.setFixedWidth(60)
+ btn_dec.clicked.connect(lambda _, a=appid: self._ext_do_decompile(a))
+ if is_decompiled:
+ btn_dec.setStyleSheet(
+ f"QPushButton {{ background: {c['success']}; color: #111; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}")
+ row_lay.addWidget(btn_dec)
+
+ btn_scan = QPushButton("提取")
+ btn_scan.setFixedWidth(50)
+ btn_scan.setEnabled(is_decompiled)
+ btn_scan.clicked.connect(lambda _, a=appid: self._ext_do_scan(a))
+ row_lay.addWidget(btn_scan)
+
+ btn_view = QPushButton("查看")
+ btn_view.setFixedWidth(50)
+ btn_view.setEnabled(is_scanned)
+ btn_view.clicked.connect(lambda _, a=appid: self._ext_view_results(a))
+ row_lay.addWidget(btn_view)
+
+ btn_del = QPushButton("删除")
+ btn_del.setFixedWidth(50)
+ btn_del.clicked.connect(lambda _, a=appid: self._ext_delete_app(a))
+ btn_del.setStyleSheet(
+ f"QPushButton {{ background: {c['error']}; color: #fff; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}"
+ f"QPushButton:hover {{ background: #dc2626; }}")
+ row_lay.addWidget(btn_del)
+
+ self._ext_app_widgets[appid] = {
+ "row": row, "btn_dec": btn_dec, "btn_scan": btn_scan,
+ "btn_view": btn_view, "btn_del": btn_del, "lbl_name": lbl_name,
+ }
+
+ # 插入在 stretch 之前
+ self._ext_list_layout.insertWidget(self._ext_list_layout.count() - 1, row)
+
+ self._ext_status_lbl.setText(f"发现 {len(appid_groups)} 个小程序")
+ # 记录当前已知 appid 集合 (供目录监控用)
+ self._ext_last_appids = set(appid_groups.keys())
+
+ def _ext_update_app_buttons(self, appid):
+ """更新指定 app 的按钮状态"""
+ state = self._ext_app_states.get(appid, {})
+ widgets = self._ext_app_widgets.get(appid, {})
+ if not widgets:
+ return
+ c = _TH[self._tn]
+ is_dec = state.get("decompiled", False)
+ is_scanned = state.get("scanned", False)
+
+ btn_dec = widgets["btn_dec"]
+ if is_dec:
+ btn_dec.setStyleSheet(
+ f"QPushButton {{ background: {c['success']}; color: #111; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}")
+ else:
+ btn_dec.setStyleSheet("")
+
+ widgets["btn_scan"].setEnabled(is_dec)
+ widgets["btn_view"].setEnabled(is_scanned)
+
+ # ============================================
+ # 提取页 - 反编译
+ # ============================================
+
+ def _ext_do_decompile(self, appid):
+ if self._ext_proc:
+ self._ext_log("有任务正在运行,请等待完成")
+ return
+
+ output_base = os.path.join(_BASE_DIR, "output")
+ app_output = os.path.join(output_base, appid)
+ os.makedirs(app_output, exist_ok=True)
+
+ pkg_dir = self._ext_path_ent.text().strip()
+ worker_path = os.path.join(_BASE_DIR, "src", "extract_worker.py")
+ cmd = [
+ sys.executable, worker_path, "decompile",
+ "--packages-dir", pkg_dir,
+ "--appid", appid,
+ "--output-dir", app_output,
+ ]
+
+ self._ext_log(f"开始反编译: {appid}")
+ self._ext_prog.setValue(0)
+ self._ext_status_lbl.setText(f"正在反编译 {appid}...")
+ self._ext_current_op = ("decompile", appid)
+
+ try:
+ self._ext_proc = subprocess.Popen(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
+ )
+ except Exception as e:
+ self._ext_log(f"启动失败: {e}")
+ return
+
+ self._ext_thread = threading.Thread(target=self._ext_reader, daemon=True)
+ self._ext_thread.start()
+
+ # ============================================
+ # 提取页 - 敏感信息扫描
+ # ============================================
+
+ def _ext_do_scan(self, appid):
+ if self._ext_proc:
+ self._ext_log("有任务正在运行,请等待完成")
+ return
+
+ state = self._ext_app_states.get(appid, {})
+ if not state.get("decompiled"):
+ self._ext_log(f"请先反编译 {appid}")
+ return
+
+ output_base = os.path.join(_BASE_DIR, "output")
+ app_output = os.path.join(output_base, appid)
+ decompile_dir = state.get("decompile_dir", os.path.join(app_output, "decompiled"))
+ result_dir = os.path.join(app_output, "result")
+ os.makedirs(result_dir, exist_ok=True)
+
+ # 保存自定义正则
+ custom_file = ""
+ if self._ext_custom_patterns:
+ custom_file = os.path.join(_BASE_DIR, ".extract_custom_patterns.json")
+ with open(custom_file, "w", encoding="utf-8") as f:
+ json.dump(self._ext_custom_patterns, f, ensure_ascii=False)
+
+ worker_path = os.path.join(_BASE_DIR, "src", "extract_worker.py")
+ cmd = [
+ sys.executable, worker_path, "scan",
+ "--scan-dir", decompile_dir,
+ "--output-dir", result_dir,
+ ]
+ if custom_file:
+ cmd += ["--custom-patterns", custom_file]
+
+ self._ext_log(f"开始提取敏感信息: {appid}")
+ self._ext_prog.setValue(0)
+ self._ext_status_lbl.setText(f"正在扫描 {appid}...")
+ self._ext_current_op = ("scan", appid)
+
+ try:
+ self._ext_proc = subprocess.Popen(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
+ )
+ except Exception as e:
+ self._ext_log(f"启动失败: {e}")
+ return
+
+ self._ext_thread = threading.Thread(target=self._ext_reader, daemon=True)
+ self._ext_thread.start()
+
+ # ============================================
+ # 提取页 - 子进程通信
+ # ============================================
+
+ def _ext_reader(self):
+ """后台线程读取子进程 stdout"""
+ proc = self._ext_proc
+ if not proc or not proc.stdout:
+ return
+ try:
+ for line in proc.stdout:
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ obj = json.loads(line)
+ self._ext_q.put(obj)
+ except json.JSONDecodeError:
+ self._ext_q.put({"type": "log", "msg": line})
+ except Exception:
+ pass
+ finally:
+ proc.wait()
+ stderr_out = ""
+ if proc.stderr:
+ try:
+ stderr_out = proc.stderr.read()
+ except Exception:
+ pass
+ if stderr_out:
+ self._ext_q.put({"type": "log", "msg": f"[stderr] {stderr_out[:500]}"})
+ self._ext_q.put({"type": "__done__", "returncode": proc.returncode})
+
+ # ============================================
+ # 提取页 - 查看结果
+ # ============================================
+
+ def _ext_view_results(self, appid):
+ """查看敏感信息子页面 — 双列布局,仿 HTML 报告样式"""
+ state = self._ext_app_states.get(appid, {})
+ result_dir = state.get("result_dir", "")
+ json_path = os.path.join(result_dir, "report.json")
+ html_path = os.path.join(result_dir, "report.html")
+
+ if not os.path.isfile(json_path):
+ self._ext_log(f"未找到 {appid} 的扫描结果")
+ return
+
+ self._ext_current_html = html_path
+ self._ext_current_json = json_path
+ self._ext_view_title.setText(f"查看敏感信息 - {appid}")
+
+ # 清除旧内容
+ layout = self._ext_view_top_layout
+ while layout.count():
+ item = layout.takeAt(0)
+ w = item.widget()
+ if w:
+ w.deleteLater()
+ elif item.layout():
+ # 递归删除子布局
+ sub = item.layout()
+ while sub.count():
+ si = sub.takeAt(0)
+ sw = si.widget()
+ if sw:
+ sw.deleteLater()
+
+ # 加载 JSON 数据
+ try:
+ with open(json_path, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ except Exception as e:
+ self._ext_log(f"加载结果失败: {e}")
+ return
+
+ c = _TH[self._tn]
+ accent = c["accent"] # 绿色
+
+ cat_labels = {
+ 'ip': 'IP', 'ip_port': 'IP:PORT', 'domain': '域名',
+ 'sfz': '身份证', 'mobile': '手机号', 'mail': '邮箱',
+ 'jwt': 'JWT', 'algorithm': '加密算法', 'secret': 'Secret/密钥',
+ 'path': 'Path', 'incomplete_path': 'IncompletePath',
+ 'url': 'URL', 'static': 'StaticUrl'
+ }
+ # 反向映射: 中文标签 → 内置key
+ label_to_key = {v: k for k, v in cat_labels.items()}
+
+ left_cats = ['ip', 'ip_port', 'domain', 'sfz', 'mobile', 'mail', 'jwt', 'algorithm', 'secret']
+ right_cats = ['path', 'incomplete_path', 'url', 'static']
+ all_builtin = set(left_cats + right_cats)
+
+ # 合并自定义结果到同名内置分类,去重
+ merged_data = {}
+ custom_only_keys = []
+ for k, v in data.items():
+ if k == "_meta":
+ continue
+ items = v if isinstance(v, list) else []
+ if k in all_builtin:
+ merged_data.setdefault(k, []).extend(items)
+ elif k in label_to_key:
+ # 自定义名称和内置中文标签相同,合并到内置分类
+ builtin_key = label_to_key[k]
+ merged_data.setdefault(builtin_key, []).extend(items)
+ else:
+ merged_data.setdefault(k, []).extend(items)
+ if k not in all_builtin:
+ custom_only_keys.append(k)
+
+ # 去重
+ for k in merged_data:
+ seen = set()
+ deduped = []
+ for item in merged_data[k]:
+ s = str(item)
+ if s not in seen:
+ seen.add(s)
+ deduped.append(item)
+ merged_data[k] = deduped
+
+ def _build_cat_widget(key, label, items):
+ """构建单个分类的 widget — 支持展开/折叠"""
+ w = QWidget()
+ lay = QVBoxLayout(w)
+ lay.setContentsMargins(0, 8, 0, 4)
+ lay.setSpacing(2)
+
+ # 标题行: ▶ 分类名 (数量) [复制]
+ title_row = QHBoxLayout()
+ title_row.setContentsMargins(0, 0, 0, 0)
+
+ # 展开/折叠按钮
+ btn_fold = QPushButton("▼")
+ btn_fold.setFixedSize(20, 20)
+ btn_fold.setCursor(Qt.CursorShape.PointingHandCursor)
+ btn_fold.setStyleSheet(
+ f"QPushButton {{ background: transparent; color: {c['text3']}; border: none;"
+ f"font-size: 10px; padding: 0; }}"
+ f"QPushButton:hover {{ color: {c['text1']}; }}")
+ title_row.addWidget(btn_fold)
+
+ title_lbl = QLabel(f"{label} ({len(items)})")
+ title_lbl.setFont(QFont(_FN, 11, QFont.Weight.Bold))
+ title_lbl.setStyleSheet(
+ f"color: {c['text1']}; border-left: 4px solid {accent}; padding-left: 8px;"
+ f"background: transparent;")
+ title_row.addWidget(title_lbl)
+ title_row.addStretch()
+
+ btn_copy = QPushButton("复制")
+ btn_copy.setFixedSize(48, 26)
+ btn_copy.setCursor(Qt.CursorShape.PointingHandCursor)
+ btn_copy.setStyleSheet(
+ f"QPushButton {{ background: {c['input']}; color: {accent}; border: 1px solid {accent};"
+ f"border-radius: 4px; font-size: 11px; padding: 2px 8px; }}"
+ f"QPushButton:hover {{ background: {accent}; color: #111; }}")
+ copy_text = "\n".join(str(i) for i in items)
+ btn_copy.clicked.connect(lambda _, t=copy_text, b=btn_copy: self._ext_copy_cat(t, b))
+ title_row.addWidget(btn_copy)
+ lay.addLayout(title_row)
+
+ # 内容区域(可折叠)
+ content_widget = QWidget()
+ content_lay = QVBoxLayout(content_widget)
+ content_lay.setContentsMargins(0, 0, 0, 0)
+ content_lay.setSpacing(0)
+ if items:
+ content_lbl = QLabel()
+ content_lbl.setFont(QFont(_FM, 9))
+ content_lbl.setWordWrap(True)
+ content_lbl.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
+ content_lbl.setStyleSheet(
+ f"color: {c['text2']}; padding-left: 14px; background: transparent;")
+ display_items = items[:200]
+ lines = [str(i) for i in display_items]
+ if len(items) > 200:
+ lines.append(f"... 共 {len(items)} 条,仅显示前 200 条")
+ content_lbl.setText("\n".join(lines))
+ content_lay.addWidget(content_lbl)
+ lay.addWidget(content_widget)
+
+ # 展开/折叠逻辑 — 默认展开
+ def toggle_fold():
+ visible = content_widget.isVisible()
+ content_widget.setVisible(not visible)
+ btn_fold.setText("▶" if visible else "▼")
+
+ btn_fold.clicked.connect(toggle_fold)
+
+ return w
+
+ # === 双列布局 ===
+ cols_widget = QWidget()
+ cols_layout = QHBoxLayout(cols_widget)
+ cols_layout.setContentsMargins(0, 0, 0, 0)
+ cols_layout.setSpacing(16)
+
+ # 左列
+ left_col = QWidget()
+ left_lay = QVBoxLayout(left_col)
+ left_lay.setContentsMargins(0, 0, 0, 0)
+ left_lay.setSpacing(0)
+ for cat in left_cats:
+ label = cat_labels.get(cat, cat)
+ items = merged_data.get(cat, [])
+ left_lay.addWidget(_build_cat_widget(cat, label, items))
+ # 自定义规则(非同名)放左列底部
+ for key in custom_only_keys:
+ items = merged_data.get(key, [])
+ left_lay.addWidget(_build_cat_widget(key, key, items))
+ left_lay.addStretch()
+ cols_layout.addWidget(left_col, 1)
+
+ # 分隔线
+ sep = QFrame()
+ sep.setFrameShape(QFrame.Shape.VLine)
+ sep.setStyleSheet("color: rgba(128,128,128,0.2);")
+ cols_layout.addWidget(sep)
+
+ # 右列
+ right_col = QWidget()
+ right_lay = QVBoxLayout(right_col)
+ right_lay.setContentsMargins(0, 0, 0, 0)
+ right_lay.setSpacing(0)
+ for cat in right_cats:
+ label = cat_labels.get(cat, cat)
+ items = merged_data.get(cat, [])
+ right_lay.addWidget(_build_cat_widget(cat, label, items))
+ right_lay.addStretch()
+ cols_layout.addWidget(right_col, 1)
+
+ layout.addWidget(cols_widget)
+ layout.addStretch()
+
+ # 切换到查看页
+ self._ext_stack.setCurrentIndex(2)
+
+ def _ext_copy_cat(self, text, btn):
+ """复制分类内容并显示反馈"""
+ QApplication.clipboard().setText(text)
+ old = btn.text()
+ btn.setText("✓")
+ QTimer.singleShot(1200, lambda: btn.setText(old))
+
+ def _ext_open_html(self):
+ """浏览器打开 HTML 报告"""
+ if self._ext_current_html and os.path.isfile(self._ext_current_html):
+ QDesktopServices.openUrl(QUrl.fromLocalFile(self._ext_current_html))
+ else:
+ self._ext_log("HTML 报告不存在")
+
+ # ============================================
+ # 提取页 - 删除
+ # ============================================
+
+ def _ext_delete_app(self, appid):
+ """删除指定小程序的解包和结果"""
+ import shutil
+ output_base = os.path.join(_BASE_DIR, "output")
+ app_output = os.path.join(output_base, appid)
+ if os.path.isdir(app_output):
+ try:
+ shutil.rmtree(app_output)
+ self._ext_log(f"已删除 {appid} 的数据")
+ except Exception as e:
+ self._ext_log(f"删除失败: {e}")
+
+ # 更新状态
+ if appid in self._ext_app_states:
+ self._ext_app_states[appid]["decompiled"] = False
+ self._ext_app_states[appid]["scanned"] = False
+ self._ext_update_app_buttons(appid)
+ # 如果开启了自动反编译,重新处理
+ QTimer.singleShot(500, self._ext_auto_process_pending)
+
+ def _ext_clear_decompiled(self):
+ """清空所有解包文件"""
+ import shutil
+ output_base = os.path.join(_BASE_DIR, "output")
+ if not os.path.isdir(output_base):
+ return
+ count = 0
+ for appid in os.listdir(output_base):
+ dec_dir = os.path.join(output_base, appid, "decompiled")
+ if os.path.isdir(dec_dir):
+ try:
+ shutil.rmtree(dec_dir)
+ count += 1
+ except Exception:
+ pass
+ self._ext_log(f"已清空 {count} 个小程序的解包文件")
+ # 刷新状态
+ for appid in self._ext_app_states:
+ self._ext_app_states[appid]["decompiled"] = False
+ self._ext_app_states[appid]["scanned"] = False
+ self._ext_update_app_buttons(appid)
+ # 如果开启了自动反编译,重新处理
+ QTimer.singleShot(500, self._ext_auto_process_pending)
+
+ def _ext_clear_applet(self):
+ """清空 Applet 目录"""
+ import shutil
+ pkg_dir = self._ext_path_ent.text().strip()
+ if not pkg_dir or not os.path.isdir(pkg_dir):
+ self._ext_log("Applet 目录不存在")
+ return
+ try:
+ for entry in os.listdir(pkg_dir):
+ full = os.path.join(pkg_dir, entry)
+ if os.path.isdir(full):
+ shutil.rmtree(full)
+ self._ext_log("已清空 Applet 目录")
+ self._ext_refresh_apps()
+ except Exception as e:
+ self._ext_log(f"清空失败: {e}")
+
+ # ============================================
+ # 提取页 - 正则管理
+ # ============================================
+
+ def _ext_goto_regex(self):
+ self._ext_stack.setCurrentIndex(1)
+
+ def _ext_fill_builtin_patterns(self):
+ """填充内置正则到树"""
+ from src.extractor import Extractor
+ builtin = Extractor.get_all_builtin_patterns()
+ self._ext_builtin_tree.clear()
+ for name, pat in builtin.items():
+ display = pat if len(pat) < 200 else pat[:200] + "..."
+ item = QTreeWidgetItem([name, display])
+ self._ext_builtin_tree.addTopLevelItem(item)
+
+ def _ext_refresh_custom_patterns(self):
+ """刷新自定义正则行列表 — 用 QFrame 行代替 QTableWidget"""
+ lay = self._ext_pat_layout
+ # 清除旧行(保留最后一个 stretch)
+ while lay.count() > 1:
+ item = lay.takeAt(0)
+ w = item.widget()
+ if w:
+ w.deleteLater()
+ c = _TH[self._tn]
+ for name, info in self._ext_custom_patterns.items():
+ if isinstance(info, str):
+ pat = info; enabled = True
+ else:
+ pat = info.get("regex", ""); enabled = info.get("enabled", True)
+ row = QFrame()
+ row.setObjectName("_ext_pat_row")
+ row.setStyleSheet(
+ f"QFrame#_ext_pat_row {{ background: {c['input']}; border-radius: 8px; }}"
+ f"QFrame#_ext_pat_row QLabel {{ background: transparent; border: none; }}")
+ row.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
+ row.customContextMenuRequested.connect(lambda pos, n=name: self._ext_pattern_ctx(pos, n))
+ rl = QHBoxLayout(row)
+ rl.setContentsMargins(12, 6, 12, 6)
+ rl.setSpacing(6)
+ lbl_n = QLabel(name)
+ lbl_n.setFixedWidth(120)
+ lbl_n.setFont(QFont(_FM, 9))
+ lbl_n.setStyleSheet(f"color: {c['text1']};")
+ rl.addWidget(lbl_n)
+ lbl_p = QLabel(pat if len(pat) < 80 else pat[:80] + "...")
+ lbl_p.setFont(QFont(_FM, 9))
+ lbl_p.setStyleSheet(f"color: {c['text2']};")
+ lbl_p.setToolTip(pat)
+ rl.addWidget(lbl_p, 1)
+ lbl_s = QLabel("启用" if enabled else "禁用")
+ lbl_s.setFixedWidth(50)
+ lbl_s.setStyleSheet(f"color: {c['success'] if enabled else c['error']};")
+ rl.addWidget(lbl_s)
+ btn_edit = QPushButton("修改")
+ btn_edit.setFixedWidth(50)
+ btn_edit.clicked.connect(lambda _, n=name: self._ext_edit_pattern(n))
+ rl.addWidget(btn_edit)
+ btn_toggle = QPushButton("禁用" if enabled else "启用")
+ btn_toggle.setFixedWidth(50)
+ if enabled:
+ btn_toggle.setStyleSheet(
+ f"QPushButton {{ background: {c['warning']}; color: #111; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}"
+ f"QPushButton:hover {{ background: #ca8a04; }}")
+ else:
+ btn_toggle.setStyleSheet(
+ f"QPushButton {{ background: {c['success']}; color: #111; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}")
+ btn_toggle.clicked.connect(lambda _, n=name: self._ext_toggle_pattern(n))
+ rl.addWidget(btn_toggle)
+ btn_del = QPushButton("删除")
+ btn_del.setFixedWidth(50)
+ btn_del.setStyleSheet(
+ f"QPushButton {{ background: {c['error']}; color: #fff; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}"
+ f"QPushButton:hover {{ background: #dc2626; }}")
+ btn_del.clicked.connect(lambda _, n=name: self._ext_delete_pattern(n))
+ rl.addWidget(btn_del)
+ lay.insertWidget(lay.count() - 1, row)
+
+ def _ext_add_pattern(self):
+ """新建空白行 — 用 QLineEdit 输入"""
+ c = _TH[self._tn]
+ row = QFrame()
+ row.setObjectName("_ext_new_row")
+ row.setStyleSheet(
+ f"QFrame#_ext_new_row {{ background: {c['input']}; border-radius: 8px; border: 1px solid {c['accent']}; }}"
+ f"QFrame#_ext_new_row QLabel {{ background: transparent; border: none; }}"
+ f"QFrame#_ext_new_row QLineEdit {{ border: 1px solid {c['border']}; border-radius: 4px; padding: 4px 6px; background: {c['card']}; color: {c['text1']}; }}")
+ rl = QHBoxLayout(row)
+ rl.setContentsMargins(12, 6, 12, 6)
+ rl.setSpacing(6)
+ ent_name = QLineEdit()
+ ent_name.setPlaceholderText("栏目名...")
+ ent_name.setFixedWidth(120)
+ rl.addWidget(ent_name)
+ ent_regex = QLineEdit()
+ ent_regex.setPlaceholderText("正则表达式...")
+ rl.addWidget(ent_regex, 1)
+ lbl_s = QLabel("启用")
+ lbl_s.setFixedWidth(50)
+ rl.addWidget(lbl_s)
+ btn_ok = QPushButton("确认")
+ btn_ok.setFixedWidth(50)
+ btn_ok.clicked.connect(lambda: self._ext_confirm_new(ent_name, ent_regex, row))
+ rl.addWidget(btn_ok)
+ btn_cancel = QPushButton("取消")
+ btn_cancel.setFixedWidth(50)
+ btn_cancel.setStyleSheet(
+ f"QPushButton {{ background: {c['error']}; color: #fff; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}"
+ f"QPushButton:hover {{ background: #dc2626; }}")
+ btn_cancel.clicked.connect(lambda: (row.deleteLater(),))
+ rl.addWidget(btn_cancel)
+ self._ext_pat_layout.insertWidget(self._ext_pat_layout.count() - 1, row)
+ ent_name.setFocus()
+
+ def _ext_confirm_new(self, ent_name, ent_regex, row_widget):
+ """确认新建正则"""
+ name = ent_name.text().strip()
+ regex = ent_regex.text().strip()
+ if not name or not regex:
+ self._ext_log("栏目名和正则不能为空")
+ return
+ import re as _re
+ try:
+ _re.compile(regex)
+ except _re.error as e:
+ self._ext_log(f"正则语法错误: {e}")
+ return
+ self._ext_custom_patterns[name] = {"regex": regex, "enabled": True}
+ row_widget.deleteLater()
+ self._ext_refresh_custom_patterns()
+ self._auto_save()
+ self._ext_log(f"已添加规则: {name}")
+
+ def _ext_pattern_ctx(self, pos, name):
+ """右键菜单"""
+ info = self._ext_custom_patterns.get(name, "")
+ regex = info.get("regex", info) if isinstance(info, dict) else info
+ menu = QMenu(self)
+ menu.addAction("测试正则", lambda: self._ext_test_pattern(name))
+ menu.addAction("复制正则", lambda: QApplication.clipboard().setText(regex))
+ # 找到发送信号的 widget
+ sender = self.sender()
+ if sender:
+ menu.exec(sender.mapToGlobal(pos))
+ else:
+ menu.exec(self.cursor().pos())
+
+ def _ext_test_pattern(self, name):
+ """弹窗测试正则"""
+ info = self._ext_custom_patterns.get(name, "")
+ regex = info.get("regex", info) if isinstance(info, dict) else info
+ import re as _re
+
+ dlg = QDialog(self)
+ dlg.setWindowTitle(f"测试正则 - {name}")
+ dlg.resize(600, 400)
+ lay = QVBoxLayout(dlg)
+
+ lay.addWidget(QLabel(f"正则: {regex}"))
+
+ input_lbl = QLabel("输入测试文本:")
+ lay.addWidget(input_lbl)
+ input_box = QTextEdit()
+ input_box.setFont(QFont(_FM, 9))
+ input_box.setPlaceholderText("在此粘贴或输入要测试的文本...")
+ lay.addWidget(input_box, 1)
+
+ result_lbl = QLabel("匹配结果:")
+ lay.addWidget(result_lbl)
+ result_box = QTextEdit()
+ result_box.setReadOnly(True)
+ result_box.setFont(QFont(_FM, 9))
+ lay.addWidget(result_box, 1)
+
+ def do_test():
+ text = input_box.toPlainText()
+ try:
+ pat = _re.compile(regex, _re.IGNORECASE)
+ matches = pat.findall(text)
+ if matches:
+ if isinstance(matches[0], tuple):
+ matches = [m.group(0) for m in pat.finditer(text)]
+ result_box.setPlainText(f"找到 {len(matches)} 个匹配:\n" + "\n".join(str(m) for m in matches))
+ else:
+ result_box.setPlainText("无匹配")
+ except _re.error as e:
+ result_box.setPlainText(f"正则错误: {e}")
+
+ btn_test = QPushButton("测试")
+ btn_test.clicked.connect(do_test)
+ lay.addWidget(btn_test)
+ dlg.exec()
+
+ def _ext_edit_pattern(self, name):
+ """编辑正则 — 替换该行为可编辑的 QLineEdit 行"""
+ info = self._ext_custom_patterns.get(name, "")
+ if isinstance(info, str):
+ pat = info; enabled = True
+ else:
+ pat = info.get("regex", ""); enabled = info.get("enabled", True)
+ c = _TH[self._tn]
+ # 找到并删除原行
+ lay = self._ext_pat_layout
+ idx = -1
+ for i in range(lay.count()):
+ w = lay.itemAt(i).widget()
+ if w:
+ rl = w.layout()
+ if rl and rl.count() > 0:
+ first = rl.itemAt(0).widget()
+ if isinstance(first, QLabel) and first.text() == name:
+ idx = i
+ w.deleteLater()
+ break
+ if idx < 0:
+ idx = lay.count() - 1 # before stretch
+ # 创建编辑行
+ row = QFrame()
+ row.setObjectName("_ext_edit_row")
+ row.setStyleSheet(
+ f"QFrame#_ext_edit_row {{ background: {c['input']}; border-radius: 8px; border: 1px solid {c['accent']}; }}"
+ f"QFrame#_ext_edit_row QLabel {{ background: transparent; border: none; }}"
+ f"QFrame#_ext_edit_row QLineEdit {{ border: 1px solid {c['border']}; border-radius: 4px; padding: 4px 6px; background: {c['card']}; color: {c['text1']}; }}")
+ rl = QHBoxLayout(row)
+ rl.setContentsMargins(12, 6, 12, 6)
+ rl.setSpacing(6)
+ ent_name = QLineEdit(name)
+ ent_name.setFixedWidth(120)
+ rl.addWidget(ent_name)
+ ent_regex = QLineEdit(pat)
+ rl.addWidget(ent_regex, 1)
+ lbl_s = QLabel("启用" if enabled else "禁用")
+ lbl_s.setFixedWidth(50)
+ rl.addWidget(lbl_s)
+ btn_save = QPushButton("保存")
+ btn_save.setFixedWidth(50)
+ btn_save.clicked.connect(lambda: self._ext_save_edit(name, ent_name, ent_regex, enabled, row))
+ rl.addWidget(btn_save)
+ btn_cancel = QPushButton("取消")
+ btn_cancel.setFixedWidth(50)
+ btn_cancel.setStyleSheet(
+ f"QPushButton {{ background: {c['error']}; color: #fff; border-radius: 6px; padding: 4px 8px; font-size: 9px; }}"
+ f"QPushButton:hover {{ background: #dc2626; }}")
+ btn_cancel.clicked.connect(lambda: (row.deleteLater(), self._ext_refresh_custom_patterns()))
+ rl.addWidget(btn_cancel)
+ lay.insertWidget(idx, row)
+ ent_name.setFocus()
+
+ def _ext_save_edit(self, old_name, ent_name, ent_regex, enabled, row_widget):
+ """保存编辑后的正则"""
+ new_name = ent_name.text().strip()
+ new_regex = ent_regex.text().strip()
+ if not new_name or not new_regex:
+ self._ext_log("栏目名和正则不能为空")
+ return
+ import re as _re
+ try:
+ _re.compile(new_regex)
+ except _re.error as e:
+ self._ext_log(f"正则语法错误: {e}")
+ return
+ self._ext_custom_patterns.pop(old_name, None)
+ self._ext_custom_patterns[new_name] = {"regex": new_regex, "enabled": enabled}
+ row_widget.deleteLater()
+ self._ext_refresh_custom_patterns()
+ self._auto_save()
+ self._ext_log(f"已保存规则: {new_name}")
+
+ def _ext_toggle_pattern(self, name):
+ """切换正则启用/禁用"""
+ info = self._ext_custom_patterns.get(name, "")
+ if isinstance(info, str):
+ self._ext_custom_patterns[name] = {"regex": info, "enabled": False}
+ else:
+ info["enabled"] = not info.get("enabled", True)
+ self._ext_refresh_custom_patterns()
+ self._auto_save()
+
+ def _ext_delete_pattern(self, name):
+ self._ext_custom_patterns.pop(name, None)
+ self._ext_refresh_custom_patterns()
+ self._auto_save()
+ self._ext_log(f"已删除规则: {name}")
+
# ── 调试开关 (vConsole) ──
def _build_vconsole(self):
@@ -1526,6 +2789,7 @@ class App(QMainWindow):
self._update_theme_label()
self._update_toggle_colors()
self._refresh_sb_app_card()
+ self._ext_refresh_custom_patterns()
self._hl_sb()
self._auto_save()
@@ -1535,7 +2799,7 @@ class App(QMainWindow):
def _update_toggle_colors(self):
c = _TH[self._tn]
- for tog in (self._tog_dm, self._tog_df):
+ for tog in (self._tog_dm, self._tog_df, self._tog_auto_dec, self._tog_auto_scan):
tog.set_colors(c["accent"], c["text4"])
def _refresh_sb_app_card(self):
@@ -1550,11 +2814,14 @@ class App(QMainWindow):
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),
+ "extract_packages_dir": self._ext_path_ent.text(),
+ "extract_custom_patterns": dict(self._ext_custom_patterns),
+ "auto_decompile": self._tog_auto_dec.isChecked(),
+ "auto_scan": self._tog_auto_scan.isChecked(),
}
_save_cfg(data)
@@ -1608,7 +2875,6 @@ class App(QMainWindow):
if self._running:
return
try:
- dp = int(self._dp_ent.text())
cp = int(self._cp_ent.text())
except ValueError:
self._log_add("error", "[gui] 端口号无效")
@@ -1616,7 +2882,7 @@ class App(QMainWindow):
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,
+ cdp_port=cp,
debug_main=self._tog_dm.isChecked(),
debug_frida=self._tog_df.isChecked(),
scripts_dir=scripts_dir,
@@ -1702,7 +2968,7 @@ class App(QMainWindow):
def _nav_btns(self, on):
for b in (self._btn_go, self._btn_relaunch,
- self._btn_back, self._btn_auto, self._btn_prev,
+ self._btn_back, self._btn_refresh, self._btn_auto, self._btn_prev,
self._btn_next, self._btn_copy_route):
b.setEnabled(on)
self._guard_switch.setEnabled(on)
@@ -1776,6 +3042,7 @@ class App(QMainWindow):
# 自动恢复云扫描
if not self._cloud_scan_active and self._auditor:
self._cloud_start_scan()
+ self._log_add("info", "[云扫描] 小程序连接后自动恢复捕获")
def _delayed_fetch_app_info(self, gen):
"""延迟调用,只有最后一次触发的 gen 匹配才执行。"""
@@ -1856,6 +3123,29 @@ class App(QMainWindow):
except Exception as e:
self._log_q.put(("error", f"[导航] 返回失败: {e}"))
+ def _do_refresh(self):
+ if self._engine and self._loop:
+ asyncio.run_coroutine_threadsafe(self._arefresh(), self._loop)
+
+ async def _arefresh(self):
+ try:
+ result = await self._navigator.refresh_page()
+ if result:
+ import json as _json
+ try:
+ info = _json.loads(result)
+ if info.get("err"):
+ self._log_q.put(("error", f"[导航] 刷新失败: {info['err']}"))
+ return
+ route = info.get("route", "")
+ self._log_q.put(("info", f"[导航] 已刷新页面: /{route}"))
+ except (_json.JSONDecodeError, TypeError):
+ self._log_q.put(("info", "[导航] 已刷新页面"))
+ else:
+ 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", "[导航] 请先获取路由")
@@ -2281,6 +3571,12 @@ class App(QMainWindow):
except queue.Empty:
break
self._handle_cld(item)
+ for _ in range(50):
+ try:
+ item = self._ext_q.get_nowait()
+ except queue.Empty:
+ break
+ self._handle_ext(item)
def _apply_sts(self, sts):
c = _TH[self._tn]
@@ -2467,11 +3763,93 @@ class App(QMainWindow):
self._cloud_result.setHtml(
f'{name} -> {detail}')
+ def _handle_ext(self, item):
+ kind = item.get("type", "")
+ c = _TH[self._tn]
+ op = self._ext_current_op # ("decompile"/"scan", appid) or None
+
+ if kind == "progress":
+ done = item.get("done", 0)
+ total = item.get("total", 1)
+ if total > 0:
+ self._ext_prog.setValue(int((done / total) * 100))
+ self._ext_status_lbl.setText(f"进度: {done}/{total}")
+
+ elif kind == "log":
+ self._ext_log(item.get("msg", ""))
+
+ elif kind == "result":
+ data = item.get("data", {})
+ if op:
+ op_type, appid = op
+ if op_type == "decompile":
+ # 反编译完成
+ state = self._ext_app_states.setdefault(appid, {})
+ state["decompiled"] = True
+ decompile_dir = data.get("decompile_dir", "")
+ if decompile_dir:
+ state["decompile_dir"] = decompile_dir
+ extracted = data.get("extracted", 0)
+ self._ext_status_lbl.setText(f"反编译完成! {appid} 提取了 {extracted} 个文件")
+ self._ext_status_lbl.setStyleSheet(f"color: {c['success']};")
+ self._ext_update_app_buttons(appid)
+ # 刷新名称显示(解包后可读取 app-config.json)
+ widgets = self._ext_app_widgets.get(appid, {})
+ if "lbl_name" in widgets:
+ pkgs = state.get("packages", [])
+ output_base = os.path.join(_BASE_DIR, "output")
+ name = self._ext_get_app_name(appid, pkgs, output_base)
+ widgets["lbl_name"].setText(name)
+
+ elif op_type == "scan":
+ # 扫描完成
+ state = self._ext_app_states.setdefault(appid, {})
+ state["scanned"] = True
+ result_dir = data.get("result_dir", "")
+ if result_dir:
+ state["result_dir"] = result_dir
+ findings = data.get("findings", 0)
+ self._ext_status_lbl.setText(f"扫描完成! {appid} 发现 {findings} 条敏感信息")
+ self._ext_status_lbl.setStyleSheet(f"color: {c['success']};")
+ self._ext_update_app_buttons(appid)
+
+ elif kind == "error":
+ self._ext_log(f"错误: {item.get('msg', '')}")
+ self._ext_status_lbl.setText("出错")
+ self._ext_status_lbl.setStyleSheet(f"color: {c['error']};")
+
+ elif kind == "__done__":
+ self._ext_proc = None
+ prev_op = self._ext_current_op
+ self._ext_current_op = None
+ self._ext_prog.setValue(100)
+ rc = item.get("returncode", -1)
+ if rc != 0 and "完成" not in self._ext_status_lbl.text():
+ self._ext_status_lbl.setText(f"进程退出 (code={rc})")
+ self._ext_status_lbl.setStyleSheet(f"color: {c['error']};")
+
+ # 自动处理链: 反编译完成后 → 自动扫描; 扫描完成后 → 处理下一个
+ if prev_op and rc == 0:
+ op_type, appid = prev_op
+ if op_type == "decompile" and self._tog_auto_scan.isChecked():
+ state = self._ext_app_states.get(appid, {})
+ if state.get("decompiled") and not state.get("scanned"):
+ QTimer.singleShot(500, lambda a=appid: self._ext_do_scan(a))
+ return
+ # 继续处理其他未完成的小程序
+ QTimer.singleShot(500, self._ext_auto_process_pending)
# ──────────────────────────────────
# 退出
# ──────────────────────────────────
def closeEvent(self, event):
+ if self._ext_proc:
+ try:
+ self._ext_proc.kill() # 强制杀死子进程
+ self._ext_proc.wait(timeout=2)
+ except Exception:
+ pass
+ self._ext_proc = None
if self._running:
self._do_stop()
QTimer.singleShot(400, lambda: QApplication.quit())
diff --git a/src/__pycache__/__init__.cpython-310.pyc b/src/__pycache__/__init__.cpython-310.pyc
index e558f38335c627800449593931d457ae508ee5c7..c5c66f878f3cf8ea9f1dbf0a34e916fba6c29667 100644
GIT binary patch
delta 65
zcmeBSoXE(X&&$ij00idi?@Z*j(K68wElw>e)-Nm0&&ez)E!KC*FVD-#PfRJ+cgrj)
TF44_R%*@j-E=ty)n4T9-cynzmWJweD36YUfR9t%u%MmT<*Q(7AD=Wb!n@@H{la$w8|xtv>uwG zju&ep`F} OaDj}HjPj|EjJV_sTm@D!3t9IYC{dxxis47OeI_MazhI> z^pv4eL&HeWie)Et{%nv5WtQ89kyT`7oR`oFhk(%2gY2u)m@LTF3iSifL0?9wg~k?) zE8kxKTTC-1w&ukvx8HlGMRtqdfA#kJziGa5wz=>-Fxm1SUT C4T9^UFUzYu9E=e6|);S#`=Cv}<0?_9}i*ESJ~Sc!nlP^+(V+>FMov zK)C4X_Ino>e{g;A+WEyR=g>_|bjhez2-}>{xI)Q)QF5#P0Mf;4?<~Fblh{;DCwBK# z%%^wcymyyiPFub5;oZA;*I}EawBa-lf$f*EfG5m`9LndU{n9a +e2&xnqRhxtt;&O8&Z|69%Wfq~A1jvYZY ^>Q5~&!oW0btwgv*`wMB8PB3AYH3iL{{Tm7N?- zZnZ9+;tnz)=GnG>K4ST7)LM!qXYsbH?TRLEmo24T?SN#vnw2|I;-81ox*Z^5BBm1T z(ue&Bumb?M3^z-+T#u>P5h;`bekD{Gx!MlY>Bt*bHD<=n1V^yGC^g`ybHe39>>^8V zk%S8=Ohs#zB>#_&_9=AqmQFG%BQ#hBqnf~|vdGh@a-VusX3N^R)UX0AG*8P1q!ZOf zisW086`9odWm&4QPncc0Wd}1~+wNb`?g<)cET{E&K_^;Z?H82>2qYA1g4P??^!9IP z)$<9xWm~;{f=;fb6Iyddm mb0-y0n9p4(rmy ~Cv!?wA2_z9^#gG&Dx5+Wc?u~bX$kS%#TzO8qy zOjELQy9|;-J(1P@$+xB3p=fFwEVRpQvT{mO_&!>>OibW~8lti&U9AVTdQe~CxWp7q(xlTW{LJc?{LRzAH2t2%iz8kOmD5T? z57bbnV;lZrh6em)=y|1Kfg7wjjlq}ZbVgsuz=zgL693YR4+q8z=9~q;wjfWVloFD3 zn4U+Fz|4yZdXU02voNNd7*i%pkxzYWRE4#p3eCAJ*?{jo9GdXGRQ+q1UJ@Dn?l U0rDY{>rLTE?&Bf`w5QswXWGb_r~)4&m)D89=;>*z9T<>Qt9nI z3J_cF3?kn`buv<>NAtNzs{)&m#&KAQ(2!gu}I~`X(#p3`S(T0os>Hh@M0-&8UGtn z{xS(IfIz_EUUar4+6g?bLD}y@l2y|3_Pc4ojUd b2C`+zc z9Cv+zFN$_r5B*m_7!&v+yTFEw;8C7Kh&nI(I0q1RN^>fefSA1$V4_w7pDI9vn&<*O z{>~RSgbG4RX-*GOhW10?W4&PjZ6-?F8y0alljmhVl2`%GU tdA5{29%1Tp2B zCh*bo;={0vao|?)M+|ZC{68;Vx@5QJWZ#R4^QRE?t#YV{KXDCTeEE7>+>K)OrT717 z@y#=cE0dMk{Vgz+#~QnB`}TX^B?G;&e PiS%Ef6YJ z3&cJs;8xj)&j?}Yq?uUHYmqKYaYBEL1q!9}DLz61l194c18k9526~a<2T7<+7t51? zbcOlTUN9l>7n$R3Kt1NuoH)z@gA?ufkxra9>zL?^Fu;!p6TDgtaDO2 t Q`+P#K&UC30md}=PD6=)Y1#U#a!UCh z<*t_D-$a8F-qnYKXM&=D;S `AoGq -8cp>GOs@67|<=dfn&9Bl{eyz z6kaZNGmW6WSYdS^M=I?kKZ+!;I4P18CUDzvO#F!?w&|WzYAd!Ek@Q#*dkUiX$fy4u zCVUFZ(>Q*zvVp@2obX{*)35=2lsecMnS}zYmI?))ML8bT0CMOJTa8OIr(KAQV77)i za+Q=QnBa7kb#g=EGS)jJVtsMJa5g2fe+ayC@6$!bs2yE?`A%xGF|MZjThh$gvPQJ7 z6|otou!d9g#5Yhv@qDDh!T9H>xQUX_Q$ojQOl-hzDG+cHlU##Wv2x;qVeOr&vU=Iw zN4)3nMWRX8eY8MXd9$pDvzstWXE JItO8Kj1(c;msL+#Wb-1*?wI=6N1{9h@k BG93T_ literal 0 HcmV?d00001 diff --git a/src/cli.py b/src/cli.py index c8f154f..3d84121 100644 --- a/src/cli.py +++ b/src/cli.py @@ -7,9 +7,9 @@ CDP_PORT = 62000 class CliOptions: - def __init__(self, debug_port: int, cdp_port: int, debug_main: bool, debug_frida: bool, + def __init__(self, cdp_port: int, debug_main: bool, debug_frida: bool, scripts_dir: str = "", script_files: list = None): - self.debug_port = debug_port + self.debug_port = DEBUG_PORT # 小程序硬编码 9421,不可修改 self.cdp_port = cdp_port self.debug_main = debug_main self.debug_frida = debug_frida @@ -35,18 +35,16 @@ def parse_cli_options() -> CliOptions: formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f"""Examples: python main.py - python main.py --debug-port 9421 --cdp-port 62000 + python main.py --cdp-port 62000 python main.py --debug-main --debug-frida Default ports: - Debug server: {DEBUG_PORT} + Debug server: {DEBUG_PORT} (fixed) CDP proxy: {CDP_PORT} Usage: Connect Chrome DevTools to: devtools://devtools/bundled/inspector.html?ws=127.0.0.1: """ ) - parser.add_argument("--debug-port", type=str, default=None, - help=f"Remote debug server port (default: {DEBUG_PORT})") parser.add_argument("--cdp-port", type=str, default=None, help=f"CDP proxy server port (default: {CDP_PORT})") parser.add_argument("--debug-main", action="store_true", default=False, @@ -72,7 +70,6 @@ Usage: script_files.append(os.path.abspath(f)) return CliOptions( - debug_port=parse_port("--debug-port", args.debug_port, DEBUG_PORT), cdp_port=parse_port("--cdp-port", args.cdp_port, CDP_PORT), debug_main=args.debug_main, debug_frida=args.debug_frida, diff --git a/src/cloud_audit.py b/src/cloud_audit.py index 5ea4bcf..ce6c3ff 100644 --- a/src/cloud_audit.py +++ b/src/cloud_audit.py @@ -25,6 +25,7 @@ class CloudAuditor: """启动全局 Hook:注入 JS,安装 hook,启动自动扫描新 frame""" if self._enabled: return {"ok": True, "already": True} + self._injected = False # 强制重新注入(连接可能已更换) await self.inject() result = await self.engine.evaluate_js( "JSON.stringify(window.cloudAudit.installHook())", timeout=5.0) @@ -52,11 +53,23 @@ class CloudAuditor: pass async def poll(self): - """拉取新捕获的云函数调用,只返回增量""" + """拉取新捕获的云函数调用,只返回增量。 + 自动检测 Hook 是否丢失(页面切换后),并重新注入。 + """ if not self._enabled: return [] try: await self.inject() + # 检查 Hook 是否仍然存活(页面导航后可能丢失) + alive_result = await self.engine.evaluate_js( + "(function(){try{return window.cloudAudit&&window.cloudAudit._hooked?'1':'0'}catch(e){return '0'}})()", + timeout=3.0) + alive = self._extract_value(alive_result) + if alive != '1': + # Hook 丢失(页面切换/刷新),重新安装 + await self.engine.evaluate_js( + "JSON.stringify(window.cloudAudit.installHook())", timeout=5.0) + self._seen_count = 0 # 重置计数器,新上下文从 0 开始 result = await self.engine.evaluate_js( "JSON.stringify(window.cloudAudit.getHookedCalls())", timeout=5.0) value = self._extract_value(result) diff --git a/src/extract_worker.py b/src/extract_worker.py new file mode 100644 index 0000000..e0d978c --- /dev/null +++ b/src/extract_worker.py @@ -0,0 +1,194 @@ +""" +extract_worker.py — 独立子进程 worker +由 GUI 通过 subprocess 启动,避免阻塞主 UI + +支持两种模式: + 1. decompile —— 解密解包单个小程序 + 2. scan —— 扫描已解包目录的敏感信息 + +通信协议 (stdout JSON lines): + {"type": "progress", "done": N, "total": N} + {"type": "log", "msg": "..."} + {"type": "result", "data": {...}} + {"type": "error", "msg": "..."} + +命令行: + python extract_worker.py decompile --packages-dir DIR --appid APPID --output-dir DIR + python extract_worker.py scan --scan-dir DIR --output-dir DIR [--custom-patterns JSON_FILE] +""" +import json +import os +import sys +import traceback + + +def _init_stdout(): + """Windows 下强制 stdout 使用 UTF-8,避免 GBK 编码错误""" + if sys.platform == "win32": + import io + sys.stdout = io.TextIOWrapper( + sys.stdout.buffer, encoding="utf-8", errors="replace" + ) + sys.stderr = io.TextIOWrapper( + sys.stderr.buffer, encoding="utf-8", errors="replace" + ) + + +def _emit(obj): + """输出一行 JSON 到 stdout""" + print(json.dumps(obj, ensure_ascii=False), flush=True) + + +def do_decompile(args): + """解密解包单个小程序""" + src_dir = os.path.dirname(os.path.abspath(__file__)) + base_dir = os.path.dirname(src_dir) + if base_dir not in sys.path: + sys.path.insert(0, base_dir) + + from src.wxapkg import find_wxapkg_files, extract_wxapkg + + packages_dir = args.packages_dir + appid = args.appid + output_dir = args.output_dir + + os.makedirs(output_dir, exist_ok=True) + + _emit({"type": "log", "msg": f"正在查找 {appid} 的 wxapkg 文件..."}) + + # 查找该 appid 的包 + all_pkgs = find_wxapkg_files(packages_dir) + pkgs = [p for p in all_pkgs if p["appid"] == appid] + + if not pkgs: + _emit({"type": "log", "msg": f"未找到 {appid} 的 wxapkg 文件"}) + _emit({"type": "result", "data": {"appid": appid, "files": 0, "extracted": 0}}) + return + + _emit({"type": "log", "msg": f"发现 {len(pkgs)} 个 wxapkg 文件"}) + + decompile_dir = os.path.join(output_dir, "decompiled") + os.makedirs(decompile_dir, exist_ok=True) + + total = len(pkgs) + extracted_total = 0 + + for i, pkg in enumerate(pkgs, 1): + try: + files = extract_wxapkg(pkg["path"], decompile_dir, appid) + extracted_total += len(files) + _emit({"type": "log", "msg": f" 解包 {pkg['name']}: {len(files)} 个文件"}) + except Exception as e: + _emit({"type": "log", "msg": f" 解包失败 {pkg['name']}: {e}"}) + _emit({"type": "progress", "done": i, "total": total}) + + _emit({"type": "result", "data": { + "appid": appid, + "files": total, + "extracted": extracted_total, + "decompile_dir": decompile_dir, + }}) + _emit({"type": "log", "msg": f"反编译完成! 共提取 {extracted_total} 个文件"}) + + +def do_scan(args): + """扫描已解包的目录""" + src_dir = os.path.dirname(os.path.abspath(__file__)) + base_dir = os.path.dirname(src_dir) + if base_dir not in sys.path: + sys.path.insert(0, base_dir) + + from src.extractor import Extractor + + scan_dir = args.scan_dir + output_dir = args.output_dir + + os.makedirs(output_dir, exist_ok=True) + + extractor = Extractor() + + # 加载自定义正则 + if args.custom_patterns and os.path.isfile(args.custom_patterns): + try: + with open(args.custom_patterns, "r", encoding="utf-8") as f: + custom = json.load(f) + loaded = 0 + for name, info in custom.items(): + # 兼容新格式 {"regex": str, "enabled": bool} 和旧格式 str + if isinstance(info, dict): + if not info.get("enabled", True): + continue + pat = info.get("regex", "") + else: + pat = info + if pat: + extractor.add_custom_pattern(name, pat) + loaded += 1 + _emit({"type": "log", "msg": f"已加载 {loaded} 个自定义正则"}) + except Exception as e: + _emit({"type": "log", "msg": f"加载自定义正则失败: {e}"}) + + _emit({"type": "log", "msg": f"正在扫描目录: {scan_dir}"}) + + def on_progress(done, total): + _emit({"type": "progress", "done": done, "total": total}) + + scan_result = extractor.scan_directory(scan_dir, on_progress=on_progress) + + # 导出 + result_dir = output_dir + html_path = os.path.join(result_dir, "report.html") + json_path = os.path.join(result_dir, "report.json") + Extractor.export_html(scan_result, html_path) + Extractor.export_json(scan_result, json_path) + + summary = Extractor.get_summary(scan_result) + total_findings = sum(summary.values()) + + _emit({"type": "result", "data": { + "files_scanned": scan_result.get("files_scanned", 0), + "elapsed": scan_result.get("elapsed", 0), + "findings": total_findings, + "summary": summary, + "html_path": html_path, + "json_path": json_path, + "result_dir": result_dir, + }}) + _emit({"type": "log", "msg": f"扫描完成! 发现 {total_findings} 条敏感信息,耗时 {scan_result.get('elapsed', 0)}s"}) + + +def main(): + import argparse + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command") + + # decompile 子命令 + p_dec = subparsers.add_parser("decompile") + p_dec.add_argument("--packages-dir", required=True) + p_dec.add_argument("--appid", required=True) + p_dec.add_argument("--output-dir", required=True) + + # scan 子命令 + p_scan = subparsers.add_parser("scan") + p_scan.add_argument("--scan-dir", required=True) + p_scan.add_argument("--output-dir", required=True) + p_scan.add_argument("--custom-patterns", default="") + + args = parser.parse_args() + + if args.command == "decompile": + do_decompile(args) + elif args.command == "scan": + do_scan(args) + else: + _emit({"type": "error", "msg": "未知命令,使用 decompile 或 scan"}) + sys.exit(1) + + +if __name__ == "__main__": + _init_stdout() + try: + main() + except Exception as e: + _emit({"type": "error", "msg": f"worker 异常: {traceback.format_exc()}"}) + sys.exit(1) diff --git a/src/extractor.py b/src/extractor.py new file mode 100644 index 0000000..48c646c --- /dev/null +++ b/src/extractor.py @@ -0,0 +1,978 @@ +""" +extractor.py — 敏感信息提取引擎 +直接使用 heartk_fast.py 的完整正则规则,保证提取结果一致 +支持自定义正则规则,自定义类别会出现在导出报告中 + +用法: + from src.extractor import Extractor + ext = Extractor() + ext.add_custom_pattern("视频", r"video_url\s*=\s*['\"](.+?)['\"]") + results = ext.scan_directory("/path/to/dir") + Extractor.export_html(results, "report.html") +""" +import json +import os +import re +import sys +import time +from concurrent.futures import ProcessPoolExecutor, as_completed +from typing import Dict, List, Optional, Any, Callable + +try: + import chardet +except ImportError: + chardet = None + +# ============================================================ +# 1. 所有正则规则 (从 heartk_fast.py 原封不动复制) +# ============================================================ + +STATIC_EXTENSIONS = ['.jpg', '.png', '.gif', '.css', '.svg', '.ico', '.js'] + +CATEGORIES = [ + "ip", "ip_port", "domain", "path", "incomplete_path", + "url", "sfz", "mobile", "mail", "jwt", "algorithm", "secret", "static" +] + +# ---------- extract_info 对应的正则 ---------- + +EXTRACT_PATTERNS: Dict[str, re.Pattern] = { + 'sfz': re.compile( + r"""['"]((\d{8}(0\d|10|11|12)([0-2]\d|30|31)\d{3})|(\d{6}(18|19|20)\d{2}(0[1-9]|10|11|12)([0-2]\d|30|31)\d{3}(\d|X|x)))['"]"""), + 'mobile': re.compile( + r"""['\"](1(3([0-35-9]\d|4[1-8])|4[14-9]\d|5([\d]\d|7[1-79])|66\d|7[2-35-8]\d|8\d{2}|9[89]\d)\d{7})['\"]"""), + 'mail': re.compile( + r"""['\"][a-zA-Z0-9\._\-]*@[a-zA-Z0-9\._\-]{1,63}\.((?!js|css|jpg|jpeg|png|ico)[a-zA-Z]{2,})['\"]"""), + 'ip': re.compile( + r"""['\"](([a-zA-Z0-9]+:)?\/\/)?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(\/.*?)?['\"]"""), + 'ip_port': re.compile( + r"""['\"](([a-zA-Z0-9]+:)?\/\/)?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\:\d{1,5}(\/.*?)?['\"]"""), + 'domain': re.compile( + r"""['\"](([a-zA-Z0-9]+:)?\/\/)?[a-zA-Z0-9\-\.]*?\.(xin|com|cn|net|com\.cn|vip|top|cc|shop|club|wang|xyz|luxe|site|news|pub|fun|online|win|red|loan|ren|mom|net\.cn|org|link|biz|bid|help|tech|date|mobi|so|me|tv|co|vc|pw|video|party|pics|website|store|ltd|ink|trade|live|wiki|space|gift|lol|work|band|info|click|photo|market|tel|social|press|game|kim|org\.cn|games|pro|men|love|studio|rocks|asia|group|science|design|software|engineer|lawyer|fit|beer|tw|我爱你|中国|公司|网络|在线|网址|网店|集团|中文网)(\:\d{1,5})?(\/)?['\"]"""), + 'path': re.compile( + r"""['\"](?:\/|\.\.\/)(?:[^\/\>\< \)\(\{\}\,\'\"\\\n][^\>\< \)\(\{\}\,\'\"\\\n]*?)['\"]"""), + 'incomplete_path': re.compile( + r"""['\"][^\/\>\< \)\(\{\}\,\'\"\\\n][\w\/]*?\/[\w\/]*?['\"]"""), + 'url': re.compile( + r"""['\"](([a-zA-Z0-9]+:)?\/\/)?[a-zA-Z0-9\-\.]*?\.(xin|com|cn|net|com\.cn|vip|top|cc|shop|club|wang|xyz|luxe|site|news|pub|fun|online|win|red|loan|ren|mom|net\.cn|org|link|biz|bid|help|tech|date|mobi|so|me|tv|co|vc|pw|video|party|pics|website|store|ltd|ink|trade|live|wiki|space|gift|lol|work|band|info|click|photo|market|tel|social|press|game|kim|org\.cn|games|pro|men|love|studio|rocks|asia|group|science|design|software|engineer|lawyer|fit|beer|tw|我爱你|中国|公司|网络|在线|网址|网店|集团|中文网)(\:\d{1,5})?(\/.*?)?['\"]"""), + 'jwt': re.compile( + r"""['\"](ey[A-Za-z0-9_-]{10,}\.[A-Za-z0-9._-]{10,}|ey[A-Za-z0-9_\/+-]{10,}\.[A-Za-z0-9._\/+-]{10,})['\"]"""), + 'algorithm': re.compile( + r"""\W(Base64\.encode|Base64\.decode|btoa|atob|CryptoJS\.AES|CryptoJS\.DES|JSEncrypt|rsa|KJUR|\$\.md5|md5|sha1|sha256|sha512)[\(\.]""", re.IGNORECASE), +} + +# ---------- nuclei_regex (727条秘密检测规则) ---------- + +_NUCLEI_REGEX_SOURCES = [ + "zopim[_-]?account[_-]?key", "zhuliang[_-]?gh[_-]?token", + "zensonatypepassword", "zendesk[_-]?travis[_-]?github", + "yt[_-]?server[_-]?api[_-]?key", "yt[_-]?partner[_-]?refresh[_-]?token", + "yt[_-]?partner[_-]?client[_-]?secret", "yt[_-]?client[_-]?secret", + "yt[_-]?api[_-]?key", "yt[_-]?account[_-]?refresh[_-]?token", + "yt[_-]?account[_-]?client[_-]?secret", "yangshun[_-]?gh[_-]?token", + "yangshun[_-]?gh[_-]?password", "www[_-]?googleapis[_-]?com", + "wpt[_-]?ssh[_-]?private[_-]?key[_-]?base64", "wpt[_-]?ssh[_-]?connect", + "wpt[_-]?report[_-]?api[_-]?key", "wpt[_-]?prepare[_-]?dir", + "wpt[_-]?db[_-]?user", "wpt[_-]?db[_-]?password", "wporg[_-]?password", + "wpjm[_-]?phpunit[_-]?google[_-]?geocode[_-]?api[_-]?key", + "wordpress[_-]?db[_-]?user", "wordpress[_-]?db[_-]?password", + "wincert[_-]?password", "widget[_-]?test[_-]?server", + "widget[_-]?fb[_-]?password[_-]?3", "widget[_-]?fb[_-]?password[_-]?2", + "widget[_-]?fb[_-]?password", + "widget[_-]?basic[_-]?password[_-]?5", "widget[_-]?basic[_-]?password[_-]?4", + "widget[_-]?basic[_-]?password[_-]?3", "widget[_-]?basic[_-]?password[_-]?2", + "widget[_-]?basic[_-]?password", + "watson[_-]?password", "watson[_-]?device[_-]?password", + "watson[_-]?conversation[_-]?password", "wakatime[_-]?api[_-]?key", + "vscetoken", "visual[_-]?recognition[_-]?api[_-]?key", + "virustotal[_-]?apikey", "vip[_-]?github[_-]?deploy[_-]?key[_-]?pass", + "vip[_-]?github[_-]?deploy[_-]?key", + "vip[_-]?github[_-]?build[_-]?repo[_-]?deploy[_-]?key", + "v[_-]?sfdc[_-]?password", "v[_-]?sfdc[_-]?client[_-]?secret", + "usertravis", "user[_-]?assets[_-]?secret[_-]?access[_-]?key", + "user[_-]?assets[_-]?access[_-]?key[_-]?id", "use[_-]?ssh", + "us[_-]?east[_-]?1[_-]?elb[_-]?amazonaws[_-]?com", + "urban[_-]?secret", "urban[_-]?master[_-]?secret", "urban[_-]?key", + "unity[_-]?serial", "unity[_-]?password", + "twitteroauthaccesstoken", "twitteroauthaccesssecret", + "twitter[_-]?consumer[_-]?secret", "twitter[_-]?consumer[_-]?key", + "twine[_-]?password", "twilio[_-]?token", "twilio[_-]?sid", + "twilio[_-]?configuration[_-]?sid", + "twilio[_-]?chat[_-]?account[_-]?api[_-]?service", + "twilio[_-]?api[_-]?secret", "twilio[_-]?api[_-]?key", + "trex[_-]?okta[_-]?client[_-]?token", "trex[_-]?client[_-]?token", + "travis[_-]?token", "travis[_-]?secure[_-]?env[_-]?vars", + "travis[_-]?pull[_-]?request", "travis[_-]?gh[_-]?token", + "travis[_-]?e2e[_-]?token", "travis[_-]?com[_-]?token", + "travis[_-]?branch", "travis[_-]?api[_-]?token", + "travis[_-]?access[_-]?token", "token[_-]?core[_-]?java", + "thera[_-]?oss[_-]?access[_-]?key", "tester[_-]?keys[_-]?password", + "test[_-]?test", "test[_-]?github[_-]?token", "tesco[_-]?api[_-]?key", + "svn[_-]?pass", "surge[_-]?token", "surge[_-]?login", + "stripe[_-]?public", "stripe[_-]?private", + "strip[_-]?secret[_-]?key", "strip[_-]?publishable[_-]?key", + "stormpath[_-]?api[_-]?key[_-]?secret", "stormpath[_-]?api[_-]?key[_-]?id", + "starship[_-]?auth[_-]?token", "starship[_-]?account[_-]?sid", + "star[_-]?test[_-]?secret[_-]?access[_-]?key", "star[_-]?test[_-]?location", + "star[_-]?test[_-]?bucket", "star[_-]?test[_-]?aws[_-]?access[_-]?key[_-]?id", + "staging[_-]?base[_-]?url[_-]?runscope", "ssmtp[_-]?config", "sshpass", + "srcclr[_-]?api[_-]?token", + "square[_-]?reader[_-]?sdk[_-]?repository[_-]?password", + "sqssecretkey", "sqsaccesskey", "spring[_-]?mail[_-]?password", + "spotify[_-]?api[_-]?client[_-]?secret", "spotify[_-]?api[_-]?access[_-]?token", + "spaces[_-]?secret[_-]?access[_-]?key", "spaces[_-]?access[_-]?key[_-]?id", + "soundcloud[_-]?password", "soundcloud[_-]?client[_-]?secret", + "sonatypepassword", "sonatype[_-]?token[_-]?user", + "sonatype[_-]?token[_-]?password", "sonatype[_-]?password", + "sonatype[_-]?pass", "sonatype[_-]?nexus[_-]?password", + "sonatype[_-]?gpg[_-]?passphrase", "sonatype[_-]?gpg[_-]?key[_-]?name", + "sonar[_-]?token", "sonar[_-]?project[_-]?key", + "sonar[_-]?organization[_-]?key", + "socrata[_-]?password", "socrata[_-]?app[_-]?token", + "snyk[_-]?token", "snyk[_-]?api[_-]?token", + "snoowrap[_-]?refresh[_-]?token", "snoowrap[_-]?password", + "snoowrap[_-]?client[_-]?secret", "slate[_-]?user[_-]?email", + "slash[_-]?developer[_-]?space[_-]?key", "slash[_-]?developer[_-]?space", + "signing[_-]?key[_-]?sid", "signing[_-]?key[_-]?secret", + "signing[_-]?key[_-]?password", "signing[_-]?key", + "setsecretkey", "setdstsecretkey", "setdstaccesskey", + "ses[_-]?secret[_-]?key", "ses[_-]?access[_-]?key", + "service[_-]?account[_-]?secret", "sentry[_-]?key", + "sentry[_-]?secret", "sentry[_-]?endpoint", + "sentry[_-]?default[_-]?org", "sentry[_-]?auth[_-]?token", + "sendwithus[_-]?key", "sendgrid[_-]?username", "sendgrid[_-]?user", + "sendgrid[_-]?password", "sendgrid[_-]?key", "sendgrid[_-]?api[_-]?key", + "sendgrid", "selion[_-]?selenium[_-]?host", "selion[_-]?log[_-]?level[_-]?dev", + "segment[_-]?api[_-]?key", "secretid", "secretkey", "secretaccesskey", + "secret[_-]?key[_-]?base", + "secret[_-]?9", "secret[_-]?8", "secret[_-]?7", "secret[_-]?6", + "secret[_-]?5", "secret[_-]?4", "secret[_-]?3", "secret[_-]?2", + "secret[_-]?11", "secret[_-]?10", "secret[_-]?1", "secret[_-]?0", + "sdr[_-]?token", "scrutinizer[_-]?token", "sauce[_-]?access[_-]?key", + "sandbox[_-]?aws[_-]?secret[_-]?access[_-]?key", + "sandbox[_-]?aws[_-]?access[_-]?key[_-]?id", "sandbox[_-]?access[_-]?token", + "salesforce[_-]?bulk[_-]?test[_-]?security[_-]?token", + "salesforce[_-]?bulk[_-]?test[_-]?password", + "sacloud[_-]?api", "sacloud[_-]?access[_-]?token[_-]?secret", + "sacloud[_-]?access[_-]?token", + "s3[_-]?user[_-]?secret", "s3[_-]?secret[_-]?key", + "s3[_-]?secret[_-]?assets", "s3[_-]?secret[_-]?app[_-]?logs", + "s3[_-]?key[_-]?assets", "s3[_-]?key[_-]?app[_-]?logs", "s3[_-]?key", + "s3[_-]?external[_-]?3[_-]?amazonaws[_-]?com", + "s3[_-]?bucket[_-]?name[_-]?assets", "s3[_-]?bucket[_-]?name[_-]?app[_-]?logs", + "s3[_-]?access[_-]?key[_-]?id", "s3[_-]?access[_-]?key", + "rubygems[_-]?auth[_-]?token", "rtd[_-]?store[_-]?pass", "rtd[_-]?key[_-]?pass", + "route53[_-]?access[_-]?key[_-]?id", + "ropsten[_-]?private[_-]?key", "rinkeby[_-]?private[_-]?key", + "rest[_-]?api[_-]?key", "repotoken", + "reporting[_-]?webdav[_-]?url", "reporting[_-]?webdav[_-]?pwd", + "release[_-]?token", "release[_-]?gh[_-]?token", + "registry[_-]?secure", "registry[_-]?pass", "refresh[_-]?token", + "rediscloud[_-]?url", "redis[_-]?stunnel[_-]?urls", + "randrmusicapiaccesstoken", "rabbitmq[_-]?password", + "quip[_-]?token", "qiita[_-]?token", "pypi[_-]?passowrd", + "pushover[_-]?token", "publish[_-]?secret", "publish[_-]?key", + "publish[_-]?access", "project[_-]?config", + "prod[_-]?secret[_-]?key", "prod[_-]?password", "prod[_-]?access[_-]?key[_-]?id", + "private[_-]?signing[_-]?password", "pring[_-]?mail[_-]?username", + "preferred[_-]?username", "prebuild[_-]?auth", + "postgresql[_-]?pass", "postgresql[_-]?db", + "postgres[_-]?env[_-]?postgres[_-]?password", + "postgres[_-]?env[_-]?postgres[_-]?db", "plugin[_-]?password", + "plotly[_-]?apikey", "places[_-]?apikey", "places[_-]?api[_-]?key", + "pg[_-]?host", "pg[_-]?database", "personal[_-]?secret", "personal[_-]?key", + "percy[_-]?token", "percy[_-]?project", + "paypal[_-]?client[_-]?secret", "passwordtravis", "parse[_-]?js[_-]?key", + "pagerduty[_-]?apikey", "packagecloud[_-]?token", + "ossrh[_-]?username", "ossrh[_-]?secret", "ossrh[_-]?password", + "ossrh[_-]?pass", "ossrh[_-]?jira[_-]?password", + "os[_-]?password", "os[_-]?auth[_-]?url", + "org[_-]?project[_-]?gradle[_-]?sonatype[_-]?nexus[_-]?password", + "org[_-]?gradle[_-]?project[_-]?sonatype[_-]?nexus[_-]?password", + "openwhisk[_-]?key", "open[_-]?whisk[_-]?key", + "onesignal[_-]?user[_-]?auth[_-]?key", "onesignal[_-]?api[_-]?key", + "omise[_-]?skey", "omise[_-]?pubkey", "omise[_-]?pkey", "omise[_-]?key", + "okta[_-]?oauth2[_-]?clientsecret", "okta[_-]?oauth2[_-]?client[_-]?secret", + "okta[_-]?client[_-]?token", + "ofta[_-]?secret", "ofta[_-]?region", "ofta[_-]?key", + "octest[_-]?password", "octest[_-]?app[_-]?username", + "octest[_-]?app[_-]?password", "oc[_-]?pass", + "object[_-]?store[_-]?creds", "object[_-]?store[_-]?bucket", + "object[_-]?storage[_-]?region[_-]?name", "object[_-]?storage[_-]?password", + "oauth[_-]?token", "numbers[_-]?service[_-]?pass", + "nuget[_-]?key", "nuget[_-]?apikey", "nuget[_-]?api[_-]?key", + "npm[_-]?token", "npm[_-]?secret[_-]?key", "npm[_-]?password", + "npm[_-]?email", "npm[_-]?auth[_-]?token", "npm[_-]?api[_-]?token", + "npm[_-]?api[_-]?key", "now[_-]?token", "non[_-]?token", + "node[_-]?pre[_-]?gyp[_-]?secretaccesskey", + "node[_-]?pre[_-]?gyp[_-]?github[_-]?token", + "node[_-]?pre[_-]?gyp[_-]?accesskeyid", "node[_-]?env", + "ngrok[_-]?token", "ngrok[_-]?auth[_-]?token", + "nexuspassword", "nexus[_-]?password", + "new[_-]?relic[_-]?beta[_-]?token", "netlify[_-]?api[_-]?key", + "nativeevents", "mysqlsecret", "mysqlmasteruser", + "mysql[_-]?username", "mysql[_-]?user", "mysql[_-]?root[_-]?password", + "mysql[_-]?password", "mysql[_-]?hostname", "mysql[_-]?database", + "my[_-]?secret[_-]?env", + "multi[_-]?workspace[_-]?sid", "multi[_-]?workflow[_-]?sid", + "multi[_-]?disconnect[_-]?sid", "multi[_-]?connect[_-]?sid", + "multi[_-]?bob[_-]?sid", + "minio[_-]?secret[_-]?key", "minio[_-]?access[_-]?key", + "mile[_-]?zero[_-]?key", "mh[_-]?password", "mh[_-]?apikey", + "mg[_-]?public[_-]?api[_-]?key", "mg[_-]?api[_-]?key", + "mapboxaccesstoken", + "mapbox[_-]?aws[_-]?secret[_-]?access[_-]?key", + "mapbox[_-]?aws[_-]?access[_-]?key[_-]?id", + "mapbox[_-]?api[_-]?token", "mapbox[_-]?access[_-]?token", + "manifest[_-]?app[_-]?url", "manifest[_-]?app[_-]?token", + "mandrill[_-]?api[_-]?key", "managementapiaccesstoken", + "management[_-]?token", "manage[_-]?secret", "manage[_-]?key", + "mailgun[_-]?secret[_-]?api[_-]?key", "mailgun[_-]?pub[_-]?key", + "mailgun[_-]?pub[_-]?apikey", "mailgun[_-]?priv[_-]?key", + "mailgun[_-]?password", "mailgun[_-]?apikey", "mailgun[_-]?api[_-]?key", + "mailer[_-]?password", "mailchimp[_-]?key", "mailchimp[_-]?api[_-]?key", + "mail[_-]?password", "magento[_-]?password", + "magento[_-]?auth[_-]?username", "magento[_-]?auth[_-]?password", + "lottie[_-]?upload[_-]?cert[_-]?key[_-]?store[_-]?password", + "lottie[_-]?upload[_-]?cert[_-]?key[_-]?password", + "lottie[_-]?s3[_-]?secret[_-]?key", + "lottie[_-]?happo[_-]?secret[_-]?key", "lottie[_-]?happo[_-]?api[_-]?key", + "looker[_-]?test[_-]?runner[_-]?client[_-]?secret", + "ll[_-]?shared[_-]?key", "ll[_-]?publish[_-]?url", + "linux[_-]?signing[_-]?key", + "lighthouse[_-]?api[_-]?key", + "lektor[_-]?deploy[_-]?username", "lektor[_-]?deploy[_-]?password", + "leanplum[_-]?key", "kxoltsn3vogdop92m", + "kubeconfig", "kubecfg[_-]?s3[_-]?path", "kovan[_-]?private[_-]?key", + "keystore[_-]?pass", + "kafka[_-]?rest[_-]?url", "kafka[_-]?instance[_-]?name", + "kafka[_-]?admin[_-]?url", "jwt[_-]?secret", + "jdbc[_-]?host", "jdbc[_-]?databaseurl", + "itest[_-]?gh[_-]?token", "ios[_-]?docs[_-]?deploy[_-]?token", + "internal[_-]?secrets", + "integration[_-]?test[_-]?appid", "integration[_-]?test[_-]?api[_-]?key", + "index[_-]?name", "ij[_-]?repo[_-]?username", "ij[_-]?repo[_-]?password", + "hub[_-]?dxia2[_-]?password", "homebrew[_-]?github[_-]?api[_-]?token", + "hockeyapp[_-]?token", + "heroku[_-]?token", "heroku[_-]?email", "heroku[_-]?api[_-]?key", + "hb[_-]?codesign[_-]?key[_-]?pass", "hb[_-]?codesign[_-]?gpg[_-]?pass", + "hab[_-]?key", "hab[_-]?auth[_-]?token", + "grgit[_-]?user", "gren[_-]?github[_-]?token", + "gradle[_-]?signing[_-]?password", "gradle[_-]?signing[_-]?key[_-]?id", + "gradle[_-]?publish[_-]?secret", "gradle[_-]?publish[_-]?key", + "gpg[_-]?secret[_-]?keys", "gpg[_-]?private[_-]?key", + "gpg[_-]?passphrase", "gpg[_-]?ownertrust", + "gpg[_-]?keyname", "gpg[_-]?key[_-]?name", + "google[_-]?private[_-]?key(?:[_-]?id)?", + "google[_-]?maps[_-]?api[_-]?key", "google[_-]?client[_-]?secret", + "google[_-]?client[_-]?id", "google[_-]?client[_-]?email", + "google[_-]?account[_-]?type", "gogs[_-]?password", + "gitlab[_-]?user[_-]?email", + "github[_-]?tokens", "github[_-]?token", "github[_-]?repo", + "github[_-]?release[_-]?token", "github[_-]?pwd", "github[_-]?password", + "github[_-]?oauth[_-]?token", "github[_-]?oauth", "github[_-]?key", + "github[_-]?hunter[_-]?username", "github[_-]?hunter[_-]?token", + "github[_-]?deployment[_-]?token", + "github[_-]?deploy[_-]?hb[_-]?doc[_-]?pass", + "github[_-]?client[_-]?secret", "github[_-]?auth[_-]?token", + "github[_-]?auth", "github[_-]?api[_-]?token", "github[_-]?api[_-]?key", + "github[_-]?access[_-]?token", + "git[_-]?token", "git[_-]?name", "git[_-]?email", + "git[_-]?committer[_-]?name", "git[_-]?committer[_-]?email", + "git[_-]?author[_-]?name", "git[_-]?author[_-]?email", + "ghost[_-]?api[_-]?key", "ghb[_-]?token", + "gh[_-]?unstable[_-]?oauth[_-]?client[_-]?secret", + "gh[_-]?token", "gh[_-]?repo[_-]?token", "gh[_-]?oauth[_-]?token", + "gh[_-]?oauth[_-]?client[_-]?secret", + "gh[_-]?next[_-]?unstable[_-]?oauth[_-]?client[_-]?secret", + "gh[_-]?next[_-]?unstable[_-]?oauth[_-]?client[_-]?id", + "gh[_-]?next[_-]?oauth[_-]?client[_-]?secret", + "gh[_-]?email", "gh[_-]?api[_-]?key", + "gcs[_-]?bucket", "gcr[_-]?password", + "gcloud[_-]?service[_-]?key", "gcloud[_-]?project", "gcloud[_-]?bucket", + "ftp[_-]?username", "ftp[_-]?user", "ftp[_-]?pw", "ftp[_-]?password", + "ftp[_-]?login", "ftp[_-]?host", "fossa[_-]?api[_-]?key", + "flickr[_-]?api[_-]?secret", "flickr[_-]?api[_-]?key", + "flask[_-]?secret[_-]?key", "firefox[_-]?secret", + "firebase[_-]?token", "firebase[_-]?project[_-]?develop", + "firebase[_-]?key", "firebase[_-]?api[_-]?token", "firebase[_-]?api[_-]?json", + "file[_-]?password", "exp[_-]?password", "eureka[_-]?awssecretkey", + "env[_-]?sonatype[_-]?password", "env[_-]?secret[_-]?access[_-]?key", + "env[_-]?secret", "env[_-]?key", + "env[_-]?heroku[_-]?api[_-]?key", "env[_-]?github[_-]?oauth[_-]?token", + "end[_-]?user[_-]?password", "encryption[_-]?password", + "elasticsearch[_-]?password", "elastic[_-]?cloud[_-]?auth", + "dsonar[_-]?projectkey", "dsonar[_-]?login", + "droplet[_-]?travis[_-]?password", "dropbox[_-]?oauth[_-]?bearer", + "doordash[_-]?auth[_-]?token", + "dockerhubpassword", "dockerhub[_-]?password", + "docker[_-]?token", "docker[_-]?postgres[_-]?url", + "docker[_-]?password", "docker[_-]?passwd", "docker[_-]?pass", + "docker[_-]?key", "docker[_-]?hub[_-]?password", + "digitalocean[_-]?ssh[_-]?key[_-]?ids", + "digitalocean[_-]?ssh[_-]?key[_-]?body", + "digitalocean[_-]?access[_-]?token", "dgpg[_-]?passphrase", + "deploy[_-]?user", "deploy[_-]?token", "deploy[_-]?secure", + "deploy[_-]?password", + "ddgc[_-]?github[_-]?token", "ddg[_-]?test[_-]?email[_-]?pw", + "ddg[_-]?test[_-]?email", + "db[_-]?username", "db[_-]?user", "db[_-]?pw", "db[_-]?password", + "db[_-]?host", "db[_-]?database", "db[_-]?connection", + "datadog[_-]?app[_-]?key", "datadog[_-]?api[_-]?key", + "database[_-]?username", "database[_-]?user", "database[_-]?port", + "database[_-]?password", "database[_-]?name", "database[_-]?host", + "danger[_-]?github[_-]?api[_-]?token", "cypress[_-]?record[_-]?key", + "coverity[_-]?scan[_-]?token", "coveralls[_-]?token", + "coveralls[_-]?repo[_-]?token", "coveralls[_-]?api[_-]?token", + "cos[_-]?secrets", + "conversation[_-]?username", "conversation[_-]?password", + "contentful[_-]?v2[_-]?access[_-]?token", + "contentful[_-]?test[_-]?org[_-]?cma[_-]?token", + "contentful[_-]?php[_-]?management[_-]?test[_-]?token", + "contentful[_-]?management[_-]?api[_-]?access[_-]?token[_-]?new", + "contentful[_-]?management[_-]?api[_-]?access[_-]?token", + "contentful[_-]?integration[_-]?management[_-]?token", + "contentful[_-]?cma[_-]?test[_-]?token", + "contentful[_-]?access[_-]?token", + "consumerkey", "consumer[_-]?key", "conekta[_-]?apikey", + "coding[_-]?token", "codecov[_-]?token", "codeclimate[_-]?repo[_-]?token", + "codacy[_-]?project[_-]?token", + "cocoapods[_-]?trunk[_-]?token", "cocoapods[_-]?trunk[_-]?email", + "cn[_-]?secret[_-]?access[_-]?key", "cn[_-]?access[_-]?key[_-]?id", + "clu[_-]?ssh[_-]?private[_-]?key[_-]?base64", "clu[_-]?repo[_-]?url", + "cloudinary[_-]?url[_-]?staging", "cloudinary[_-]?url", + "cloudflare[_-]?email", "cloudflare[_-]?auth[_-]?key", + "cloudflare[_-]?auth[_-]?email", "cloudflare[_-]?api[_-]?key", + "cloudant[_-]?service[_-]?database", "cloudant[_-]?processed[_-]?database", + "cloudant[_-]?password", "cloudant[_-]?parsed[_-]?database", + "cloudant[_-]?order[_-]?database", "cloudant[_-]?instance", + "cloudant[_-]?database", "cloudant[_-]?audited[_-]?database", + "cloudant[_-]?archived[_-]?database", + "cloud[_-]?api[_-]?key", "clojars[_-]?password", "client[_-]?secret", + "cli[_-]?e2e[_-]?cma[_-]?token", + "claimr[_-]?token", "claimr[_-]?superuser", "claimr[_-]?db", + "claimr[_-]?database", + "ci[_-]?user[_-]?token", "ci[_-]?server[_-]?name", + "ci[_-]?registry[_-]?user", "ci[_-]?project[_-]?url", + "ci[_-]?deploy[_-]?password", + "chrome[_-]?refresh[_-]?token", "chrome[_-]?client[_-]?secret", + "cheverny[_-]?token", "cf[_-]?password", "certificate[_-]?password", + "censys[_-]?secret", + "cattle[_-]?secret[_-]?key", "cattle[_-]?agent[_-]?instance[_-]?auth", + "cattle[_-]?access[_-]?key", "cargo[_-]?token", + "cache[_-]?s3[_-]?secret[_-]?key", + "bx[_-]?username", "bx[_-]?password", + "bundlesize[_-]?github[_-]?token", "built[_-]?branch[_-]?deploy[_-]?key", + "bucketeer[_-]?aws[_-]?secret[_-]?access[_-]?key", + "bucketeer[_-]?aws[_-]?access[_-]?key[_-]?id", + "browserstack[_-]?access[_-]?key", "browser[_-]?stack[_-]?access[_-]?key", + "brackets[_-]?repo[_-]?oauth[_-]?token", + "bluemix[_-]?username", "bluemix[_-]?pwd", "bluemix[_-]?password", + "bluemix[_-]?pass[_-]?prod", "bluemix[_-]?pass", + "bluemix[_-]?auth", "bluemix[_-]?api[_-]?key", + "bintraykey", "bintray[_-]?token", "bintray[_-]?key", + "bintray[_-]?gpg[_-]?password", "bintray[_-]?apikey", "bintray[_-]?api[_-]?key", + "b2[_-]?bucket", "b2[_-]?app[_-]?key", + "awssecretkey", "awscn[_-]?secret[_-]?access[_-]?key", + "awscn[_-]?access[_-]?key[_-]?id", "awsaccesskeyid", + "aws[_-]?ses[_-]?secret[_-]?access[_-]?key", + "aws[_-]?ses[_-]?access[_-]?key[_-]?id", "aws[_-]?secrets", + "aws[_-]?secret[_-]?key", "aws[_-]?secret[_-]?access[_-]?key", + "aws[_-]?secret", "aws[_-]?key", + "aws[_-]?config[_-]?secretaccesskey", "aws[_-]?config[_-]?accesskeyid", + "aws[_-]?access[_-]?key[_-]?id", "aws[_-]?access[_-]?key", "aws[_-]?access", + "author[_-]?npm[_-]?api[_-]?key", "author[_-]?email[_-]?addr", + "auth0[_-]?client[_-]?secret", "auth0[_-]?api[_-]?clientsecret", + "auth[_-]?token", + "assistant[_-]?iam[_-]?apikey", + "artifacts[_-]?secret", "artifacts[_-]?key", "artifacts[_-]?bucket", + "artifacts[_-]?aws[_-]?secret[_-]?access[_-]?key", + "artifacts[_-]?aws[_-]?access[_-]?key[_-]?id", "artifactory[_-]?key", + "argos[_-]?token", "apple[_-]?id[_-]?password", "appclientsecret", + "app[_-]?token", "app[_-]?secrete", "app[_-]?report[_-]?token[_-]?key", + "app[_-]?bucket[_-]?perm", + "apigw[_-]?access[_-]?token", "apiary[_-]?api[_-]?key", + "api[_-]?secret", "api[_-]?key[_-]?sid", "api[_-]?key[_-]?secret", + "api[_-]?key", "aos[_-]?sec", "aos[_-]?key", + "ansible[_-]?vault[_-]?password", "android[_-]?docs[_-]?deploy[_-]?token", + "anaconda[_-]?token", + "amazon[_-]?secret[_-]?access[_-]?key", "amazon[_-]?bucket[_-]?name", + "alicloud[_-]?secret[_-]?key", "alicloud[_-]?access[_-]?key", + "alias[_-]?pass", + "algolia[_-]?search[_-]?key[_-]?1", "algolia[_-]?search[_-]?key", + "algolia[_-]?search[_-]?api[_-]?key", + "algolia[_-]?api[_-]?key[_-]?search", "algolia[_-]?api[_-]?key[_-]?mcm", + "algolia[_-]?api[_-]?key", "algolia[_-]?admin[_-]?key[_-]?mcm", + "algolia[_-]?admin[_-]?key[_-]?2", "algolia[_-]?admin[_-]?key[_-]?1", + "adzerk[_-]?api[_-]?key", "admin[_-]?email", + "account[_-]?sid", "access[_-]?token", "access[_-]?secret", + "access[_-]?key[_-]?secret", + "account", "password", "username", +] + +# 通配符型规则 +_NUCLEI_WILDCARD_SOURCES = [ + r"""[\w_-]*?password[\w_-]*?""", + r"""[\w_-]*?username[\w_-]*?""", + r"""[\w_-]*?accesskey[\w_-]*?""", + r"""[\w_-]*?secret[\w_-]*?""", + r"""[\w_-]*?bucket[\w_-]*?""", + r"""[\w_-]*?token[\w_-]*?""", +] + +# 特殊格式的规则 (非 key=value 型) +_NUCLEI_SPECIAL_PATTERNS = [ + r"""[\"']?[-]+BEGIN \w+ PRIVATE KEY[-]+""", + r"""[\"']?huawei\.oss\.(ak|sk|bucket\.name|endpoint|local\.path)[\"']?\s*[=:]\s*[\"']?[\w-]+[\"']?""", + r"""[\"']?private[_-]?key(?:[_-]?id)?[\"']?\s*[=:]\s*[\"']?[\w-]+[\"']?""", + r"""[\"']?account[_-]?(?:name|key)?[\"']?\s*[=:]\s*[\"']?[\w-]+[\"']?""", + r"""LTAI[A-Za-z\d]{12,30}""", + r"""AKID[A-Za-z\d]{13,40}""", + r"""JDC_[0-9A-Z]{25,40}""", + r"""[\"']?(?:A3T[A-Z0-9]|AKIA|AGPA|AIDA|AROA|AIPA|ANPA|ANVA|ASIA)[A-Z0-9]{16}[\"']?""", + r"""(?:AKLT|AKTP)[a-zA-Z0-9]{35,50}""", + r"""AKLT[a-zA-Z0-9\-_]{16,28}""", + r"""AIza[0-9A-Za-z_\-]{35}""", + r"""[Bb]earer\s+[a-zA-Z0-9\-=._+/\\]{20,500}""", + r"""[Bb]asic\s+[A-Za-z0-9+/]{18,}={0,2}""", + r"""[\"'\[]*[Aa]uthorization[\"'\]]*\s*[:=]\s*['"]?\b(?:[Tt]oken\s+)?[a-zA-Z0-9\-_+/]{20,500}['"]?""", + r"""(glpat-[a-zA-Z0-9\-=_]{20,22})""", + r"""((?:ghp|gho|ghu|ghs|ghr|github_pat)_[a-zA-Z0-9_]{36,255})""", + r"""APID[a-zA-Z0-9]{32,42}""", + r"""[\"'](wx[a-z0-9]{15,18})[\"']""", + r"""[\"'](ww[a-z0-9]{15,18})[\"']""", + r"""[\"'](gh_[a-z0-9]{11,13})[\"']""", + r"""(?:admin_?pass|password|[a-z]{3,15}_?password|user_?pass|user_?pwd|admin_?pwd)\\?['\"]*\s*[:=]\s*\\?['"][a-z0-9!@#$%&*]{5,20}\\?['"]""", + r"""https:\/\/qyapi\.weixin\.qq\.com\/cgi\-bin\/webhook\/send\?key=[a-zA-Z0-9\-]{25,50}""", + r"""https:\/\/oapi\.dingtalk\.com\/robot\/send\?access_token=[a-z0-9]{50,80}""", + r"""https:\/\/open\.feishu\.cn\/open\-apis\/bot\/v2\/hook\/[a-z0-9\-]{25,50}""", + r"""https:\/\/hooks\.slack\.com\/services\/[a-zA-Z0-9\-_]{6,12}\/[a-zA-Z0-9\-_]{6,12}\/[a-zA-Z0-9\-_]{15,24}""", + r"""eyJrIjoi[a-zA-Z0-9\-_+/]{50,100}={0,2}""", + r"""glc_[A-Za-z0-9\-_+/]{32,200}={0,2}""", + r"""glsa_[A-Za-z0-9]{32}_[A-Fa-f0-9]{8}""", + r"""[\"']?air[-_]?table[-_]?api[-_]?key[\"']?[=:][\"']?.+[\"']""", + r"""[\"']?jdbc:mysql[\"']?\s*[=:]\s*[\"']?[\w-]+[\"']?""", +] + + +def _build_nuclei_patterns() -> List[re.Pattern]: + """编译所有nuclei正则为Python Pattern对象 (与heartk_fast.py完全一致)""" + compiled = [] + all_kv_keys = list(_NUCLEI_REGEX_SOURCES) + list(_NUCLEI_WILDCARD_SOURCES) + batch_size = 120 + for i in range(0, len(all_kv_keys), batch_size): + batch = all_kv_keys[i:i + batch_size] + alternation = '|'.join(batch) + pat = r"""[\"']?(?:""" + alternation + r""")[\"']?[^\S\r\n]*[=:][^\S\r\n]*[\"']?[\w-]+[\"']?""" + compiled.append(re.compile(pat, re.IGNORECASE)) + + for pat_str in _NUCLEI_SPECIAL_PATTERNS: + try: + compiled.append(re.compile(pat_str, re.IGNORECASE)) + except re.error: + compiled.append(re.compile(pat_str)) + + return compiled + + +NUCLEI_PATTERNS = _build_nuclei_patterns() + + +# ============================================================ +# 2. 核心分析函数 (与 heartk_fast.py 完全一致) +# ============================================================ + +def strip_quotes(items: List[str]) -> List[str]: + """移除字符串两端的引号""" + result = [] + for item in items: + s = 0 + e = len(item) + if item and item[0] in ("'", '"'): + s = 1 + if item and item[-1] in ("'", '"'): + e -= 1 + result.append(item[s:e]) + return result + + +def extract_info(data: str) -> Dict[str, Optional[List[str]]]: + """从文本内容中提取各类敏感信息 (对应JS中的 extract_info + get_secret)""" + result: Dict[str, Optional[List[str]]] = {} + + # 运行各分类正则 + for key, pattern in EXTRACT_PATTERNS.items(): + matches = pattern.findall(data) + if matches: + # findall 对于有组的正则返回元组,取第一个完整匹配或全匹配 + if isinstance(matches[0], tuple): + # 需要完整匹配,用 finditer + matches = [m.group(0) for m in pattern.finditer(data)] + result[key] = matches if matches else None + else: + result[key] = None + + # 从URL中额外提取IP/域名 (对应JS中 extract_info 末尾逻辑) + if result.get('url'): + ip_pat = re.compile(r"""['\"](([a-zA-Z0-9]+:)?\/\/)?\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}""") + for url in result['url']: + ip_matches = ip_pat.findall(url) + if ip_matches: + full_matches = [m.group(0) for m in ip_pat.finditer(url)] + if result['ip'] is None: + result['ip'] = full_matches + else: + result['ip'].extend(full_matches) + + # 运行nuclei秘密检测 + secrets = [] + for pat in NUCLEI_PATTERNS: + try: + matches = pat.findall(data) + if matches: + if isinstance(matches[0], tuple): + for m in pat.finditer(data): + secrets.append(m.group(0)) + else: + secrets.extend(matches) + except Exception: + pass + result['secret'] = secrets if secrets else None + + return result + + +def init_source(source: str) -> List[str]: + """从HTML中提取href和src属性值 (对应JS中的 init_source)""" + target_list = [] + + # href="..." + for m in re.finditer(r"""href=['\"](.+?)['\"]""", source): + target_list.append(m.group(1)) + + # src="..." + for m in re.finditer(r"""src=['\"](.+?)['\"]""", source): + target_list.append(m.group(1)) + + return list(set(target_list)) + + +def collect_static(items: List[str], static_list: List[str]) -> tuple: + """分离静态资源""" + remaining = [] + for item in items: + is_static = False + for ext in STATIC_EXTENSIONS: + if ext in item: + if ext == '.js' and '.jsp' in item: + continue + is_static = True + if item not in static_list: + static_list.append(item) + break + if not is_static: + remaining.append(item) + return remaining, static_list + + +def get_info(content: str) -> Dict[str, Optional[List[str]]]: + """完整分析一个文件内容""" + tmp_data = extract_info(content) + + source_links = init_source(content) + static_data: List[str] = [] + + if source_links: + _, static_data = collect_static(source_links, static_data) + + for key in ['domain', 'path', 'url']: + if tmp_data.get(key): + _, static_data = collect_static(tmp_data[key], static_data) + + for key in ['domain', 'path', 'url']: + if tmp_data.get(key): + tmp_data[key], _ = collect_static(tmp_data[key], list(static_data)) + + tmp_data['static'] = static_data if static_data else None + + # 去引号 (secret 保留原始) + not_sub_keys = {'secret'} + for key in CATEGORIES: + if key in tmp_data and tmp_data[key] is not None and key not in not_sub_keys: + tmp_data[key] = strip_quotes(tmp_data[key]) + + return tmp_data + + +# ============================================================ +# 3. 文件扫描 +# ============================================================ + +def detect_encoding(filepath: str) -> Optional[str]: + """检测文件编码""" + if chardet is None: + return "utf-8" + try: + with open(filepath, 'rb') as f: + raw = f.read(65536) + if not raw: + return None + result = chardet.detect(raw) + return result.get('encoding') + except Exception: + return None + + +def scan_single_file(filepath: str) -> Optional[Dict[str, Optional[List[str]]]]: + """扫描单个文件,返回分析结果""" + encoding = detect_encoding(filepath) + if not encoding: + return None + + try: + with open(filepath, 'r', encoding=encoding, errors='ignore') as f: + content = f.read() + if not content.strip(): + return None + return get_info(content) + except Exception: + return None + + +def scan_single_file_with_custom(args): + """带自定义正则的扫描 (用于进程池)""" + filepath, custom_patterns_raw = args + + # 基础扫描 + result = scan_single_file(filepath) + if result is None: + result = {} + + # 自定义正则扫描 + if custom_patterns_raw: + encoding = detect_encoding(filepath) + if not encoding: + return filepath, result if result else None + try: + with open(filepath, 'r', encoding=encoding, errors='ignore') as f: + content = f.read() + except Exception: + return filepath, result if result else None + + if content.strip(): + for name, pat_str in custom_patterns_raw.items(): + try: + pat = re.compile(pat_str, re.IGNORECASE) + matches = [m.group(0) for m in pat.finditer(content)] + if matches: + result[f"custom_{name}"] = matches + except Exception: + pass + + return filepath, result if result else None + + +_SCAN_EXTS = {'.js', '.html', '.htm', '.json', '.css', '.wxml', '.wxss', '.txt', '.xml', '.svg', '.ts', '.jsx', '.tsx', '.md', '.yaml', '.yml', '.cfg', '.ini', '.conf', '.env', '.properties'} + + +def collect_files(path: str) -> List[str]: + """递归收集所有可扫描的文本文件""" + files = [] + for root, dirs, filenames in os.walk(path): + for fn in filenames: + ext = os.path.splitext(fn)[1].lower() + if ext in _SCAN_EXTS: + files.append(os.path.join(root, fn)) + return files + + +def merge_results(all_info: Dict, file_result: Dict): + """合并单文件结果到总结果""" + for key, value in file_result.items(): + if value is None: + continue + if all_info.get(key) is None: + all_info[key] = list(value) + else: + all_info[key].extend(value) + + +def dedup_and_sort(all_info: Dict): + """去重并排序""" + for key in list(all_info.keys()): + if all_info.get(key) and isinstance(all_info[key], list): + all_info[key] = sorted(set(all_info[key])) + + +# ============================================================ +# 4. Extractor 类 (GUI 接口) +# ============================================================ + +class Extractor: + """敏感信息提取引擎 — GUI 接口""" + + def __init__(self): + self.custom_patterns: Dict[str, str] = {} # {名称: 正则} + + def add_custom_pattern(self, name: str, regex_str: str): + self.custom_patterns[name] = regex_str + + def remove_custom_pattern(self, name: str): + self.custom_patterns.pop(name, None) + + def scan_directory(self, path: str, num_workers: int = 0, + on_progress: Optional[Callable] = None) -> Dict[str, Any]: + """扫描目录,返回合并后的结果""" + files = collect_files(path) + if not files: + return {"files_scanned": 0, "elapsed": 0, "results": {k: [] for k in CATEGORIES}, + "custom_results": {}} + + if num_workers <= 0: + num_workers = min(os.cpu_count() or 4, len(files)) + num_workers = max(1, min(num_workers, 16)) + + all_info: Dict[str, Optional[List[str]]] = {k: None for k in CATEGORIES} + custom_results: Dict[str, List[str]] = {} # 自定义正则结果 + done_count = 0 + total = len(files) + + start_time = time.time() + + custom_raw = dict(self.custom_patterns) if self.custom_patterns else {} + + if num_workers == 1 or total <= 3: + for fpath in files: + _, result = scan_single_file_with_custom((fpath, custom_raw)) + done_count += 1 + if result: + # 分离自定义结果 + for key in list(result.keys()): + if key.startswith("custom_"): + cat_name = key[7:] # 去掉 "custom_" 前缀 + if cat_name not in custom_results: + custom_results[cat_name] = [] + custom_results[cat_name].extend(result.pop(key)) + merge_results(all_info, result) + if on_progress: + on_progress(done_count, total) + else: + args_list = [(f, custom_raw) for f in files] + with ProcessPoolExecutor(max_workers=num_workers) as executor: + future_to_file = {executor.submit(scan_single_file_with_custom, a): a[0] for a in args_list} + for future in as_completed(future_to_file): + done_count += 1 + try: + fpath, result = future.result() + if result: + for key in list(result.keys()): + if key.startswith("custom_"): + cat_name = key[7:] + if cat_name not in custom_results: + custom_results[cat_name] = [] + custom_results[cat_name].extend(result.pop(key)) + merge_results(all_info, result) + except Exception: + pass + if on_progress: + on_progress(done_count, total) + + # domain 合并到 url + if all_info.get('domain'): + if all_info.get('url'): + all_info['url'].extend(all_info['domain']) + else: + all_info['url'] = list(all_info['domain']) + + # 去重排序 + dedup_and_sort(all_info) + for k in custom_results: + custom_results[k] = sorted(set(custom_results[k])) + + elapsed = time.time() - start_time + + return { + "files_scanned": total, + "elapsed": round(elapsed, 2), + "results": {k: all_info.get(k) or [] for k in CATEGORIES}, + "custom_results": custom_results, + } + + @staticmethod + def export_json(scan_result: Dict, output_path: str): + """导出 JSON (包含自定义正则结果)""" + clean = {} + for key in CATEGORIES: + clean[key] = scan_result.get("results", {}).get(key, []) + # 自定义正则结果 + for name, items in scan_result.get("custom_results", {}).items(): + clean[name] = items + clean["_meta"] = { + "files_scanned": scan_result.get("files_scanned", 0), + "elapsed": scan_result.get("elapsed", 0), + } + with open(output_path, 'w', encoding='utf-8') as f: + json.dump(clean, f, ensure_ascii=False, indent=2) + + @staticmethod + def export_html(scan_result: Dict, output_path: str): + """生成 HTML 报告 (包含自定义正则结果)""" + data = dict(scan_result.get("results", {})) + custom = scan_result.get("custom_results", {}) + # 合并自定义到 data 以供 JS 使用 + for name, items in custom.items(): + data[name] = items + + data_json = json.dumps(data, ensure_ascii=False) + + # 所有类别 key 列表 (用于 JS) + all_keys = list(CATEGORIES) + list(custom.keys()) + keys_js = json.dumps(all_keys, ensure_ascii=False) + + popup_js = r""" + var key = """ + keys_js + r"""; + let messages = {"popupCopy": "复制", "popupCopyurl": "复制URL"}; + function getMessage(key) { return messages[key] || ""; } + function init_copy() { + var elements = document.getElementsByClassName("copy"); + if(elements){ + for (var i=0, len=elements.length|0; i {label} ({count})\n 🈚️
\n' + + right_html = "" + for cat in right_cats: + label = cat_labels.get(cat, cat) + count = len(data.get(cat, [])) + right_html += f'{label} ({count})\n🈚️
\n' + + # 自定义正则类别 + custom_html = "" + for name, items in custom.items(): + count = len(items) + custom_html += f'{name} ({count})\n🈚️
\n' + + meta = scan_result.get("files_scanned", 0) + elapsed = scan_result.get("elapsed", 0) + total_findings = sum(len(data.get(k, [])) for k in all_keys) + + report_html = f""" + +敏感信息提取报告 + ++++ First 敏感信息提取报告 ++++ {"+ {left_html} +++ {right_html} ++" if custom_html else ""} + + +""" + with open(output_path, 'w', encoding='utf-8') as f: + f.write(report_html) + + @staticmethod + def get_summary(scan_result: Dict) -> Dict[str, int]: + """返回各分类的计数 (含自定义)""" + summary = {} + results = scan_result.get("results", {}) + for key in CATEGORIES: + summary[key] = len(results.get(key, [])) + for name, items in scan_result.get("custom_results", {}).items(): + summary[name] = len(items) + return summary + + @staticmethod + def get_category_label(key: str) -> str: + labels = { + 'sfz': '身份证', 'mobile': '手机号', 'mail': '邮箱', + 'ip': 'IP地址', 'ip_port': 'IP端口', 'domain': '域名', + 'path': '路径', 'incomplete_path': '不完整路径', + 'url': 'URL', 'jwt': 'JWT', 'algorithm': '加密算法', + 'secret': '密钥/凭证', 'static': '静态资源' + } + return labels.get(key, key) + + @staticmethod + def get_all_builtin_patterns() -> Dict[str, str]: + """获取所有内置正则的字符串表示 (供 UI 显示)""" + result = {} + # extract patterns + for key, pat in EXTRACT_PATTERNS.items(): + result[f"[提取] {key}"] = pat.pattern + # nuclei kv keys count + result["[密钥] key=value 关键字"] = f"({len(_NUCLEI_REGEX_SOURCES)} 个关键字匹配)" + # nuclei wildcards + for i, pat in enumerate(_NUCLEI_WILDCARD_SOURCES): + result[f"[密钥] 通配符 #{i+1}"] = pat + # nuclei specials + for i, pat in enumerate(_NUCLEI_SPECIAL_PATTERNS): + result[f"[密钥] 特殊规则 #{i+1}"] = pat + return result diff --git a/src/navigator.py b/src/navigator.py index 9d8b401..306da8d 100644 --- a/src/navigator.py +++ b/src/navigator.py @@ -83,6 +83,34 @@ class MiniProgramNavigator: await self._ensure() await self.engine.evaluate_js(f"window.nav.back({delta})", timeout=5.0) + async def refresh_page(self): + """刷新当前小程序页面:获取当前路由后 reLaunch 到同一页面。""" + await self._ensure() + result = await self.engine.evaluate_js( + "(function(){" + "try{" + "var nav=window.nav;" + "if(!nav||!nav.wxFrame)return JSON.stringify({err:'no nav'});" + "var p=nav.wxFrame.getCurrentPages();" + "if(!p||!p.length)return JSON.stringify({err:'no page'});" + "var cur=p[p.length-1];" + "var route=cur.route||cur.__route__||'';" + "if(!route)return JSON.stringify({err:'no route'});" + "var url='/'+route;" + "var opts=cur.options||{};" + "var qs=Object.keys(opts).map(function(k){return k+'='+opts[k]}).join('&');" + "if(qs)url+='?'+qs;" + "nav.wxFrame.wx.reLaunch({url:url," + "success:function(){},fail:function(e){" + "nav.wxFrame.wx.redirectTo({url:url,success:function(){},fail:function(){}})" + "}});" + "return JSON.stringify({ok:true,route:route})" + "}catch(e){return JSON.stringify({err:e.message})}" + "})()", + timeout=5.0, + ) + return self._extract_value(result) or "" + async def get_current_route(self): """Get current page route via the detected wxFrame.""" await self._ensure() diff --git a/src/wxapkg.py b/src/wxapkg.py new file mode 100644 index 0000000..2fd936c --- /dev/null +++ b/src/wxapkg.py @@ -0,0 +1,223 @@ +""" +wxapkg.py — 微信小程序 wxapkg 解密 + 解包 (WeChat 4.0+) +纯 Python 实现,基于 KillWxapkg 算法 + +加密格式: + - Magic: "V1MMWX" (6 bytes) + - AES-256-CBC 加密区: file[6:1030] → 解密取前 1023 bytes + - XOR 加密区: file[1030:] → 逐字节 XOR appID[-2] + - Key = PBKDF2(appID, "saltiest", 1000, 32, SHA1) + - IV = "the iv: 16 bytes" + +解包格式: + - 0xBE marker (1 byte) + - info1 (4 bytes, big-endian uint32) + - indexInfoLength (4 bytes) + - bodyInfoLength (4 bytes) + - 0xED marker (1 byte) + - fileCount (4 bytes) + - index entries: nameLen(4) + name(nameLen) + offset(4) + size(4) + - body data +""" +import hashlib +import os +import struct +from pathlib import Path +from typing import List, Optional, Tuple + +try: + from Crypto.Cipher import AES + from Crypto.Util.Padding import unpad +except ImportError: + try: + from Cryptodome.Cipher import AES + from Cryptodome.Util.Padding import unpad + except ImportError: + AES = None + unpad = None + +MAGIC = b"V1MMWX" +SALT = b"saltiest" +IV = b"the iv: 16 bytes" +PBKDF2_ITER = 1000 +KEY_LEN = 32 + + +def _derive_key(app_id: str) -> bytes: + """PBKDF2-SHA1 派生 AES-256 密钥""" + return hashlib.pbkdf2_hmac("sha1", app_id.encode("utf-8"), SALT, PBKDF2_ITER, KEY_LEN) + + +def decrypt_wxapkg(data: bytes, app_id: str) -> bytes: + """解密单个 wxapkg 文件内容,返回解密后的 bytes""" + if AES is None: + raise ImportError("需要安装 pycryptodome: pip install pycryptodome") + + if len(data) < 1030: + raise ValueError("文件太小,不是有效的加密 wxapkg") + + # 检查 magic header + if data[:6] != MAGIC: + # 可能是未加密的 wxapkg,直接返回 + if data[0:1] == b'\xbe': + return data + raise ValueError(f"未知文件格式 (magic: {data[:6].hex()})") + + key = _derive_key(app_id) + + # AES-256-CBC 解密前 1024 bytes (file[6:1030]) + cipher = AES.new(key, AES.MODE_CBC, IV) + decrypted_header = cipher.decrypt(data[6:1030]) + # 取前 1023 bytes(去掉 PKCS7 padding 的最后一个字节) + decrypted_header = decrypted_header[:1023] + + # XOR 解密剩余部分 + xor_key = ord(app_id[-2]) if len(app_id) >= 2 else 0 + tail = data[1030:] + xor_decrypted = bytes(b ^ xor_key for b in tail) + + return decrypted_header + xor_decrypted + + +def unpack_wxapkg(data: bytes) -> List[Tuple[str, bytes]]: + """解包 wxapkg,返回 [(文件路径, 文件内容), ...]""" + if len(data) < 14: + raise ValueError("数据太短,不是有效的 wxapkg") + + pos = 0 + + # 0xBE marker + marker1 = data[pos] + if marker1 != 0xBE: + raise ValueError(f"无效的 wxapkg 格式 (marker1: 0x{marker1:02x}, 期望 0xBE)") + pos += 1 + + # info1 (4 bytes) + _info1 = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + # indexInfoLength (4 bytes) + index_info_length = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + # bodyInfoLength (4 bytes) + _body_info_length = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + # 0xED marker + marker2 = data[pos] + if marker2 != 0xED: + raise ValueError(f"无效的 wxapkg 格式 (marker2: 0x{marker2:02x}, 期望 0xED)") + pos += 1 + + # fileCount (4 bytes) + file_count = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + files = [] + for _ in range(file_count): + if pos + 4 > len(data): + break + + # nameLen (4 bytes) + name_len = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + if pos + name_len > len(data): + break + + # name (nameLen bytes) + name = data[pos:pos + name_len].decode("utf-8", errors="replace") + pos += name_len + + if pos + 8 > len(data): + break + + # offset (4 bytes) + offset = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + # size (4 bytes) + size = struct.unpack(">I", data[pos:pos + 4])[0] + pos += 4 + + files.append((name, offset, size)) + + # 提取文件内容 + result = [] + for name, offset, size in files: + if offset + size <= len(data): + content = data[offset:offset + size] + result.append((name, content)) + + return result + + +def extract_wxapkg(wxapkg_path: str, output_dir: str, app_id: str) -> List[str]: + """解密并解包 wxapkg 到指定目录,返回提取的文件路径列表""" + with open(wxapkg_path, "rb") as f: + raw = f.read() + + decrypted = decrypt_wxapkg(raw, app_id) + files = unpack_wxapkg(decrypted) + + extracted = [] + for name, content in files: + # 规范化路径,防止路径穿越 + name = name.lstrip("/").replace("\\", "/") + if ".." in name: + continue + + out_path = os.path.join(output_dir, name) + os.makedirs(os.path.dirname(out_path), exist_ok=True) + + with open(out_path, "wb") as f: + f.write(content) + extracted.append(out_path) + + return extracted + + +def find_wxapkg_files(packages_dir: str) -> List[dict]: + """扫描 packages 目录,返回所有 wxapkg 文件信息 + + 目录结构: packages/{appid}/ 下有 .wxapkg 文件 + 返回: [{"appid": str, "path": str, "name": str, "size": int}, ...] + """ + results = [] + if not os.path.isdir(packages_dir): + return results + + for entry in os.listdir(packages_dir): + app_dir = os.path.join(packages_dir, entry) + if not os.path.isdir(app_dir): + continue + # 检查是否像 appid (wx开头 或 纯字母数字) + appid = entry + + for root, dirs, filenames in os.walk(app_dir): + for fn in filenames: + if fn.endswith(".wxapkg"): + fpath = os.path.join(root, fn) + results.append({ + "appid": appid, + "path": fpath, + "name": fn, + "size": os.path.getsize(fpath), + }) + + return results + + +def get_default_packages_dir() -> Optional[str]: + """获取默认的微信小程序包路径(Windows)""" + if os.name != "nt": + return None + user_home = os.path.expanduser("~") + pkg_dir = os.path.join( + user_home, "AppData", "Roaming", "Tencent", + "xwechat", "radium", "Applet", "packages" + ) + if os.path.isdir(pkg_dir): + return pkg_dir + return None自定义规则结果" + custom_html + "