traffic-database/readpcap.py at master · DeepShield-AI/traffic-database · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
import struct
import socket
import sys

def read_pcap(file_path):
    with open(file_path, 'rb') as f:
        # 读取 pcap 全局头（24字节）
        # global_header = f.read(24)
        # magic_number = struct.unpack('I', global_header[:4])[0]

        # # 判断字节序（小端或大端）
        # if magic_number == 0xa1b2c3d4:
        #     endian = '<'  # little endian
        # elif magic_number == 0xd4c3b2a1:
        #     endian = '>'  # big endian
        # else:
        #     raise ValueError('不是有效的 pcap 文件')

        endian = '<'
        count = 0

        # f.seek(2147479704)
        f.seek(int(sys.argv[2]))

        # 逐个读取数据包
        while True:
            count += 1
            if count > int(sys.argv[3]):
                break
            packet_header = f.read(16)
            if len(packet_header) < 16:
                break  # 文件结束

            ts_sec, ts_usec, incl_len, orig_len = struct.unpack(endian + 'IIII', packet_header)
            # print(orig_len)
            packet_data = f.read(incl_len)

            if len(packet_data) < 20:
                continue  # 非法帧

            version = packet_data[0] >> 4

            if version == 4:
                # IPv4
                if len(packet_data) < 20:
                    continue
                ihl = (packet_data[0] & 0x0F) * 4
                if len(packet_data) < ihl + 4:
                    continue
                protocol = packet_data[9]
                src_ip = socket.inet_ntoa(packet_data[12:16])
                dst_ip = socket.inet_ntoa(packet_data[16:20])

                if protocol in (6, 17):  # TCP or UDP
                    trans_header = packet_data[ihl:ihl+4]
                    if len(trans_header) < 4:
                        continue
                    src_port, dst_port = struct.unpack('!HH', trans_header)
                    print(f"{src_ip}:{src_port} --> {dst_ip}:{dst_port} ({protocol})")


            elif version == 6:
                # IPv6
                if len(packet_data) < 40:
                    continue
                next_header = packet_data[6]
                src_ip = socket.inet_ntop(socket.AF_INET6, packet_data[8:24])
                dst_ip = socket.inet_ntop(socket.AF_INET6, packet_data[24:40])
                protocol = next_header

                if next_header in (6, 17):  # TCP or UDP
                    trans_header = packet_data[40:44]
                    if len(trans_header) < 4:
                        continue
                    src_port, dst_port = struct.unpack('!HH', trans_header)
                    print(f"{src_ip}:{src_port} --> {dst_ip}:{dst_port} ({protocol})")

            else:
                # 其他 IP 协议暂不支持
                print(version)
                continue

if __name__ == "__main__":
    # for id in range(int(sys.argv[1])):
    #     print(f"read pcap file: ./data/input/0-{id}.pcap")
    #     read_pcap(f"./data/input/0-{id}.pcap")
    #     print()
    filename = sys.argv[1]
    print(f"read pcap file: {filename}")
    read_pcap(filename)