ping_sweep.py

#!/usr/bin/env python3
import argparse
import concurrent.futures as cf
import ipaddress
import platform
import subprocess
from typing import Iterable

def is_pingable(host: str, timeout: float = 1.0) -> bool:
    """
    Returns True if host responds to a single ICMP echo.
    Note: Firewalls may drop ICMP; non-response != definitely down.
    """
    system = platform.system()
    try:
        if system == "Windows":
            # -n 1 = one echo, -w timeout(ms)
            cmd = ["ping", "-n", "1", "-w", str(int(timeout * 1000)), host]
        elif system == "Linux":
            # -c 1 = one echo, -W timeout(s)
            cmd = ["ping", "-c", "1", "-W", str(int(timeout)), host]
        else:
            # Fallback (macOS/others): one echo; timeout handling varies by OS
            cmd = ["ping", "-c", "1", host]

        result = subprocess.run(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False
        )
        return result.returncode == 0
    except Exception:
        return False

def iterate_targets(target: str) -> Iterable[str]:
    """
    If target is a CIDR, iterate usable hosts; else just yield the single IP.
    """
    try:
        net = ipaddress.ip_network(target, strict=False)
        if isinstance(net, ipaddress.IPv4Network):
            for host in net.hosts():  # skips network & broadcast
                yield str(host)
        else:
            # Avoid IPv6 sweeping (enormous space); just yield if it's a /128
            if net.prefixlen == 128:
                yield str(net.network_address)
            else:
                raise ValueError("IPv6 subnet sweeping is impractical. Use single IPv6 addresses.")
    except ValueError:
        # Not a network — treat as single IP (validity left to ping)
        yield target

def main():
    parser = argparse.ArgumentParser(description="Ping a host or sweep a subnet.")
    parser.add_argument("target", help="IP or CIDR (e.g., 10.10.71.0/24)")
    parser.add_argument("--timeout", type=float, default=1.0, help="Per-host timeout (seconds)")
    parser.add_argument("--workers", type=int, default=128, help="Max concurrent pings")
    parser.add_argument("--only-up", action="store_true", help="Print only responsive hosts")
    args = parser.parse_args()

    targets = list(iterate_targets(args.target))
    up = []

    with cf.ThreadPoolExecutor(max_workers=args.workers) as ex:
        futures = {ex.submit(is_pingable, t, args.timeout): t for t in targets}
        for fut in cf.as_completed(futures):
            t = futures[fut]
            alive = fut.result()
            if args.only_up:
                if alive:
                    print(t)
            else:
                print(f"{t:>15}  {'UP' if alive else 'DOWN'}")
            if alive:
                up.append(t)

    print(f"\nSummary: {len(up)}/{len(targets)} hosts responded.")

if __name__ == "__main__":
    main()