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

rules/mqtt-events.rules

@@ -13,3 +13,4 @@ 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;)

rust/src/mqtt/mqtt.rs

@@ -35,6 +35,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, AppLayerEvent)]
@@ -48,6 +51,7 @@ pub enum MQTTEvent {
     InvalidQosLevel,
     MissingMsgId,
     UnassignedMsgType,
+    TooManyTransactions,
 }
 
 #[derive(Debug)]
@@ -163,6 +167,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;
     }