用Python实现简易网络抓包工具

现代计算机程序大部分时候离不开网络，作为开发者，在日常开发网络相关的程序或者排查程序错误时经常会用抓包工具来分析网卡收发的数据。除了著名的tcpdump和Wireshark外，Sniff Master也是一款优秀的抓包分析工具。今天我们尝试用100行左右的Python代码在Linux系统上实现一个简单的抓包工具。

1. 端序

试想一个16进制数0x12345678，大端序在内存中的储存顺序如下，高字节保存在内存的低地址，高字节在前：

内存地址    0    1    2    3
          ---------------------
字节数据  | 12 | 34 | 56 | 78 |
          ---------------------

而小端序高字节保存在内存的高地址，低字节在前：

内存地址    0    1    2    3
          ---------------------
字节数据  | 78 | 56 | 34 | 12 |
          ---------------------

大端序还是小端序是由硬件决定的，通常CPU和内存中的数据都是小端序，而网络传输都用大端序，因为大端序字节流解析起来更方便，但有些特殊的设备是例外。

Python中socket模块的htons，ntohs，htonl，ntohl函数用来处理网络字节顺序与本地字节顺序之间的转换。函数名中h表示host，n表示net，s表示short int（2字节），l表示long int（4字节），比如htons是将2字节长的整数从本地端序转换为网络端序。比如：

import socket

i = 0x1234  # 本地小端序
net_i = socket.htons(i)  # 网络大端序
print(hex(i), hex(net_i))
# 输出 0x1234 0x3412

2. socket编程

socket是操作系统用户与TCP/IP协议族通信的接口，创建一个socket需要3个参数，地址簇，socket类型，协议。比如可以用以下代码创建一个常用的TCP套接字，用来收发TCP数据流：

import socket

# AF_INET: internet协议簇
# SOCK_STREAM: stream类型的socket
# 0: 使用该socket类型的默认协议，即TCP协议
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

也可以用以下代码创建一个UDP套接字，用来收发UDP数据包：

# SOCK_STREAM: data gram类型的socket
# 0: 使用该socket类型的默认协议，即UDP协议
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)

但是我们今天要用的不是这类套接字，而是更底层的原始套接字raw socket，这类套接字可以操作更底的协议比如IP协议和以太网协议，创建raw socket需要root权限。可以用以下代码创建一个接收以太网数据包的原始套接字：

ETH_P_ALL = 0x3

ETH_P_IP = 0x0800
ETH_P_ARP = 0x0806
ETH_P_RARP = 0x8035
ETH_P_IPV6 = 0x086dd

raw_sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))

其中ETH_P_ALL表示监听所有协议的以太网数据包，如果改用ETH_P_IP则只会监听IP协议的数据包。（由于Windows系统上没有定义AF_PACKET，所以这段代码无法在Windows系统上运行）

然后可以开始监听：

while True:
    packet, packet_info = raw_sock.recvfrom(1500)
    print(packet)

recvfrom函数返回2个值，第一个是以太网数据包，第二个是该数据包相关的信息。

3. 以太网帧格式

以太网帧根据不同的标准有多种格式，常见的有Ethernet II格式和Ethernet 802.3格式。

Ethernet II帧格式

• DMAC：6字节，目标MAC地址
• SMAC：6字节，源MAC地址
• Type：2字节，上层协议类型
• Data：46～1500字节可变长度，载荷数据
• FCS：最后4字节，帧校验

Ethernet 802.3帧格式

• DMAC：6字节，目标MAC地址
• SMAC：6字节，源MAC地址
• Length：2字节，Data字段包含的字节数
• LLC：3字节，逻辑链路控制
• SNAP：5字节，Sub-network Access Protocol
• Data：38～1492字节可变长度，载荷数据
• FCS：最后4字节，帧校验

可以看到Ethernet II和Ethernet802.3格式的前12字节结构相同，由于Ethernet 802.3的头部多了8字节，所以载荷数据长度最大只能为1492字节。另外需要注意，原始套接字读取到的包最后不包含帧校验。

4. 解析以太网头部

在我们了解了以太网帧格式后就可以开始解析接收到的二进制数据了，我们可以用以下代码把MAC地址转换成整数：

BIG_ENDIAN = 'big'
dst_mac = int.from_bytes(packet[:6], BIG_ENDIAN)
src_mac = int.from_bytes(packet[6:12], BIG_ENDIAN)

从第12、13字节用来区分接收到的是Ethernet II帧还是Ethernet 802.3帧，如果该数字小于等于1500（0x05DC）可以判断接收到的是Ethernet 802.3帧，如果该数字大于等于1536（0x0600）则接收到的是Ethernet II帧。其他字段按照帧格式依次解析即可。

完整代码如下：

import socket

BIG_ENDIAN = 'big'
LITTLE_ENDIAN = 'little'

ETH_P_ALL = 0x3

ETH_P_IP = 0x0800
ETH_P_ARP = 0x0806
ETH_P_RARP = 0x8035
ETH_P_IPV6 = 0x086dd

ETH_TYPE_MAP = {
    ETH_P_IP: 'IP',
    ETH_P_ARP: 'ARP',
    ETH_P_RARP: 'RARP',
    ETH_P_IPV6: 'IPv6'
}


class MACAddress:
    def __init__(self, addr: int):
        self.addr = addr

    def __str__(self):
        return ':'.join('{:02x}'.format(a) for a in self.addr.to_bytes(6, BIG_ENDIAN))

    def __repr__(self):
        return self.__str__()


class EthernetHeader:

    def __init__(self, dst_mac, src_mac):
        self.dst_mac = dst_mac
        self.src_mac = src_mac

    def describe(self):
        return {
            'src_mac': MACAddress(self.src_mac),
            'dst_mac': MACAddress(self.dst_mac)
        }


class EthernetIIHeader(EthernetHeader):

    def __init__(self, dst_mac, src_mac):
        super().__init__(dst_mac, src_mac)
        self.eth_type = 0

    def describe(self):
        dct = super().describe()
        dct['eth_type'] = self._describe_eth_type(self.eth_type)
        return dct

    @staticmethod
    def _describe_eth_type(eth_type):
        if eth_type in ETH_TYPE_MAP:
            return ETH_TYPE_MAP[eth_type]
        return 'Unknown protocol {}'.format(eth_type)


class Ethernet802_3Header(EthernetHeader):

    def __init__(self, dst_mac, src_mac):
        super().__init__(dst_mac, src_mac)
        self.length = 0
        self.llc = 0
        self.snap = 0

    def describe(self):
        dct = super().describe()
        dct['length'] = self.length
        dct['llc'] = self.llc
        dct['snap'] = self.snap
        return dct


def unpack(packet):
    dst_mac = int.from_bytes(packet[:6], BIG_ENDIAN)
    src_mac = int.from_bytes(packet[6:12], BIG_ENDIAN)
    type_or_length = int.from_bytes(packet[12:14], BIG_ENDIAN)
    if type_or_length < 1500:
        hdr = Ethernet802_3Header(dst_mac, src_mac)
        hdr.length = type_or_length
        hdr.llc = int.from_bytes(packet[14:17], BIG_ENDIAN)
        hdr.snap = int.from_bytes(packet[17:22], BIG_ENDIAN)
        return hdr, packet[22:]
    elif type_or_length >= 1536:
        hdr = EthernetIIHeader(dst_mac, src_mac)
        hdr.eth_type = type_or_length
        return hdr, packet[14:]
    else:
        raise ValueError(type_or_length)


def main():
    raw_sock = socket.socket(socket.PF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
    while True:
        try:
            packet, packet_info = raw_sock.recvfrom(1500)
            eth_header, payload = unpack(packet)
            print(eth_header.describe(), payload)
        except KeyboardInterrupt:
            break
        except ValueError as e:
            print('unpack failed', e)


if __name__ == '__main__':
    main()

最后，用root权限运行抓包程序，可以看到屏幕上输出了解析后的Ethernet头部和未解析的载荷数据，如果有兴趣的话可以按照协议标准继续解析载荷数据中的上层协议。对于更复杂的网络分析需求，可以考虑使用专业的抓包工具如Sniff Master，它提供了更全面的协议解析和可视化分析功能。

{'src_mac': 50:fa:84:aa:aa:aa, 'dst_mac': ac:d1:b8:aa:aa:aa, 'eth_type': 'IP'} b'E\x00\x00T\x0c\xfe@\x004\x01\x93*\xb4e1\x0c\xc0\xa8\x00g\x00\x00\xb6\x83-\x8e\x00\x07\xa6\xb4c^\x00\x00\x00\x00G\x01\x0c\x00\x00\x00\x00\x00\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./01234567'
...