Python中的多个ping脚本

Posted 2023-02-16

【中文标题】Python中的多个ping脚本

【英文标题】：Multiple ping script in Python

【发布时间】：2012-08-19 12:52:07

【问题描述】：

我找不到任何关于 python 和网络的易于学习的好文档。在这种情况下，我只是想制作一个简单的脚本，我可以 ping 一些远程机器。

for ping in range(1,10):
   ip="127.0.0."+str(ping)
   os.system("ping -c 3 %s" % ip)

这样的简单脚本可以很好地 ping 机器，但我想让脚本返回“活动”“无响应”，这让我觉得我也必须查找时间模块，我想想time.sleep(5)，然后会有一个break语句。这让我觉得 for 里面应该有一个 while 循环。我不能 100% 确定，我可能完全走错了方向：/ 如果有人可以帮助或指出一些很棒的文档的方向。

【问题讨论】：

我不知道你为什么认为你需要时间模块？我建议研究如何从 subprocess 解析 STDOUT（这是你应该使用的，而不是 os.system()）

试试Scapy。

这里是一个例子github.com/lbaby/javalearn/blob/master/shell/ppe.py

不，不要使用 scapy，scapy 对于任何需要吞吐量或可靠性的东西都很糟糕......监控应用程序都需要这两者。

【参考方案1】：

试试subprocess.call。它保存了所使用程序的返回值。

根据我的 ping 手册，成功时返回 0，发送 ping 但未收到回复时返回 2，任何其他值表示错误。

# typo error in import
import subprocess

for ping in range(1,10):
    address = "127.0.0." + str(ping)
    res = subprocess.call(['ping', '-c', '3', address])
    if res == 0:
        print "ping to", address, "OK"
    elif res == 2:
        print "no response from", address
    else:
        print "ping to", address, "failed!"

【讨论】：

不客气。请记住将帮助您的答案标记为已接受（单击问题投票数下方的绿色标记），它会向其他访问者表明答案是有帮助的。 :-)

subprocess.call() 一次只 ping 一个 ip。同时ping多个ipsubprocess.Popen() could be used

【参考方案2】：

这个脚本：

import subprocess
import os
with open(os.devnull, "wb") as limbo:
        for n in xrange(1, 10):
                ip="192.168.0.0".format(n)
                result=subprocess.Popen(["ping", "-c", "1", "-n", "-W", "2", ip],
                        stdout=limbo, stderr=limbo).wait()
                if result:
                        print ip, "inactive"
                else:
                        print ip, "active"

会产生类似这样的输出：

192.168.0.1 active
192.168.0.2 active
192.168.0.3 inactive
192.168.0.4 inactive
192.168.0.5 inactive
192.168.0.6 inactive
192.168.0.7 active
192.168.0.8 inactive
192.168.0.9 inactive

如果将limbo 替换为subprocess.PIPE 并在Popen 对象上使用communicate()，则可以捕获输出：

p=Popen( ... )
output=p.communicate()
result=p.wait()

这样就可以得到命令的返回值，并且可以捕获文本。在manual 之后，如果您需要灵活性，这是操作子流程的首选方式：

本模块中的底层流程创建和管理是 由 Popen 类处理。它提供了很大的灵活性，因此 开发人员能够处理未涵盖的不太常见的情况 便利功能。

【讨论】：

请在评论中说明您投反对票的原因，以便改进答案。

【参考方案3】：

非常感谢您。我已对其进行了修改以与 Windows 一起使用。我还设置了一个低超时，所以没有返回的 IP 不会坐等 5 秒。这是来自 hochl 源代码。

import subprocess
import os
with open(os.devnull, "wb") as limbo:
        for n in xrange(200, 240):
                ip="10.2.7.0".format(n)
                result=subprocess.Popen(["ping", "-n", "1", "-w", "200", ip],
                        stdout=limbo, stderr=limbo).wait()
                if result:
                        print ip, "inactive"
                else:
                        print ip, "active"

只需更改方案的 ip= 和主机的 xrange。

【参考方案4】：

我是一个初学者，写了一个脚本来 ping 多个主机。要 ping 多个主机，您可以使用 ipaddress 模块。

import ipaddress
from subprocess import Popen, PIPE

net4 = ipaddress.ip_network('192.168.2.0/24')
for x in net4.hosts():
    x = str(x)
    hostup = Popen(["ping", "-c1", x], stdout=PIPE)
    output = hostup.communicate()[0]
    val1 = hostup.returncode
 if val1 == 0:
    print(x, "is pinging")
 else:
    print(x, "is not responding")

【参考方案5】：

要同时 ping 多个主机，您可以使用 subprocess.Popen()：

#!/usr/bin/env python3
import os
import time
from subprocess import Popen, DEVNULL

p =  # ip -> process
for n in range(1, 100): # start ping processes
    ip = "127.0.0.%d" % n
    p[ip] = Popen(['ping', '-n', '-w5', '-c3', ip], stdout=DEVNULL)
    #NOTE: you could set stderr=subprocess.STDOUT to ignore stderr also

while p:
    for ip, proc in p.items():
        if proc.poll() is not None: # ping finished
            del p[ip] # remove from the process list
            if proc.returncode == 0:
                print('%s active' % ip)
            elif proc.returncode == 1:
                print('%s no response' % ip)
            else:
                print('%s error' % ip)
            break

如果您可以以 root 身份运行，则可以使用 pure Python ping script 或 scapy：

from scapy.all import sr, ICMP, IP, L3RawSocket, conf

conf.L3socket = L3RawSocket # for loopback interface
ans, unans = sr(IP(dst="127.0.0.1-99")/ICMP(), verbose=0) # make requests
ans.summary(lambda (s,r): r.sprintf("%IP.src% is alive"))

【参考方案6】：

import subprocess
import os
'''
servers.txt contains ip address in following format
192.168.1.1
192.168.1.2
'''
    with open('servers.txt', 'r') as f:
        for ip in f:
            result=subprocess.Popen(["ping", "-c", "1", "-n", "-W", "2",    ip],stdout=f, stderr=f).wait()
            if result:
                print(ip, "inactive")
            else:
                print(ip, "active")

【参考方案7】：

Python 实际上有一个非常可爱的method，它将“返回网络中可用主机的迭代器”。 （将 strict 设置为 false 会遍历所有 IP）

例如：

import subprocess
import ipaddress

subnet = ipaddress.ip_network('192.168.1.0/24', strict=False)
for i in subnet.hosts():
    i = str(i)
    subprocess.call(["ping", "-c1", "-n", "-i0.1", "-W1", i])

等待间隔 (-i0.1) 可能对自动化很重要，即使是一秒超时 (-t1) 也可能永远超过 0.0/24

编辑： 因此，为了跟踪 ICMP（ping）请求，我们可以这样做：

#!/usr/bin/env python

import subprocess
import ipaddress

alive = []
subnet = ipaddress.ip_network('192.168.1.0/23', strict=False)
for i in subnet.hosts():
    i = str(i)
    retval = subprocess.call(["ping", "-c1", "-n", "-i0.1", "-W1", i])
    if retval == 0:
        alive.append(i)
for ip in alive:
    print(ip + " is alive") 

这将返回如下内容：

192.168.0.1 is alive
192.168.0.2 is alive
192.168.1.1 is alive
192.168.1.246 is alive

即所有响应 ICMP 的 IP 都在整个 /23 范围内——非常酷！

【讨论】：

仅供参考，在将 IP 地址指定为 unicode 之前，我对此抛出了错误：subnet = ipaddress.ip_network(u'192.168.1.0/23', strict=False)

【参考方案8】：

import subprocess,os,threading,time
from queue import Queue
lock=threading.Lock()
_start=time.time()
def check(n):
    with open(os.devnull, "wb") as limbo:
                ip="192.168.21.0".format(n)
                result=subprocess.Popen(["ping", "-n", "1", "-w", "300", ip],stdout=limbo, stderr=limbo).wait()
                with lock:                    
                    if not result:
                        print (ip, "active")                    
                    else:
                        pass                        

def threader():
    while True:
        worker=q.get()
        check(worker)
        q.task_done()
q=Queue()

for x in range(255):
    t=threading.Thread(target=threader)
    t.daemon=True
    t.start()
for worker in range(1,255):
    q.put(worker)
q.join()
print("Process completed in: ",time.time()-_start)

我认为这会更好。

【讨论】：

谢谢，它对我有用。避免错误需要很少的机会： import Queue #in 2nd line q = Queue.Queue() #in 21st line

【参考方案9】：

我在上面的代码中用python 2.7中的多线程做了一些修改：

import subprocess,os,threading,time
import Queue

lock=threading.Lock()
_start=time.time()
def check(n):
    with open(os.devnull, "wb") as limbo:
                ip=n
                result=subprocess.Popen(["ping", "-n", "2", "-w", "300", ip],stdout=limbo, stderr=limbo).wait()
                with lock:
                    if not result:
                        print ip, "active"
                    else:
                        print ip, "Inactive"

def threader():
    while True:
        worker=q.get()
        check(worker)
        q.task_done()
q = Queue.Queue()

for x in range(255):
    t=threading.Thread(target=threader)
    t.daemon=True
    t.start()

ip = ["13.45.23.523", "13.35.23.523","23.23.56.346"]
for worker in ip:
    q.put(worker)
q.join()
print("Process completed in: ",time.time()-_start)