#!/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)

© 2023 acidvegas, inc • generated with stagit