[雪峰磁针石博客]python网络作业:使用python的socket库实现ICMP协议的ping

简介:

ICMP ping是您遇到过的最常见的网络扫描类型。 打开命令行提示符或终端并输入ping www.google.com非常容易。

为什么要在python中实现?

  • 很多名牌大学喜欢考试用python的socket库实现ICMP协议的ping
  • 个别环境没有ping

直接上代码:

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# 技术支持:https://www.jianshu.com/u/69f40328d4f0 
# 技术支持 https://china-testing.github.io/
# https://github.com/china-testing/python-api-tesing/blob/master/practices/ping.py
#qq群144081101 567351477
# CreateDate: 2018-11-22

import os 
import argparse 
import socket
import struct
import select
import time


ICMP_ECHO_REQUEST = 8 # Platform specific
DEFAULT_TIMEOUT = 2
DEFAULT_COUNT = 4 


class Pinger(object):
    """ Pings to a host -- the Pythonic way"""

    def __init__(self, target_host, count=DEFAULT_COUNT, timeout=DEFAULT_TIMEOUT):
        self.target_host = target_host
        self.count = count
        self.timeout = timeout


    def do_checksum(self, source_string):
        """  Verify the packet integritity """
        sum = 0
        max_count = (len(source_string)/2)*2
        count = 0
        while count < max_count:

            val = source_string[count + 1]*256 + source_string[count]                   
            sum = sum + val
            sum = sum & 0xffffffff 
            count = count + 2

        if max_count<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 receive_pong(self, sock, ID, timeout):
        """
        Receive ping from the socket.
        """
        time_remaining = timeout
        while True:
            start_time = time.time()
            readable = select.select([sock], [], [], time_remaining)
            time_spent = (time.time() - start_time)
            if readable[0] == []: # Timeout
                return

            time_received = time.time()
            recv_packet, addr = sock.recvfrom(1024)
            icmp_header = recv_packet[20:28]
            type, code, checksum, packet_ID, sequence = struct.unpack(
       "bbHHh", icmp_header
   )
            if packet_ID == ID:
                bytes_In_double = struct.calcsize("d")
                time_sent = struct.unpack("d", recv_packet[28:28 + bytes_In_double])[0]
                return time_received - time_sent

            time_remaining = time_remaining - time_spent
            if time_remaining <= 0:
                return


    def send_ping(self, sock,  ID):
        """
        Send ping to the target host
        """
        target_addr  =  socket.gethostbyname(self.target_host)

        my_checksum = 0

        # Create a dummy heder with a 0 checksum.
        header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, ID, 1)
        bytes_In_double = struct.calcsize("d")
        data = (192 - bytes_In_double) * "Q"
        data = struct.pack("d", time.time()) + bytes(data.encode('utf-8'))

        # Get the checksum on the data and the dummy header.
        my_checksum = self.do_checksum(header + data)
        header = struct.pack(
      "bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), ID, 1
  )
        packet = header + data
        sock.sendto(packet, (target_addr, 1))


    def ping_once(self):
        """
        Returns the delay (in seconds) or none on timeout.
        """
        icmp = socket.getprotobyname("icmp")
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp)
        except socket.error as e:
            if e.errno == 1:
                # Not superuser, so operation not permitted
                e.msg +=  "ICMP messages can only be sent from root user processes"
                raise socket.error(e.msg)
        except Exception as e:
            print ("Exception: %s" %(e))

        my_ID = os.getpid() & 0xFFFF

        self.send_ping(sock, my_ID)
        delay = self.receive_pong(sock, my_ID, self.timeout)
        sock.close()
        return delay


    def ping(self):
        """
        Run the ping process
        """
        for i in range(self.count):
            print ("Ping to %s..." % self.target_host,)
            try:
                delay  =  self.ping_once()
            except socket.gaierror as e:
                print ("Ping failed. (socket error: '%s')" % e[1])
                break

            if delay  ==  None:
                print ("Ping failed. (timeout within %ssec.)" % self.timeout)
            else:
                delay  =  delay * 1000
                print ("Get pong in %0.4fms" % delay)



if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Python ping')
    parser.add_argument('host', action="store", help=u'主机名')
    given_args = parser.parse_args()  
    target_host = given_args.host
    pinger = Pinger(target_host=target_host)
    pinger.ping()

参考资料

执行

注意要有root或管理员权限:

# python3 ping.py china-testing.github.io
Ping to china-testing.github.io...
Get pong in 160.7175ms
Ping to china-testing.github.io...
Get pong in 160.8465ms
Ping to china-testing.github.io...
Get pong in 12.0983ms
Ping to china-testing.github.io...
Get pong in 161.3324ms
# python3 ping.py www.so.com
Ping to www.so.com...
Get pong in 29.0303ms
Ping to www.so.com...
Get pong in 28.8599ms
Ping to www.so.com...
Get pong in 28.9860ms
Ping to www.so.com...
Get pong in 29.0167ms
相关文章
|
1天前
|
XML 前端开发 数据格式
BeautifulSoup 是一个 Python 库,用于从 HTML 和 XML 文件中提取数据
BeautifulSoup 是 Python 的一个库,用于解析 HTML 和 XML 文件,即使在格式不规范的情况下也能有效工作。通过创建 BeautifulSoup 对象并使用方法如 find_all 和 get,可以方便地提取和查找文档中的信息。以下是一段示例代码,展示如何安装库、解析 HTML 数据以及打印段落、链接和特定类名的元素。BeautifulSoup 还支持更复杂的查询和文档修改功能。
6 1
|
1天前
|
机器学习/深度学习 自然语言处理 算法
Gensim详细介绍和使用:一个Python文本建模库
Gensim详细介绍和使用:一个Python文本建模库
10 1
|
2天前
|
JSON 数据格式 Python
Python 的 requests 库是一个强大的 HTTP 客户端库,用于发送各种类型的 HTTP 请求
`requests` 库是 Python 中用于HTTP请求的强大工具。要开始使用,需通过 `pip install requests` 进行安装。发送GET请求可使用 `requests.get(url)`,而POST请求则需结合 `json.dumps(data)` 以JSON格式发送数据。PUT和DELETE请求类似,分别调用 `requests.put()` 和 `requests.delete()`。
12 2
|
2天前
|
JSON 数据格式 索引
python之JMESPath:JSON 查询语法库示例详解
python之JMESPath:JSON 查询语法库示例详解
13 0
|
10天前
|
Python
在Python中绘制K线图,可以使用matplotlib和mplfinance库
使用Python的matplotlib和mplfinance库可绘制金融K线图。mplfinance提供便利的绘图功能,示例代码显示如何加载CSV数据(含开盘、最高、最低、收盘价及成交量),并用`mpf.plot()`绘制K线图,设置类型为&#39;candle&#39;,显示移动平均线(mav)和成交量信息。可通过调整参数自定义图表样式,详情参考mplfinance文档。
29 2
|
10天前
|
机器学习/深度学习 边缘计算 TensorFlow
【Python机器学习专栏】Python机器学习工具与库的未来展望
【4月更文挑战第30天】本文探讨了Python在机器学习中的关键角色,重点介绍了Scikit-learn、TensorFlow和PyTorch等流行库。随着技术进步,未来Python机器学习工具将聚焦自动化、智能化、可解释性和可信赖性,并促进跨领域创新,结合云端与边缘计算,为各领域应用带来更高效、可靠的解决方案。
|
11天前
|
监控 网络协议 安全
【亮剑】当设备IP能ping通但无法上网时,可能是DNS解析、网关/路由设置、防火墙限制、网络配置错误或ISP问题
【4月更文挑战第30天】当设备IP能ping通但无法上网时,可能是DNS解析、网关/路由设置、防火墙限制、网络配置错误或ISP问题。解决步骤包括检查网络配置、DNS设置、网关路由、防火墙规则,以及联系ISP。预防措施包括定期备份配置、更新固件、监控网络性能和实施网络安全策略。通过排查和维护,可确保网络稳定和安全。
|
11天前
|
Serverless Python
使用Python的pandas和matplotlib库绘制移动平均线(MA)示例
使用Python的pandas和matplotlib库绘制移动平均线(MA)示例:加载CSV数据,计算5日、10日和20日MA,然后在K线图上绘制。通过`rolling()`计算平均值,`plot()`函数展示图表,`legend()`添加图例。可利用matplotlib参数自定义样式。查阅matplotlib文档以获取更多定制选项。
25 1
|
2天前
|
域名解析 网络协议 Linux
linux网络配置详解
linux网络配置详解
12 0
|
3天前
|
网络协议 Java Linux
【探索Linux】P.29(网络编程套接字 —— 简单的TCP网络程序模拟实现)
【探索Linux】P.29(网络编程套接字 —— 简单的TCP网络程序模拟实现)
10 0