Provide an explicit state for the flow classification process (#2942)

Application should keep calling nDPI until flow state became
`NDPI_STATE_CLASSIFIED`.

The main loop in the application is simplified to something like:
```
res = ndpi_detection_process_packet(...);
if(res->state == NDPI_STATE_CLASSIFIED) {
  /* Done: you can get finale classification and all metadata.
     nDPI doesn't need more packets for this flow */
} else {
  /* nDPI needs more packets for this flow. The provided
     classification is not final and more metadata might be
     extracted.
     If `res->state` is `NDPI_STATE_PARTIAL`, partial/initial
     classification is available in `res->proto`
     as usual but it can be updated later.
  */
}

/*
    Example A (QUIC flow):
     pkt 1: proto QUIC state NDPI_STATE_PARTIAL
     pkt 2: proto QUIC/Youtube  state NDPI_STATE_CLASSIFIED
    Example B (GoogleMeet call):
     pkt 1:   proto STUN state NDPI_STATE_PARTIAL
     pkt N:   proto DTLS state NDPI_STATE_PARTIAL
     pkt N+M: proto DTLS/GoogleCall state NDPI_STATE_CLASSIFIED
    Example C (standard TLS flow):
     pkt 1:   proto Unknown state NDPI_STATE_INSPECTING
     pkt 2:   proto Unknown state NDPI_STATE_INSPECTING
     pkt 3:   proto Unknown state NDPI_STATE_INSPECTING
     pkt 4:   proto TLS/Facebook state NDPI_STATE_PARTIAL
     pkt N:   proto TLS/Facebook state NDPI_STATE_CLASSIFIED
 */
}
```
You can take a look at `ndpiReader` for a slightly more complex example.

API changes:
* remove the third parameter from `ndpi_detection_giveup()`. If you need
to know if the classification flow has been guessed, you can access
`flow->protocol_was_guessed`
* remove `ndpi_extra_dissection_possible()`
* change some prototypes from accepting `ndpi_protocol foo` to
`ndpi_master_app_protocol bar`. The update is trivial: from `foo` to
`foo.proto`

Author: IvanNardi

example/ndpiReader.c

@@ -563,7 +563,7 @@ void ndpiCheckHostStringMatch(char *testChar) {
            detected_protocol.proto.app_protocol,
            detected_protocol.breed,
            detected_protocol.category,
-           ndpi_protocol2name(ndpi_str, detected_protocol, appBufStr,
+           ndpi_protocol2name(ndpi_str, detected_protocol.proto, appBufStr,
                               sizeof(appBufStr)),
            ndpi_get_proto_breed_name(detected_protocol.breed),
            ndpi_category_get_name(ndpi_str, detected_protocol.category));
@@ -640,7 +640,7 @@ void ndpiCheckHostsFileStringMatch(const char *domains_file) {
 
       printf("Domain [%s] -> %s %s %s\n",
              line,
-             ndpi_protocol2name(ndpi_str, detected_protocol, appBufStr,
+             ndpi_protocol2name(ndpi_str, detected_protocol.proto, appBufStr,
                                 sizeof(appBufStr)),
              ndpi_get_proto_breed_name(detected_protocol.breed),
              ndpi_category_get_name(ndpi_str, detected_protocol.category));
@@ -725,7 +725,7 @@ static void ndpiCheckIPMatch(char *testChar) {
     memset(&detected_protocol, 0, sizeof(ndpi_protocol));
     detected_protocol.proto.app_protocol = ndpi_map_ndpi_id_to_user_proto_id(ndpi_str, ret);
 
-    ndpi_protocol2name(ndpi_str, detected_protocol, appBufStr,
+    ndpi_protocol2name(ndpi_str, detected_protocol.proto, appBufStr,
                        sizeof(appBufStr));
 
     printf("Match Found for IP %s, port %d -> %s (%d)\n",
@@ -1947,11 +1947,11 @@ static void printFlow(u_int32_t id, struct ndpi_flow_info *flow, u_int16_t threa
             );
 
     fprintf(csv_fp, "%s|",
-            ndpi_protocol2id(flow->detected_protocol, buf, sizeof(buf)));
+            ndpi_protocol2id(flow->detected_protocol.proto, buf, sizeof(buf)));
 
     fprintf(csv_fp, "%s|%s|%s|%s|",
             ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-                               flow->detected_protocol, buf, sizeof(buf)),
+                               flow->detected_protocol.proto, buf, sizeof(buf)),
             ndpi_stack2str(ndpi_thread_info[thread_id].workflow->ndpi_struct,
                            &flow->detected_protocol.protocol_stack, buf2, sizeof(buf2)),
             ndpi_get_proto_name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
@@ -2090,7 +2090,7 @@ static void printFlow(u_int32_t id, struct ndpi_flow_info *flow, u_int16_t threa
 	       flow->detected_protocol.proto.master_protocol,
 	       flow->detected_protocol.proto.app_protocol,
 	       ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-				  flow->detected_protocol, buf1, sizeof(buf1))
+				  flow->detected_protocol.proto, buf1, sizeof(buf1))
 	  );
       }
     }
@@ -2132,14 +2132,14 @@ static void printFlow(u_int32_t id, struct ndpi_flow_info *flow, u_int16_t threa
 
 #ifdef NDPI_EXTENDED_SANITY_CHECKS
     /* Be sure new stack logic is compatible with legacy code */
-    assert(ndpi_stack_get_upper_proto(&flow->detected_protocol.protocol_stack) == ndpi_get_upper_proto(flow->detected_protocol));
-    assert(ndpi_stack_get_lower_proto(&flow->detected_protocol.protocol_stack) == ndpi_get_lower_proto(flow->detected_protocol));
+    assert(ndpi_stack_get_upper_proto(&flow->detected_protocol.protocol_stack) == ndpi_get_upper_proto(flow->detected_protocol.proto));
+    assert(ndpi_stack_get_lower_proto(&flow->detected_protocol.protocol_stack) == ndpi_get_lower_proto(flow->detected_protocol.proto));
 #endif
 
     fprintf(out, "%s/%s][Stack: %s][IP: %u/%s]",
-	    ndpi_protocol2id(flow->detected_protocol, buf, sizeof(buf)),
+	    ndpi_protocol2id(flow->detected_protocol.proto, buf, sizeof(buf)),
 	    ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-			       flow->detected_protocol, buf1, sizeof(buf1)),
+			       flow->detected_protocol.proto, buf1, sizeof(buf1)),
 	    ndpi_stack2str(ndpi_thread_info[thread_id].workflow->ndpi_struct,
 		           &flow->detected_protocol.protocol_stack, buf2, sizeof(buf2)),
 	    flow->detected_protocol.protocol_by_ip,
@@ -2173,7 +2173,7 @@ static void printFlow(u_int32_t id, struct ndpi_flow_info *flow, u_int16_t threa
 
     fprintf(out, "[%s]",
 	    ndpi_is_encrypted_proto(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-				    flow->detected_protocol) ? "Encrypted" : "ClearText");
+				    flow->detected_protocol.proto) ? "Encrypted" : "ClearText");
 
     fprintf(out, "[Confidence: %s]", ndpi_confidence_get_name(flow->confidence));
 
@@ -2748,14 +2748,13 @@ static void node_proto_guess_walker(const void *node, ndpi_VISIT which, int dept
 
   if((which == ndpi_preorder) || (which == ndpi_leaf)) { /* Avoid walking the same node multiple times */
     if((!flow->detection_completed) && flow->ndpi_flow) {
-      u_int8_t proto_guessed;
 
       malloc_size_stats = 1;
       flow->detected_protocol = ndpi_detection_giveup(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-                                                      flow->ndpi_flow, &proto_guessed);
+                                                      flow->ndpi_flow);
       malloc_size_stats = 0;
 
-      if(proto_guessed) ndpi_thread_info[thread_id].workflow->stats.guessed_flow_protocols++;
+      if(flow->ndpi_flow->protocol_was_guessed) ndpi_thread_info[thread_id].workflow->stats.guessed_flow_protocols++;
     }
 
     process_ndpi_collected_info(ndpi_thread_info[thread_id].workflow, flow);
@@ -3157,7 +3156,7 @@ static void port_stats_walker(const void *node, ndpi_VISIT which, int depth, voi
     /* get app level protocol */
     if(flow->detected_protocol.proto.master_protocol) {
       ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
-                         flow->detected_protocol, proto, sizeof(proto));
+                         flow->detected_protocol.proto, proto, sizeof(proto));
     } else {
       strncpy(proto, ndpi_get_proto_name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
                                          flow->detected_protocol.proto.app_protocol),sizeof(proto) - 1);
@@ -3265,7 +3264,7 @@ static void dump_realtime_protocol(struct ndpi_workflow * workflow, struct ndpi_
     snprintf(dstip, sizeof(dstip), "[%s]", flow->dst_name ? flow->dst_name : "");
   }
 
-  ndpi_protocol2name(workflow->ndpi_struct, flow->detected_protocol, app_name, sizeof(app_name));
+  ndpi_protocol2name(workflow->ndpi_struct, flow->detected_protocol.proto, app_name, sizeof(app_name));
 
   if (ret == 1) {
     fprintf(out, "Detected Realtime protocol %s --> [%s] %s:%d <--> %s:%d app=%s <%s>\n",
@@ -4114,7 +4113,7 @@ static void printFlowsStats() {
               fprintf(out, "\t%u\t%-10s\t%s:%u <-> %s:%u\t[",
                       i,
                       ndpi_protocol2name(ndpi_thread_info[0].workflow->ndpi_struct,
-                                         all_flows[i].flow->detected_protocol, buf, sizeof(buf)),
+                                         all_flows[i].flow->detected_protocol.proto, buf, sizeof(buf)),
                       all_flows[i].flow->src_name ? all_flows[i].flow->src_name : "",
                       ntohs(all_flows[i].flow->src_port),
                       all_flows[i].flow->dst_name ? all_flows[i].flow->dst_name : "",
@@ -5113,7 +5112,7 @@ static void ndpi_process_packet(u_char *args,
     }
     trailer->flow_risk_info[sizeof(trailer->flow_risk_info) - 1] = '\0';
     trailer->proto.master_protocol = htons(p.proto.master_protocol), trailer->proto.app_protocol = htons(p.proto.app_protocol);
-    ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct, p, trailer->name, sizeof(trailer->name));
+    ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct, p.proto, trailer->name, sizeof(trailer->name));
 
     /* Metadata */
     /* Metadata are (all) available in `flow` only after nDPI completed its work!

example/ndpiSimpleIntegration.c

@@ -876,12 +876,10 @@ static void ndpi_process_packet(uint8_t * const args,
     return;
   } else if (flow_to_process->ndpi_flow->num_processed_pkts == 0xFE) {
     /* last chance to guess something, better then nothing */
-    uint8_t protocol_was_guessed = 0;
     flow_to_process->guessed_protocol =
       ndpi_detection_giveup(workflow->ndpi_struct,
-			    flow_to_process->ndpi_flow,
-			    &protocol_was_guessed);
-    if (protocol_was_guessed != 0) {
+			    flow_to_process->ndpi_flow);
+    if (flow_to_process->ndpi_flow->protocol_was_guessed != 0) {
       printf("[%8llu, %d, %4d][GUESSED] protocol: %s | app protocol: %s | category: %s\n",
 	     workflow->packets_captured,
 	     reader_thread->array_index,

example/reader_util.c

@@ -1118,7 +1118,7 @@ static void dump_flow_fingerprint(struct ndpi_workflow * workflow,
     ndpi_serialize_string_uint32(&serializer, "srv_port", ntohs(flow->dst_port));
     ndpi_serialize_string_string(&serializer, "proto",
 				 ndpi_protocol2name(workflow->ndpi_struct,
-						    flow->detected_protocol,
+						    flow->detected_protocol.proto,
 						    buf, sizeof(buf)));
 
     if(flow->server_hostname)
@@ -1165,7 +1165,7 @@ static void process_ndpi_monitoring_info(struct ndpi_flow_info *flow) {
     return;
 
   if(flow->monitoring_state == 0 &&
-     flow->ndpi_flow->monitoring) {
+     flow->ndpi_flow->state == NDPI_STATE_MONITORING) {
     /* We just moved to monitoring state */
     flow->monitoring_state = 1;
     flow->num_packets_before_monitoring = flow->ndpi_flow->packet_direction_complete_counter[0] + flow->ndpi_flow->packet_direction_complete_counter[1];
@@ -1680,7 +1680,7 @@ void process_ndpi_collected_info(struct ndpi_workflow * workflow, struct ndpi_fl
     ndpi_serialize_string_uint32(&flow->ndpi_flow_serializer, "detection_completed", flow->detection_completed);
     ndpi_serialize_string_uint32(&flow->ndpi_flow_serializer, "check_extra_packets", flow->check_extra_packets);
 
-    if(flow->ndpi_flow->monitoring) {
+    if(flow->ndpi_flow->state == NDPI_STATE_MONITORING) {
       serialize_monitoring_metadata(flow);
     }
 
@@ -1994,26 +1994,19 @@ static struct ndpi_proto packet_processing(struct ndpi_workflow * workflow,
 							    ipsize, time_ms, &input_info);
     if(monitoring_enabled)
       process_ndpi_monitoring_info(flow);
-    enough_packets |= ndpi_flow->fail_with_unknown;
-    if(enough_packets || (flow->detected_protocol.proto.app_protocol != NDPI_PROTOCOL_UNKNOWN)) {
-      if((!enough_packets)
-	 && ndpi_extra_dissection_possible(workflow->ndpi_struct, ndpi_flow))
-	; /* Wait for further metadata */
-      else {
-	/* New protocol detected or give up */
-	flow->detection_completed = 1;
-
-	if(flow->detected_protocol.proto.app_protocol == NDPI_PROTOCOL_UNKNOWN) {
-	  u_int8_t proto_guessed;
-
-	  flow->detected_protocol = ndpi_detection_giveup(workflow->ndpi_struct, flow->ndpi_flow,
-							  &proto_guessed);
-	  if(proto_guessed) workflow->stats.guessed_flow_protocols++;
-	}
+    if(flow->detected_protocol.state == NDPI_STATE_CLASSIFIED ||
+       enough_packets) {
+
+      flow->detection_completed = 1;
 
-	process_ndpi_collected_info(workflow, flow);
+      if(flow->detected_protocol.state != NDPI_STATE_CLASSIFIED) {
+         flow->detected_protocol = ndpi_detection_giveup(workflow->ndpi_struct, flow->ndpi_flow);
       }
+
+      if(flow->ndpi_flow->protocol_was_guessed) workflow->stats.guessed_flow_protocols++;
+      process_ndpi_collected_info(workflow, flow);
     }
+
     /* Let's try to save client-server direction */
     flow->current_pkt_from_client_to_server = input_info.in_pkt_dir;
 

fuzz/fuzz_config.cpp

@@ -12,7 +12,7 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
   FuzzedDataProvider fuzzed_data(data, size);
   struct ndpi_detection_module_struct *ndpi_info_mod;
   struct ndpi_flow_struct flow;
-  u_int8_t protocol_was_guessed, unused;
+  u_int8_t unused;
   u_int32_t i, ret;
   u_int16_t bool_value;
   struct ndpi_lru_cache_stats lru_stats;
@@ -833,15 +833,15 @@ extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
   ndpi_detection_get_l4(pkt.data(), pkt.size(), &l4_return, &l4_len_return, &l4_protocol_return, NDPI_DETECTION_ONLY_IPV4);
 
   ndpi_detection_process_packet(ndpi_info_mod, &flow, pkt.data(), pkt.size(), 0, &input_info);
-  p = ndpi_detection_giveup(ndpi_info_mod, &flow, &protocol_was_guessed);
+  p = ndpi_detection_giveup(ndpi_info_mod, &flow);
 
   assert(p.proto.master_protocol == ndpi_get_flow_masterprotocol(&flow));
   assert(p.proto.app_protocol == ndpi_get_flow_appprotocol(&flow));
   assert(p.category == ndpi_get_flow_category(&flow));
   ndpi_is_master_only_protocol(ndpi_info_mod, p.proto.app_protocol);
   ndpi_normalize_protocol(ndpi_info_mod, &p.proto);
-  assert(ndpi_stack_get_upper_proto(&p.protocol_stack) == ndpi_get_upper_proto(p));
-  assert(ndpi_stack_get_lower_proto(&p.protocol_stack) == ndpi_get_lower_proto(p));
+  assert(ndpi_stack_get_upper_proto(&p.protocol_stack) == ndpi_get_upper_proto(p.proto));
+  assert(ndpi_stack_get_lower_proto(&p.protocol_stack) == ndpi_get_lower_proto(p.proto));
   ndpi_get_flow_error_code(&flow);
   ndpi_get_flow_risk_info(&flow, out, sizeof(out), 1);
   ndpi_get_flow_ndpi_proto(&flow, &p2);

fuzz/fuzz_ndpi_reader.c

@@ -47,10 +47,8 @@ static void node_cleanup_walker(const void *node, ndpi_VISIT which, int depth, v
 
   if((which == ndpi_preorder) || (which == ndpi_leaf)) { /* Avoid walking the same node multiple times */
     if((!flow->detection_completed) && flow->ndpi_flow) {
-      u_int8_t proto_guessed;
-
       flow->detected_protocol = ndpi_detection_giveup(workflow->ndpi_struct,
-                                                      flow->ndpi_flow, &proto_guessed);
+                                                      flow->ndpi_flow);
     }
 
     process_ndpi_collected_info(workflow, flow);

fuzz/fuzz_process_packet.c

@@ -10,8 +10,6 @@ static ndpi_serializer json_serializer = {};
 static ndpi_serializer csv_serializer = {};
 
 int LLVMFuzzerTestOneInput(const uint8_t *Data, size_t Size) {
-  uint8_t protocol_was_guessed;
-
   if (ndpi_info_mod == NULL) {
     fuzz_init_detection_module(&ndpi_info_mod, NULL);
 
@@ -22,19 +20,12 @@ int LLVMFuzzerTestOneInput(const uint8_t *Data, size_t Size) {
   memset(&ndpi_flow, 0, SIZEOF_FLOW_STRUCT);
   ndpi_protocol detected_protocol =
     ndpi_detection_process_packet(ndpi_info_mod, &ndpi_flow, Data, Size, 0, NULL);
-  ndpi_protocol guessed_protocol =
-    ndpi_detection_giveup(ndpi_info_mod, &ndpi_flow, &protocol_was_guessed);
+  detected_protocol = ndpi_detection_giveup(ndpi_info_mod, &ndpi_flow);
 
   ndpi_reset_serializer(&json_serializer);
   ndpi_reset_serializer(&csv_serializer);
-  if (protocol_was_guessed == 0)
-  {
-    ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, detected_protocol, &json_serializer);
-    ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, detected_protocol, &csv_serializer);
-  } else {
-    ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, guessed_protocol, &json_serializer);
-    ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, guessed_protocol, &csv_serializer);
-  }
+  ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, detected_protocol, &json_serializer);
+  ndpi_dpi2json(ndpi_info_mod, &ndpi_flow, detected_protocol, &csv_serializer);
   ndpi_free_flow_data(&ndpi_flow);
 
   return 0;

python/ndpi/ndpi.py

@@ -59,8 +59,7 @@ def process_packet(self, flow, packet, packet_time_ms, input_info):
 
     def giveup(self, flow):
         p = lib.ndpi_detection_giveup(self._detection_module,
-                                      flow.C,
-                                      ffi.new("uint8_t*", 0))
+                                      flow.C)
         return ndpi_protocol(C=p,
                              master_protocol=p.proto.master_protocol,
                              app_protocol=p.proto.app_protocol,

python/ndpi/ndpi_build.py

@@ -56,11 +56,10 @@
                                             const u_int64_t packet_time_ms,
                                             struct ndpi_flow_input_info *input_info);
 ndpi_protocol ndpi_detection_giveup(struct ndpi_detection_module_struct *ndpi_struct,
-                                    struct ndpi_flow_struct *flow,
-                                    u_int8_t *protocol_was_guessed);
+                                    struct ndpi_flow_struct *flow);
 void ndpi_py_setup_detection_module(struct ndpi_detection_module_struct *mod);
 struct ndpi_flow_struct * ndpi_py_initialize_flow(void);
-char* ndpi_protocol2name(struct ndpi_detection_module_struct *ndpi_mod, ndpi_protocol proto, char *buf, u_int buf_len);
+char* ndpi_protocol2name(struct ndpi_detection_module_struct *ndpi_mod, ndpi_master_app_protocol proto, char *buf, u_int buf_len);
 const char* ndpi_category_get_name(struct ndpi_detection_module_struct *ndpi_mod, ndpi_protocol_category_t category);
 const char* ndpi_confidence_get_name(ndpi_confidence_t confidence);
 """

python/ndpi_example.py

@@ -151,8 +151,8 @@ def parse_arguments():
             flow.detected_protocol = nDPI.giveup(flow.ndpi_flow)  # We try to guess it (port matching, LRU, etc.)
         FLOW_EXPORT = FLOW_STR.format(flow.index,
                                       key,
-                                      nDPI.protocol_name(flow.detected_protocol),
-                                      nDPI.protocol_category_name(flow.detected_protocol),
+                                      nDPI.protocol_name(flow.detected_protocol.proto),
+                                      nDPI.protocol_category_name(flow.detected_protocol.proto),
                                       flow.ndpi_flow.confidence.name,
                                       flow.pkts,
                                       flow.bytes)