快捷導(dǎo)航

python構(gòu)造icmp echo請(qǐng)求和實(shí)現(xiàn)網(wǎng)絡(luò)探測(cè)器功能代碼分享

更新時(shí)間：2014年01月10日 10:26:12 作者：

本文分享了二個(gè)python示例，python構(gòu)造icmp echo請(qǐng)求、實(shí)現(xiàn)網(wǎng)絡(luò)探測(cè)器功能代碼，類似nmap功能

python發(fā)送icmp echo requesy請(qǐng)求

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

掃描探測(cè)網(wǎng)絡(luò)功能(網(wǎng)絡(luò)探測(cè)器)

復(fù)制代碼代碼如下:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

'''
探測(cè)網(wǎng)絡(luò)主機(jī)存活。
'''

import os
import struct
import array
import time
import socket
import IPy
import threading

class SendPingThr(threading.Thread):
    '''
    發(fā)送ICMP請(qǐng)求報(bào)文的線程。

    參數(shù)：
        ipPool      -- 可迭代的IP地址池
        icmpPacket  -- 構(gòu)造的icmp報(bào)文
        icmpSocket  -- icmp套字接
        timeout     -- 設(shè)置發(fā)送超時(shí)
    '''
    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)  #等待接收線程啟動(dòng)
        for ip in self.ipPool:
            try:
                self.Sock.sendto(self.packet, (ip, 0))
            except socket.timeout:
                break
        time.sleep(self.timeout)

class Nscan:
    '''
    參數(shù)：
        timeout    -- Socket超時(shí)，默認(rèn)3秒
        IPv6       -- 是否是IPv6，默認(rèn)為False
    '''
    def __init__(self, timeout=3, IPv6=False):
        self.timeout = timeout
        self.IPv6 = IPv6

        self.__data = struct.pack('d', time.time())   #用于ICMP報(bào)文的負(fù)荷字節(jié)（8bit）
        self.__id = os.getpid()   #構(gòu)造ICMP報(bào)文的ID字段，無(wú)實(shí)際意義

    @property   #屬性裝飾器
    def __icmpSocket(self):
        '''創(chuàng)建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 報(bào)文效驗(yàn)和計(jì)算方法'''
        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):
        '''構(gòu)造 ICMP 報(bào)文'''
        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是否是一個(gè)合法的單播地址'''
        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):
        '''生產(chǎn) 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報(bào)文探測(cè)網(wǎng)絡(luò)主機(jī)存活

        參數(shù)：
            ipPool  -- 可迭代的IP地址池
        '''
        Sock = self.__icmpSocket
        Sock.settimeout(self.timeout)
        packet = self.__icmpPacket
        recvFroms = set()   #接收線程的來(lái)源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) )