Index: .qmake.conf =================================================================== diff -u -ra5781739bcbe58c754aff8861561495624bc5b67 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- .qmake.conf (.../.qmake.conf) (revision a5781739bcbe58c754aff8861561495624bc5b67) +++ .qmake.conf (.../.qmake.conf) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -8,5 +8,5 @@ BUILD_ROOT = $$shadowed($$PWD) # individual .pro files can also set these options for standalone builds -CONFIG += c++20 +CONFIG += c++17 QMAKE_CXXFLAGS += -Wall -Werror -Wextra Index: CMakeLists.txt =================================================================== diff -u -rb915ccc1a72fcca21908c35140af7ac7df1f1f37 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CMakeLists.txt (.../CMakeLists.txt) (revision b915ccc1a72fcca21908c35140af7ac7df1f1f37) +++ CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -8,7 +8,7 @@ include(cmake/Debug.cmake) include(cmake/PythonVenv.cmake) -set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_DISABLE_IN_SOURCE_BUILD ON) Index: CloudConnect/CMakeLists.txt =================================================================== diff -u -rb915ccc1a72fcca21908c35140af7ac7df1f1f37 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/CMakeLists.txt (.../CMakeLists.txt) (revision b915ccc1a72fcca21908c35140af7ac7df1f1f37) +++ CloudConnect/CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -3,7 +3,7 @@ LANGUAGES CXX ) -set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_INCLUDE_CURRENT_DIR ON) Index: CloudConnect/CloudConnect.pro =================================================================== diff -u -rb915ccc1a72fcca21908c35140af7ac7df1f1f37 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/CloudConnect.pro (.../CloudConnect.pro) (revision b915ccc1a72fcca21908c35140af7ac7df1f1f37) +++ CloudConnect/CloudConnect.pro (.../CloudConnect.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -1,14 +1,14 @@ TEMPLATE = app TARGET = CloudConnect -CONFIG += c++20 moc +CONFIG += c++17 moc QT += core network QT -= gui QMAKE_CXXFLAGS += -Wall -Werror -Wextra DESTDIR = $$PWD/../bin -include($$PWD/../lib/MsgUtils/codegen.pri) +include($$PWD/../lib/MsgUtils/codegen_tools.pri) include($$PWD/../lib/MsgUtils/MsgUtils.pri) LEAHI_MSG_CONF = $$PWD/../data/LeahiUnhandled.conf Index: CloudConnect/CloudConnectController.cpp =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -39,7 +39,7 @@ _settings(configPath, QSettings::IniFormat), _canInterface(this), _dispatcher(this), - _publisher(this) + _mqttClient(this) { loadMsgHandling(msgHandlingPath); @@ -109,10 +109,10 @@ /*! * \brief CloudConnectController::connectToCloud - * \details Builds the MQTT transport from the [Mqtt] settings and starts connecting. + * \details Configures the MQTT interface from the [Mqtt] settings and starts connecting. * \return true if the attempt was started, or if MQTT is disabled. - * \note Must run on the controller thread, after initThread(). MqttTcpTransport - * owns a QTcpSocket, which belongs to the thread that constructs it. + * \note Must run on the controller thread, after initThread(). The interface's + * QMqttClient socket belongs to the thread that constructs it. */ bool CloudConnectController::connectToCloud() { @@ -125,25 +125,23 @@ return true; } - MqttTransport::Config config; - config.endpoint = _settings.value("Mqtt/ServerAddress", QStringLiteral("127.0.0.1")).toString(); - config.port = static_cast(_settings.value("Mqtt/Port", 1883).toUInt()); - _topicPrefix = _settings.value("Mqtt/TopicPrefix", QStringLiteral("diality/v1/devices")).toString(); _deviceId = _settings.value("Mqtt/DeviceId", QStringLiteral("test_device")).toString(); - _mqttTransport = QSharedPointer::create(); - if (!_publisher.init(_mqttTransport, config)) { - qCritical().noquote() << "CloudConnect: could not initialise the MQTT publisher"; + MqttClient::Config config; + config.endpoint = _settings.value("Mqtt/ServerAddress", QStringLiteral("127.0.0.1")).toString(); + config.port = static_cast(_settings.value("Mqtt/Port", 1883).toUInt()); + config.clientId = _deviceId; + config.caPath = _settings.value("Mqtt/CaFile").toString(); + config.certPath = _settings.value("Mqtt/CertFile").toString(); + config.keyPath = _settings.value("Mqtt/KeyFile").toString(); + + if (!_mqttClient.init(config)) { + qCritical().noquote() << "CloudConnect: could not initialise the MQTT interface"; return false; } - connect(&_publisher, &MqttPublisher::didConnectionChange, - this, &CloudConnectController::onCloudConnectionChange); - connect(&_publisher, &MqttPublisher::didPublishAck, - this, &CloudConnectController::onPublishAck); - - return _publisher.open(); + return _mqttClient.open(); } /*! @@ -260,7 +258,7 @@ // TODO: temporary for capture protobuf // captureProtobuf(payload); - const bool published = _mqttEnabled && _publisher.publish(mqttTopic(it->topic), payload); + const bool published = _mqttEnabled && _mqttClient.publish(mqttTopic(it->topic), payload); if (_mqttEnabled && !published) { qWarning().noquote() << QString("CloudConnect: could not publish %1 (0x%2), message lost") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); @@ -298,34 +296,6 @@ } /*! - * \brief CloudConnectController::onCloudConnectionChange - * \details Logs MQTT session transitions. - * \param connected - true when the session came up - */ -void CloudConnectController::onCloudConnectionChange(bool connected) -{ - if (connected) { - qInfo().noquote() << "CloudConnect: MQTT session established"; - } - else { - qWarning().noquote() << "CloudConnect: MQTT session lost, messages are not reaching the cloud"; - } -} - -/*! - * \brief CloudConnectController::onPublishAck - * \details Reports the outcome of attempting to publish a message. - * \param messageId - correlation token passed to publish(); -1 when unused - * \param success - true when the broker acknowledged the message - */ -void CloudConnectController::onPublishAck(qint64 messageId, bool success) -{ - if (!success) { - qWarning().noquote() << "CloudConnect: publish was not acknowledged, id=" << messageId; - } -} - -/*! * \brief CloudConnectController::captureProtobuf * \details Appends one varint-delimited protobuf record to the .ser capture file and * the equivalent json to .json capture file. Index: CloudConnect/CloudConnectController.h =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -30,8 +30,7 @@ #include "CanMessage.h" #include "CloudConnectServer.h" #include "MessageDispatcher.h" -#include "MqttPublisher.h" -#include "MqttTcpTransport.h" +#include "MqttClient.h" using namespace Can; @@ -80,8 +79,7 @@ QMap> _msgCache; QHash _msgHandling; QSharedPointer _appServer; - QSharedPointer _mqttTransport; - MqttPublisher _publisher; + MqttClient _mqttClient; QString _topicPrefix; // TODO: define in INI? QString _deviceId; // TODO: this needs to be sent from Leahi app or retrieved from somewhere bool _mqttEnabled = false; @@ -93,6 +91,4 @@ private Q_SLOTS: void onFrameReceive(const QCanBusFrame frame); void onMessageReceive(const Can::Message &msg); - void onCloudConnectionChange(bool connected); - void onPublishAck(qint64 messageId, bool success); }; Index: CloudConnect/config/CloudConnect.ini =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/config/CloudConnect.ini (.../CloudConnect.ini) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ CloudConnect/config/CloudConnect.ini (.../CloudConnect.ini) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -3,10 +3,13 @@ [Mqtt] ServerAddress=127.0.0.1 -Port=1883 +Port=8883 TopicPrefix=diality/v1/devices ; TODO: DeviceId should be sent by Leahi app or retrieved from somewhere else DeviceId=test_device +CaFile= +CertFile= +KeyFile= ; TODO: temporary for protobuf capture [Capture] Index: CloudConnect/config/LeahiMsgHandling.ini =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- CloudConnect/config/LeahiMsgHandling.ini (.../LeahiMsgHandling.ini) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ CloudConnect/config/LeahiMsgHandling.ini (.../LeahiMsgHandling.ini) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -18,7 +18,6 @@ [0x0100] msg_id = MSG_ID_ALARM_STATUS_DATA -topic = HighPriority [0x0180] msg_id = MSG_ID_TD_SOFTWARE_RESET_REQUEST @@ -31,7 +30,6 @@ [0x0200] msg_id = MSG_ID_ALARM_TRIGGERED -topic = HighPriority [0x0280] msg_id = MSG_ID_TD_SEND_TEST_CONFIGURATION @@ -44,7 +42,6 @@ [0x0300] msg_id = MSG_ID_ALARM_CLEARED -topic = HighPriority [0x0380] msg_id = MSG_ID_TD_BUBBLE_OVERRIDE_REQUEST @@ -225,8 +222,6 @@ [0x1200] msg_id = MSG_ID_TD_OP_MODE_DATA -action = SendAlways -topic = HighPriority [0x1280] msg_id = MSG_ID_TD_BACKUP_ALARM_AUDIO_CURRENT_OVERRIDE_REQUEST @@ -383,7 +378,6 @@ [0x1F00] msg_id = MSG_ID_DD_CONDUCTIVITY_DATA -action = SendDelta [0x1F80] msg_id = MSG_ID_TD_ALARM_CLEAR_ALL_ALARMS_REQUEST @@ -468,7 +462,6 @@ [0x2600] msg_id = MSG_ID_DD_TEMPERATURE_DATA -action = SendDelta [0x2680] msg_id = MSG_ID_TD_BLOOD_PUMP_SET_FLOW_RATE_REQUEST @@ -481,7 +474,6 @@ [0x2700] msg_id = MSG_ID_DIALYSATE_PUMPS_DATA -action = SendDelta [0x2780] msg_id = MSG_ID_TD_BLOOD_PUMP_SET_SPEED_REQUEST @@ -563,7 +555,6 @@ [0x2E00] msg_id = MSG_ID_DD_BAL_CHAMBER_DATA -action = SendDelta [0x2E80] msg_id = MSG_ID_TD_RSP_CURRENT_TREATMENT_PARAMETERS @@ -771,7 +762,6 @@ [0x3FA0] msg_id = MSG_ID_DD_SEND_BLOOD_LEAK_EMB_MODE_RESPONSE -action = SendDelta [0x3FB0] msg_id = MSG_ID_FP_FLUSH_CONCENTRATE_TIMER_OVERRIDE_REQUEST @@ -970,7 +960,6 @@ [0x5300] msg_id = MSG_ID_TD_TEMPERATURE_DATA -action = SendDelta [0x5380] msg_id = MSG_ID_TD_SYRINGE_PUMP_POSITION_OVERRIDE_REQUEST @@ -1436,4 +1425,4 @@ [0xFFFF] msg_id = MSG_ID_ACK_MESSAGE_THAT_REQUIRES_ACK -action = Drop + Index: lib/Comms/CMakeLists.txt =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -6,7 +6,7 @@ set(CMAKE_AUTOMOC ON) find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core) -find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network SerialBus) +find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Mqtt Network SerialBus) set(INCLUDES include/CanInterface.h @@ -20,10 +20,7 @@ include/main.h include/MessageBuilder.h include/MessageDispatcher.h - include/MqttMessage.h - include/MqttPublisher.h - include/MqttTcpTransport.h - include/MqttTransport.h + include/MqttClient.h include/types.h ) @@ -37,9 +34,7 @@ src/FrameInterface.cpp src/MessageBuilder.cpp src/MessageDispatcher.cpp - src/MqttMessage.cpp - src/MqttPublisher.cpp - src/MqttTcpTransport.cpp + src/MqttClient.cpp src/types.cpp ) @@ -59,6 +54,7 @@ target_link_libraries(${PROJECT_NAME} PUBLIC Qt${QT_VERSION_MAJOR}::Core + Qt${QT_VERSION_MAJOR}::Mqtt Qt${QT_VERSION_MAJOR}::Network Qt${QT_VERSION_MAJOR}::SerialBus ) Index: lib/Comms/Comms.pro =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- lib/Comms/Comms.pro (.../Comms.pro) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ lib/Comms/Comms.pro (.../Comms.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -2,8 +2,8 @@ TEMPLATE = lib TARGET = Comms -CONFIG += shared c++20 moc -QT += core network serialbus +CONFIG += shared c++17 moc +QT += core mqtt network serialbus QMAKE_CXXFLAGS += -Wall -Werror -Wextra @@ -21,10 +21,7 @@ include/main.h \ include/MessageBuilder.h \ include/MessageDispatcher.h \ - include/MqttPacket.h \ - include/MqttPublisher.h \ - include/MqttTcpTransport.h \ - include/MqttTransport.h \ + include/MqttClient.h \ include/types.h SOURCES = \ @@ -37,9 +34,7 @@ src/FrameInterface.cpp \ src/MessageBuilder.cpp \ src/MessageDispatcher.cpp \ - src/MqttPacket.cpp \ - src/MqttPublisher.cpp \ - src/MqttTcpTransport.cpp \ + src/MqttClient.cpp \ src/types.cpp INCLUDEPATH += $$PWD/include Index: lib/Comms/include/MqttClient.h =================================================================== diff -u --- lib/Comms/include/MqttClient.h (revision 0) +++ lib/Comms/include/MqttClient.h (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -0,0 +1,80 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttClient.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include +#include +#include +#include + +class QMqttClient; + +/*! + * \brief Publishes messages to the cloud over MQTT. + * \details Wraps QtMqtt's QMqttClient for a QoS 1 publishing session. + * + * The lifecycle is init() -> open() -> close(). init() only validates + * the configuration and builds the TLS configuration from it; the + * QMqttClient itself is created by open() and destroyed by close(), + * or by an unsolicited disconnect. A close()/open() pair is therefore + * a full reset, which is what allows the session to be reopened with + * rotated credentials. + * \note Connecting is asynchronous: open() only starts the attempt, and a + * true return does not mean the broker accepted the session. + * \note Must live on the thread that calls open(): the client's socket + * belongs to the thread that constructs it. + * \note The connection is always TLS. init() fails unless keyPath names a + * readable private key. + */ +class MqttClient : public QObject +{ + Q_OBJECT + +public: + /*! + * \brief Parameters for one MQTT session. + */ + struct Config + { + QString endpoint; // Broker FQDN or address; no scheme or trailing slash. + quint16 port = 8883; // Broker port; the session is always TLS. + QString certPath; // Client certificate for mTLS, PEM; optional. + QString keyPath; // Unencrypted private key matching certPath, PEM; required. + QString caPath; // CA bundle to verify the broker, PEM; optional, falls back + // to the system CA set. + QString clientId; // Must equal the IoT Thing name; the device policy scopes iot:Connect to it. + quint16 keepAliveSecs = 30; // MQTT keep-alive interval; PINGREQ is QMqttClient's job. + bool cleanSession = true; // No broker-side queue; durability is the device's job. + }; + + explicit MqttClient(QObject *parent = nullptr); + ~MqttClient() override; + + bool init(const Config &config); + +public Q_SLOTS: + bool open(); + void close(); + bool publish(const QString &topic, const QByteArray &payload, qint64 messageId = -1); + +private: + void onStateChanged(); + void onMessageSent(qint32 id); + void failPending(); + + QMqttClient *_client = nullptr; + Config _config; + QSslConfiguration _sslConfig; + QHash _pending; +}; Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/include/MqttPacket.h'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/include/MqttPublisher.h'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/include/MqttTcpTransport.h'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/include/MqttTransport.h'. Fisheye: No comparison available. Pass `N' to diff? Index: lib/Comms/src/MqttClient.cpp =================================================================== diff -u --- lib/Comms/src/MqttClient.cpp (revision 0) +++ lib/Comms/src/MqttClient.cpp (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -0,0 +1,277 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttClient.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include +#include +#include +#include +#include +#include +#include + +#include "MqttClient.h" + +Q_LOGGING_CATEGORY(logMqtt, "cloudconnect.mqtt") + +/*! + * \brief MqttClient::MqttClient + * \details Constructor + * \param parent - optional QObject parent + */ +MqttClient::MqttClient(QObject *parent) : + QObject(parent) +{ +} + +/*! + * \brief MqttClient::~MqttClient + * \details Ends any open session. + */ +MqttClient::~MqttClient() +{ + close(); +} + +/*! + * \brief MqttClient::init + * \details Validates the configuration and builds the TLS configuration from + * the certificate paths it names. No client is created and no + * connection is attempted here; that is open()'s job. + * \param config - parameters for the session, retained for the next open() + * \return true if the configuration is usable, otherwise false + * \note keyPath is required. caPath and certPath are optional: without + * caPath the system CA set verifies the broker, and without certPath + * the session is server-authenticated only rather than mTLS. + */ +bool MqttClient::init(const Config &config) +{ + if (_client != nullptr) { + qCCritical(logMqtt).noquote() << "client already initialized"; + return false; + } + + QSslConfiguration ssl = QSslConfiguration::defaultConfiguration(); + if (!config.caPath.isEmpty()) { + const QList cas = QSslCertificate::fromPath(config.caPath); + if (cas.isEmpty()) { + qCCritical(logMqtt).noquote() << QString("%1 is not a CA certificate").arg(config.caPath); + return false; + } + ssl.setCaCertificates(cas); + } + + if (!config.certPath.isEmpty()) { + const QList certs = QSslCertificate::fromPath(config.certPath); + if (certs.isEmpty()) { + qCCritical(logMqtt).noquote() << QString("%1 is not a certificate").arg(config.certPath); + return false; + } + ssl.setLocalCertificate(certs.first()); + } + + QFile keyFile(config.keyPath); + if (!keyFile.open(QIODevice::ReadOnly)) { + qCCritical(logMqtt).noquote() << QString("could not read private key %1").arg(config.keyPath); + return false; + } + const QByteArray pem = keyFile.readAll(); + QSslKey key(pem, QSsl::Rsa, QSsl::Pem); + if (key.isNull()) { + key = QSslKey(pem, QSsl::Ec, QSsl::Pem); + } + if (key.isNull()) { + qCCritical(logMqtt).noquote() << QString("private key %1 is not a PEM, RSA, or EC key").arg(config.keyPath); + return false; + } + ssl.setPrivateKey(key); + + _sslConfig = ssl; + _config = config; + + return true; +} + +/*! + * \brief MqttClient::open + * \details Creates the QMqttClient from the configuration validated by init(), + * wires its signals and starts a TLS connection attempt. + * \return true if the attempt was started + * \note Must be called on this object's thread; the client's socket belongs + * to the thread that constructs it. + * \note The attempt is asynchronous, so a true return only means the client + * accepted the request. The outcome arrives via onStateChanged(). + */ +bool MqttClient::open() +{ + if (_client != nullptr) { + qCCritical(logMqtt).noquote() << "client connection already opened"; + return false; + } + + _client = new QMqttClient(this); + _client->setProtocolVersion(QMqttClient::MQTT_3_1_1); + _client->setHostname(_config.endpoint); + _client->setPort(_config.port); + _client->setClientId(_config.clientId); + _client->setKeepAlive(_config.keepAliveSecs); + _client->setCleanSession(_config.cleanSession); + + connect(_client, &QMqttClient::stateChanged, this, &MqttClient::onStateChanged); + connect(_client, &QMqttClient::messageSent, this, &MqttClient::onMessageSent); + connect(_client, &QMqttClient::errorChanged, this, + [](QMqttClient::ClientError error) { + if (error != QMqttClient::NoError) { + qCWarning(logMqtt).noquote() << "client error" << error; + } + } + ); + + qCInfo(logMqtt).noquote() << QString("attempting to connect to %1:%2 as %3") + .arg(_config.endpoint).arg(_config.port).arg(_config.clientId); + _client->connectToHostEncrypted(_sslConfig); + return true; +} + +/*! + * \brief MqttClient::close + * \details Ends the session and discards the client. Sends a DISCONNECT only + * when the socket drains, otherwise aborts it. + */ +void MqttClient::close() +{ + if (_client == nullptr) { + return; + } + + // clear the handle and disconnect signals first so when client state changes + // to Disconnected it does not trigger a reconnect in onStateChanged. + QMqttClient *client = _client; + _client = nullptr; + + failPending(); + client->disconnect(this); + + // flush to force the socket to send the Disconnect so bytesToWrite hopefully + // does not have any pending data to send (bytesToWrite() == 0) + auto *socket = qobject_cast(client->transport()); + if (socket != nullptr) { + socket->flush(); + } + + // if bytesToWrite is zero and state is not Disconnected, then send Disconnect + if (socket == nullptr || socket->bytesToWrite() == 0) { + if (client->state() != QMqttClient::Disconnected) { + client->disconnectFromHost(); + } + } + else { + // if there are pending bytes to write, then disconnectFromHost() blocks 30s + // waiting for the bytes to be sent, just abort, which forcees the write + // to fail immediately and skip the wait + socket->abort(); + } + + // clean up the client + client->deleteLater(); +} + +/*! + * \brief MqttClient::publish + * \details Hands one message to the client for QoS 1 delivery. + * \param topic - fully resolved MQTT topic + * \param payload - serialised message bytes + * \param messageId - caller token echoed back on acknowledgement, or -1 for none + * \return true if the client accepted the message. + * \note Return status does NOT indicate delivery, only that the client + * accepted the message for sending. The PUBACK that confirms delivery + * arrives later in onMessageSent(). + */ +bool MqttClient::publish(const QString &topic, const QByteArray &payload, qint64 messageId) +{ + if (_client == nullptr || _client->state() != QMqttClient::Connected) { + return false; + } + + const qint32 id = _client->publish(QMqttTopicName(topic), payload, 1, false); + if (id == -1) { + qCWarning(logMqtt).noquote() << QString("client rejected publishing to topic %1").arg(topic); + return false; + } + + _pending.insert(id, messageId); + // TODO: determine if messages waiting on a PUBACK need to be re-sent + // set a timeout interval to indicate when a message needs to be + // resent and have a timer periodically check _pending for messages + // that have met that threshold + + return true; +} + +/*! + * \brief MqttClient::onStateChanged + * \details Handle QMqttClient state changes and perform additional actions as + * a result of state transition, if applicable. + */ +void MqttClient::onStateChanged() +{ + if (_client != nullptr) { + switch (_client->state()) { + case QMqttClient::Disconnected: + // NOTE: Receiving state change to Disconnected here means that connection + // was not done by user of this class calling close() and is most + // likely caused by externally factors, like the server side closing + // the connection. + qCInfo(logMqtt).noquote() << "client disconnected"; + failPending(); + _client->deleteLater(); + _client = nullptr; + // TODO: reconnect on disconnect + // set a flag reconnect flag indicating if connection should try to + // be re-established + break; + case QMqttClient::Connecting: + qCInfo(logMqtt).noquote() << "client connecting"; + break; + case QMqttClient::Connected: + qCInfo(logMqtt).noquote() << "client connected"; + break; + } + } +} + +/*! + * \brief MqttClient::onMessageSent + * \details Clears one QoS 1 publish from _pending on PUBACK. + * \param id - QMqttClient's identifier for the acknowledged publish + */ +void MqttClient::onMessageSent(qint32 id) +{ + if (!_pending.contains(id)) { + qCWarning(logMqtt).noquote() << QString("received PUBACK for an unknown published message with id %1").arg(id); + return; + } + _pending.take(id); +} + +/*! + * \brief MqttClient::failPending + * \details Abandons every publish still awaiting a PUBACK. + */ +void MqttClient::failPending() +{ + if (_pending.isEmpty()) { + return; + } + + qCWarning(logMqtt).noquote() << QString("Failing %1 unacknowledged publishes").arg(_pending.size()); + _pending.clear(); +} Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/src/MqttPacket.cpp'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/src/MqttPublisher.cpp'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/Comms/src/MqttTcpTransport.cpp'. Fisheye: No comparison available. Pass `N' to diff? Index: lib/MsgUtils/MsgUtils.pro =================================================================== diff -u -r4dccc470aedf6f68a0ad73c1101f8a96188cbdb1 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- lib/MsgUtils/MsgUtils.pro (.../MsgUtils.pro) (revision 4dccc470aedf6f68a0ad73c1101f8a96188cbdb1) +++ lib/MsgUtils/MsgUtils.pro (.../MsgUtils.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -2,14 +2,14 @@ TEMPLATE = lib TARGET = MsgUtils -CONFIG += shared c++20 moc +CONFIG += shared c++17 moc QT += core network serialbus QMAKE_CXXFLAGS += -Wall -Werror -Wextra DESTDIR = $$PWD/lib -include(codegen.pri) +include(codegen_tools.pri) LEAHI_MSG_CONF = $$PWD/../../data/LeahiUnhandled.conf Fisheye: Tag 51e99f2578e0901d9da91a4cb60d1b8858cfe971 refers to a dead (removed) revision in file `lib/MsgUtils/codegen.pri'. Fisheye: No comparison available. Pass `N' to diff? Index: lib/MsgUtils/codegen_tools.pri =================================================================== diff -u --- lib/MsgUtils/codegen_tools.pri (revision 0) +++ lib/MsgUtils/codegen_tools.pri (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -0,0 +1,59 @@ +# Project Python venv and generator script paths. +# Included by the components that generate code (e.g. MsgUtils and CloudConnect). +# +# qmake evaluates system() at qmake time, while QMAKE_EXTRA_COMPILERS / +# QMAKE_EXTRA_TARGETS run at make time. The venv is created here once and shared. +# The generators are wired as make-time rules in the including .pro so they +# re-run on input change. + +PROJECT_ROOT = $$clean_path($$PWD/../..) +MSGUTILS_SCRIPTS_DIR = $$PROJECT_ROOT/scripts/MsgUtils + +# BUILD_ROOT is set by the repo-root .qmake.conf — identical for every component +# in the tree, so all components share ONE venv. Fall back to OUT_PWD when built +# without the .qmake.conf in scope (e.g. a leaf .pro copied elsewhere). +isEmpty(BUILD_ROOT): BUILD_ROOT = $$OUT_PWD + +PROJECT_VENV_DIR = $$BUILD_ROOT/.venv +PROJECT_PYTHON = $$PROJECT_VENV_DIR/bin/python + +# Create the venv, then editable-install the msgutils package so the generator +# scripts can `import msgutils`. +# +# A missing python3-venv reports nothing here and later becomes a generator rule +# failing to find an interpreter. This checks and reports here if it is missing. +!exists($$PROJECT_PYTHON) { + message("Creating project Python venv at $$PROJECT_VENV_DIR") + + !system(python3 -m venv $$shell_quote($$PROJECT_VENV_DIR)) { + # $${} braces: qmake allows '.' in a variable name, so "$$VAR." reads as a + # variable named "VAR." and expands to nothing. + error("Could not create the Python venv at $${PROJECT_VENV_DIR}. On Debian/Ubuntu this usually means the venv module is missing: sudo apt install python3-venv") + } + + # venv can exit 0 and still leave no interpreter (interrupted, disk full). + !exists($$PROJECT_PYTHON) { + error("The venv at $${PROJECT_VENV_DIR} was created but has no interpreter at $${PROJECT_PYTHON}. Remove the directory and re-run qmake.") + } +} + +# Allow a machine with a working venv and no network can still build. +!system($$shell_quote($$PROJECT_PYTHON) -m pip install --upgrade pip --quiet) { + warning("Could not upgrade pip in $${PROJECT_VENV_DIR}; continuing with the installed version.") +} + +# The generators import msgutils, but a failed reinstall is survivable when the venv +# already has it. +!system($$shell_quote($$PROJECT_PYTHON) -m pip install -e $$shell_quote($$MSGUTILS_SCRIPTS_DIR) --quiet) { + !system($$shell_quote($$PROJECT_PYTHON) -c $$shell_quote(import msgutils)) { + error("Could not install msgutils from $${MSGUTILS_SCRIPTS_DIR} into $${PROJECT_VENV_DIR}, and it is not importable. The code generators cannot run.") + } + warning("Could not reinstall msgutils from $${MSGUTILS_SCRIPTS_DIR}; continuing with the copy already in the venv.") +} + +PROTOC = protoc +# Missing protoc reports nothing here and later becomes a generator rule failing with "command not found". +# This checks and reports here if it is missing. +!system($$PROTOC --version > /dev/null 2>&1) { + error("Could not run '$${PROTOC}'. On Debian/Ubuntu install it with: sudo apt install protobuf-compiler") +} Index: tools/CANDumpPlayer/CANDumpPlayer.pro =================================================================== diff -u -ra5781739bcbe58c754aff8861561495624bc5b67 -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/CANDumpPlayer/CANDumpPlayer.pro (.../CANDumpPlayer.pro) (revision a5781739bcbe58c754aff8861561495624bc5b67) +++ tools/CANDumpPlayer/CANDumpPlayer.pro (.../CANDumpPlayer.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -1,6 +1,6 @@ TEMPLATE = app TARGET = CANDumpPlayer -CONFIG += c++20 +CONFIG += c++17 QT += core serialbus QT -= gui Index: tools/CANDumpPlayer/CMakeLists.txt =================================================================== diff -u -rd7719d4a8f8209f11f34905d29272d21f448195b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/CANDumpPlayer/CMakeLists.txt (.../CMakeLists.txt) (revision d7719d4a8f8209f11f34905d29272d21f448195b) +++ tools/CANDumpPlayer/CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -2,7 +2,7 @@ project(CANDumpPlayer LANGUAGES CXX) -set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_INCLUDE_CURRENT_DIR ON) Index: tools/DCSsim/CMakeLists.txt =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/DCSsim/CMakeLists.txt (.../CMakeLists.txt) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ tools/DCSsim/CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -2,7 +2,7 @@ project(DCSsim LANGUAGES CXX) -set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_INCLUDE_CURRENT_DIR ON) set(CMAKE_AUTOMOC ON) @@ -18,11 +18,13 @@ set(INCLUDES DCSSimController.h + MqttPacket.h ) set(SRCS DCSSimController.cpp main.cpp + MqttPacket.cpp ) add_executable(${PROJECT_NAME} Index: tools/DCSsim/DCSSimController.cpp =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/DCSsim/DCSSimController.cpp (.../DCSSimController.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ tools/DCSsim/DCSSimController.cpp (.../DCSSimController.cpp) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -14,41 +14,148 @@ #include #include #include +#include +#include +#include +#include #include #include "DCSSimController.h" #include "LeahiMsgDefs.h" #include "LeahiMsgProtoUtils.h" /*! + * \brief SslTcpServer::setTls + * \details Installs the TLS material; connections accepted from here on are encrypted. + * \param tls Certificate, key and optional client CA bundle. + */ +void SslTcpServer::setTls(const Tls &tls) +{ + _tls = tls; +} + +/*! + * \brief SslTcpServer::incomingConnection + * \details Adopts the accepted descriptor and, with TLS material set, starts the + * server-side handshake before surfacing the connection. + * \param socketDescriptor Native descriptor of the accepted connection. + */ +void SslTcpServer::incomingConnection(qintptr socketDescriptor) +{ + if (_tls.cert.isNull()) { + QTcpServer::incomingConnection(socketDescriptor); + return; + } + + auto *socket = new QSslSocket(this); + if (!socket->setSocketDescriptor(socketDescriptor)) { + qWarning().noquote() << "SslTcpServer: cannot adopt the accepted socket"; + delete socket; + return; + } + + QSslConfiguration ssl = QSslConfiguration::defaultConfiguration(); + ssl.setLocalCertificate(_tls.cert); + ssl.setPrivateKey(_tls.key); + if (!_tls.caCerts.isEmpty()) { + ssl.setCaCertificates(_tls.caCerts); + // mTLS: refuse clients that do not present a certificate the CA signs. + ssl.setPeerVerifyMode(QSslSocket::VerifyPeer); + } + socket->setSslConfiguration(ssl); + + connect(socket, QOverload &>::of(&QSslSocket::sslErrors), + this, [](const QList &errors) { + for (const QSslError &error : errors) { + qWarning().noquote() << "SslTcpServer: TLS error:" << error.errorString(); + } + }); + + socket->startServerEncryption(); + addPendingConnection(socket); +} + +/*! * \brief DCSSimController::DCSSimController * \details Constructor * \param port TCP port to listen on. * \param parent QObject parent. */ -DCSSimController::DCSSimController(quint16 port, QObject *parent) : +DCSSimController::DCSSimController(quint16 port, const TlsConfig &tls, QObject *parent) : QObject(parent), - _port(port) + _port(port), + _tlsConfig(tls) { connect(&_server, &QTcpServer::newConnection, this, &DCSSimController::onNewConnection); } /*! * \brief DCSSimController::listen * \details Starts the controller listening on all interfaces. - * \return true on success, false if the port cannot be bound. + * \return true on success, false if the port cannot be bound or TLS material + * cannot be loaded. */ bool DCSSimController::listen() { + if (!_tlsConfig.certPath.isEmpty() && !initTls()) { + return false; + } + if (!_server.listen(QHostAddress::Any, _port)) { qCritical().noquote() << "DCSSimController: cannot listen on port" << _port << "—" << _server.errorString(); return false; } - qInfo().noquote() << "DCSSimController: listening on port" << _port; + qInfo().noquote() << "DCSSimController: listening on port" << _port + << (_tlsConfig.certPath.isEmpty() ? "(plain)" + : _tlsConfig.caPath.isEmpty() ? "(TLS)" : "(mTLS)"); return true; } /*! + * \brief DCSSimController::initTls + * \details Loads the PEM material from TlsConfig into the server. + * \return true if everything referenced by TlsConfig loaded. + * \note The key is tried as RSA first, then EC. + */ +bool DCSSimController::initTls() +{ + SslTcpServer::Tls tls; + + const QList certs = QSslCertificate::fromPath(_tlsConfig.certPath); + if (certs.isEmpty()) { + qCritical().noquote() << "DCSSimController: no server certificate in" << _tlsConfig.certPath; + return false; + } + tls.cert = certs.first(); + + QFile keyFile(_tlsConfig.keyPath); + if (!keyFile.open(QIODevice::ReadOnly)) { + qCritical().noquote() << "DCSSimController: cannot read private key" << _tlsConfig.keyPath; + return false; + } + const QByteArray pem = keyFile.readAll(); + tls.key = QSslKey(pem, QSsl::Rsa, QSsl::Pem); + if (tls.key.isNull()) { + tls.key = QSslKey(pem, QSsl::Ec, QSsl::Pem); + } + if (tls.key.isNull()) { + qCritical().noquote() << "DCSSimController: private key" << _tlsConfig.keyPath << "is not a PEM RSA or EC key"; + return false; + } + + if (!_tlsConfig.caPath.isEmpty()) { + tls.caCerts = QSslCertificate::fromPath(_tlsConfig.caPath); + if (tls.caCerts.isEmpty()) { + qCritical().noquote() << "DCSSimController: no CA certificates in" << _tlsConfig.caPath; + return false; + } + } + + _server.setTls(tls); + return true; +} + +/*! * \brief DCSSimController::onNewConnection * \details Accepts the pending connection, if another client is not already connect. */ Index: tools/DCSsim/DCSSimController.h =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/DCSsim/DCSSimController.h (.../DCSSimController.h) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ tools/DCSsim/DCSSimController.h (.../DCSSimController.h) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -15,21 +15,65 @@ #include #include #include +#include +#include #include #include #include "MqttPacket.h" /*! + * \brief QTcpServer that upgrades accepted connections to server-side TLS. + * \details Plain QTcpServer while no TLS material is set. With material set, + * each accepted descriptor becomes a QSslSocket and the handshake is + * started before the connection is surfaced. + */ +class SslTcpServer : public QTcpServer +{ + Q_OBJECT + +public: + /*! + * \brief Server-side TLS material; TLS is disabled while cert is null. + */ + struct Tls + { + QSslCertificate cert; ///< Server certificate. + QSslKey key; ///< Server private key. + QList caCerts; ///< When set, clients must present a certificate these sign (mTLS). + }; + + using QTcpServer::QTcpServer; + + void setTls(const Tls &tls); + +protected: + void incomingConnection(qintptr socketDescriptor) override; + +private: + Tls _tls; +}; + +/*! * \brief Diality Cloud System (DCS) simulator */ class DCSSimController : public QObject { Q_OBJECT public: - explicit DCSSimController(quint16 port, QObject *parent = nullptr); + /*! + * \brief TLS configuration, all PEM; TLS is off while certPath is empty. + */ + struct TlsConfig + { + QString certPath; ///< Server certificate. + QString keyPath; ///< Server private key. + QString caPath; ///< When set, clients must present a certificate this bundle signs (mTLS). + }; + explicit DCSSimController(quint16 port, const TlsConfig &tls = {}, QObject *parent = nullptr); + bool listen(); private Q_SLOTS: @@ -38,14 +82,16 @@ void onDisconnected(); private: + bool initTls(); void handlePacket(const Mqtt::Packet &packet); void handleConnectPacket(const Mqtt::Packet &packet); void handlePublishPacket(const Mqtt::Packet &packet); void dumpPayload(const QByteArray &payload); - QTcpServer _server; + SslTcpServer _server; QPointer _client; QByteArray _rxBuf; quint16 _port; + TlsConfig _tlsConfig; quint64 _publishCount = 0; }; Index: tools/DCSsim/DCSsim.pro =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/DCSsim/DCSsim.pro (.../DCSsim.pro) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ tools/DCSsim/DCSsim.pro (.../DCSsim.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -1,6 +1,6 @@ TEMPLATE = app TARGET = DCSsim -CONFIG += c++20 +CONFIG += c++17 QT += core network QT -= gui @@ -11,11 +11,13 @@ DESTDIR = $$PWD/../bin HEADERS = \ - DCSSimController.h + DCSSimController.h \ + MqttPacket.h SOURCES = \ DCSSimController.cpp \ - main.cpp + main.cpp \ + MqttPacket.cpp INCLUDEPATH += $$PWD Index: tools/DCSsim/MqttPacket.cpp =================================================================== diff -u --- tools/DCSsim/MqttPacket.cpp (revision 0) +++ tools/DCSsim/MqttPacket.cpp (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -0,0 +1,380 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPacket.cpp + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#include "MqttPacket.h" + +namespace Mqtt { + +constexpr int MAX_REMAINING_LENGTH_BYTES = 4; +constexpr quint8 PROTOCOL_LEVEL_311 = 0x04; +constexpr quint8 CONNECT_FLAG_CLEAN_SESSION = 0x02; + +/*! + * \brief encodeRemainingLength + * \details Encodes a length as the MQTT variable-length field: seven value bits + * per byte, high bit set while more bytes follow. + * \param length Byte count of everything after the fixed header. + * \return One to four encoded bytes. + */ +QByteArray encodeRemainingLength(quint32 length) +{ + QByteArray out; + do { + quint8 byte = static_cast(length % 128); + length /= 128; + if (length > 0) { + byte = static_cast(byte | 0x80); + } + out.append(static_cast(byte)); + } while (length > 0); + return out; +} + +/*! + * \brief decodeRemainingLength + * \details Reads the variable-length field starting at offset. + * \param bytes Buffer to read from. + * \param offset Index of the first length byte. + * \param value Receives the decoded length. + * \return Bytes consumed, 0 when more input is needed, or -1 when malformed. + */ +int decodeRemainingLength(const QByteArray &bytes, int offset, quint32 &value) +{ + quint32 multiplier = 1; + value = 0; + + for (int i = 0; i < MAX_REMAINING_LENGTH_BYTES; ++i) { + if (offset + i >= bytes.size()) { + return 0; + } + const quint8 byte = static_cast(bytes.at(offset + i)); + value += static_cast(byte & 0x7F) * multiplier; + if ((byte & 0x80) == 0) { + return i + 1; + } + multiplier *= 128; + } + return -1; +} + +/*! + * \brief appendString + * \details Appends a UTF-8 string in MQTT form: a two-byte big-endian length + * followed by the encoded bytes. + * \param out Buffer to append to. + * \param value String to encode. + */ +void appendString(QByteArray &out, const QString &value) +{ + const QByteArray utf8 = value.toUtf8(); + out.append(static_cast((utf8.size() >> 8) & 0xFF)); + out.append(static_cast(utf8.size() & 0xFF)); + out.append(utf8); +} + +/*! + * \brief readString + * \details Reads one length-prefixed UTF-8 string and advances pos past it. + * \param bytes Buffer to read from. + * \param pos Read cursor, advanced on success. + * \param end One past the last readable index. + * \param out Receives the decoded string. + * \return true if a complete string was read. + */ +bool readString(const QByteArray &bytes, int &pos, int end, QString &out) +{ + if (pos + 2 > end) { + return false; + } + const int length = (static_cast(bytes.at(pos)) << 8) | static_cast(bytes.at(pos + 1)); + pos += 2; + + if (pos + length > end) { + return false; + } + out = QString::fromUtf8(bytes.constData() + pos, length); + pos += length; + return true; +} + +/*! + * \brief readUint16 + * \details Reads one big-endian 16-bit value and advances pos past it. + * \param bytes Buffer to read from. + * \param pos Read cursor, advanced on success. + * \param end One past the last readable index. + * \param out Receives the decoded value. + * \return true if two bytes were available. + */ +bool readUint16(const QByteArray &bytes, int &pos, int end, quint16 &out) +{ + if (pos + 2 > end) { + return false; + } + out = static_cast((static_cast(bytes.at(pos)) << 8) | static_cast(bytes.at(pos + 1))); + pos += 2; + return true; +} + +/*! + * \brief buildFixedHeader + * \details Prepends the fixed header to an already-built packet body. + * \param type Control packet type. + * \param flags Low nibble of byte one. + * \param body Variable header and payload. + * \return The complete packet. + */ +QByteArray buildFixedHeader(Mqtt::PacketType type, quint8 flags, const QByteArray &body) +{ + QByteArray out; + out.append(static_cast((static_cast(type) << 4) | (flags & 0x0F))); + out.append(encodeRemainingLength(static_cast(body.size()))); + out.append(body); + return out; +} + +} // namespace + +namespace Mqtt { + +/*! + * \brief Mqtt::buildConnect + * \details Builds a CONNECT for a publish-only session: no will, no credentials. + * \param clientId Client identifier; must match the IoT Thing name on AWS. + * \param keepAliveSecs Keep-alive interval advertised to the broker. + * \param cleanSession True to ask the broker to retain no session state. + * \return The encoded packet. + */ +QByteArray buildConnect(const QString &clientId, quint16 keepAliveSecs, bool cleanSession) +{ + QByteArray body; + appendString(body, QStringLiteral("MQTT")); + body.append(static_cast(PROTOCOL_LEVEL_311)); + body.append(static_cast(cleanSession ? CONNECT_FLAG_CLEAN_SESSION : 0x00)); + body.append(static_cast((keepAliveSecs >> 8) & 0xFF)); + body.append(static_cast(keepAliveSecs & 0xFF)); + appendString(body, clientId); + + return buildFixedHeader(PacketType::Connect, 0, body); +} + +/*! + * \brief Mqtt::buildConnAck + * \details Builds the broker's response to a CONNECT. + * \param code Acceptance or the reason for refusal. + * \param sessionPresent True when the broker resumed an existing session. + * \return The encoded packet. + */ +QByteArray buildConnAck(ConnAckCode code, bool sessionPresent) +{ + QByteArray body; + body.append(static_cast(sessionPresent ? 0x01 : 0x00)); + body.append(static_cast(code)); + + return buildFixedHeader(PacketType::ConnAck, 0, body); +} + +/*! + * \brief Mqtt::buildPublish + * \details Builds a PUBLISH. The packet identifier is only present at QoS > 0. + * \param topic Fully resolved topic name. + * \param payload Application bytes, passed through untouched. + * \param qos Delivery quality, 0 or 1. + * \param packetId Identifier to correlate the PUBACK; ignored at QoS 0. + * \return The encoded packet. + */ +QByteArray buildPublish(const QString &topic, const QByteArray &payload, quint8 qos, quint16 packetId) +{ + QByteArray body; + appendString(body, topic); + if (qos > 0) { + body.append(static_cast((packetId >> 8) & 0xFF)); + body.append(static_cast(packetId & 0xFF)); + } + body.append(payload); + + return buildFixedHeader(PacketType::Publish, static_cast((qos & 0x03) << 1), body); +} + +/*! + * \brief Mqtt::buildPubAck + * \details Builds the QoS 1 acknowledgement for a received PUBLISH. + * \param packetId Identifier copied from the PUBLISH being acknowledged. + * \return The encoded packet. + */ +QByteArray buildPubAck(quint16 packetId) +{ + QByteArray body; + body.append(static_cast((packetId >> 8) & 0xFF)); + body.append(static_cast(packetId & 0xFF)); + + return buildFixedHeader(PacketType::PubAck, 0, body); +} + +/*! + * \brief Mqtt::buildPingReq + * \details Builds the keep-alive request. + * \return The encoded packet. + */ +QByteArray buildPingReq() +{ + return buildFixedHeader(PacketType::PingReq, 0, QByteArray()); +} + +/*! + * \brief Mqtt::buildPingResp + * \details Builds the keep-alive response. + * \return The encoded packet. + */ +QByteArray buildPingResp() +{ + return buildFixedHeader(PacketType::PingResp, 0, QByteArray()); +} + +/*! + * \brief Mqtt::buildDisconnect + * \details Builds the graceful shutdown notification. + * \return The encoded packet. + */ +QByteArray buildDisconnect() +{ + return buildFixedHeader(PacketType::Disconnect, 0, QByteArray()); +} + +/*! + * \brief Mqtt::readPacket + * \details Decodes one packet from the front of bytes, consuming it on success. + * \param bytes Accumulated stream data; modified in place. + * \param packet Receives the decoded packet. + * \return Complete when a packet was consumed, Incomplete when more data is + * needed, or Error when the stream is malformed. + */ +ReadState readPacket(QByteArray &bytes, Packet &packet) +{ + if (bytes.size() < 2) { + return ReadState::Incomplete; + } + + const quint8 header = static_cast(bytes.at(0)); + const quint8 rawType = static_cast(header >> 4); + const quint8 flags = static_cast(header & 0x0F); + + quint32 remainingLength = 0; + const int lengthBytes = decodeRemainingLength(bytes, 1, remainingLength); + if (lengthBytes == 0) { + return ReadState::Incomplete; + } + if (lengthBytes < 0) { + return ReadState::Error; + } + + const int headerSize = 1 + lengthBytes; + const int totalSize = headerSize + static_cast(remainingLength); + if (bytes.size() < totalSize) { + return ReadState::Incomplete; + } + + packet = Packet(); + packet.type = static_cast(rawType); + + int pos = headerSize; + const int end = totalSize; + bool ok = true; + + switch (packet.type) { + case PacketType::Connect: { + QString protocolName; + ok = readString(bytes, pos, end, protocolName) && protocolName == QStringLiteral("MQTT"); + if (ok && pos + 2 <= end) { + // Protocol level is validated by the caller, which decides the CONNACK code. + pos++; + packet.cleanSession = (static_cast(bytes.at(pos)) & CONNECT_FLAG_CLEAN_SESSION) != 0; + pos++; + ok = readUint16(bytes, pos, end, packet.keepAliveSecs) && readString(bytes, pos, end, packet.clientId); + } + else { + ok = false; + } + break; + } + + case PacketType::ConnAck: { + if (pos + 2 <= end) { + packet.sessionPresent = (static_cast(bytes.at(pos)) & 0x01) != 0; + packet.returnCode = static_cast(static_cast(bytes.at(pos + 1))); + pos += 2; + } + else { + ok = false; + } + break; + } + + case PacketType::Publish: { + packet.dup = (flags & 0x08) != 0; + packet.qos = static_cast((flags >> 1) & 0x03); + packet.retain = (flags & 0x01) != 0; + + ok = readString(bytes, pos, end, packet.topic); + if (ok && packet.qos > 0) { + ok = readUint16(bytes, pos, end, packet.packetId); + } + if (ok) { + packet.payload = bytes.mid(pos, end - pos); + pos = end; + } + break; + } + + case PacketType::PubAck: + ok = readUint16(bytes, pos, end, packet.packetId); + break; + + case PacketType::PingReq: + case PacketType::PingResp: + case PacketType::Disconnect: + break; + + default: + ok = false; + break; + } + + if (!ok) { + return ReadState::Error; + } + + bytes.remove(0, totalSize); + return ReadState::Complete; +} + +/*! + * \brief Mqtt::typeName + * \details Maps a packet type to its specification name, for logging. + * \param type Packet type to name. + * \return The name, or "UNKNOWN" for a type this codec does not handle. + */ +QString typeName(PacketType type) +{ + switch (type) { + case PacketType::Connect: return QStringLiteral("CONNECT"); + case PacketType::ConnAck: return QStringLiteral("CONNACK"); + case PacketType::Publish: return QStringLiteral("PUBLISH"); + case PacketType::PubAck: return QStringLiteral("PUBACK"); + case PacketType::PingReq: return QStringLiteral("PINGREQ"); + case PacketType::PingResp: return QStringLiteral("PINGRESP"); + case PacketType::Disconnect: return QStringLiteral("DISCONNECT"); + } + return QStringLiteral("UNKNOWN"); +} + +} // namespace Mqtt Index: tools/DCSsim/MqttPacket.h =================================================================== diff -u --- tools/DCSsim/MqttPacket.h (revision 0) +++ tools/DCSsim/MqttPacket.h (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -0,0 +1,86 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MqttPacket.h + * \author (original) Stephen Quong + * \date (original) 30-Jul-2026 + * + */ +#pragma once + +#include +#include + +/*! + * \brief MQTT packet encoding and decoding. + */ +namespace Mqtt { + +enum class PacketType : quint8 +{ + Connect = 1, + ConnAck = 2, + Publish = 3, + PubAck = 4, + PingReq = 12, + PingResp = 13, + Disconnect = 14, +}; + +enum class ConnAckCode : quint8 +{ + Accepted = 0, + UnacceptableProtocol = 1, + IdentifierRejected = 2, + ServerUnavailable = 3, + BadCredentials = 4, + NotAuthorized = 5, +}; + +// Outcome of one readPacket() call. +enum class ReadState +{ + Incomplete, // Need more bytes; nothing was consumed. + Complete, // One packet was decoded and consumed. + Error, // Malformed; the caller should drop the connection. +}; + +/*! + * \brief One decoded MQTT packet. + */ +struct Packet +{ + PacketType type = PacketType::Disconnect; + + QString topic; + quint16 packetId = 0; + quint8 qos = 0; + QByteArray payload; + bool dup = false; + bool retain = false; + + QString clientId; + quint16 keepAliveSecs = 0; + bool cleanSession = true; + + ConnAckCode returnCode = ConnAckCode::Accepted; + bool sessionPresent = false; +}; + +QByteArray buildConnect(const QString &clientId, quint16 keepAliveSecs, bool cleanSession); +QByteArray buildConnAck(ConnAckCode code, bool sessionPresent = false); +QByteArray buildPublish(const QString &topic, const QByteArray &payload, quint8 qos, quint16 packetId); +QByteArray buildPubAck(quint16 packetId); +QByteArray buildPingReq(); +QByteArray buildPingResp(); +QByteArray buildDisconnect(); + +ReadState readPacket(QByteArray &bytes, Packet &packet); + +QString typeName(PacketType type); + +} // namespace Mqtt Index: tools/DCSsim/main.cpp =================================================================== diff -u -r59b4c22f45a1d098064a886452769204e90cfb4b -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 --- tools/DCSsim/main.cpp (.../main.cpp) (revision 59b4c22f45a1d098064a886452769204e90cfb4b) +++ tools/DCSsim/main.cpp (.../main.cpp) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) @@ -29,7 +29,13 @@ QCommandLineOption portOption(QStringList() << "p" << "port", "TCP listening port", "port", QString::number(DEFAULT_MQTT_PORT)); + QCommandLineOption certOption("cert", "Server certificate PEM; enables TLS (requires --key).", "file"); + QCommandLineOption keyOption("key", "Server private key PEM.", "file"); + QCommandLineOption caOption("ca", "CA bundle PEM; clients must present a certificate it signs (mTLS).", "file"); parser.addOption(portOption); + parser.addOption(certOption); + parser.addOption(keyOption); + parser.addOption(caOption); parser.process(app); bool ok = false; @@ -39,7 +45,21 @@ return 1; } - DCSSimController simController(static_cast(port)); + if (parser.isSet(certOption) != parser.isSet(keyOption)) { + qCritical().noquote() << "DCSsim: --cert and --key must be given together"; + return 1; + } + if (parser.isSet(caOption) && !parser.isSet(certOption)) { + qCritical().noquote() << "DCSsim: --ca requires --cert and --key"; + return 1; + } + + DCSSimController::TlsConfig tls; + tls.certPath = parser.value(certOption); + tls.keyPath = parser.value(keyOption); + tls.caPath = parser.value(caOption); + + DCSSimController simController(static_cast(port), tls); if (!simController.listen()) { return 1; }