From 96493e59738882d8759642e26c8949dd67decf19 Mon Sep 17 00:00:00 2001 From: Russell Schmidt Date: Thu, 16 Jan 2025 07:49:01 -0600 Subject: [PATCH] Add traceroute support --- message_handlers/tx_handler.py | 79 +++++++++++++++++++++++++++++++++- ui/curses_ui.py | 9 ++-- 2 files changed, 84 insertions(+), 4 deletions(-) diff --git a/message_handlers/tx_handler.py b/message_handlers/tx_handler.py index 0a50903..a301cda 100644 --- a/message_handlers/tx_handler.py +++ b/message_handlers/tx_handler.py @@ -1,7 +1,9 @@ from meshtastic import BROADCAST_NUM from db_handler import save_message_to_db, update_ack_nak -from utilities.utils import get_nodeNum +from meshtastic.protobuf import mesh_pb2, portnums_pb2 +from utilities.utils import get_nodeNum, get_name_from_number import globals +import google.protobuf.json_format ack_naks = {} @@ -35,6 +37,70 @@ def onAckNak(packet): draw_messages_window() +def on_response_traceroute(packet): + """on response for trace route""" + from ui.curses_ui import draw_channel_list, draw_messages_window, add_notification + + UNK_SNR = -128 # Value representing unknown SNR + + route_discovery = mesh_pb2.RouteDiscovery() + route_discovery.ParseFromString(packet["decoded"]["payload"]) + msg_dict = google.protobuf.json_format.MessageToDict(route_discovery) + + msg_str = "Route traced towards destination:\n" + + route_str = get_name_from_number(packet["to"]) or f"{packet['to']:08x}" # Start with destination of response + + # SNR list should have one more entry than the route, as the final destination adds its SNR also + lenTowards = 0 if "route" not in msg_dict else len(msg_dict["route"]) + snrTowardsValid = "snrTowards" in msg_dict and len(msg_dict["snrTowards"]) == lenTowards + 1 + if lenTowards > 0: # Loop through hops in route and add SNR if available + for idx, node_num in enumerate(msg_dict["route"]): + route_str += " --> " + (get_name_from_number(node_num) or f"{node_num:08x}") \ + + " (" + (str(msg_dict["snrTowards"][idx] / 4) if snrTowardsValid and msg_dict["snrTowards"][idx] != UNK_SNR else "?") + "dB)" + + # End with origin of response + route_str += " --> " + (get_name_from_number(packet["from"]) or f"{packet['from']:08x}") \ + + " (" + (str(msg_dict["snrTowards"][-1] / 4) if snrTowardsValid and msg_dict["snrTowards"][-1] != UNK_SNR else "?") + "dB)" + + msg_str += route_str + "\n" # Print the route towards destination + + # Only if hopStart is set and there is an SNR entry (for the origin) it's valid, even though route might be empty (direct connection) + lenBack = 0 if "routeBack" not in msg_dict else len(msg_dict["routeBack"]) + backValid = "hopStart" in packet and "snrBack" in msg_dict and len(msg_dict["snrBack"]) == lenBack + 1 + if backValid: + msg_str += "Route traced back to us:\n" + route_str = get_name_from_number(packet["from"]) or f"{packet['from']:08x}" # Start with origin of response + + if lenBack > 0: # Loop through hops in routeBack and add SNR if available + for idx, node_num in enumerate(msg_dict["routeBack"]): + route_str += " --> " + (get_name_from_number(node_num) or f"{node_num:08x}") \ + + " (" + (str(msg_dict["snrBack"][idx] / 4) if msg_dict["snrBack"][idx] != UNK_SNR else "?") + "dB)" + + # End with destination of response (us) + route_str += " --> " + (get_name_from_number(packet["to"]) or f"{p['to']:08x}") \ + + " (" + (str(msg_dict["snrBack"][-1] / 4) if msg_dict["snrBack"][-1] != UNK_SNR else "?") + "dB)" + + msg_str += route_str + "\n" # Print the route back to us + + if(packet['from'] not in globals.channel_list): + globals.channel_list.append(packet['from']) + + channel_number = globals.channel_list.index(packet['from']) + + if globals.channel_list[channel_number] != globals.channel_list[globals.selected_channel]: + add_notification(channel_number) + + message_from_string = get_name_from_number(packet['from'], type='short') + ":\n" + + if globals.channel_list[channel_number] not in globals.all_messages: + globals.all_messages[globals.channel_list[channel_number]] = [] + globals.all_messages[globals.channel_list[channel_number]].append((f"{globals.message_prefix} {message_from_string}", msg_str)) + + draw_channel_list() + draw_messages_window() + save_message_to_db(globals.channel_list[channel_number], packet['from'], msg_str) + def send_message(message, destination=BROADCAST_NUM, channel=0): myid = globals.myNodeNum @@ -65,3 +131,14 @@ def send_message(message, destination=BROADCAST_NUM, channel=0): ack_naks[sent_message_data.id] = {'channel' : channel_id, 'messageIndex' : len(globals.all_messages[channel_id]) - 1, 'timestamp' : timestamp } +def send_traceroute(): + r = mesh_pb2.RouteDiscovery() + globals.interface.sendData( + r, + destinationId=globals.node_list[globals.selected_node], + portNum=portnums_pb2.PortNum.TRACEROUTE_APP, + wantResponse=True, + onResponse=on_response_traceroute, + channelIndex=0, + hopLimit=3, + ) diff --git a/ui/curses_ui.py b/ui/curses_ui.py index 615cc8e..394284a 100644 --- a/ui/curses_ui.py +++ b/ui/curses_ui.py @@ -3,8 +3,7 @@ import textwrap import globals from utilities.utils import get_name_from_number, get_channels from settings import settings -from message_handlers.tx_handler import send_message - +from message_handlers.tx_handler import send_message, send_traceroute def add_notification(channel_number): handle_notification(channel_number, add=True) @@ -324,7 +323,11 @@ def main_ui(stdscr): # Check for Esc elif char == 27: break - + + # Check for Ctrl + t + elif char == 20: + send_traceroute() + elif char == curses.KEY_ENTER or char == 10 or char == 13: if globals.current_window == 2: node_list = globals.node_list