Python—网络抓包与解包（pcap、dpkt）

pcap安装

[root@localhost ~]# pip install pypcap

抓包与解包

# -*- coding:utf-8 -*-

import pcap, dpkt

import re, threading, requests

__black_ip = ['103.224.249.123', '203.66.1.212']

# 抓包：param1 eth_name 网卡名，如：eth0,eth3。 param2 p_type 日志捕获类型 1：sdk日志用例分析 2：目标域名过滤输出 3：原始数据包

def catch_pack(eth_name="enp5s0", packet_type=None):

    sniffer = pcap.pcap(eth_name)

    sniffer.setfilter("tcp")            # 只抓取TCP包

    # sniffer.setfilter('tcp port 80')  # 设置监听过滤器

    if sniffer:

        for packet_time, packet_data in sniffer:  # packet_time为收到的时间，packet_data为收到的数据

            th = threading.Thread(target=check_pack, args=(packet_time, packet_data, packet_type))

            th.setDaemon(True)

            th.start()

# 解包

def check_pack(packet_time, packet_data, packet_type):

    packet = dpkt.ethernet.Ethernet(packet_data)

    # 判断是否HTTP协议javascript:;

    try:

        m_tips = packet.data.data.data[:3]

    except:

        return False

    if m_tips != 'GET' and m_tips != 'POS': return False

    # 取回头信息

    tmp = get_header(packet, packet_time)

    if not tmp: return False

    print "tmp===>", tmp

    # self.input_database(tmp)

# 取header数据

def get_header(packet, packet_time):

    if not packet: return False

    tmp = {}

    # 获取基础头信息

    tmp['dst'] = "%d.%d.%d.%d" % tuple(map(ord, list(packet.data.dst)))

    if tmp['dst'] in __black_ip: return False

    tmp['src'] = "%d.%d.%d.%d" % tuple(map(ord, list(packet.data.src)))

    tmp['time'] = packet_time

    tmp['dport'] = packet.data.data.dport

    tmp['sport'] = packet.data.data.sport

    # if not self.ip_exists(tmp['dst']): self.input_ip(tmp['dst'])

    # 截取前1024个字符

    t = packet.data.data.data[:1024]

    # 获取host,uri,referer,method

    o = re.search('(GET|POST)\s+(.+)\s+HTTP(.|\n)+Host:\s+(.+)\r(.|\n)+Referer:\s+(.+)\r', t)

    if not o: return False

    o1 = o.groups()

    if len(o1) < 6: return False

    tmp['method'] = o1[0]

    tmp['host'] = o1[3]

    tmp['uri'] = o1[1]

    tmp['referer'] = o1[5]

    # if self.is_exists(tmp['host'],tmp['dport']): return False

    return tmp

if __name__ == "__main__":

    catch_pack("eth0", 1)

解析pcap文件

try:

    import scapy.all as scapy

except ImportError:

    import scapy

from scapy.utils import PcapReader

packets = scapy.rdpcap('test.pcap')

for data in packets:

    if 'UDP' in data:

        s = repr(data)

        print(data['UDP'].sport)

        break

https://www.cnblogs.com/guanfuchang/p/6495533.html

https://www.cnblogs.com/KevinGeorge/p/8088915.html