iarv/meshing-around
1
0
Fork 1
mirror of https://github.com/SpudGunMan/meshing-around.git synced 2026-03-28 17:32:36 +01:00
Code Issues Packages Projects Releases Wiki Activity

coreFix

Browse Source 
Enhance Packet hop and MQTT detection
This commit is contained in:
SpudGunMan
2025-10-19 22:16:19 -07:00
parent d99698e7f3
commit a859f830bb
3 changed files with 142 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
mesh_bot.py
View File

@@ -1449,7 +1449,13 @@ def onReceive(packet, interface):
pkiStatus = (False, 'ABC')
replyIDset = False
emojiSeen = False
simulator_flag = False
isDM = False
channel_number = 0
hop_away = 0
hop_start = 0
hop_count = 0
channel_name = "unknown"
playingGame = False
if DEBUGpacket:
@@ -1496,7 +1502,23 @@ def onReceive(packet, interface):
# check if the packet has a channel flag use it
if packet.get('channel'):
channel_number = packet.get('channel', 0)
channel_number = packet.get('channel')
channel_name = "unknown"
# get channel hashes for the interface
device = next((d for d in channel_list if d["interface_id"] == rxNode), None)
if device:
# Find the channel name whose hash matches channel_number
for chan_name, info in device['channels'].items():
if info['hash'] == channel_number:
print(f"Matched channel hash {info['hash']} to channel name {chan_name}")
channel_name = chan_name
break
# check if the packet has a simulator flag
simulator_flag = packet['decoded'].get('simulator', False)
if isinstance(simulator_flag, dict):
# assume Software Simulator
simulator_flag = True
# set the message_from_id
message_from_id = packet['from']
@@ -1521,7 +1543,13 @@ def onReceive(packet, interface):
message_bytes = packet['decoded']['payload']
message_string = message_bytes.decode('utf-8')
via_mqtt = packet['decoded'].get('viaMqtt', False)
transport_mechanism = packet['decoded'].get('transport_mechanism', 'unknown')
transport_mechanism = (
packet.get('transport_mechanism')
or packet.get('transportMechanism')
or (packet.get('decoded', {}).get('transport_mechanism'))
or (packet.get('decoded', {}).get('transportMechanism'))
or 'unknown'
)
rx_time = packet['decoded'].get('rxTime', time.time())
# check if the packet is from us
@@ -1548,40 +1576,33 @@ def onReceive(packet, interface):
# check if the packet has a hop count flag use it
if packet.get('hopsAway'):
hop_away = packet.get('hopsAway', 0)
if packet.get('hopStart'):
hop_start = packet.get('hopStart', 0)
if packet.get('hopLimit'):
hop_limit = packet.get('hopLimit', 0)
# calculate hop count
hop = ""
if hop_limit > 0 and hop_start >= hop_limit:
hop_count = hop_away + (hop_start - hop_limit)
elif hop_limit > 0 and hop_start < hop_limit:
hop_count = hop_away + (hop_limit - hop_start)
else:
# if the packet does not have a hop count try other methods
if packet.get('hopLimit'):
hop_limit = packet.get('hopLimit', 0)
else:
hop_limit = 0
if packet.get('hopStart'):
hop_start = packet.get('hopStart', 0)
else:
hop_start = 0
if enableHopLogs:
logger.debug(f"System: Packet HopDebugger: hop_away:{hop_away} hop_limit:{hop_limit} hop_start:{hop_start}")
hop_count = hop_away
if hop_away == 0 and hop_limit == 0 and hop_start == 0:
hop = "Last Hop"
hop_count = 0
if hop_start == hop_limit:
hop = "Direct"
hop_count = 0
elif hop_start == 0 and hop_limit > 0 or via_mqtt or transport_mechanism == "TRANSPORT_MQTT":
hop = "MQTT"
hop_count = 0
else:
# set hop to Direct if the message was sent directly otherwise set the hop count
if hop_away > 0:
hop_count = hop_away
else:
hop_count = hop_start - hop_limit
#print (f"calculated hop count: {hop_start} - {hop_limit} = {hop_count}")
hop = f"{hop_count} hops"
if ((hop_start == 0 and hop_limit >= 0) or via_mqtt or ("mqtt" in str(transport_mechanism).lower())):
hop = "MQTT"
if enableHopLogs:
logger.debug(f"System: Packet HopDebugger: hop_away:{hop_away} hop_limit:{hop_limit} hop_start:{hop_start} calculated_hop_count:{hop_count} final_hop_value:{hop} via_mqtt:{via_mqtt} transport_mechanism:{transport_mechanism}")
# check with stringSafeChecker if the message is safe
if stringSafeCheck(message_string) is False:

56
modules/system.py
View File

@@ -7,6 +7,7 @@ import meshtastic.ble_interface
import time
import asyncio
import random
import base64
# not ideal but needed?
import contextlib # for suppressing output on watchdog
import io # for suppressing output on watchdog
@@ -315,6 +316,24 @@ if ble_count > 1:
logger.critical(f"System: Multiple BLE interfaces detected. Only one BLE interface is allowed. Exiting")
exit()
def xor_hash(data: bytes) -> int:
"""Compute an XOR hash from bytes."""
result = 0
for char in data:
result ^= char
return result
def generate_hash(name: str, key: str) -> int:
"""generate the channel number by hashing the channel name and psk"""
if key == "AQ==":
key = "1PG7OiApB1nwvP+rz05pAQ=="
replaced_key = key.replace("-", "+").replace("_", "/")
key_bytes = base64.b64decode(replaced_key.encode("utf-8"))
h_name = xor_hash(bytes(name, "utf-8"))
h_key = xor_hash(key_bytes)
result: int = h_name ^ h_key
return result
# Initialize interfaces
logger.debug(f"System: Initializing Interfaces")
interface1 = interface2 = interface3 = interface4 = interface5 = interface6 = interface7 = interface8 = interface9 = None
@@ -365,6 +384,43 @@ for i in range(1, 10):
else:
globals()[f'myNodeNum{i}'] = 777
# Fetch channel list from each device
channel_list = []
for i in range(1, 10):
if globals().get(f'interface{i}') and globals().get(f'interface{i}_enabled'):
try:
node = globals()[f'interface{i}'].getNode('^local')
channels = node.channels
channel_dict = {}
for channel in channels:
if hasattr(channel, 'role') and channel.role:
channel_name = getattr(channel.settings, 'name', '').strip()
channel_number = getattr(channel, 'index', 0)
# Only add channels with a non-empty name
if channel_name:
channel_dict[channel_name] = channel_number
channel_list.append({
"interface_id": i,
"channels": channel_dict
})
logger.debug(f"System: Fetched Channel List from Device{i}")
except Exception as e:
logger.error(f"System: Error fetching channel list from Device{i}: {e}")
# add channel hash to channel_list
for device in channel_list:
interface_id = device["interface_id"]
for channel_name, channel_number in device["channels"].items():
psk_base64 = base64.b64encode(channel.settings.psk).decode('utf-8')
channel_hash = generate_hash(channel_name, psk_base64)
# add hash to the channel entry in channel_list under key 'hash'
for entry in channel_list:
if entry["interface_id"] == interface_id:
entry["channels"][channel_name] = {
"number": channel_number,
"hash": channel_hash
}
#### FUN-ctions ####
def cleanup_memory():

64
pong_bot.py
View File

@@ -282,6 +282,7 @@ def onReceive(packet, interface):
message_bytes = packet['decoded']['payload']
message_string = message_bytes.decode('utf-8')
via_mqtt = packet['decoded'].get('viaMqtt', False)
transport_mechanism = packet['decoded'].get('transport_mechanism', 'unknown')
# check if the packet is from us
if message_from_id == myNodeNum1 or message_from_id == myNodeNum2:
@@ -294,46 +295,51 @@ def onReceive(packet, interface):
# check if the packet has a publicKey flag use it
if packet.get('publicKey'):
pkiStatus = (packet.get('pkiEncrypted', False), packet.get('publicKey', 'ABC'))
pkiStatus = packet.get('pkiEncrypted', False), packet.get('publicKey', 'ABC')
# check if the packet has replyId flag // currently unused in the code
if packet.get('replyId'):
replyIDset = packet.get('replyId', False)
# check if the packet has emoji flag set it // currently unused in the code
if packet.get('emoji'):
emojiSeen = packet.get('emoji', False)
# check if the packet has a hop count flag use it
if packet.get('hopsAway'):
hop_away = packet.get('hopsAway', 0)
else:
# if the packet does not have a hop count try other methods
if packet.get('hopLimit'):
hop_limit = packet.get('hopLimit', 0)
else:
hop_limit = 0
if packet.get('hopStart'):
hop_start = packet.get('hopStart', 0)
else:
hop_start = 0
if enableHopLogs:
logger.debug(f"System: Packet HopDebugger: hop_away:{hop_away} hop_limit:{hop_limit} hop_start:{hop_start}")
if packet.get('hopStart'):
hop_start = packet.get('hopStart', 0)
if packet.get('hopLimit'):
hop_limit = packet.get('hopLimit', 0)
# calculate hop count
hop = ""
if hop_limit > 0 and hop_start >= hop_limit:
hop_count = hop_away + (hop_start - hop_limit)
elif hop_limit > 0 and hop_start < hop_limit:
hop_count = hop_away + (hop_limit - hop_start)
else:
hop_count = hop_away
if hop_away == 0 and hop_limit == 0 and hop_start == 0:
hop = "Last Hop"
hop_count = 0
if hop_start == hop_limit:
hop = "Direct"
hop_count = 0
elif hop_start == 0 and hop_limit > 0 or via_mqtt:
hop = "MQTT"
hop_count = 0
else:
# set hop to Direct if the message was sent directly otherwise set the hop count
if hop_away > 0:
hop_count = hop_away
else:
hop_count = hop_start - hop_limit
#print (f"calculated hop count: {hop_start} - {hop_limit} = {hop_count}")
hop = f"{hop_count} hops"
if ((hop_start == 0 and hop_limit >= 0) or via_mqtt or ("mqtt" in str(transport_mechanism).lower())):
hop = "MQTT"
if enableHopLogs:
logger.debug(f"System: Packet HopDebugger: hop_away:{hop_away} hop_limit:{hop_limit} hop_start:{hop_start} calculated_hop_count:{hop_count} final_hop_value:{hop} via_mqtt:{via_mqtt} transport_mechanism:{transport_mechanism}")
# check with stringSafeChecker if the message is safe
if stringSafeCheck(message_string) is False:
logger.warning(f"System: Possibly Unsafe Message from {get_name_from_number(message_from_id, 'long', rxNode)}")
if help_message in message_string or welcome_message in message_string or "CMD?:" in message_string:
# ignore help and welcome messages
logger.warning(f"Got Own Welcome/Help header. From: {get_name_from_number(message_from_id, 'long', rxNode)}")