461 lines
17 KiB
Python
461 lines
17 KiB
Python
#!/usr/bin/env python
|
|
# Meshtastic MQTT Interface - Developed by acidvegas in Python (https://acid.vegas/meshtastic)
|
|
|
|
import argparse
|
|
import base64
|
|
import logging
|
|
|
|
try:
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
except ImportError:
|
|
raise SystemExit('missing the cryptography module (pip install cryptography)')
|
|
|
|
try:
|
|
from meshtastic import mesh_pb2, mqtt_pb2, portnums_pb2, telemetry_pb2
|
|
except ImportError:
|
|
raise SystemExit('missing the meshtastic module (pip install meshtastic)')
|
|
|
|
try:
|
|
import paho.mqtt.client as mqtt
|
|
except ImportError:
|
|
raise SystemExit('missing the paho-mqtt module (pip install paho-mqtt)')
|
|
|
|
|
|
# Initialize the logging module
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %I:%M:%S')
|
|
|
|
|
|
def clean_dict(dictionary: dict) -> dict:
|
|
'''
|
|
Remove empty fields from a dictionary.
|
|
|
|
:param dictionary: The dictionary to remove empty fields from
|
|
'''
|
|
|
|
return {key: value for key, value in dictionary.items() if value}
|
|
|
|
|
|
class MeshtasticMQTT(object):
|
|
def __init__(self):
|
|
'''Initialize the Meshtastic MQTT client'''
|
|
|
|
self.broadcast_id = 4294967295 # Our channel ID
|
|
self.key = None
|
|
|
|
|
|
def connect(self, broker: str, port: int, root: str, tls: bool, username: str, password: str, key: str):
|
|
'''
|
|
Connect to the MQTT broker
|
|
|
|
:param broker: The MQTT broker address
|
|
:param port: The MQTT broker port
|
|
:param root: The root topic to subscribe to
|
|
:param tls: Enable TLS/SSL
|
|
:param username: The MQTT username
|
|
:param password: The MQTT password
|
|
:param key: The encryption key
|
|
'''
|
|
|
|
# Initialize the MQTT client (these arguments were the only way to get it to work properly..)
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='', clean_session=True, userdata=None)
|
|
|
|
# Set the username and password for the MQTT broker
|
|
client.username_pw_set(username=username, password=password)
|
|
|
|
# Set the encryption key global in the client class (the default key is padded to ensure it's the correct length for AES)
|
|
self.key = '1PG7OiApB1nwvP+rz05pAQ==' if key == 'AQ==' else key
|
|
|
|
# Enable TLS/SSL if the user specified it
|
|
if tls:
|
|
client.tls_set()
|
|
#client.tls_insecure_set(False)
|
|
|
|
# Set the MQTT callbacks
|
|
client.on_connect = self.on_connect
|
|
client.on_message = self.on_message
|
|
client.on_subscribe = self.on_subscribe
|
|
client.on_unsubscribe = self.on_unsubscribe
|
|
|
|
# Connect to the MQTT broker and subscribe to the root topic
|
|
client.connect(broker, port, 60)
|
|
client.subscribe(root, 0)
|
|
|
|
# Keep-alive loop
|
|
client.loop_forever()
|
|
|
|
|
|
def decrypt_message_packet(self, message_packet):
|
|
'''
|
|
Decrypt an encrypted message packet.
|
|
|
|
:param message_packet: The message packet to decrypt
|
|
'''
|
|
|
|
# Ensure the key is formatted and padded correctly before turning it into bytes
|
|
padded_key = self.key.ljust(len(self.key) + ((4 - (len(self.key) % 4)) % 4), '=')
|
|
key = padded_key.replace('-', '+').replace('_', '/')
|
|
key_bytes = base64.b64decode(key.encode('ascii'))
|
|
|
|
# Extract the nonce from the packet
|
|
nonce_packet_id = getattr(message_packet, 'id').to_bytes(8, 'little')
|
|
nonce_from_node = getattr(message_packet, 'from').to_bytes(8, 'little')
|
|
nonce = nonce_packet_id + nonce_from_node
|
|
|
|
# Decrypt the message
|
|
cipher = Cipher(algorithms.AES(key_bytes), modes.CTR(nonce), backend=default_backend())
|
|
decryptor = cipher.decryptor()
|
|
decrypted_bytes = decryptor.update(getattr(message_packet, 'encrypted')) + decryptor.finalize()
|
|
|
|
# Parse the decrypted message
|
|
data = mesh_pb2.Data()
|
|
data.ParseFromString(decrypted_bytes)
|
|
message_packet.decoded.CopyFrom(data)
|
|
|
|
return message_packet
|
|
|
|
|
|
def on_connect(self, client, userdata, flags, rc, properties):
|
|
'''
|
|
Callback for when the client receives a CONNACK response from the server.
|
|
|
|
:param client: The client instance for this callback
|
|
:param userdata: The private user data as set in Client() or user_data_set()
|
|
:param flags: Response flags sent by the broker
|
|
:param rc: The connection result
|
|
:param properties: The properties returned by the broker
|
|
'''
|
|
|
|
if rc == 0:
|
|
logging.info('Connected to MQTT broker')
|
|
else:
|
|
logging.error(f'Failed to connect to MQTT broker: {rc}')
|
|
|
|
|
|
def on_message(self, client, userdata, msg):
|
|
'''
|
|
Callback for when a message is received from the server.
|
|
|
|
:param client: The client instance for this callback
|
|
:param userdata: The private user data as set in Client() or user_data_set()
|
|
:param msg: An instance of MQTTMessage. This is a
|
|
'''
|
|
|
|
# Define the service envelope
|
|
service_envelope = mqtt_pb2.ServiceEnvelope()
|
|
|
|
try:
|
|
# Parse the message payload
|
|
service_envelope.ParseFromString(msg.payload)
|
|
|
|
# Extract the message packet from the service envelope
|
|
message_packet = service_envelope.packet
|
|
except Exception as e:
|
|
#logging.error(f'Failed to parse message: {str(e)}')
|
|
return
|
|
|
|
# Check if the message is encrypted before decrypting it
|
|
if message_packet.HasField('encrypted') and not message_packet.HasField('decoded'):
|
|
message_packet = self.decrypt_message_packet(message_packet)
|
|
|
|
text = {
|
|
'from' : getattr(message_packet, 'from'),
|
|
'to' : getattr(message_packet, 'to'),
|
|
'channel' : getattr(message_packet, 'channel'),
|
|
'id' : getattr(message_packet, 'id'),
|
|
'rx_time' : getattr(message_packet, 'rx_time'),
|
|
'hop_limit' : getattr(message_packet, 'hop_limit'),
|
|
'priority' : getattr(message_packet, 'priority'),
|
|
'hop_start' : getattr(message_packet, 'hop_start')
|
|
}
|
|
logging.info(text)
|
|
|
|
if message_packet.decoded.portnum == portnums_pb2.UNKNOWN_APP:
|
|
logging.warning('Received an unknown app message:')
|
|
logging.info(message_packet)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.TEXT_MESSAGE_APP:
|
|
text_payload = message_packet.decoded.payload.decode('utf-8')
|
|
text = {
|
|
'message' : text_payload,
|
|
'from' : getattr(message_packet, 'from'),
|
|
'id' : getattr(message_packet, 'id'),
|
|
'to' : getattr(message_packet, 'to')
|
|
}
|
|
logging.info('Received text message:')
|
|
logging.info(text)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.REMOTE_HARDWARE_APP:
|
|
data = mesh_pb2.RemoteHardware()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received remote hardware:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.POSITION_APP:
|
|
data = mesh_pb2.Position()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
|
|
data_dict = {key: value for key, value in data}
|
|
print(data_dict)
|
|
|
|
logging.info('Received position:')
|
|
loc = {
|
|
'lattitude' : getattr(data, 'latitude_i') / 1e7,
|
|
'longitude' : getattr(data, 'longitude_i') / 1e7,
|
|
'altitude' : getattr(data, 'altitude') / 1000,
|
|
'location_source' : getattr(data, 'location_source'),
|
|
'altitude_source' : getattr(data, 'altitude_source'),
|
|
'pdop' : getattr(data, 'PDOP'),
|
|
'hdop' : getattr(data, 'HDOP'),
|
|
'vdop' : getattr(data, 'VDOP'),
|
|
'gps_accuracy' : getattr(data, 'gps_accuracy'),
|
|
'ground_speed' : getattr(data, 'ground_speed'),
|
|
'ground_track' : getattr(data, 'ground_track'),
|
|
'fix_quality' : getattr(data, 'fix_quality'),
|
|
'fix_type' : getattr(data, 'fix_type'),
|
|
'sats_in_view' : getattr(data, 'sats_in_view'),
|
|
'sensor_id' : getattr(data, 'sensor_id'),
|
|
'next_update' : getattr(data, 'next_update'),
|
|
'seq_number' : getattr(data, 'seq_number'),
|
|
'precision_bits' : getattr(data, 'precision_bits')
|
|
}
|
|
|
|
if (loc := clean_dict(loc)):
|
|
logging.info(loc)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.NODEINFO_APP:
|
|
data = mesh_pb2.NodeInfo()
|
|
#data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received node info:')
|
|
logging.info(message_packet)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.ROUTING_APP:
|
|
data = mesh_pb2.Routing()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received routing:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.ADMIN_APP:
|
|
data = mesh_pb2.Admin()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received admin:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.TEXT_MESSAGE_COMPRESSED_APP:
|
|
data = mesh_pb2.TextMessageCompressed()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received compressed text message:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.WAYPOINT_APP:
|
|
data = mesh_pb2.Waypoint()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received waypoint:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.AUDIO_APP:
|
|
data = mesh_pb2.Audio()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received audio:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.DETECTION_SENSOR_APP:
|
|
data = mesh_pb2.DetectionSensor()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received detection sensor:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.REPLY_APP:
|
|
data = mesh_pb2.Reply()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received reply:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.IP_TUNNEL_APP:
|
|
data = mesh_pb2.IPTunnel()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received IP tunnel:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.PAXCOUNTER_APP:
|
|
data = mesh_pb2.Paxcounter()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received paxcounter:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.SERIAL_APP:
|
|
data = mesh_pb2.Serial()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received serial:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.STORE_FORWARD_APP:
|
|
logging.info('Received store and forward:')
|
|
logging.info(message_packet)
|
|
logging.info(message_packet.decoded.payload)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.RANGE_TEST_APP:
|
|
data = mesh_pb2.RangeTest()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received range test:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.TELEMETRY_APP:
|
|
data = telemetry_pb2.Telemetry()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received telemetry:')
|
|
|
|
data_dict = {}
|
|
for field, value in data.ListFields():
|
|
if field.name == 'device_metrics':
|
|
text = clean_dict({item.name: getattr(value, item.name) for item in value.DESCRIPTOR.fields if hasattr(value, item.name)})
|
|
if text:
|
|
logging.info(text)
|
|
else:
|
|
data_dict[field.name] = value
|
|
|
|
logging.info(data_dict)
|
|
|
|
if getattr(data, 'device_metrics'):
|
|
text = {
|
|
'battery_level' : getattr(data.device_metrics, 'battery_level'),
|
|
'voltage' : getattr(data.device_metrics, 'voltage'),
|
|
'channel_utilization' : getattr(data.device_metrics, 'channel_utilization'),
|
|
'air_util_tx' : getattr(data.device_metrics, 'air_util_tx'),
|
|
'uptime_seconds' : getattr(data.device_metrics, 'uptime_seconds')
|
|
}
|
|
if (text := clean_dict(text)):
|
|
logging.info(text)
|
|
|
|
if getattr(data, 'environment_metrics'):
|
|
env_metrics = {
|
|
'barometric_pressure' : getattr(data.environment_metrics, 'barometric_pressure'),
|
|
'current' : getattr(data.environment_metrics, 'current'),
|
|
'distance' : getattr(data.environment_metrics, 'distance'),
|
|
'gas_resistance' : getattr(data.environment_metrics, 'gas_resistance'),
|
|
'iaq' : getattr(data.environment_metrics, 'iaq'),
|
|
'relative_humidity' : getattr(data.environment_metrics, 'relative_humidity'),
|
|
'temperature' : getattr(data.environment_metrics, 'temperature'),
|
|
'voltage' : getattr(data.environment_metrics, 'voltage')
|
|
}
|
|
if (env_metrics := clean_dict(env_metrics)):
|
|
logging.info(env_metrics)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.ZPS_APP:
|
|
data = mesh_pb2.Zps()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received ZPS:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.SIMULATOR_APP:
|
|
data = mesh_pb2.Simulator()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received simulator:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.TRACEROUTE_APP:
|
|
routeDiscovery = mesh_pb2.RouteDiscovery()
|
|
routeDiscovery.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received traceroute:')
|
|
logging.info(routeDiscovery)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.NEIGHBORINFO_APP:
|
|
neighborInfo = mesh_pb2.NeighborInfo()
|
|
neighborInfo.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received neighbor info:')
|
|
info = {
|
|
'node_id' : getattr(neighborInfo, 'node_id'),
|
|
'last_sent_by_id' : getattr(neighborInfo, 'last_sent_by_id'),
|
|
'node_broadcast_interval_secs' : getattr(neighborInfo, 'node_broadcast_interval_secs'),
|
|
'neighbors' : getattr(neighborInfo, 'neighbors')
|
|
}
|
|
logging.info(info)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.ATAK_PLUGIN:
|
|
data = mesh_pb2.AtakPlugin()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received ATAK plugin:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.PRIVATE_APP:
|
|
data = mesh_pb2.Private()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received private:')
|
|
logging.info(data)
|
|
|
|
elif message_packet.decoded.portnum == portnums_pb2.ATAK_FORWARDER:
|
|
data = mesh_pb2.AtakForwarder()
|
|
data.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received ATAK forwarder:')
|
|
logging.info(data)
|
|
|
|
else:
|
|
logging.warning('Received an unknown message:')
|
|
logging.info(message_packet)
|
|
|
|
# Unencrypted messages
|
|
else:
|
|
if message_packet.decoded.portnum == portnums_pb2.MAP_REPORT_APP:
|
|
pos = mesh_pb2.Position()
|
|
pos.ParseFromString(message_packet.decoded.payload)
|
|
logging.info('Received map report:')
|
|
logging.info(pos)
|
|
|
|
else:
|
|
logging.warning('Received an unencrypted message')
|
|
logging.info(f'Payload: {message_packet}')
|
|
|
|
|
|
def on_subscribe(self, client, userdata, mid, reason_code_list, properties):
|
|
'''
|
|
Callback for when the client receives a SUBACK response from the server.
|
|
|
|
:param client: The client instance for this callback
|
|
:param userdata: The private user data as set in Client() or user_data_set()
|
|
:param mid: The message ID of the subscribe request
|
|
:param reason_code_list: A list of SUBACK reason codes
|
|
:param properties: The properties returned by the broker
|
|
'''
|
|
|
|
# Since we subscribed only for a single channel, reason_code_list contains a single entry
|
|
if reason_code_list[0].is_failure:
|
|
logging.error(f'Broker rejected you subscription: {reason_code_list[0]}')
|
|
else:
|
|
logging.info(f'Broker granted the following QoS: {reason_code_list[0].value}')
|
|
|
|
|
|
def on_unsubscribe(self, client, userdata, mid, reason_code_list, properties):
|
|
'''
|
|
Callback for when the client receives a UNSUBACK response from the server.
|
|
|
|
:param client: The client instance for this callback
|
|
:param userdata: The private user data as set in Client() or user_data_set()
|
|
:param mid: The message ID of the unsubscribe request
|
|
:param reason_code_list: A list of UNSUBACK reason codes
|
|
:param properties: The properties returned by the broker
|
|
'''
|
|
|
|
# reason_code_list is only present in MQTTv5, it will always be empty in MQTTv3
|
|
if len(reason_code_list) == 0 or not reason_code_list[0].is_failure:
|
|
logging.info('Broker accepted the unsubscription(s)')
|
|
else:
|
|
logging.error(f'Broker replied with failure: {reason_code_list[0]}')
|
|
|
|
# Disconnect from the broker
|
|
client.disconnect()
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description='Meshtastic MQTT Interface')
|
|
parser.add_argument('--broker', default='mqtt.meshtastic.org', help='MQTT broker address')
|
|
parser.add_argument('--port', default=1883, type=int, help='MQTT broker port')
|
|
parser.add_argument('--root', default='#', help='Root topic')
|
|
parser.add_argument('--tls', action='store_true', help='Enable TLS/SSL')
|
|
parser.add_argument('--username', default='meshdev', help='MQTT username')
|
|
parser.add_argument('--password', default='large4cats', help='MQTT password')
|
|
parser.add_argument('--key', default='AQ==', help='Encryption key')
|
|
args = parser.parse_args()
|
|
|
|
client = MeshtasticMQTT()
|
|
client.connect(args.broker, args.port, args.root, args.tls, args.username, args.password, args.key) |