mqtt: bounds number of messages per tx

Ticket: 8525

rules/mqtt-events.rules

@@ -13,3 +13,4 @@ alert mqtt any any -> any any (msg:"SURICATA MQTT missing message ID"; app-layer
 alert mqtt any any -> any any (msg:"SURICATA MQTT unassigned message type (0 or >15)"; app-layer-event:mqtt.unassigned_msg_type; classtype:protocol-command-decode; sid:2229008; rev:1;)
 alert mqtt any any -> any any (msg:"SURICATA MQTT too many transactions"; app-layer-event:mqtt.too_many_transactions; classtype:protocol-command-decode; sid:2229009; rev:1;)
 alert mqtt any any -> any any (msg:"SURICATA MQTT malformed traffic"; app-layer-event:mqtt.malformed_traffic; classtype:protocol-command-decode; sid:2229010; rev:1;)
+alert mqtt any any -> any any (msg:"SURICATA MQTT too many messages"; app-layer-event:mqtt.too_many_messages; classtype:protocol-command-decode; sid:2229011; rev:1;)

rust/src/mqtt/mqtt.rs

@@ -46,6 +46,8 @@ static mut MAX_MSG_LEN: u32 = 1048576;
 
 static mut MQTT_MAX_TX: usize = 1024;
 
+static mut MQTT_MAX_MSGS: usize = 1024;
+
 pub(super) static mut ALPROTO_MQTT: AppProto = ALPROTO_UNKNOWN;
 
 #[derive(AppLayerFrameType)]
@@ -68,6 +70,7 @@ pub enum MQTTEvent {
     UnassignedMsgType,
     TooManyTransactions,
     MalformedTraffic,
+    TooManyMessages,
 }
 
 #[derive(Debug)]
@@ -318,7 +321,11 @@ impl MQTTState {
             }
             MQTTOperation::PUBREC(ref v) | MQTTOperation::PUBREL(ref v) => {
                 if let Some(tx) = self.get_tx_by_pkt_id(v.message_id as u32) {
-                    tx.msg.push(msg);
+                    if tx.msg.len() >= unsafe { MQTT_MAX_MSGS } {
+                        tx.tx_data.set_event(MQTTEvent::TooManyMessages as u8);
+                    } else {
+                        tx.msg.push(msg);
+                    }
                 } else {
                     let mut tx = self.new_tx(msg, toclient);
                     MQTTState::set_event(&mut tx, MQTTEvent::MissingPublish);
@@ -801,6 +808,13 @@ pub unsafe extern "C" fn SCMqttRegisterParser() {
                 SCLogError!("Invalid value for mqtt.max-tx");
             }
         }
+        if let Some(val) = conf_get("app-layer.protocols.mqtt.max-messages") {
+            if let Ok(v) = val.parse::<usize>() {
+                MQTT_MAX_MSGS = v;
+            } else {
+                SCLogWarning!("Invalid value for mqtt.max-messages");
+            }
+        }
         if let Some(val) = conf_get("app-layer.protocols.mqtt.max-msg-length") {
             if let Ok(v) = get_memval(val) {
                 MAX_MSG_LEN = v as u32;

suricata.yaml.in

@@ -954,6 +954,8 @@ app-layer:
       # unsubscribe-topic-match-limit: 100
       # Maximum number of live MQTT transactions per flow
       # max-tx: 4096
+      # Maximum number of messages per transaction
+      # max-messages: 1024
     krb5:
       enabled: yes
     bittorrent-dht: