700字范文 > 利用ICMP协议使用python原始套接字实现主机存活探测工具

利用ICMP协议使用python原始套接字实现主机存活探测工具

时间：2021-10-02 09:12:36

一.课题概述。

一学期一次的课程设计终于开始了（停课两周，马上放寒假了，哈哈哈哈哈哈。。。）这次我们课程设计的科目是计算机协议，我们小组抽到的题目是利用ICMP模仿ping命令写一个主机存活探测的工具。具体描述和需求如下：

【实验目的】

1. 加深对ICMP协议的理解

2. 掌握原始套接字进行网络程序设计的方法

【案例描述】

Ping工具是使用ICMP协议进行网络连通性检测的工具，在日常生活中使用广泛。请根据ICMP协议的相关原理，使用开发工具为同学开发一个Ping工具。

【需求分析】

根据案例描述，可以总结出用户有以下需求：

1. 使用该工具可以测试目标主机的状态

根据ICMP回显请求和回显应答报文，使用该工具测试目标主机的状态。

2. 程序应该提供帮助信息

为了方便用户使用，该工具应该提供帮助信息，当用户需要帮助时可以进行查询。

【总体设计】

1. 使用原始套接字

由于该工具需要发送和接收ICMP报文，所以应该使用原始套接字。

2. 发送ICMP回显请求报文

可以发送ICMP回显请求报文给目的主机

3. 接收ICMP回显应答报文

可以接收来自目的主机的ICMP回显应答报文。

4. 程序应该提供帮助信息

为了方便用户使用，该工具应该提供帮助信息。可以在程序中设计一个用于提供帮助的函数，当用户需要帮助时调用该函数。

二、分析。

题目不是很难，在正式开始之前要首先弄清楚一个概念什么是原始套接字和标准套接字？它们有什么不同？

标准套接字，指在传输层下面使用的套接字。流式套接字和数据报套接字这两种套接字工作在传输层，主要为应用层的应用程序提供服务，并且在接收和发送时只能操作数据部分，而不能对IP首部或TCP和UDP首部进行操作，通常把这两种套接字称为标准套接字。原始套接字，它是一种对原始网络报文进行处理的套接字。原始套接字的用途主要有：

（1）发送自定义的IP 数据报

（2）发送ICMP 数据报

（3）网卡的侦听模式，监听网络上的数据包。

（4）伪装IP地址。

（5）自定义协议的实现。

比如发送一个自定义的IP包、UDP包、TCP包或ICMP包，捕获所有经过本机网卡的数据包，伪装本机的IP，想要操作IP首部或传输层协议首部，等等。

弄清楚这个概念后，我们主要在学习怎样构造ICMP Request报文。先学习ICMP Request报文的格式如下图：

下表是各字段的含义：

三、代码实现。

1.代码结构。

main.py用来实现各个代码的调用和命令行化，networkscan实现单个主机的探测，networkscans实现网段探测主机存活数，readlogs实现读取日志文件，logs是日志文件，下图是主要功能代码的各个方法。

2.主要功能分析。

首先是代码使用到的模块。

import osimport structimport array import timeimport socketimport loggingfrom queue import Queue

struct：生成用于网络传输的而二进制数据。array：是python中实现的一种高效的数组存储类型。socket：套接字。logging：主要用于输出运行日志，可以设置输出日志的等级、日志保存路径、日志文件回滚等。Queue：实现了一个基本的先进先出(FIFO)容器，使用put()将元素添加到序列尾端，get()从队列尾部移除元素。

代码主要用两个类来实现其中SendPing类中的run()方法主要用来实现对自定义ICMP数据包的发送。

class SendPing():'''发送ICMP请求报文的线程。参数：ipPool-- IP地址icmpPacket -- 构造的icmp报文icmpSocket -- icmp套字接timeout-- 设置发送超时'''def __init__(self, ipPool, icmpPacket, icmpSocket, timeout=3):self.Sock = icmpSocketself.ipPool = ipPoolself.packet = icmpPacketself.timeout = timeoutself.Sock.settimeout( timeout + 3 )def run(self):try:self.Sock.sendto(self.packet, (self.ipPool, 0))except OSError:print("你不能输入一个内部保留地址！")exit()

def makeIpPool(self, startIP, lastIP):'''生产 IP 地址池'''IPver = 6 if self.IPv6 else 4intIP = lambda ip: IPy.IP(ip).int() #将IP地址转换为整型格式ipPool = {IPy.intToIp(ip, IPver) for ip in range(intIP(startIP), intIP(lastIP)+1)}return {ip for ip in ipPool if self.isUnIP(ip)}

NetworkScan()方法用来实现探测过程， isUnIP()主要用来判断IP地址的合法性，mPing方法中又调用了__icmpSocket和__icmpPacket方法来建立和构造套接字和ICMP数据包，再通过调SendPing类实现发送给目标主机。

# -*- coding: utf-8 -*-import osimport structimport array #array模块是python中实现的一种高效的数组存储类型。import timeimport socketimport loggingfrom queue import Queue'''Queue类实现了一个基本的先进先出(FIFO)容器，使用put()将元素添加到序列尾端，get()从队列尾部移除元素。''''''filename：指定日志文件名；filemode：和file函数意义相同，指定日志文件的打开模式，'w'或者'a'；format：指定输出的格式和内容，format可以输出很多有用的信息，'''logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S",filename="logs", filemode="a")class SendPing():'''发送ICMP请求报文的线程。参数：ipPool-- 可迭代的IP地址池icmpPacket -- 构造的icmp报文icmpSocket -- icmp套字接timeout-- 设置发送超时'''def __init__(self, ip, icmpPacket, icmpSocket, timeout=3):self.Sock = icmpSocketself.ip = ipself.packet = icmpPacketself.timeout = timeoutself.Sock.settimeout( timeout + 3 )def run(self):try:self.Sock.sendto(self.packet, (self.ip, 0))except OSError:print("你不能输入一个内部保留地址！")exit()class NetworkScan():'''参数：timeout -- Socket超时，默认3秒IPv6 -- 是否是IPv6，默认为False'''def __init__(self, timeout=3, IPv6=False):self.timeout = timeoutself.IPv6 = IPv6self._LOGS = Queue()'''按照给定的格式(fmt),把数据转换成字符串(字节流),并将该字符串返回.'''self.__data = struct.pack('d', time.time()) #用于ICMP报文的负荷字节（8bit）self.__id = os.getpid() #构造ICMP报文的ID字段，无实际意义@property #属性装饰器def __icmpSocket(self):'''socket.getprotobyname('icmp')创建ICMP Socket'''if not self.IPv6:#socket.SOCK_RAW 原始套接字#作用：获得网络协议名（如：'icmp'）对应的值Sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.getprotobyname("icmp"))else:Sock = socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.getprotobyname("ipv6-icmp"))return Sockdef __inCksum(self, packet):'''ICMP 报文效验和计算方法& 按位与运算符：参与运算的两个值,如果两个相应位都为1,则该位的结果为1,否则为0'''if len(packet) & 1:packet = packet + '\\0'words = array.array('h', packet)sum = 0for word in words:sum += (word & 0xffff)'''右移动运算符：把">>"左边的运算数的各二进位全部右移若干位，>> 右边的数字指定了移动的位数'''sum = (sum >> 16) + (sum & 0xffff)sum = sum + (sum >> 16)#按位取反运算符：对数据的每个二进制位取反,即把1变为0,把0变为1 。~x 类似于 -x-1return (~sum) & 0xffff@property #负责把一个方法变成属性调用def __icmpPacket(self):'''构造 ICMP 报文'''if not self.IPv6:header = struct.pack('bbHHh', 8, 0, 0, self.__id, 0) # TYPE、CODE、CHKSUM、ID、SEQelse:header = struct.pack('BbHHh', 128, 0, 0, self.__id, 0)packet = header + self.__data# packet without checksumchkSum = self.__inCksum(packet) # make checksumif not self.IPv6:header = struct.pack('bbHHh', 8, 0, chkSum, self.__id, 0)else:header = struct.pack('BbHHh', 128, 0, chkSum, self.__id, 0)return header + self.__data # packet *with* checksumdef isUnIP(self, IP):'''判断IP是否是一个合法的单播地址'''IP = [int(x) for x in IP.split('.') if x.isdigit()]if len(IP) == 4:if (0 < IP[0] < 223 and IP[1] < 256 and IP[2] < 256 and 0 < IP[3] < 255):return Truereturn Falsedef mPing(self, ip):'''利用ICMP报文探测网络主机存活参数：ipPool -- 可迭代的IP地址池'''Sock = self.__icmpSocketSock.settimeout(self.timeout)packet = self.__icmpPacketrecvFroms = ''sendThr = SendPing(ip, packet, Sock, self.timeout)sendThr.run()try:recvFroms = Sock.recvfrom(1024)[1][0]except Exception:passreturn recvFromsdef NetworkScan(self, network): #设置要扫描的网段self.print_logs(" 等待中。。。。。。")if self.isUnIP(network):alive_ip = self.mPing(network)if alive_ip != '':self.print_logs("%s is alive." % network)else:self.print_logs("%s is die." % network)else:self.print_logs("输入的IP地址有误！")def print_logs(self, msg):print(time.strftime("[%Y-%m-%d %H:%M:%S] ", time.localtime()) + msg)logging.info(msg)self._LOGS.put(time.strftime("[%Y-%m-%d %H:%M:%S] ", time.localtime()) + msg)

主要看一下这个语句 header = struct.pack(‘bbHHh’, 8, 0, 0, self.__id, 0) 其中bbHHh是用来控制网络传输的数据格式，其与的参数分别对应ICMP报文的TYPE、CODE、CHKSUM、ID、SEQ字段。然后使用packet = header + self.__data，自定义完整的ICMP数据包，再使用 self.__inCksum(packet)语句，调用计算检验和的方法计算出检验后和，最后在重新封装ICMP数据包。

3.多线程实现网段探测主机存活数。

SendPingThr类中主要构建了run()方法用来实现发送ICMP数据包和建立多线程扫描提高代码运行速度。

class SendPingThr(threading.Thread):def __init__(self, ipPool, icmpPacket, icmpSocket, timeout=3):threading.Thread.__init__(self)self.Sock = icmpSocketself.ipPool = ipPoolself.packet = icmpPacketself.timeout = timeoutself.Sock.settimeout( timeout + 3 )def run(self):time.sleep(0.01) #等待接收线程启动for ip in self.ipPool:try:self.Sock.sendto(self.packet, (ip, 0))except socket.timeout:breaktime.sleep(self.timeout)

NetworkScan类中添加了makeIpPool()方法生成地址池，修改了NetworkScan()方法用来实现扫描过程，首先它有设置网段的功能，然后通过设置的网段调用makeIpPool()方法生成地址池，再将地址池中的每个地址通过for循环传给mPing()方法实现发送数据包的过程。

def makeIpPool(self, startIP, lastIP):'''生产 IP 地址池'''IPver = 6 if self.IPv6 else 4intIP = lambda ip: IPy.IP(ip).int() #将IP地址转换为整型格式ipPool = {IPy.intToIp(ip, IPver) for ip in range(intIP(startIP), intIP(lastIP)+1)}return {ip for ip in ipPool if self.isUnIP(ip)}def NetworkScan(self, network): #设置要扫描的网段args = "".join(network)ip_prefix = '.'.join(args.split('.')[:-1])ip_start = ip_prefix + ".1"ip_end = ip_prefix + ".255"self.print_logs(" [*] 开始内网主机扫描")ipPool = self.makeIpPool(ip_start, ip_end)alive_ip = self.mPing(ipPool)for i in alive_ip:self.print_logs(" [+] %s is alive." % i)

3.命令行化。

实现命令行化主要使用argparse模块，想具体学习argparse模块的可以查看官方文档。

import networkscansimport networkscanimport argparsefrom readlogs import readif __name__ == '__main__':#定义一个容器parser = argparse.ArgumentParser(description="这是一个探测工具！", formatter_class=argparse.RawTextHelpFormatter,epilog='''use examples:python main.py -i 192.168.1.1 python main.py -s 192.168.1.0python main.py -r logs''')#设置需要的参数parser.add_argument('-i', metavar = '', help = '探测主机存活，-i参数后面输入主机IP')parser.add_argument('-s', metavar = '',help = '探测内网存活的主机，-s参数后面输入一个内网网段')parser.add_argument('-r', metavar='',help = '-r参数后加logs,查看日志文件')args = parser.parse_args()#实现参数的各个功能if args.i:s = workScan()workScan(args.i)elif args.s:s = workScan()workScan(args.s)elif args.r:read(args.r)else:print("输入的参数有误，请使用-h参数查看帮助信息！")