ping.py

import os
import sys
import time
import socket
import struct
import select

# ICMP code
ICMP_TYPE_ECHO_REQ = 8
ICMP_TYPE_ECHO_REP = 0

# RFC1071 checksum
def inet_cksum(buf: bytes) -> int:
    if len(buf) & 1:
        buf += b"\x00"
    total = 0
    for i in range(0, len(buf), 2):
        total += (buf[i] << 8) + buf[i + 1]
        total = (total & 0xFFFF) + (total >> 16)
    checksum = (~total) & 0xFFFF
    return checksum

# craft ICMP echo request packet
def craft_echo(pid_id: int, seq_no: int, payload_len: int = 32) -> bytes:
    hdr = struct.pack("!BBHHH", ICMP_TYPE_ECHO_REQ, 0, 0, pid_id, seq_no)
    body = struct.pack("!d", time.time()) + b"P" * max(0, payload_len - 8)
    csum = inet_cksum(hdr + body)
    hdr = struct.pack("!BBHHH", ICMP_TYPE_ECHO_REQ, 0, csum, pid_id, seq_no)    # [type, code, checksum, identifier, sequence] + [payload]
    return hdr + body

# receive and validate one echo reply
def await_echo(sock: socket.socket, pid_id: int, seq_no: int, timeout_s: float):
    deadline = time.time() + timeout_s
    while True:
        remaining = deadline - time.time()
        if remaining <= 0:
            return None
        rlist, _, _ = select.select([sock], [], [], remaining)
        if not rlist:
            return None

        t_recv = time.time()
        packet, (src_ip, _) = sock.recvfrom(65535)

        # IPv4 header
        iphdr = packet[:20]
        _ver_ihl, _tos, _len, _id, _flags_frag, ttl, proto, _sum, src_raw, _dst_raw = \
            struct.unpack("!BBHHHBBHII", iphdr)

        # ICMP header
        icmph = packet[20:28]
        icmp_type, icmp_code, _icmp_sum, recv_id, recv_seq = struct.unpack("!BBHHH", icmph)

        # accept only our echo reply
        if icmp_type == ICMP_TYPE_ECHO_REP and recv_id == pid_id and recv_seq == seq_no:
            t_send = struct.unpack("!d", packet[28:36])[0]  # timestamp from payload
            rtt_ms = (t_recv - t_send) * 1000.0
            data_sz = len(packet) - 20
            return (rtt_ms, ttl, data_sz, src_ip)

# ping loop
def run_ping(target_host: str, probes: int = 6, per_timeout: float = 1.0, payload_len: int = 32):
    try:
        dst_ip = socket.gethostbyname(target_host)
    except socket.gaierror as e:
        print(f"DNS resolution failed for {target_host}: {e}")
        return 2

    icmp_proto = socket.getprotobyname("icmp")
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp_proto)
    except PermissionError:
        print("ERROR: run as Administrator/root (raw sockets required).")
        return 2

    pid_id = os.getpid() & 0xFFFF
    tx_count = rx_count = 0
    samples_ms = []

    print("ping destination IP addresses in Python")
    print(f"PING {target_host} ({dst_ip}): {payload_len} data bytes")

    for seq_no in range(probes):
        pkt = craft_echo(pid_id, seq_no, payload_len)
        t0 = time.time()
        sock.sendto(pkt, (dst_ip, 0))
        tx_count += 1

        result = await_echo(sock, pid_id, seq_no, per_timeout)
        if result is None:
            print(f"Request timed out for icmp_seq={seq_no}")
        else:
            rtt_ms, ttl, nbytes, src = result
            rx_count += 1
            samples_ms.append(rtt_ms)
            print(f"{nbytes} bytes from {src}: ICMP_seq = {seq_no}, TTL = {ttl}, time = {rtt_ms:.3f}ms")

        # probe time
        pause = 1.0 - (time.time() - t0)
        if pause > 0:
            time.sleep(pause)

    # ping & RTT stats
    loss_pct = (1 - rx_count / tx_count) * 100 if tx_count else 0.0
    print(f"\n{target_host} ping statistics:")
    print(f"{tx_count} packets transmitted, {rx_count} received, {loss_pct:.1f}% packet loss")
    if samples_ms:
        mn, avg, mx = min(samples_ms), sum(samples_ms) / len(samples_ms), max(samples_ms)
        print(f"RTT min: {mn:.3f}ms, avg: {avg:.3f}ms, max: {mx:.3f}ms")

    sock.close()
    return 0 if rx_count else 1

# continue?
def ask_continue(prompt="Another ping? (y/n): "):
    while True:
        ans = input(prompt).strip().lower()
        if ans in ("y"):
            return True
        elif ans in ("n"):
            return False
        else:
            print("Invalid input. Please enter 'y' or 'n'.")


# main function
if __name__ == "__main__":
    first_arg = sys.argv[1] if len(sys.argv) > 1 else None
    while True:
        host = first_arg or input("\nEnter a host to ping: ").strip()
        first_arg = None
        run_ping(host)
        if not ask_continue("Another ping? (y/n): "):
            print("Cya")
            break