加入收藏 | 设为首页 | 会员中心 | 我要投稿 李大同 (https://www.lidatong.com.cn/)- 科技、建站、经验、云计算、5G、大数据,站长网!
当前位置: 首页 > 编程开发 > Python > 正文

python实现的ping功能

发布时间:2020-12-17 17:16:32 所属栏目:Python 来源:网络整理
导读:今天PHP站长网 52php.cn把收集自互联网的代码分享给大家,仅供参考。 #!/usr/bin/env python#coding:utf-8import os,sys,socket,struct,select,time # From /usr/include/linux/icmp.h; your milage may vary.ICMP_ECHO_R

以下代码由PHP站长网 52php.cn收集自互联网

现在PHP站长网小编把它分享给大家,仅供参考

#!/usr/bin/env python
#coding:utf-8
import os,sys,socket,struct,select,time
  
# From /usr/include/linux/icmp.h; your milage may vary.
ICMP_ECHO_REQUEST = 8 # Seems to be the same on Solaris.
  
  
def checksum(source_string):
    """
    I'm not too confident that this is right but testing seems
    to suggest that it gives the same answers as in_cksum in ping.c
    """
    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 # Necessary?
        count = count + 2
  
    if countTo<len(source_string):
        sum = sum + ord(source_string[len(source_string) - 1])
        sum = sum & 0xffffffff # Necessary?
  
    sum = (sum >> 16)  +  (sum & 0xffff)
    sum = sum + (sum >> 16)
    answer = ~sum
    answer = answer & 0xffff
  
    # Swap bytes. Bugger me if I know why.
    answer = answer >> 8 | (answer << 8 & 0xff00)
  
    return answer
  
  
def receive_one_ping(my_socket,ID,timeout):
    """
    receive the ping from the socket.
    """
    timeLeft = timeout
    while True:
        startedSelect = time.time()
        whatReady = select.select([my_socket],[],timeLeft)
        howLongInSelect = (time.time() - startedSelect)
        if whatReady[0] == []: # Timeout
            return
  
        timeReceived = time.time()
        recPacket,addr = my_socket.recvfrom(1024)
        icmpHeader = recPacket[20:28]
        type,code,checksum,packetID,sequence = struct.unpack(
            "bbHHh",icmpHeader
        )
        if packetID == ID:
            bytesInDouble = struct.calcsize("d")
            timeSent = struct.unpack("d",recPacket[28:28 + bytesInDouble])[0]
            return timeReceived - timeSent
  
        timeLeft = timeLeft - howLongInSelect
        if timeLeft <= 0:
            return
  
  
def send_one_ping(my_socket,dest_addr,ID):
    """
    Send one ping to the given >dest_addr<.
    """
    dest_addr  =  socket.gethostbyname(dest_addr)
  
    # Header is type (8),code (8),checksum (16),id (16),sequence (16)
    my_checksum = 0
  
    # Make a dummy heder with a 0 checksum.
    header = struct.pack("bbHHh",ICMP_ECHO_REQUEST,my_checksum,1)  #压包
    #a1 = struct.unpack("bbHHh",header)    #my test
    bytesInDouble = struct.calcsize("d")
    data = (192 - bytesInDouble) * "Q"
    data = struct.pack("d",time.time()) + data
  
    # Calculate the checksum on the data and the dummy header.
    my_checksum = checksum(header + data)
  
    # Now that we have the right checksum,we put that in. It's just easier
    # to make up a new header than to stuff it into the dummy.
    header = struct.pack("bbHHh",socket.htons(my_checksum),1)
    packet = header + data
    my_socket.sendto(packet,(dest_addr,1)) # Don't know about the 1
  
  
def do_one(dest_addr,timeout):
    """
    Returns either the delay (in seconds) or none on timeout.
    """
    icmp = socket.getprotobyname("icmp")
    try:
        my_socket = socket.socket(socket.AF_INET,socket.SOCK_RAW,icmp)
    except socket.error,(errno,msg):
        if errno == 1:
            # Operation not permitted
            msg = msg + (
                " - Note that ICMP messages can only be sent from processes"
                " running as root."
            )
            raise socket.error(msg)
        raise # raise the original error
  
    my_ID = os.getpid() & 0xFFFF
  
    send_one_ping(my_socket,my_ID)
    delay = receive_one_ping(my_socket,my_ID,timeout)
  
    my_socket.close()
    return delay
  
  
def verbose_ping(dest_addr,timeout = 2,count = 100):
    """
    Send >count< ping to >dest_addr< with the given >timeout< and display
    the result.
    """
    for i in xrange(count):
        print "ping %s..." % dest_addr,try:
            delay  =  do_one(dest_addr,timeout)
        except socket.gaierror,e:
            print "failed. (socket error: '%s')" % e[1]
            break
 
        if delay  ==  None:
            print "failed. (timeout within %ssec.)" % timeout
        else:
            delay  =  delay * 1000
            print "get ping in %0.4fms" % delay
  
if __name__ == '__main__':
    verbose_ping("www.163.com",2,1)

以上内容由PHP站长网【52php.cn】收集整理供大家参考研究

如果以上内容对您有帮助,欢迎收藏、点赞、推荐、分享。

(编辑:李大同)

【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容!

    推荐文章
      热点阅读