Add the readBufferSize property to limit the read buffer for clients

- this function should be used to protect your socket against receiving
too much data, which may eventually cause your application to run out
of memory.
- setSocketReadBufferSize(size) sets the size of internal read buffer
to be 'size' bytes.

[ChangeLog][QMqttClient] Added a readBufferSize property to limit
the transport buffer size.

Task-number: QTBUG-149866
Change-Id: I6ff843b66b50d019ecf620c25c47e0d779518113
Reviewed-by: Ivan Solovev <ivan.solovev@qt.io>

src/mqtt/qmqttclient.cpp

@@ -157,6 +157,21 @@ Q_LOGGING_CATEGORY(lcMqttClient, "qt.mqtt.client")
     \sa keepAlive(), requestPing(), serverConnectionProperties(), pingResponseReceived()
 */
 
+/*!
+    \property QMqttClient::readBufferSize
+    \since 6.13
+    \brief This property holds the size of the internal read buffer.
+
+    The property limits the amount of data that the client can receive
+    before you call read() or readAll().
+
+    A read buffer size of 0 (the default) means that the buffer has no size
+    limit, ensuring that no data is lost.
+
+    \note The limit is not applicable for a plain \c TransportType::IODevice
+    transport type.
+*/
+
 /*!
     \enum QMqttClient::TransportType
 
@@ -324,6 +339,15 @@ Q_LOGGING_CATEGORY(lcMqttClient, "qt.mqtt.client")
     \sa authenticationRequested(), authenticate()
 */
 
+/*!
+    \since 6.13
+    \fn QMqttClient::readBufferSizeChanged(qint64 size)
+
+    This signal is emitted when the \a size of internal read buffer is changed.
+
+    \sa readBufferSize()
+*/
+
 /*!
     Creates a new MQTT client instance with the specified \a parent.
  */
@@ -368,6 +392,8 @@ void QMqttClient::setTransport(QIODevice *device, QMqttClient::TransportType tra
         return;
     }
     d->m_connection.setTransport(device, transport);
+    // Set the currently active buffer size for a new transport.
+    d->m_connection.setReadBufferSize(d->m_readBufferSize);
 }
 
 /*!
@@ -379,6 +405,74 @@ QIODevice *QMqttClient::transport() const
     return d->m_connection.transport();
 }
 
+/*!
+    \since 6.13
+
+    Limits the read buffer of the transport used by this client to
+    \a size bytes. Triggers warning message if the limit could not
+    be applied.
+
+    The transport does not buffer more than \a size bytes: it stops
+    reading from the network until the client has taken the data out of
+    the buffer. Use this function to protect the application against a
+    broker that sends more data than the application can handle, which
+    could make it run out of memory. A size of \c 0 means that the buffer
+    has no size limit, which is the default of the underlying transport.
+
+    \code
+    QMqttClient client;
+    client.setHostname(QStringLiteral("broker.example.com"));
+    client.setPort(1883);
+    client.setReadBufferSize(64 * 1024);
+    client.connectToHost();
+    \endcode
+
+    The size can be set before connecting. It is applied to the transport
+    when connectToHost() creates it, or when setTransport() sets it. The size
+    has no effect if the transport is a plain \l QMqttClient::IODevice.
+
+    Before Qt 6.13, the \c QMqttClient class provided no way to limit the
+    amount of incoming data. However, applications using older versions of
+    \c QMqttClient with certain types of transport, such as transports derived
+    from \c AbstractSocket, can protect themselves against memory exhaustion
+    caused by a broker, as shown in the example below:
+    \code
+    QMqttClient client;
+    client.setHostname(QStringLiteral("broker.example.com"));
+    client.setPort(1883);
+    QObject::connect(&client, &QMqttClient::connected, &client, [&client]() {
+        QAbstractSocket *socket = dynamic_cast<QAbstractSocket *>(client.transport());
+        if (socket)
+            socket->setReadBufferSize(64 * 1024);
+    });
+    client.connectToHost();
+    \endcode
+
+    \note If you use a custom QIODevice-based transport, implement
+    support for a read buffer limit in the device and configure it
+    before calling setTransport(). Some devices already support it.
+    For example, \l QSerialPort provides QSerialPort::setReadBufferSize().
+
+    \note This limit and the MQTT 5 maximum packet size protect against
+    different things. The maximum packet size limits the size of an
+    individual packet, not the total size of a buffer.
+
+    \sa setTransport(), QMqttConnectionProperties::setMaximumPacketSize()
+ */
+void QMqttClient::setReadBufferSize(qint64 size)
+{
+    Q_D(QMqttClient);
+
+    if (d->m_readBufferSize == size)
+        return;
+
+    // Store the size even if there is no transport yet. setTransport() and
+    // connectToHost() apply it to the transport they set or create.
+    d->m_readBufferSize = size;
+    d->m_connection.setReadBufferSize(size);
+    emit readBufferSizeChanged(d->m_readBufferSize);
+}
+
 /*!
     Adds a new subscription to receive notifications on \a topic. The parameter
     \a qos specifies the level at which security messages are received. For more
@@ -630,6 +724,11 @@ void QMqttClient::connectToHost(bool encrypted, const QString &sslPeerName)
         return;
     }
 
+    // We can safely set the active buffer size for the new transport here,
+    // as the transport type has been determined by ensureTransport() or
+    // connectToHostWebSocket()/connectToHostWebSocketEncrypted() at this point.
+    d->m_connection.setReadBufferSize(d->m_readBufferSize);
+
     // Once transport has connected, it will invoke
     // QMqttConnection::sendControlConnect to
     // handshake with the broker
@@ -715,6 +814,12 @@ bool QMqttClient::autoKeepAlive() const
     return d->m_autoKeepAlive;
 }
 
+qint64 QMqttClient::readBufferSize() const
+{
+    Q_D(const QMqttClient);
+    return d->m_readBufferSize;
+}
+
 /*!
     \since 5.12
 

src/mqtt/qmqttclient.h

@@ -84,12 +84,15 @@ private:
     Q_PROPERTY(quint8 willQoS READ willQoS WRITE setWillQoS NOTIFY willQoSChanged)
     Q_PROPERTY(bool willRetain READ willRetain WRITE setWillRetain NOTIFY willRetainChanged)
     Q_PROPERTY(bool autoKeepAlive READ autoKeepAlive WRITE setAutoKeepAlive NOTIFY autoKeepAliveChanged)
+    Q_PROPERTY(qint64 readBufferSize READ readBufferSize WRITE setReadBufferSize NOTIFY readBufferSizeChanged)
+
 public:
     explicit QMqttClient(QObject *parent = nullptr);
     ~QMqttClient() override;
 
     void setTransport(QIODevice *device, TransportType transport);
     QIODevice *transport() const;
+    void setReadBufferSize(qint64 size);
 
     QMqttSubscription *subscribe(const QMqttTopicFilter &topic, quint8 qos = 0);
     QMqttSubscription *subscribe(const QMqttTopicFilter &topic,
@@ -132,6 +135,7 @@ public:
     QByteArray willMessage() const;
     bool willRetain() const;
     bool autoKeepAlive() const;
+    qint64 readBufferSize() const;
 
     void setConnectionProperties(const QMqttConnectionProperties &prop);
     QMqttConnectionProperties connectionProperties() const;
@@ -167,6 +171,7 @@ Q_SIGNALS:
     void willMessageChanged(QByteArray willMessage);
     void willRetainChanged(bool willRetain);
     void autoKeepAliveChanged(bool autoKeepAlive);
+    void readBufferSizeChanged(qint64 size);
 
     void authenticationRequested(const QMqttAuthenticationProperties &p);
     void authenticationFinished(const QMqttAuthenticationProperties &p);

src/mqtt/qmqttclient_p.h

@@ -48,6 +48,7 @@ public:
     quint8 m_willQoS{0};
     bool m_willRetain{false};
     bool m_autoKeepAlive{true};
+    qint64 m_readBufferSize{0};
     QString m_username;
     QString m_password;
     bool m_cleanSession{true};

src/mqtt/qmqttconnection.cpp

@@ -420,6 +420,33 @@ bool QMqttConnection::sendControlConnect()
     return true;
 }
 
+void QMqttConnection::setReadBufferSize(qint64 size)
+{
+    // Nothing to apply yet. QMqttClient applies the size again once a transport is
+    // set or created.
+    if (!m_transport)
+        return;
+    switch (m_transportType) {
+    case QMqttClient::TransportType::IODevice:
+        if (size != 0)
+            qCWarning(lcMqttClient) << "TransportType::IODevice doesn't support a read buffer limit.";
+        break;
+    case QMqttClient::TransportType::AbstractSocket:
+    case QMqttClient::TransportType::SecureSocket:
+        // QSslSocket is a QAbstractSocket and setReadBufferSize() is declared on the
+        // base class, so this also works in a build without SSL support.
+        setSocketBufferSize<QAbstractSocket>(size);
+        break;
+    case QMqttClient::TransportType::WebSocket:
+    case QMqttClient::TransportType::SecureWebSocket:
+#ifdef QT_MQTT_WITH_WEBSOCKETS
+        // Both web socket transports share the base class holding the QWebSocket.
+        setSocketBufferSize<QMqttWebSocketIOBase>(size);
+#endif
+        break;
+    }
+}
+
 bool QMqttConnection::sendControlAuthenticate(const QMqttAuthenticationProperties &properties)
 {
     qCDebug(lcMqttConnection) << Q_FUNC_INFO;

src/mqtt/qmqttconnection_p.h

@@ -69,7 +69,7 @@ public:
     bool sendControlUnsubscribe(const QMqttTopicFilter &topic, const QMqttUnsubscriptionProperties &properties);
     bool sendControlPingRequest(bool isAuto = true);
     bool sendControlDisconnect(std::optional<QMqtt::ReasonCode> reasonCode = std::nullopt);
-
+    void setReadBufferSize(qint64 size);
     void setClientPrivate(QMqttClientPrivate *clientPrivate);
 
     inline quint16 unusedPacketIdentifier() const;
@@ -125,6 +125,12 @@ private:
     template<typename T> T readBufferTyped(qint64 *dataSize = nullptr);
     bool isPendingUnsubscribe(QMqttSubscription *) const;
     QMqttSubscription *findActiveSubscription(const QMqttTopicFilter &topic) const;
+    template <typename T>
+    void setSocketBufferSize(qint64 size) {
+        auto *socket = qobject_cast<T *>(m_transport.get());
+        Q_ASSERT(socket);
+        socket->setReadBufferSize(size);
+    }
 
     QByteArray m_readBuffer;
     int m_readPosition{0};

src/mqtt/transportLayers/mqtt_websocket_io_base_p.h

@@ -77,6 +77,11 @@ public:
     {
         return m_externalSocket;
     }
+
+    void setReadBufferSize(qint64 size)
+    {
+        m_socket->setReadBufferSize(size);
+    }
 Q_SIGNALS:
     void disconnected();
     void errorOccurred(QAbstractSocket::SocketError error);