Source code for duct.protocol.icmp

"""
.. module:: icmp
   :synopsis: Native ICMP protocol implementation

.. moduleauthor:: Colin Alston <colin@tamvera.com>
"""
import socket
import time
import fcntl
import random
import struct
import asyncio


[docs] class IP(object): """IP header decoder""" def __init__(self, packet): self.readPacket(packet)
[docs] def readPacket(self, packet): """Decode an IP packet header""" vl = struct.unpack('!b', packet[0:1])[0] l = (vl & 0xf) * 4 self.offset = struct.unpack('!H', packet[6:8]) self.payload = packet[l:]
[docs] class EchoPacket(object): """ICMP Echo packet encoder and decoder""" def __init__(self, seq=0, eid=None, data=None, packet=None): if packet: self.decodePacket(packet) self.packet = packet else: self.eid = eid self.seq = seq self.data = data self.encodePacket()
[docs] def calculateChecksum(self, buf): """Calculate ICMP checksum over buf""" if isinstance(buf, str): buf = buf.encode('latin-1') nleft = len(buf) chksum = 0 pos = 0 while nleft > 1: chksum = buf[pos] * 256 + (buf[pos + 1] + chksum) pos += 2 nleft -= 2 if nleft == 1: chksum = chksum + buf[pos] * 256 chksum = (chksum >> 16) + (chksum & 0xFFFF) chksum += (chksum >> 16) return (~chksum) & 0xFFFF
[docs] def encodePacket(self): """Encode an ICMP echo request packet""" head = struct.pack('!bb', 8, 0) echo = struct.pack('!HH', self.seq, self.eid) if isinstance(self.data, str): data_bytes = self.data.encode('latin-1') else: data_bytes = self.data chk = self.calculateChecksum(head + b'\x00\x00' + echo + data_bytes) chk = struct.pack('!H', chk) self.packet = head + chk + echo + data_bytes
[docs] def decodePacket(self, packet): """Decode an ICMP echo reply packet""" self.icmp_type, self.code, self.chk, self.seq, self.eid = struct.unpack( '!bbHHH', packet[:8]) self.data = packet[8:] rc = packet[:2] + b'\x00\x00' + packet[4:] mychk = self.calculateChecksum(rc) self.valid = mychk == self.chk
def __repr__(self): return (f"<type={self.icmp_type} code={self.code} chk={self.chk}" f" seq={self.seq} data={len(self.data)} valid={self.valid}>")
class _ICMPPinger: """Internal state machine for an async ICMP ping""" def __init__(self, sock, dst, count, inter, maxwait, size, loop): self.sock = sock self.dst = dst self.count = count self.inter = inter self.maxwait = maxwait self.size = size - 36 self.loop = loop self.seq = 0 self.start = 0 self.id_base = random.randint(0, 40000) self.recv = [] self._done = loop.create_future() self._send_handle = None def _data_received(self): try: datagram, _ = self.sock.recvfrom(4096) except BlockingIOError: return now = int(time.time() * 1000000) packet = IP(datagram) icmp = EchoPacket(packet=packet.payload) if icmp.valid and icmp.code == 0 and icmp.icmp_type == 0: if (icmp.eid - icmp.seq) == self.id_base: ts = icmp.data[:8] delta = (now - struct.unpack('!Q', ts)[0]) / 1000.0 self.maxwait = (self.maxwait + delta) / 2.0 self.recv.append((icmp.seq, delta)) def _create_data(self, n): s = b'' c = 33 for _ in range(n): s += bytes([c]) c = c + 1 if c < 126 else 33 return s def _send_echo(self): data = struct.pack('!Q', int(time.time() * 1000000)) data += self._create_data(self.size) pkt = EchoPacket(seq=self.seq, eid=self.id_base + self.seq, data=data) self.sock.sendto(pkt.packet, (self.dst, 0)) self.seq += 1 if self.seq < self.count: self._send_handle = self.loop.call_later( self.inter, self._send_echo) else: tdelay = (self.maxwait * self.count) / 1000.0 elapsed = time.time() - self.start remaining = max(tdelay - elapsed, 0.05) self.loop.call_later(remaining, self._finish) def _finish(self): r = len(self.recv) loss = int(100 * (self.count - r) / float(self.count)) avg_latency = sum(d for _, d in self.recv) / float(r) if r else None if not self._done.done(): self._done.set_result((loss, avg_latency)) async def run(self): """Run the ping and return (packet_loss_percent, avg_latency_ms).""" self.sock.connect((self.dst, random.randint(33434, 33534))) self.start = time.time() self.loop.add_reader(self.sock.fileno(), self._data_received) try: self._send_echo() return await asyncio.wait_for( self._done, timeout=(self.maxwait * self.count / 1000.0) + 2.0 ) except asyncio.TimeoutError: r = len(self.recv) loss = int(100 * (self.count - r) / float(self.count)) avg_latency = sum(d for _, d in self.recv) / float(r) if r else None return (loss, avg_latency) finally: self.loop.remove_reader(self.sock.fileno()) if self._send_handle: self._send_handle.cancel()
[docs] async def ping(dst, count, inter=0.2, maxwait=1000, size=64): """Send ICMP echo requests to `dst` `count` times. Returns (packet_loss_percent, avg_latency_ms) or (100, None) on failure. Requires CAP_NET_RAW or root privileges. """ loop = asyncio.get_event_loop() sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) sock.setblocking(False) fd = sock.fileno() flags = fcntl.fcntl(fd, fcntl.F_GETFD) fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC) try: pinger = _ICMPPinger(sock, dst, count, inter, maxwait, size, loop) return await pinger.run() finally: sock.close()