使用Python实现网络探测（Ping）功能
本文展示了一个使用Python实现类似Ping功能的尝试。原始代码试图通过scapy库构造ICMP包来探测主机存活，但遇到了问题：尽管系统中原生的ping命令可以正常工作，Python版本的实现却始终无法收到响应。下面分析代码、错误原因，并给出替代方案。
基于Scapy的ICMP探测实现
import threading
import time
from random import randint
from scapy.all import IP, TCP, ICMP, sr1
class NetworkProber(threading.Thread):
def __init__(self, target_ip: str):
super().__init__()
self.daemon = True
self.active = None
self.ip = target_ip
self._stop_flag = threading.Event()
self.start()
@staticmethod
def icmp_check(host: str) -> bool:
result = False
# 生成随机标识，避免冲突
ip_id = randint(1, 65535) # IP层标识
ping_id = randint(1, 65535) # ICMP标识
ping_seq = randint(1, 65535) # 序列号
# 构造完整的ICMP请求包
packet = IP(dst=host, ttl=64, id=ip_id) / ICMP(id=ping_id, seq=ping_seq) / b'Hello from Python'
print(packet)
# 发送并等待响应，超时3秒
reply = sr1(packet, timeout=3, verbose=False)
if reply:
src = reply[IP].src
if src == host:
# 检查响应类别
if reply[1].listname == 'Unanswered':
print(f"No response received from {host}")
else:
print(f"Reply received from {host}")
result = True
else:
print(f"No response from {host}")
return result
def run(self):
print(f"Starting probe for {self.ip}...")
while not self._stop_flag.is_set():
time.sleep(6)
NetworkProber.icmp_check(self.ip)
def stop(self):
print("Stopping probe.")
self.active = False
self._stop_flag.set()
self.join()
运行日志与问题分析
ping 172.20.3.34 start...
running count No.1
...
WARNING: MAC address to reach destination not found. Using broadcast.
...
Ping:没有回应: 172.20.3.34
关键问题在于日志中的警告信息："MAC address to reach destination not found. Using broadcast."。这说明scapy未能通过ARP协议解析目标IP对应的MAC地址，转而使用广播方式发送。但广播的ICMP请求很可能被防火墙或路由器丢弃，导致无法收到响应。
对比之下，系统原生的ping命令能正常工作，而Python的scapy版本却失败，通常由以下原因造成：
- 目标主机或中间网络设备过滤了来自Python发送的ICMP包（例如检测到非标准payload）。
- scapy未能正确解析ARP表（可能需要手动绑定或等待ARP缓存更新）。
- 随机生成的ICMP标识或序列号与系统已有冲突，被目标忽略。
尝试ARP探测方案
以下代码试图通过ARP请求来替代ICMP探测，但同样未能成功：
class ARPProber(threading.Thread):
def __init__(self, target_ip: str):
super().__init__()
self.daemon = True
self.active = None
self.ip = target_ip
self._stop_flag = threading.Event()
self.start()
@staticmethod
def arp_check(host: str) -> bool:
result = False
# 构造ARP请求
arp_request = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=host + "/24")
reply = srp(arp_request, timeout=2)
if reply:
ans, unans = reply[0], reply[1]
print(f"ARP response: {ans} {unans}")
if unans.listname == "Unanswered":
print(f"ARP: No reply from {host}")
else:
print(f"ARP: Reply from {host}")
else:
print(f"ARP: No response from {host}")
return result
def run(self):
print(f"Starting ARP probe for {self.ip}...")
while not self._stop_flag.is_set():
time.sleep(6)
ARPProber.arp_check(self.ip)
def stop(self):
print("Stopping ARP probe.")
self.active = False
self._stop_flag.set()
self.join()
可行的解决方案：调用系统Ping命令
当直接构造网络包不可行时，最简单的方案是封装系统命令：
import subprocess
import threading
import time
class SystemPing(threading.Thread):
def __init__(self, target_ip: str):
super().__init__()
self.daemon = True
self.active = None
self.ip = target_ip
self._stop_flag = threading.Event()
self.start()
@staticmethod
def ping_host(host: str) -> bool:
# Windows使用 -n，Linux/macOS使用 -c
import platform
param = '-n' if platform.system().lower() == 'windows' else '-c'
cmd = ['ping', param, '4', host]
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
output = result.stdout
# 检查返回结果
if "TTL" in output or "ttl" in output or "Reply from" in output:
print(f"Ping successful for {host}")
return True
else:
print(f"Ping failed for {host}")
return False
def run(self):
print(f"Starting system ping for {self.ip}...")
while not self._stop_flag.is_set():
time.sleep(6)
SystemPing.ping_host(self.ip)
def stop(self):
print("Stopping system ping.")
self.active = False
self._stop_flag.set()
self.join()
此方案直接利用操作系统内核的Ping实现，避免了scapy在底层网络栈上的兼容性问题。如果需要更细粒度的控制（如自定义超时、包大小），可以通过subprocess传递额外的命令行参数。
总结
Python中使用scapy实现Ping功能时，常因ARP解析问题或ICMP包过滤导致失败。通过调用系统原生的ping命令可确保可靠性，同时保持代码简洁。若坚持使用scapy，需确保网络环境允许发送和接收原始ICMP包，并提前处理好ARP缓存（例如先执行一次系统ping以填充ARP表）。
