原始套接字和流量嗅探

admin 2022年6月7日00:00:32评论16 views字数 16277阅读54分15秒阅读模式


原始套接字和流量嗅探


前言

《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_ICMP
sniffer = 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_ICMP
sniffer = 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版本

#!/usr/bin/env python#-*- coding:utf8 -*-
"""源代码在kali2 64上运行会出现错误:`Buffer size too small (20 instead of at least 32 bytes)`解决方法可参考:https://stackoverflow.com/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 structfrom ctypes import *
# 监听的主机host = "192.168.1.145"
# ip头定义class IP(Structure): _fields_ =[ ("ih1", 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): # 协议字段与协议名称对应 self.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)
# 下面的代码类似于之前的例子
# 创建原始套接字, 然后绑定在公开接口上if os.name == "nt": socket_protocol = socket.IPPROTO_IPelse: socket_protocol = socket.IPPROTO_ICMP
sniffer = 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))
# 处理Ctrl+Cexcept KeyboardInterrupt: # 如果运行在windows上, 关闭混杂模式 if os.name == "nt":        sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

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): pass

if os.name == "nt": socket_protocol = socket.IPPROTO_IPelse: socket_protocol = socket.IPPROTO_ICMP
sniffer = 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)`解决方法可参考:https://stackoverflow.com/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 threading
from 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 names self.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_ICMP
sniffer = 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 * 4 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)

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): pass

if os.name == "nt": socket_protocol = socket.IPPROTO_IPelse: socket_protocol = socket.IPPROTO_ICMP
sniffer = 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)
# 检查类型和代码值是否都为2 if 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)`解决方法可参考:https://stackoverflow.com/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: pass
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 names self.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_ICMP
sniffer = 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 * 4 buf = raw_buffer[offset:offset + sizeof(ICMP)]
# 解析ICMP数据 icmp_header = ICMP(buf)
# print("ICMP -> Type: %d Code: %d" % (icmp_header.type, icmp_header.code))
# 检查类型和代码值是否为3 if 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)

结语

利用socket,根据关闭端口的ICMP不可达响应,一步一步做了一个简单的主机扫描器




红客突击队于2019年由队长k龙牵头,联合国内多位顶尖高校研究生成立。其团队从成立至今多次参加国际网络安全竞赛并取得良好成绩,积累了丰富的竞赛经验。团队现有三十多位正式成员及若干预备人员,下属联合分队数支。红客突击队始终秉承先做人后技术的宗旨,旨在打造国际顶尖网络安全团队。

原文始发于微信公众号(红客突击队):原始套接字和流量嗅探

  • 左青龙
  • 微信扫一扫
  • weinxin
  • 右白虎
  • 微信扫一扫
  • weinxin
admin
  • 本文由 发表于 2022年6月7日00:00:32
  • 转载请保留本文链接(CN-SEC中文网:感谢原作者辛苦付出):
                   原始套接字和流量嗅探http://cn-sec.com/archives/1091224.html

发表评论

匿名网友 填写信息