Add monitoring capability (#2588)

Allow nDPI to process the entire flows and not only the first N packets.
Usefull when the application is interested in some metadata spanning the
entire life of the session.

As initial step, only STUN flows can be put in monitoring.

See `doc/monitoring.md` for further details.

This feature is disabled by default.

Close #2583

4 files changed, 119 insertions, 20 deletions

diff --git a/src/include/ndpi_private.h b/src/include/ndpi_private.h
index 2278428a5..a72a956d7 100644
--- a/src/include/ndpi_private.h
+++ b/src/include/ndpi_private.h
@@ -284,6 +284,7 @@ struct ndpi_detection_module_config_struct {
 
   NDPI_PROTOCOL_BITMASK debug_bitmask;
   NDPI_PROTOCOL_BITMASK ip_list_bitmask;
+  NDPI_PROTOCOL_BITMASK monitoring;
 
   int flow_risk_lists_enabled;
   int risk_anonymous_subscriber_list_icloudprivaterelay_enabled;
@@ -637,6 +638,8 @@ bool ndpi_cache_address(struct ndpi_detection_module_struct *ndpi_struct,
 			ndpi_ip_addr_t ip_addr, char *hostname,
 			u_int32_t epoch_now, u_int32_t ttl);
 
+int is_monitoring_enabled(struct ndpi_detection_module_struct *ndpi_str, int protoId);
+
   /* TLS */
 int processClientServerHello(struct ndpi_detection_module_struct *ndpi_struct,
                              struct ndpi_flow_struct *flow, uint32_t quic_version);
diff --git a/src/include/ndpi_typedefs.h b/src/include/ndpi_typedefs.h
index b401aad11..82eaabcc3 100644
--- a/src/include/ndpi_typedefs.h
+++ b/src/include/ndpi_typedefs.h
@@ -1232,6 +1232,18 @@ struct ndpi_risk_information {
   char *info;  
 };
 
+struct ndpi_metadata_monitoring {
+  union {
+    struct {
+      ndpi_address_port mapped_address;
+      ndpi_address_port peer_address;
+      ndpi_address_port relayed_address;
+      ndpi_address_port response_origin;
+      ndpi_address_port other_address;
+    } dtls_stun_rtp;
+  } protos;
+};
+
 struct ndpi_flow_struct {
   u_int16_t detected_protocol_stack[NDPI_PROTOCOL_SIZE];
 
@@ -1239,6 +1251,7 @@ struct ndpi_flow_struct {
   u_int16_t guessed_protocol_id, guessed_protocol_id_by_ip, guessed_category, guessed_header_category;
   u_int8_t l4_proto, protocol_id_already_guessed:1, fail_with_unknown:1,
     init_finished:1, client_packet_direction:1, packet_direction:1, is_ipv6:1, first_pkt_fully_encrypted:1, skip_entropy_check: 1;
+  u_int8_t monitoring: 1, _pad:7;
 
   u_int16_t num_dissector_calls;
   ndpi_confidence_t confidence; /* ndpi_confidence_t */
@@ -1486,7 +1499,8 @@ struct ndpi_flow_struct {
     } slp;
   } protos;
 
-  /*** ALL protocol specific 64 bit variables here ***/
+  /* **Packet** metadata for flows where monitoring is enabled. It is reset after each packet! */
+  struct ndpi_metadata_monitoring *monit;
 
   /* protocols which have marked a connection as this connection cannot be protocol XXX, multiple u_int64_t */
   NDPI_PROTOCOL_BITMASK excluded_protocol_bitmask;
@@ -1573,8 +1587,8 @@ struct ndpi_flow_struct {
 _Static_assert(sizeof(((struct ndpi_flow_struct *)0)->protos) <= 264,
                "Size of the struct member protocols increased to more than 264 bytes, "
                "please check if this change is necessary.");
-_Static_assert(sizeof(struct ndpi_flow_struct) <= 1160,
-               "Size of the flow struct increased to more than 1160 bytes, "
+_Static_assert(sizeof(struct ndpi_flow_struct) <= 1176,
+               "Size of the flow struct increased to more than 1176 bytes, "
                "please check if this change is necessary.");
 #endif
 #endif
diff --git a/src/lib/ndpi_main.c b/src/lib/ndpi_main.c
index c1d5e39dc..bcd3520fb 100644
--- a/src/lib/ndpi_main.c
+++ b/src/lib/ndpi_main.c
@@ -3590,6 +3590,15 @@ static int is_ip_list_enabled(struct ndpi_detection_module_struct *ndpi_str, int
 
 /* *********************************************** */
 
+int is_monitoring_enabled(struct ndpi_detection_module_struct *ndpi_str, int protoId)
+{
+  if(NDPI_COMPARE_PROTOCOL_TO_BITMASK(ndpi_str->cfg.monitoring, protoId) == 0)
+    return 0;
+  return 1;
+}
+
+/* *********************************************** */
+
 int ndpi_finalize_initialization(struct ndpi_detection_module_struct *ndpi_str) {
   u_int i;
 
@@ -6742,6 +6751,9 @@ void ndpi_free_flow_data(struct ndpi_flow_struct* flow) {
     if(flow->kerberos_buf.pktbuf)
       ndpi_free(flow->kerberos_buf.pktbuf);
 
+    if(flow->monit)
+      ndpi_free(flow->monit);
+
    if(flow_is_proto(flow, NDPI_PROTOCOL_QUIC) ||
        flow_is_proto(flow, NDPI_PROTOCOL_TLS) ||
        flow_is_proto(flow, NDPI_PROTOCOL_DTLS) ||
@@ -7997,11 +8009,10 @@ void ndpi_process_extra_packet(struct ndpi_detection_module_struct *ndpi_str,
 
   /* call the extra packet function (which may add more data/info to flow) */
   if(flow->extra_packets_func) {
-    if((flow->extra_packets_func(ndpi_str, flow)) == 0)
-      flow->extra_packets_func = NULL; /* Enough packets detected */
-
-    if(++flow->num_extra_packets_checked == flow->max_extra_packets_to_check)
-      flow->extra_packets_func = NULL; /* Enough packets detected */
+    if((flow->extra_packets_func(ndpi_str, flow) == 0) ||
+       (!flow->monitoring && ++flow->num_extra_packets_checked == flow->max_extra_packets_to_check)) {
+      flow->extra_packets_func = NULL; /* Done */
+    }
   }
 }
 
@@ -8610,12 +8621,17 @@ static ndpi_protocol ndpi_internal_detection_process_packet(struct ndpi_detectio
   ret.protocol_by_ip = flow->guessed_protocol_id_by_ip;
   ret.category = flow->category;
 
+  if(flow->monit)
+    memset(flow->monit, '\0', sizeof(*flow->monit));
+
   if(flow->fail_with_unknown) {
     // printf("%s(): FAIL_WITH_UNKNOWN\n", __FUNCTION__);
     return(ret);
   }
 
-  if(ndpi_str->cfg.max_packets_to_process > 0 && flow->num_processed_pkts >= ndpi_str->cfg.max_packets_to_process) {
+  if(ndpi_str->cfg.max_packets_to_process > 0 &&
+     flow->num_processed_pkts >= ndpi_str->cfg.max_packets_to_process &&
+     !flow->monitoring) {
     flow->extra_packets_func = NULL; /* To allow ndpi_extra_dissection_possible() to fail */
     flow->fail_with_unknown = 1;
     /* Let's try to update ndpi_str->input_info->in_pkt_dir even in this case.
@@ -9320,6 +9336,13 @@ void ndpi_set_detected_protocol(struct ndpi_detection_module_struct *ndpi_str, s
 				ndpi_confidence_t confidence) {
   ndpi_protocol ret;
 
+  if(flow->monitoring) {
+    NDPI_LOG_ERR(ndpi_str, "Impossible to update classification while in monitoring state! %d/%d->%d/%d\n",
+                 flow->detected_protocol_stack[1], flow->detected_protocol_stack[0],
+                 upper_detected_protocol, lower_detected_protocol);
+    return;
+  }
+
   ndpi_int_change_protocol(flow, upper_detected_protocol, lower_detected_protocol, confidence);
   ret.proto.master_protocol = flow->detected_protocol_stack[1], ret.proto.app_protocol = flow->detected_protocol_stack[0];
   ndpi_reconcile_protocols(ndpi_str, flow, &ret);
@@ -11380,6 +11403,7 @@ static const struct cfg_param {
 
   { "$PROTO_NAME_OR_ID", "log",                                 "disable", NULL, NULL, CFG_PARAM_PROTOCOL_ENABLE_DISABLE, __OFF(debug_bitmask), NULL },
   { "$PROTO_NAME_OR_ID", "ip_list.load",                        "1", NULL, NULL, CFG_PARAM_PROTOCOL_ENABLE_DISABLE, __OFF(ip_list_bitmask), NULL },
+  { "$PROTO_NAME_OR_ID", "monitoring",                          "disable", NULL, NULL, CFG_PARAM_PROTOCOL_ENABLE_DISABLE, __OFF(monitoring), NULL },
 
   /* Global parameters */
 
diff --git a/src/lib/protocols/stun.c b/src/lib/protocols/stun.c
index acc0e1e91..7207c1b9f 100644
--- a/src/lib/protocols/stun.c
+++ b/src/lib/protocols/stun.c
@@ -187,7 +187,8 @@ static void add_to_cache(struct ndpi_detection_module_struct *ndpi_struct,
 }
 
 static void parse_ip_port_attribute(const u_int8_t *payload, u_int16_t payload_length,
-                                    int off, u_int16_t real_len,ndpi_address_port *ap)
+                                    int off, u_int16_t real_len, ndpi_address_port *ap,
+                                    ndpi_address_port *ap_monit)
 {
   if(off + 4 + real_len <= payload_length &&
      (real_len == 8 || real_len == 20)) {
@@ -201,6 +202,9 @@ static void parse_ip_port_attribute(const u_int8_t *payload, u_int16_t payload_l
       ap->port = port;
       ap->address.v4 = htonl(ip);
       ap->is_ipv6 = 0;
+
+      if(ap_monit)
+        memcpy(ap_monit, ap, sizeof(*ap_monit));
     } else if(protocol_family == 0x02 /* IPv6 */ &&
               real_len == 20) {
       u_int16_t port = ntohs(*((u_int16_t*)&payload[off+6]));
@@ -214,6 +218,9 @@ static void parse_ip_port_attribute(const u_int8_t *payload, u_int16_t payload_l
       ap->port = port;
       memcpy(&ap->address, &ip, 16);
       ap->is_ipv6 = 1;
+
+      if(ap_monit)
+        memcpy(ap_monit, ap, sizeof(*ap_monit));
     }
   }
 }
@@ -221,7 +228,8 @@ static void parse_ip_port_attribute(const u_int8_t *payload, u_int16_t payload_l
 static void parse_xor_ip_port_attribute(struct ndpi_detection_module_struct *ndpi_struct,
                                         struct ndpi_flow_struct *flow,
                                         const u_int8_t *payload, u_int16_t payload_length,
-                                        int off, u_int16_t real_len,ndpi_address_port *ap,
+                                        int off, u_int16_t real_len,
+                                        ndpi_address_port *ap, ndpi_address_port *ap_monit,
                                         u_int32_t transaction_id[3], u_int32_t magic_cookie,
                                         int add_to_cache)
 {
@@ -245,6 +253,9 @@ static void parse_xor_ip_port_attribute(struct ndpi_detection_module_struct *ndp
       ap->address.v4 = ip;
       ap->is_ipv6 = 0;
 
+      if(ap_monit)
+        memcpy(ap_monit, ap, sizeof(*ap_monit));
+
       if(add_to_cache) {
         NDPI_LOG_DBG(ndpi_struct, "Peer %s:%d [proto %d]\n",
                      inet_ntop(AF_INET, &ip, buf, sizeof(buf)), port,
@@ -277,6 +288,9 @@ static void parse_xor_ip_port_attribute(struct ndpi_detection_module_struct *ndp
       memcpy(&ap->address, &ip, 16);
       ap->is_ipv6 = 1;
 
+      if(ap_monit)
+        memcpy(ap_monit, ap, sizeof(*ap_monit));
+
       if(add_to_cache) {
         NDPI_LOG_DBG(ndpi_struct, "Peer %s:%d [proto %d]\n",
                      inet_ntop(AF_INET6, &ip, buf, sizeof(buf)), port,
@@ -407,6 +421,10 @@ int is_stun(struct ndpi_detection_module_struct *ndpi_struct,
 
   /* STUN */
 
+  if(flow->monit == NULL &&
+     is_monitoring_enabled(ndpi_struct, NDPI_PROTOCOL_STUN))
+    flow->monit = ndpi_calloc(1, sizeof(struct ndpi_metadata_monitoring));
+
   if(msg_type == 0x0800 || msg_type == 0x0801 || msg_type == 0x0802) {
     *app_proto = NDPI_PROTOCOL_WHATSAPP_CALL;
     return 1;
@@ -440,19 +458,22 @@ int is_stun(struct ndpi_detection_module_struct *ndpi_struct,
     switch(attribute) {
     case 0x0001: /* MAPPED-ADDRESS */
       if(ndpi_struct->cfg.stun_mapped_address_enabled) {
-        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.mapped_address);
+        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.mapped_address,
+                                flow->monit ? &flow->monit->protos.dtls_stun_rtp.mapped_address : NULL);
       }
       break;
 
     case 0x802b: /* RESPONSE-ORIGIN */
       if(ndpi_struct->cfg.stun_response_origin_enabled) {
-        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.response_origin);
+        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.response_origin,
+                                flow->monit ? &flow->monit->protos.dtls_stun_rtp.response_origin : NULL);
       }
       break;
 
     case 0x802c: /* OTHER-ADDRESS */
       if(ndpi_struct->cfg.stun_other_address_enabled) {
-        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.other_address);
+        parse_ip_port_attribute(payload, payload_length, off, real_len, &flow->stun.other_address,
+                                flow->monit ? &flow->monit->protos.dtls_stun_rtp.other_address : NULL);
       }
       break;
 
@@ -461,6 +482,7 @@ int is_stun(struct ndpi_detection_module_struct *ndpi_struct,
         parse_xor_ip_port_attribute(ndpi_struct, flow,
                                     payload, payload_length, off, real_len,
                                     &flow->stun.peer_address,
+                                    flow->monit ? &flow->monit->protos.dtls_stun_rtp.peer_address : NULL,
                                     transaction_id, magic_cookie, 1);
       }
       break;
@@ -555,6 +577,7 @@ int is_stun(struct ndpi_detection_module_struct *ndpi_struct,
         parse_xor_ip_port_attribute(ndpi_struct, flow,
                                     payload, payload_length, off, real_len,
                                     &flow->stun.mapped_address,
+                                    flow->monit ? &flow->monit->protos.dtls_stun_rtp.mapped_address : NULL,
                                     transaction_id, magic_cookie, 0);
       }
       break;
@@ -564,6 +587,7 @@ int is_stun(struct ndpi_detection_module_struct *ndpi_struct,
         parse_xor_ip_port_attribute(ndpi_struct, flow,
                                     payload, payload_length, off, real_len,
                                     &flow->stun.relayed_address,
+                                    flow->monit ? &flow->monit->protos.dtls_stun_rtp.relayed_address : NULL,
                                     transaction_id, magic_cookie, 0);
       }
       break;
@@ -593,14 +617,28 @@ static int keep_extra_dissection(struct ndpi_detection_module_struct *ndpi_struc
          * for the other protocols, we stop after we have all metadata (if enabled)
          * for some specific protocol, we might know that some attributes
            are never used
+
+     **After** extra dissection is ended, we might move to monitoring. Note that:
+     * classification doesn't change while in monitoring!
   */
 
-  if(!is_subclassification_real(flow))
+  if(flow->monitoring)
     return 1;
 
   if(flow->detected_protocol_stack[0] == NDPI_PROTOCOL_ZOOM)
     return 0;
 
+  if(flow->num_extra_packets_checked + 1 == flow->max_extra_packets_to_check) {
+    if(is_monitoring_enabled(ndpi_struct, NDPI_PROTOCOL_STUN)) {
+      NDPI_LOG_DBG(ndpi_struct, "Enabling monitoring (end extra dissection)\n");
+      flow->monitoring = 1;
+      return 1;
+    }
+  }
+
+  if(!is_subclassification_real(flow))
+    return 1;
+
   if(flow->detected_protocol_stack[0] == NDPI_PROTOCOL_TELEGRAM_VOIP &&
      ndpi_struct->cfg.stun_peer_address_enabled)
     return 1;
@@ -610,14 +648,26 @@ static int keep_extra_dissection(struct ndpi_detection_module_struct *ndpi_struc
      (flow->stun.peer_address.port || !ndpi_struct->cfg.stun_peer_address_enabled) &&
      (flow->stun.relayed_address.port || !ndpi_struct->cfg.stun_relayed_address_enabled) &&
      (flow->stun.response_origin.port || !ndpi_struct->cfg.stun_response_origin_enabled) &&
-     (flow->stun.other_address.port || !ndpi_struct->cfg.stun_other_address_enabled))
+     (flow->stun.other_address.port || !ndpi_struct->cfg.stun_other_address_enabled)) {
+    if(is_monitoring_enabled(ndpi_struct, NDPI_PROTOCOL_STUN)) {
+      NDPI_LOG_DBG(ndpi_struct, "Enabling monitoring (found all metadata)\n");
+      flow->monitoring = 1;
+      return 1;
+    }
     return 0;
+  }
 
   /* Exception WA: only relayed and mapped address attributes */
   if(flow->detected_protocol_stack[0] == NDPI_PROTOCOL_WHATSAPP_CALL &&
      (flow->stun.mapped_address.port || !ndpi_struct->cfg.stun_mapped_address_enabled) &&
-     (flow->stun.relayed_address.port || !ndpi_struct->cfg.stun_relayed_address_enabled))
+     (flow->stun.relayed_address.port || !ndpi_struct->cfg.stun_relayed_address_enabled)) {
+    if(is_monitoring_enabled(ndpi_struct, NDPI_PROTOCOL_STUN)) {
+      NDPI_LOG_DBG(ndpi_struct, "Enabling monitor (found all metadata; wa case)\n");
+      flow->monitoring = 1;
+      return 1;
+    }
     return 0;
+  }
 
   return 1;
 }
@@ -642,8 +692,10 @@ static int stun_search_again(struct ndpi_detection_module_struct *ndpi_struct,
   int first_dtls_pkt = 0;
   u_int16_t old_proto_stack[2] = {NDPI_PROTOCOL_UNKNOWN, NDPI_PROTOCOL_UNKNOWN};
 
-  NDPI_LOG_DBG2(ndpi_struct, "Packet counter %d protos %d/%d\n", flow->packet_counter,
-                flow->detected_protocol_stack[0], flow->detected_protocol_stack[1]);
+  NDPI_LOG_DBG2(ndpi_struct, "Packet counter %d protos %d/%d Monitoring? %d\n",
+                flow->packet_counter,
+                flow->detected_protocol_stack[0], flow->detected_protocol_stack[1],
+                flow->monitoring);
 
   /* TODO: check TCP support. We need to pay some attention because:
      * multiple msg in the same TCP segment
@@ -694,7 +746,7 @@ static int stun_search_again(struct ndpi_detection_module_struct *ndpi_struct,
 
       if(flow->tls_quic.certificate_processed == 1) {
         NDPI_LOG_DBG(ndpi_struct, "Interesting DTLS stuff already processed. Ignoring\n");
-      } else {
+      } else if(!flow->monitoring) {
         NDPI_LOG_DBG(ndpi_struct, "Switch to DTLS (%d/%d)\n",
                      flow->detected_protocol_stack[0], flow->detected_protocol_stack[1]);
 
@@ -746,6 +798,8 @@ static int stun_search_again(struct ndpi_detection_module_struct *ndpi_struct,
 
         NDPI_LOG_DBG(ndpi_struct, "(%d/%d)\n",
                      flow->detected_protocol_stack[0], flow->detected_protocol_stack[1]);
+      } else {
+        NDPI_LOG_DBG(ndpi_struct, "Skip DTLS packet because in monitoring\n");
       }
     }
   } else if(first_byte <= 79) {
@@ -897,6 +951,10 @@ static void ndpi_int_stun_add_connection(struct ndpi_detection_module_struct *nd
   ndpi_confidence_t confidence = NDPI_CONFIDENCE_DPI;
   u_int16_t new_app_proto;
 
+  /* In monitoring the classification can't change again */
+  if(flow->monitoring)
+    return;
+
   NDPI_LOG_DBG(ndpi_struct, "Wanting %d/%d\n", master_proto, app_proto);
 
   if(app_proto == NDPI_PROTOCOL_UNKNOWN) {