hero123457/qtmqtt
1
0
Fork 0
mirror of https://github.com/qt/qtmqtt.git synced 2026-10-25 16:16:26 +00:00
Code Issues Projects Releases Wiki Activity
06d301aa50
qtmqtt/tests/auto/qmqttclient/tst_qmqttclient.cpp
Tatiana Borisova 06d301aa50 Read the Remaining Length as a variable byte integer everywhere 
- Every MQTT packet begins with a fixed header, whose second field is the
Remaining Length: the number of bytes that follow. It is not a byte. It
is spread over one to four bytes, seven value bits each, with the top
bit meaning "another byte follows". So values up to 127 take one byte,
and larger values take more.

Six packet types read it as a single quint8: SUBACK, UNSUBACK, PUBACK,
PUBREC, PUBREL and PUBCOMP. For most of them that used to be correct.
In MQTT 3.1.1 those packets contain nothing but a two byte identifier,
so the length was always exactly 2, which always fits in one byte. The
code did not even use the byte it read: it checked that the value was 2
and then hardcoded the length.

MQTT 5.0 support (ac66658) gave those packets an optional reason code
and properties, so the length can now be far above 127. That commit
switched off the "must be 2" check for MQTT 5.0 and began using the
byte as the real length, but kept reading only one byte. The check had
been the only thing making a one byte read safe, because a multi-byte
length always has its top bit set and so could never equal 2.

SUBACK never had a fixed length at all, since its payload carries one
reason code per subscribed topic filter. It has been misparsed for any
SUBACK above 127 bytes since the parser was written (d96011b).

Reading one byte of a two byte length consumes one byte too few, so the
remainder of the packet is then read as the next packet's header. The
client rejects a valid acknowledgement and stays out of step for the
rest of the connection. Brokers trigger this on ordinary traffic as
soon as they attach a reason string or user properties.

All six now use readVariableByteInteger(), and each rejects a length too
short to hold the fields the specification requires.

PUBREL also accepted a length of exactly 2 even for MQTT 5.0, which
rejected valid PUBREL packets carrying a reason code. It now accepts 2
or more there, and still requires exactly 2 for MQTT 3.1.1.

- Adds three regression tests driving the parser through the existing
FakeServer, so they need no broker.

Pick-to: 6.11 6.12 6.8
Task-number: QTBUG-149569
Change-Id: Ib239d836556e0cc14e13c116dee757fdd1a7ae81
Reviewed-by: Ivan Solovev <ivan.solovev@qt.io>
2026-09-01 17:40:37 +00:00

1209 lines
42 KiB
C++
Raw Blame History

// Copyright (C) 2017 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include "broker_connection.h"
#include <QtCore/QString>
#include <QtNetwork/QTcpServer>
#include <QtTest/QtTest>
#include <QtTest/QSignalSpy>
#include <QtMqtt/QMqttClient>
#include <limits>
class Tst_QMqttClient : public QObject
{
Q_OBJECT
public:
Tst_QMqttClient();
private Q_SLOTS:
void initTestCase();
void cleanupTestCase();
void getSetCheck_data();
void getSetCheck();
void sendReceive_data();
void sendReceive();
void retainMessage_data();
void retainMessage();
void willMessage_data();
void willMessage();
void compliantTopic_data();
void compliantTopic();
void subscribeLongTopic_data();
void subscribeLongTopic();
void dataIncludingZero_data();
void dataIncludingZero();
void publishLongTopic_data();
void publishLongTopic();
void reconnect_QTBUG65726_data();
void reconnect_QTBUG65726();
void openIODevice_QTBUG66955_data();
void openIODevice_QTBUG66955();
void subackLongRemainingLength();
void pubackLongRemainingLength();
void pubrelWithReasonCode();
void shortRemainingLength_data();
void shortRemainingLength();
void staticProperties_QTBUG_67176_data();
void staticProperties_QTBUG_67176();
void authentication();
void messageStatus_data();
void messageStatus();
void messageStatusReceive_data();
void messageStatusReceive();
void subscriptionIdsOverlap();
void keepAlive_data();
void keepAlive();
void reconnect();
private:
QProcess m_brokerProcess;
QString m_testBroker;
quint16 m_port{0};
};
Tst_QMqttClient::Tst_QMqttClient()
{
}
void Tst_QMqttClient::initTestCase()
{
m_testBroker = invokeOrInitializeBroker(&m_brokerProcess);
if (m_testBroker.isEmpty())
qFatal("No MQTT broker present to test against.");
}
void Tst_QMqttClient::cleanupTestCase()
{
}
DefaultVersionTestData(Tst_QMqttClient::getSetCheck_data)
void Tst_QMqttClient::getSetCheck()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
VersionClient(mqttVersion, client);
QVERIFY(client.clientId().size() > 0);
const QString clientId = QLatin1String("testclient123");
client.setClientId(clientId);
QCOMPARE(client.clientId(), clientId);
QCOMPARE(client.hostname(), QString());
const QString hostname = QLatin1String("qt.io");
client.setHostname(hostname);
QCOMPARE(client.hostname(), hostname);
QCOMPARE(client.port(), quint16(0));
client.setPort(1883);
QCOMPARE(client.port(), quint16(1883));
QCOMPARE(client.keepAlive(), quint16(60));
client.setKeepAlive(10);
QCOMPARE(client.keepAlive(), quint16(10));
// Available protocol versions
QMqttClient client2;
QCOMPARE(client2.protocolVersion(), QMqttClient::MQTT_3_1_1);
client2.setProtocolVersion(QMqttClient::ProtocolVersion(0));
QCOMPARE(client2.protocolVersion(), QMqttClient::MQTT_3_1_1);
client2.setProtocolVersion(QMqttClient::ProtocolVersion(6));
QCOMPARE(client2.protocolVersion(), QMqttClient::MQTT_3_1_1);
client2.setProtocolVersion(QMqttClient::MQTT_3_1);
QCOMPARE(client2.protocolVersion(), QMqttClient::MQTT_3_1);
client2.setProtocolVersion(QMqttClient::MQTT_5_0);
QCOMPARE(client2.protocolVersion(), QMqttClient::MQTT_5_0);
#ifdef QT_BUILD_INTERNAL
if (qEnvironmentVariableIsSet("QT_MQTT_TEST_USERNAME"))
QEXPECT_FAIL("", "Default username has been overwritten.", Continue);
#endif
QCOMPARE(client.username(), QString());
#ifdef QT_BUILD_INTERNAL
if (qEnvironmentVariableIsSet("QT_MQTT_TEST_PASSWORD"))
QEXPECT_FAIL("", "Default username has been overwritten.", Continue);
#endif
QCOMPARE(client.password(), QString());
QCOMPARE(client.cleanSession(), true);
QCOMPARE(client.willTopic(), QString());
QCOMPARE(client.willMessage(), QByteArray());
QCOMPARE(client.willQoS(), quint8(0));
QCOMPARE(client.willRetain(), false);
QCOMPARE(client.autoKeepAlive(), true);
client.setAutoKeepAlive(false);
QCOMPARE(client.autoKeepAlive(), false);
}
void Tst_QMqttClient::sendReceive_data()
{
QTest::addColumn<QMqttClient::ProtocolVersion>("mqttVersion");
QTest::addColumn<QByteArray>("data");
QList<QMqttClient::ProtocolVersion> versions{QMqttClient::MQTT_3_1_1, QMqttClient::MQTT_5_0};
for (int i = 0; i < 2; ++i) {
QTest::newRow(qPrintable(QString::number(versions[i]) + ":empty")) << versions[i] << QByteArray();
QTest::newRow(qPrintable(QString::number(versions[i]) + ":simple")) << versions[i] << QByteArray("This is a test message");
QByteArray d;
d.fill('A', 500);
QTest::newRow(qPrintable(QString::number(versions[i]) + ":big")) << versions[i] << d;
d.fill('B', (128 * 128 * 128) + 4);
QTest::newRow(qPrintable(QString::number(versions[i]) + ":huge")) << versions[i] << d;
}
}
void Tst_QMqttClient::sendReceive()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
QFETCH(QByteArray, data);
const QString testTopic = QLatin1String("Qt/QMqttClient/Topic");
VersionClient(mqttVersion, publisher);
publisher.setClientId(QLatin1String("publisher"));
publisher.setHostname(m_testBroker);
publisher.setPort(m_port);
publisher.connectToHost();
QTRY_COMPARE(publisher.state(), QMqttClient::Connected);
VersionClient(mqttVersion, subscriber);
subscriber.setClientId(QLatin1String("subscriber"));
subscriber.setHostname(m_testBroker);
subscriber.setPort(m_port);
subscriber.connectToHost();
QTRY_COMPARE(subscriber.state(), QMqttClient::Connected);
bool received = false;
bool verified = false;
auto sub = subscriber.subscribe(testTopic, 1);
QVERIFY(sub);
connect(sub, &QMqttSubscription::messageReceived, sub, [&](QMqttMessage msg) {
verified = msg.payload() == data;
received = true;
});
QTRY_COMPARE(sub->state(), QMqttSubscription::Subscribed);
if (subscriber.protocolVersion() == QMqttClient::MQTT_5_0 &&
(quint32)data.size() > subscriber.serverConnectionProperties().maximumPacketSize())
QSKIP("The MQTT 5 test broker does not support huge packages.", SkipOnce);
publisher.publish(testTopic, data, 1);
QTRY_VERIFY2_WITH_TIMEOUT(received, "Subscriber did not receive message", std::chrono::seconds(180));
QVERIFY2(verified, "Subscriber received different message");
}
DefaultVersionTestData(Tst_QMqttClient::retainMessage_data)
void Tst_QMqttClient::retainMessage()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
const QString testTopic = QLatin1String("Qt/QMqttClient/Topic2");
const QByteArray testMessage("retainedMessage");
// Publisher
VersionClient(mqttVersion, publisher);
publisher.setClientId(QLatin1String("publisher"));
publisher.setHostname(m_testBroker);
publisher.setPort(m_port);
publisher.connectToHost();
QTRY_COMPARE(publisher.state(), QMqttClient::Connected);
// [MQTT-3.3.1-7]
// If the Server receives a QoS 0 message with the RETAIN flag set to 1 it MUST
// discard any message previously retained for that topic. It SHOULD store the
// new QoS 0 message
publisher.publish(testTopic, QByteArray(), 0, true);
// We cannot wait as 0 QoS does not send confirmation
QTest::qWait(500);
for (int i = 0; i < 2; i++) {
int msgCount = 0;
QSignalSpy publishSpy(&publisher, SIGNAL(messageSent(qint32)));
publisher.publish(testTopic, testMessage, 1, i == 1 ? true : false);
QTRY_COMPARE(publishSpy.size(), 1);
VersionClient(mqttVersion, sub);
sub.setClientId(QLatin1String("SubA"));
sub.setHostname(m_testBroker);
sub.setPort(m_port);
connect(&sub, &QMqttClient::messageReceived, &sub,
[&msgCount, testMessage](const QByteArray &msg) {
if (msg == testMessage)
msgCount++;
});
QSignalSpy messageSpy(&sub, SIGNAL(messageReceived(QByteArray,QMqttTopicName)));
sub.connectToHost();
QTRY_COMPARE(sub.state(), QMqttClient::Connected);
auto subscription = sub.subscribe(testTopic);
QTRY_COMPARE(subscription->state(), QMqttSubscription::Subscribed);
QTRY_VERIFY(msgCount == i);
}
publisher.disconnect();
}
void Tst_QMqttClient::willMessage_data()
{
QTest::addColumn<QMqttClient::ProtocolVersion>("mqttVersion");
QTest::addColumn<QByteArray>("willMessage");
QList<QMqttClient::ProtocolVersion> versions{QMqttClient::MQTT_3_1_1, QMqttClient::MQTT_5_0};
for (int i = 0; i < 2; ++i) {
QTest::newRow(qPrintable(QString::number(versions[i]) + ":simple")) << versions[i] << QByteArray("The client connection is gone.");
QTest::newRow(qPrintable(QString::number(versions[i]) + ":empty")) << versions[i] << QByteArray();
}
}
void Tst_QMqttClient::willMessage()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
QFETCH(QByteArray, willMessage);
const QString willTopic = QLatin1String("Qt/QMqttClient/will/topic");
// Client A connects
VersionClient(mqttVersion, client1);
client1.setHostname(m_testBroker);
client1.setPort(m_port);
client1.connectToHost();
QTRY_COMPARE(client1.state(), QMqttClient::Connected);
if (!client1.isTcp())
QSKIP("This test is only for tcp");
auto client1Sub = client1.subscribe(willTopic, 1);
connect(client1Sub, &QMqttSubscription::messageReceived,
client1Sub, [=](QMqttMessage message) {
Q_UNUSED(message);
// Just debug purposes
//qDebug() << "Got something:" << message;
});
QTRY_COMPARE(client1Sub->state(), QMqttSubscription::Subscribed);
QSignalSpy messageSpy(client1Sub, SIGNAL(messageReceived(QMqttMessage)));
// Client B connects (via TcpSocket)
QTcpSocket sock;
sock.connectToHost(client1.hostname(), client1.port());
QVERIFY(sock.waitForConnected());
for (int i = 1; i > 0; --i) {
VersionClient(mqttVersion, willClient);
if (i == 1)
willClient.setTransport(&sock, QMqttClient::AbstractSocket);
else {
willClient.setHostname(m_testBroker);
willClient.setPort(m_port);
}
willClient.setWillQoS(1);
willClient.setWillTopic(willTopic);
willClient.setWillMessage(willMessage);
willClient.connectToHost();
QTRY_COMPARE(willClient.state(), QMqttClient::Connected);
willClient.publish(QLatin1String("noninteresting"), "just something");
// Be evil and kill the connection without DISCONNECT
// Should send will message to client1.
// When you manually disconnect (send the DISCONNECT command) no will message
// is sent
if (i == 1)
sock.disconnectFromHost();
else
willClient.disconnectFromHost();
QTest::qWait(500);
QTRY_COMPARE(messageSpy.size(), i);
}
}
void Tst_QMqttClient::compliantTopic_data()
{
QTest::addColumn<QMqttClient::ProtocolVersion>("mqttVersion");
QTest::addColumn<QString>("topic");
QList<QMqttClient::ProtocolVersion> versions{QMqttClient::MQTT_3_1_1, QMqttClient::MQTT_5_0};
for (int i = 0; i < 2; ++i) {
QTest::newRow(qPrintable(QString::number(versions[i]) + ":simple")) << versions[i] << QString::fromLatin1("Qt/QMqttClient/topic");
QTest::newRow(qPrintable(QString::number(versions[i]) + ":subPath")) << versions[i] << QString::fromLatin1("Qt/QMqttClient/topic/subtopic");
if (versions[i] != QMqttClient::MQTT_5_0) {
QString l;
l.fill(QLatin1Char('T'), std::numeric_limits<std::uint16_t>::max());
QTest::newRow(qPrintable(QString::number(versions[i]) + ":maxSize")) << versions[i] << l;
}
}
}
void Tst_QMqttClient::compliantTopic()
{
QFETCH(QString, topic);
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
QString truncTopic = topic;
truncTopic.truncate(std::numeric_limits<std::uint16_t>::max());
VersionClient(mqttVersion, publisher);
publisher.setClientId(QLatin1String("publisher"));
publisher.setHostname(m_testBroker);
publisher.setPort(m_port);
publisher.connectToHost();
QTRY_COMPARE(publisher.state(), QMqttClient::Connected);
VersionClient(mqttVersion, subscriber);
subscriber.setClientId(QLatin1String("subscriber"));
subscriber.setHostname(m_testBroker);
subscriber.setPort(m_port);
subscriber.connectToHost();
QTRY_COMPARE(subscriber.state(), QMqttClient::Connected);
bool received = false;
bool verified = false;
connect(&subscriber, &QMqttClient::messageReceived,
&subscriber, [&](const QByteArray &, const QMqttTopicName &t) {
received = true;
verified = t == truncTopic;
});
auto client1Sub = subscriber.subscribe(truncTopic, 1);
QTRY_COMPARE(client1Sub->state(), QMqttSubscription::Subscribed);
publisher.publish(topic, QByteArray("Msgs"), 1);
QTRY_VERIFY2(received, "Subscriber did not receive message");
QVERIFY2(verified, "Subscriber received different message");
}
DefaultVersionTestData(Tst_QMqttClient::subscribeLongTopic_data)
void Tst_QMqttClient::subscribeLongTopic()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
VersionClient(mqttVersion, subscriber);
subscriber.setClientId(QLatin1String("subscriber"));
subscriber.setHostname(m_testBroker);
subscriber.setPort(m_port);
subscriber.connectToHost();
QTRY_COMPARE(subscriber.state(), QMqttClient::Connected);
QString topic;
topic.fill(QLatin1Char('s'), 2 * std::numeric_limits<std::uint16_t>::max());
auto sub = subscriber.subscribe(topic);
QCOMPARE(sub, nullptr);
}
DefaultVersionTestData(Tst_QMqttClient::dataIncludingZero_data)
void Tst_QMqttClient::dataIncludingZero()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
QByteArray data;
const int dataSize = 200;
data.fill('A', dataSize);
data[100] = '\0';
VersionClient(mqttVersion, client);
client.setHostname(m_testBroker);
client.setPort(m_port);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
bool received = false;
bool verified = false;
bool correctSize = false;
const QString testTopic(QLatin1String("Qt/QMqttClient/some/topic"));
auto sub = client.subscribe(testTopic, 1);
QVERIFY(sub);
connect(sub, &QMqttSubscription::messageReceived,
sub, [&](QMqttMessage msg) {
verified = msg.payload() == data;
correctSize = msg.payload().size() == dataSize;
received = true;
});
QTRY_COMPARE(sub->state(), QMqttSubscription::Subscribed);
client.publish(testTopic, data, 1);
QTRY_VERIFY2(received, "Subscriber did not receive message");
QVERIFY2(verified, "Subscriber received different message");
QVERIFY2(correctSize, "Subscriber received message of different size");
}
DefaultVersionTestData(Tst_QMqttClient::publishLongTopic_data)
void Tst_QMqttClient::publishLongTopic()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
VersionClient(mqttVersion, publisher);
publisher.setClientId(QLatin1String("publisher"));
publisher.setHostname(m_testBroker);
publisher.setPort(m_port);
publisher.connectToHost();
QTRY_COMPARE(publisher.state(), QMqttClient::Connected);
QString topic;
topic.fill(QLatin1Char('s'), 2 * std::numeric_limits<std::uint16_t>::max());
auto pub = publisher.publish(topic);
QCOMPARE(pub, -1);
}
class FakeServer : public QObject
{
Q_OBJECT
public:
FakeServer() {
server = new QTcpServer(this);
connect(server, &QTcpServer::newConnection, this, &FakeServer::createSocket);
server->listen(QHostAddress::Any, 5726);
}
// Sends raw bytes to the client, so a test can drive the parser with an exact
// byte sequence.
void send(const QByteArray &data) { socket->write(data); }
public slots:
void createSocket() {
socket = server->nextPendingConnection();
connectionAcknowledged = false;
lastRequest.clear();
connect(socket, &QTcpSocket::readyRead, this, &FakeServer::connectionRequested);
}
void connectionRequested() {
// We assume the first packet is always a connect statement, so no verification
// is done. Everything after that is recorded for the test to answer itself.
if (connectionAcknowledged) {
lastRequest.append(socket->readAll());
return;
}
connectionAcknowledged = true;
socket->readAll();
QByteArray response;
response += 0x20;
response += quint8(2); // Payload size
if (!connectionSuccess) {
response += quint8(255); // Causes ProtocolViolation
response += quint8(13);
} else {
response += char(0); // ackFlags
response += char(0); // result
if (version == QMqttClient::MQTT_5_0) {
response += char(0); // No properties
response[1] = response[1] + 1; // Adjust payload size
}
}
qDebug() << "Fake server response:" << connectionSuccess;
socket->write(response);
}
public:
QTcpServer *server;
QTcpSocket *socket;
QMqttClient::ProtocolVersion version{QMqttClient::MQTT_3_1_1};
bool connectionSuccess{false};
bool connectionAcknowledged{false};
QByteArray lastRequest;
};
DefaultVersionTestData(Tst_QMqttClient::reconnect_QTBUG65726_data)
void Tst_QMqttClient::reconnect_QTBUG65726()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
FakeServer server;
VersionClient(mqttVersion, client);
client.setClientId(QLatin1String("bugclient"));
client.setHostname(QLatin1String("localhost"));
client.setPort(5726);
if (!client.isTcp())
QSKIP("This test is only for tcp");
server.version = client.protocolVersion();
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
QTRY_COMPARE(client.error(), QMqttClient::ProtocolViolation);
server.connectionSuccess = true;
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
QTRY_COMPARE(client.error(), QMqttClient::NoError);
}
class IOTransport : public QIODevice
{
public:
bool open(OpenMode mode) override {
return QIODevice::open(mode);
}
qint64 readData(char *data, qint64 maxlen) override {
Q_UNUSED(data);
Q_UNUSED(maxlen);
return 0;
}
qint64 writeData(const char *data, qint64 len) override {
Q_UNUSED(data);
Q_UNUSED(len);
if (data[0] == 0x10)
written = 1;
else
qDebug() << "Received unknown/invalid data";
return len;
}
QAtomicInt written{0};
};
DefaultVersionTestData(Tst_QMqttClient::openIODevice_QTBUG66955_data)
void Tst_QMqttClient::openIODevice_QTBUG66955()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
IOTransport trans;
trans.open(QIODevice::ReadWrite);
VersionClient(mqttVersion, client);
client.setTransport(&trans, QMqttClient::IODevice);
client.connectToHost();
QTRY_COMPARE(trans.written, 1);
}
// A reason string property (0x1f) of the requested content length.
static QByteArray reasonStringProperty(quint16 length)
{
QByteArray property;
property.append(char(0x1f));
property.append(char(length >> 8));
property.append(char(length & 0xff));
property.append(QByteArray(length, 'r'));
return property;
}
// The packet identifier of a packet the client wrote, skipping the packet type and
// its variable byte remaining length.
static quint16 packetIdentifier(const QByteArray &packet)
{
qsizetype pos = 1;
while (pos < packet.size() && (packet.at(pos) & 128) != 0)
++pos;
Q_ASSERT(pos < packet.size() - 2);
++pos;
return quint16(quint8(packet.at(pos)) << 8 | quint8(packet.at(pos + 1)));
}
static void appendPacketIdentifier(QByteArray *payload, quint16 id)
{
payload->append(char(id >> 8));
payload->append(char(id & 0xff));
}
// MQTT-2.2.3 (3.1.1) / MQTT-2.1.4 (5.0) The remaining length is a variable byte
// integer, so a packet longer than 127 bytes carries a two byte length whose first
// byte has the continuation bit set. Reading only that first byte consumes one byte
// too few and leaves the rest of the payload to be parsed as the next packet's
// fixed header.
void Tst_QMqttClient::subackLongRemainingLength()
{
FakeServer server;
server.version = QMqttClient::MQTT_5_0;
server.connectionSuccess = true;
VersionClient(QMqttClient::MQTT_5_0, client);
if (!client.isTcp())
QSKIP("This test is only for tcp");
client.setHostname(QLatin1String("localhost"));
client.setPort(5726);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
auto sub = client.subscribe(QMqttTopicFilter(QLatin1String("a/b")), 1);
QVERIFY(sub);
QTRY_VERIFY(server.lastRequest.size() >= 4); // The SUBSCRIBE arrived
// SUBACK with a reason string, which pushes the remaining length past 127.
QByteArray payload;
appendPacketIdentifier(&payload, packetIdentifier(server.lastRequest));
const QByteArray properties = reasonStringProperty(123);
payload.append(char(properties.size())); // Property length, 126
payload.append(properties);
payload.append(char(0x01)); // Reason code: maximum QoS 1
QCOMPARE(payload.size(), 130);
QByteArray packet;
packet.append(char(0x90));
packet.append(char(0x82)); // Remaining length 130, low seven bits plus continuation
packet.append(char(0x01)); // Remaining length 130, remaining bits
packet.append(payload);
server.send(packet);
QTRY_COMPARE(sub->state(), QMqttSubscription::Subscribed);
QCOMPARE(client.state(), QMqttClient::Connected);
QCOMPARE(client.error(), QMqttClient::NoError);
}
void Tst_QMqttClient::pubackLongRemainingLength()
{
FakeServer server;
server.version = QMqttClient::MQTT_5_0;
server.connectionSuccess = true;
VersionClient(QMqttClient::MQTT_5_0, client);
if (!client.isTcp())
QSKIP("This test is only for tcp");
client.setHostname(QLatin1String("localhost"));
client.setPort(5726);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
const qint32 id = client.publish(QMqttTopicName(QLatin1String("a/b")), QByteArray("x"), 1);
QVERIFY(id > 0);
QTRY_VERIFY(server.lastRequest.size() >= 4); // The PUBLISH arrived
QSignalSpy sentSpy(&client, &QMqttClient::messageSent);
// PUBACK with a reason string, which pushes the remaining length past 127.
QByteArray payload;
appendPacketIdentifier(&payload, quint16(id));
payload.append(char(0x00)); // Reason code: Success
const QByteArray properties = reasonStringProperty(123);
payload.append(char(properties.size())); // Property length, 126
payload.append(properties);
QCOMPARE(payload.size(), 130);
QByteArray packet;
packet.append(char(0x40));
packet.append(char(0x82)); // Remaining length 130, low seven bits plus continuation
packet.append(char(0x01)); // Remaining length 130, remaining bits
packet.append(payload);
server.send(packet);
QTRY_COMPARE(sentSpy.count(), 1);
QCOMPARE(sentSpy.at(0).at(0).toInt(), id);
QCOMPARE(client.state(), QMqttClient::Connected);
QCOMPARE(client.error(), QMqttClient::NoError);
}
// MQTT-3.6.2, MQTT-3.6.2.1 In MQTT 5.0 the PUBREL variable header holds a reason
// code and properties in addition to the packet identifier. A remaining length of 2
// only occurs when both are omitted, so a longer PUBREL must be accepted.
void Tst_QMqttClient::pubrelWithReasonCode()
{
FakeServer server;
server.version = QMqttClient::MQTT_5_0;
server.connectionSuccess = true;
VersionClient(QMqttClient::MQTT_5_0, client);
if (!client.isTcp())
QSKIP("This test is only for tcp");
client.setHostname(QLatin1String("localhost"));
client.setPort(5726);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
QMqtt::MessageStatus status = QMqtt::MessageStatus::Unknown;
QString reasonString;
connect(&client, &QMqttClient::messageStatusChanged, &client,
[&](qint32, QMqtt::MessageStatus s, const QMqttMessageStatusProperties &properties)
{
status = s;
reasonString = properties.reason();
});
// PUBLISH, QoS 2, topic "a/b", packet identifier 7, no properties, payload "x".
// The client answers with a PUBREC, which puts it in the middle of the QoS 2
// exchange and makes it expect a PUBREL.
server.send(QByteArray::fromHex("34090003612f6200070078"));
QTRY_VERIFY(server.lastRequest.size() >= 4); // The PUBREC arrived
// PUBREL, packet identifier 7, reason code Success, and a reason string. Remaining
// length 12: 2 identifier + 1 reason code + 1 property length + 8 property bytes.
server.send(QByteArray::fromHex("620c000700081f000568656c6c6f"));
QTRY_COMPARE(status, QMqtt::MessageStatus::Released);
QCOMPARE(reasonString, QLatin1String("hello"));
QCOMPARE(client.state(), QMqttClient::Connected);
QCOMPARE(client.error(), QMqttClient::NoError);
}
// The remaining length counts the bytes that follow the fixed header, so a packet
// declaring less than the fields the specification requires is malformed. Accepting
// one leaves the parser reading past the end of the packet: the reason code loops in
// finalize_suback() and finalize_unsuback() read at least once regardless of what is
// left, and finalize_pubAckRecRelComp() always reads a packet identifier. Each of
// these packets must be rejected as a protocol violation instead.
void Tst_QMqttClient::shortRemainingLength_data()
{
QTest::addColumn<QMqttClient::ProtocolVersion>("mqttVersion");
QTest::addColumn<QByteArray>("packet");
// SUBACK, packet identifier 7, and nothing else. MQTT-3.9.3 requires at least one
// reason code, so the shortest legal remaining length is 3.
QTest::newRow("SUBACK 3.1.1, no reason code")
<< QMqttClient::MQTT_3_1_1 << QByteArray::fromHex("90020007");
// The same, plus an empty property block. MQTT-3.9.2.1 makes the property length
// mandatory in MQTT 5.0, so the shortest legal remaining length is 4.
QTest::newRow("SUBACK 5.0, no reason code")
<< QMqttClient::MQTT_5_0 << QByteArray::fromHex("9003000700");
// UNSUBACK, packet identifier 7, empty property block, no reason code. The
// property length is mandatory (MQTT-3.11.2.1) and at least one reason code is
// required (MQTT-3.11.3), so the shortest legal remaining length is 4.
QTest::newRow("UNSUBACK 5.0, no reason code")
<< QMqttClient::MQTT_5_0 << QByteArray::fromHex("b003000700");
// PUBACK with half a packet identifier. The reason code and the property length may
// both be omitted (MQTT-3.4.2.1), but the identifier may not, so the shortest legal
// remaining length is 2.
QTest::newRow("PUBACK 5.0, truncated packet identifier")
<< QMqttClient::MQTT_5_0 << QByteArray::fromHex("400100");
// The same for PUBREL, MQTT-3.6.2.1.
QTest::newRow("PUBREL 5.0, truncated packet identifier")
<< QMqttClient::MQTT_5_0 << QByteArray::fromHex("620100");
}
void Tst_QMqttClient::shortRemainingLength()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
QFETCH(QByteArray, packet);
FakeServer server;
server.version = mqttVersion;
server.connectionSuccess = true;
VersionClient(mqttVersion, client);
if (!client.isTcp())
QSKIP("This test is only for tcp");
client.setHostname(QLatin1String("localhost"));
client.setPort(5726);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
server.send(packet);
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
QCOMPARE(client.error(), QMqttClient::ProtocolViolation);
}
DefaultVersionTestData(Tst_QMqttClient::staticProperties_QTBUG_67176_data)
void Tst_QMqttClient::staticProperties_QTBUG_67176()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
VersionClient(mqttVersion, client);
client.setHostname(m_testBroker);
client.setPort(m_port);
auto clientVersion = client.protocolVersion();
const QString hostname = client.hostname();
const int port = client.port();
const QString clientId = client.clientId();
const quint16 keepAlive = client.keepAlive();
const bool clean = client.cleanSession();
client.connectToHost();
QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
client.setClientId(QLatin1String("someclient"));
QCOMPARE(client.clientId(), clientId);
client.setHostname(QLatin1String("some.domain.foo"));
QCOMPARE(client.hostname(), hostname);
client.setPort(1234);
QCOMPARE(client.port(), port);
client.setKeepAlive(keepAlive + 10);
QCOMPARE(client.keepAlive(), keepAlive);
client.setProtocolVersion(QMqttClient::MQTT_3_1);
QCOMPARE(client.protocolVersion(), clientVersion);
client.setUsername(QLatin1String("someUser"));
QCOMPARE(client.username(), QLatin1String());
client.setPassword(QLatin1String("somePassword"));
QCOMPARE(client.password(), QLatin1String());
client.setCleanSession(!clean);
QCOMPARE(client.cleanSession(), clean);
}
void Tst_QMqttClient::authentication()
{
VersionClient(QMqttClient::MQTT_5_0, client);
client.setHostname(m_testBroker);
client.setPort(m_port);
QMqttConnectionProperties connectionProperties;
connectionProperties.setAuthenticationMethod(QLatin1String("SCRAM-SHA-1"));
client.setConnectionProperties(connectionProperties);
connect(&client, &QMqttClient::authenticationRequested, &client,
[](const QMqttAuthenticationProperties &prop)
{
qDebug() << "Authentication requested:" << prop.authenticationMethod();
});
connect(&client, &QMqttClient::authenticationFinished, &client,
[](const QMqttAuthenticationProperties &prop)
{
qDebug() << "Authentication finished:" << prop.authenticationMethod();
});
// ### FIXME : There is no public test broker yet able to handle authentication methods
// Theoretically the broker should send an AUTH request, followed by AUTH call including
// authentication data. See 4.12 of MQTT v5 specs.
QSKIP("No broker available with enhanced authentication.");
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
}
Q_DECLARE_METATYPE(QMqtt::MessageStatus)
void Tst_QMqttClient::messageStatus_data()
{
QTest::addColumn<int>("qos");
QTest::addColumn<QList<QMqtt::MessageStatus>>("expectedStatus");
QTest::newRow("QoS1") << 1 << (QList<QMqtt::MessageStatus>() << QMqtt::MessageStatus::Acknowledged);
QTest::newRow("QoS2") << 2 << (QList<QMqtt::MessageStatus>() << QMqtt::MessageStatus::Received
<< QMqtt::MessageStatus::Completed);
}
void Tst_QMqttClient::messageStatus()
{
QFETCH(int, qos);
QFETCH(QList<QMqtt::MessageStatus>, expectedStatus);
VersionClient(QMqttClient::MQTT_5_0, client);
client.setHostname(m_testBroker);
client.setPort(m_port);
client.connectToHost();
QTRY_VERIFY2(client.state() == QMqttClient::Connected, "Could not connect to broker.");
const QString topic = QLatin1String("Qt/client/statusCheck");
connect(&client, &QMqttClient::messageStatusChanged,
&client, [&expectedStatus](qint32,
QMqtt::MessageStatus s,
const QMqttMessageStatusProperties &)
{
QCOMPARE(s, expectedStatus.first());
expectedStatus.takeFirst();
});
QSignalSpy publishSpy(&client, &QMqttClient::messageSent);
client.publish(topic, QByteArray("someContent"), quint8(qos));
QTRY_VERIFY2(publishSpy.size() == 1, "Could not publish message");
QTRY_VERIFY2(expectedStatus.isEmpty(), "Did not receive all status updates.");
}
void Tst_QMqttClient::messageStatusReceive_data()
{
QTest::addColumn<int>("qos");
QTest::addColumn<QList<QMqtt::MessageStatus>>("expectedStatus");
QTest::newRow("QoS1") << 1 << (QList<QMqtt::MessageStatus>() << QMqtt::MessageStatus::Published);
QTest::newRow("QoS2") << 2 << (QList<QMqtt::MessageStatus>() << QMqtt::MessageStatus::Published
<< QMqtt::MessageStatus::Released);
}
void Tst_QMqttClient::messageStatusReceive()
{
QFETCH(int, qos);
QFETCH(QList<QMqtt::MessageStatus>, expectedStatus);
VersionClient(QMqttClient::MQTT_5_0, publisher);
publisher.setHostname(m_testBroker);
publisher.setPort(m_port);
publisher.connectToHost();
QTRY_VERIFY2(publisher.state() == QMqttClient::Connected, "Could not connect to broker.");
VersionClient(QMqttClient::MQTT_5_0, subscriber);
subscriber.setHostname(m_testBroker);
subscriber.setPort(m_port);
subscriber.connectToHost();
QTRY_VERIFY2(subscriber.state() == QMqttClient::Connected, "Could not connect to broker.");
const QString topic = QLatin1String("Qt/client/statusCheckReceive");
auto subscription = subscriber.subscribe(topic, quint8(qos));
QTRY_VERIFY2(subscription->state() == QMqttSubscription::Subscribed, "Could not subscribe to topic");
QVERIFY(subscription->qos() >= qos);
connect(&subscriber, &QMqttClient::messageStatusChanged, &subscriber,
[&expectedStatus](qint32, QMqtt::MessageStatus s, const QMqttMessageStatusProperties &)
{
QCOMPARE(s, expectedStatus.first());
expectedStatus.takeFirst();
});
QSignalSpy publishSpy(&publisher, &QMqttClient::messageSent);
QSignalSpy receiveSpy(&subscriber, &QMqttClient::messageReceived);
publisher.publish(topic, QByteArray("someContent"), quint8(qos));
QTRY_VERIFY2(publishSpy.size() == 1, "Could not publish message");
QTRY_VERIFY2(receiveSpy.size() == 1, "Did not receive message");
QTRY_VERIFY2(expectedStatus.isEmpty(), "Did not receive all status updates.");
}
void Tst_QMqttClient::subscriptionIdsOverlap()
{
// If the Server sends a single copy of the message it MUST include in the
// PUBLISH packet the Subscription Identifiers for all matching
// subscriptions which have a Subscription Identifiers, their order is not
// significant [MQTT-3.3.4-4].
// If the Server sends multiple PUBLISH packets it MUST send, in each of
// them, the Subscription Identifier of the matching subscription if it has
// a Subscription Identifier [MQTT-3.3.4-5].
const QString topic = QLatin1String("Qt/client/idcheck");
// Connect publisher
VersionClient(QMqttClient::MQTT_5_0, pub);
pub.setHostname(m_testBroker);
pub.setPort(m_port);
pub.connectToHost();
QTRY_VERIFY2(pub.state() == QMqttClient::Connected, "Could not connect publisher.");
// Connect subA
VersionClient(QMqttClient::MQTT_5_0, subClientA);
subClientA.setHostname(m_testBroker);
subClientA.setPort(m_port);
subClientA.connectToHost();
QTRY_VERIFY2(subClientA.state() == QMqttClient::Connected, "Could not connect subscriber A.");
QMqttSubscriptionProperties subAProp;
subAProp.setSubscriptionIdentifier(8);
auto subA = subClientA.subscribe(topic, subAProp, 1);
QTRY_VERIFY2(subA->state() == QMqttSubscription::Subscribed, "Could not subscibe A.");
int receiveACounter = 0;
connect(subA, &QMqttSubscription::messageReceived, subA,
[&receiveACounter](QMqttMessage msg) {
qDebug() << "Sub A received:" << msg.publishProperties().subscriptionIdentifiers();
// ### TODO: Wait for fix at https://github.com/eclipse/paho.mqtt.testing/issues/56
//QVERIFY(msg.publishProperties().subscriptionIdentifiers().size() == 1);
//QVERIFY(msg.publishProperties().subscriptionIdentifiers().at(0) == 8); // Use sub->id();
receiveACounter++;
});
// Connect subB
VersionClient(QMqttClient::MQTT_5_0, subClientB);
subClientB.setHostname(m_testBroker);
subClientB.setPort(m_port);
subClientB.connectToHost();
QTRY_VERIFY2(subClientB.state() == QMqttClient::Connected, "Could not connect subscriber A.");
QMqttSubscriptionProperties subBProp;
subBProp.setSubscriptionIdentifier(9);
auto subB = subClientB.subscribe(topic, subBProp, 1);
QTRY_VERIFY2(subB->state() == QMqttSubscription::Subscribed, "Could not subscibe A.");
int receiveBCounter = 2;
connect(subB, &QMqttSubscription::messageReceived,
subB, [&receiveBCounter](QMqttMessage msg) {
qDebug() << "Sub B received:" << msg.publishProperties().subscriptionIdentifiers();
QVERIFY(msg.publishProperties().subscriptionIdentifiers().size() > 0);
receiveBCounter -= msg.publishProperties().subscriptionIdentifiers().size();
});
QMqttSubscriptionProperties subB2Prop;
subB2Prop.setSubscriptionIdentifier(14);
auto subB2 = subClientB.subscribe(topic + "/#", subB2Prop, 1);
QTRY_VERIFY2(subB2->state() == QMqttSubscription::Subscribed, "Could not subscibe A.");
int receiveB2Counter = 2;
connect(subB2, &QMqttSubscription::messageReceived,
subB2, [&receiveB2Counter](QMqttMessage msg) {
qDebug() << "Sub B2 received:" << msg.publishProperties().subscriptionIdentifiers();
QVERIFY(msg.publishProperties().subscriptionIdentifiers().size() > 0);
receiveB2Counter -= msg.publishProperties().subscriptionIdentifiers().size();
});
QSignalSpy publishSpy(&pub, &QMqttClient::messageSent);
pub.publish(topic, "SomeData", 1);
QTRY_VERIFY2(publishSpy.size() == 1, "Could not finalize publication.");
QTRY_VERIFY2(receiveBCounter == 0, "Did not receive both messages.");
QTRY_VERIFY2(receiveB2Counter == 0, "Did not receive both messages.");
QTRY_VERIFY2(receiveACounter == 1, "Did not receive non-overlapping message.");
}
DefaultVersionTestData(Tst_QMqttClient::keepAlive_data)
void Tst_QMqttClient::keepAlive()
{
QFETCH(QMqttClient::ProtocolVersion, mqttVersion);
const quint16 keepAlive = 1;
VersionClient(mqttVersion, client);
client.setHostname(m_testBroker);
client.setPort(m_port);
client.setKeepAlive(keepAlive);
QCOMPARE(client.keepAlive(), keepAlive);
QCOMPARE(client.autoKeepAlive(), true); // default
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
if (client.protocolVersion() == QMqttClient::MQTT_5_0) {
const auto serverProps = client.serverConnectionProperties();
if (serverProps.isValid() && serverProps.availableProperties() & QMqttServerConnectionProperties::ServerKeepAlive) {
if (serverProps.serverKeepAlive() == 0 || serverProps.serverKeepAlive() > keepAlive) {
qDebug() << "Server specifies keepAlive which cannot be used for this test";
return;
}
}
}
// Changing autoKeepAlive is not possible while connected
client.setAutoKeepAlive(false);
QCOMPARE(client.autoKeepAlive(), true);
// Sending a manual ping request with autoKeepAlive is not allowed
// and will be suppressed.
const bool sent = client.requestPing();
QCOMPARE(sent, false);
// Verify auto keep alive works
QSignalSpy spy(&client, &QMqttClient::pingResponseReceived);
QTRY_VERIFY_WITH_TIMEOUT(spy.size() == 5, keepAlive * 1000 * 5 * 2);
spy.clear();
client.disconnectFromHost();
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
// Set manual keepAlive
client.setAutoKeepAlive(false);
QCOMPARE(client.autoKeepAlive(), false);
client.setKeepAlive(keepAlive);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
// Check for timeout / disconnect when not ping sent
QTRY_COMPARE_WITH_TIMEOUT(client.state(), QMqttClient::Disconnected, keepAlive * 1000 * 5 * 2);
// Check manual ping
QTimer t;
t.setInterval(keepAlive * 1000);
t.setSingleShot(false);
t.connect(&t, &QTimer::timeout, &client, [&client]() {
client.requestPing();
});
connect(&client, &QMqttClient::connected, &t, [&t]() {
t.start();
});
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
QTRY_VERIFY_WITH_TIMEOUT(spy.size() == 5, keepAlive * 1000 * 5 * 2);
client.disconnectFromHost();
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
}
void Tst_QMqttClient::reconnect()
{
VersionClient(QMqttClient::ProtocolVersion(0), client);
client.setHostname(m_testBroker);
client.setPort(m_port);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
client.disconnectFromHost();
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
/* The rest requires a mosquitto installation that runs
on multiple ports.
*/
#if defined(MQTT_TEST_BROKER)
client.setPort(client.port() + 1000);
client.connectToHost();
QTRY_COMPARE(client.state(), QMqttClient::Connected);
client.disconnectFromHost();
QTRY_COMPARE(client.state(), QMqttClient::Disconnected);
QMqttClient client2;
client2.setHostname(client.hostname());
for (int i = 0; i < 2; ++i) {
#if defined(QT_MQTT_WITH_WEBSOCKETS)
client2.setPort(9080);
client2.connectToHostWebSocket();
QTRY_COMPARE(client2.state(), QMqttClient::Connected);
client2.disconnectFromHost();
QTRY_COMPARE(client2.state(), QMqttClient::Disconnected);
#endif
#if defined(QT_MQTT_WITH_WEBSOCKETS) && !defined(QT_NO_SSL)
client2.setPort(9081);
QUrl url(QString(u"wss://") + client2.hostname() + QString(u":") + QString::number(client2.port()));
QWebSocketHandshakeOptions options;
options.setSubprotocols(QStringList{ QString::fromUtf8("mqttv3.1") });
QWebSocket socket;
socket.setSslConfiguration(makeConfig(QString::fromUtf8(getenv("MQTT_TEST_BROKER_CERTIFICATE"))));
socket.open(url, options);
client2.connectToHostWebSocketEncrypted(&socket);
QTRY_COMPARE(client2.state(), QMqttClient::Connected);
client2.disconnectFromHost();
QTRY_COMPARE(client2.state(), QMqttClient::Disconnected);
#endif
#if !defined(QT_NO_SSL)
client2.setPort(9883);
client2.connectToHostEncrypted(makeConfig(QString::fromUtf8(getenv("MQTT_TEST_BROKER_CERTIFICATE"))));
QTRY_COMPARE(client2.state(), QMqttClient::Connected);
client2.disconnectFromHost();
QTRY_COMPARE(client2.state(), QMqttClient::Disconnected);
#endif
client2.setPort(2883);
client2.connectToHost();
QTRY_COMPARE(client2.state(), QMqttClient::Connected);
client2.disconnectFromHost();
QTRY_COMPARE(client2.state(), QMqttClient::Disconnected);
}
#endif
}
QTEST_MAIN(Tst_QMqttClient)
#include "tst_qmqttclient.moc"
Reference in New Issue View Git Blame Copy Permalink