python 处理传输层的报文 TCP/UDP

总所周知，对于python而言实现tcp/udp的传输一般靠的是socket这个库，而区分两者的是建立socket的参数

SOCK_STREAM 为TCP连接

SOCK_DGRAM 为UDP连接

而一般情况下接收报文需要遵从某一些协议，这样双方可以通过特定的粘包解包操作处理数据。

很多情况自定义协议都是比较简单，先接收报文头，获取消息长度，再获取消息体。

但是有很多协议写起来很麻烦，于是就用到scapy这个库，就可以每次获取一个报文，一般报文头信息都是类似的

都带有源地址，目标地址，消息长度，校验码

 

这里是一段监听DMS报警系统的代码

 

from scapy.all import *
alarm_map = {"211": "未系安全带", "205": "疲劳驾驶", "206": "疲劳驾驶", "208": "抽烟", "209": "出现异常"}
def scan(target, port):
    has_no_connect = True
    while True:
        # 根据接口进行监听报文
        # 接口获取由 IFACES 决定
        try:
            # sniff开始获取报文，iface是接口，filter可以选择过滤报文类型， count为一次性获取多少个pkt
            pkt = sniff(iface=IFACES.dev_from_index(12), filter="udp", count=1)
            # 每个pkt格式类型于YAML,如果没有IP信息则代表这个包不完整
            if pkt[0][IP].src == "192.168.43.1":
                if has_no_connect:
                    # 展示建立连接的第一个UDP包
                    pkt[0].show()
                if Raw in pkt[0]:
                    has_no_connect = False
                    # load的值为bytes类型
                    body = pkt[0][Raw].load
                    if b'alarm":2' in body:
                        result = json.loads(body.decode())
                        if str(result.get('dms').get('alarm')) in alarm_map.keys():
                            now_time = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
                            message = {
                                "event_name": alarm_map[str(result.get('dms').get('alarm'))],
                                "event_time": now_time,
                            }
                            print(message)
                        else:
                            print(result)
        except Exception as e:
            print(pkt[0])
            continue
 
"""            
 调用show()可以展示数据包，抓到第一个数据包格式如下：
###[ Ethernet ]### 
  dst       = 20:0d:b0:17:cf:d4
  src       = 02:08:22:b2:bb:fb
  type      = IPv4
###[ IP ]### 
     version   = 4
     ihl       = 5
     tos       = 0x0
     len       = 208
     id        = 19582
     flags     = DF
     frag      = 0
     ttl       = 64
     proto     = udp
     chksum    = 0x163a
     src       = 192.168.43.1
     dst       = 192.168.43.19
     options   
###[ UDP ]### 
        sport     = 23456
        dport     = 62472
        len       = 188
        chksum    = 0x202f
###[ Raw ]### 
           load      = '{"code":3,"dms":{"alarm":0,"num":1,"id":0,"eye":0,"p":15.220612,"y":-29.808479,"r":6.351517,"fr":0.493671,"fmi":0,"fmon":0,"fx":0.389583,"fy":0.201852,"fw":0.192708,"fh":0.457407}}'
"""

 

scapy这个库确实能够满足一部分需求，不用头疼于各种协议的实现。上手也比较简单，在一些需要抓TCP包或者UDP的场景下使用效果极佳。

如果知道协议了，这里有段比较不错的监听tcp/udp ip报文的代码。 【需要用管理员权限运行，不要在IDE下运行】

 

from __future__ import print_function
import ctypes, sys
import struct
import socket
import traceback

ip_header_fmt = '!BBHHHBBH4s4s'
tcp_header_fmt = '!HHLLBBHHH'
udp_header_fmt = '!HHHH'
IP_HEAD_LEN = struct.calcsize(ip_header_fmt)  # 20字节
TCP_HEADER_LEN = struct.calcsize(tcp_header_fmt)  # 20字节
UDP_HEAD_LEN = struct.calcsize(udp_header_fmt)  # 8字节
IP_MAX_LEN = 65535

def is_admin():
    try:
        return ctypes.windll.shell32.IsUserAnAdmin()
    except:
        return False


def unpack_ip_header(buf):
    if len(buf) < IP_HEAD_LEN:
        return
    try:
        iph = struct.unpack(ip_header_fmt, buf[:IP_HEAD_LEN])
        protocol_map = {1: 'ICMP', 6: 'TCP', 17: 'UDP'}
        return {
            'version': iph[0] >> 4,  # 高4位
            'ihl': (iph[0] & 0xF) * 4,  # 低4位，每个长度单位表示4字节，最大为60字节
            'tos': iph[1],  # 8位
            'len': iph[2],  # 16位
            'id': iph[3],  # 16位
            'offset': iph[4],  # 16位
            'ttl': iph[5],  # 8位
            'protocol': protocol_map.get(iph[6], str(iph[6])),  # 8位
            'cks': iph[7],  # 16位
            'src': iph[8],  # 32位
            'dst': iph[9],  # 32位
        }
    except Exception as e:
        raise e


def unpack_tcp_header(buf):
    if len(buf) < TCP_HEADER_LEN:
        return
    try:
        tcph = struct.unpack(tcp_header_fmt, buf[:TCP_HEADER_LEN])
        return {
            'src': tcph[0],  # 16位
            'dst': tcph[1],  # 16位
            'seq': tcph[2],  # 32位
            'ack': tcph[3],  # 32位
            'thl': (tcph[4] >> 4) * 4,  # 高4位
            'wlen': tcph[6],  # 16位
        }
    except Exception as e:
        raise e


def unpack_udp_header(buf):
    if len(buf) < UDP_HEAD_LEN:
        return
    try:
        udph = struct.unpack(udp_header_fmt, buf[:UDP_HEAD_LEN])
        return {
            'src': udph[0],  # 16位
            'dst': udph[1],  # 16位
            'dlen': udph[2],  # 16位
            'cks': udph[3],  # 16位
        }
    except Exception as e:
        raise e


def get_host_ip():
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.connect(('8.8.8.8', 80))
        ip = sock.getsockname()[0]
    except Exception:
        pass
    finally:
        sock.close()
    return ip


if __name__ == '__main__':
    if sys.version_info[0] == 3:
        ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, __file__, None, 1)

    host = get_host_ip()  # 获取当前联网网卡的ip地址
    try:
        sock = socket.socket(
            socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
        sock.bind((host, 0))
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
        while True:
            buf = sock.recvfrom(IP_MAX_LEN)[0]
            if len(buf) < IP_HEAD_LEN:
                continue
            ipheader = unpack_ip_header(buf[:IP_HEAD_LEN])
            if ipheader:
                buf = buf[ipheader['ihl']:]
                if ipheader['protocol'] == 'TCP':
                    tcpheader = unpack_tcp_header(buf[:TCP_HEADER_LEN])
                    if tcpheader:
                        data = buf[tcpheader['thl']:(
                            ipheader['len'] - ipheader['ihl'])]
                        if data:
                            print('TCP {}'.format(data))
                elif ipheader['protocol'] == 'UDP':
                    udpheader = unpack_udp_header(buf[:UDP_HEAD_LEN])
                    if udpheader:
                        data = buf[UDP_HEAD_LEN:][:udpheader['dlen']]
                        if data:
                            print('UDP {}'.format(data))
                # 可以增加其他协议的头解析
    except Exception as e:
        traceback.print_exc()
    finally:
        sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
        sock.close()