mqtt: limits the number of active transactions per flow

Ticket: 4530

So, that we do not get DOS by quadratic complexity, while
looking for a new pkt_id over the ever growing list
of active transactions

(cherry picked from commit a8079dc978)

rules/mqtt-events.rules

@@ -13,4 +13,5 @@ alert mqtt any any -> any any (msg:"SURICATA MQTT message seen before CONNECT/CO
 alert mqtt any any -> any any (msg:"SURICATA MQTT invalid QOS level"; app-layer-event:mqtt.invalid_qos_level; classtype:protocol-command-decode; sid:2229006; rev:1;)
 alert mqtt any any -> any any (msg:"SURICATA MQTT missing message ID"; app-layer-event:mqtt.missing_msg_id; classtype:protocol-command-decode; sid:2229007; rev:1;)
 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;)

rust/src/mqtt/mqtt.rs

@@ -37,6 +37,9 @@ const MQTT_CONNECT_PKT_ID: u32 = std::u32::MAX;
 // this value, it will be truncated. Default: 1MB.
 static mut MAX_MSG_LEN: u32 = 1048576;
 
+//TODO make this configurable
+const MQTT_MAX_TX: usize = 1024;
+
 static mut ALPROTO_MQTT: AppProto = ALPROTO_UNKNOWN;
 
 #[derive(FromPrimitive, Debug)]
@@ -51,6 +54,7 @@ pub enum MQTTEvent {
     InvalidQosLevel,
     MissingMsgId,
     UnassignedMsgtype,
+    TooManyTransactions,
     MalformedTraffic,
 }
 
@@ -178,6 +182,15 @@ impl MQTTState {
         } else {
             tx.toserver = true;
         }
+        if self.transactions.len() > MQTT_MAX_TX {
+            for tx_old in &mut self.transactions {
+                if !tx_old.complete {
+                    tx_old.complete = true;
+                    MQTTState::set_event(tx_old, MQTTEvent::TooManyTransactions);
+                    break;
+                }
+            }
+        }
         return tx;
     }
 
@@ -738,6 +751,7 @@ pub extern "C" fn rs_mqtt_state_get_event_info_by_id(event_id: std::os::raw::c_i
             MQTTEvent::InvalidQosLevel     => { "invalid_qos_level\0" },
             MQTTEvent::MissingMsgId        => { "missing_msg_id\0" },
             MQTTEvent::UnassignedMsgtype   => { "unassigned_msg_type\0" },
+            MQTTEvent::TooManyTransactions => { "too_many_transactions\0" },
             MQTTEvent::MalformedTraffic    => { "malformed_traffic\0" },
         };
         unsafe{