Add more flow lock assertions to the debug validation code.

src/app-layer-parser.c

@@ -82,6 +82,8 @@ static uint32_t al_result_pool_elmts = 0;
 FileContainer *AppLayerGetFilesFromFlow(Flow *f, uint8_t direction) {
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     uint16_t alproto = f->alproto;
 
     if (alproto == ALPROTO_UNKNOWN)
@@ -709,6 +711,8 @@ static int AppLayerDoParse(void *local_data, Flow *f,
                            uint16_t proto)
 {
     SCEnter();
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     int retval = 0;
     AppLayerParserResult result = { NULL, NULL, 0 };
 
@@ -1016,6 +1020,8 @@ error:
 int AppLayerTransactionGetBaseId(Flow *f) {
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     AppLayerParserStateStore *parser_state_store =
         (AppLayerParserStateStore *)f->alparser;
 

src/app-layer.c

@@ -40,7 +40,7 @@
 extern uint8_t engine_mode;
 
 /** \brief Get the active app layer proto from the packet
- *  \param p packet pointer
+ *  \param p packet pointer with a LOCKED flow
  *  \retval alstate void pointer to the state
  *  \retval proto (ALPROTO_UNKNOWN if no proto yet) */
 uint16_t AppLayerGetProtoFromPacket(Packet *p) {
@@ -50,6 +50,8 @@ uint16_t AppLayerGetProtoFromPacket(Packet *p) {
         SCReturnUInt(ALPROTO_UNKNOWN);
     }
 
+    DEBUG_ASSERT_FLOW_LOCKED(p->flow);
+
     SCLogDebug("p->flow->alproto %"PRIu16"", p->flow->alproto);
 
     SCReturnUInt(p->flow->alproto);
@@ -66,6 +68,8 @@ void *AppLayerGetProtoStateFromPacket(Packet *p) {
         SCReturnPtr(NULL, "void");
     }
 
+    DEBUG_ASSERT_FLOW_LOCKED(p->flow);
+
     SCLogDebug("p->flow->alproto %"PRIu16"", p->flow->alproto);
 
     SCLogDebug("p->flow %p", p->flow);

src/detect-app-layer-event.c

@@ -76,7 +76,9 @@ int DetectAppLayerEventMatch(ThreadVars *t, DetectEngineThreadCtx *det_ctx,
 
     DetectAppLayerEventData *aled = (DetectAppLayerEventData *)m->ctx;
 
+    SCMutexLock(&f->m);
     AppLayerDecoderEvents *decoder_events = AppLayerGetDecoderEventsForFlow(f);
+    SCMutexUnlock(&f->m);
     if (decoder_events == NULL) {
         SCReturnInt(0);
     }

src/detect.c

@@ -1055,6 +1055,8 @@ SigGroupHead *SigMatchSignaturesGetSgh(DetectEngineCtx *de_ctx, DetectEngineThre
 static StreamMsg *SigMatchSignaturesGetSmsg(Flow *f, Packet *p, uint8_t flags) {
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     StreamMsg *smsg = NULL;
 
     if (p->proto == IPPROTO_TCP && f->protoctx != NULL) {

src/flow-timeout.c

@@ -406,8 +406,14 @@ int FlowForceReassemblyForFlowV2(Flow *f)
  * \internal
  * \brief Forces reassembly for flows that need it.
  *
- *        Please note we don't use locks anywhere.  This function is to be
- *        called right when the engine is not doing anything.
+ * When this function is called we're running in virtually dead engine,
+ * so locking the flows is not strictly required. The reasons it is still
+ * done are:
+ * - code consistency
+ * - silence complaining profilers
+ * - allow us to aggressively check using debug valdation assertions
+ * - be robust in case of future changes
+ * - locking overhead if neglectable when no other thread fights us
  *
  * \param q The queue to process flows from.
  */
@@ -417,9 +423,7 @@ static inline void FlowForceReassemblyForQ(FlowQueue *q)
     TcpSession *ssn;
     int client_ok;
     int server_ok;
-
-    /* no locks needed, since the engine is virtually dead.
-     * We are the kings here */
+    int tcp_needs_inspection;
 
     /* get the topmost flow from the QUEUE */
     f = q->top;
@@ -431,6 +435,8 @@ static inline void FlowForceReassemblyForQ(FlowQueue *q)
 
     /* we need to loop through all the flows in the queue */
     while (f != NULL) {
+        SCMutexLock(&f->m);
+
         PACKET_RECYCLE(reassemble_p);
 
         /* Get the tcp session for the flow */
@@ -438,6 +444,7 @@ static inline void FlowForceReassemblyForQ(FlowQueue *q)
 
         /* \todo Also skip flows that shouldn't be inspected */
         if (ssn == NULL) {
+            SCMutexUnlock(&f->m);
             f = f->lnext;
             continue;
         }
@@ -469,11 +476,20 @@ static inline void FlowForceReassemblyForQ(FlowQueue *q)
             StreamTcpReassembleProcessAppLayer(stt->ra_ctx);
         }
 
+        if (ssn->state >= TCP_ESTABLISHED && ssn->state != TCP_CLOSED)
+            tcp_needs_inspection = 1;
+        else
+            tcp_needs_inspection = 0;
+
+        SCMutexUnlock(&f->m);
+
         /* insert a pseudo packet in the toserver direction */
-        if (client_ok ||
-                (ssn->state >= TCP_ESTABLISHED && ssn->state != TCP_CLOSED))
+        if (client_ok || tcp_needs_inspection)
         {
+            SCMutexLock(&f->m);
             Packet *p = FlowForceReassemblyPseudoPacketGet(0, f, ssn, 1);
+            SCMutexUnlock(&f->m);
+
             if (p == NULL) {
                 TmqhOutputPacketpool(NULL, reassemble_p);
                 return;
@@ -496,10 +512,12 @@ static inline void FlowForceReassemblyForQ(FlowQueue *q)
                 }
             }
         } /* if (ssn->client.seg_list != NULL) */
-        if (server_ok ||
-                (ssn->state >= TCP_ESTABLISHED && ssn->state != TCP_CLOSED))
+        if (server_ok || tcp_needs_inspection)
         {
+            SCMutexLock(&f->m);
             Packet *p = FlowForceReassemblyPseudoPacketGet(1, f, ssn, 1);
+            SCMutexUnlock(&f->m);
+
             if (p == NULL) {
                 TmqhOutputPacketpool(NULL, reassemble_p);
                 return;

src/stream-tcp.c

@@ -67,6 +67,7 @@
 #include "util-privs.h"
 #include "util-profiling.h"
 #include "util-misc.h"
+#include "util-validate.h"
 
 //#define DEBUG
 
@@ -3594,6 +3595,8 @@ static int StreamTcpPacket (ThreadVars *tv, Packet *p, StreamTcpThread *stt,
 {
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(p->flow);
+
     SCLogDebug("p->pcap_cnt %"PRIu64, p->pcap_cnt);
 
     TcpSession *ssn = (TcpSession *)p->flow->protoctx;

src/util-file.c

@@ -589,6 +589,8 @@ void FileDisableStoring(Flow *f, uint8_t direction) {
 
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     f->flags |= FLOW_FILE_NO_STORE;
 
     FileContainer *ffc = AppLayerGetFilesFromFlow(f, direction);
@@ -613,6 +615,8 @@ void FileDisableMagic(Flow *f, uint8_t direction) {
 
     SCEnter();
 
+    DEBUG_ASSERT_FLOW_LOCKED(f);
+
     f->flags |= FLOW_FILE_NO_MAGIC;
 
     FileContainer *ffc = AppLayerGetFilesFromFlow(f, direction);