python构造icmp echo请求和实现网络探测器功能代码分享
python发送icmp echo requesy请求
import socket
import struct
def checksum(source_string):
sum = 0
countto = (len(source_string)/2)*2
count = 0
while count<countto:
thisval = ord(source_string[count + 1])*256 + ord(source_string[count])
sum = sum + thisval
sum = sum & 0xffffffff
count = count + 2
if countto<len(source_string):
sum = sum + ord(source_string[len(source_string) - 1])
sum = sum & 0xffffffff
sum = (sum >> 16) + (sum & 0xffff)
sum = sum + (sum >> 16)
answer = ~sum
answer = answer & 0xffff
answer = answer >> 8 | (answer << 8 & 0xff00)
return answer
def ping(ip):
s = socket.socket(socket.af_inet, socket.sock_raw, 1)
packet = struct.pack(
"!bbhhh", 8, 0, 0, 0, 0
)
chksum=checksum(packet)
packet = struct.pack(
"!bbhhh", 8, 0, chksum, 0, 0
)
s.sendto(packet, (ip, 1))
if __name__=='__main__':
ping('192.168.41.56')
扫描探测网络功能(网络探测器)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
探测网络主机存活。
'''
import os
import struct
import array
import time
import socket
import ipy
import threading
class sendpingthr(threading.thread):
'''
发送icmp请求报文的线程。
参数:
ippool -- 可迭代的ip地址池
icmppacket -- 构造的icmp报文
icmpsocket -- icmp套字接
timeout -- 设置发送超时
'''
def __init__(self, ippool, icmppacket, icmpsocket, timeout=3):
threading.thread.__init__(self)
self.sock = icmpsocket
self.ippool = ippool
self.packet = icmppacket
self.timeout = timeout
self.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:
break
time.sleep(self.timeout)
class nscan:
'''
参数:
timeout -- socket超时,默认3秒
ipv6 -- 是否是ipv6,默认为false
'''
def __init__(self, timeout=3, ipv6=false):
self.timeout = timeout
self.ipv6 = ipv6
self.__data = struct.pack('d', time.time()) #用于icmp报文的负荷字节(8bit)
self.__id = os.getpid() #构造icmp报文的id字段,无实际意义
@property #属性装饰器
def __icmpsocket(self):
'''创建icmp socket'''
if not self.ipv6:
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 sock
def __incksum(self, packet):
'''icmp 报文效验和计算方法'''
if len(packet) & 1:
packet = packet + '\0'
words = array.array('h', packet)
sum = 0
for word in words:
sum += (word & 0xffff)
sum = (sum >> 16) + (sum & 0xffff)
sum = sum + (sum >> 16)
return (~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、seq
else:
header = struct.pack('bbhhh', 128, 0, 0, self.__id, 0)
packet = header + self.__data # packet without checksum
chksum = self.__incksum(packet) # make checksum
if 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* checksum
def 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[0] != 127 and ip[1] < 256 and ip[2] < 256 and 0 < ip[3] < 255):
return true
return false
def makeippool(self, startip, lastip):
'''生产 ip 地址池'''
ipver = 6 if self.ipv6 else 4
intip = lambda ip: ipy.ip(ip).int()
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 mping(self, ippool):
'''利用icmp报文探测网络主机存活
参数:
ippool -- 可迭代的ip地址池
'''
sock = self.__icmpsocket
sock.settimeout(self.timeout)
packet = self.__icmppacket
recvfroms = set() #接收线程的来源ip地址容器
sendthr = sendpingthr(ippool, packet, sock, self.timeout)
sendthr.start()
while true:
try:
recvfroms.add(sock.recvfrom(1024)[1][0])
except exception:
pass
finally:
if not sendthr.isalive():
break
return recvfroms & ippool
if __name__=='__main__':
s = nscan()
ippool = s.makeippool('192.168.0.1', '192.168.0.254')
print( s.mping(ippool) )