From: Eric Leblond <eric@regit.org>
Date: Sun, 24 Mar 2019 18:47:02 +0000 (+0100)
Subject: bypass: use flow storage for bypass counter
X-Git-Tag: suricata-5.0.0-rc1~323
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6ab1cbcb8ef87ba6df05a14e045170088ab2c19b;p=thirdparty%2Fsuricata.git

bypass: use flow storage for bypass counter

There is a synchronization issue occuring when a flow is
added to the eBPF bypass maps. The flow can have packets
in the ring buffer that have already passed the eBPF stage.
By consequences, they are not accounted in the eBPF counter
but are accounted by Suricata flow engine.

This was causing counters to be completely wrong. This code
fixes the issue by avoiding the counter change in invalid
case.

To avoid adding 4 64bits integers to the Flow structure for the
bypass accounting, we use instead a FlowStorage. This limits the
memory usage to the size of a pointer.
---

diff --git a/src/decode.c b/src/decode.c
index b19365d054..77c4199044 100644
--- a/src/decode.c
+++ b/src/decode.c
@@ -66,6 +66,7 @@
 #include "util-hash-string.h"
 #include "output.h"
 #include "output-flow.h"
+#include "flow-storage.h"
 
 extern bool stats_decoder_events;
 const char *stats_decoder_events_prefix;
@@ -407,6 +408,12 @@ void PacketBypassCallback(Packet *p)
         return;
     }
 
+    FlowCounters *fc = SCCalloc(sizeof(FlowCounters), 1);
+    if (fc) {
+        FlowSetStorageById(p->flow, GetFlowBypassCounterID(), fc);
+    } else {
+        return;
+    }
     if (p->BypassPacketsFlow && p->BypassPacketsFlow(p)) {
         FlowUpdateState(p->flow, FLOW_STATE_CAPTURE_BYPASSED);
     } else {
diff --git a/src/flow-util.c b/src/flow-util.c
index 0dc08a6d57..ed95cfc261 100644
--- a/src/flow-util.c
+++ b/src/flow-util.c
@@ -203,3 +203,22 @@ void FlowInit(Flow *f, const Packet *p)
 
     SCReturn;
 }
+
+int g_bypass_counter_id = -1;
+
+int GetFlowBypassCounterID(void)
+{
+    return g_bypass_counter_id;
+}
+
+static void FlowCounterFree(void *x)
+{
+    if (x)
+        SCFree(x);
+}
+
+void RegisterFlowBypassCounter(void)
+{
+    g_bypass_counter_id = FlowStorageRegister("bypass_counters", sizeof(void *),
+                                              NULL, FlowCounterFree);
+}
diff --git a/src/flow.h b/src/flow.h
index 50951e5970..19a9229917 100644
--- a/src/flow.h
+++ b/src/flow.h
@@ -482,6 +482,13 @@ typedef struct FlowProtoFreeFunc_ {
     void (*Freefunc)(void *);
 } FlowProtoFreeFunc;
 
+typedef struct FlowCounters_ {
+    uint64_t tosrcpktcnt;
+    uint64_t tosrcbytecnt;
+    uint64_t todstpktcnt;
+    uint64_t todstbytecnt;
+} FlowCounters;
+
 /** \brief prepare packet for a life with flow
  *  Set PKT_WANTS_FLOW flag to incidate workers should do a flow lookup
  *  and calc the hash value to be used in the lookup and autofp flow
@@ -522,6 +529,9 @@ int FlowSetMemcap(uint64_t size);
 uint64_t FlowGetMemcap(void);
 uint64_t FlowGetMemuse(void);
 
+int GetFlowBypassCounterID(void);
+void RegisterFlowBypassCounter(void);
+
 /** ----- Inline functions ----- */
 
 /** \brief Set the No Packet Inspection Flag without locking the flow.
diff --git a/src/output-json-flow.c b/src/output-json-flow.c
index 09f2a7ccee..ed1db7a76f 100644
--- a/src/output-json-flow.c
+++ b/src/output-json-flow.c
@@ -48,6 +48,7 @@
 #include "output-json-flow.h"
 
 #include "stream-tcp-private.h"
+#include "flow-storage.h"
 
 #ifdef HAVE_LIBJANSSON
 
@@ -199,14 +200,38 @@ void JsonAddFlow(Flow *f, json_t *js, json_t *hjs)
                 json_string(AppProtoToString(f->alproto_expect)));
     }
 
-    json_object_set_new(hjs, "pkts_toserver",
-            json_integer(f->todstpktcnt));
-    json_object_set_new(hjs, "pkts_toclient",
-            json_integer(f->tosrcpktcnt));
-    json_object_set_new(hjs, "bytes_toserver",
-            json_integer(f->todstbytecnt));
-    json_object_set_new(hjs, "bytes_toclient",
-            json_integer(f->tosrcbytecnt));
+    FlowCounters *fc = FlowGetStorageById(f, GetFlowBypassCounterID());
+    if (fc) {
+        json_object_set_new(hjs, "pkts_toserver",
+                json_integer(f->todstpktcnt + fc->todstpktcnt));
+        json_object_set_new(hjs, "pkts_toclient",
+                json_integer(f->tosrcpktcnt + fc->tosrcpktcnt));
+        json_object_set_new(hjs, "bytes_toserver",
+                json_integer(f->todstbytecnt + fc->todstbytecnt));
+        json_object_set_new(hjs, "bytes_toclient",
+                json_integer(f->tosrcbytecnt + fc->tosrcbytecnt));
+        json_t *bhjs = json_object();
+        if (bhjs != NULL) {
+            json_object_set_new(bhjs, "pkts_toserver",
+                    json_integer(fc->todstpktcnt));
+            json_object_set_new(bhjs, "pkts_toclient",
+                    json_integer(fc->tosrcpktcnt));
+            json_object_set_new(bhjs, "bytes_toserver",
+                    json_integer(fc->todstbytecnt));
+            json_object_set_new(bhjs, "bytes_toclient",
+                    json_integer(fc->tosrcbytecnt));
+            json_object_set_new(hjs, "bypassed", bhjs);
+        }
+    } else {
+        json_object_set_new(hjs, "pkts_toserver",
+                json_integer(f->todstpktcnt));
+        json_object_set_new(hjs, "pkts_toclient",
+                json_integer(f->tosrcpktcnt));
+        json_object_set_new(hjs, "bytes_toserver",
+                json_integer(f->todstbytecnt));
+        json_object_set_new(hjs, "bytes_toclient",
+                json_integer(f->tosrcbytecnt));
+    }
 
     char timebuf1[64];
     CreateIsoTimeString(&f->startts, timebuf1, sizeof(timebuf1));
diff --git a/src/source-af-packet.c b/src/source-af-packet.c
index 19bb12d027..94aa68bece 100644
--- a/src/source-af-packet.c
+++ b/src/source-af-packet.c
@@ -2307,13 +2307,8 @@ static int AFPInsertHalfFlow(int mapd, void *key, uint32_t hash,
     }
 
     /* We use a per CPU structure so we have to set an array of values as the kernel
-     * is not duplicating the data on each CPU by itself. We set the first entry to
-     * the actual flow pkts and bytes count as we need to continue from actual point
-     * to detect an absence of packets in the future. */
-    BPF_PERCPU(value,0).packets = pkts_cnt;
-    BPF_PERCPU(value,0).bytes = bytes_cnt;
-    BPF_PERCPU(value,0).hash = hash;
-    for (i = 1; i < nr_cpus; i++) {
+     * is not duplicating the data on each CPU by itself. */
+    for (i = 0; i < nr_cpus; i++) {
         BPF_PERCPU(value, i).packets = 0;
         BPF_PERCPU(value, i).bytes = 0;
         BPF_PERCPU(value, i).hash = hash;
diff --git a/src/suricata.c b/src/suricata.c
index c5d6d22b6a..514f1207f0 100644
--- a/src/suricata.c
+++ b/src/suricata.c
@@ -2705,6 +2705,7 @@ static int PostConfLoadedSetup(SCInstance *suri)
     EBPFRegisterExtension();
     LiveDevRegisterExtension();
 #endif
+    RegisterFlowBypassCounter();
     AppLayerSetup();
 
     /* Suricata will use this umask if provided. By default it will use the
diff --git a/src/util-ebpf.c b/src/util-ebpf.c
index 149af41828..10a07b7c74 100644
--- a/src/util-ebpf.c
+++ b/src/util-ebpf.c
@@ -508,12 +508,20 @@ static int EBPFCreateFlowForKey(struct flows_stats *flowstats, FlowKey *flow_key
      * server then if we already have something in to server to client. We need
      * these numbers as we will use it to see if we have new traffic coming
      * on the flow */
-    if (f->todstbytecnt == 0) {
-        f->todstpktcnt = pkts_cnt;
-        f->todstbytecnt = bytes_cnt;
+    FlowCounters *fc = FlowGetStorageById(f, GetFlowBypassCounterID());
+    if (fc == NULL) {
+        fc = SCCalloc(sizeof(FlowCounters), 1);
+        if (fc) {
+            FlowSetStorageById(f, GetFlowBypassCounterID(), fc);
+            fc->todstpktcnt = pkts_cnt;
+            fc->todstbytecnt = bytes_cnt;
+        } else {
+            FLOWLOCK_UNLOCK(f);
+            return 0;
+        }
     } else {
-        f->tosrcpktcnt = pkts_cnt;
-        f->tosrcbytecnt = bytes_cnt;
+        fc->tosrcpktcnt = pkts_cnt;
+        fc->tosrcbytecnt = bytes_cnt;
     }
     FLOWLOCK_UNLOCK(f);
     return 0;
@@ -527,22 +535,27 @@ static int EBPFUpdateFlowForKey(struct flows_stats *flowstats, FlowKey *flow_key
     if (f != NULL) {
         SCLogDebug("bypassed flow found %d -> %d, doing accounting",
                     f->sp, f->dp);
+        FlowCounters *fc = FlowGetStorageById(f, GetFlowBypassCounterID());
+        if (fc == NULL) {
+            FLOWLOCK_UNLOCK(f);
+            return 0;
+        }
         if (flow_key->sp == f->sp) {
-            if (pkts_cnt != f->todstpktcnt) {
-                flowstats->packets += pkts_cnt - f->todstpktcnt;
-                flowstats->bytes += bytes_cnt - f->todstbytecnt;
-                f->todstpktcnt = pkts_cnt;
-                f->todstbytecnt = bytes_cnt;
+            if (pkts_cnt != fc->todstpktcnt) {
+                flowstats->packets += pkts_cnt - fc->todstpktcnt;
+                flowstats->bytes += bytes_cnt - fc->todstbytecnt;
+                fc->todstpktcnt = pkts_cnt;
+                fc->todstbytecnt = bytes_cnt;
                 /* interval based so no meaning to update the millisecond.
                  * Let's keep it fast and simple */
                 f->lastts.tv_sec = ctime->tv_sec;
             }
         } else {
-            if (pkts_cnt != f->tosrcpktcnt) {
-                flowstats->packets += pkts_cnt - f->tosrcpktcnt;
-                flowstats->bytes += bytes_cnt - f->tosrcbytecnt;
-                f->tosrcpktcnt = pkts_cnt;
-                f->tosrcbytecnt = bytes_cnt;
+            if (pkts_cnt != fc->tosrcpktcnt) {
+                flowstats->packets += pkts_cnt - fc->tosrcpktcnt;
+                flowstats->bytes += bytes_cnt - fc->tosrcbytecnt;
+                fc->tosrcpktcnt = pkts_cnt;
+                fc->tosrcbytecnt = bytes_cnt;
                 f->lastts.tv_sec = ctime->tv_sec;
             }
         }