2021年的时候写了一个安全事件分析小工具，这是我为安全驻场大头兵写的第一个小工具，基于pyinstaller打包的pe软件，使用的时候非常简单，只需要将态势感知上的安全事件列表导出，导入到小工具中，即可实现自动分析，一方面是帮助安全驻场理解安全事件，另一方面是收集每个现场的安全事件，以便于后续对运营效果进行评估，今年偶然一个机会，发现需要增强对逆向技能的学习，了解到可以对pyinstaller打包的exe软件逆向出python文件，于是想起之前github上有上传过自己写的小工具，于是有了本次的逆向工程～
提醒：故事有后续，逆向出pyinstaller打包的exe软件的所有源代码：ailx10：逆向pyinstaller打包的exe软件，获取python源码(4)
ailx10
网络安全优秀回答者
网络安全硕士
去咨询
小工具说明：
- 本工具是没有经过专业测试的v3.0
- 帮助解决安全事件分析、处置相关的常见问题，辅助一线快速分析
- 使用过程中可能由于安全事件数据字段内容缺失，软件会自动退出
- 如果遇到软件退出的问题，请将安全事件发给我，并提供自己的输入信息
- 如果你有一些好的想法，也可以给我提建议哦~
优化：
- 规则联动，无需用户手动输入
- 自适应屏幕分辨率
- 自动联网查询Virus Total情报IOC
- 优化事件分析逻辑函数，更加友好平滑
- 优化部分事件的描述信息，更加准确
- 添加数据校验码，验证数据的完整性
- VT API是我的个人账号，查询次数受限制，为正常现象
第一步：对exe程序进行反编译[1]
python pyinstxtractor.py 安全事件分析main.exe第二步：进入新获得的extracted文件夹
第三步：查看struct.pyc和main.pyc前12字节之间的区别
第四步：反编译pyc文件得到python源代码
uncompyle6 安全事件分析main.pyc > main.py第五步：欣赏一下反编译的代码
非常遗憾，暂时只能看到主函数，看不到其他函数
# uncompyle6 version 3.9.0
# Python bytecode version base 3.6 (3379)
# Decompiled from: Python 3.6.13 |Anaconda, Inc.| (default, Mar 16 2021, 11:37:27) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: 安全事件分析main.py
"""
@File : 安全事件分析main.py
@Time : 2021/8/3 17:53
@Author : ailx10
@Software: PyCharm
"""
import sys
from datetime import datetime
import hashlib
from PyQt5.QtWidgets import QApplication, QMainWindow, QFileDialog
from pandas import read_excel
from pandas import DataFrame
from 安全事件说明 import *
from 学习sqllite import create_tables, insert_tables, update_tables_ANALYSIS, update_tables_HELP
from 情报联网 import https_get_ip, https_get_domain, is_ip, is_domain, thead_network_detect, get_network_flag
from 安全事件分析 import Ui_Form
class MyMainForm(QMainWindow, Ui_Form):
def __init__(self, parent=None):
super(MyMainForm, self).__init__(parent)
self.setupUi(self)
self.data_frame = []
self.malicious = -999
self.daily_occurrence = 0
self.delta_to_now = 0
self.affected_in_ip_num = 0
self.threat_scoring = 0
self.credibility = 0
self.send_msg_str = ''
self.openFileButton.clicked.connect(self.openFile)
self.pushButton_clear.clicked.connect(self.clearInput)
self.pushButton_analysis.clicked.connect(self.eventAnalysis)
self.pushButton_help.clicked.connect(self.get_help)
self.QComboBox_ruleName.currentIndexChanged[int].connect(self.rule_changed)
self.QComboBox_eventName.currentIndexChanged[int].connect(self.event_change)
self.QComboBox_evnetMsg.currentIndexChanged[int].connect(self.msg_change)
self.QComboBox_focus.currentIndexChanged[int].connect(self.focus_change)
self.QComboBox_srcIP.currentIndexChanged[int].connect(self.srcip_change)
self.QComboBox_destIP.currentIndexChanged[int].connect(self.destip_change)
self.QComboBox_lastTime.currentIndexChanged[int].connect(self.last_change)
def rule_changed(self, rule_idx):
if len(self.data_frame) > 0:
self.QComboBox_eventName.clear()
df = self.data_frame[self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()]
events = set(df['事件名称'].tolist())
for event in events:
self.QComboBox_eventName.addItem(event)
def event_change(self, event_idx):
if len(self.data_frame) > 0:
self.QComboBox_evnetMsg.clear()
df = self.data_frame[(self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()) & (self.data_frame['事件名称'] == self.QComboBox_eventName.currentText())]
event_msgs = set(df['事件描述'].tolist())
for event_msg in event_msgs:
self.QComboBox_evnetMsg.addItem(event_msg)
def msg_change(self, msg_idx):
if len(self.data_frame) > 0:
self.QComboBox_focus.clear()
df = self.data_frame[(self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()) & (self.data_frame['事件名称'] == self.QComboBox_eventName.currentText()) & (self.data_frame['事件描述'] == self.QComboBox_evnetMsg.currentText())]
focus = set(df['关注点'].tolist())
for focu in focus:
self.QComboBox_focus.addItem(focu)
def focus_change(self, focus_idx):
if len(self.data_frame) > 0:
self.QComboBox_srcIP.clear()
df = self.data_frame[(self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()) & (self.data_frame['事件名称'] == self.QComboBox_eventName.currentText()) & (self.data_frame['事件描述'] == self.QComboBox_evnetMsg.currentText()) & (self.data_frame['关注点'] == self.QComboBox_focus.currentText())]
src_ips = set(df['源IP'].tolist())
for src_ip in src_ips:
self.QComboBox_srcIP.addItem(src_ip)
def srcip_change(self, srcip_idx):
if len(self.data_frame) > 0:
self.QComboBox_destIP.clear()
df = self.data_frame[(self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()) & (self.data_frame['事件名称'] == self.QComboBox_eventName.currentText()) & (self.data_frame['关注点'] == self.QComboBox_focus.currentText()) & (self.data_frame['源IP'] == self.QComboBox_srcIP.currentText())]
dest_ips = set(df['目的IP'].tolist())
for dest_ip in dest_ips:
self.QComboBox_destIP.addItem(dest_ip)
def destip_change(self, destip_idx):
if len(self.data_frame) > 0:
self.QComboBox_lastTime.clear()
df = self.data_frame[(self.data_frame['规则名称'] == self.QComboBox_ruleName.currentText()) & (self.data_frame['事件名称'] == self.QComboBox_eventName.currentText()) & (self.data_frame['关注点'] == self.QComboBox_focus.currentText()) & (self.data_frame['源IP'] == self.QComboBox_srcIP.currentText()) & (self.data_frame['目的IP'] == self.QComboBox_destIP.currentText())]
last_times = set(df['最近发生时间'].tolist())
for last_time in last_times:
if isinstance(last_time, datetime):
self.QComboBox_lastTime.addItem(last_time.strftime('%Y-%m-%d %H:%M:%S'))
elif isinstance(last_time, str):
self.QComboBox_lastTime.addItem(last_time)
else:
self.textBrowser_.setText('【最近发生时间】字段里面存在非时间类型的字符')
break
def last_change(self, last_idx):
pass
def base_analysis(self, df):
self.textBrowser_.insertPlainText('------------基本分析：-----------\n')
ruleName = df.loc[(0, '规则名称')]
eventMsg = df.loc[(0, '事件描述')]
credibility = df.loc[(0, '确信度')]
ioc = df.loc[(0, '情报IOC')]
self.credibility = get_credibility(credibility)
info = get_rule_info(ruleName)
self.textBrowser_.insertPlainText(info + '\n')
endtime = df.loc[(0, '最近发生时间')]
startime = df.loc[(0, '首次发生时间')]
eventnums = df.loc[(0, '聚合次数')]
end = datetime.strptime(str(endtime), '%Y-%m-%d %H:%M:%S')
start = datetime.strptime(str(startime), '%Y-%m-%d %H:%M:%S')
intervalday = (end - start).days + 1
today = datetime.now()
self.daily_occurrence = round(eventnums / intervalday, 2)
self.delta_to_now = (today - end).days
df_src = self.data_frame[(self.data_frame['关注点'] == '源') & (self.data_frame['事件描述'] == eventMsg)]
df_dest = self.data_frame[(self.data_frame['关注点'] == '目的') & (self.data_frame['事件描述'] == eventMsg)]
src_ips = [i[0] for i in list(df_src.groupby('源IP'))]
dest_ips = [i[0] for i in list(df_dest.groupby('目的IP'))]
cross_ips = set(src_ips) | set(dest_ips)
self.affected_in_ip_num = len(cross_ips)
try:
if ruleName in ('恶意主机外联', '恶意域名事件'):
print('测试联网：{}'.format(get_network_flag()))
if get_network_flag():
print('联网成功，正在检测IOC...')
if is_ip(ioc):
self.malicious = https_get_ip(ioc, 1)
else:
if is_domain(ioc):
self.malicious = https_get_domain(ioc, 1)
print(self.malicious)
except:
pass
def false_positives_analysis(self, df):
self.textBrowser_.insertPlainText('\n------------误报分析：------------\n')
if self.credibility < 0:
self.textBrowser_.insertPlainText('事件本身是低可疑的，误报可能性高，可信度扣0.5分\n')
self.threat_scoring -= 0.5
if self.delta_to_now >= 7:
self.textBrowser_.insertPlainText('一周内从未发生过，误报可能性高，可信度扣0.5分\n')
self.threat_scoring -= 0.5
if self.daily_occurrence <= 1:
self.textBrowser_.insertPlainText('平均日发生次数：' + str(self.daily_occurrence) + ' 疑似误报，可信度扣1分\n')
self.threat_scoring -= 1
if self.affected_in_ip_num > 99:
self.textBrowser_.insertPlainText('事件影响主机数：' + str(self.affected_in_ip_num) + ' 疑似误报，可信度扣1分\n')
self.threat_scoring -= 1
else:
if self.affected_in_ip_num > 49:
self.textBrowser_.insertPlainText('事件影响主机数：' + str(self.affected_in_ip_num) + ' 疑似误报，可信度扣0.5分\n')
self.threat_scoring -= 0.5
if self.malicious < 0:
if self.malicious > -999:
self.textBrowser_.insertPlainText('VT情报命中为正常：' + str(self.malicious) + ' 疑似误报，可信度扣0.5分\n')
self.threat_scoring -= 0.5
if self.malicious == 0:
self.textBrowser_.insertPlainText('VT情报命中为正常：' + str(self.malicious) + ' 疑似误报，可信度扣0.2分\n')
self.threat_scoring -= 0.2
def poisoning_analysis(self, df):
self.textBrowser_.insertPlainText('\n------------确认分析：------------\n')
if self.credibility == 0.5:
self.textBrowser_.insertPlainText('事件本身是高可疑的，基本可信，可信度加0.5分\n')
self.threat_scoring += 0.5
else:
if self.credibility == 1:
self.textBrowser_.insertPlainText('事件本身是已失陷的，基本可信，可信度加1分\n')
self.threat_scoring += 1
if self.delta_to_now < 3:
self.textBrowser_.insertPlainText('3天内发生过，基本可信，可信度加0.5分\n')
self.threat_scoring += 0.5
else:
if self.delta_to_now < 7:
self.textBrowser_.insertPlainText('7天内发生过，但是3天内没再发生，基本可信，可信度加0.2分\n')
self.threat_scoring += 0.2
if self.daily_occurrence >= 3:
self.textBrowser_.insertPlainText('平均日发生次数：' + str(self.daily_occurrence) + ' 基本可信，可信度加1分\n')
self.threat_scoring += 1
else:
if (self.daily_occurrence > 1) & (self.daily_occurrence < 3):
self.textBrowser_.insertPlainText('平均日发生次数：' + str(self.daily_occurrence) + ' 基本可信，可信度加0.5分\n')
self.threat_scoring += 0.5
if self.affected_in_ip_num <= 49:
self.textBrowser_.insertPlainText('事件影响主机数：' + str(self.affected_in_ip_num) + ' 基本可信，可信度加0.5分\n')
self.threat_scoring += 0.5
if self.malicious > 0:
self.textBrowser_.insertPlainText('VT情报命中为恶意：' + str(self.malicious) + ' 基本可信，可信度加0.5分\n')
self.threat_scoring += 0.5
def conclusion_analysis(self, df):
self.textBrowser_.insertPlainText('\n------------结论：------------\n')
self.textBrowser_.insertPlainText('综合打分：' + str(self.threat_scoring) + '\n')
if self.threat_scoring >= 1:
self.textBrowser_.insertPlainText('事件基本可信')
else:
if self.threat_scoring >= 0:
self.textBrowser_.insertPlainText('事件可信度不高，但好像不是误报，需要再看看')
else:
self.textBrowser_.insertPlainText('事件好像是误报')
self.threat_scoring = 0
def disposal_advice(self):
pass
def eventAnalysis(self):
ruleName = self.QComboBox_ruleName.currentText()
eventName = self.QComboBox_eventName.currentText()
eventMsg = self.QComboBox_evnetMsg.currentText()
focus = self.QComboBox_focus.currentText()
srcIP = self.QComboBox_srcIP.currentText()
destIP = self.QComboBox_destIP.currentText()
lastTime = self.QComboBox_lastTime.currentText()
self.textBrowser_.clear()
self.textBrowser_.insertPlainText('------------您输入的安全事件基本信息：------------\n规则名称：' + ruleName + '\n事件名称：' + eventName + '\n事件描述：' + eventMsg + '\n关注点：' + focus + '\n源IP：' + srcIP + '\n目的IP：' + destIP + '\n最近发生时间：' + lastTime + '\n\n')
if len(self.data_frame) >= 1:
df = self.data_frame.loc[(self.data_frame['规则名称'] == ruleName) & (self.data_frame['事件名称'] == eventName) & (self.data_frame['事件描述'] == eventMsg) & (self.data_frame['关注点'] == focus) & (self.data_frame['源IP'] == srcIP) & (self.data_frame['目的IP'] == destIP)]
if len(df) > 1:
if len(lastTime) > 1:
df = self.data_frame.loc[(self.data_frame['规则名称'] == ruleName) & (self.data_frame['事件名称'] == eventName) & (self.data_frame['事件描述'] == eventMsg) & (self.data_frame['关注点'] == focus) & (self.data_frame['源IP'] == srcIP) & (self.data_frame['目的IP'] == destIP) & (self.data_frame['最近发生时间'] == lastTime)]
else:
self.textBrowser_.setText('请输入最近发生时间，确保选中唯一事件')
df = df.reset_index(drop=True)
if len(df) == 1:
self.base_analysis(df)
self.false_positives_analysis(df)
self.poisoning_analysis(df)
self.conclusion_analysis(df)
try:
update_tables_ANALYSIS()
except:
pass
else:
if len(df) == 0:
self.textBrowser_.insertPlainText('输入错误：未找到安全事件\n')
elif len(df) > 1:
self.textBrowser_.insertPlainText('输入告警：存在重复安全事件\n')
self.base_analysis(df.ix[0])
self.false_positives_analysis(df.ix[0])
self.poisoning_analysis(df.ix[0])
self.conclusion_analysis(df.ix[0])
try:
update_tables_ANALYSIS()
except:
pass
else:
self.textBrowser_.setText('先按照要求导入安全事件\n')
def get_time_to_stamp(self, x):
return datetime.timestamp(datetime.strptime(str(x), '%Y-%m-%d %H:%M:%S'))
def event_collect(self, df):
g_df = df.groupby(["'事件描述'", "'事件名称'", "'规则名称'", "'关注点'", "'确信度'", "'攻击阶段'"])
g_df = g_df['聚合次数'].sum().reset_index(name='聚合总次数')
c_df = DataFrame(g_df)
c_df.sort_values(by=['聚合总次数'], ascending=False, inplace=True)
c_df.to_csv('事件详情.csv', index=False, header=True)
df_temp = df.copy(deep=True)
df_temp['最近发生时间'] = df_temp['最近发生时间'].apply((lambda x: self.get_time_to_stamp(x)))
max_stamp = df_temp['最近发生时间'].max()
recent_7day = max_stamp - 604800
df_event_7 = df_temp[((df_temp['规则名称'] == '恶意主机外联') | (df_temp['规则名称'] == '恶意域名事件')) & (df_temp['最近发生时间'] > recent_7day)]
df_event_7.to_csv('情报事件.csv', index=False, header=True)
def openFile(self):
thead_network_detect(has_proxy=0)
self.textBrowser_.clear()
get_filename_path, ok = QFileDialog.getOpenFileName(self, '选取单个文件', 'C:/', 'All Files (*);;Text Files (*.txt)')
if ok:
self.filePathlineEdit.setText(str(get_filename_path))
if 'xls' in get_filename_path:
self.data_frame = read_excel(get_filename_path)
self.data_frame = self.data_frame.fillna('')
self.textBrowser_.insertPlainText('导入数据成功：安全事件为 ' + get_filename_path + '\n表格中一共有' + str(len(self.data_frame)) + '条安全事件\n')
core_field = [
"'事件描述'", "'事件名称'", "'规则名称'",
"'确信度'", "'攻击阶段'", "'关注点'", "'源IP'", "'目的IP'",
"'聚合次数'", "'情报IOC'", "'首次发生时间'", "'最近发生时间'",
"'处理状态'"]
miss_field = list(set(core_field).difference(set(self.data_frame.columns.values)))
if len(miss_field) > 0:
self.textBrowser_.insertPlainText('【错误】安全事件缺少关键字段：【{}】，请在态势感知上添加列定制后重新下载，重新导入'.format(' 】【'.join(miss_field)))
else:
self.QComboBox_ruleName.clear()
df_event_status_ed = self.data_frame[self.data_frame['处理状态'] == '已处理']
df_event_status_ing = self.data_frame[self.data_frame['处理状态'] == '处理中']
df_event_status_ignore = self.data_frame[self.data_frame['处理状态'] == '忽略']
df_event_status_ed_fall_2 = self.data_frame[(self.data_frame['确信度'] == '已失陷') & (self.data_frame['处理状态'] == '已处理')]
self.textBrowser_.insertPlainText('----------总的处理现状：----------\n已处理事件数：{}\t 忽略事件数：{}\t 处理中事件数：{}\t 已处理&已失陷事件数：{}\n'.format(len(df_event_status_ed), len(df_event_status_ignore), len(df_event_status_ing), len(df_event_status_ed_fall_2)))
df_event_fall_2 = self.data_frame[(self.data_frame['确信度'] == '已失陷') & (self.data_frame['处理状态'] == '未处理')]
df_event_fall_1 = self.data_frame[(self.data_frame['确信度'] == '高可疑') & (self.data_frame['处理状态'] == '未处理')]
df_event_fall_0 = self.data_frame[(self.data_frame['确信度'] == '低可疑') & (self.data_frame['处理状态'] == '未处理')]
self.textBrowser_.insertPlainText('总的残余风险：\n未处置事件数：{}\t已失陷事件数：{}\t高可疑事件数：{}\t低可疑事件数：{}\n'.format(len(df_event_fall_2) + len(df_event_fall_1) + len(df_event_fall_0), len(df_event_fall_2), len(df_event_fall_1), len(df_event_fall_0)))
df_temp = self.data_frame.copy(deep=True)
df_temp['最近发生时间'] = df_temp['最近发生时间'].apply((lambda x: self.get_time_to_stamp(x)))
max_stamp = df_temp['最近发生时间'].max()
recent_7day = max_stamp - 604800
df_event_status_ed_7 = df_temp[(df_temp['处理状态'] == '已处理') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
df_event_status_ing_7 = df_temp[(df_temp['处理状态'] == '处理中') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
df_event_status_ignore_7 = df_temp[(df_temp['处理状态'] == '忽略') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
df_event_status_ed_fall_2_7 = df_temp[(df_temp['确信度'] == '已失陷') & (df_temp['处理状态'] == '已处理') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
self.textBrowser_.insertPlainText('----------最近7天处理现状：----------\n已处理事件数：{}\t 忽略事件数：{}\t 处理中事件数：{}\t 已处理&已失陷事件数：{}\n'.format(len(df_event_status_ed_7), len(df_event_status_ignore_7), len(df_event_status_ing_7), len(df_event_status_ed_fall_2_7)))
df_event_fall_2_7 = df_temp[(df_temp['确信度'] == '已失陷') & (df_temp['处理状态'] == '未处理') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
df_event_fall_1_7 = df_temp[(df_temp['确信度'] == '高可疑') & (df_temp['处理状态'] == '未处理') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
df_event_fall_0_7 = df_temp[(df_temp['确信度'] == '低可疑') & (df_temp['处理状态'] == '未处理') & (df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
self.textBrowser_.insertPlainText('最近7天残余风险：\n未处置事件数：{}\t已失陷事件数：{}\t高可疑事件数：{}\t低可疑事件数：{}\n'.format(len(df_event_fall_2_7) + len(df_event_fall_1_7) + len(df_event_fall_0_7), len(df_event_fall_2_7), len(df_event_fall_1_7), len(df_event_fall_0_7)))
recall_level_df = df_temp[(df_temp['最近发生时间'] <= max_stamp) & (df_temp['最近发生时间'] > recent_7day)]
recall_level = len(set(recall_level_df['事件名称'].tolist()))
precision = len(df_event_status_ed_fall_2_7)
false_alarm = len(df_event_status_ignore_7)
residual_risks = len(df_event_fall_2_7) * 10 + len(df_event_fall_1_7) + len(df_event_fall_0_7) * 0.2
del df_temp
self.send_msg_str = self.textBrowser_.toPlainText()
try:
create_tables(self.textBrowser_)
data = str(['recall_level', 'precision', 'false_alarm', 'residual_risks',
'1993'])
check_code = hashlib.md5(data.encode(encoding='UTF-8')).hexdigest()
insert_tables(get_filename_path, check_code, recall_level, precision, false_alarm, residual_risks)
except:
print('db采集出bug了')
self.textBrowser_.insertPlainText('db采集出bug了\n')
try:
self.event_collect(self.data_frame)
except:
print('采集事件有bug')
self.textBrowser_.insertPlainText('采集事件有bug')
rules = set(self.data_frame['规则名称'].tolist())
for rule in rules:
self.QComboBox_ruleName.addItem(rule)
else:
self.textBrowser_.setText('导入数据错误：请选择安全事件（excel文件）\n')
else:
self.textBrowser_.setText('导入数据错误：你单击的是文件夹，要选择excel文件\n')
def clearInput(self):
self.QComboBox_eventName.clear()
self.QComboBox_evnetMsg.clear()
self.QComboBox_focus.clear()
self.QComboBox_srcIP.clear()
self.QComboBox_destIP.clear()
self.QComboBox_lastTime.clear()
self.textBrowser_.clear()
def get_help(self):
self.textBrowser_.clear()
self.textBrowser_.setText('小工具说明：\n1.本工具是没有经过专业测试的v3.0\n2.帮助解决安全事件分析、处置相关的常见问题，辅助一线快速分析\n3.使用过程中可能由于安全事件数据字段内容缺失，软件会自动退出\n4.如果遇到软件退出的问题，请将安全事件发给我，并提供自己的输入信息...\n')
try:
update_tables_HELP()
except:
pass
if __name__ == '__main__':
app = QApplication(sys.argv)
myWin = MyMainForm()
myWin.show()
sys.exit(app.exec_())
# okay decompiling 安全事件分析main.pyc参考
- ^pyinstxtractor https://github.com/extremecoders-re/pyinstxtractor
发布于 2023-01-13 21:39IP 属地江苏