天晴安全分析台 v7.0 浏览数 - 95 发布于 - 2026-06-13 - 22:05
功能 监控火绒拦截日志，自动识别并清除指定文件夹内的病毒。
工作流程：
火绒拦截任何程序写入注册表 Run 键
本工具读取火绒日志，提取病毒文件路径
判断病毒是否在用户指定的文件夹内
如果是，调用 KVRT 清除病毒（优先剥离，保留原文件）
砍掉的功能 v7 之前的复杂模块已全部移除：
静态分析引擎（查壳/签名/字符串/YARA）
三引擎联网查询（VT/微步/魔盾）
智能解包
HTML 报告
评分逻辑
病毒库管理
其他冗余代码
当前代码约 400 行，只做一件事：自动清除指定文件夹内的病毒。
依赖
火绒安全 6.0 或更高版本
KVRT.exe（卡巴斯基免费清除工具）
Python 3.8 或更高版本
配置步骤 1. 下载 KVRT 放入 天晴启动器/KVRT.exe
2. 配置火绒自定义规则 防护中心 → 高级防护 → 自定义防护 → 添加规则：
注册表路径：HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Windows\CurrentVersion\Run*
保护动作：创建、修改、删除
处理方式：直接阻止
记录日志：勾选
3. 运行程序 双击 main.py → 同意免责声明 → 选择游戏文件夹 → 点击启动监控
文件结构 text
天晴安全分析台/
├── main.py
├── 天晴启动器/
│ └── KVRT.exe
├── cruise.log
└── .config.json
免责声明 本工具仅供安全研究使用，使用后果自行承担。
复制（代码） #!/usr/bin/env python3
-- coding: utf-8 - -
"""
天晴安全分析台 v11.0 - 全自动静默巡航版
功能：监控火绒拦截日志，自动提取病毒路径，静默调用 KVRT Cure
特点：全程无弹窗、无用户交互、无打扰
用法：拖入文件夹 → 后台自动运行 → 病毒自动清除
Developer: 天晴
"""
import sys
import os
import json
import time
import sqlite3
import subprocess
import logging
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
from pathlib import Path
from datetime import datetime
from threading import Thread, Lock
================================================================
免责声明
================================================================
DISCLAIMER_TEXT = """
╔══════════════════════════════════════════════════════════════════╗
║ 天晴安全分析台 免责声明 ║
╠══════════════════════════════════════════════════════════════════╣
║ ║
║ 1. 本工具仅供安全研究、学习交流使用，严禁用于任何非法目的。 ║
║ ║
║ 2. 本工具依赖火绒安全软件和卡巴斯基 KVRT，请确保已安装并配置。 ║
║ ║
║ 3. 使用本工具产生的任何直接或间接后果，由使用者自行承担。 ║
║ ║
║ 4. 本工具不替代专业杀毒软件，检测结果仅供参考。 ║
║ ║
║ 5. 如涉及侵权问题，请联系开发者处理。 ║
║ ║
║ 6. 继续使用即表示您已阅读、理解并同意本免责声明的全部条款。 ║
║ ║
╚══════════════════════════════════════════════════════════════════╝
"""
================================================================
配置
================================================================
SCRIPT_DIR = Path(file ).parent.absolute()
TOOLS_DIR = SCRIPT_DIR / "天晴启动器"
KVRT_EXE = TOOLS_DIR / "KVRT.exe"
火绒日志数据库路径
HUORONG_DATA_DIR = Path("C:/ProgramData/Huorong/Sysdiag")
LOG_DB_PATH = HUORONG_DATA_DIR / "log.db"
全局变量
TARGET_FOLDER = None
last_processed_id = 0
monitoring = False
monitor_thread = None
lock = Lock()
日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(SCRIPT_DIR / "cruise.log", encoding='utf-8'),
logging.StreamHandler()
]
)
================================================================
免责声明界面
================================================================
def show_disclaimer_gui():
"""显示免责声明窗口"""
agreed_file = SCRIPT_DIR / ".disclaimer_agreed"
# 检查是否已同意且在有效期内
if agreed_file.exists():
try:
with open(agreed_file, 'r', encoding='utf-8') as f:
data = json.load(f)
agreed_date = datetime.strptime(data['date'], "%Y-%m-%d")
if (datetime.now() - agreed_date).days < 7:
return True
else:
agreed_file.unlink()
except:
if agreed_file.exists():
agreed_file.unlink()
# 创建窗口
win = tk.Tk()
win.title("天晴安全分析台 - 免责声明")
win.geometry("650x550")
win.resizable(False, False)
# 居中
win.update_idletasks()
x = (win.winfo_screenwidth() // 2) - (650 // 2)
y = (win.winfo_screenheight() // 2) - (550 // 2)
win.geometry(f"+{x}+{y}")
# 标题
tk.Label(win, text="天晴安全分析台", font=('微软雅黑', 20, 'bold'), fg='#333').pack(pady=25)
# 免责声明文本
text_area = scrolledtext.ScrolledText(win, font=('微软雅黑', 10), height=18, wrap=tk.WORD)
text_area.pack(fill=tk.BOTH, expand=True, padx=30, pady=10)
text_area.insert(tk.END, DISCLAIMER_TEXT)
text_area.config(state=tk.DISABLED)
# 按钮
btn_frame = tk.Frame(win)
btn_frame.pack(pady=25)
agreed = False
def on_agree():
nonlocal agreed
with open(agreed_file, 'w', encoding='utf-8') as f:
json.dump({'date': datetime.now().strftime("%Y-%m-%d")}, f)
agreed = True
win.destroy()
def on_disagree():
win.destroy()
tk.Button(btn_frame, text="同意并继续", command=on_agree, bg='#4CAF50', fg='white',
font=('微软雅黑', 12), padx=30, pady=6).pack(side=tk.LEFT, padx=20)
tk.Button(btn_frame, text="拒绝并退出", command=on_disagree, bg='#f44336', fg='white',
font=('微软雅黑', 12), padx=30, pady=6).pack(side=tk.LEFT, padx=20)
win.protocol("WM_DELETE_WINDOW", on_disagree)
win.mainloop()
return agreed
================================================================
KVRT 清除
================================================================
def kvrt_cure(virus_path):
"""静默调用 KVRT 修复病毒文件"""
if not virus_path or not Path(virus_path).exists():
return False
if not KVRT_EXE.exists():
logging.error(f"KVRT.exe 未找到: {KVRT_EXE}")
return False
logging.info(f"正在清除病毒: {virus_path}")
try:
result = subprocess.run([
str(KVRT_EXE),
"-accepteula",
"-silent",
"-processlevel", "1",
"-customonly",
"-custom", virus_path
], capture_output=True, timeout=300)
if result.returncode == 0:
logging.info(f"清除成功: {virus_path}")
return True
else:
logging.warning(f"清除失败，返回码: {result.returncode}")
return False
except subprocess.TimeoutExpired:
logging.error(f"清除超时: {virus_path}")
return False
except Exception as e:
logging.error(f"清除异常: {e}")
return False
================================================================
日志解析
================================================================
def extract_virus_path_from_log(row, columns):
"""从日志行中提取真正的病毒文件路径"""
if not row:
return None
# 查找字段索引
opt_process_idx = None
parent_idx = None
for i, col in enumerate(columns):
if col in ['opt_process', 'OperateProcess', 'process']:
opt_process_idx = i
if col in ['parent_process', 'ParentProcess', 'parent']:
parent_idx = i
if opt_process_idx is None:
return None
opt_process = row[opt_process_idx] if opt_process_idx < len(row) else ''
parent_process = row[parent_idx] if parent_idx and parent_idx < len(row) else ''
# 如果操作进程是系统工具，则使用父进程
if opt_process and parent_process:
lower_op = opt_process.lower()
system_tools = ['regedit.exe', 'cmd.exe', 'powershell.exe', 'reg.exe', 'wmic.exe']
if any(tool in lower_op for tool in system_tools):
return parent_process
return opt_process
================================================================
获取最新拦截记录
================================================================
def get_latest_blocked_record():
"""从火绒日志数据库获取最新的已阻止记录"""
global last_processed_id
if not LOG_DB_PATH.exists():
return None, None
try:
conn = sqlite3.connect(f"file:{LOG_DB_PATH}?mode=ro", uri=True)
cursor = conn.cursor()
# 获取表名
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'HrLogV3%'")
tables = cursor.fetchall()
if not tables:
conn.close()
return None, None
table_name = tables[0][0]
# 获取列名
cursor.execute(f"PRAGMA table_info({table_name})")
columns = [col[1] for col in cursor.fetchall()]
# 确定需要的字段名
id_col = 'id' if 'id' in columns else 'Id'
opt_result_col = 'opt_result' if 'opt_result' in columns else 'OptResult'
# 查询最新的一条已阻止记录
query = f"""
SELECT *
FROM {table_name}
WHERE {opt_result_col} = '已阻止'
ORDER BY {id_col} DESC
LIMIT 1
"""
cursor.execute(query)
row = cursor.fetchone()
conn.close()
if row:
record_id = row[0]
if record_id > last_processed_id:
with lock:
last_processed_id = record_id
return row, columns
return None, None
except Exception as e:
logging.debug(f"读取日志失败: {e}")
return None, None
================================================================
监控循环
================================================================
def monitor_loop(status_callback=None):
"""轮询监控火绒日志"""
global monitoring
logging.info(f"开始监控目标文件夹: {TARGET_FOLDER}")
logging.info(f"KVRT 路径: {KVRT_EXE}")
while monitoring:
try:
row, columns = get_latest_blocked_record()
if row:
virus_path = extract_virus_path_from_log(row, columns)
if virus_path and Path(virus_path).exists():
virus_file = Path(virus_path)
is_in_target = False
if TARGET_FOLDER:
try:
virus_file.relative_to(TARGET_FOLDER)
is_in_target = True
except ValueError:
pass
if is_in_target:
logging.info(f"跳过圈内文件: {virus_path}")
else:
logging.warning(f"检测到圈外病毒: {virus_path}")
kvrt_cure(virus_path)
if status_callback:
status_callback(f"已清除: {Path(virus_path).name}")
time.sleep(2)
except Exception as e:
logging.error(f"监控循环异常: {e}")
time.sleep(5)
logging.info("监控已停止")
================================================================
兼容性检测
================================================================
def check_compatibility():
"""检查火绒日志和KVRT"""
issues = []
# 检查火绒日志
if not LOG_DB_PATH.exists():
issues.append("火绒日志文件不存在，请确保火绒已安装并运行")
return False, issues
# 检查日志表结构
try:
conn = sqlite3.connect(f"file:{LOG_DB_PATH}?mode=ro", uri=True)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'HrLogV3%'")
if not cursor.fetchall():
issues.append("火绒版本可能不兼容，需要 6.0 或更高版本")
conn.close()
return False, issues
conn.close()
except Exception as e:
issues.append(f"数据库检测失败: {e}")
return False, issues
# 检查 KVRT
if not KVRT_EXE.exists():
issues.append(f"KVRT.exe 未找到，请放入 {TOOLS_DIR}")
return False, issues
return True, issues
================================================================
主界面
================================================================
class TianQingApp:
def init (self):
self.root = tk.Tk()
self.root.title("天晴安全分析台 v11.0")
self.root.geometry("650x500")
self.root.resizable(False, False)
# 居中
self.root.update_idletasks()
x = (self.root.winfo_screenwidth() // 2) - (650 // 2)
y = (self.root.winfo_screenheight() // 2) - (500 // 2)
self.root.geometry(f"+{x}+{y}")
self.setup_ui()
# 监控状态
self.monitoring = False
self.monitor_thread = None
# 加载配置
self.load_config()
def setup_ui(self):
"""设置界面"""
# 标题区域
title_frame = tk.Frame(self.root)
title_frame.pack(fill='x', pady=15)
tk.Label(title_frame, text="天晴安全分析台", font=('微软雅黑', 18, 'bold'),
fg='#333').pack()
tk.Label(title_frame, text="全自动静默巡航版", font=('微软雅黑', 10),
fg='#666').pack()
# 分隔线
ttk.Separator(self.root, orient='horizontal').pack(fill='x', padx=20, pady=10)
# 监控文件夹选择
folder_frame = tk.LabelFrame(self.root, text="监控设置", font=('微软雅黑', 10, 'bold'),
padx=10, pady=10)
folder_frame.pack(fill='x', padx=20, pady=10)
tk.Label(folder_frame, text="目标文件夹：", font=('微软雅黑', 10), width=12,
anchor='w').grid(row=0, column=0, padx=5, pady=5)
self.folder_var = tk.StringVar()
self.folder_entry = tk.Entry(folder_frame, textvariable=self.folder_var,
font=('微软雅黑', 10), state='readonly')
self.folder_entry.grid(row=0, column=1, sticky='ew', padx=5, pady=5)
tk.Button(folder_frame, text="浏览...", command=self.select_folder,
font=('微软雅黑', 9), width=8).grid(row=0, column=2, padx=5, pady=5)
folder_frame.columnconfigure(1, weight=1)
# 状态显示
status_frame = tk.LabelFrame(self.root, text="运行状态", font=('微软雅黑', 10, 'bold'),
padx=10, pady=10)
status_frame.pack(fill='x', padx=20, pady=10)
self.status_var = tk.StringVar(value="● 未启动")
self.status_label = tk.Label(status_frame, textvariable=self.status_var,
font=('微软雅黑', 11), fg='#999')
self.status_label.pack(anchor='w')
# 日志区域
log_frame = tk.LabelFrame(self.root, text="运行日志", font=('微软雅黑', 10, 'bold'),
padx=10, pady=5)
log_frame.pack(fill='both', expand=True, padx=20, pady=10)
self.log_text = scrolledtext.ScrolledText(log_frame, font=('Consolas', 9),
height=10, wrap=tk.WORD)
self.log_text.pack(fill='both', expand=True)
# 按钮区域
btn_frame = tk.Frame(self.root)
btn_frame.pack(pady=15)
self.start_btn = tk.Button(btn_frame, text="启动监控", command=self.start_monitoring,
bg='#4CAF50', fg='white', font=('微软雅黑', 10),
width=12, height=1)
self.start_btn.pack(side=tk.LEFT, padx=8)
self.stop_btn = tk.Button(btn_frame, text="停止监控", command=self.stop_monitoring,
bg='#f44336', fg='white', font=('微软雅黑', 10),
width=12, height=1, state=tk.DISABLED)
self.stop_btn.pack(side=tk.LEFT, padx=8)
tk.Button(btn_frame, text="查看日志文件", command=self.view_log,
bg='#2196F3', fg='white', font=('微软雅黑', 10),
width=12, height=1).pack(side=tk.LEFT, padx=8)
tk.Button(btn_frame, text="清空日志", command=self.clear_log,
bg='#FF9800', fg='white', font=('微软雅黑', 10),
width=10, height=1).pack(side=tk.LEFT, padx=8)
# 底部状态栏
status_bar = tk.Frame(self.root, height=28, bg='#f0f0f0')
status_bar.pack(fill='x', side=tk.BOTTOM)
self.status_bar_label = tk.Label(status_bar, text="就绪", font=('微软雅黑', 9),
bg='#f0f0f0', anchor='w')
self.status_bar_label.pack(fill='x', padx=15, pady=5)
def log(self, message):
"""添加日志"""
timestamp = datetime.now().strftime("%H:%M:%S")
self.log_text.insert(tk.END, f"[{timestamp}] {message}\n")
self.log_text.see(tk.END)
logging.info(message)
def select_folder(self):
"""选择文件夹"""
folder = filedialog.askdirectory(title="选择要监控的游戏文件夹（可选中整个盘符）")
if folder:
self.folder_var.set(folder)
self.save_config()
self.log(f"已选择监控文件夹：{folder}")
def load_config(self):
"""加载配置"""
config_file = SCRIPT_DIR / ".config.json"
if config_file.exists():
try:
with open(config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
if 'target_folder' in config:
self.folder_var.set(config['target_folder'])
except:
pass
def save_config(self):
"""保存配置"""
config_file = SCRIPT_DIR / ".config.json"
try:
with open(config_file, 'w', encoding='utf-8') as f:
json.dump({'target_folder': self.folder_var.get()}, f)
except:
pass
def status_callback(self, message):
"""状态回调"""
self.log(message)
self.status_bar_label.config(text=message)
def start_monitoring(self):
"""启动监控"""
global TARGET_FOLDER, monitoring, monitor_thread
folder = self.folder_var.get()
if not folder:
messagebox.showwarning("提示", "请先选择要监控的文件夹")
return
target_path = Path(folder)
if not target_path.exists():
messagebox.showwarning("提示", "文件夹不存在")
return
# 兼容性检测
compatible, issues = check_compatibility()
if not compatible:
messagebox.showerror("兼容性错误", "\n".join(issues))
return
TARGET_FOLDER = target_path
monitoring = True
# 启动监控线程
monitor_thread = Thread(target=monitor_loop, args=(self.status_callback,), daemon=True)
monitor_thread.start()
self.monitoring = True
self.status_var.set("● 运行中")
self.status_label.config(fg='#4CAF50')
self.start_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.NORMAL)
self.status_bar_label.config(text=f"监控中：{folder}")
self.log(f"监控已启动，目标文件夹：{folder}")
def stop_monitoring(self):
"""停止监控"""
global monitoring
monitoring = False
self.monitoring = False
self.status_var.set("● 已停止")
self.status_label.config(fg='#999')
self.start_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.status_bar_label.config(text="监控已停止")
self.log("监控已停止")
def view_log(self):
"""查看日志文件"""
log_file = SCRIPT_DIR / "cruise.log"
if log_file.exists():
os.startfile(str(log_file))
else:
messagebox.showinfo("提示", "日志文件不存在")
def clear_log(self):
"""清空日志"""
self.log_text.delete(1.0, tk.END)
self.log("日志已清空")
def run(self):
"""运行程序"""
# 显示免责声明
if not show_disclaimer_gui():
sys.exit(0)
# 检查 KVRT
if not KVRT_EXE.exists():
messagebox.showerror("错误", f"KVRT.exe 未找到\n请放入：{TOOLS_DIR}")
sys.exit(1)
self.root.mainloop()
