欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

python构造icmp echo请求和实现网络探测器功能代码分享

程序员文章站 2023-01-31 10:33:44
python发送icmp echo requesy请求复制代码 代码如下:import socketimport struct def checksum(source_s...

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