diff --git a/RNS/Utilities/rnprobe.py b/RNS/Utilities/rnprobe.py new file mode 100644 index 0000000..cd3182c --- /dev/null +++ b/RNS/Utilities/rnprobe.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 + +import RNS +import os +import sys +import time +import argparse + +from RNS._version import __version__ + +DEFAULT_PROBE_SIZE = 16 + +def program_setup(configdir, destination_hexhash, size=DEFAULT_PROBE_SIZE, full_name = None): + 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("Destination length is invalid, must be {hex} hexadecimal characters ({byte} bytes).".format(hex=dest_len, byte=dest_len//2)) + 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() + + + reticulum = RNS.Reticulum(configdir = configdir) + + if not RNS.Transport.has_path(destination_hash): + RNS.Transport.request_path(destination_hash) + print("Path to "+RNS.prettyhexrep(destination_hash)+" requested ", end=" ") + sys.stdout.flush() + + i = 0 + syms = "⢄⢂⢁⡁⡈⡐⡠" + while not RNS.Transport.has_path(destination_hash): + time.sleep(0.1) + print(("\b\b"+syms[i]+" "), end="") + sys.stdout.flush() + i = (i+1)%len(syms) + + server_identity = RNS.Identity.recall(destination_hash) + + request_destination = RNS.Destination( + server_identity, + RNS.Destination.OUT, + RNS.Destination.SINGLE, + app_name, + *aspects + ) + + probe = RNS.Packet(request_destination, os.urandom(size)) + receipt = probe.send() + + print("\rSent "+str(size)+" byte probe to "+RNS.prettyhexrep(destination_hash)+" via "+RNS.prettyhexrep(RNS.Transport.next_hop(destination_hash))+" on "+str(RNS.Transport.next_hop_interface(destination_hash))+" ", end=" ") + + i = 0 + while not receipt.status == RNS.PacketReceipt.DELIVERED: + time.sleep(0.1) + print(("\b\b"+syms[i]+" "), end="") + sys.stdout.flush() + i = (i+1)%len(syms) + + print("\b\b ") + sys.stdout.flush() + + hops = str(RNS.Transport.hops_to(destination_hash)) + rtt = receipt.get_rtt() + if (rtt >= 1): + rtt = round(rtt, 3) + rttstring = str(rtt)+" seconds" + else: + rtt = round(rtt*1000, 3) + rttstring = str(rtt)+" milliseconds" + + print( + "Valid reply received from "+ + RNS.prettyhexrep(receipt.destination.hash)+ + ", round-trip time is "+rttstring+ + " over "+hops+" hops" + ) + + + +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( + "--version", + action="version", + version="rnpath {version}".format(version=__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 + ) + + 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, + full_name = args.full_name + ) + + except KeyboardInterrupt: + print("") + exit() + +if __name__ == "__main__": + main() \ No newline at end of file