目录
- 贝叶斯恶意流量检测可视化
- 1.界面视图
- 2.源码分类
(1)主界面
（2）过滤抓包
（3）流量检测
以上检测会根据指定的网卡名称和过滤规则进行流量嗅探。
2.源码分类
（1）界面代码
class mainwindow(Ui_MainWindow,QMainWindow):
data_all=''
data_all_f = ''
def __init__(self):
super().__init__()
self.setupUi(self)
# 实例化状态栏
self.statusBar = QStatusBar()
# 设置状态栏，类似布局设置
self.setStatusBar(self.statusBar)
self.pushButton.clicked.connect(self.flow_detect)
self.pushButton_2.clicked.connect(self.wireshark)
def wireshark(self):
try:
if self.lineEdit.text() == '' or self.lineEdit_2.text()=='':
self.status_msg("不能为空!",5000)
else:
port_name = self.lineEdit.text()
port_filter = self.lineEdit_2.text()
sniff(iface=port_name,filter=port_filter, prn=lambda x: self.show_wireshark(x))
except Exception as e:
print(e)
self.status_msg(str(e),5000)
def show_wireshark(self,data):
pkt_str = data.summary()+'n'
print(pkt_str)
# pkt_str = str(raw(data).decode(encoding='ISO-8859-1'))
# print(pkt_str)
self.data_all += str(pkt_str)
# print(self.data_all)
try:
self.textEdit.setPlainText(str(self.data_all))
except Exception as e:
print(e)
self.status_msg(str(e), 5000)
cursor = self.textEdit.textCursor()
cursor.movePosition(QTextCursor.End) # 还可以有别的位置
self.textEdit.setTextCursor(cursor)
QApplication.processEvents()
def show_flow(self,data):
try:
if data=='':
pass
else:
self.data_all_f += (str(data)+'n')
# print(self.data_all_f)
self.textEdit.setPlainText(str(self.data_all_f))
cursor = self.textEdit.textCursor()
cursor.movePosition(QTextCursor.End) # 还可以有别的位置
self.textEdit.setTextCursor(cursor)
QApplication.processEvents()
except Exception as e:
print(e)
self.status_msg(str(e), 5000)
def flow_detect(self):
try:
if self.lineEdit.text() == '' or self.lineEdit_2.text()=='':
self.status_msg("不能为空!",5000)
else:
port_name = self.lineEdit.text()
port_filter = self.lineEdit_2.text()
ListenTraffic().run(port_name=port_name, port_filter=port_filter)
except Exception as e:
print(e)
self.status_msg(str(e), 5000)
# 状态栏消息
def status_msg(self, msg, time):
# 设置状态栏的显示文本以及显示时间
self.statusBar.showMessage(str(msg), int(time))
QApplication.processEvents()
（2）检测算法
class ListenTraffic():
def __init__(self): # 检测到通过第一个数据包的时候加载算法
self.model = shell_detect()
if isinstance(self.model,shell_detect): #machine_learning_model.shell_detect
print(u"贝叶斯算法加载完成，引擎启动")
window.show_flow('贝叶斯算法加载完成，引擎启动')
else:
print(u"贝叶斯算法加载失败。")
window.show_flow('贝叶斯算法加载失败')
def __result__(self, packet): # POST 和GET 请求分别处理
self.http_request = packet
try:
self.Method = self.http_request["HTTPRequest"].Method
print(self.Method)
if self.Method in b"POST":
http_request = unquote(str(self.http_request["Raw"].load), 'utf-8')
http_ip_src = unquote(str(self.http_request["IP"].src), 'utf-8')
http_ip_dst = unquote(str(self.http_request["IP"].dst), 'utf-8')
ip_list = [http_ip_src,http_ip_dst]
self.model.try_classify(unquote(http_request, 'utf-8'),ip_list)
elif self.Method in b"GET":
http_request = self.http_request["HTTPRequest"].Path
http_request = unquote(str(http_request), 'utf-8')
http_ip_src = unquote(str(self.http_request["IP"].src), 'utf-8')
http_ip_dst = unquote(str(self.http_request["IP"].dst), 'utf-8')
ip_list = [http_ip_src, http_ip_dst]
self.model.try_classify(unquote(http_request, 'utf-8'), ip_list)
except Exception as e:
print(str(e))
window.status_msg(str(e),2000)
# print(packet.show())
QApplication.processEvents()
def __flow_separa__(self, packet): # 区分POST 和GET
# print(packet.show())
if "POST" in str(packet):
return packet
# return packet
elif "GET" in str(packet):
return packet
# return packet
else:
pass
def run(self, port_name, port_filter): # 运行监听
sniff(
iface=port_name,
prn=self.__result__,
lfilter=lambda p: self.__flow_separa__(p),
filter=port_filter, # 设置端口为 tcp port 80 和8080
)
（3）训练算法
class shell_detect:
@staticmethod
def read_file(file_path):
file = open(file_path)
data = file.read()
file.close()
return data
@staticmethod
def code_word_to_vector(php_code):
filter_flag_list = ['@', '[', ']', '(', ')', '{', '}', ''', '"', ',', ';', '=', '.', 't', 'n', 'rn']
keyword = ['$_GET', '$_POST', '$_REQUEST', '$_COOKIE']
for filter_flag_index in filter_flag_list:
php_code = php_code.replace(filter_flag_index, ' ')
vector = php_code.split(' ')
for index in range(len(vector)): # filter $ variant
if vector[index].startswith('$') and not vector[index] in keyword:
vector[index] = ''
elif vector[index] in keyword:
vector[index] = '$'
while vector.count(''): # filter empty item ..
vector.remove('')
return vector
@staticmethod
def load_and_train_model(data_set_path='./shell'):
file_list = os.listdir(data_set_path)
shell_sample = {} # classfy set ..
for file_index in file_list:
try:
file_information = file_index.split('-')
classfy_type = file_information[0] + '-' + file_information[1] + '-' + file_information[2]
php_code_vector = shell_detect.code_word_to_vector(
shell_detect.read_file(data_set_path + '\' + file_index))
if classfy_type not in shell_sample:
shell_sample[classfy_type] = []
shell_sample[classfy_type].append(php_code_vector)
except Exception as e:
print(e)
return shell_sample
def __init__(self, data_set_path='./shell'):
self.shell_sample = shell_detect.load_and_train_model(data_set_path)
def try_classify(self, php_code, ip_list):
php_code_vector = shell_detect.code_word_to_vector(php_code.replace("n", ""))
alpha = 1
p_list = {}
# print 'Debug for try_classify : ' ,php_code_vector
for key_index in self.shell_sample.keys():
max_p_value = 0
for shell_sample_index in self.shell_sample[key_index]:
found_vector_in_shell_sample_count = 0
for php_code_vector_index in php_code_vector:
if php_code_vector_index in shell_sample_index:
found_vector_in_shell_sample_count += shell_sample_index.count(php_code_vector_index)
p_value = (found_vector_in_shell_sample_count + alpha) / float(len(shell_sample_index) * 2 + alpha)
# print shell_sample_index , p_value
if p_value >= max_p_value:
max_p_value = p_value
p_list[key_index] = max_p_value
# print key_index ,p_list[key_index]
max_p_value = 0
max_p_type_name = ''
for p_type_name_index in p_list.keys():
p_value = p_list[p_type_name_index]
if p_value >= max_p_value:
max_p_value = p_value
max_p_type_name = p_type_name_index
# print php_code , max_p_type_name , max_p_value
print('n', "*" * 20, 'HEAD', "*" * 20)
print("HTTP_REQUEST:n", php_code, 'nShell Type :n', max_p_type_name)
print(f"IP_GO:n{ip_list[0]}----->>>>>{ip_list[1]}")
print("*" * 20, 'TAIL', "*" * 20, "n")
data = 'n'+"*" * 20+'HEAD'+"*" * 20+'n'+"HTTP_REQUEST:n"+str(php_code)+'nShell Type :n'+str(max_p_type_name)+'n'
data+=f"IP_GO:n{ip_list[0]}----->>>>>{ip_list[1]}n"
data+=("*" * 20+'TAIL'+"*" * 20+"n")
window.show_flow(data)
QApplication.processEvents()
return max_p_type_name
本文只作为学习使用
传送门
