 Easy MQTT Handler is a MQTT Client implemented in Python3 that will connect to a MQTT Broker, listening for freely configurable messages and reacting by executing arbitrary pre-defined commands. It comes with a Qt5-based GUI.
#! /usr/bin/python3
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  __main__.py
*
*  Main class of the program
*
*  Copyright (C) 2023 A. Zeil
"""
import sys
import argparse

from PyQt5.QtWidgets import QApplication
from easy_mqtt_handler.qt.MainWindow import MainWindow
from easy_mqtt_handler.util.Tools import Utils


# entry point
if __name__ == "__main__":

    arguments = argparse.ArgumentParser(description="Easy MQTT Handler")
    arguments.add_argument("-mqtt-conf", "--mqtt-configuration-file", type=str, default="")
    arguments.add_argument("-payload-conf", "--payload-configuration-file", type=str, default="")
    args = arguments.parse_args()

    # create configuration folder, if it doesn't exist, yet
    firstStart = Utils.create_path_if_not_exists(Utils.get_config_path())

    # create the application
    app = QApplication(sys.argv)

    # create the main window
    main_window = MainWindow(app, args.mqtt_configuration_file, args.payload_configuration_file)

    # if this is our first run, show the main window
    if firstStart:
        main_window.show()

    # run the application
    sys.exit(app.exec_())

src/easy_mqtt_handler/__main__.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  qt/AboutDialog.py
*
*  Defines the About Dialog
*
*  Copyright (C) 2023 A. Zeil
"""
import gettext

from PyQt5.QtWidgets import QWidget, QTabWidget, QTextEdit, QVBoxLayout, QPushButton

from easy_mqtt_handler.util.Icons import *
from easy_mqtt_handler.util.Tools import Utils

MY_LICENSE = Utils.resource_path("licenses/COPYING")
PAHO_LICENSE = Utils.resource_path("licenses/paho.txt")
QT5_LICENSE = Utils.resource_path("licenses/qt5.txt")
BRIEFCASE_LICENSE = Utils.resource_path("licenses/briefcase.txt")
CRYPTOGRAPHY_LICENSE = Utils.resource_path("licenses/cryptography.txt")
PYOPENSSL_LICENSE = Utils.resource_path("licenses/pyopenssl.txt")
HUMANITY_ICONS_LICENSE = Utils.resource_path("licenses/humanity.txt")

# Set the local directory
localedir = Utils.resource_path("./locale")

# Set up your magic function
translate = gettext.translation('AboutDialog', localedir, fallback=True)
_ = translate.gettext


class AboutTabMain(QWidget):
    def __init__(self):
        super().__init__()

        # create the multi-line text box
        self.text_box = QTextEdit(self)
        self.text_box.setReadOnly(True)

        # make it always fill the QWidget
        layout = QVBoxLayout()
        layout.addWidget(self.text_box)
        self.setLayout(layout)

        self.text_box.setText(Utils.load_license_file(MY_LICENSE))


class LicenseTab(QWidget):

    def __init__(self, license_file):
        super().__init__()

        # create the multi-line text box
        self.text_box = QTextEdit(self)
        self.text_box.setReadOnly(True)

        # make it always fill the QWidget
        layout = QVBoxLayout()
        layout.addWidget(self.text_box)
        self.setLayout(layout)

        self.text_box.setText(Utils.load_license_file(license_file))


class AboutDialog(QWidget):

    def __init__(self, app):
        super().__init__()

        self.setWindowTitle(_("About Program"))

        # load the dialog icon
        self.setWindowIcon(Icons.load_icon(APP_ICON))

        # create tab container widget
        self.tab_widget = QTabWidget(self)

        # create the tabs
        self.main_tab = AboutTabMain()
        self.paho_tab = LicenseTab(PAHO_LICENSE)
        self.qt_tab = LicenseTab(QT5_LICENSE)
        self.briefcase_tab = LicenseTab(BRIEFCASE_LICENSE)
        self.cryptography_tab = LicenseTab(CRYPTOGRAPHY_LICENSE)
        self.pyopenssl_tab = LicenseTab(PYOPENSSL_LICENSE)
        self.humanity_icons_tab = LicenseTab(HUMANITY_ICONS_LICENSE)

        # add them to the tab widget
        self.tab_widget.addTab(self.main_tab, _('About Easy MQTT Handler'))
        self.tab_widget.addTab(self.paho_tab, _('paho mqtt License'))
        self.tab_widget.addTab(self.qt_tab, _('Qt 5 License'))
        self.tab_widget.addTab(self.briefcase_tab, _('briefcase License'))
        self.tab_widget.addTab(self.cryptography_tab, _('cryptography License'))
        self.tab_widget.addTab(self.pyopenssl_tab, _('pyOpenSSL License'))
        self.tab_widget.addTab(self.humanity_icons_tab, _('Humanity Icons License'))

        self.close_button = QPushButton(_('Close'))
        self.close_button.clicked.connect(self.close)

        layout = QVBoxLayout()
        layout.addWidget(self.tab_widget)
        layout.addWidget(self.close_button)
        self.setLayout(layout)

        # resize and reposition about dialog
        self.width = app.primaryScreen().size().width() // 4
        self.height = (app.primaryScreen().size().height() // 4) + 75
        pos_x = app.primaryScreen().size().width() // 2
        pos_y = app.primaryScreen().size().height() // 2
        self.setGeometry(pos_x - (self.width // 2), pos_y - (self.height // 2), self.width, self.height)

src/easy_mqtt_handler/qt/AboutDialog.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  qt/MainWindow.py
*
*  Main window of the program based on QMainWindow (QT5)
*
*  Copyright (C) 2023 A. Zeil
"""
import gettext
import sys

from PyQt5.QtCore import QSize, QThread
from PyQt5.QtWidgets import QMainWindow, QToolBar, QAction, QTabWidget, QSystemTrayIcon, QMenu, QStatusBar, QLabel, \
    QMessageBox

from easy_mqtt_handler.qt.AboutDialog import AboutDialog
from easy_mqtt_handler.qt.tabs.ConnectionTabWidget import ConnectionTabWidget
from easy_mqtt_handler.qt.tabs.LogTabWidget import LogTabWidget
from easy_mqtt_handler.qt.tabs.PayloadTabWidget import PayloadTabWidget

from easy_mqtt_handler.util.Icons import *
from easy_mqtt_handler.util.MQTTPayloads import MQTTPayloads
from easy_mqtt_handler.util.MQTTSettings import MQTTSettings
from easy_mqtt_handler.util.MQTTWorkerThread import MQTTWorkerThread
from easy_mqtt_handler.util.Tools import Utils

# Set the local directory
localedir = Utils.resource_path("./locale")

# Set up your magic function
translate = gettext.translation("MainWindow", localedir, fallback=True)
_ = translate.gettext

# consts go here
PROG_NAME = _("Easy MQTT Handler")

# get locations of config files
config_path = Utils.get_config_path()
payload_file = config_path + "default-payloads.json"
settings_file = config_path + "default-settings.json"


class MainWindow(QMainWindow):
    tab_widget: QTabWidget
    settings: MQTTSettings
    payloads: MQTTPayloads
    unsaved_changes = False

    """
    Main Window init()
    """

    def __init__(self, app, mqtt_config_file, payload_config_file):
        super().__init__()

        self.worker_thread: QThread

        self.setWindowTitle(PROG_NAME)
        app.setQuitOnLastWindowClosed(False)

        # load current configs
        self.settings = MQTTSettings(mqtt_config_file if mqtt_config_file != "" else settings_file)
        self.payloads = MQTTPayloads(payload_config_file if payload_config_file != "" else payload_file)

        # load the dialog icon
        self.setWindowIcon(Icons.load_icon(APP_ICON))

        # create toolbar
        toolbar = QToolBar(self)
        toolbar.setIconSize(QSize(32, 32))
        self.addToolBar(toolbar)

        # create tab container widget
        self.tab_widget = QTabWidget()
        self.setCentralWidget(self.tab_widget)

        # as long as we init. the tabs, block signals
        self.tab_widget.blockSignals(True)

        # create the tabs
        self.log_tab = LogTabWidget()
        self.connection_tab = ConnectionTabWidget()
        self.payload_editor = PayloadTabWidget()

        # add them to the tab widget
        self.tab_widget.addTab(self.connection_tab, _("Connection"))
        self.tab_widget.addTab(self.payload_editor, _("Payload Handlers"))
        self.tab_widget.addTab(self.log_tab, _("Logs"))

        # wire change handlers
        self.connection_tab.settings_changed.connect(self.on_connection_settings_changed)
        self.payload_editor.settings_changed.connect(self.on_payloads_changed)

        # now that init is done receive signals again
        self.tab_widget.blockSignals(False)

        # create a worker thread and connect the signal to update the main form
        self.worker_thread = MQTTWorkerThread()
        self.worker_thread.add_log_line.connect(self.add_log_line)
        self.worker_thread.track_status.connect(self.track_status)

        # start the worker thread
        self.worker_thread.start()

        # create an About dialog ... we might need it later
        self.about_dialog = AboutDialog(app)

        # add actions to toolbar
        self.save_action = QAction(Icons.load_icon(TOOLBAR_SAVE), _("Save"), self)
        self.save_action.triggered.connect(self.save_settings)
        self.save_action.setVisible(False)
        toolbar.addAction(self.save_action)

        self.action_seperator = toolbar.addSeparator()
        self.action_seperator.setVisible(False)

        self.connect_action = QAction(Icons.load_icon(TOOLBAR_CONNECT), _("Connect"), self)
        self.connect_action.triggered.connect(self.on_connect_action)
        toolbar.addAction(self.connect_action)

        self.disconnect_action = QAction(Icons.load_icon(TOOLBAR_DISCONNECT), _("Disconnect"), self)
        self.disconnect_action.triggered.connect(self.worker_thread.mqtt_disconnect)
        toolbar.addAction(self.disconnect_action)

        self.reconnect_action = QAction(Icons.load_icon(TOOLBAR_RECONNECT), _("Reconnect"), self)
        self.reconnect_action.triggered.connect(self.worker_thread.mqtt_reconnect)
        toolbar.addAction(self.reconnect_action)

        toolbar.addSeparator()

        self.about_action = QAction(Icons.load_icon(TOOLBAR_ABOUT), _("About"), self)
        self.about_action.triggered.connect(self.about_dialog.show)
        toolbar.addAction(self.about_action)

        self.quit_action = QAction(Icons.load_icon(TOOLBAR_QUIT), _("Quit"), self)
        self.quit_action.triggered.connect(self.on_close)
        toolbar.addAction(self.quit_action)

        # let's create a statusbar
        self.statusbar = QStatusBar()
        self.setStatusBar(self.statusbar)
        self.status_icon = QLabel()
        self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_DISCONNECTED).pixmap(32))
        self.status_icon.setToolTip(_("Disconnected"))
        self.statusbar.addPermanentWidget(self.status_icon)

        # let's set up a tray icon
        self.tray_icon = QSystemTrayIcon(self)
        self.tray_icon.setIcon(Icons.load_icon(APP_ICON))
        self.tray_icon.setToolTip(PROG_NAME)
        self.tray_icon.setVisible(True)

        # add a tiny context menu to the tray icon
        tray_menu = QMenu(self)

        # assign some actions to the context menu
        action_open = tray_menu.addAction(_("Open"))
        action_open.triggered.connect(self.on_tray_open)

        action_close = tray_menu.addAction(_("Close"))
        action_close.triggered.connect(self.on_close)

        # add the context menu to the tray icon
        self.tray_icon.setContextMenu(tray_menu)

        # wire event handler
        self.tray_icon.activated.connect(self.on_tray_icon_activated)

        # resize and reposition main window
        self.width = app.primaryScreen().size().width() // 4
        self.height = app.primaryScreen().size().height() // 4
        pos_x = app.primaryScreen().size().width() // 2
        pos_y = app.primaryScreen().size().height() // 2
        self.setGeometry(pos_x - (self.width // 2), pos_y - (self.height // 2), self.width, self.height)

    """
    Event Handlers
    """
    def on_close(self):
        if self.unsaved_changes:
            confirm_quit = QMessageBox(self)
            confirm_quit.setWindowTitle(_("Unsaved changes detected."))
            confirm_quit.setText(_("You haven't saved your changes, yet. Do you still want to quit?"))
            confirm_quit.setStandardButtons(QMessageBox.Ok | QMessageBox.Save | QMessageBox.Cancel)
            selected_button = confirm_quit.exec()
            if selected_button == QMessageBox.Ok:
                sys.exit(0)
            elif selected_button == QMessageBox.Save:
                self.save_settings()
                confirm_final_quit = QMessageBox(self)
                confirm_final_quit.setWindowTitle(_("Settings saved."))
                confirm_final_quit.setText(_("Do you want to quit the application now?"))
                confirm_final_quit.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
                selected_button = confirm_final_quit.exec()
                if selected_button == QMessageBox.Yes:
                    sys.exit(0)
                elif selected_button == QMessageBox.No:
                    return

            elif selected_button == QMessageBox.Cancel:
                return
        else:
            sys.exit(0)

    # will be called when the tray icon is interacted with
    def on_tray_icon_activated(self, reason):
        if reason == QSystemTrayIcon.Trigger:
            self.show()

    def on_tray_open(self):
        self.show()
        self.activateWindow()
        self.raise_()

    def on_connection_settings_changed(self, changed):
        self.setWindowTitle("*" + PROG_NAME)
        self.tab_widget.setTabText(0, "*" + _("Connection"))
        self.unsaved_changes = True
        self.save_action.setVisible(True)
        self.action_seperator.setVisible(True)

    def on_payloads_changed(self, changed):
        self.setWindowTitle("*" + PROG_NAME)
        self.tab_widget.setTabText(1, "*" + _("Payload Handlers"))
        self.unsaved_changes = True
        self.save_action.setVisible(True)
        self.action_seperator.setVisible(True)

    def on_connect_action(self):
        if self.worker_thread.mqtt_connect():
            self.connect_action.setEnabled(False)
            self.disconnect_action.setVisible(True)
            self.add_log_line(_("Connecting to Broker, please wait ..."))

    def add_log_line(self, text):
        logline = f"[{Utils.get_timestamp()}] {text}"
        self.statusbar.showMessage(logline)
        self.log_tab.text_box.append(logline)
        self.statusbar.setToolTip(text)

    def track_status(self, code):
        match code:
            case 100:
                self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_DISCONNECTED).pixmap(32))
                self.status_icon.setToolTip(_("Disconnected"))

            case 102 | 103 | 104:
                self.disconnect_action.setVisible(False)
                self.reconnect_action.setVisible(False)
                self.tab_widget.setCurrentIndex(2)

            case 101:
                self.connect_action.setVisible(False)
                self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_CONNECTED).pixmap(32))
                self.status_icon.setToolTip(_("Connected"))
                self.connect_action.setEnabled(True)

            case 105:
                self.connect_action.setEnabled(True)

            case 200:
                self.tab_widget.setCurrentIndex(0)

            case 300:
                self.disconnect_action.setVisible(True)
                self.reconnect_action.setVisible(True)
                self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_CONNECTED).pixmap(32))
                self.status_icon.setToolTip(_("Connected"))

            case 301:
                self.connect_action.setVisible(True)
                self.disconnect_action.setVisible(False)
                self.reconnect_action.setVisible(False)
                self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_DISCONNECTED).pixmap(32))
                self.status_icon.setToolTip(_("Disconnected"))

            case 302:
                self.connect_action.setVisible(False)
                self.disconnect_action.setVisible(True)
                self.reconnect_action.setVisible(True)
                self.status_icon.setPixmap(Icons.load_icon(STATUSBAR_DISCONNECTED).pixmap(32))
                self.status_icon.setToolTip(_("Disconnected"))

            case _:
                self.add_log_line(_("Statuscode {0} not implemented!").format(code))

    def closeEvent(self, event):
        # don't allow to close the main window
        event.ignore()
        self.hide()

    """
    Helpers
    """

    def save_settings(self):
        if MQTTSettings.get_instance().save_settings():
            # remove mark from tab title
            self.tab_widget.setTabText(0, _("Connection"))

        if MQTTPayloads.get_instance().save_payload_data():
            # remove mark from tab title
            self.tab_widget.setTabText(1, _("Payload Handlers"))

        self.setWindowTitle(PROG_NAME)
        self.unsaved_changes = False
        self.save_action.setVisible(False)
        self.action_seperator.setVisible(False)

src/easy_mqtt_handler/qt/MainWindow.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  qt/tabs/ConnectionTabWidget.py
*
*  Defines the connection settings tab
*
*  Copyright (C) 2023 A. Zeil
"""
import gettext
import os

from PyQt5 import QtGui
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLineEdit, QLabel, QCheckBox, QHBoxLayout, QFileDialog, QPushButton
from easy_mqtt_handler.util.MQTTSettings import MQTTSettings
from easy_mqtt_handler.util.Tools import Utils

# Set the local directory
localedir = Utils.resource_path("./locale")

# Set up your magic function
translate = gettext.translation("ConnectionTabWidget", localedir, fallback=True)
_ = translate.gettext


class ConnectionTabWidget(QWidget):
    settings_changed = pyqtSignal(bool)

    def __init__(self):
        super().__init__()

        # Create the textboxes and labels
        self.hostname_label = QLabel(_("Hostname:"))
        self.hostname_textbox = QLineEdit()
        self.port_label = QLabel(_("Port:"))
        self.port_textbox = QLineEdit()

        self.insecure_ssl_checkbox = QCheckBox()
        self.insecure_ssl_label = QLabel(_("Allow insecure SSL/TLS"))
        self.insecure_ssl_label.setStyleSheet("border: 1px solid red;")

        self.insecure_ssl_layout = QHBoxLayout()
        self.insecure_ssl_layout.setSpacing(10)
        self.insecure_ssl_layout.addWidget(self.insecure_ssl_checkbox)
        self.insecure_ssl_layout.addWidget(self.insecure_ssl_label)
        self.insecure_ssl_layout.addStretch()

        self.username_label = QLabel(_("Username:"))
        self.username_textbox = QLineEdit()
        self.password_label = QLabel(_("Password:"))
        self.password_textbox = QLineEdit()
        # Set the password textbox to hide input
        self.password_textbox.setEchoMode(QLineEdit.Password)
        self.topic_label = QLabel(_("MQTT Topic:"))
        self.topic_textbox = QLineEdit()

        self.enable_ssl_checkbox = QCheckBox()
        self.enable_ssl_label = QLabel(_("Use SSL/TLS"))
        self.enable_ssl_layout = QHBoxLayout()
        self.enable_ssl_layout.setSpacing(10)
        self.enable_ssl_layout.addWidget(self.enable_ssl_checkbox)
        self.enable_ssl_layout.addWidget(self.enable_ssl_label)
        self.enable_ssl_layout.addStretch()

        self.server_certificate_label = QLabel(_("Server Certificate / Certificate Chain:"))
        self.server_certificate_file = QLineEdit()
        self.server_certificate_browse_btn = QPushButton('...')
        self.server_certificate_browse_btn.clicked.connect(lambda: self.cryptofile_browse("server-cert"))
        self.server_certificate_browse_btn.setFixedWidth(40)

        self.server_certificate_layout = QHBoxLayout()
        self.server_certificate_layout.setSpacing(10)
        self.server_certificate_layout.addWidget(self.server_certificate_file)
        self.server_certificate_layout.addWidget(self.server_certificate_browse_btn)

        self.enable_client_ssl_auth_checkbox = QCheckBox()
        self.enable_client_ssl_auth_label = QLabel(_("Enable client certificate authentication"))
        self.enable_client_ssl_auth_layout = QHBoxLayout()
        self.enable_client_ssl_auth_layout.setSpacing(10)
        self.enable_client_ssl_auth_layout.addWidget(self.enable_client_ssl_auth_checkbox)
        self.enable_client_ssl_auth_layout.addWidget(self.enable_client_ssl_auth_label)
        self.enable_client_ssl_auth_layout.addStretch()

        self.client_certificate_label = QLabel(_("Client certificate file:"))
        self.client_certificate_file = QLineEdit()
        self.client_certificate_browse_btn = QPushButton('...')
        self.client_certificate_browse_btn.clicked.connect(lambda: self.cryptofile_browse("client-cert"))
        self.client_certificate_browse_btn.setFixedWidth(40)

        self.client_certificate_layout = QHBoxLayout()
        self.client_certificate_layout.setSpacing(10)
        self.client_certificate_layout.addWidget(self.client_certificate_file)
        self.client_certificate_layout.addWidget(self.client_certificate_browse_btn)

        self.client_key_label = QLabel(_("Client key file:"))
        self.client_key_file = QLineEdit()
        self.client_key_browse_btn = QPushButton('...')
        self.client_key_browse_btn.clicked.connect(lambda: self.cryptofile_browse("client-key"))
        self.client_key_browse_btn.setFixedWidth(40)

        self.client_key_layout = QHBoxLayout()
        self.client_key_layout.setSpacing(10)
        self.client_key_layout.addWidget(self.client_key_file)
        self.client_key_layout.addWidget(self.client_key_browse_btn)

        self.client_ssl_auth_widget = QWidget()
        self.layout_client_ssl_auth_settings = QVBoxLayout(self.client_ssl_auth_widget)
        self.layout_client_ssl_auth_settings.addWidget(self.client_certificate_label)
        self.layout_client_ssl_auth_settings.addLayout(self.client_certificate_layout)
        self.layout_client_ssl_auth_settings.addWidget(self.client_key_label)
        self.layout_client_ssl_auth_settings.addLayout(self.client_key_layout)

        self.ssl_settings_widget = QWidget()
        self.layout_ssl_settings = QVBoxLayout(self.ssl_settings_widget)
        self.layout_ssl_settings.addWidget(self.server_certificate_label)
        self.layout_ssl_settings.addLayout(self.server_certificate_layout)
        self.layout_ssl_settings.addLayout(self.insecure_ssl_layout)
        self.layout_ssl_settings.addLayout(self.enable_client_ssl_auth_layout)
        self.layout_ssl_settings.addWidget(self.client_ssl_auth_widget)

        # Create a layout for the textboxes and labels
        layout = QVBoxLayout()
        layout.addWidget(self.hostname_label)
        layout.addWidget(self.hostname_textbox)
        layout.addWidget(self.port_label)
        layout.addWidget(self.port_textbox)
        layout.addWidget(self.username_label)
        layout.addWidget(self.username_textbox)
        layout.addWidget(self.password_label)
        layout.addWidget(self.password_textbox)
        layout.addWidget(self.topic_label)
        layout.addWidget(self.topic_textbox)
        layout.addLayout(self.enable_ssl_layout)
        layout.addWidget(self.ssl_settings_widget)

        layout.addStretch(1)
        # Set the layout for the QWidget
        self.setLayout(layout)

    def cryptofile_browse(self, cert_type):

        # set some options for the file open dialog if we are on windows
        if os.name == "nt":
            start_dir = "C:\\"
            file_filter = "*.*"
        # or on linux and macOS
        else:
            start_dir = "/"
            file_filter = "*"

        filedialog = QFileDialog()

        if cert_type == "client-cert":
            filedialog.setWindowTitle(_('Select a client certificate'))
        elif cert_type == "server-cert":
            filedialog.setWindowTitle(_('Select a server certificate'))
        elif cert_type == "client-key":
            filedialog.setWindowTitle(_('Select client key file'))

        filedialog.setDirectory(start_dir)
        filedialog.setFileMode(QFileDialog.FileMode.ExistingFile)
        filedialog.setNameFilter(file_filter)
        filedialog.setViewMode(QFileDialog.ViewMode.List)

        if filedialog.exec_() and len(filedialog.selectedFiles()) == 1:
            selected_file = filedialog.selectedFiles()[0]
            if cert_type == "client-cert":
                self.client_certificate_file.setText(selected_file)
            elif cert_type == "server-cert":
                self.server_certificate_file.setText(selected_file)
            elif cert_type == "client-key":
                self.client_key_file.setText(selected_file)
            self.settings_changed.emit(True)

    def setting_changed_event(self):
        self.settings_changed.emit(True)
        self.set_new_connection_settings()

    def enable_ssl_changed_event(self):
        self.ssl_settings_widget.setHidden(not self.enable_ssl_checkbox.isChecked())
        self.setting_changed_event()

    def enable_client_ssl_auth_changed_event(self):
        self.client_ssl_auth_widget.setHidden(not self.enable_client_ssl_auth_checkbox.isChecked())
        self.setting_changed_event()

    def set_new_connection_settings(self):
        settings_data = []

        # Create a dictionary from the textbox data
        settings_data = {
            "hostname": self.hostname_textbox.text(),
            "port": self.port_textbox.text(),
            "username": self.username_textbox.text(),
            "password": self.password_textbox.text(),
            "topic": self.topic_textbox.text(),
            "enable_ssl": self.enable_ssl_checkbox.isChecked(),
            "server_certificate_file": self.server_certificate_file.text(),
            "enable_client_ssl_auth": self.enable_client_ssl_auth_checkbox.isChecked(),
            "client_certificate_file": self.client_certificate_file.text(),
            "client_key_file": self.client_key_file.text(),
            "allow_insecure_ssl": self.insecure_ssl_checkbox.isChecked()
        }

        MQTTSettings.get_instance().refresh_settings(settings_data)

    def showEvent(self, a0: QtGui.QShowEvent) -> None:
        settings = MQTTSettings.get_instance()

        self.hostname_textbox.setText(settings.hostname)
        self.port_textbox.setText(settings.port)
        self.username_textbox.setText(settings.username)
        self.password_textbox.setText(settings.password)
        self.topic_textbox.setText(settings.topic)
        self.enable_ssl_checkbox.setChecked(settings.enable_ssl)
        self.server_certificate_file.setText(settings.server_certificate_file)
        self.enable_client_ssl_auth_checkbox.setChecked(settings.enable_client_ssl_auth)
        self.client_certificate_file.setText(settings.client_certificate_file)
        self.client_key_file.setText(settings.client_key_file)
        self.insecure_ssl_checkbox.setChecked(settings.allow_insecure_ssl)

        # now that we've loaded the settings, enable event handlers to catch changes
        self.hostname_textbox.textChanged.connect(self.setting_changed_event)
        self.port_textbox.textChanged.connect(self.setting_changed_event)
        self.username_textbox.textChanged.connect(self.setting_changed_event)
        self.password_textbox.textChanged.connect(self.setting_changed_event)
        self.topic_textbox.textChanged.connect(self.setting_changed_event)
        self.enable_ssl_checkbox.stateChanged.connect(self.enable_ssl_changed_event)
        self.server_certificate_file.textChanged.connect(self.setting_changed_event)
        self.enable_client_ssl_auth_checkbox.stateChanged.connect(self.enable_client_ssl_auth_changed_event)
        self.client_certificate_file.textChanged.connect(self.setting_changed_event)
        self.client_key_file.textChanged.connect(self.setting_changed_event)
        self.insecure_ssl_checkbox.stateChanged.connect(self.setting_changed_event)

        # check whether we need to show SSL/TLS settings
        self.ssl_settings_widget.setHidden(not self.enable_ssl_checkbox.isChecked())
        self.client_ssl_auth_widget.setHidden(not self.enable_client_ssl_auth_checkbox.isChecked())

src/easy_mqtt_handler/qt/tabs/ConnectionTabWidget.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  qt/tabs/LogTabWidget.py
*
*  Defines the Logging tab
*
*  Copyright (C) 2023 A. Zeil
"""
from PyQt5.QtWidgets import QWidget, QTextEdit, QVBoxLayout


class LogTabWidget(QWidget):
    def __init__(self):
        super().__init__()

        # create the multi-line text box
        self.text_box = QTextEdit(self)
        self.text_box.setReadOnly(True)

        # make it always fill the QWidget
        layout = QVBoxLayout()
        layout.addWidget(self.text_box)
        self.setLayout(layout)

src/easy_mqtt_handler/qt/tabs/LogTabWidget.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  qt/tabs/PayloadTabWidget.py
*
*  Defines the Payload Editor Tab
*
*  Copyright (C) 2023 A. Zeil
"""
import gettext
import os

from PyQt5 import QtGui
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtWidgets import QWidget, QTableWidget, QAbstractItemView, QPushButton, QVBoxLayout, QHBoxLayout, \
    QHeaderView, QSizePolicy, QTableWidgetItem, QFileDialog

from easy_mqtt_handler.util.MQTTPayloads import MQTTPayloads
from easy_mqtt_handler.util.Tools import Utils

# Set the local directory
localedir = Utils.resource_path("./locale")

# Set up your magic function
translate = gettext.translation("PayloadTabWidget", localedir, fallback=True)
_ = translate.gettext


class PayloadTabWidget(QWidget):

    settings_changed = pyqtSignal(bool)

    def __init__(self):
        super().__init__()

        # create the table
        self.table = QTableWidget()
        self.table.setColumnCount(5)
        self.table.setHorizontalHeaderLabels([_('Payload Command'), _('Payload Argument'),
                                              _('Command to Run'), "", _('Command line arguments')])

        self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
        self.table.setSelectionMode(QAbstractItemView.SingleSelection)

        # create the buttons
        self.save_button = QPushButton(_('Add Payload'))
        self.save_button.clicked.connect(self.add_payload)
        self.cancel_button = QPushButton(_('Remove Payload'))
        self.cancel_button.clicked.connect(self.remove_payload)

        # create the layout
        layout = QVBoxLayout()
        layout.addWidget(self.table)

        button_layout = QHBoxLayout()
        button_layout.addWidget(self.save_button)
        button_layout.addWidget(self.cancel_button)
        layout.addLayout(button_layout)

        self.setLayout(layout)

        self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeToContents)
        self.table.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
        self.table.horizontalHeader().setStretchLastSection(True)

    def setting_changed_event(self, text):
        self.settings_changed.emit(True)
        self.set_new_payload_data()

    def add_data(self, payload_command, payload_argument, command_to_run, command_line_arguments):
        row_count = self.table.rowCount()
        self.table.setRowCount(row_count + 1)

        self.table.setItem(row_count, 0, QTableWidgetItem(payload_command))
        self.table.setItem(row_count, 1, QTableWidgetItem(payload_argument))
        self.table.setItem(row_count, 2, QTableWidgetItem(command_to_run))

        button = QPushButton("...")
        button.setFixedWidth(40)
        button.setProperty("row", row_count)
        self.table.setCellWidget(row_count, 3, button)
        button.clicked.connect(lambda: self.browse_executable(button))

        self.table.setItem(row_count, 4, QTableWidgetItem(command_line_arguments))

    def browse_executable(self, cur_button):

        # set some options for the file open dialog if we are on windows
        if os.name == "nt":
            start_dir = "C:\\"
            file_filter = "*.*"
        # or on linux and macOS
        else:
            start_dir = "/"
            file_filter = "*"

        filedialog = QFileDialog()
        filedialog.setWindowTitle(_('Select an executable file'))
        filedialog.setDirectory(start_dir)
        filedialog.setFileMode(QFileDialog.FileMode.ExistingFile)
        filedialog.setNameFilter(file_filter)
        filedialog.setViewMode(QFileDialog.ViewMode.List)

        if filedialog.exec_() and len(filedialog.selectedFiles()) == 1:
            selected_file = filedialog.selectedFiles()[0]
            self.table.item(cur_button.property("row"), 2).setText(selected_file)
            self.table.viewport().update()

    def add_payload(self):
        row_count = self.table.rowCount()
        self.table.setRowCount(row_count + 1)

        self.table.setItem(row_count, 0, QTableWidgetItem(""))
        self.table.setItem(row_count, 1, QTableWidgetItem(""))
        self.table.setItem(row_count, 2, QTableWidgetItem(""))

        button = QPushButton("...")
        button.setFixedWidth(40)
        button.setProperty("row", row_count)
        self.table.setCellWidget(row_count, 3, button)
        button.clicked.connect(lambda: self.browse_executable(button))

        self.setting_changed_event(True)

    def set_new_payload_data(self):
        # ensure payload data is empty
        new_payload_data = []
        # payload_data.clear()

        # for each line of the table append one item to the payload config
        for row in range(self.table.rowCount()):
            payload_command = "" if self.table.item(row, 0) is None else self.table.item(row, 0).text()
            payload_argument = "" if self.table.item(row, 1) is None else self.table.item(row, 1).text()
            command_to_run = "" if self.table.item(row, 2) is None else self.table.item(row, 2).text()
            command_line_arguments = "" if self.table.item(row, 4) is None else self.table.item(row, 4).text()
            new_payload_data.append({
                'payload_command': payload_command,
                'payload_argument': payload_argument,
                'command_to_run': command_to_run,
                'command_line_arguments': command_line_arguments
            })

        MQTTPayloads.get_instance().payload_data = new_payload_data

    def remove_payload(self):
        selected_row = self.table.currentRow()
        self.table.removeRow(selected_row)
        self.setting_changed_event(True)

    def showEvent(self, a0: QtGui.QShowEvent) -> None:
        # unbind dataChanged event until we've loaded the new payload data
        self.table.model().dataChanged.disconnect()

        payload_settings = MQTTPayloads.get_instance().payload_data

        # clear the table to get a fresh copy of the payload config
        self.table.clearContents()
        self.table.setRowCount(0)

        # fill table with current payload config
        for item in payload_settings:
            self.add_data(str(item['payload_command']),
                          str(item['payload_argument']),
                          str(item['command_to_run']),
                          str(item['command_line_arguments']))

        # now that we've loaded data: enable listening to dataChanged event and send a signal on changes
        self.table.model().dataChanged.connect(self.setting_changed_event)

src/easy_mqtt_handler/qt/tabs/PayloadTabWidget.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  util/Icons.py
*
*  Class to handle program icons
*
*  Copyright (C) 2023 A. Zeil
"""
from PyQt5.QtGui import QIcon

from easy_mqtt_handler.util.Tools import Utils

# the default icons are all part of the "Humanity" Icon set (see license-humanity.txt)
APP_ICON = Utils.resource_path("assets/app-icon/app-icon.svg")

BUTTON_ADD = Utils.resource_path("assets/button/add.svg")
BUTTON_CLEAR = Utils.resource_path("assets/button/clear.svg")
BUTTON_COPY = Utils.resource_path("assets/button/copy.svg")
BUTTON_REMOVE = Utils.resource_path("assets/button/remove.svg")

TOOLBAR_ABOUT = Utils.resource_path("assets/toolbar/about.svg")
TOOLBAR_CONNECT = Utils.resource_path("assets/toolbar/connect.svg")
TOOLBAR_DISCONNECT = Utils.resource_path("assets/toolbar/disconnect.svg")
TOOLBAR_QUIT = Utils.resource_path("assets/toolbar/quit.svg")
TOOLBAR_RECONNECT = Utils.resource_path("assets/toolbar/reconnect.svg")
TOOLBAR_SAVE = Utils.resource_path("assets/toolbar/save.svg")

STATUSBAR_CONNECTED = Utils.resource_path("assets/statusbar/connected.svg")
STATUSBAR_DISCONNECTED = Utils.resource_path("assets/statusbar/disconnected.svg")


class Icons(object):
    @staticmethod
    def load_icon(icon_filename):
        try:
            return QIcon(icon_filename)
        # TODO: implement better exception handling
        except:
            return None

src/easy_mqtt_handler/util/Icons.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  util/MQTTPayloads.py
*
*  Class to handle MQTT payloads
*
*  Copyright (C) 2023 A. Zeil
"""
import json
import os


class MQTTPayloads(object):
    @property
    def payload_data(self):
        return self._payload_data

    @payload_data.setter
    def payload_data(self, value):
        self._payload_data = value

    _instance = None
    _payload_file = ""

    _payload_data = {}

    @staticmethod
    def get_instance():
        if MQTTPayloads._instance is None:
            MQTTPayloads()
        return MQTTPayloads._instance

    def __init__(self, filename):
        if MQTTPayloads._instance is not None:
            raise Exception("This is a Singleton Class. Only once instance allowed!")
        else:
            MQTTPayloads._instance = self
            self._payload_file = filename
            self._payload_data = self.load_payload_data()

    def load_payload_data(self):
        self._payload_data.clear()

        if os.path.exists(self._payload_file):
            try:
                pf = open(self._payload_file, 'r')
                return json.load(pf)
            # TODO: implement better exception handling
            except IOError:
                # self.add_log_line("Payload File couldn't be loaded.")
                return ""
        else:
            # self.add_log_line("Payload File doesn't exist, yet.")
            return ""

    def save_payload_data(self):
        # create a JSON object from the data and save it to the payload config file
        try:
            with open(self._payload_file, 'w') as pf:
                json.dump(self._payload_data, pf)
                pf.close()

                return True
        # TODO: implement better exception handling
        except:
            return False

src/easy_mqtt_handler/util/MQTTPayloads.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  util/MQTTSettings.py
*
*  Class to handle MQTT connection settings
*
*  Copyright (C) 2023 A. Zeil
"""
import json
import os


class MQTTSettings(object):

    @property
    def hostname(self):
        return self.get_property('hostname')

    @hostname.setter
    def hostname(self, value):
        self.hostname = value

    @property
    def port(self):
        return self.get_property('port')

    @port.setter
    def port(self, value):
        self.port = value

    @property
    def username(self):
        return self.get_property('username')

    @username.setter
    def username(self, value):
        self.username = value

    @property
    def password(self):
        return self.get_property('password')

    @password.setter
    def password(self, value):
        self.password = value

    @property
    def topic(self):
        return self.get_property('topic')

    @topic.setter
    def topic(self, value):
        self.topic = value

    @property
    def enable_ssl(self):
        if self.get_property('enable_ssl') is None:
            return False
        else:
            return self.get_property('enable_ssl')

    @enable_ssl.setter
    def enable_ssl(self, value):
        self.enable_ssl = value

    @property
    def server_certificate_file(self):
        return self.get_property('server_certificate_file')

    @server_certificate_file.setter
    def server_certificate_file(self, value):
        self.server_certificate_file = value

    @property
    def enable_client_ssl_auth(self):
        if self.get_property('enable_client_ssl_auth') is None:
            return False
        else:
            return self.get_property('enable_client_ssl_auth')

    @enable_client_ssl_auth.setter
    def enable_client_ssl_auth(self, value):
        self.enable_client_ssl_auth = value

    @property
    def client_certificate_file(self):
        return self.get_property('client_certificate_file')

    @client_certificate_file.setter
    def client_certificate_file(self, value):
        self.client_certificate_file = value

    @property
    def client_key_file(self):
        return self.get_property('client_key_file')

    @client_key_file.setter
    def client_key_file(self, value):
        self.client_key_file = value

    @property
    def allow_insecure_ssl(self):
        if self.get_property('allow_insecure_ssl') is None:
            return False
        else:
            return self.get_property('allow_insecure_ssl')

    @allow_insecure_ssl.setter
    def allow_insecure_ssl(self, value):
        self.allow_insecure_ssl = value

    _instance = None
    _filename = ""
    _settings = {}

    @staticmethod
    def get_instance():
        if MQTTSettings._instance is None:
            MQTTSettings()
        return MQTTSettings._instance

    def __init__(self, filename):
        if MQTTSettings._instance is not None:
            raise Exception("This is a Singleton Class. Only once instance allowed!")
        else:
            MQTTSettings._instance = self
            self._filename = filename
            self._settings = self.load_settings()

    def get_property(self, property_name):
        try:
            if property_name not in self._settings.keys():
                return None
            return self._settings[property_name]
        except AttributeError:
            return None

    def load_settings(self):
        self._settings.clear()

        if os.path.exists(self._filename):
            try:
                sf = open(self._filename, 'r')
                return json.load(sf)
            # TODO: implement better exception handling
            except IOError:
                # self.add_log_line("Settings File couldn't be loaded.")
                return ""
        else:
            # self.add_log_line("Settings File doesn't exist, yet.")
            return ""

    def refresh_settings(self, new_settings):
        self._settings = new_settings

    def save_settings(self):
        try:
            with open(self._filename, 'w') as sf:
                json.dump(self._settings, sf)
                sf.close()

                return True
        # TODO: implement better exception handling
        except:
            return False

src/easy_mqtt_handler/util/MQTTSettings.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  util/MQTTWorkerThread.py
*
*  Class defines the WorkerThread to handle the MQTT connection
*
*  Copyright (C) 2023 A. Zeil
"""
import gettext
import json
import os
from ssl import SSLError, SSLZeroReturnError

from PyQt5.QtCore import QThread, pyqtSignal
import paho.mqtt.client as mqtt
from paho.mqtt.client import ssl

from easy_mqtt_handler.util.MQTTPayloads import MQTTPayloads
from easy_mqtt_handler.util.MQTTSettings import MQTTSettings
from easy_mqtt_handler.util.Tools import Utils

# Set the local directory
localedir = Utils.resource_path("./locale")

# Set up your magic function
translate = gettext.translation('MQTTWorkerThread', localedir, fallback=True)
_ = translate.gettext


def extract_payload(payload):
    try:
        return json.loads(payload.decode())
    except Exception:
        return payload.decode()


def find_command_to_run(command, arg):
    payload_data = MQTTPayloads.get_instance().payload_data

    for item in payload_data:
        if item['payload_command'] == command and item['payload_argument'] == arg:
            return item['command_to_run']
    return None


def find_command_line_arguments(command, arg):
    payload_data = MQTTPayloads.get_instance().payload_data

    for item in payload_data:
        if item['payload_command'] == command and item['payload_argument'] == arg:
            return item['command_line_arguments']
    return None


class MQTTWorkerThread(QThread):
    # create a signal for adding a log line -> we need this cause we are running in a separate thread
    add_log_line = pyqtSignal(str)
    # create a signal for track status and controlling the MainWindow GUI
    track_status = pyqtSignal(int)

    client = mqtt.Client()
    settings: MQTTSettings

    def on_connect(self, client, userdata, flags, rc):

        self.settings = MQTTSettings.get_instance()

        if rc != 0:
            self.add_log_line.emit(_("Couldn't connect to MQTT Broker \"{0}\".").format(self.settings.hostname))
            self.track_status.emit(100)
            return rc

        self.add_log_line.emit(_("Connected to MQTT Broker \"{0}\" on port {1}.").format(self.settings.hostname,
                                                                                         self.settings.port))
        self.track_status.emit(101)

        try:
            client.subscribe(f"{self.settings.topic}/#")
            self.add_log_line.emit(_("Subscribing to topic \"{0}/#\".").format(self.settings.topic))
        except ValueError:
            self.add_log_line.emit(_("Couldn't subscribe to topic \"{0}/#\". You provided an invalid or emtpy topic.").
                                   format(self.settings.topic))
            self.track_status.emit(103)
        self.add_log_line.emit(_("Listening to Broker."))
        self.track_status.emit(300)

    def router(self, data):
        mqtt_command = data.get("command")
        mqtt_arg = data.get("args")

        if mqtt_command is not None:
            self.add_log_line.emit(_("Received command \"{0}\" with argument \"{1}\".").format(mqtt_command, mqtt_arg))

        command_to_run = find_command_to_run(mqtt_command, mqtt_arg)
        command_line_args = find_command_line_arguments(mqtt_command, mqtt_arg)

        if command_to_run is not None:
            # when there are spaces in the given command we should set it in quotes
            if ' ' in command_to_run:
                command_to_run = f"\"{command_to_run}\""

            self.add_log_line.emit(_("Executing command \"{0}\".").format(command_to_run))
            os.system(f"{command_to_run} {command_line_args}")
            self.add_log_line.emit(_("Command \"{0}\" was executed.").format(command_to_run))
        else:
            if mqtt_command is None:
                self.add_log_line.emit(
                    _("Couldn't read command from the MQTT payload. Ensure that you are sending a \"command\" attribute"
                      " in the MQTT payload.")
                    .format(mqtt_command))
            else:
                self.add_log_line.emit(_("Payload command \"{0}\" is not defined. Please add it via the Payload "
                                         "Handler tab.")
                                       .format(mqtt_command))
                self.track_status.emit(104)

    def on_message(self, client, userdata, msg):
        topic = msg.topic.replace(f"{self.settings.topic}/", "")
        data = extract_payload(msg.payload)
        self.add_log_line.emit(_("Received MQTT payload \"{0}\" via topic \"{1}\".").format(data, topic))
        self.router(data)

    def __init__(self):
        super().__init__()
        self.settings = MQTTSettings.get_instance()

    def run(self):
        self.mqtt_connect()

    def mqtt_connect(self):
        if self.settings.hostname is None or self.settings.port is None or self.settings.topic is None:
            return False

        self.client = mqtt.Client()

        if self.settings.hostname != "" and self.settings.port != "" and self.settings.topic != "":
            self.client.on_connect = self.on_connect
            self.client.on_message = self.on_message
            self.client.on_log = self.mqtt_log
            self.client.username_pw_set(self.settings.username, self.settings.password)

            tmp_pem_file = ""

            try:
                if self.settings.enable_ssl:
                    # always check whether server certificate was given and exists
                    if self.settings.server_certificate_file == "" or \
                            not os.path.isfile(self.settings.server_certificate_file):

                        # if not try to fetch the certificate chain from the server
                        pem_file = Utils.get_certificate_chain(self.settings.hostname, self.settings.port)
                        if not pem_file:
                            self.add_log_line.emit(_("You didn't provide a server certificate file. It's "
                                                     "needed for SSL/TLS connections to the broker and couldn't be"
                                                     "fetched automatically from the broker. Canceling."))
                            self.track_status.emit(301)
                            return False
                        else:
                            tmp_pem_file = pem_file

                    if self.settings.enable_client_ssl_auth:
                        # in case of enabled client authentication, always ensure that client certificate file and
                        # client key file are given and exist, otherwise refuse to connect via SSL
                        if self.settings.client_certificate_file == "":
                            self.add_log_line.emit(_("You didn't provide a client certificate file. It's "
                                                     "needed for SSL/TLS client certificate authentication. "
                                                     "Canceling."))
                            self.track_status.emit(301)
                            return False

                        if not os.path.isfile(self.settings.client_certificate_file):
                            self.add_log_line.emit(_("The client certificate file you've provided ({0}) doesn't exist. "
                                                     "It's needed for SSL/TLS client certificate authentication. "
                                                     "Canceling.").format(self.settings.client_certificate_file))
                            self.track_status.emit(301)
                            return False

                        if self.settings.client_key_file == "":
                            self.add_log_line.emit(_("You didn't provide a client client key file. It's "
                                                     "needed for SSL/TLS client certificate authentication. "
                                                     "Canceling."))
                            self.track_status.emit(301)
                            return False

                        if not os.path.isfile(self.settings.client_key_file):
                            self.add_log_line.emit(_("The client key file you've provided ({0}) doesn't exist. "
                                                     "It's needed for SSL/TLS client certificate authentication. "
                                                     "Canceling.").format(self.settings.client_key_file))
                            self.track_status.emit(301)
                            return False

                        self.client.tls_set(ca_certs=(self.settings.server_certificate_file if tmp_pem_file == ""
                                                      else tmp_pem_file),
                                            certfile=self.settings.client_certificate_file,
                                            keyfile=self.settings.client_key_file,
                                            tls_version=ssl.PROTOCOL_TLSv1_2)
                    elif self.settings.allow_insecure_ssl and self.settings.server_certificate_file != "":
                        # disable checks for certificates and allow insecure TLS
                        self.client.tls_set(ca_certs=self.settings.server_certificate_file,
                                            tls_version=ssl.PROTOCOL_TLSv1_2)
                        self.client.tls_insecure_set(True)
                        # but warn the user, because this is not a recommended setup!
                        self.add_log_line.emit(_("WARNING: INSECURE SSL/TLS ENABLED! PROCEED WITH CAUTION!"))
                    elif self.settings.allow_insecure_ssl:
                        # disable checks for certificates and allow insecure TLS
                        self.client.tls_set(cert_reqs=ssl.CERT_NONE)
                        self.client.tls_insecure_set(True)
                        # but warn the user, because this is not a recommended setup!
                        self.add_log_line.emit(_("WARNING: INSECURE SSL/TLS ENABLED! PROCEED WITH CAUTION!"))
                    else:
                        # we just want in transit encryption
                        self.client.tls_set(ca_certs=(self.settings.server_certificate_file if tmp_pem_file == ""
                                                      else tmp_pem_file),
                                            tls_version=ssl.PROTOCOL_TLSv1_2)

                self.client.connect(self.settings.hostname, int(self.settings.port), 60)
                self.client.loop_start()

            except ConnectionRefusedError:
                self.add_log_line.emit(_("Broker refused the connection. Did you maybe provide wrong credentials?"))
                return False
            except ValueError:
                return False
            except TimeoutError:
                return False
            except SSLError as e:
                if hasattr(e, 'message'):
                    self.add_log_line.emit(f"openSSL: {str(e.message)}")
                else:
                    self.add_log_line.emit(f"openSSL: {str(e)}")

                return False
            except ConnectionResetError:
                if self.settings.enable_ssl and self.settings.port == "1883":
                    self.add_log_line.emit(_("Broker has reset the connection. You are trying to connect using SSL/TLS "
                                             "on plaintext MQTT port 1883. Retry with port 8883!"))

                else:
                    self.add_log_line.emit(_("Connection reset by broker."))

                return False
        else:
            self.add_log_line.emit(_("MQTT Connection not set up. Please provide a Hostname, Port and Topic."))
            self.track_status.emit(200)

    def mqtt_log(self, client, userdata, level, buff):
        self.add_log_line.emit(f"paho-mqtt: {buff}")

    def mqtt_disconnect(self):
        self.add_log_line.emit(_("Disconnecting from Broker."))
        self.track_status.emit(301)
        self.client.loop_stop()
        try:
            self.client.disconnect()
        except SSLZeroReturnError:
            return
        self.add_log_line.emit(_("Disconnected from Broker."))
        self.track_status.emit(105)

    def mqtt_reconnect(self):
        self.mqtt_disconnect()
        self.add_log_line.emit(_("Reconnecting to Broker."))
        self.track_status.emit(302)
        self.mqtt_connect()

src/easy_mqtt_handler/util/MQTTWorkerThread.py
andzeil/easy-mqtt-handler
"""
SPDX-License-Identifier: GPL-3.0-or-later
*
*  Tools.py
*
*  This class contains a collection of various utility functions
*
*  Copyright (C) 2023 A. Zeil
"""
import datetime
import os
import socket
import sys

from cryptography.hazmat.primitives import serialization
from OpenSSL import SSL
from pathlib import Path


class Utils:
    @staticmethod
    def get_config_path():
        # on windows, we want to store the configuration in %appdata%\easy-mqtt-handler, while
        # on *nix-based OSes we want to store the configuration in ~/.config/easy-mqtt-handler
        return os.path.expandvars("%appdata%\\easy-mqtt-handler\\") if os.name == "nt" else os.path.expanduser("~/.config/easy-mqtt-handler/")

    @staticmethod
    def create_path_if_not_exists(path):
        if not os.path.exists(path):
            os.makedirs(path, exist_ok=True)
            return True
        else:
            return False

    @staticmethod
    def get_timestamp():
        return datetime.datetime.now().strftime("%d.%m.%Y, %H:%M:%S")

    @staticmethod
    def load_license_file(license_file):
        if os.path.exists(license_file):
            try:
                sf = open(license_file, 'r')
                return sf.read()
            # TODO: implement better exception handling
            except IOError:
                return "License File couldn't be loaded, please check our Git Repository: " \
                       "https://github.com/andzeil/easy-mqtt-handler/"
        else:
            return "License File couldn't be loaded, please check our Git Repository: " \
                   "https://github.com/andzeil/easy-mqtt-handler/"

    @staticmethod
    def resource_path(relative_path):
        base_path = getattr(sys, '_MEIPASS', str(Path(os.path.dirname(os.path.abspath(__file__))).parent.absolute()))
        return os.path.join(base_path, relative_path)

    # this function tries to establish a connection and initiate an SSL handshake to fetch the certificate chain from a
    # server. it returns False, should it not be able to do so for whatever reason
    @classmethod
    def get_certificate_chain(cls, host, port):
        try:
            ssl_context = SSL.Context(method=SSL.SSLv23_METHOD)
            ssl_connection = SSL.Connection(ssl_context,
                                            socket=socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM))
            # ssl_connection.settimeout(1)
            ssl_connection.setblocking(1)
            ssl_connection.connect((host, int(port)))
            ssl_connection.do_handshake()
            cert_chain = ssl_connection.get_peer_cert_chain()
            ssl_connection.close()

            pem_file_bytes = bytearray()
            for cert in cert_chain:
                pem_file_bytes = pem_file_bytes + cert.to_cryptography().public_bytes(serialization.Encoding.PEM)

            tmp_pem_file = f"{cls.get_config_path()}tmp.pem"

            with open(tmp_pem_file, 'wb') as pf:
                pf.write(pem_file_bytes)
                pf.close()

                return tmp_pem_file
        except:
            return False

src/easy_mqtt_handler/util/Tools.py
andzeil/easy-mqtt-handler