!/usr/bin/env python3
"""
蓝队白加黑检测辅助 - 签名扫描器 (GUI版)
功能：
- 递归扫描用户选定目录中的指定后缀文件（默认 .exe .dll .sys）
- 用户可自定义要扫描的文件后缀名（增/删/改），配置自动保存到 config.json
- 操作日志自动追加到 log.txt（含时间戳、启动/异常等信息）
- 判断文件是否有数字签名（提取“颁发给”的公司名）
- 调用 Windows 原生 WinVerifyTrust API 验证签名有效性
- 提供图形界面，实时展示扫描进度和结果
- 支持过滤视图：仅显示无签名、签名无效或指定公司的文件
- 支持中途停止扫描，并更新已收集的公司列表
依赖：pip install pefile cryptography
运行环境：仅 Windows（需要 WinVerifyTrust API）
"""
import os
import sys
import json
import ctypes
import warnings
import threading
import queue
import traceback
from datetime import datetime
from pathlib import Path
from ctypes import wintypes, POINTER, Structure, byref
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, simpledialog
import pefile
from cryptography.hazmat.primitives.serialization import pkcs7
from cryptography import x509
from cryptography.x509.oid import NameOID
---- 忽略 PKCS#7 解析时 BER 回退的警告（不影响功能） ----
warnings.filterwarnings("ignore", message=".PKCS#7 certificates could not be parsed as DER.")
===========================================================================
全局路径常量（脚本所在目录）
===========================================================================
SCRIPT_DIR = os.path.dirname(os.path.abspath(file))
CONFIG_FILE = os.path.join(SCRIPT_DIR, "config.json")
LOG_FILE = os.path.join(SCRIPT_DIR, "log.txt")
===========================================================================
日志工具函数
===========================================================================
def log_message(level, message):
"""
向 log.txt 追加一条带时间戳的日志记录。
格式：[YYYY-MM-DD HH:MM:SS] [LEVEL] message
"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_line = f"[{timestamp}] [{level}] {message}\n"
try:
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(log_line)
except Exception:
pass # 写入日志本身出错时静默忽略，避免干扰主程序
===========================================================================
Windows 信任验证 API (WinVerifyTrust) 的类型与常量定义
===========================================================================
WTD_UI_NONE = 2 # 无用户界面
WTD_REVOKE_NONE = 0x00000000 # 不执行吊销检查（加速扫描）
WTD_CHOICE_FILE = 1 # 验证对象为文件
WTD_STATEACTION_VERIFY = 1 # 开始验证
WTD_STATEACTION_CLOSE = 2 # 关闭验证句柄
class GUID(Structure):
"""Windows GUID 结构体"""
fields = [
("Data1", wintypes.DWORD),
("Data2", wintypes.WORD),
("Data3", wintypes.WORD),
("Data4", wintypes.BYTE * 8),
]
class WINTRUST_FILE_INFO(Structure):
"""WinVerifyTrust 使用的文件信息结构体"""
fields = [
("cbStruct", wintypes.DWORD), # 结构体大小
("pcwszFilePath", wintypes.LPCWSTR), # 文件路径（Unicode）
("hFile", wintypes.HANDLE), # 文件句柄（可为 NULL）
("pgKnownSubject", ctypes.c_void_p), # 已知主题（保留）
]
class WINTRUST_DATA(Structure):
"""WinVerifyTrust 的验证数据结构体"""
fields = [
("cbStruct", wintypes.DWORD), # 结构体大小
("pPolicyCallbackData", ctypes.c_void_p), # 策略回调数据
("pSIPClientData", ctypes.c_void_p), # SIP 客户端数据
("dwUIChoice", wintypes.DWORD), # UI 选项
("fdwRevocationChecks", wintypes.DWORD), # 吊销检查标志
("dwUnionChoice", wintypes.DWORD), # 联合选择 (文件/Blob 等)
("pFile", POINTER(WINTRUST_FILE_INFO)), # 文件信息指针
("dwStateAction", wintypes.DWORD), # 状态操作（验证/关闭）
("hWVTStateData", wintypes.HANDLE), # 状态数据句柄
("pwszURLReference", wintypes.LPCWSTR), # URL 参考
("dwProvFlags", wintypes.DWORD), # 提供程序标志
("dwUIContext", wintypes.DWORD), # UI 上下文
("pSignatureSettings", ctypes.c_void_p), # 签名设置（保留）
]
信任验证动作 GUID（通用文件验证）
WINTRUST_ACTION_GENERIC_VERIFY_V2 = GUID(
0x00AAC56B, 0xCD44, 0x11D0,
(0x8C, 0xC2, 0x00, 0xC0, 0x4F, 0xC2, 0x95, 0xEE)
)
def win_verify_signature(file_path):
"""
使用 Windows 原生 WinVerifyTrust API 验证文件数字签名是否有效。
返回：(有效布尔值, 错误描述字符串)
"""
if not os.path.isfile(file_path):
return False, "文件不存在"
# 准备绝对路径
file_path = os.path.abspath(file_path)
# 构造文件信息
file_info = WINTRUST_FILE_INFO()
file_info.cbStruct = ctypes.sizeof(file_info)
file_info.pcwszFilePath = file_path
file_info.hFile = None
file_info.pgKnownSubject = None
# 构造验证数据
trust_data = WINTRUST_DATA()
trust_data.cbStruct = ctypes.sizeof(trust_data)
trust_data.dwUIChoice = WTD_UI_NONE
trust_data.fdwRevocationChecks = WTD_REVOKE_NONE # 忽略吊销检查以加速
trust_data.dwUnionChoice = WTD_CHOICE_FILE
trust_data.pFile = ctypes.pointer(file_info)
trust_data.dwStateAction = WTD_STATEACTION_VERIFY
trust_data.dwProvFlags = 0
# 加载 wintrust.dll 的 WinVerifyTrust 函数
wintrust = ctypes.windll.wintrust
WinVerifyTrust = wintrust.WinVerifyTrust
WinVerifyTrust.argtypes = [wintypes.HWND, POINTER(GUID), ctypes.c_void_p]
WinVerifyTrust.restype = ctypes.c_long
action = WINTRUST_ACTION_GENERIC_VERIFY_V2
status = WinVerifyTrust(None, byref(action), byref(trust_data))
# 关闭信任状态数据（必须释放）
trust_data.dwStateAction = WTD_STATEACTION_CLOSE
WinVerifyTrust(None, byref(action), byref(trust_data))
if status == 0: # ERROR_SUCCESS 即签名有效
return True, ""
else:
# 常见错误码翻译
errors = {
0x800B0100: "TRUST_E_NOSIGNATURE", # 无签名
0x800B0101: "CERT_E_EXPIRED", # 证书过期
0x800B0102: "CERT_E_UNTRUSTEDROOT", # 不受信任的根证书
0x800B0103: "CERT_E_CHAINING", # 证书链错误
0x800B0104: "TRUST_E_FAIL", # 信任验证失败
0x800B0105: "TRUST_E_TIME_STAMP", # 时间戳无效
0x800B0109: "CERT_E_VALIDITYPERIODNESTING",
0x800B010A: "CERT_E_REVOKED", # 证书已吊销
}
err_msg = errors.get(status & 0xFFFFFFFF, f"0x{status:08X}")
return False, err_msg
def extract_company_from_signature(file_path):
"""
从 PE 文件的 Authenticode 数字签名中提取“颁发给”的公司名。
内部流程：
- 解析 PE，定位安全目录（IMAGE_DIRECTORY_ENTRY_SECURITY）
- 读取原始签名数据（WIN_CERTIFICATE 结构）
- 提取 PKCS#7 签名并解析出所有证书
- 过滤掉证书颁发机构 (CA) 证书
- 过滤掉时间戳证书（根据主题关键词）
-
从真正的签名者证书中提取 Organization (O) 字段
返回：公司名字符串，若无签名或提取失败则返回 None
"""
try:
---- 1. 加载 PE 文件，完整解析 ----
pe = pefile.PE(file_path)
# 获取安全目录（包含了签名的原始数据地址和大小）
security_entry = pe.OPTIONAL_HEADER.DATA_DIRECTORY[
pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_SECURITY']
]
if security_entry.VirtualAddress == 0 or security_entry.Size == 0:
return None # 没有签名数据
# ---- 2. 读取签名原始字节 ----
with open(file_path, 'rb') as f:
f.seek(security_entry.VirtualAddress)
raw_sig = f.read(security_entry.Size)
if len(raw_sig) < 8:
return None
# WIN_CERTIFICATE 结构: dwLength(4) + wRevision(2) + wCertificateType(2)
cert_len = int.from_bytes(raw_sig[0:4], byteorder='little') # 结构总长度
cert_type = int.from_bytes(raw_sig[6:8], byteorder='little') # 证书类型
if cert_type != 2: # WIN_CERT_TYPE_PKCS_SIGNED_DATA 是 2
return None
# 提取真正的 PKCS#7 数据（跳过头部 8 字节）
pkcs_data = raw_sig[8:cert_len]
# ---- 3. 解析证书链 ----
try:
certs = pkcs7.load_der_pkcs7_certificates(pkcs_data)
except Exception:
# 回退到 BER 解析（兼容非严格 DER 编码的签名）
certs = pkcs7.load_pem_pkcs7_certificates(pkcs_data)
if not certs:
return None
# ---- 4. 筛选真正的签名者证书（跳过 CA 和时间戳证书） ----
timestamp_keywords = ['timestamp', 'tsa', 'time stamp']
signer_cert = None
for cert in certs:
# 检查 BasicConstraints 扩展，跳过 CA 证书
try:
bc = cert.extensions.get_extension_for_class(x509.BasicConstraints)
if bc.value.ca:
continue # 这是颁发者，跳过
except x509.extensions.ExtensionNotFound:
pass # 没有 BasicConstraints 通常是最终实体证书
# 检查主题是否包含时间戳相关关键词
subject_str = cert.subject.rfc4514_string().lower()
if any(kw in subject_str for kw in timestamp_keywords):
continue # 这是时间戳证书，跳过
# 符合条件，将其作为签名者证书
signer_cert = cert
break
# 如果所有证书都被过滤（极少情况），回退到第一个非 CA 证书
if signer_cert is None:
for cert in certs:
try:
bc = cert.extensions.get_extension_for_class(x509.BasicConstraints)
if bc.value.ca:
continue
except x509.extensions.ExtensionNotFound:
pass
signer_cert = cert
break
if signer_cert is None:
return None
# ---- 5. 提取公司名（优先 Organization，其次 Common Name） ----
org = signer_cert.subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)
if org:
return org[0].value
cn = signer_cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
if cn:
return cn[0].value
return "Unknown"
except Exception:
return None
===========================================================================
自定义后缀名管理窗口
===========================================================================
class ExtensionManager(tk.Toplevel):
"""
弹窗，允许用户增/删/改要扫描的文件后缀名。
会自动将用户输入格式化为带前导点的形式（如 'exe' -> '.exe'）。
每次修改后会调用 save_callback 以保存配置。
"""
def init(self, parent, extensions, save_callback=None):
super().init(parent)
self.title("管理扫描后缀")
self.geometry("300x300")
self.resizable(False, False)
# extensions 是父窗口的列表对象，直接修改它
self.extensions = extensions
self.save_callback = save_callback # 保存配置的回调函数
# ---- 当前后缀列表（带滚动条） ----
list_frame = ttk.Frame(self)
list_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
ttk.Label(list_frame, text="当前后缀名:").pack(anchor=tk.W)
self.listbox = tk.Listbox(list_frame, height=8)
self.listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.listbox.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.listbox.config(yscrollcommand=scrollbar.set)
# 填充列表
self._refresh_listbox()
# ---- 输入框 ----
entry_frame = ttk.Frame(self)
entry_frame.pack(fill=tk.X, padx=5, pady=2)
ttk.Label(entry_frame, text="后缀名:").pack(side=tk.LEFT)
self.ext_var = tk.StringVar()
self.ext_entry = ttk.Entry(entry_frame, textvariable=self.ext_var, width=10)
self.ext_entry.pack(side=tk.LEFT, padx=5)
# ---- 操作按钮 ----
btn_frame = ttk.Frame(self)
btn_frame.pack(fill=tk.X, padx=5, pady=5)
ttk.Button(btn_frame, text="添加", command=self.add_extension).pack(side=tk.LEFT, padx=2)
ttk.Button(btn_frame, text="删除选中", command=self.delete_extension).pack(side=tk.LEFT, padx=2)
ttk.Button(btn_frame, text="修改为", command=self.modify_extension).pack(side=tk.LEFT, padx=2)
ttk.Button(btn_frame, text="重置默认", command=self.reset_default).pack(side=tk.LEFT, padx=2)
# 关闭按钮
ttk.Button(self, text="关闭", command=self.on_close).pack(pady=5)
# 绑定双击列表项进行修改
self.listbox.bind("<Double-Button-1>", self.on_double_click)
# ---- 工具函数：格式化后缀，确保以点开头 ----
def _normalize_ext(self, ext):
"""去掉首尾空格，确保以 '.' 开头，并转为小写"""
ext = ext.strip()
if not ext:
return None
if not ext.startswith('.'):
ext = '.' + ext
return ext.lower()
# ---- 刷新列表显示 ----
def _refresh_listbox(self):
self.listbox.delete(0, tk.END)
for ext in self.extensions:
self.listbox.insert(tk.END, ext)
# ---- 保存配置回调包装 ----
def _save(self):
if self.save_callback:
self.save_callback()
# ---- 添加后缀 ----
def add_extension(self):
raw = self.ext_var.get()
norm = self._normalize_ext(raw)
if norm is None:
messagebox.showwarning("输入错误", "后缀名不能为空")
return
if norm in self.extensions:
messagebox.showinfo("重复", f"后缀 {norm} 已存在")
return
self.extensions.append(norm)
self._refresh_listbox()
self.ext_var.set("")
self._save()
# ---- 删除选中后缀 ----
def delete_extension(self):
selection = self.listbox.curselection()
if not selection:
messagebox.showwarning("未选择", "请先在列表中选中一个后缀")
return
idx = selection[0]
# 至少保留一个后缀，避免扫描时无后缀
if len(self.extensions) <= 1:
messagebox.showwarning("操作禁止", "至少需要保留一个后缀名")
return
del self.extensions[idx]
self._refresh_listbox()
self._save()
# ---- 修改选中后缀 ----
def modify_extension(self):
raw = self.ext_var.get()
norm = self._normalize_ext(raw)
if norm is None:
messagebox.showwarning("输入错误", "请输入有效的新后缀名")
return
selection = self.listbox.curselection()
if not selection:
messagebox.showwarning("未选择", "请先选中要修改的后缀")
return
idx = selection[0]
# 检查是否与新后缀重复（除了当前位置）
if norm in self.extensions and self.extensions.index(norm) != idx:
messagebox.showinfo("重复", f"后缀 {norm} 已存在")
return
self.extensions[idx] = norm
self._refresh_listbox()
self.ext_var.set("")
self._save()
# ---- 重置为默认后缀 ----
def reset_default(self):
if messagebox.askyesno("确认重置", "重置后将恢复到默认后缀 (.exe .dll .sys)，确定吗？"):
default = ['.exe', '.dll', '.sys']
self.extensions.clear()
self.extensions.extend(default)
self._refresh_listbox()
self.ext_var.set("")
self._save()
# ---- 双击列表项，快速填充到输入框 ----
def on_double_click(self, event):
selection = self.listbox.curselection()
if selection:
ext = self.extensions[selection[0]]
self.ext_var.set(ext)
# ---- 关闭窗口 ----
def on_close(self):
self.destroy()
===========================================================================
图形界面主类
===========================================================================
class SigScannerGUI:
def init(self, root):
self.root = root
self.root.title("PE 签名扫描器 - 白加黑检测辅助")
self.root.geometry("800x560") # 窗口大小
# 线程控制与数据存储
self.scan_thread = None # 后台扫描线程
self.stop_flag = False # 停止标志
self.need_update_companies = False # 是否需要在扫描结束后更新公司列表
self.queue = queue.Queue() # 线程间安全队列
self.all_results = [] # 存储所有扫描结果 (path, has_sig, valid, company)
self.filter_mode = "all" # 当前过滤模式：all / unsigned / invalid / company
self.selected_company = None # 按公司过滤时选中的公司名
# 自定义扫描后缀名列表：先从配置文件加载，失败则使用默认值
self.extensions = self.load_extensions_from_config()
# 记录启动信息
log_message("INFO", "程序启动")
# ---------- 顶部控制栏 ----------
control_frame = ttk.Frame(root, padding=5)
control_frame.pack(fill=tk.X)
ttk.Label(control_frame, text="扫描目录:").pack(side=tk.LEFT, padx=5)
self.dir_var = tk.StringVar()
self.dir_entry = ttk.Entry(control_frame, textvariable=self.dir_var, width=40)
self.dir_entry.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True)
ttk.Button(control_frame, text="选择目录", command=self.choose_directory).pack(side=tk.LEFT, padx=2)
ttk.Button(control_frame, text="管理后缀", command=self.open_extension_manager).pack(side=tk.LEFT, padx=2)
self.scan_btn = ttk.Button(control_frame, text="开始扫描", command=self.start_scan)
self.scan_btn.pack(side=tk.LEFT, padx=2)
self.stop_btn = ttk.Button(control_frame, text="停止", command=self.stop_scan, state=tk.DISABLED)
self.stop_btn.pack(side=tk.LEFT, padx=2)
# ---------- 结果表格 ----------
table_frame = ttk.Frame(root)
table_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
columns = ("path", "has_sig", "valid", "company")
self.tree = ttk.Treeview(table_frame, columns=columns, show="headings", height=10)
self.tree.heading("path", text="文件路径", anchor=tk.W)
self.tree.heading("has_sig", text="有签名", anchor=tk.CENTER)
self.tree.heading("valid", text="签名有效", anchor=tk.CENTER)
self.tree.heading("company", text="公司名称", anchor=tk.W)
self.tree.column("path", width=350, stretch=True)
self.tree.column("has_sig", width=50, anchor=tk.CENTER, stretch=False)
self.tree.column("valid", width=70, anchor=tk.CENTER, stretch=False)
self.tree.column("company", width=150, stretch=False)
vsb = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=vsb.set)
self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
vsb.pack(side=tk.RIGHT, fill=tk.Y)
# 为列头绑定排序功能
for col in columns:
self.tree.heading(col, command=lambda c=col: self.sort_treeview(c, False))
# ---------- 底部过滤控制区 ----------
bottom_container = ttk.Frame(root, padding=5)
bottom_container.pack(fill=tk.X)
# 第一行：通用过滤按钮
filter_row = ttk.Frame(bottom_container)
filter_row.pack(fill=tk.X, pady=2)
ttk.Button(filter_row, text="显示无签名", command=lambda: self.apply_filter("unsigned")).pack(side=tk.LEFT, padx=5)
ttk.Button(filter_row, text="显示签名无效", command=lambda: self.apply_filter("invalid")).pack(side=tk.LEFT, padx=5)
ttk.Button(filter_row, text="显示全部", command=lambda: self.apply_filter("all")).pack(side=tk.LEFT, padx=5)
# 第二行：按公司过滤
company_row = ttk.Frame(bottom_container)
company_row.pack(fill=tk.X, pady=2)
self.company_label = ttk.Label(company_row, text="共有 0 家不同公司的签名")
self.company_label.pack(side=tk.LEFT, padx=5)
self.company_combo = ttk.Combobox(company_row, state="readonly", width=30)
self.company_combo.pack(side=tk.LEFT, padx=5)
ttk.Button(company_row, text="查看该公司签名", command=self.filter_by_company).pack(side=tk.LEFT, padx=5)
# 状态栏（底边）
self.status_var = tk.StringVar(value="就绪")
status_bar = ttk.Label(root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W, padding=2)
status_bar.pack(fill=tk.X, side=tk.BOTTOM)
# ========== 配置文件的加载与保存 ==========
def load_extensions_from_config(self):
"""从 config.json 加载后缀列表，失败则使用默认值"""
default = ['.exe', '.dll', '.sys']
if os.path.isfile(CONFIG_FILE):
try:
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, dict) and 'extensions' in data:
exts = data['extensions']
if isinstance(exts, list) and all(isinstance(e, str) for e in exts):
log_message("INFO", f"从配置文件加载后缀: {exts}")
return exts
except Exception as e:
log_message("ERROR", f"加载配置文件失败: {e}")
# 加载失败或文件不存在，使用默认
log_message("INFO", "使用默认后缀列表")
return default
def save_extensions_to_config(self):
"""将当前后缀列表保存到 config.json"""
try:
data = {'extensions': self.extensions}
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
log_message("INFO", f"保存后缀配置: {self.extensions}")
except Exception as e:
log_message("ERROR", f"保存配置文件失败: {e}")
# ========== 后缀管理窗口 ==========
def open_extension_manager(self):
"""
弹出后缀名管理窗口，允许用户自定义要扫描的文件后缀。
每次修改会立即保存配置。
"""
ExtensionManager(self.root, self.extensions, save_callback=self.save_extensions_to_config)
# ---- 目录选择 ----
def choose_directory(self):
directory = filedialog.askdirectory(title="选择要扫描的根目录")
if directory:
self.dir_var.set(directory)
# ---- 开始扫描 ----
def start_scan(self):
target_dir = self.dir_var.get().strip()
if not target_dir:
messagebox.showwarning("警告", "请先选择扫描目录")
return
if not os.path.isdir(target_dir):
messagebox.showerror("错误", "目录不存在")
return
if not self.extensions:
messagebox.showwarning("警告", "请至少设置一个扫描后缀名")
return
# 记录扫描开始
log_message("INFO", f"开始扫描目录: {target_dir}，后缀: {self.extensions}")
# 重置 UI 与数据
for item in self.tree.get_children():
self.tree.delete(item)
self.all_results.clear()
self.filter_mode = "all"
self.selected_company = None
self.company_combo['values'] = []
self.company_combo.set('')
self.company_label.config(text="共有 0 家不同公司的签名")
self.need_update_companies = False
self.stop_flag = False
self.scan_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.NORMAL)
self.status_var.set(f"正在扫描... (后缀: {', '.join(self.extensions)})")
self.queue.queue.clear()
# 启动后台扫描线程
self.scan_thread = threading.Thread(target=self.scan_worker, args=(target_dir,), daemon=True)
self.scan_thread.start()
self.root.after(100, self.process_queue) # 开始定期处理队列
# ---- 后台扫描线程 ----
def scan_worker(self, root_dir):
"""
在后台执行扫描，支持随时中断：
1. 先用 os.walk 收集目标文件（每一步检查停止标志）
2. 逐个分析签名，将结果放入队列
"""
target_exts = {ext.lower() for ext in self.extensions}
files = []
# ---- 文件收集阶段 ----
try:
for dirpath, dirnames, filenames in os.walk(root_dir):
if self.stop_flag:
self.queue.put(("progress", "扫描已停止"))
return
for fname in filenames:
if self.stop_flag:
self.queue.put(("progress", "扫描已停止"))
return
if Path(fname).suffix.lower() in target_exts:
files.append(os.path.join(dirpath, fname))
except Exception as e:
err_msg = f"扫描目录失败: {e}"
log_message("ERROR", err_msg)
self.queue.put(("error", err_msg))
return
total = len(files)
self.queue.put(("progress", f"发现 {total} 个文件，开始分析..."))
# ---- 文件分析阶段 ----
for idx, full_path in enumerate(files, 1):
if self.stop_flag:
self.queue.put(("progress", "扫描已停止"))
return
# 提取公司名（公司名存在即视为有签名）
try:
company = extract_company_from_signature(full_path)
except Exception as e:
log_message("ERROR", f"提取签名时异常 [{full_path}]: {e}")
company = None
has_sig = company is not None
valid = False
if has_sig:
try:
valid, _ = win_verify_signature(full_path)
except Exception as e:
log_message("ERROR", f"验证签名时异常 [{full_path}]: {e}")
result_data = (full_path, has_sig, valid, company)
self.queue.put(("result", result_data))
self.queue.put(("progress", f"进度: {idx}/{total}"))
self.queue.put(("done", "扫描完成"))
# ---- 定期处理队列，更新 GUI ----
def process_queue(self):
"""
每隔 100ms 调用一次，从队列取出消息并更新界面。
消息类型：
- "result" : 一条新的扫描结果
- "progress" : 进度文本，更新状态栏
- "done" : 扫描正常完成
- "error" : 扫描出错
额外处理：检测到线程已结束但未收到 done 时，作为手动停止处理。
"""
try:
while True:
msg_type, data = self.queue.get_nowait()
if msg_type == "result":
path, has_sig, valid, company = data
sig_display = "有" if has_sig else "无"
valid_display = "有效" if (has_sig and valid) else ("无效" if has_sig else "-")
company_display = company if company else "-"
# 存储原始数据
self.all_results.append((path, has_sig, valid, company))
# 只有符合当前过滤条件才插入表格
if self._match_filter(has_sig, valid, company):
self.tree.insert("", tk.END, values=(path, sig_display, valid_display, company_display))
elif msg_type == "progress":
self.status_var.set(data)
elif msg_type == "done":
# 扫描正常结束：更新公司下拉列表
self._update_company_list()
self.status_var.set(data)
self.scan_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.need_update_companies = False
log_message("INFO", "扫描完成")
elif msg_type == "error":
messagebox.showerror("错误", data)
self.status_var.set("扫描出错")
self.scan_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.need_update_companies = False
log_message("ERROR", data)
except queue.Empty:
pass
finally:
# 如果扫描线程已经结束但还没有收到 done，说明是手动停止
if self.scan_thread and not self.scan_thread.is_alive():
if self.need_update_companies:
self._update_company_list()
self.status_var.set("扫描已停止")
self.scan_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.need_update_companies = False
log_message("INFO", "用户手动停止扫描")
# 如果线程还在运行，继续定期检查
if self.scan_thread and self.scan_thread.is_alive():
self.root.after(100, self.process_queue)
# ---- 过滤匹配判断 ----
def _match_filter(self, has_sig, valid, company):
"""根据当前过滤模式判断某条记录是否应该显示"""
if self.filter_mode == "all":
return True
elif self.filter_mode == "unsigned":
return not has_sig
elif self.filter_mode == "invalid":
return has_sig and (not valid)
elif self.filter_mode == "company":
return has_sig and company and company == self.selected_company
return True
# ---- 更新公司下拉列表和统计标签 ----
def _update_company_list(self):
"""从 self.all_results 提取所有唯一的有效签名公司名称"""
companies = set()
for _, has_sig, _, company in self.all_results:
if has_sig and company and company != "-":
companies.add(company)
sorted_companies = sorted(companies)
self.company_combo['values'] = sorted_companies
self.company_label.config(text=f"共有 {len(sorted_companies)} 家不同公司的签名")
if sorted_companies:
self.company_combo.current(0)
self.selected_company = sorted_companies[0]
else:
self.company_combo.set('')
self.selected_company = None
# ---- 通用过滤（无签名/无效/全部） ----
def apply_filter(self, mode):
self.filter_mode = mode
self.selected_company = None
self._refresh_table()
# ---- 按公司过滤 ----
def filter_by_company(self):
company = self.company_combo.get()
if not company:
messagebox.showinfo("提示", "请先选择一家公司")
return
self.filter_mode = "company"
self.selected_company = company
self._refresh_table()
# ---- 刷新表格（根据当前过滤条件重新渲染） ----
def _refresh_table(self):
for item in self.tree.get_children():
self.tree.delete(item)
for path, has_sig, valid, company in self.all_results:
if self._match_filter(has_sig, valid, company):
sig_display = "有" if has_sig else "无"
valid_display = "有效" if (has_sig and valid) else ("无效" if has_sig else "-")
company_display = company if company else "-"
self.tree.insert("", tk.END, values=(path, sig_display, valid_display, company_display))
count = len(self.tree.get_children())
total = len(self.all_results)
mode_desc = {
"all": "全部",
"unsigned": "无签名",
"invalid": "签名无效",
"company": f"公司: {self.selected_company}"
}.get(self.filter_mode, self.filter_mode)
self.status_var.set(f"过滤模式: {mode_desc} 显示 {count} / {total} 条结果")
# ---- 用户点击停止 ----
def stop_scan(self):
self.stop_flag = True
self.need_update_companies = True
self.status_var.set("正在停止...")
self.stop_btn.config(state=tk.DISABLED)
self.root.update_idletasks()
# ---- 表格列排序（简单字符串比较） ----
def sort_treeview(self, col, reverse):
items = [(self.tree.set(item, col), item) for item in self.tree.get_children()]
items.sort(key=lambda x: x[0], reverse=reverse)
for index, (val, item) in enumerate(items):
self.tree.move(item, '', index)
self.tree.heading(col, command=lambda: self.sort_treeview(col, not reverse))
===========================================================================
程序入口
===========================================================================
if name == "main":
if sys.platform != "win32":
print("此工具只能在 Windows 上运行（需要 WinVerifyTrust API）")
sys.exit(1)
root = tk.Tk()
app = SigScannerGUI(root)
root.mainloop()
发表于 2026-7-14 01:47
分享到朋友圈
