1200字范文 > 《Python黑帽子：黑客与渗透测试编程之道》读书笔记（二）：原始套接字和流量嗅探

《Python黑帽子：黑客与渗透测试编程之道》读书笔记（二）：原始套接字和流量嗅探

时间：2023-01-01 00:25:07

前言

《Python黑帽子：黑客与渗透测试编程之道》的读书笔记，会包括书中源码，并自己将其中一些改写成Python3版本。书是比较老了，anyway，还是本很好的书

本篇是第3章原始套接字和流量嗅探

1、Windows和Linux上的包嗅探

为了多平台使用，先创建SOCKET，再判断平台

windows允许嗅探所有协议linux只能嗅探ICMP

我们需要开启混杂模式以允许嗅探网卡上所有数据包，这就需要管理员权限

#!/usr/bin/env python#-*- coding:utf8 -*-import socketimport os# 监听主机，即监听那个网络接口，下面的为我的kali的iphost = "10.10.10.145"# 创建原始套接字,然后绑定在公开接口上if os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol) #raw的中文是生的意思,大概就是原始套接字的意思吧sniffer.bind((host, 0)) #这里端口为0,监听所有端口吧~# 设置在捕获的数据包中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在Windows平台上,我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)# 读取单个数据包print sniffer.recvfrom(65565)# 在Windows平台上关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

2、解码IP层

嗅探到数据包后，需要解码IP层，获取协议类型、源IP、目的IP等信息

#!/usr/bin/env python#-*- coding:utf8 -*-import socketimport osimport structfrom ctypes import *# 监听主机，即监听那个网络接口，下面的ip为我的kali的iphost = "10.10.10.145"# ip头定义,参考C语言的结构体class IP(Structure):_fields_ = [("ihl", c_ubyte, 4), #ip head length:头长度("version", c_ubyte, 4), #版本("tos", c_ubyte), #服务类型("len", c_ushort),#ip数据包总长度("id", c_ushort), #标识符("offset",c_ushort),#片偏移("ttl", c_ubyte), #生存时间("protocol_num", c_ubyte), #协议数字,应该是协议类型,这里用数字来代表时哪个协议,下面构造函数有设置映射表("sum", c_ushort),#头部校验和("src", c_ulong), #源ip地址("dst", c_ulong) #目的ip地址]# __new__(cls, *args, **kwargs) 创建对象时调用，返回当前对象的一个实例;注意：这里的第一个参数是cls即class本身def __new__(self, socket_buffer=None):return self.from_buffer_copy(socket_buffer)# __init__(self, *args, **kwargs) 创建完对象后调用，对当前对象的实例的一些初始化，无返回值,即在调用__new__之后，根据返回的实例初始化；注意，这里的第一个参数是self即对象本身【注意和new的区别】def __init__(self, socket_buffer=None):# 协议字段与协议名称的对应self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}# 可读性更强的ip地址(转换32位打包的IPV4地址为IP地址的标准点号分隔字符串表示。)self.src_address = socket.inet_ntoa(struct.pack("<L", self.src))self.dst_address = socket.inet_ntoa(struct.pack("<L", self.dst))# 协议类型try:self.protocol = self.protocol_map[self.protocol_num]except:self.protocol = str(self.protocol_num)if os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol) #raw的中文是生的意思,大概就是原始套接字的意思吧sniffer.bind((host, 0)) #这里端口为0,监听所有端口吧~# 设置在捕获的数据包中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在Windows平台上,我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)try:while True:# 读取数据包raw_buffer = sniffer.recvfrom(65565)[0]# 将缓冲区的前20个字节按IP头进行解析ip_header = IP(raw_buffer[0:20])# 输出协议和通信双方IP地址print "Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)# 处理CTRL-Cexcept KeyboardInterrupt:# 如果运行再Windows上,关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

Python3版本

3、解码ICMP层

如果要主机发现，关闭的端口会对UDP包返回一个ICMP端口不可达的响应，以此判断主机是否存活，所以需要解码ICMP

#!/usr/bin/env python#-*- coding:utf8 -*-import socketimport osimport structfrom ctypes import *# 监听主机，即监听那个网络接口，下面的ip为我的kali的iphost = "10.10.10.145"# ip头定义class IP(Structure):_fields_ = [("ihl", c_ubyte, 4), #ip head length:头长度("version", c_ubyte, 4), #版本("tos", c_ubyte), #服务类型("len", c_ushort),#ip数据包总长度("id", c_ushort), #标识符("offset",c_ushort),#片偏移("ttl", c_ubyte), #生存时间("protocol_num", c_ubyte), #协议数字,应该是协议类型,这里用数字来代表时哪个协议,下面构造函数有设置映射表("sum", c_ushort),#头部校验和("src", c_ulong), #源ip地址("dst", c_ulong) #目的ip地址]# __new__(cls, *args, **kwargs) 创建对象时调用，返回当前对象的一个实例;注意：这里的第一个参数是cls即class本身def __new__(self, socket_buffer=None):return self.from_buffer_copy(socket_buffer)# __init__(self, *args, **kwargs) 创建完对象后调用，对当前对象的实例的一些初始化，无返回值,即在调用__new__之后，根据返回的实例初始化；注意，这里的第一个参数是self即对象本身【注意和new的区别】def __init__(self, socket_buffer=None):# 协议字段与协议名称的对应self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}# 可读性更强的ip地址(转换32位打包的IPV4地址为IP地址的标准点号分隔字符串表示。)self.src_address = socket.inet_ntoa(struct.pack("<L", self.src))self.dst_address = socket.inet_ntoa(struct.pack("<L", self.dst))# 协议类型try:self.protocol = self.protocol_map[self.protocol_num]except:self.protocol = str(self.protocol_num)class ICMP(Structure):#_fields_ = [("type", c_ubyte), #类型("code", c_ubyte), #代码值("checksum", c_ubyte), #头部校验和("unused",c_ubyte), #未使用("next_hop_mtu", c_ubyte) #下一跳的MTU]def __new__(self, socket_buffer):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer):passif os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol) #raw的中文是生的意思,大概就是原始套接字的意思吧sniffer.bind((host, 0)) #这里端口为0,监听所有端口吧~# 设置在捕获的数据包中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在Windows平台上,我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)try:while True:# 读取数据包raw_buffer = sniffer.recvfrom(65565)[0]# 将缓冲区的前20个字节按IP头进行解析ip_header = IP(raw_buffer[0:20])# 输出协议和通信双方IP地址print "Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)# 如果为ICMP,进行处理if ip_header.protocol == "ICMP":# 计算ICMP包的起始位置,并获取ICMP包的数据offset = ip_header.ihl * 4#ihl是头部长度,代表32位(即4字节)长的分片的个数 [我的理解是因为一个字节表示一个符号,所以这里的offset要搞成以字节为单位的,为的是下一句的提取数据]buf = raw_buffer[offset:offset+sizeof(ICMP)]# 解析ICMP数据icmp_header = ICMP(buf)print "ICMP -> Type: %d Code: %d" % (icmp_header.type, icmp_header.code)# 处理CTRL-Cexcept KeyboardInterrupt:# 如果运行再Windows上,关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

Python3版本

#!/usr/bin/env python#-*- coding:utf8 -*-"""源代码在kali2 64上运行会出现错误：`Buffer size too small (20 instead of at least 32 bytes)`解决方法可参考：/questions/29306747/python-sniffing-from-black-hat-python-book修改("src", c_ulong),("dst", c_ulong) self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))为("src", c_uint32),("dst", c_uint32) self.src_address = socket.inet_ntoa(struct.pack("@I",self.src))self.dst_address = socket.inet_ntoa(struct.pack("@I",self.dst))"""import socketimport osimport struct# import threadingfrom ctypes import *# 监听的主机host = "192.168.1.145"class IP(Structure):_fields_ = [("ihl", c_ubyte, 4),("version", c_ubyte, 4),("tos", c_ubyte),("len", c_ushort),("id", c_ushort),("offset", c_ushort),("ttl", c_ubyte),("protocol_num", c_ubyte),("sum", c_ushort),("src", c_uint32),("dst", c_uint32)]def __new__(self, socket_buffer=None):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer=None):# map protocol constants to their namesself.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"}# 转换为可读性更强的ip地址self.src_address = socket.inet_ntoa(struct.pack("@I", self.src))self.dst_address = socket.inet_ntoa(struct.pack("@I", self.dst))# 可读性更强的协议try:self.protocol = self.protocol_map[self.protocol_num]except:self.protocol = str(self.protocol_num)class ICMP(Structure):_fields_ = [("type", c_ubyte),("code", c_ubyte),("checksum", c_ushort),("unused", c_ushort),("next_hop_mtu", c_ushort)]def __new__(self, socket_buffer):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer):pass# 创建一个新的套接字，并绑定到公开接口上if os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol)sniffer.bind((host, 0))# 让捕获的数据中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在windows平台上，我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)try:while True:# 读取数据包raw_buffer = sniffer.recvfrom(65565)[0]# 将缓冲区的前20个字节按IP头进行解析ip_header = IP(raw_buffer[0:20])# 输出协议和通信双方IP地址print("Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address))# 如果是ICMP协议，进行处理if ip_header.protocol == "ICMP":# 计算ICMP包的起始位置offset = ip_header.ihl * 4buf = raw_buffer[offset:offset + sizeof(ICMP)]# 解析ICMP数据icmp_header = ICMP(buf)print("ICMP -> Type: %d Code: %d" % (icmp_header.type, icmp_header.code))# 处理Ctrl+Cexcept KeyboardInterrupt:# 如果运行在windows上，关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

4、发现主机

最终我们是希望能有一个主机扫描器

所以还要加一个netaddr模块

#!/usr/bin/env python#-*- coding:utf8 -*-import socketimport osimport structimport threadingimport timeimport sysfrom netaddr import IPNetwork,IPAddressfrom ctypes import *# 监听主机，即监听那个网络接口，下面的ip为我的kali的iphost = "10.10.10.145"# 扫描的目标子网# subnet = "192.168.1.0/24"# 没有命令行参数,默认192.168.1.0/24if len(sys.argv) == 1:subnet = "192.168.1.0/24"else:subnet = sys.argv[1]# 自定义的字符串,我们将在ICMP响应中进行核对magic_message = "PYTHONRULES!"# 批量发送UDP数据包def udp_sender(subnet, magic_message):time.sleep(5) #可以说程序暂停5秒吧# 建立一个socket对象(SOCK_DGRAM:UDP客户端)sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)for ip in IPNetwork(subnet):try:# 尝试发送magic_message这个消息到子网的每个ip,还用了个不怎么可能用的65212端口sender.sendto(magic_message, ("%s" % ip, 65212))except:pass #代表什么也不做# ip头定义class IP(Structure):_fields_ = [("ihl", c_ubyte, 4), #ip head length:头长度("version", c_ubyte, 4), #版本("tos", c_ubyte), #服务类型("len", c_ushort),#ip数据包总长度("id", c_ushort), #标识符("offset",c_ushort),#片偏移("ttl", c_ubyte), #生存时间("protocol_num", c_ubyte), #协议数字,应该是协议类型,这里用数字来代表时哪个协议,下面构造函数有设置映射表("sum", c_ushort),#头部校验和("src", c_ulong), #源ip地址("dst", c_ulong) #目的ip地址]# __new__(cls, *args, **kwargs) 创建对象时调用，返回当前对象的一个实例;注意：这里的第一个参数是cls即class本身def __new__(self, socket_buffer=None):return self.from_buffer_copy(socket_buffer)# __init__(self, *args, **kwargs) 创建完对象后调用，对当前对象的实例的一些初始化，无返回值,即在调用__new__之后，根据返回的实例初始化；注意，这里的第一个参数是self即对象本身【注意和new的区别】def __init__(self, socket_buffer=None):# 协议字段与协议名称的对应self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}# 可读性更强的ip地址(转换32位打包的IPV4地址为IP地址的标准点号分隔字符串表示。)self.src_address = socket.inet_ntoa(struct.pack("<L", self.src))self.dst_address = socket.inet_ntoa(struct.pack("<L", self.dst))# 协议类型try:self.protocol = self.protocol_map[self.protocol_num]except:self.protocol = str(self.protocol_num)class ICMP(Structure):#_fields_ = [("type", c_ubyte), #类型("code", c_ubyte), #代码值("checksum", c_ubyte), #头部校验和("unused",c_ubyte), #未使用("next_hop_mtu", c_ubyte) #下一跳的MTU]def __new__(self, socket_buffer):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer):passif os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol) #raw的中文是生的意思,大概就是原始套接字的意思吧sniffer.bind((host, 0)) #这里端口为0,监听所有端口吧~# 设置在捕获的数据包中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在Windows平台上,我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)# 开启多线程发送udp数据包t = threading.Thread(target=udp_sender, args=(subnet, magic_message))t.start()try:while True:# 读取数据包raw_buffer = sniffer.recvfrom(65565)[0]# 将缓冲区的前20个字节按IP头进行解析ip_header = IP(raw_buffer[0:20])# 输出协议和通信双方IP地址#print "Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)# 如果为ICMP,进行处理if ip_header.protocol == "ICMP":# 计算ICMP包的起始位置,并获取ICMP包的数据offset = ip_header.ihl * 4#ihl是头部长度,代表32位(即4字节)长的分片的个数 [我的理解是因为一个字节表示一个符号,所以这里的offset要搞成以字节为单位的,为的是下一句的提取数据]buf = raw_buffer[offset:offset+sizeof(ICMP)]# 解析ICMP数据icmp_header = ICMP(buf)#print "ICMP -> Type: %d Code: %d" % (icmp_header.type, icmp_header.code)# 检查类型和代码值是否都为2if icmp_header.type == 3 and icmp_header.code == 3:# 确认响应的主机再我们的目标子网之内if IPAddress(ip_header.src_address) in IPNetwork(subnet):# 确认ICMP包中包含我们发送的自定义的字符串if raw_buffer[len(raw_buffer) - len(magic_message):] == magic_message:print "Host Up: %s" % ip_header.src_address# 处理CTRL-Cexcept KeyboardInterrupt:# 如果运行再Windows上,关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

Python3版本

#!/usr/bin/env python#-*- coding:utf8 -*-"""源代码在kali2 64上运行会出现错误：`Buffer size too small (20 instead of at least 32 bytes)`解决方法可参考：/questions/29306747/python-sniffing-from-black-hat-python-book修改("src", c_ulong),("dst", c_ulong)self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))为("src", c_uint32),("dst", c_uint32)self.src_address = socket.inet_ntoa(struct.pack("@I",self.src))self.dst_address = socket.inet_ntoa(struct.pack("@I",self.dst))"""import timeimport socketimport osimport structimport threadingfrom netaddr import IPNetwork, IPAddressfrom ctypes import *# 监听的主机host = "192.168.1.145"# 扫描的目标子网subnet = "192.168.1.0/24"# 自定义的字符串，我们将在ICMP响应中进行核对magic_message = "PYTHONRULES!"# 批量发送UDP数据包def udp_sender(subnet, magic_message):time.sleep(5)sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)for ip in IPNetwork(subnet):try:sender.sendto(magic_message, ("%s" % ip, 65212))except:passclass IP(Structure):_fields_ = [("ihl", c_ubyte, 4),("version", c_ubyte, 4),("tos", c_ubyte),("len", c_ushort),("id", c_ushort),("offset", c_ushort),("ttl", c_ubyte),("protocol_num", c_ubyte),("sum", c_ushort),("src", c_uint32),("dst", c_uint32)]def __new__(self, socket_buffer=None):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer=None):# map protocol constants to their namesself.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"}# 转换为可读性更强的ip地址self.src_address = socket.inet_ntoa(struct.pack("@I", self.src))self.dst_address = socket.inet_ntoa(struct.pack("@I", self.dst))# 可读性更强的协议try:self.protocol = self.protocol_map[self.protocol_num]except:self.protocol = str(self.protocol_num)class ICMP(Structure):_fields_ = [("type", c_ubyte),("code", c_ubyte),("checksum", c_ushort),("unused", c_ushort),("next_hop_mtu", c_ushort)]def __new__(self, socket_buffer):return self.from_buffer_copy(socket_buffer)def __init__(self, socket_buffer):pass# 创建一个新的套接字，并绑定到公开接口上if os.name == "nt":socket_protocol = socket.IPPROTO_IPelse:socket_protocol = socket.IPPROTO_ICMPsniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol)sniffer.bind((host, 0))# 让捕获的数据中包含IP头sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)# 在windows平台上，我们需要设置IOCTL以启用混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)# 开始发送数据包t = threading.Thread(target=udp_sender, args=(subnet,magic_message))t.start()try:while True:# 读取数据包raw_buffer = sniffer.recvfrom(65565)[0]# 将缓冲区的前20个字节按IP头进行解析ip_header = IP(raw_buffer[0:20])# 输出协议和通信双方IP地址# print("Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address))# 如果是ICMP协议，进行处理if ip_header.protocol == "ICMP":# 计算ICMP包的起始位置offset = ip_header.ihl * 4buf = raw_buffer[offset:offset + sizeof(ICMP)]# 解析ICMP数据icmp_header = ICMP(buf)# print("ICMP -> Type: %d Code: %d" % (icmp_header.type, icmp_header.code))# 检查类型和代码值是否为3if icmp_header.code == 3 and icmp_header.type == 3:# 确认响应的主机在我们的目标子网之内if IPAddress(ip_header.src_address) in IPNetwork(subnet):# 确认ICMP数据中包含我们发送的自定义的字符串if raw_buffer[len(raw_buffer) - len(magic_message):] == magic_message:print("Host Up: %s" % ip_header.src_address)# 处理Ctrl+Cexcept KeyboardInterrupt:# 如果运行在windows上，关闭混杂模式if os.name == "nt":sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)