flow: add callbacks for flow init and flow updates

Adds user registerable callbacks for flow initialization, flow
update and flow finish.

Some plugins, such as other DPI libraries like nDPI need a way to hook
into these flow lifecycle events.

Ticket: #7319
Ticket: #7320

src/Makefile.am

@@ -317,6 +317,7 @@ noinst_HEADERS = \
 	feature.h \
 	flow-bit.h \
 	flow-bypass.h \
+	flow-callbacks.h \
 	flow.h \
 	flow-hash.h \
 	flow-manager.h \
@@ -875,6 +876,7 @@ libsuricata_c_a_SOURCES = \
 	feature.c \
 	flow-bit.c \
 	flow-bypass.c \
+	flow-callbacks.c \
 	flow.c \
 	flow-hash.c \
 	flow-manager.c \

src/flow-callbacks.c

@@ -0,0 +1,129 @@
+/* Copyright (C) 2024 Open Information Security Foundation
+ *
+ * You can copy, redistribute or modify this Program under the terms of
+ * the GNU General Public License version 2 as published by the Free
+ * Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301, USA.
+ */
+
+#include "flow-callbacks.h"
+
+typedef struct FlowInitCallback_ {
+    SCFlowInitCallbackFn Callback;
+    void *user;
+    struct FlowInitCallback_ *next;
+} FlowInitCallback;
+
+static FlowInitCallback *init_callbacks = NULL;
+
+typedef struct FlowUpdateCallback_ {
+    SCFlowUpdateCallbackFn Callback;
+    void *user;
+    struct FlowUpdateCallback_ *next;
+} FlowUpdateCallback;
+
+static FlowUpdateCallback *update_callbacks = NULL;
+
+typedef struct FlowFinishCallback_ {
+    SCFlowFinishCallbackFn Callback;
+    void *user;
+    struct FlowFinishCallback_ *next;
+} FlowFinishCallback;
+
+static FlowFinishCallback *finish_callbacks = NULL;
+
+bool SCFlowRegisterInitCallback(SCFlowInitCallbackFn fn, void *user)
+{
+    FlowInitCallback *cb = SCCalloc(1, sizeof(*cb));
+    if (cb == NULL) {
+        return false;
+    }
+    cb->Callback = fn;
+    cb->user = user;
+    if (init_callbacks == NULL) {
+        init_callbacks = cb;
+    } else {
+        FlowInitCallback *current = init_callbacks;
+        while (current->next != NULL) {
+            current = current->next;
+        }
+        current->next = cb;
+    }
+    return true;
+}
+
+void SCFlowRunInitCallbacks(ThreadVars *tv, Flow *f, const Packet *p)
+{
+    FlowInitCallback *cb = init_callbacks;
+    while (cb != NULL) {
+        cb->Callback(tv, f, p, cb->user);
+        cb = cb->next;
+    }
+}
+
+bool SCFlowRegisterUpdateCallback(SCFlowUpdateCallbackFn fn, void *user)
+{
+    FlowUpdateCallback *cb = SCCalloc(1, sizeof(*cb));
+    if (cb == NULL) {
+        return false;
+    }
+    cb->Callback = fn;
+    cb->user = user;
+    if (update_callbacks == NULL) {
+        update_callbacks = cb;
+    } else {
+        FlowUpdateCallback *current = update_callbacks;
+        while (current->next != NULL) {
+            current = current->next;
+        }
+        current->next = cb;
+    }
+    return true;
+}
+
+void SCFlowRunUpdateCallbacks(ThreadVars *tv, Flow *f, Packet *p)
+{
+    FlowUpdateCallback *cb = update_callbacks;
+    while (cb != NULL) {
+        cb->Callback(tv, f, p, cb->user);
+        cb = cb->next;
+    }
+}
+
+bool SCFlowRegisterFinishCallback(SCFlowFinishCallbackFn fn, void *user)
+{
+    FlowFinishCallback *cb = SCCalloc(1, sizeof(*cb));
+    if (cb == NULL) {
+        return false;
+    }
+    cb->Callback = fn;
+    cb->user = user;
+    if (finish_callbacks == NULL) {
+        finish_callbacks = cb;
+    } else {
+        FlowFinishCallback *current = finish_callbacks;
+        while (current->next != NULL) {
+            current = current->next;
+        }
+        current->next = cb;
+    }
+    return true;
+}
+
+void SCFlowRunFinishCallbacks(ThreadVars *tv, Flow *f)
+{
+    FlowFinishCallback *cb = finish_callbacks;
+    while (cb != NULL) {
+        cb->Callback(tv, f, cb->user);
+        cb = cb->next;
+    }
+}

src/flow-callbacks.h

@@ -0,0 +1,121 @@
+/* Copyright (C) 2024 Open Information Security Foundation
+ *
+ * You can copy, redistribute or modify this Program under the terms of
+ * the GNU General Public License version 2 as published by the Free
+ * Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301, USA.
+ */
+
+#ifndef SURICATA_FLOW_CALLBACKS_H
+#define SURICATA_FLOW_CALLBACKS_H
+
+#include "suricata-common.h"
+#include "flow.h"
+
+/** \brief Function type for flow initialization callbacks.
+ *
+ * Once registered with SCFlowRegisterInitCallback, this function will
+ * be called every time a flow is initialized, or in other words,
+ * every time Suricata picks up a flow.
+ *
+ * \param tv The ThreadVars data structure for the thread creating the
+ *     flow.
+ * \param f The newly initialized flow.
+ * \param p The packet related to creating the new flow.
+ * \param user The user data provided during callback registration.
+ */
+typedef void (*SCFlowInitCallbackFn)(ThreadVars *tv, Flow *f, const Packet *p, void *user);
+
+/** \brief Register a flow init callback.
+ *
+ * Register a user provided function to be called every time a flow is
+ * initialized for use.
+ *
+ * \param fn Pointer to function to be called
+ * \param user Additional user data to be passed to callback
+ *
+ * \returns true if callback was registered, otherwise false if the
+ *     callback could not be registered due to memory allocation error.
+ */
+bool SCFlowRegisterInitCallback(SCFlowInitCallbackFn fn, void *user);
+
+/** \internal
+ *
+ * Run all registered flow init callbacks.
+ */
+void SCFlowRunInitCallbacks(ThreadVars *tv, Flow *f, const Packet *p);
+
+/** \brief Function type for flow update callbacks.
+ *
+ * Once registered with SCFlowRegisterUpdateCallback, this function
+ * will be called every time a flow is updated by a packet (basically
+ * everytime a packet is seen on a flow).
+ *
+ * \param tv The ThreadVars data structure for the thread updating the
+ *     flow.
+ * \param f The flow being updated.
+ * \param p The packet responsible for the flow update.
+ * \param user The user data provided during callback registration.
+ */
+typedef void (*SCFlowUpdateCallbackFn)(ThreadVars *tv, Flow *f, Packet *p, void *user);
+
+/** \brief Register a flow update callback.
+ *
+ * Register a user provided function to be called everytime a flow is
+ * updated.
+ *
+ * \param fn Pointer to function to be called
+ * \param user Additional user data to be passed to callback
+ *
+ * \returns true if callback was registered, otherwise false if the
+ *     callback could not be registered due to memory allocation error.
+ */
+bool SCFlowRegisterUpdateCallback(SCFlowUpdateCallbackFn fn, void *user);
+
+/** \internal
+ *
+ * Run all registered flow update callbacks.
+ */
+void SCFlowRunUpdateCallbacks(ThreadVars *tv, Flow *f, Packet *p);
+
+/** \brief Function type for flow finish callbacks.
+ *
+ * Once registered with SCFlowRegisterFinshCallback, this function
+ * will be called when Suricata is done with a flow.
+ *
+ * \param tv The ThreadVars data structure for the thread finishing
+ *     the flow.
+ * \param f The flow being finshed.
+ * \param user The user data provided during callback registration.
+ */
+typedef void (*SCFlowFinishCallbackFn)(ThreadVars *tv, Flow *f, void *user);
+
+/** \brief Register a flow init callback.
+ *
+ * Register a user provided function to be called every time a flow is
+ * finished.
+ *
+ * \param fn Pointer to function to be called
+ * \param user Additional user data to be passed to callback
+ *
+ * \returns true if callback was registered, otherwise false if the
+ *     callback could not be registered due to memory allocation error.
+ */
+bool SCFlowRegisterFinishCallback(SCFlowFinishCallbackFn fn, void *user);
+
+/** \internal
+ *
+ * Run all registered flow init callbacks.
+ */
+void SCFlowRunFinishCallbacks(ThreadVars *tv, Flow *f);
+
+#endif /* SURICATA_FLOW_CALLBACKS_H */

src/flow-hash.c

@@ -38,6 +38,7 @@
 #include "flow-storage.h"
 #include "flow-timeout.h"
 #include "flow-spare-pool.h"
+#include "flow-callbacks.h"
 #include "app-layer-parser.h"
 
 #include "util-time.h"
@@ -781,7 +782,7 @@ static Flow *TcpReuseReplace(ThreadVars *tv, FlowLookupStruct *fls, FlowBucket *
     fb->head = f;
 
     /* initialize and return */
-    FlowInit(f, p);
+    FlowInit(tv, f, p);
     f->flow_hash = hash;
     f->fb = fb;
     FlowUpdateState(f, FLOW_STATE_NEW);
@@ -886,7 +887,7 @@ Flow *FlowGetFlowFromHash(ThreadVars *tv, FlowLookupStruct *fls, Packet *p, Flow
         fb->head = f;
 
         /* got one, now lock, initialize and return */
-        FlowInit(f, p);
+        FlowInit(tv, f, p);
         f->flow_hash = hash;
         f->fb = fb;
         FlowUpdateState(f, FLOW_STATE_NEW);
@@ -951,7 +952,7 @@ flow_removed:
             fb->head = f;
 
             /* initialize and return */
-            FlowInit(f, p);
+            FlowInit(tv, f, p);
             f->flow_hash = hash;
             f->fb = fb;
             FlowUpdateState(f, FLOW_STATE_NEW);
@@ -1242,6 +1243,7 @@ static Flow *FlowGetUsedFlow(ThreadVars *tv, DecodeThreadVars *dtv, const SCTime
         }
 #endif
 
+        SCFlowRunFinishCallbacks(tv, f);
         FlowClearMemory(f, f->protomap);
 
         /* leave locked */

src/flow-manager.c

@@ -39,6 +39,7 @@
 #include "flow-manager.h"
 #include "flow-storage.h"
 #include "flow-spare-pool.h"
+#include "flow-callbacks.h"
 
 #include "stream-tcp.h"
 #include "stream-tcp-cache.h"
@@ -1059,7 +1060,7 @@ static void Recycler(ThreadVars *tv, FlowRecyclerThreadData *ftd, Flow *f)
         StatsDecr(tv, ftd->counter_tcp_active_sessions);
     }
     StatsDecr(tv, ftd->counter_flow_active);
-
+    SCFlowRunFinishCallbacks(tv, f);
     FlowClearMemory(f, f->protomap);
     FLOWLOCK_UNLOCK(f);
 }

src/flow-util.c

@@ -29,6 +29,7 @@
 #include "flow.h"
 #include "flow-private.h"
 #include "flow-util.h"
+#include "flow-callbacks.h"
 #include "flow-var.h"
 #include "app-layer.h"
 
@@ -142,7 +143,7 @@ static inline void FlowSetICMPv6CounterPart(Flow *f)
 
 /* initialize the flow from the first packet
  * we see from it. */
-void FlowInit(Flow *f, const Packet *p)
+void FlowInit(ThreadVars *tv, Flow *f, const Packet *p)
 {
     SCEnter();
     SCLogDebug("flow %p", f);
@@ -203,6 +204,8 @@ void FlowInit(Flow *f, const Packet *p)
         FlowSetStorageById(f, MacSetGetFlowStorageID(), ms);
     }
 
+    SCFlowRunInitCallbacks(tv, f, p);
+
     SCReturn;
 }
 

src/flow-util.h

@@ -140,7 +140,7 @@
 Flow *FlowAlloc(void);
 void FlowFree(Flow *);
 uint8_t FlowGetProtoMapping(uint8_t);
-void FlowInit(Flow *, const Packet *);
+void FlowInit(ThreadVars *, Flow *, const Packet *);
 uint8_t FlowGetReverseProtoMapping(uint8_t rproto);
 
 /* flow end counter logic */

src/flow.c

@@ -44,6 +44,7 @@
 #include "flow-storage.h"
 #include "flow-bypass.h"
 #include "flow-spare-pool.h"
+#include "flow-callbacks.h"
 
 #include "stream-tcp-private.h"
 
@@ -503,6 +504,8 @@ void FlowHandlePacketUpdate(Flow *f, Packet *p, ThreadVars *tv, DecodeThreadVars
         SCLogDebug("setting FLOW_NOPAYLOAD_INSPECTION flag on flow %p", f);
         DecodeSetNoPayloadInspectionFlag(p);
     }
+
+    SCFlowRunUpdateCallbacks(tv, f, p);
 }
 
 /** \brief Entry point for packet flow handling