coding: utf-8
import socket import struct import time
ICMP_ECHO_REQUEST = 8 # 设定Ping类型为8
def checksum(data): n = len(data) m = n % 2 sum = 0 for i in range(0, n - m, 2): sum += ord(chr(data[i])) + (ord(chr(data[i + 1])) << 8) if m: sum += ord(data[-1]) sum = (sum >> 16) + (sum & 0xffff) sum += (sum >> 16) answer = ~sum & 0xffff answer = answer >> 8 | (answer << 8 & 0xff00) return answer
def receiveOnePing(mySocket, ID, timeout): while 1: timeReceived = time.time() recPacket, addr = mySocket.recvfrom(1024) header = recPacket[20: 28] type, code, checksum, packetID, sequence = struct.unpack(“!bbHHh”, header)
if type == 0 and packetID == ID:
byte_in_double = struct.calcsize("!d")
timeSent = struct.unpack("!d", recPacket[28: 28 + byte_in_double])[0]
delay = timeReceived - timeSent
ttl = ord(struct.unpack("!c", recPacket[8:9])[0].decode())
if timeout <= delay:
return None
return delay, ttl, byte_in_double
def doOnePing(destAddr, ID, sequence, timeout): # ********* Begin *********# # 创建ICMP协议的socket对象 mySocket = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) # ********* End *********# myChecksum = 0 header = struct.pack(“!bbHHh”, ICMP_ECHO_REQUEST, 0, myChecksum, ID, sequence) data = struct.pack(“!d”, time.time()) myChecksum = checksum(header + data) header = struct.pack(“!bbHHh”, ICMP_ECHO_REQUEST, 0, myChecksum, ID, sequence) packet = header + data # ********* Begin *********# # 生成校验和,并生成ping数据报 myChecksum = 0 header = struct.pack(“!bbHHh”, ICMP_ECHO_REQUEST, 0, myChecksum, ID, sequence) data = struct.pack(“!d”, time.time()) myChecksum = checksum(header + data) header = struct.pack(“!bbHHh”, ICMP_ECHO_REQUEST, 0, myChecksum, ID, sequence) packet = header + data # ********* End *********#
# ********* Begin *********#
# 发送该数据包
mySocket.sendto(packet, (destAddr, 12345))
# ********* End *********#
ans = receiveOnePing(mySocket, ID, timeout)
# ********* Begin *********#
# 关闭socket对象
mySocket.close()
# ********* End *********#
return ans
def ping(host, timeout=1): # ********* Begin *********# # 获取地址,并输出发送信息,参考预期输出的第一行 dest = socket.gethostbyname(host) print(“Pinging " + dest + " using Python:”) # ********* End *********#
myID = 2468
loss = 0 # 用来统计丢包数量
for i in range(5):
result = doOnePing(dest, myID, i, timeout)
# ********* Begin *********#
# 如果没有收到pong报文,输出 Request timed out.
# 及时记录丢包数量
if not result:
print("Request timed out.")
loss += 1
# ********* End *********#
# ********* Begin *********#
# 收到pong报文,获取并输出相应的信息,具体输出格式参考预期输出
else:
delay = int(result[0] * 1000)
ttl = result[1]
bytes = result[2]
print("Received from " + dest + ": byte(s)=" + str(bytes) + " delay=" + str(delay) + "ms TTL=" + str(ttl))
# ********* End *********#
time.sleep(1)
# ********* Begin *********#
# 输出统计的信息,包括发送、接收以及丢失数据包的个数,具体输出格式参考预期输出
print("Packet: sent = " + str(5) + " received = " + str(5-loss) + " lost = " + str(loss))
# ********* End *********#
return
ping(“localhost”)
|