网络流量分析

全栈程序员站长

发布于 2022-08-11 19:18:02

1.3K00

代码可运行

文章被收录于专栏：全栈程序员必看全栈程序员必看

运行总次数：0

代码可运行

大家好，又见面了，我是你们的朋友全栈君。

网络流量分析

具体要求

收集自己本机的网络流量数据（至少1小时）并进行数据显示。
可用wireshark软件抓包
网络流量大小的时序图，可按每半分钟、每分钟、每五分钟、每十分钟进行分别显示。
流量协议类型直方图
可设置过滤条件，显示指定协议数据包、显示时间段数据包、显示长度范围内的数据包
提示：由于代码导入pyshark模块，注意wireshark安装路径为C盘programfils 文件夹下，否则无法运行。

具体思路

要想对数据进行分析，首先要有数据，所以第一步要抓取数据
抓取数据我所知道的有两种方法，第一种为通过代码进行抓取，然后保存在文件中进行读取，第二种通过wireshark等软件进行抓取，然后通过代码分析。
前者更倾向于分析实时数据包，后者则耗时间比较少（具体根据需要选择）
拿到数据包以后，在分析之前，我们要通过代码把数据包中的内容拿出来，我选择pyshark.FileCapture方法
作图我选择导入matplotlib模块，作图会方便很多
具体的分析过程是一些简单的选择结构（ps：不懂得可以看一下Python基础篇）

python代码实现

# -*- coding: utf-8 -*-
import pyshark
from scapy.all import *
import matplotlib.pyplot as plt

# 读取pcap文件
packets = pyshark.FileCapture("./net_package.pcap")


def protocal(packets):
    """
    制作流量协议类型直方图
    :param packets: 读取的pcap文件数据
    """
    # 新建空字典
    dict = {}
    for packet in packets:
        if packet.highest_layer not in dict.keys():
            dict[packet.highest_layer] = 1
        else:
            dict[packet.highest_layer] += 1
    # print(dict)
    keys = dict.keys()
    values = dict.values()
    plt.figure(figsize=(8, 20), dpi=80)
    plt.bar(keys, values)
    plt.xticks(rotation=45)
    plt.xlabel('protocal')
    plt.ylabel('amount')
    plt.title('the amounts of all protocals')
    plt.show()


# print(proto_sum)
def graph_size(packets):
    """
    作流量大小时序图
    :param packets: 读取的pcap文件数据
    """
    time_stamps = []
    print("正在统计中。。。")
    for packet in packets:
        # print(int(float(packet.sniff_timestamp)))
        time_stamps.append(int(float(packet.sniff_timestamp)))
    # print(time_stamps)
    print("统计完成！")
    d = int(float(input("请输入时间间隔(单位：分钟):")) * 60)
    # d = 30 #半分钟
    num_bins = (max(time_stamps) - min(time_stamps)) // d
    step = len(time_stamps) // num_bins
    time_labels = [time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(i)) for i in time_stamps[::step]]
    # 新建20*8英寸图形，分辨率为80
    plt.figure(figsize=(20, 8), dpi=80)
    # X轴分布数据以及num_bins条柱状图
    plt.hist(time_stamps, num_bins)
    # 标签旋转角度45
    plt.xticks(range(min(time_stamps), max(time_stamps) + d, d), time_labels, rotation=45)
    # plt.xticks(range(min(time_stamps),max(time_stamps)+d,d),rotation = 45)
    plt.xlabel("timestamp")
    plt.ylabel("amount")
    plt.title("amount of per " + str(d) + " s")
    plt.show()


def filter(packets):
    """
    显示过滤器
    :param packets: 读取的pcap文件数据
    """

    protocal = input("请输入协议类型:")
    begin_time = input("请输入开始时间(Example:2019-09-09 10:58:42):")
    end_time = input("请输入结束时间(Example:2019-09-09 11:40:00):")
    length = int(input("请输入最大长度限制(单位：字节):"))
    # time.strptime把固定格式时间转换为时间元组
    array_begin_time = time.strptime(begin_time, "%Y-%m-%d %H:%M:%S")
    # time.mktime把时间元组转换为以秒表示的时间
    begin_time_stamp = float(time.mktime(array_begin_time))
    # print("begin_time_stamp:"+str(begin_time_stamp))
    array_end_time = time.strptime(end_time, "%Y-%m-%d %H:%M:%S")
    end_time_stamp = float(time.mktime(array_end_time))
    # print("end_time_stamp:"+str(end_time_stamp))
    packlist = []
    for packet in packets:
        # sniff_timestamp获取开始嗅探的时间戳
        time_stamp = float(packet.sniff_timestamp)
        # 获取数据包的捕获长度
        size = float(packet.captured_length)
        if packet.highest_layer == protocal and time_stamp > begin_time_stamp and time_stamp < end_time_stamp and size <= length:
            print(packet)
            packlist.append(packet)
    print("过滤出的数据包个数为 %s" % len(packlist))


# 调用函数进行操作

protocal(packets)
graph_size(packets)
filter(packets)