mqtt_sniffer/mqtt_sniffer.py

import argparse
try:
    import bottombar as bb
except ModuleNotFoundError:
    bb = None
try:
    from console_bottombar import BottomBar
except ModuleNotFoundError:
    BottomBar = None
import getpass
import json
import mqtt
import os
import re
import time

VERSION = "0.2.0"
logfile = None

HELPTEXT = """
F1: Get this help message
F2: Set a filter (regular expression) for the topic of a message
    Examples:
    * "/gfw/.*/main_light/" to get everything with "/gfw/" before "/main_light/"
    * "^zigbee.*(?>!logging)$" to get everything starting with "zigbee" and not ending with "logging"
    * "^(?!shellies).*/dirk/.*temperature$" to get everything not starting with "shellies" followed by "/dirk/" and ends with "temperature"
F9: Start / Stop logging to mqtt-sniffer.log
F12: Quit the mqtt sniffer

'c': Clear screen
'q': Quit
"""


def msg_print_log(message, topic_regex, log2file):
    global logfile

    try:
        match = len(re.findall(topic_regex, message.topic)) > 0
    except re.error:
        print('No valid regular expression (%s). No filter active.' % topic_regex)
        match = True

    data = None
    try:
        data = json.loads(message.payload)
    except:
        pass
    try:
        data = message.payload.decode("utf-8")
    except:
        data = message.payload

    if match:
        print("%25s::%75s::%s" % (time.asctime(), message.topic, data))
        if log2file:
            if logfile is None:
                logfile = open('mqtt_sniffer.log', 'w')
            logfile.write("%25s::%s::%s\n" % (time.asctime(), message.topic, data))
            logfile.flush()



def rx_mqtt(mc, userdata, message):
    global my_bb
    global args

    if BottomBar is not None and bb is not None:
        msg_print_log(message, my_bb.get_entry('msg_re'), my_bb.get_entry('log2file'))
        bb.redraw()
    else:
        msg_print_log(message, args.topicfilter, args.logtofile)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='This is a mqtt sniffer.')
    parser.add_argument('-f', dest='hostname', default='localhost', help='Hostname of the mqtt server')
    parser.add_argument('-p', dest='port', default=1883, help='Port of the mqtt server')
    parser.add_argument('-n', dest='no_credentials', action='store_true', help='Avoid asking for credentials')
    parser.add_argument('-u', dest='username', default=None, help='Set username for mqtt server')
    parser.add_argument('-t', dest='topicfilter', default="", help='Set topic filter')
    parser.add_argument('-l', dest='logtofile', action='store_true', help='Enable logging to file')

    args = parser.parse_args()

    if not args.no_credentials:
        if args.username == None:
            args.username = input("Username: ")
        password = getpass.getpass(prompt='Password: ', stream=None)
    else:
        args.username = None
        password = None

    if BottomBar != None:
        my_bb = BottomBar(VERSION, label='MQTT-Sniffer')
        my_bb.add_entry('help', 1, my_bb.FUNC_INFO, label='[F1] Help', infotext=HELPTEXT)
        my_bb.add_entry('msg_re', 2, my_bb.FUNC_TEXT, label='[F2] Filter', default=args.topicfilter)
        my_bb.add_entry('quit', 12, my_bb.FUNC_QUIT, "Quit", label='[F12]', right=True)
        my_bb.add_entry('log2file', 9, my_bb.FUNC_BOOL, label='[F9] Log2File', default=args.logtofile, right=True)

    mc = mqtt.mqtt_client("mqtt_sniffer", args.hostname, port=args.port, username=args.username, password=password)
    mc.add_callback("#", rx_mqtt)
    if BottomBar != None:
        my_bb.run()
    else:
        while True:
            time.sleep(1)