Reticulum/RNS/Utilities/rnprobe.py

#!/usr/bin/env python3

# MIT License
#
# Copyright (c) 2016-2022 Mark Qvist / unsigned.io
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import RNS
import os
import sys
import time
import argparse

from RNS._version import __version__

DEFAULT_PROBE_SIZE = 16
DEFAULT_TIMEOUT = 12

def program_setup(configdir, destination_hexhash, size=None, full_name = None, verbosity = 0, timeout=None, wait=0, probes=1):
    if size == None: size = DEFAULT_PROBE_SIZE
    if full_name == None:
        print("The full destination name including application name aspects must be specified for the destination")
        exit()

    try:
        app_name, aspects = RNS.Destination.app_and_aspects_from_name(full_name)

    except Exception as e:
        print(str(e))
        exit()

    try:
        dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2
        if len(destination_hexhash) != dest_len:
            raise ValueError(f"Destination length is invalid, must be {dest_len} hexadecimal characters ({dest_len // 2} bytes).")
        try:
            destination_hash = bytes.fromhex(destination_hexhash)
        except Exception as e:
            raise ValueError("Invalid destination entered. Check your input.")
    except Exception as e:
        print(str(e))
        exit()

    if verbosity > 0:
        more_output = True
        verbosity -= 1
    else:
        more_output = False
        verbosity -= 1


    reticulum = RNS.Reticulum(configdir = configdir, loglevel = 3+verbosity)

    if not RNS.Transport.has_path(destination_hash):
        RNS.Transport.request_path(destination_hash)
        print(f"Path to {RNS.prettyhexrep(destination_hash)} requested  ", end=" ")
        sys.stdout.flush()

    _timeout = time.time() + (timeout or DEFAULT_TIMEOUT+reticulum.get_first_hop_timeout(destination_hash))
    i = 0
    syms = "⢄⢂⢁⡁⡈⡐⡠"
    while not RNS.Transport.has_path(destination_hash) and not time.time() > _timeout:
        time.sleep(0.1)
        print(f'\x08\x08{syms[i]} ', end="")
        sys.stdout.flush()
        i = (i+1)%len(syms)

    if time.time() > _timeout:
        print("\r                                                          \rPath request timed out")
        exit(1)

    server_identity = RNS.Identity.recall(destination_hash)

    request_destination = RNS.Destination(
        server_identity,
        RNS.Destination.OUT,
        RNS.Destination.SINGLE,
        app_name,
        *aspects
    )

    sent = 0
    replies = 0
    while probes:

        if sent > 0:
            time.sleep(wait)

        try:
            probe = RNS.Packet(request_destination, os.urandom(size))
            probe.pack()
        except OSError:
            print(f"Error: Probe packet size of {len(probe.raw)} bytes exceed MTU of {RNS.Reticulum.MTU} bytes")
            exit(3)

        receipt = probe.send()
        sent += 1

        if more_output:
            nhd = reticulum.get_next_hop(destination_hash)
            via_str = f" via {RNS.prettyhexrep(nhd)}" if nhd != None else ""
            if_str  = f" on {reticulum.get_next_hop_if_name(destination_hash)}" if reticulum.get_next_hop_if_name(destination_hash) != "None" else ""
            more = via_str+if_str
        else:
            more = ""

        print(f'\rSent probe {sent} ({size} bytes) to {RNS.prettyhexrep(destination_hash)}{more}  ', end=" ")

        _timeout = time.time() + (timeout or DEFAULT_TIMEOUT+reticulum.get_first_hop_timeout(destination_hash))
        i = 0
        while receipt.status == RNS.PacketReceipt.SENT and not time.time() > _timeout:
            time.sleep(0.1)
            print(f'\x08\x08{syms[i]} ', end="")
            sys.stdout.flush()
            i = (i+1)%len(syms)

        if time.time() > _timeout:
            print("\r                                                                \rProbe timed out")

        else:
            print("\b\b ")
            sys.stdout.flush()

            if receipt.status == RNS.PacketReceipt.DELIVERED:
                replies += 1
                hops = RNS.Transport.hops_to(destination_hash)
                if hops != 1:
                    ms = "s"
                else:
                    ms = ""

                rtt = receipt.get_rtt()
                if (rtt >= 1):
                    rtt = round(rtt, 3)
                    rttstring = f"{rtt} seconds"
                else:
                    rtt = round(rtt*1000, 3)
                    rttstring = f"{rtt} milliseconds"

                reception_stats = ""
                if reticulum.is_connected_to_shared_instance:
                    reception_rssi = reticulum.get_packet_rssi(receipt.proof_packet.packet_hash)
                    reception_snr  = reticulum.get_packet_snr(receipt.proof_packet.packet_hash)
                    reception_q    = reticulum.get_packet_q(receipt.proof_packet.packet_hash)

                    if reception_rssi != None:
                        reception_stats += f" [RSSI {reception_rssi} dBm]"

                    if reception_snr != None:
                        reception_stats += f" [SNR {reception_snr} dB]"

                    if reception_q != None:
                        reception_stats += f" [Link Quality {reception_q}%]"

                else:
                    if receipt.proof_packet != None:
                        if receipt.proof_packet.rssi != None:
                            reception_stats += f" [RSSI {receipt.proof_packet.rssi} dBm]"

                        if receipt.proof_packet.snr != None:
                            reception_stats += f" [SNR {receipt.proof_packet.snr} dB]"

                print(
                    f"Valid reply from {RNS.prettyhexrep(receipt.destination.hash)}\nRound-trip time is {rttstring} over {hops} hop{ms}{reception_stats}\n"
                )

            else:
                print("\r                                                          \rProbe timed out")

        probes -= 1

    loss = round((1-(replies/sent))*100, 2)
    print(f"Sent {sent}, received {replies}, packet loss {loss}%")
    if loss > 0:
        exit(2)
    else:
        exit(0)


def main():
    try:
        parser = argparse.ArgumentParser(description="Reticulum Probe Utility")

        parser.add_argument("--config", action="store", default=None, help="path to alternative Reticulum config directory", type=str)
        parser.add_argument("-s", "--size", action="store", default=None, help="size of probe packet payload in bytes", type=int)
        parser.add_argument("-n", "--probes", action="store", default=1, help="number of probes to send", type=int)
        parser.add_argument("-t", "--timeout", metavar="seconds", action="store", default=None, help="timeout before giving up", type=float)
        parser.add_argument("-w", "--wait", metavar="seconds", action="store", default=0, help="time between each probe", type=float)
        parser.add_argument("--version", action="version", version=f"rnprobe {__version__}")
        parser.add_argument("full_name", nargs="?", default=None, help="full destination name in dotted notation", type=str)
        parser.add_argument("destination_hash", nargs="?", default=None, help="hexadecimal hash of the destination", type=str)

        parser.add_argument('-v', '--verbose', action='count', default=0)

        args = parser.parse_args()

        if args.config:
            configarg = args.config
        else:
            configarg = None

        if not args.destination_hash:
            print("")
            parser.print_help()
            print("")
        else:
            program_setup(
                configdir = configarg,
                destination_hexhash = args.destination_hash,
                size = args.size,
                full_name = args.full_name,
                verbosity = args.verbose,
                probes = args.probes,
                wait = args.wait,
                timeout = args.timeout,
            )

    except KeyboardInterrupt:
        print("")
        exit()

if __name__ == "__main__":
    main()