总所周知，对于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 决定
# 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():
return ctypes.windll.shell32.IsUserAnAdmin()
except:
return False
def unpack_ip_header(buf):
if len(buf) < IP_HEAD_LEN:
return
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
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
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():
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.connect(('8.8.8.8', 80))
ip = sock.getsockname()[0]
except Exception:
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地址
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()
