Simplified Windows' tracert command

Question

I wrote a simplified versions of Windows' tracert command as an exercise and to learn about TTL in packets. It shows the path that your internet traffic will (likely) take on the way to the given destination.

Here's an example (some stuff is redacted out of paranoia):

python tracert.py google.com -h
usage: tracert.py [-h] [-d] [-o O] [-w W] [-t T] ip

positional arguments:
  ip

optional arguments:
  -h, --help  show this help message and exit
  -d          Do not resolve addresses to hostnames.
  -o O        Maximum number of hops to search for target.
  -w W        Wait timeout milliseconds for each reply.
  -t T        How many per packets to send per hop.

Unfortunately, due to using Scapy, the timings seem to be somewhat broken. Sending packets seems to have an inherent overhead, and that overhead dwarfs the actual time. Unless Scapy provides a way to access round-trip times though, I don't think I can fix that without doing away with Scapy and doing the packet construction/sending manually. I mention this since anyone familiar with tracert will likely find the pattern of times odd. They were included mostly as a "semi-functional" placeholder so I could try to implement the incremental behavior of tracert.

I tried to match a lot of tracerts behavior, but I didn't go overboard. I opted to have the target first instead of last for simplicity.

The hardest part was getting displaying incremental as they came in to match tracert. I went through several redesigns of _tracert_hop_row where I tried to return an iterable of results, and the consumer could print results as they were produced. This ended up being a mess though, because I also needed to communicate back whether we had found the destination yet. I ended up printing the result directly instead.

My main concern is _tracert_hop_row. It's a mess, and it's long. Any tips regarding it or anything else would be appreciated.

reverse_dns.py

from typing import Union

from scapy.layers.inet import UDP, IP
from scapy.layers.dns import DNS, DNSQR, DNSRR
from scapy.sendrecv import sr1


_REVERSE_DOMAIN = ".in-addr.arpa"
_OCTET_DELIMITER = "."


def _reverse_dns_packet(ip: str, dns_server: str) -> DNS:
    reversed_ip = _OCTET_DELIMITER.join(reversed(ip.split(_OCTET_DELIMITER)))
    question = DNSQR(qname=f"{reversed_ip}{_REVERSE_DOMAIN}", qtype=12)
    return IP(dst=dns_server) / UDP(dport=53) / DNS(qd=question)


def reverse_dns_lookup(ip: str, dns_server: str, **sr1_kwargs) -> Union[None, int, str]:
    """
    Returns the str host-name if it could be resolved,
      or an int response code if there was an error,
      or None if the DNS server couldn't be reached.
    Unless you know the DNS server will be reachable, setting a timeout via the kwargs is advised.
    """
    request = _reverse_dns_packet(ip, dns_server)
    resp = sr1(request, **sr1_kwargs)
    if resp is None:
        return None
    resp_code = resp[1][DNS].rcode
    if resp_code == 0:
        raw = resp[1][DNSRR].rdata
        return raw.decode("UTF-8")
    else:
        return resp_code

tracert.py

from typing import Tuple, List, Optional, Callable, TypeVar
from collections import Counter
from time import perf_counter
import argparse as ap

from scapy.config import conf
from scapy.layers.inet import IP, ICMP
from scapy.sendrecv import sr1

from reverse_dns import reverse_dns_lookup


_MAX_TTL = 255
_DEFAULT_TIMEOUT = 5
_DEFAULT_VERBOSITY = False
_DEFAULT_TESTS_PER = 3
_DEFAULT_RESOLVE_HOSTNAMES = True
_TABLE_SPACING = 10

NO_INFO_SYM = "*"

T = TypeVar("T")


def _new_trace_packet(destination_ip: str, hop_n: int) -> ICMP:
    return IP(dst=destination_ip, ttl=hop_n) / ICMP()


# Credit: https://stackoverflow.com/a/60845191/3000206
def get_gateway_of(ip: str) -> str:
    return conf.route.route(ip)[2]


def _check_hop_n(destination_ip: str, hop_n: int, **sr1_kwargs) -> Optional[Tuple[str, bool]]:
    """
    Returns a tuple of (hop_ip, destination_reached?)
    """
    sr1_kwargs.setdefault("timeout", _DEFAULT_TIMEOUT)
    sr1_kwargs.setdefault("verbose", _DEFAULT_VERBOSITY)

    packet = _new_trace_packet(destination_ip, hop_n)
    reply = sr1(packet, **sr1_kwargs)
    return reply and (reply[IP].src, reply[ICMP].type == 0)


def _find_proper_route(replies: List[ICMP]) -> Optional[Tuple[str, bool]]:
    if not replies:
        return None
    ip_isfin_pairs = [(resp[IP].src, resp[ICMP].type == 0) for resp in replies]
    found_destination = next((ip for ip, isfin in ip_isfin_pairs if isfin), None)
    selected_ip = found_destination or Counter(ip for ip, _ in ip_isfin_pairs).most_common(1)[0][0]
    return selected_ip, bool(found_destination)


def _time_exec(f: Callable[[], T]) -> Tuple[float, T]:
    """
    Executes the function and returns a tuple of [exec_seconds, result].
    """
    start = perf_counter()
    result = f()
    end = perf_counter()
    return end - start, result


def _cell_print(x: str) -> None:
    print(x.ljust(_TABLE_SPACING), end="", flush=True)


def _tracert_hop_row(destination_ip: str,
                     n_tests: int,
                     hop_n: int,
                     resolve_hostname: bool,
                     **sr_kwargs) -> bool:
    sr_kwargs.setdefault("timeout", _DEFAULT_TIMEOUT)
    sr_kwargs.setdefault("verbose", _DEFAULT_VERBOSITY)
    packet = _new_trace_packet(destination_ip, hop_n)
    replies = []
    for _ in range(n_tests):
        secs, reply = _time_exec(lambda: sr1(packet, **sr_kwargs))
        _cell_print(NO_INFO_SYM if reply is None else f"{int(secs * 1000)} ms")
        if reply:
            replies.append(reply)
    if not replies:
        _cell_print(NO_INFO_SYM)
        return False
    best_route, found_destination = _find_proper_route(replies)
    if resolve_hostname:
        host = reverse_dns_lookup(best_route, get_gateway_of(best_route), **sr_kwargs)
        if isinstance(host, str):
            _cell_print(f"{host} [{best_route}]")
        else:
            _cell_print(best_route)
    else:
        _cell_print(best_route)
    return found_destination


def tracert_internal(destination_ip: str,
                     n_tests_per_hop: int = _DEFAULT_TESTS_PER,
                     resolve_hostnames: bool = _DEFAULT_RESOLVE_HOSTNAMES,
                     max_hops: int = _MAX_TTL,
                     **sr_kwargs) -> None:
    for hop_n in range(1, max_hops + 1):
        _cell_print(str(hop_n))
        found_destination = _tracert_hop_row(destination_ip, n_tests_per_hop, hop_n, resolve_hostnames, **sr_kwargs)
        print()
        if found_destination:
            break


def main():
    parser = ap.ArgumentParser()
    parser.add_argument("ip")
    parser.add_argument("-d", action="store_false", default=True,
                        help="Do not resolve addresses to hostnames.")
    parser.add_argument("-o", type=int, default=_MAX_TTL,
                        help="Maximum number of hops to search for target.")
    parser.add_argument("-w", type=float, default=_DEFAULT_TIMEOUT,
                        help="Wait timeout milliseconds for each reply.")
    parser.add_argument("-t", type=int, default=_DEFAULT_TESTS_PER,
                        help="How many per packets to send per hop.")
    args = parser.parse_args()

    try:
        tracert_internal(args.ip, args.t, args.d, args.o, timeout=args.w)
    except KeyboardInterrupt:
        pass


if __name__ == '__main__':
    main()

Answer 1

if resolve_hostname:
    host = reverse_dns_lookup(best_route, get_gateway_of(best_route), **sr_kwargs)
    if isinstance(host, str):
        _cell_print(f"{host} [{best_route}]")
    else:
        _cell_print(best_route)
else:
    _cell_print(best_route)

You are doing _cell_print and using best_route in each case, which is also the default value. It might be motivated to use a different structure here, such as

output = best_route  #default value
if resolve_hostname:
    host = reversee_dns_lookup(best_route, get_gateway_of(best_route), **sr_kwargs)
    if isinstance(host, str):
        output = f"{host} [{best_route}]"

_cell_print(output)

```

Answer 2

Thanks to @Cukic0d's suggestion, the timing can be easily improved by making use of the .time and .sent_time attributes of the packets.

I removed _time_exec, and rejigged the for loop to be just:

for _ in range(n_tests):
    reply = sr1(packet, **sr_kwargs)
    if reply is None:
        _cell_print(NO_INFO_SYM)
    else:
        _cell_print(f"{int((reply.time - packet.sent_time) * 1000)} ms")
        replies.append(reply)

The timings now roughly match that of tracert.