Reticulum/RNS/Utilities/rnprobe.py

214 lines
6.4 KiB
Python
Raw Normal View History

2021-09-24 08:20:12 -04:00
#!/usr/bin/env python3
2022-04-01 11:18:18 -04:00
# 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.
2021-09-24 08:20:12 -04:00
import RNS
import os
import sys
import time
import argparse
from RNS._version import __version__
DEFAULT_PROBE_SIZE = 16
2021-09-24 09:34:03 -04:00
def program_setup(configdir, destination_hexhash, size=DEFAULT_PROBE_SIZE, full_name = None, verbosity = 0):
2021-09-24 08:20:12 -04:00
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()
2021-09-24 10:49:07 -04:00
if verbosity > 0:
more_output = True
verbosity -= 1
else:
more_output = False
verbosity -= 1
2021-09-24 08:20:12 -04:00
2021-09-24 09:34:03 -04:00
reticulum = RNS.Reticulum(configdir = configdir, loglevel = 3+verbosity)
2021-09-24 08:20:12 -04:00
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()
2021-09-24 10:49:07 -04:00
if more_output:
2021-09-25 17:22:33 -04:00
more = " via "+RNS.prettyhexrep(reticulum.get_next_hop(destination_hash))+" on "+str(reticulum.get_next_hop_if_name(destination_hash))
2021-09-24 10:49:07 -04:00
else:
more = ""
print("\rSent "+str(size)+" byte probe to "+RNS.prettyhexrep(destination_hash)+more+" ", end=" ")
2021-09-24 08:20:12 -04:00
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()
2021-09-24 09:34:03 -04:00
hops = RNS.Transport.hops_to(destination_hash)
2021-09-25 17:22:33 -04:00
if hops != 1:
2021-09-24 09:34:03 -04:00
ms = "s"
else:
ms = ""
2021-09-24 08:20:12 -04:00
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"
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)
if reception_rssi != None:
reception_stats += " [RSSI "+str(reception_rssi)+" dBm]"
if reception_snr != None:
2021-10-12 12:04:55 -04:00
reception_stats += " [SNR "+str(reception_snr)+" dB]"
else:
if receipt.proof_packet != None:
if receipt.proof_packet.rssi != None:
reception_stats += " [RSSI "+str(receipt.proof_packet.rssi)+" dBm]"
if receipt.proof_packet.snr != None:
2021-10-12 10:36:29 -04:00
reception_stats += " [SNR "+str(receipt.proof_packet.snr)+" dB]"
2021-09-24 08:20:12 -04:00
print(
"Valid reply received from "+
RNS.prettyhexrep(receipt.destination.hash)+
2021-09-24 09:34:03 -04:00
"\nRound-trip time is "+rttstring+
" over "+str(hops)+" hop"+ms+
reception_stats
2021-09-24 08:20:12 -04:00
)
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",
2021-09-24 14:10:04 -04:00
version="rnprobe {version}".format(version=__version__)
2021-09-24 08:20:12 -04:00
)
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
)
2021-09-24 09:34:03 -04:00
parser.add_argument('-v', '--verbose', action='count', default=0)
2021-09-24 08:20:12 -04:00
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,
2021-09-24 09:34:03 -04:00
full_name = args.full_name,
verbosity = args.verbose
2021-09-24 08:20:12 -04:00
)
except KeyboardInterrupt:
print("")
exit()
if __name__ == "__main__":
main()