SteamFrame/FrameAssistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6e1bed79f07ea9dacb29a1b7edfc4cc9a07ab350
FrameAssistant/tests/tst_mqtt.cpp
T
matzmanandClaude Opus 5.5 148ae663cf TLS for the MQTT connection (iteration 2, step 5) 
ca_file or the system's CAs (MOSQ_OPT_TLS_USE_OS_CERTS), client
certificate, broker certificate and host name verified, settings errors
reported before connecting. The client thread blocks SIGPIPE: with TLS,
a write to a connection the broker closed killed the process. The
reason of a failed handshake (libmosquitto's log) is appended to the
error. tst_mqtt covers TLS with test certificates in tests/data (D44).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 16:29:32 +02:00

509 lines
18 KiB
C++
Raw Blame History

// MosquittoClient against a real broker (docs/DESIGN.md, D30 level 2): a
// local mosquitto on a free port. Connect, messages, retained messages,
// reconnect, last will, login, TLS (certificates in tests/data/, D44). Skips
// when mosquitto isn't installed.
//
// The binary is also its own helper process for the last-will test:
// `tst_mqtt --will-helper <port> <topic>` connects with a will and waits to be
// killed.
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <QCoreApplication>
#include <QFile>
#include <QProcess>
#include <QStandardPaths>
#include <QTemporaryDir>
#include <QTest>
#include <memory>
#include "mqtt/MosquittoClient.h"
namespace {
constexpr int kTimeoutMs = 10000;
QString data(const char* file) {
return QStringLiteral(TEST_DATA_DIR "/") + QLatin1String(file);
}
// Broker configuration for a TLS listener with the test CA's server
// certificate.
QString tlsListener(bool requireClientCertificate = false) {
return QStringLiteral(
"allow_anonymous true\ncafile %1\ncertfile %2\nkeyfile %3\n"
"require_certificate %4\n")
.arg(data("ca.crt"), data("server.crt"), data("server.key"),
requireClientCertificate ? QStringLiteral("true")
: QStringLiteral("false"));
}
MosquittoClient::Settings tlsSettings(int port, const char* caFile) {
MosquittoClient::Settings s;
s.broker.host = QStringLiteral("127.0.0.1");
s.broker.port = port;
s.broker.tls.enabled = true;
if (caFile) s.broker.tls.caFile = data(caFile);
s.clientId = QStringLiteral("tls");
s.reconnectMinMs = 100;
s.reconnectMaxMs = 500;
return s;
}
QString findTool(const QString& name) {
QString path = QStandardPaths::findExecutable(name);
if (path.isEmpty())
path = QStandardPaths::findExecutable(name, {QStringLiteral("/usr/sbin")});
return path;
}
// A TCP port nobody listens on right now.
int freePort() {
const int fd = socket(AF_INET, SOCK_STREAM, 0);
sockaddr_in address{};
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
socklen_t length = sizeof(address);
bind(fd, reinterpret_cast<sockaddr*>(&address), length);
getsockname(fd, reinterpret_cast<sockaddr*>(&address), &length);
close(fd);
return ntohs(address.sin_port);
}
bool portOpen(int port) {
const int fd = socket(AF_INET, SOCK_STREAM, 0);
sockaddr_in address{};
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
address.sin_port = htons(port);
const bool open = ::connect(fd, reinterpret_cast<sockaddr*>(&address),
sizeof(address)) == 0;
close(fd);
return open;
}
// A mosquitto process with its own configuration.
class Broker {
public:
explicit Broker(const QString& dir) : m_dir(dir), m_port(freePort()) {}
~Broker() { stop(); }
int port() const { return m_port; }
// `extra`: further configuration lines; without them anonymous clients may
// connect.
bool start(const QString& extra = QStringLiteral("allow_anonymous true\n")) {
const QString config = m_dir + QStringLiteral("/mosquitto.conf");
QFile file(config);
if (!file.open(QIODevice::WriteOnly)) return false;
file.write(QStringLiteral("listener %1 127.0.0.1\npersistence false\n%2")
.arg(m_port)
.arg(extra)
.toUtf8());
file.close();
m_process = std::make_unique<QProcess>();
m_process->setProcessChannelMode(QProcess::ForwardedErrorChannel);
m_process->start(findTool(QStringLiteral("mosquitto")),
{QStringLiteral("-c"), config});
if (!m_process->waitForStarted()) return false;
QDeadlineTimer deadline(5000);
while (!portOpen(m_port)) {
if (deadline.hasExpired() || m_process->state() != QProcess::Running)
return false;
QThread::msleep(20);
}
return true;
}
void stop() {
if (!m_process) return;
m_process->terminate();
if (!m_process->waitForFinished(5000)) m_process->kill();
m_process->waitForFinished();
m_process.reset();
}
private:
QString m_dir;
int m_port;
std::unique_ptr<QProcess> m_process;
};
MosquittoClient::Settings settings(int port, const QString& clientId) {
MosquittoClient::Settings s;
s.broker.host = QStringLiteral("127.0.0.1");
s.broker.port = port;
s.clientId = clientId;
s.reconnectMinMs = 100;
s.reconnectMaxMs = 500;
return s;
}
// Collects a client's signals in the main thread (queued connections).
class Recorder : public QObject {
public:
explicit Recorder(MqttClient* client) {
connect(client, &MqttClient::connectionChanged, this,
[this](MqttClient::State state, const QString& error) {
states.append(state);
if (!error.isEmpty()) errors.append(error);
});
connect(client, &MqttClient::messageReceived, this,
[this](const QString& topic, const QByteArray& payload) {
messages.append({topic, payload});
});
}
QList<MqttClient::State> states;
QStringList errors;
QList<QPair<QString, QByteArray>> messages;
};
// Connects and waits for it; then subscribes to `topic` if given.
bool connectClient(MosquittoClient& client, Recorder& recorder,
const QString& topic = {}) {
client.connectToBroker();
if (!QTest::qWaitFor(
[&] {
return recorder.states.contains(MqttClient::State::Connected);
},
kTimeoutMs))
return false;
if (topic.isEmpty()) return true;
if (!client.subscribe(topic)) return false;
// No SUBACK signal: give the broker a moment to register it.
QTest::qWait(200);
return true;
}
int runWillHelper(int argc, char** argv) {
QCoreApplication app(argc, argv);
MosquittoClient::Settings s =
settings(QByteArray(argv[2]).toInt(), QStringLiteral("will-helper"));
s.will = {QString::fromUtf8(argv[3]), "offline", true};
MosquittoClient client(s);
QObject::connect(&client, &MqttClient::connectionChanged, &app,
[](MqttClient::State state) {
if (state == MqttClient::State::Connected) {
fputs("connected\n", stdout);
fflush(stdout);
}
});
client.connectToBroker();
return app.exec();
}
} // namespace
class TestMqtt : public QObject {
Q_OBJECT
private slots:
void initTestCase() {
if (findTool(QStringLiteral("mosquitto")).isEmpty())
QSKIP("mosquitto (the broker) is not installed");
QVERIFY(m_dir.isValid());
}
void connectsAndExchangesMessages() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
Recorder ra(&a), rb(&b);
QVERIFY(connectClient(a, ra));
QCOMPARE(ra.states.first(), MqttClient::State::Connecting);
QCOMPARE(a.state(), MqttClient::State::Connected);
QVERIFY(connectClient(b, rb, QStringLiteral("test/#")));
QVERIFY(a.publish(QStringLiteral("test/x"), "hello", false));
QVERIFY(a.waitForPublished(kTimeoutMs));
QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
QCOMPARE(rb.messages.first().first, QStringLiteral("test/x"));
QCOMPARE(rb.messages.first().second, QByteArray("hello"));
QVERIFY(ra.errors.isEmpty());
}
void retainedMessageReachesLaterSubscriber() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
Recorder ra(&a);
QVERIFY(connectClient(a, ra));
QVERIFY(a.publish(QStringLiteral("test/retained"), "kept", true));
QVERIFY(a.waitForPublished(kTimeoutMs));
MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
Recorder rb(&b);
QVERIFY(connectClient(b, rb, QStringLiteral("test/retained")));
QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
QCOMPARE(rb.messages.first().second, QByteArray("kept"));
}
void publishFailsWhileDisconnected() {
MosquittoClient a(settings(freePort(), QStringLiteral("a")));
QVERIFY(!a.publish(QStringLiteral("test/x"), "lost", false));
QVERIFY(!a.subscribe(QStringLiteral("test/x")));
QVERIFY(a.waitForPublished(0));
}
// The broker isn't running yet: the error is reported, and the client
// connects once it runs.
void firstConnectRetriesUntilBrokerRuns() {
Broker broker(m_dir.path());
MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
Recorder ra(&a);
a.connectToBroker();
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QCOMPARE(a.state(), MqttClient::State::Disconnected);
QVERIFY2(ra.errors.first().contains(QStringLiteral("refused")),
qPrintable(ra.errors.first()));
QVERIFY(broker.start());
QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
kTimeoutMs);
}
void reconnectsAfterBrokerRestart() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
Recorder ra(&a);
QVERIFY(connectClient(a, ra));
broker.stop();
QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Disconnected,
kTimeoutMs);
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
ra.states.clear();
QVERIFY(broker.start());
QTRY_VERIFY_WITH_TIMEOUT(ra.states.contains(MqttClient::State::Connected),
kTimeoutMs);
}
void unresolvableHostIsReported() {
MosquittoClient::Settings s = settings(1883, QStringLiteral("a"));
s.broker.host = QStringLiteral("no-such-host.invalid");
MosquittoClient a(s);
Recorder ra(&a);
a.connectToBroker();
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QVERIFY2(ra.errors.first().startsWith(QStringLiteral("Cannot resolve")),
qPrintable(ra.errors.first()));
}
// CON-3: a killed process (no DISCONNECT) → the broker publishes the will.
void lastWillWhenKilled() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient watcher(settings(broker.port(), QStringLiteral("w")));
Recorder rw(&watcher);
QVERIFY(connectClient(watcher, rw, QStringLiteral("test/availability")));
QProcess helper;
helper.start(
QCoreApplication::applicationFilePath(),
{QStringLiteral("--will-helper"), QString::number(broker.port()),
QStringLiteral("test/availability")});
QVERIFY(helper.waitForStarted());
// QTRY_* evaluates its expression again at the end: collect the output.
QByteArray output;
QTRY_VERIFY_WITH_TIMEOUT(
(output += helper.readAllStandardOutput()).contains("connected"),
kTimeoutMs);
QVERIFY(rw.messages.isEmpty());
helper.kill();
helper.waitForFinished();
QTRY_COMPARE_WITH_TIMEOUT(rw.messages.size(), 1, kTimeoutMs);
QCOMPARE(rw.messages.first().second, QByteArray("offline"));
}
// A normal exit publishes the will itself (retained), then disconnects.
void disconnectPublishesWill() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a"));
s.will = {QStringLiteral("test/availability"), "offline", true};
MosquittoClient a(s);
Recorder ra(&a);
QVERIFY(connectClient(a, ra));
QVERIFY(a.disconnectFromBroker(kTimeoutMs));
QCOMPARE(a.state(), MqttClient::State::Disconnected);
QTRY_COMPARE_WITH_TIMEOUT(ra.states.last(), MqttClient::State::Disconnected,
kTimeoutMs);
MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
Recorder rb(&b);
QVERIFY(connectClient(b, rb, QStringLiteral("test/availability")));
QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
QCOMPARE(rb.messages.first().second, QByteArray("offline"));
}
void login_data() {
QTest::addColumn<QString>("password");
QTest::addColumn<bool>("accepted");
QTest::newRow("right") << QStringLiteral("secret") << true;
QTest::newRow("wrong") << QStringLiteral("wrong") << false;
}
void login() {
QFETCH(QString, password);
QFETCH(bool, accepted);
const QString passwd = findTool(QStringLiteral("mosquitto_passwd"));
if (passwd.isEmpty()) QSKIP("mosquitto_passwd is not installed");
const QString file = m_dir.path() + QStringLiteral("/passwd");
QFile::remove(file);
QCOMPARE(QProcess::execute(
passwd, {QStringLiteral("-c"), QStringLiteral("-b"), file,
QStringLiteral("frame"), QStringLiteral("secret")}),
0);
Broker broker(m_dir.path());
QVERIFY(broker.start(
QStringLiteral("allow_anonymous false\npassword_file %1\n").arg(file)));
MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a"));
s.broker.username = QStringLiteral("frame");
s.broker.password = password;
MosquittoClient a(s);
Recorder ra(&a);
a.connectToBroker();
if (accepted) {
QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
kTimeoutMs);
} else {
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QVERIFY2(ra.errors.first().contains(QStringLiteral("not authori")),
qPrintable(ra.errors.first()));
QVERIFY(!ra.states.contains(MqttClient::State::Connected));
}
}
void tlsConnects() {
Broker broker(m_dir.path());
QVERIFY(broker.start(tlsListener()));
MosquittoClient a(tlsSettings(broker.port(), "ca.crt"));
Recorder ra(&a);
QVERIFY(connectClient(a, ra, QStringLiteral("test/#")));
QVERIFY(a.publish(QStringLiteral("test/x"), "secret", false));
QTRY_COMPARE_WITH_TIMEOUT(ra.messages.size(), 1, kTimeoutMs);
QVERIFY(ra.errors.isEmpty());
}
// The broker's certificate isn't from the given CA: no connection, and
// the reason says why.
void tlsWrongCaRejected() {
Broker broker(m_dir.path());
QVERIFY(broker.start(tlsListener()));
MosquittoClient a(tlsSettings(broker.port(), "other-ca.crt"));
Recorder ra(&a);
a.connectToBroker();
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QVERIFY2(ra.errors.first().contains(QStringLiteral("verify failed")),
qPrintable(ra.errors.first()));
QTest::qWait(300);
QVERIFY(!ra.states.contains(MqttClient::State::Connected));
}
// Without ca_file the system's CAs are used (they don't know the test
// CA). Fails the same way, which shows that TLS is on.
void tlsWithSystemCertificates() {
Broker broker(m_dir.path());
QVERIFY(broker.start(tlsListener()));
MosquittoClient a(tlsSettings(broker.port(), nullptr));
Recorder ra(&a);
a.connectToBroker();
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QVERIFY2(ra.errors.first().contains(QStringLiteral("verify failed")),
qPrintable(ra.errors.first()));
}
// A TLS error ends libmosquitto's own loop for good (D45); ours retries.
// First a plain listener on the port (handshake fails), then TLS.
void retriesAfterTlsError() {
Broker broker(m_dir.path());
QVERIFY(broker.start());
MosquittoClient a(tlsSettings(broker.port(), "ca.crt"));
Recorder ra(&a);
a.connectToBroker();
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
broker.stop();
QVERIFY(broker.start(tlsListener()));
QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
kTimeoutMs);
}
void tlsClientCertificate_data() {
QTest::addColumn<bool>("withCertificate");
QTest::newRow("with") << true;
QTest::newRow("without") << false;
}
void tlsClientCertificate() {
QFETCH(bool, withCertificate);
Broker broker(m_dir.path());
QVERIFY(broker.start(tlsListener(true)));
MosquittoClient::Settings s = tlsSettings(broker.port(), "ca.crt");
if (withCertificate) {
s.broker.tls.certFile = data("client.crt");
s.broker.tls.keyFile = data("client.key");
}
MosquittoClient a(s);
Recorder ra(&a);
a.connectToBroker();
if (withCertificate) {
QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
kTimeoutMs);
} else {
QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
QVERIFY(!ra.states.contains(MqttClient::State::Connected));
}
}
// Settings errors show up at once, without a connection attempt.
void tlsSettingsErrors_data() {
QTest::addColumn<QString>("caFile");
QTest::addColumn<QString>("certFile");
QTest::addColumn<QString>("keyFile");
QTest::addColumn<QString>("error");
QTest::newRow("missing CA file")
<< QStringLiteral("/no/such/ca.crt") << QString() << QString()
<< QStringLiteral("TLS: cannot read /no/such/ca.crt");
QTest::newRow("certificate without key")
<< data("ca.crt") << data("client.crt") << QString()
<< QStringLiteral("TLS: cert_file and key_file go together");
QTest::newRow("client certificate without CA file")
<< QString() << data("client.crt") << data("client.key")
<< QStringLiteral("TLS: a client certificate needs ca_file");
}
void tlsSettingsErrors() {
QFETCH(QString, caFile);
QFETCH(QString, certFile);
QFETCH(QString, keyFile);
QFETCH(QString, error);
MosquittoClient::Settings s = tlsSettings(freePort(), nullptr);
s.broker.tls.caFile = caFile;
s.broker.tls.certFile = certFile;
s.broker.tls.keyFile = keyFile;
MosquittoClient a(s);
Recorder ra(&a);
a.connectToBroker();
QTRY_COMPARE_WITH_TIMEOUT(ra.errors.size(), 1, kTimeoutMs);
QCOMPARE(ra.errors.first(), error);
QCOMPARE(a.state(), MqttClient::State::Disconnected);
}
private:
QTemporaryDir m_dir;
};
int main(int argc, char** argv) {
if (argc == 4 && qstrcmp(argv[1], "--will-helper") == 0)
return runWillHelper(argc, argv);
QCoreApplication app(argc, argv);
TestMqtt test;
return QTest::qExec(&test, argc, argv);
}
#include "tst_mqtt.moc"
Reference in New Issue View Git Blame Copy Permalink