Mercurial Repositories > MQTT Monitor / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2018 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing a dialog to edit the MQTT connection profiles.
"""

from __future__ import unicode_literals

import collections

from PyQt5.QtCore import pyqtSlot, Qt, QUuid
from PyQt5.QtWidgets import QDialog, QDialogButtonBox, QAbstractButton, \
    QListWidgetItem, QInputDialog, QLineEdit

from E5Gui import E5MessageBox
from E5Gui.E5PathPicker import E5PathPickerModes

from .Ui_MqttConnectionProfilesDialog import Ui_MqttConnectionProfilesDialog

import UI.PixmapCache
from Utilities.crypto import pwConvert


class MqttConnectionProfilesDialog(QDialog, Ui_MqttConnectionProfilesDialog):
    """
    Class implementing a dialog to edit the MQTT connection profiles.
    """
    def __init__(self, client, profiles, parent=None):
        """
        Constructor
        
        @param client reference to the MQTT client object
        @type MqttClient
        @param profiles dictionary containing dictionaries containing the
            connection parameters. Each entry must have the keys
            "BrokerAddress", "BrokerPort", "ClientId",
            "Keepalive", "CleanSession", "Username", "Password", "WillTopic",
            "WillMessage", "WillQos", "WillRetain", "TlsEnable", "TlsCaCert",
            "TlsClientCert", "TlsClientKey", "ConnectionTimeout".
        @type dict
        @param parent reference to the parent widget
        @type QWidget
        """
        super(MqttConnectionProfilesDialog, self).__init__(parent)
        self.setupUi(self)
        
        self.__client = client
        
        self.__profiles = collections.defaultdict(self.__defaultProfile)
        self.__profiles.update(profiles)
        self.__profilesChanged = False
        
        self.plusButton.setIcon(UI.PixmapCache.getIcon("plus.png"))
        self.copyButton.setIcon(UI.PixmapCache.getIcon("editCopy.png"))
        self.minusButton.setIcon(UI.PixmapCache.getIcon("minus.png"))
        
        self.tlsCertsFilePicker.setMode(E5PathPickerModes.OpenFileMode)
        self.tlsCertsFilePicker.setFilters(
            self.tr("Certificate Files (*.crt *.pem);;All Files (*)"))
        self.tlsSelfSignedCertsFilePicker.setMode(
            E5PathPickerModes.OpenFileMode)
        self.tlsSelfSignedCertsFilePicker.setFilters(
            self.tr("Certificate Files (*.crt *.pem);;All Files (*)"))
        self.tlsSelfSignedClientCertFilePicker.setMode(
            E5PathPickerModes.OpenFileMode)
        self.tlsSelfSignedClientCertFilePicker.setFilters(
            self.tr("Certificate Files (*.crt *.pem);;All Files (*)"))
        self.tlsSelfSignedClientKeyFilePicker.setMode(
            E5PathPickerModes.OpenFileMode)
        self.tlsSelfSignedClientKeyFilePicker.setFilters(
            self.tr("Key Files (*.key *.pem);;All Files (*)"))
        
        self.profileTabWidget.setCurrentIndex(0)
        
        if len(self.__profiles) == 0:
            self.minusButton.setEnabled(False)
            self.copyButton.setEnabled(False)
        
        self.profileFrame.setEnabled(False)
        self.__populatingProfile = False
        self.__deletingProfile = False
        
        self.__populateProfilesList()
    
    @pyqtSlot(str)
    def on_profileEdit_textChanged(self, name):
        """
        Private slot to handle changes of the profile name.
        
        @param name name of the profile
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(QAbstractButton)
    def on_profileButtonBox_clicked(self, button):
        """
        Private slot handling presses of the profile buttons.
        
        @param button reference to the pressed button
        @type QAbstractButton
        """
        if button == self.profileButtonBox.button(QDialogButtonBox.Apply):
            currentProfile = self.__applyProfile()
            self.__populateProfilesList(currentProfile)
        
        elif button == self.profileButtonBox.button(QDialogButtonBox.Reset):
            self.__resetProfile()
        
        elif button == self.profileButtonBox.button(
                QDialogButtonBox.RestoreDefaults):
            self.__populateProfileDefault()
    
    @pyqtSlot(QListWidgetItem, QListWidgetItem)
    def on_profilesList_currentItemChanged(self, current, previous):
        """
        Private slot to handle a change of the current profile.
        
        @param current new current item
        @type QListWidgetItem
        @param previous previous current item
        @type QListWidgetItem
        """
        self.minusButton.setEnabled(current is not None)
        self.copyButton.setEnabled(current is not None)
        
        if current is not previous:
            if not self.__deletingProfile and self.__isChangedProfile():
                # modified profile belongs to previous
                yes = E5MessageBox.yesNo(
                    self,
                    self.tr("Changed Connection Profile"),
                    self.tr("""The current profile has unsaved changes."""
                            """ Shall these be saved?"""),
                    icon=E5MessageBox.Warning,
                    yesDefault=True)
                if yes:
                    self.__applyProfile()
        
        if current:
            profileName = current.text()
            self.__populateProfile(profileName)
        else:
            self.__clearProfile()
    
    @pyqtSlot()
    def on_plusButton_clicked(self):
        """
        Private slot to add a new empty profile entry.
        """
        profileName, ok = QInputDialog.getText(
            self,
            self.tr("New Connection Profile"),
            self.tr("Enter name for the new Connection Profile:"),
            QLineEdit.Normal)
        if ok and bool(profileName):
            if profileName in self.__profiles:
                E5MessageBox.warning(
                    self,
                    self.tr("New Connection Profile"),
                    self.tr("""<p>A connection named <b>{0}</b> exists"""
                            """ already. Aborting...</p>""").format(
                        profileName))
            else:
                itm = QListWidgetItem(profileName, self.profilesList)
                self.profilesList.setCurrentItem(itm)
                self.brokerAddressEdit.setFocus(Qt.OtherFocusReason)
    
    @pyqtSlot()
    def on_copyButton_clicked(self):
        """
        Private slot to copy the selected profile entry.
        """
        itm = self.profilesList.currentItem()
        if itm:
            profileName = itm.text()
            newProfileName, ok = QInputDialog.getText(
                self,
                self.tr("Copy Connection Profile"),
                self.tr("Enter name for the copied Connection Profile:"),
                QLineEdit.Normal)
            if ok and bool(newProfileName):
                if newProfileName in self.__profiles:
                    E5MessageBox.warning(
                        self,
                        self.tr("Copy Connection Profile"),
                        self.tr("""<p>A connection named <b>{0}</b> exists"""
                                """ already. Aborting...</p>""").format(
                            newProfileName))
                else:
                    profile = self.__defaultProfile()
                    profile.update(self.__profiles[profileName])
                    self.__profiles[newProfileName] = profile
                    
                    itm = QListWidgetItem(newProfileName, self.profilesList)
                    self.profilesList.setCurrentItem(itm)
                    self.brokerAddressEdit.setFocus(Qt.OtherFocusReason)
    
    @pyqtSlot()
    def on_minusButton_clicked(self):
        """
        Private slot to delete the selected entry.
        """
        itm = self.profilesList.currentItem()
        if itm:
            profileName = itm.text()
            yes = E5MessageBox.yesNo(
                self,
                self.tr("Delete Connection Profile"),
                self.tr("""<p>Shall the Connection Profile <b>{0}</b>"""
                        """ really be deleted?</p>""").format(profileName)
            )
            if yes:
                self.__deletingProfile = True
                del self.__profiles[profileName]
                self.__profilesChanged = True
                self.__populateProfilesList()
                self.__deletingProfile = False
        
        self.profilesList.setFocus(Qt.OtherFocusReason)
    
    def getProfiles(self):
        """
        Public method to return a dictionary of profiles.
        
        @return dictionary containing dictionaries containing the defined
            connection profiles. Each entry have the keys "BrokerAddress",
            "BrokerPort", "ClientId", "Keepalive", "CleanSession", "Username",
            "Password", "WillTopic", "WillMessage", "WillQos", "WillRetain",
            "TlsEnable", "TlsCaCert", "TlsClientCert", "TlsClientKey",
            "ConnectionTimeout".
        @rtype dict
        """
        profilesDict = {}
        profilesDict.update(self.__profiles)
        return profilesDict
    
    def __applyProfile(self):
        """
        Private method to apply the entered data to the list of profiles.
        
        @return name of the applied profile
        @rtype str
        """
        profileName = self.profileEdit.text()
        profile = {
            "BrokerAddress": self.brokerAddressEdit.text(),
            "BrokerPort": self.brokerPortSpinBox.value(),
            "ClientId": self.clientIdEdit.text(),
            "ConnectionTimeout": self.connectionTimeoutSpinBox.value(),
            "Keepalive": self.keepaliveSpinBox.value(),
            "CleanSession": self.cleanSessionCheckBox.isChecked(),
            "Username": self.usernameEdit.text(),
            "Password": pwConvert(self.passwordEdit.text(), encode=True),
            "WillTopic": self.willTopicEdit.text(),
            "WillMessage": self.willMessageEdit.toPlainText(),
            "WillQos": self.willQosSpinBox.value(),
            "WillRetain": self.willRetainCheckBox.isChecked(),
            "TlsEnable": self.tlsGroupBox.isChecked(),
            "TlsCaCert": "",
            "TlsClientCert": "",
            "TlsClientKey": "",
        }
        if profile["TlsEnable"]:
            if self.tlsCertsFileButton.isChecked():
                profile["TlsCaCert"] = self.tlsCertsFilePicker.text()
            elif self.tlsSelfSignedCertsButton.isChecked():
                profile["TlsCaCert"] = \
                    self.tlsSelfSignedCertsFilePicker.text()
                profile["TlsClientCert"] = \
                    self.tlsSelfSignedClientCertFilePicker.text()
                profile["TlsClientKey"] = \
                    self.tlsSelfSignedClientKeyFilePicker.text()
        
        self.__profiles[profileName] = profile
        self.__profilesChanged = True
        
        return profileName
    
    def __defaultProfile(self):
        """
        Private method to populate non-existing profile items.
        
        @return default dictionary entry
        @rtype dict
        """
        defaultProfile = self.__client.defaultConnectionOptions()
        defaultProfile["BrokerAddress"] = ""
        if defaultProfile["TlsEnable"]:
            defaultProfile["BrokerPort"] = 8883
        else:
            defaultProfile["BrokerPort"] = 1883
        
        return defaultProfile
    
    def __populateProfilesList(self, currentProfile=""):
        """
        Private method to populate the list of defined profiles.
        
        @param currentProfile name of the current profile
        @type str
        """
        if not currentProfile:
            currentItem = self.profilesList.currentItem()
            if currentItem:
                currentProfile = currentItem.text()
        
        self.profilesList.clear()
        self.profilesList.addItems(sorted(self.__profiles.keys()))
        
        if currentProfile:
            items = self.profilesList.findItems(
                currentProfile, Qt.MatchExactly)
            if items:
                self.profilesList.setCurrentItem(items[0])
        
        if len(self.__profiles) == 0:
            self.profileFrame.setEnabled(False)
    
    def __populateProfile(self, profileName):
        """
        Private method to populate the profile data entry fields.
        
        @param profileName name of the profile to get data from
        @type str
        """
        profile = self.__defaultProfile()
        if profileName:
            profile.update(self.__profiles[profileName])
        
        self.__populatingProfile = True
        if profileName is not None:
            self.profileEdit.setText(profileName)
        self.brokerAddressEdit.setText(profile["BrokerAddress"])
        self.brokerPortSpinBox.setValue(profile["BrokerPort"])
        self.clientIdEdit.setText(profile["ClientId"])
        self.connectionTimeoutSpinBox.setValue(profile["ConnectionTimeout"])
        self.keepaliveSpinBox.setValue(profile["Keepalive"])
        self.cleanSessionCheckBox.setChecked(profile["CleanSession"])
        self.usernameEdit.setText(profile["Username"])
        self.passwordEdit.setText(pwConvert(profile["Password"], encode=False))
        self.willTopicEdit.setText(profile["WillTopic"])
        self.willMessageEdit.setPlainText(profile["WillMessage"])
        self.willQosSpinBox.setValue(profile["WillQos"])
        self.willRetainCheckBox.setChecked(profile["WillRetain"])
        self.tlsGroupBox.setChecked(profile["TlsEnable"])
        if profile["TlsCaCert"] and profile["TlsClientCert"]:
            self.tlsSelfSignedCertsButton.setChecked(True)
            self.tlsSelfSignedCertsFilePicker.setText(profile["TlsCaCert"])
            self.tlsSelfSignedClientCertFilePicker.setText(
                profile["TlsClientCert"])
            self.tlsSelfSignedClientKeyFilePicker.setText(
                profile["TlsClientKey"])
        elif profile["TlsCaCert"]:
            self.tlsCertsFileButton.setChecked(True)
            self.tlsCertsFilePicker.setText(profile["TlsCaCert"])
        else:
            self.tlsDefaultCertsButton.setChecked(True)
        self.__populatingProfile = False
        
        self.profileFrame.setEnabled(True)
        self.__updateApplyButton()
    
    def __clearProfile(self):
        """
        Private method to clear the profile data entry fields.
        """
        self.__populatingProfile = True
        self.profileEdit.setText("")
        self.brokerAddressEdit.setText("")
        self.brokerPortSpinBox.setValue(1883)
        self.clientIdEdit.setText("")
        self.keepaliveSpinBox.setValue(60)
        self.cleanSessionCheckBox.setChecked(True)
        self.usernameEdit.setText("")
        self.passwordEdit.setText("")
        self.willTopicEdit.setText("")
        self.willMessageEdit.setPlainText("")
        self.willQosSpinBox.setValue(0)
        self.willRetainCheckBox.setChecked(False)
        self.tlsGroupBox.setChecked(False)
        self.tlsDefaultCertsButton.setChecked(True)
        self.tlsCertsFileButton.setChecked(True)
        self.tlsCertsFilePicker.setText("")
        self.tlsSelfSignedCertsButton.setChecked(False)
        self.tlsSelfSignedCertsFilePicker.setText("")
        self.tlsSelfSignedClientCertFilePicker.setText("")
        self.tlsSelfSignedClientKeyFilePicker.setText("")
        self.__populatingProfile = False
        
        self.profileFrame.setEnabled(False)
        self.__updateApplyButton()
    
    def __resetProfile(self):
        """
        Private method to reset the profile data entry fields to their stored
        values.
        """
        profileName = self.profileEdit.text()
        if profileName in self.__profiles:
            self.__populateProfile(profileName)
    
    def __populateProfileDefault(self):
        """
        Private method to populate the profile data entry fields with default
        profile values.
        """
        self.__populateProfile(None)
    
    def __isChangedProfile(self):
        """
        Private method to check, if the currently shown profile contains some
        changed data.
        
        @return flag indicating changed data
        @rtype bool
        """
        profileName = self.profileEdit.text()
        if profileName == "":
            return False
        
        elif profileName in self.__profiles:
            profile = self.__defaultProfile()
            profile.update(self.__profiles[profileName])
            changed = (
                self.brokerAddressEdit.text() != profile["BrokerAddress"] or
                self.brokerPortSpinBox.value() != profile["BrokerPort"] or
                self.clientIdEdit.text() != profile["ClientId"] or
                self.connectionTimeoutSpinBox.value() !=
                profile["ConnectionTimeout"] or
                self.keepaliveSpinBox.value() != profile["Keepalive"] or
                self.cleanSessionCheckBox.isChecked() !=
                profile["CleanSession"] or
                self.usernameEdit.text() != profile["Username"] or
                self.passwordEdit.text() !=
                pwConvert(profile["Password"], encode=False) or
                self.willTopicEdit.text() != profile["WillTopic"] or
                self.willMessageEdit.toPlainText() != profile["WillMessage"] or
                self.willQosSpinBox.value() != profile["WillQos"] or
                self.willRetainCheckBox.isChecked() != profile["WillRetain"] or
                self.tlsGroupBox.isChecked() != profile["TlsEnable"]
            )
            # check TLS stuff only, if not yet changed
            if not changed:
                if self.tlsCertsFileButton.isChecked():
                    changed |= (
                        self.tlsCertsFilePicker.text() != profile["TlsCaCert"]
                    )
                elif self.tlsSelfSignedCertsButton.isChecked():
                    changed |= (
                        self.tlsSelfSignedCertsFilePicker.text() !=
                        profile["TlsCaCert"] or
                        self.tlsSelfSignedClientCertFilePicker.text() !=
                        profile["TlsClientCert"] or
                        self.tlsSelfSignedClientKeyFilePicker.text() !=
                        profile["TlsClientKey"]
                    )
            return changed
        
        else:
            return True
    
    def __updateApplyButton(self):
        """
        Private method to set the state of the Apply button.
        """
        # condition 1: profile name and broker address need to be given
        enable = (bool(self.profileEdit.text()) and
                  bool(self.brokerAddressEdit.text()))
        
        # condition 2: if client ID is empty, clean session must be selected
        if not self.__populatingProfile:
            if self.clientIdEdit.text() == "" and \
                    not self.cleanSessionCheckBox.isChecked():
                enable = False
                E5MessageBox.critical(
                    self,
                    self.tr("Invalid Connection Parameters"),
                    self.tr("An empty Client ID requires a clean session."))
        
        if self.tlsGroupBox.isChecked():
            if self.tlsCertsFileButton.isChecked():
                # condition 3a: if CA certificates file shall be used, it must
                # be given
                enable &= bool(self.tlsCertsFilePicker.text())
            elif self.tlsSelfSignedCertsButton.isChecked():
                # condition 3b: if client certificates shall be used, all files
                # must be given
                enable &= (
                    bool(self.tlsSelfSignedCertsFilePicker.text()) and
                    bool(self.tlsSelfSignedClientCertFilePicker.text()) and
                    bool(self.tlsSelfSignedClientKeyFilePicker.text())
                )
        
        self.profileButtonBox.button(QDialogButtonBox.Apply).setEnabled(enable)
    
    @pyqtSlot(str)
    def on_brokerAddressEdit_textChanged(self, address):
        """
        Private slot handling a change of the broker address.
        
        @param address broker address
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot()
    def on_generateIdButton_clicked(self):
        """
        Private slot to generate a client ID.
        """
        uuid = QUuid.createUuid()
        self.clientIdEdit.setText(uuid.toString(QUuid.WithoutBraces))
    
    @pyqtSlot(str)
    def on_clientIdEdit_textChanged(self, clientId):
        """
        Private slot handling a change of the client ID string.
        
        @param clientId client ID
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(bool)
    def on_cleanSessionCheckBox_clicked(self, checked):
        """
        Private slot to handle a change of the clean session selection.
        
        @param checked current state of the clean session selection
        @type bool
        """
        self.__updateApplyButton()
    
    @pyqtSlot(str)
    def on_tlsCertsFilePicker_textChanged(self, path):
        """
        Private slot handling a change of the TLS CA certificates file.
        
        @param path file path
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(str)
    def on_tlsSelfSignedCertsFilePicker_textChanged(self, path):
        """
        Private slot handling a change of the TLS CA certificates file.
        
        @param path file path
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(str)
    def on_tlsSelfSignedClientCertFilePicker_textChanged(self, path):
        """
        Private slot handling a change of the TLS client certificate file.
        
        @param path file path
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(str)
    def on_tlsSelfSignedClientKeyFilePicker_textChanged(self, path):
        """
        Private slot handling a change of the TLS client key file.
        
        @param path file path
        @type str
        """
        self.__updateApplyButton()
    
    @pyqtSlot(bool)
    def on_tlsGroupBox_toggled(self, checked):
        """
        Private slot handling the selection of TLS mode.
        
        @param checked state of the selection
        @type bool
        """
        if checked and self.brokerPortSpinBox.value() == 1883:
            # port is still standard non-TLS port
            yes = E5MessageBox.yesNo(
                self,
                self.tr("SSL/TLS Enabled"),
                self.tr(
                    """Encrypted connection using SSL/TLS has been enabled."""
                    """ However, the broker port is still the default"""
                    """ unencrypted port (port 1883). Shall this be"""
                    """ changed?"""),
                icon=E5MessageBox.Warning,
                yesDefault=True)
            if yes:
                self.brokerPortSpinBox.setValue(8883)
        elif not checked and self.brokerPortSpinBox.value() == 8883:
            # port is still standard TLS port
            yes = E5MessageBox.yesNo(
                self,
                self.tr("SSL/TLS Disabled"),
                self.tr(
                    """Encrypted connection using SSL/TLS has been disabled."""
                    """ However, the broker port is still the default"""
                    """ encrypted port (port 8883). Shall this be"""
                    """ changed?"""),
                icon=E5MessageBox.Warning,
                yesDefault=True)
            if yes:
                self.brokerPortSpinBox.setValue(1883)
        
        self.__updateApplyButton()
    
    @pyqtSlot(bool)
    def on_tlsDefaultCertsButton_toggled(self, checked):
        """
        Private slot handling the selection of using the default
        certificates file.
        
        @param checked state of the selection
        @type bool
        """
        self.__updateApplyButton()
    
    @pyqtSlot(bool)
    def on_tlsCertsFileButton_toggled(self, checked):
        """
        Private slot handling the selection of using a non-default
        certificates file.
        
        @param checked state of the selection
        @type bool
        """
        self.__updateApplyButton()
    
    @pyqtSlot(bool)
    def on_tlsSelfSignedCertsButton_toggled(self, checked):
        """
        Private slot handling the selection of using self signed
        client certificate and key files.
        
        @param checked state of the selection
        @type bool
        """
        self.__updateApplyButton()
    
    @pyqtSlot()
    def reject(self):
        """
        Public slot to reject the dialog changes.
        """
        if self.__isChangedProfile():
            button = E5MessageBox.warning(
                self,
                self.tr("Changed Connection Profile"),
                self.tr("""The current profile has unsaved changes. Shall"""
                        """ these be saved?"""),
                E5MessageBox.StandardButtons(
                    E5MessageBox.Discard |
                    E5MessageBox.Save),
                E5MessageBox.Save)
            if button == E5MessageBox.Save:
                self.__applyProfile()
                return
        
        if self.__profilesChanged:
            button = E5MessageBox.warning(
                self,
                self.tr("Changed Connection Profiles"),
                self.tr("""The list of connection profiles has unsaved"""
                        """ changes."""),
                E5MessageBox.StandardButtons(
                    E5MessageBox.Abort |
                    E5MessageBox.Discard |
                    E5MessageBox.Save),
                E5MessageBox.Save)
            if button == E5MessageBox.Save:
                super(MqttConnectionProfilesDialog, self).accept()
                return
            elif button == E5MessageBox.Abort:
                return
        
        super(MqttConnectionProfilesDialog, self).reject()
    
    @pyqtSlot()
    def accept(self):
        """
        Public slot to accept the dialog.
        """
        if self.__isChangedProfile():
            yes = E5MessageBox.yesNo(
                self,
                self.tr("Changed Connection Profile"),
                self.tr("""The current profile has unsaved changes. Shall"""
                        """ these be saved?"""),
                icon=E5MessageBox.Warning,
                yesDefault=True)
            if yes:
                self.__applyProfile()
        
        super(MqttConnectionProfilesDialog, self).accept()