supported tcp retransmission

src/amqp_plugin.h

@@ -37,6 +37,7 @@
 EXT void amqp_plugin(int, struct configuration *, void *);
 EXT void amqp_cache_purge(struct chained_cache *[], int);
 
+#if 0
 /* global vars */
 EXT void (*insert_func)(struct primitives_ptrs *, struct insert_data *); /* pointer to INSERT function */
 EXT void (*purge_func)(struct chained_cache *[], int); /* pointer to purge function */ 
@@ -46,6 +47,7 @@ EXT struct chained_cache **queries_queue;
 EXT struct timeval flushtime;
 EXT int qq_ptr, pp_size, pb_size, pn_size, pm_size, dbc_size, quit; 
 EXT time_t refresh_deadline;
+#endif
 
 EXT struct timeval sbasetime;
 #undef EXT

src/cfg_handlers.c

@@ -259,6 +259,7 @@ int cfg_key_aggregate(char *filename, char *name, char *value_ptr)
     else if (!strcmp(count_token, "label")) cfg_set_aggregate(filename, value, COUNT_INT_LABEL, count_token);
     else if (!strcmp(count_token, "export_proto_seqno")) cfg_set_aggregate(filename, value, COUNT_INT_EXPORT_PROTO_SEQNO, count_token);
     else if (!strcmp(count_token, "export_proto_version")) cfg_set_aggregate(filename, value, COUNT_INT_EXPORT_PROTO_VERSION, count_token);
+    else if (!strcmp(count_token, "tcp_retransmission")) cfg_set_aggregate(filename, value, COUNT_INT_TCP_RETRANSMISSION, count_token);
     else {
       cpptrs.primitive[cpptrs.num].name = count_token;
       cpptrs.num++;

src/ip_flow.c

@@ -98,6 +98,30 @@ void evaluate_tcp_flags(struct timeval *now, struct packet_ptrs *pptrs, struct i
   unsigned int rev = idx ? 0 : 1;
 
   if (fp->proto == IPPROTO_TCP) {
+    /* check tcp retransmission */
+    u_int8_t tcphdrlen = ((struct my_tcphdr *)pptrs->tlh_ptr)->th_off;
+    u_int16_t tcp_payload_len = 0;
+    u_int8_t tcp_flags = ((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags;
+    u_int32_t seq = ntohl(((struct my_tcphdr *)pptrs->tlh_ptr)->th_seq);
+    u_int32_t ack = ntohl(((struct my_tcphdr *)pptrs->tlh_ptr)->th_ack);
+    tcphdrlen <<= 2;
+    tcp_payload_len = pptrs->l3_payload_len - tcphdrlen;
+
+    if ((tcp_payload_len > 0 || tcp_flags & (TH_SYN | TH_FIN))) {
+      if (fp->next_tcp_seq[idx] && ((int64_t)seq - (int64_t)fp->next_tcp_seq[idx]) < 0) {
+        /* this packet is tcp retransmission */
+        pptrs->tcp_retransmission = 1;
+      }
+      else {
+        u_int32_t nextseq = seq + tcp_payload_len;
+        pptrs->tcp_retransmission = 0;
+        if (tcp_flags & (TH_SYN | TH_FIN)) {
+          nextseq += 1;
+        }
+        fp->next_tcp_seq[idx] = nextseq;
+      }
+    }
+
     /* evaluating the transition to the ESTABLISHED state: we need to be as much
        precise as possible as the lifetime for an established flow is quite high.
        We check that we have a) SYN flag on a forward direction, b) SYN+ACK on the

src/ip_flow.h

@@ -45,9 +45,10 @@ struct ip_flow_common {
      [0] = forward flow data
      [1] = reverse flow data
   */
-  u_int16_t bucket;
+  u_int32_t bucket;
   struct timeval last[2];
   u_int32_t last_tcp_seq;
+  u_int32_t next_tcp_seq[2];
   u_int8_t tcp_flags[2];
   u_int8_t proto;
   /* classifier hooks */

src/network.h

@@ -377,6 +377,9 @@ struct packet_ptrs {
   MMDB_lookup_result_s geoipv2_src;
   MMDB_lookup_result_s geoipv2_dst;
 #endif
+  u_int8_t tcp_retransmission;
+  u_int16_t l3_payload_len;
+  u_int8_t tunnel_level;
 };
 
 struct host_addr {
@@ -435,6 +438,7 @@ struct pkt_primitives {
   u_int16_t pkt_len_distrib;
   u_int32_t export_proto_seqno;
   u_int16_t export_proto_version;
+  u_int8_t tcp_retransmission;
 };
 
 struct pkt_data {

src/nl.c

@@ -108,13 +108,17 @@ int ip_handler(register struct packet_ptrs *pptrs)
   register unsigned char *ptr;
   register u_int16_t off = pptrs->iph_ptr-pptrs->packet_ptr, off_l4;
   int ret = TRUE, num, is_fragment = 0;
+  u_int16_t dgram_len = 0;
 
   /* len: number of 32bit words forming the header */
   len = IP_HL(((struct my_iphdr *) pptrs->iph_ptr));
   len <<= 2;
   ptr = pptrs->iph_ptr+len;
   off += len;
 
+  dgram_len = ntohs(((struct my_iphdr *) pptrs->iph_ptr)->ip_len);
+  pptrs->l3_payload_len = dgram_len - len;
+
   /* check len */
   if (off > caplen) return FALSE; /* IP packet truncated */
   pptrs->l4_proto = ((struct my_iphdr *)pptrs->iph_ptr)->ip_p;
@@ -166,21 +170,33 @@ int ip_handler(register struct packet_ptrs *pptrs)
     }
 
     if (config.handle_flows) {
-      pptrs->tcp_flags = FALSE;
-
-      if (pptrs->l4_proto == IPPROTO_TCP) {
-        if (off_l4+TCPFlagOff+1 > caplen) {
-          Log(LOG_INFO, "INFO ( %s/core ): short IPv4 packet read (%u/%u/flows). Snaplen issue ?\n",
-			config.name, caplen, off_l4+TCPFlagOff+1);
-          return FALSE;
+      u_int8_t do_flow_handler = 0;
+      if (config.tunnel0 || pptrs->tun_stack) {
+        if (pptrs->tunnel_level) {
+          do_flow_handler = 1;
         }
-        if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_SYN) pptrs->tcp_flags |= TH_SYN;
-        if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_FIN) pptrs->tcp_flags |= TH_FIN;
-        if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_RST) pptrs->tcp_flags |= TH_RST;
-        if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_ACK && pptrs->tcp_flags) pptrs->tcp_flags |= TH_ACK;
       }
+      else {
+        do_flow_handler = 1;
+      }
+
+      if (do_flow_handler) {
+        pptrs->tcp_flags = FALSE;
 
-      ip_flow_handler(pptrs);
+        if (pptrs->l4_proto == IPPROTO_TCP) {
+          if (off_l4+TCPFlagOff+1 > caplen) {
+            Log(LOG_INFO, "INFO ( %s/core ): short IPv4 packet read (%u/%u/flows). Snaplen issue ?\n",
+          		config.name, caplen, off_l4+TCPFlagOff+1);
+            return FALSE;
+          }
+          if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_SYN) pptrs->tcp_flags |= TH_SYN;
+          if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_FIN) pptrs->tcp_flags |= TH_FIN;
+          if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_RST) pptrs->tcp_flags |= TH_RST;
+          if (((struct my_tcphdr *)pptrs->tlh_ptr)->th_flags & TH_ACK && pptrs->tcp_flags) pptrs->tcp_flags |= TH_ACK;
+        }
+
+        ip_flow_handler(pptrs);
+      }
     }
 
     /* XXX: optimize/short circuit here! */
@@ -194,6 +210,7 @@ int ip_handler(register struct packet_ptrs *pptrs)
         if (tunnel_registry[0][num].proto == pptrs->l4_proto) {
 	  if (!tunnel_registry[0][num].port || (pptrs->tlh_ptr && tunnel_registry[0][num].port == ntohs(((struct my_tlhdr *)pptrs->tlh_ptr)->dst_port))) {
 	    pptrs->tun_stack = num;
+	    pptrs->tunnel_level++;
 	    ret = (*tunnel_registry[0][num].tf)(pptrs);
 	  }
         }
@@ -202,6 +219,7 @@ int ip_handler(register struct packet_ptrs *pptrs)
     else if (pptrs->tun_stack) { 
       if (tunnel_registry[pptrs->tun_stack][pptrs->tun_layer].proto == pptrs->l4_proto) {
         if (!tunnel_registry[pptrs->tun_stack][pptrs->tun_layer].port || (pptrs->tlh_ptr && tunnel_registry[pptrs->tun_stack][pptrs->tun_layer].port == ntohs(((struct my_tlhdr *)pptrs->tlh_ptr)->dst_port))) {
+	  pptrs->tunnel_level++;
           ret = (*tunnel_registry[pptrs->tun_stack][pptrs->tun_layer].tf)(pptrs);
         }
       }

src/pkt_handlers.c

@@ -671,6 +671,12 @@ void evaluate_packet_handlers()
       primitives++;
     }
 
+    if (channels_list[index].aggregation_2 & COUNT_TCP_RETRANSMISSION) {
+      if (config.acct_type == ACCT_PM) channels_list[index].phandler[primitives] = tcp_retransmission_handler;
+      else primitives--;
+      primitives++;
+    }
+
     /* if cpptrs.num > 0 one or multiple custom primitives are defined */
     if (channels_list[index].plugin->cfg.cpptrs.num) {
       if (config.acct_type == ACCT_PM) {
@@ -1087,6 +1093,13 @@ void tcp_flags_handler(struct channels_list_entry *chptr, struct packet_ptrs *pp
   if (pptrs->l4_proto == IPPROTO_TCP) pdata->tcp_flags = pptrs->tcp_flags;
 }
 
+void tcp_retransmission_handler(struct channels_list_entry *chptr, struct packet_ptrs *pptrs, char **data)
+{
+  struct pkt_data *pdata = (struct pkt_data *) *data;
+
+  if (pptrs->l4_proto == IPPROTO_TCP) pdata->primitives.tcp_retransmission = pptrs->tcp_retransmission;
+}
+
 void counters_handler(struct channels_list_entry *chptr, struct packet_ptrs *pptrs, char **data)
 {
   struct pkt_data *pdata = (struct pkt_data *) *data;

src/pkt_handlers.h

@@ -52,6 +52,7 @@ EXT void dst_port_handler(struct channels_list_entry *, struct packet_ptrs *, ch
 EXT void ip_tos_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
 EXT void ip_proto_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
 EXT void tcp_flags_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
+EXT void tcp_retransmission_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
 EXT void counters_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
 EXT void counters_renormalize_handler(struct channels_list_entry *, struct packet_ptrs *, char **);
 EXT void post_tag_handler(struct channels_list_entry *, struct packet_ptrs *, char **);

src/pmacct-defines.h

@@ -209,7 +209,8 @@
 #define COUNT_INT_LABEL			0x0002000000008000ULL
 #define COUNT_INT_EXPORT_PROTO_SEQNO	0x0002000000010000ULL
 #define COUNT_INT_EXPORT_PROTO_VERSION  0x0002000000020000ULL
-#define COUNT_INT_CUSTOM_PRIMITIVES	0x0002000000040000ULL
+#define COUNT_INT_TCP_RETRANSMISSION	0x0002000000040000ULL
+#define COUNT_INT_CUSTOM_PRIMITIVES	0x0002000000080000ULL
 
 #define COUNT_INDEX_MASK	0xFFFF
 #define COUNT_INDEX_CP		0xFFFF000000000000ULL  /* index 0xffff reserved to custom primitives */
@@ -285,6 +286,7 @@
 #define COUNT_LABEL			(COUNT_INT_LABEL & COUNT_REGISTRY_MASK)
 #define COUNT_EXPORT_PROTO_SEQNO	(COUNT_INT_EXPORT_PROTO_SEQNO & COUNT_REGISTRY_MASK)
 #define COUNT_EXPORT_PROTO_VERSION	(COUNT_INT_EXPORT_PROTO_VERSION & COUNT_REGISTRY_MASK)
+#define COUNT_TCP_RETRANSMISSION        (COUNT_INT_TCP_RETRANSMISSION & COUNT_REGISTRY_MASK)
 #define COUNT_CUSTOM_PRIMITIVES		(COUNT_INT_CUSTOM_PRIMITIVES & COUNT_REGISTRY_MASK)
 /* PRIMITIVES DEFINITION: END */
 

src/print_plugin.c

@@ -737,6 +737,8 @@ void P_cache_purge(struct chained_cache *queue[], int index)
         }
 
         if (config.what_to_count & COUNT_IP_TOS) fprintf(f, "%-3u    ", data->tos);
+
+        if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) fprintf(f, "%-3u     ", data->tcp_retransmission);
 
   #if defined WITH_GEOIP
         if (config.what_to_count_2 & COUNT_SRC_HOST_COUNTRY) fprintf(f, "%-5s       ", GeoIP_code_by_id(data->src_ip_country.id));
@@ -1071,6 +1073,8 @@ void P_cache_purge(struct chained_cache *queue[], int index)
         }
 
         if (config.what_to_count & COUNT_IP_TOS) fprintf(f, "%s%u", write_sep(sep, &count), data->tos);
+
+        if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) fprintf(f, "%s%u", write_sep(sep, &count), data->tcp_retransmission);
 
   #if defined WITH_GEOIP
         if (config.what_to_count_2 & COUNT_SRC_HOST_COUNTRY) fprintf(f, "%s%s", write_sep(sep, &count), GeoIP_code_by_id(data->src_ip_country.id));
@@ -1320,6 +1324,7 @@ void P_write_stats_header_formatted(FILE *f, int is_event)
   if (config.what_to_count & COUNT_TCPFLAGS) fprintf(f, "TCP_FLAGS  ");
   if (config.what_to_count & COUNT_IP_PROTO) fprintf(f, "PROTOCOL    ");
   if (config.what_to_count & COUNT_IP_TOS) fprintf(f, "TOS    ");
+  if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) fprintf(f, "RETRANS ");
 #if defined (WITH_GEOIP) || (WITH_GEOIPV2)
   if (config.what_to_count_2 & COUNT_SRC_HOST_COUNTRY) fprintf(f, "SH_COUNTRY  ");
   if (config.what_to_count_2 & COUNT_DST_HOST_COUNTRY) fprintf(f, "DH_COUNTRY  ");
@@ -1425,6 +1430,7 @@ void P_write_stats_header_csv(FILE *f, int is_event)
   if (config.what_to_count & COUNT_TCPFLAGS) fprintf(f, "%sTCP_FLAGS", write_sep(sep, &count));
   if (config.what_to_count & COUNT_IP_PROTO) fprintf(f, "%sPROTOCOL", write_sep(sep, &count));
   if (config.what_to_count & COUNT_IP_TOS) fprintf(f, "%sTOS", write_sep(sep, &count));
+  if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) fprintf(f, "%sRETRANS", write_sep(sep, &count));
 #if defined (WITH_GEOIP) || defined (WITH_GEOIPV2)
   if (config.what_to_count_2 & COUNT_SRC_HOST_COUNTRY) fprintf(f, "%sSH_COUNTRY", write_sep(sep, &count));
   if (config.what_to_count_2 & COUNT_DST_HOST_COUNTRY) fprintf(f, "%sDH_COUNTRY", write_sep(sep, &count));

src/sql_common.c

@@ -1247,6 +1247,7 @@ int sql_evaluate_primitives(int primitive)
     if (config.what_to_count_2 & COUNT_EXPORT_PROTO_SEQNO) what_to_count_2 |= COUNT_EXPORT_PROTO_SEQNO;
     if (config.what_to_count_2 & COUNT_EXPORT_PROTO_VERSION) what_to_count_2 |= COUNT_EXPORT_PROTO_VERSION;
     if (config.what_to_count_2 & COUNT_LABEL) what_to_count_2 |= COUNT_LABEL;
+    if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) what_to_count_2 |= COUNT_TCP_RETRANSMISSION;
   }
 
   /* sorting out delimiter */
@@ -2579,6 +2580,34 @@ int sql_evaluate_primitives(int primitive)
     primitive++;
   }
 
+  if (what_to_count_2 & COUNT_TCP_RETRANSMISSION) {
+    int count_it = FALSE;
+
+    if ((config.sql_table_version < 3 || config.sql_table_version >= SQL_TABLE_VERSION_BGP) && !assume_custom_table) {
+      if (config.what_to_count_2 & COUNT_TCP_RETRANSMISSION) {
+        Log(LOG_ERR, "ERROR ( %s/%s ): TCP Retransmission accounting not supported for selected sql_table_version/_type."
+            " Read about SQL table versioning or consider using sql_optimize_clauses.\n", config.name, config.type);
+        exit_plugin(1);
+      }
+      else what_to_count_2 ^= COUNT_TCP_RETRANSMISSION;
+    }
+    else count_it = TRUE;
+
+    if (count_it) {
+      if (primitive) {
+        strncat(insert_clause, ", ", SPACELEFT(insert_clause));
+        strncat(values[primitive].string, delim_buf, SPACELEFT(values[primitive].string));
+        strncat(where[primitive].string, " AND ", SPACELEFT(where[primitive].string));
+      }
+      strncat(insert_clause, "tcp_retransmission", SPACELEFT(insert_clause));
+      strncat(values[primitive].string, "%u", SPACELEFT(values[primitive].string));
+      strncat(where[primitive].string, "tcp_retransmission=%u", SPACELEFT(where[primitive].string));
+      values[primitive].type = where[primitive].type = COUNT_INT_TCP_RETRANSMISSION;
+      values[primitive].handler = where[primitive].handler = count_tcp_retransmission_handler;
+      primitive++;
+    }
+  }
+
   /* all custom primitives printed here */
   {
     struct custom_primitive_ptrs *cp_entry;

src/sql_common.h

@@ -266,6 +266,7 @@ EXT void PG_copy_count_timestamp_max_handler(const struct db_cache *, struct ins
 EXT void count_export_proto_seqno_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
 EXT void count_export_proto_version_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
 EXT void count_timestamp_max_residual_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
+EXT void count_tcp_retransmission_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
 EXT void count_custom_primitives_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
 EXT void fake_mac_handler(const struct db_cache *, struct insert_data *, int, char **, char **);
 EXT void fake_host_handler(const struct db_cache *, struct insert_data *, int, char **, char **);

src/sql_handlers.c

@@ -475,6 +475,14 @@ void count_export_proto_version_handler(const struct db_cache *cache_elem, struc
   *ptr_values += strlen(*ptr_values);
 }
 
+void count_tcp_retransmission_handler(const struct db_cache *cache_elem, struct insert_data *idata, int num, char **ptr_values, char **ptr_where)
+{
+  snprintf(*ptr_where, SPACELEFT(where_clause), where[num].string, cache_elem->primitives.tcp_retransmission);
+  snprintf(*ptr_values, SPACELEFT(values_clause), values[num].string, cache_elem->primitives.tcp_retransmission);
+  *ptr_where += strlen(*ptr_where);
+  *ptr_values += strlen(*ptr_values);
+}
+
 void count_custom_primitives_handler(const struct db_cache *cache_elem, struct insert_data *idata, int num, char **ptr_values, char **ptr_where)
 {
   struct custom_primitive_ptrs *cp_entry;

src/util.c

@@ -2076,6 +2076,12 @@ void *compose_json(u_int64_t wtc, u_int64_t wtc_2, u_int8_t flow_type, struct pk
     json_decref(kv);
   }
 
+  if (wtc_2 & COUNT_TCP_RETRANSMISSION) {
+    kv = json_pack("{sI}", "retrans", pbase->tcp_retransmission);
+    json_object_update_missing(obj, kv);
+    json_decref(kv);
+  }
+
   /* all custom primitives printed here */
   {
     int cp_idx;