acidvegas/meshtastic
acidvegas/meshtastic
1
Fork 0
Code Issues Pull Requests Activity

Entire code is in a class now for modular usage

Browse Source
This commit is contained in:
Dionysus 2024-05-07 16:47:30 -04:00
parent d33930ee37
commit 053c5e9d32
Signed by: acidvegas
GPG Key ID: EF4B922DB85DC9DE
1 changed files with 172 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

322
meshmqtt.py
View File

@ -6,8 +6,8 @@ import base64
import logging import logging
try: try:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
except ImportError: except ImportError:
raise SystemExit('missing the cryptography module (pip install cryptography)') raise SystemExit('missing the cryptography module (pip install cryptography)')
@ -26,27 +26,66 @@ except ImportError:
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %I:%M:%S') logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %I:%M:%S')
def decode_encrypted(message_packet): class MeshtasticMQTT(object):
''' def __init__(self):
Decrypt an encrypted message packet. '''Initialize the Meshtastic MQTT client'''
:param message_packet: The message packet to decrypt self.broadcast_id = 4294967295
''' 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
'''
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='', clean_session=True, userdata=None)
client.username_pw_set(username=username, password=password)
self.key = '1PG7OiApB1nwvP+rz05pAQ==' if key == 'AQ==' else key
if tls:
client.tls_set()
#client.tls_insecure_set(False)
client.on_connect = self.on_connect
client.on_message = self.on_message
client.on_subscribe = self.on_subscribe
client.on_unsubscribe = self.on_unsubscribe
client.connect(broker, port, 60)
client.subscribe(root, 0)
client.loop_forever()
def decrypt_message_packet(self, message_packet):
'''
Decrypt an encrypted message packet.
:param message_packet: The message packet to decrypt
'''
try:
# Ensure the key is formatted and padded correctly before turning it into bytes # Ensure the key is formatted and padded correctly before turning it into bytes
padded_key = args.key.ljust(len(args.key) + ((4 - (len(args.key) % 4)) % 4), '=') padded_key = self.key.ljust(len(self.key) + ((4 - (len(self.key) % 4)) % 4), '=')
key = padded_key.replace('-', '+').replace('_', '/') key = padded_key.replace('-', '+').replace('_', '/')
key_bytes = base64.b64decode(key.encode('ascii')) key_bytes = base64.b64decode(key.encode('ascii'))
# Extract the nonce from the packet # Extract the nonce from the packet
nonce_packet_id = getattr(message_packet, 'id').to_bytes(8, 'little') nonce_packet_id = getattr(message_packet, 'id').to_bytes(8, 'little')
nonce_from_node = getattr(message_packet, 'from').to_bytes(8, 'little') nonce_from_node = getattr(message_packet, 'from').to_bytes(8, 'little')
nonce = nonce_packet_id + nonce_from_node nonce = nonce_packet_id + nonce_from_node
# Decrypt the message # Decrypt the message
cipher = Cipher(algorithms.AES(key_bytes), modes.CTR(nonce), backend=default_backend()) cipher = Cipher(algorithms.AES(key_bytes), modes.CTR(nonce), backend=default_backend())
decryptor = cipher.decryptor() decryptor = cipher.decryptor()
decrypted_bytes = decryptor.update(getattr(message_packet, 'encrypted')) + decryptor.finalize() decrypted_bytes = decryptor.update(getattr(message_packet, 'encrypted')) + decryptor.finalize()
# Parse the decrypted message # Parse the decrypted message
@ -54,140 +93,151 @@ def decode_encrypted(message_packet):
data.ParseFromString(decrypted_bytes) data.ParseFromString(decrypted_bytes)
message_packet.decoded.CopyFrom(data) message_packet.decoded.CopyFrom(data)
if message_packet.decoded.portnum == portnums_pb2.TEXT_MESSAGE_APP: return message_packet
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.MAP_REPORT_APP:
pos = mesh_pb2.Position()
pos.ParseFromString(message_packet.decoded.payload)
logging.info('Received map report:')
logging.info(pos)
elif message_packet.decoded.portnum == portnums_pb2.NODEINFO_APP:
info = mesh_pb2.User()
info.ParseFromString(message_packet.decoded.payload)
logging.info('Received node info:')
logging.info(info)
elif message_packet.decoded.portnum == portnums_pb2.POSITION_APP:
pos = mesh_pb2.Position()
pos.ParseFromString(message_packet.decoded.payload)
logging.info('Received position:')
logging.info(pos)
elif message_packet.decoded.portnum == portnums_pb2.TELEMETRY_APP:
env = telemetry_pb2.Telemetry()
env.ParseFromString(message_packet.decoded.payload)
logging.info('Received telemetry:')
logging.info(env)
except Exception as e:
logging.error(f'Failed to decrypt message: {str(e)}')
def on_connect(client, userdata, flags, rc, properties): def on_connect(self, client, userdata, flags, rc, properties):
''' '''
Callback for when the client receives a CONNACK response from the server. Callback for when the client receives a CONNACK response from the server.
:param client: The client instance for this callback :param client: The client instance for this callback
:param userdata: The private user data as set in Client() or user_data_set() :param userdata: The private user data as set in Client() or user_data_set()
:param flags: Response flags sent by the broker :param flags: Response flags sent by the broker
:param rc: The connection result :param rc: The connection result
:param properties: The properties returned by the broker :param properties: The properties returned by the broker
''' '''
if rc == 0: if rc == 0:
logging.info('Connected to MQTT broker') logging.info('Connected to MQTT broker')
else: else:
logging.error(f'Failed to connect to MQTT broker: {rc}') logging.error(f'Failed to connect to MQTT broker: {rc}')
def on_message(client, userdata, msg): def on_message(self, client, userdata, msg):
''' '''
Callback for when a message is received from the server. Callback for when a message is received from the server.
:param client: The client instance for this callback :param client: The client instance for this callback
:param userdata: The private user data as set in Client() or user_data_set() :param userdata: The private user data as set in Client() or user_data_set()
:param msg: An instance of MQTTMessage. This is a :param msg: An instance of MQTTMessage. This is a
''' '''
# Define the service envelope # Define the service envelope
service_envelope = mqtt_pb2.ServiceEnvelope() service_envelope = mqtt_pb2.ServiceEnvelope()
try: try:
# Parse the message payload # Parse the message payload
service_envelope.ParseFromString(msg.payload) service_envelope.ParseFromString(msg.payload)
logging.info('Received a packet:') # logging.info('Received a packet:')
logging.info(service_envelope) # logging.info(service_envelope)
# Extract the message packet from the service envelope # Extract the message packet from the service envelope
message_packet = service_envelope.packet message_packet = service_envelope.packet
except Exception as e: except Exception as e:
#logging.error(f'Failed to parse message: {str(e)}') #logging.error(f'Failed to parse message: {str(e)}')
return return
# Check if the message is encrypted before decrypting it # Check if the message is encrypted before decrypting it
if message_packet.HasField('encrypted') and not message_packet.HasField('decoded'): if message_packet.HasField('encrypted') and not message_packet.HasField('decoded'):
decode_encrypted(message_packet) message_packet = self.decrypt_message_packet(message_packet)
# If the message is not encrypted, log the payload (this should not happen) if message_packet.decoded.portnum == portnums_pb2.TEXT_MESSAGE_APP:
else: text_payload = message_packet.decoded.payload.decode('utf-8')
logging.warning('Received an unencrypted message') text = {
logging.info(f'Payload: {message_packet}') '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)
h = [portnums_pb2.REMOTE_HARDWARE_APP, portnums_pb2.ROUTING_APP,
elif message_packet.decoded.portnum == portnums_pb2.MAP_REPORT_APP: # comes unencrypted
pos = mesh_pb2.Position()
pos.ParseFromString(message_packet.decoded.payload)
logging.info('Received map report:')
logging.info(pos)
elif message_packet.decoded.portnum == portnums_pb2.NODEINFO_APP:
info = mesh_pb2.User()
info.ParseFromString(message_packet.decoded.payload)
logging.info('Received node info:')
logging.info(info)
elif message_packet.decoded.portnum == portnums_pb2.POSITION_APP:
pos = mesh_pb2.Position()
pos.ParseFromString(message_packet.decoded.payload)
logging.info('Received position:')
logging.info(pos)
elif message_packet.decoded.portnum == portnums_pb2.TELEMETRY_APP:
env = telemetry_pb2.Telemetry()
env.ParseFromString(message_packet.decoded.payload)
logging.info('Received telemetry:')
logging.info(env)
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)
else:
logging.warning('Received an unknown message:')
logging.info(message_packet)
# If the message is not encrypted, log the payload (this should not happen)
else:
logging.warning('Received an unencrypted message')
logging.info(f'Payload: {message_packet}')
def on_subscribe(client, userdata, mid, reason_code_list, properties): def on_subscribe(self, client, userdata, mid, reason_code_list, properties):
''' '''
Callback for when the client receives a SUBACK response from the server. Callback for when the client receives a SUBACK response from the server.
:param client: The client instance for this callback :param client: The client instance for this callback
:param userdata: The private user data as set in Client() or user_data_set() :param userdata: The private user data as set in Client() or user_data_set()
:param mid: The message ID of the subscribe request :param mid: The message ID of the subscribe request
:param reason_code_list: A list of SUBACK reason codes :param reason_code_list: A list of SUBACK reason codes
:param properties: The properties returned by the broker :param properties: The properties returned by the broker
''' '''
# Since we subscribed only for a single channel, reason_code_list contains a single entry # Since we subscribed only for a single channel, reason_code_list contains a single entry
if reason_code_list[0].is_failure: if reason_code_list[0].is_failure:
logging.error(f'Broker rejected you subscription: {reason_code_list[0]}') logging.error(f'Broker rejected you subscription: {reason_code_list[0]}')
else: else:
logging.info(f'Broker granted the following QoS: {reason_code_list[0].value}') logging.info(f'Broker granted the following QoS: {reason_code_list[0].value}')
def on_unsubscribe(client, userdata, mid, reason_code_list, properties): def on_unsubscribe(self, client, userdata, mid, reason_code_list, properties):
''' '''
Callback for when the client receives a UNSUBACK response from the server. Callback for when the client receives a UNSUBACK response from the server.
:param client: The client instance for this callback :param client: The client instance for this callback
:param userdata: The private user data as set in Client() or user_data_set() :param userdata: The private user data as set in Client() or user_data_set()
:param mid: The message ID of the unsubscribe request :param mid: The message ID of the unsubscribe request
:param reason_code_list: A list of UNSUBACK reason codes :param reason_code_list: A list of UNSUBACK reason codes
:param properties: The properties returned by the broker :param properties: The properties returned by the broker
''' '''
# reason_code_list is only present in MQTTv5, it will always be empty in MQTTv3 # 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: if len(reason_code_list) == 0 or not reason_code_list[0].is_failure:
logging.info('Broker accepted the unsubscription(s)') logging.info('Broker accepted the unsubscription(s)')
else: else:
logging.error(f'Broker replied with failure: {reason_code_list[0]}') logging.error(f'Broker replied with failure: {reason_code_list[0]}')
# Disconnect from the broker # Disconnect from the broker
client.disconnect() client.disconnect()
if __name__ == '__main__': if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Mesh MQTT') parser = argparse.ArgumentParser(description='Meshtastic MQTT Interface')
parser.add_argument('--broker', default='mqtt.meshtastic.org', help='MQTT broker address') 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('--port', default=1883, type=int, help='MQTT broker port')
parser.add_argument('--root', default='#', help='Root topic') parser.add_argument('--root', default='#', help='Root topic')
@ -197,33 +247,5 @@ if __name__ == '__main__':
parser.add_argument('--key', default='AQ==', help='Encryption key') parser.add_argument('--key', default='AQ==', help='Encryption key')
args = parser.parse_args() args = parser.parse_args()
# Set the broadcast ID (Do we need to change this for a custom channel?) client = MeshtasticMQTT()
broadcast_id = 4294967295 client.connect(args.broker, args.port, args.root, args.tls, args.username, args.password, args.key)
# Create the MQTT client
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='', clean_session=True, userdata=None) # Defaults to mqtt.MQTTv311 (change with protocol=mqtt.MQTTv5)
# Set the authentication details
client.username_pw_set(username=args.username, password=args.password)
# Enable TLS/SSL if the --tls flag is set
if args.tls:
import ssl
client.tls_set()
#client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2)
#client.tls_insecure_set(False)
# Set the callbacks
client.on_connect = on_connect
client.on_message = on_message
client.on_subscribe = on_subscribe
client.on_unsubscribe = on_unsubscribe
# Connect to the broker
client.connect(args.broker, args.port, 60)
# Subscribe to the root topic
client.subscribe(args.root, 0)
# Start the keep-alive loop
client.loop_forever()