Inspect the reassembled stream together with the packet payload in the same direction.

src/app-layer.c

@@ -52,7 +52,7 @@ uint16_t AppLayerGetProtoFromPacket(Packet *p) {
 }
 
 /** \brief Get the active app layer state from the packet
- *  \param p packet pointer
+ *  \param p packet pointer with a LOCKED flow
  *  \retval alstate void pointer to the state
  *  \retval NULL in case we have no state */
 void *AppLayerGetProtoStateFromPacket(Packet *p) {
@@ -76,7 +76,7 @@ void *AppLayerGetProtoStateFromPacket(Packet *p) {
 }
 
 /** \brief Get the active app layer state from the flow
- *  \param f flow pointer
+ *  \param f flow pointer to a LOCKED flow
  *  \retval alstate void pointer to the state
  *  \retval NULL in case we have no state */
 void *AppLayerGetProtoStateFromFlow(Flow *f) {
@@ -182,23 +182,40 @@ int AppLayerHandleMsg(AlpProtoDetectThreadCtx *dp_ctx, StreamMsg *smsg)
             }
         }
 
-        /* put the smsg in the stream list */
-        if (ssn->smsg_head == NULL) {
-            ssn->smsg_head = smsg;
-            ssn->smsg_tail = smsg;
-            smsg->next = NULL;
-            smsg->prev = NULL;
+        /* store the smsg in the tcp stream */
+        if (smsg->flags & STREAM_TOSERVER) {
+            /* put the smsg in the stream list */
+            if (ssn->toserver_smsg_head == NULL) {
+                ssn->toserver_smsg_head = smsg;
+                ssn->toserver_smsg_tail = smsg;
+                smsg->next = NULL;
+                smsg->prev = NULL;
+            } else {
+                StreamMsg *cur = ssn->toserver_smsg_tail;
+                cur->next = smsg;
+                smsg->prev = cur;
+                smsg->next = NULL;
+                ssn->toserver_smsg_tail = smsg;
+            }
         } else {
-            StreamMsg *cur = ssn->smsg_tail;
-            cur->next = smsg;
-            smsg->prev = cur;
-            smsg->next = NULL;
-            ssn->smsg_tail = smsg;
+            /* put the smsg in the stream list */
+            if (ssn->toclient_smsg_head == NULL) {
+                ssn->toclient_smsg_head = smsg;
+                ssn->toclient_smsg_tail = smsg;
+                smsg->next = NULL;
+                smsg->prev = NULL;
+            } else {
+                StreamMsg *cur = ssn->toclient_smsg_tail;
+                cur->next = smsg;
+                smsg->prev = cur;
+                smsg->next = NULL;
+                ssn->toclient_smsg_tail = smsg;
+            }
         }
+
         /* flow is free again */
         FlowDecrUsecnt(smsg->flow);
         /* dereference the flow */
-        SCLogDebug("deref smsg->flow, smsg %p, smsg->flow %p", smsg, smsg->flow);
         smsg->flow = NULL;
 
     } else { /* no ssn ptr */

src/detect-engine-mpm.c

@@ -41,6 +41,9 @@
 
 #include "detect-content.h"
 #include "detect-uricontent.h"
+
+#include "stream.h"
+
 #include "util-cuda-handlers.h"
 #include "util-mpm-b2g-cuda.h"
 
@@ -168,6 +171,38 @@ uint32_t UriPatternSearch(ThreadVars *tv, DetectEngineThreadCtx *det_ctx,
     SCReturnUInt(ret);
 }
 
+/** \brief Pattern match -- searches for only one pattern per signature.
+ *
+ *  \param tv threadvars
+ *  \param det_ctx detection engine thread ctx
+ *  \param smsg stream msg (reassembled stream data)
+ *
+ *  \retval ret number of matches
+ */
+uint32_t StreamPatternSearch(ThreadVars *tv, DetectEngineThreadCtx *det_ctx,
+                           StreamMsg *smsg)
+{
+    SCEnter();
+
+    uint32_t ret = 0;
+    uint8_t cnt = 0;
+
+    for ( ; smsg != NULL; smsg = smsg->next) {
+        uint32_t r = mpm_table[det_ctx->sgh->mpm_ctx->mpm_type].Search(det_ctx->sgh->mpm_ctx,
+                &det_ctx->mtc, &det_ctx->smsg_pmq[cnt], smsg->data.data, smsg->data.data_len);
+        if (r > 0) {
+            ret += r;
+
+            /* merge results with overall pmq */
+            PmqMerge(&det_ctx->smsg_pmq[cnt], &det_ctx->pmq);
+        }
+
+        cnt++;
+    }
+
+    SCReturnInt(ret);
+}
+
 
 /** \brief cleans up the mpm instance after a match */
 void PacketPatternCleanup(ThreadVars *t, DetectEngineThreadCtx *det_ctx) {

src/detect-engine-mpm.h

@@ -29,10 +29,13 @@
 #include "detect-content.h"
 #include "detect-uricontent.h"
 
+#include "stream.h"
+
 uint16_t PatternMatchDefaultMatcher(void);
 
 uint32_t PacketPatternSearch(ThreadVars *, DetectEngineThreadCtx *, Packet *);
 uint32_t UriPatternSearch(ThreadVars *, DetectEngineThreadCtx *, uint8_t *, uint16_t);
+uint32_t StreamPatternSearch(ThreadVars *, DetectEngineThreadCtx *, StreamMsg *);
 
 void PacketPatternCleanup(ThreadVars *, DetectEngineThreadCtx *);
 

src/detect-engine-payload.c

@@ -61,7 +61,7 @@
  *  \param det_ctx Detection engine thread context
  *  \param s Signature to inspect
  *  \param sm SigMatch to inspect
- *  \param p Packet
+ *  \param f flow (for pcre flowvar storage)
  *  \param payload ptr to the payload to inspect
  *  \param payload_len length of the payload
  *
@@ -70,7 +70,7 @@
  */
 static int DoInspectPacketPayload(DetectEngineCtx *de_ctx,
         DetectEngineThreadCtx *det_ctx, Signature *s, SigMatch *sm,
-        Packet *p, uint8_t *payload, uint32_t payload_len)
+        Packet *p, Flow *f, uint8_t *payload, uint32_t payload_len)
 {
     SCEnter();
 
@@ -211,7 +211,7 @@ static int DoInspectPacketPayload(DetectEngineCtx *de_ctx,
                     /* see if the next payload keywords match. If not, we will
                      * search for another occurence of this content and see
                      * if the others match then until we run out of matches */
-                    int r = DoInspectPacketPayload(de_ctx,det_ctx,s,sm->next, p, payload, payload_len);
+                    int r = DoInspectPacketPayload(de_ctx,det_ctx,s,sm->next, p, f, payload, payload_len);
                     if (r == 1) {
                         SCReturnInt(1);
                     }
@@ -250,7 +250,7 @@ static int DoInspectPacketPayload(DetectEngineCtx *de_ctx,
         {
             SCLogDebug("inspecting pcre");
 
-            int r = DetectPcrePayloadMatch(det_ctx, p, s, sm);
+            int r = DetectPcrePayloadMatch(det_ctx, s, sm, p, f, payload, payload_len);
             if (r == 1) {
                 goto match;
             }
@@ -286,20 +286,20 @@ match:
     /* this sigmatch matched, inspect the next one. If it was the last,
      * the payload portion of the signature matched. */
     if (sm->next != NULL) {
-        int r = DoInspectPacketPayload(de_ctx,det_ctx,s,sm->next, p, payload, payload_len);
+        int r = DoInspectPacketPayload(de_ctx,det_ctx,s,sm->next, p, f, payload, payload_len);
         SCReturnInt(r);
     } else {
         SCReturnInt(1);
     }
 }
 
-/** \brief Do the content inspection & validation for a signature
+/**
+ *  \brief Do the content inspection & validation for a signature
  *
  *  \param de_ctx Detection engine context
  *  \param det_ctx Detection engine thread context
  *  \param s Signature to inspect
- *  \param sm SigMatch to inspect
- *  \param f Flow
+ *  \param f flow (for pcre flowvar storage)
  *  \param flags app layer flags
  *  \param state App layer state
  *  \param p Packet
@@ -320,7 +320,45 @@ int DetectEngineInspectPacketPayload(DetectEngineCtx *de_ctx,
 
     det_ctx->payload_offset = 0;
 
-    r = DoInspectPacketPayload(de_ctx, det_ctx, s, s->pmatch, p, p->payload, p->payload_len);
+    r = DoInspectPacketPayload(de_ctx, det_ctx, s, s->pmatch, p, f, p->payload, p->payload_len);
+    if (r == 1) {
+        SCReturnInt(1);
+    }
+
+    SCReturnInt(0);
+}
+
+/**
+ *  \brief Do the content inspection & validation for a signature for a stream chunk
+ *
+ *  \param de_ctx Detection engine context
+ *  \param det_ctx Detection engine thread context
+ *  \param s Signature to inspect
+ *  \param f flow (for pcre flowvar storage)
+ *  \param payload ptr to the payload to inspect
+ *  \param payload_len length of the payload
+ *
+ *  \retval 0 no match
+ *  \retval 1 match
+ *
+ *  \todo we might also pass the packet to this function for the pktvar
+ *        storage. Only, would that be right? We're not inspecting data
+ *        from the current packet here.
+ */
+int DetectEngineInspectStreamPayload(DetectEngineCtx *de_ctx,
+        DetectEngineThreadCtx *det_ctx, Signature *s, Flow *f,
+        uint8_t *payload, uint32_t payload_len)
+{
+    SCEnter();
+    int r = 0;
+
+    if (s->pmatch == NULL) {
+        SCReturnInt(0);
+    }
+
+    det_ctx->payload_offset = 0;
+
+    r = DoInspectPacketPayload(de_ctx, det_ctx, s, s->pmatch, NULL, f, payload, payload_len);
     if (r == 1) {
         SCReturnInt(1);
     }

src/detect-engine-payload.h

@@ -27,6 +27,9 @@
 int DetectEngineInspectPacketPayload(DetectEngineCtx *,
         DetectEngineThreadCtx *, Signature *, Flow *, uint8_t,
         void *, Packet *);
+int DetectEngineInspectStreamPayload(DetectEngineCtx *,
+        DetectEngineThreadCtx *, Signature *, Flow *,
+        uint8_t *, uint32_t);
 
 void PayloadRegisterTests(void);
 

src/detect-engine.c

@@ -336,6 +336,10 @@ TmEcode DetectEngineThreadCtxInit(ThreadVars *tv, void *initdata, void **data) {
 
     //PmqSetup(&det_ctx->pmq, DetectEngineGetMaxSigId(de_ctx), DetectContentMaxId(de_ctx));
     PmqSetup(&det_ctx->pmq, 0, DetectContentMaxId(de_ctx));
+    int i;
+    for (i = 0; i < 256; i++) {
+        PmqSetup(&det_ctx->smsg_pmq[i], 0, DetectContentMaxId(de_ctx));
+    }
 
     /* IP-ONLY */
     DetectEngineIPOnlyThreadInit(de_ctx,&det_ctx->io_ctx);

src/detect-pcre.c

@@ -261,6 +261,105 @@ int DetectPcreALMatch(ThreadVars *t, DetectEngineThreadCtx *det_ctx, Flow *f,
     SCReturnInt(r);
 }
 
+/**
+ * \brief match a regex on a single payload'
+ *
+ * \param det_ctx thread detection ctx
+ * \param s signature
+ * \param sm sig match to match against
+ * \param p packet to set PktVars if any
+ * \param f flow to set FlowVars if any
+ * \param payload payload to inspect
+ * \param payload_len lenght of the payload
+ *
+ * \retval 1 match
+ * \retval 0 no match
+ */
+int DetectPcrePayloadMatch(DetectEngineThreadCtx *det_ctx, Signature *s,
+        SigMatch *sm, Packet *p, Flow *f, uint8_t *payload,
+        uint32_t payload_len)
+{
+    SCEnter();
+#define MAX_SUBSTRINGS 30
+    int ret = 0;
+    int ov[MAX_SUBSTRINGS];
+    uint8_t *ptr = NULL;
+    uint16_t len = 0;
+
+    if (payload_len == 0)
+        SCReturnInt(0);
+
+    DetectPcreData *pe = (DetectPcreData *)sm->ctx;
+
+    /* If we want to inspect the http body, we will use HTP L7 parser */
+    if (pe->flags & DETECT_PCRE_HTTP_BODY_AL)
+        SCReturnInt(0);
+
+    if (s->flags & SIG_FLAG_RECURSIVE) {
+        ptr = payload + det_ctx->payload_offset;
+        len = payload_len - det_ctx->payload_offset;
+    } else if (pe->flags & DETECT_PCRE_RELATIVE) {
+        ptr = payload + det_ctx->payload_offset;
+        len = payload_len - det_ctx->payload_offset;
+        if (ptr == NULL || len == 0)
+            SCReturnInt(0);
+    } else {
+        ptr = payload;
+        len = payload_len;
+    }
+
+    /* run the actual pcre detection */
+    ret = pcre_exec(pe->re, pe->sd, (char *)ptr, len, 0, 0, ov, MAX_SUBSTRINGS);
+    SCLogDebug("ret %d (negating %s)", ret, pe->negate ? "set" : "not set");
+
+    if (ret == PCRE_ERROR_NOMATCH) {
+        if (pe->negate == 1) {
+            /* regex didn't match with negate option means we
+             * consider it a match */
+            ret = 1;
+        } else {
+            ret = 0;
+        }
+    } else if (ret >= 0) {
+        if (pe->negate == 1) {
+            /* regex matched but we're negated, so not
+             * considering it a match */
+            ret = 0;
+        } else {
+            /* regex matched and we're not negated,
+             * considering it a match */
+
+            /* see if we need to do substring capturing. */
+            if (ret > 1 && pe->capidx != 0) {
+                const char *str_ptr;
+                ret = pcre_get_substring((char *)ptr, ov, MAX_SUBSTRINGS, 1, &str_ptr);
+                if (ret) {
+                    if (pe->flags & DETECT_PCRE_CAPTURE_PKT) {
+                        if (p != NULL) {
+                            PktVarAdd(p, pe->capname, (uint8_t *)str_ptr, ret);
+                        }
+                    } else if (pe->flags & DETECT_PCRE_CAPTURE_FLOW) {
+                        if (f != NULL) {
+                            /* flow will be locked be FlowVarAddStr */
+                            FlowVarAddStr(f, pe->capidx, (uint8_t *)str_ptr, ret);
+                        }
+                    }
+                }
+            }
+
+            /* update offset for pcre RELATIVE */
+            det_ctx->payload_offset = (ptr+ov[1]) - payload;
+
+            ret = 1;
+        }
+
+    } else {
+        SCLogDebug("pcre had matching error");
+        ret = 0;
+    }
+    SCReturnInt(ret);
+}
+
 /**
  * \brief match a regex on a single payload'
  *
@@ -269,10 +368,10 @@ int DetectPcreALMatch(ThreadVars *t, DetectEngineThreadCtx *det_ctx, Flow *f,
  * \param s signature
  * \param sm sig match to match against
  *
- * \retval 1: match
- * \retval 0: no match
+ * \retval 1 match
+ * \retval 0 no match
  */
-int DetectPcrePayloadMatch(DetectEngineThreadCtx *det_ctx, Packet *p, Signature *s, SigMatch *sm) {
+int DetectPcrePacketPayloadMatch(DetectEngineThreadCtx *det_ctx, Packet *p, Signature *s, SigMatch *sm) {
     SCEnter();
 #define MAX_SUBSTRINGS 30
     int ret = 0;
@@ -366,7 +465,7 @@ int DetectPcreMatch (ThreadVars *t, DetectEngineThreadCtx *det_ctx, Packet *p,
                      Signature *s, SigMatch *sm)
 {
     SCEnter();
-    int r = DetectPcrePayloadMatch(det_ctx, p, s, sm);
+    int r = DetectPcrePacketPayloadMatch(det_ctx, p, s, sm);
     SCReturnInt(r);
 }
 

src/detect-pcre.h

@@ -48,7 +48,8 @@ typedef struct DetectPcreData_ {
 } DetectPcreData;
 
 /* prototypes */
-int DetectPcrePayloadMatch(DetectEngineThreadCtx *, Packet *, Signature *, SigMatch *);
+int DetectPcrePayloadMatch(DetectEngineThreadCtx *, Signature *, SigMatch *, Packet *, Flow *, uint8_t *, uint32_t);
+int DetectPcrePacketPayloadMatch(DetectEngineThreadCtx *, Packet *, Signature *, SigMatch *);
 void DetectPcreRegister (void);
 
 #endif /* __DETECT_PCRE_H__ */

src/detect.c

@@ -487,7 +487,8 @@ SigGroupHead *SigMatchSignaturesGetSgh(DetectEngineCtx *de_ctx, DetectEngineThre
     SCReturnPtr(sgh, "SigGroupHead");
 }
 
-/** \brief Signature match function
+/**
+ *  \brief Signature match function
  *
  *  \retval 1 one or more signatures matched
  *  \retval 0 no matches were found
@@ -530,10 +531,21 @@ int SigMatchSignatures(ThreadVars *th_v, DetectEngineCtx *de_ctx, DetectEngineTh
         if (p->proto == IPPROTO_TCP) {
             TcpSession *ssn = (TcpSession *)p->flow->protoctx;
             if (ssn != NULL) {
-                smsg = ssn->smsg_head;
-                /* deref from the ssn */
-                ssn->smsg_head = NULL;
-                ssn->smsg_tail = NULL;
+                if (p->flowflags & FLOW_PKT_TOSERVER) {
+                    smsg = ssn->toserver_smsg_head;
+                    /* deref from the ssn */
+                    ssn->toserver_smsg_head = NULL;
+                    ssn->toserver_smsg_tail = NULL;
+
+                    //BUG_ON(ssn->toclient_smsg_head != NULL);
+                } else {
+                    smsg = ssn->toclient_smsg_head;
+                    /* deref from the ssn */
+                    ssn->toclient_smsg_head = NULL;
+                    ssn->toclient_smsg_tail = NULL;
+
+                    //BUG_ON(ssn->toserver_smsg_head != NULL);
+                }
 
                 SCLogDebug("smsg %p", smsg);
             }
@@ -588,6 +600,12 @@ int SigMatchSignatures(ThreadVars *th_v, DetectEngineCtx *de_ctx, DetectEngineTh
         goto end;
     }
 
+    /* have a look at the reassembled stream (if any) */
+    if (smsg != NULL) {
+        cnt = StreamPatternSearch(th_v, det_ctx, smsg);
+        SCLogDebug("cnt %u", cnt);
+    }
+
     if (p->payload_len > 0 && det_ctx->sgh->mpm_ctx != NULL &&
         !(p->flags & PKT_NOPAYLOAD_INSPECTION))
     {
@@ -686,7 +704,6 @@ int SigMatchSignatures(ThreadVars *th_v, DetectEngineCtx *de_ctx, DetectEngineTh
                     SCLogDebug("mpm sig without matches (pat id check in content).");
                     goto next;
                 }
-
             }
         }
 
@@ -741,8 +758,40 @@ int SigMatchSignatures(ThreadVars *th_v, DetectEngineCtx *de_ctx, DetectEngineTh
         /* Check the payload keywords. If we are a MPM sig and we've made
          * to here, we've had at least one of the patterns match */
         if (s->pmatch != NULL) {
-            if (DetectEngineInspectPacketPayload(de_ctx, det_ctx, s, p->flow, flags, alstate, p) != 1)
-                goto next;
+            if (smsg != NULL) {
+                char pmatch = 0;
+                int i = 0;
+                StreamMsg *smsg_inspect = smsg;
+                for ( ; smsg_inspect != NULL; smsg_inspect = smsg_inspect->next, i++) {
+                    if (det_ctx->smsg_pmq[i].pattern_id_array_size != 0)
+                        continue;
+
+                    if (det_ctx->smsg_pmq[i].pattern_id_bitarray != NULL) {
+                        /* filter out sigs that want pattern matches, but
+                         * have no matches */
+                        if (!(det_ctx->smsg_pmq[i].pattern_id_bitarray[(s->mpm_pattern_id / 8)] & (1<<(s->mpm_pattern_id % 8))) &&
+                                (s->flags & SIG_FLAG_MPM) && !(s->flags & SIG_FLAG_MPM_NEGCONTENT)) {
+                            SCLogDebug("no match in this smsg");
+                            continue;
+                        }
+
+                        if (DetectEngineInspectStreamPayload(de_ctx, det_ctx, s, p->flow, smsg_inspect->data.data, smsg_inspect->data.data_len) == 1) {
+                            SCLogDebug("match in smsg %p", smsg);
+                            pmatch = 1;
+                            break;
+                        }
+                    }
+                }
+
+                /* no match? then inspect packet payload */
+                if (pmatch == 0) {
+                    if (DetectEngineInspectPacketPayload(de_ctx, det_ctx, s, p->flow, flags, alstate, p) != 1)
+                        goto next;
+                }
+            } else {
+                if (DetectEngineInspectPacketPayload(de_ctx, det_ctx, s, p->flow, flags, alstate, p) != 1)
+                    goto next;
+            }
         }
 
         SCLogDebug("s->amatch %p", s->amatch);

src/detect.h

@@ -483,6 +483,7 @@ typedef struct DetectionEngineThreadCtx_ {
     MpmThreadCtx mtcu;
     struct SigGroupHead_ *sgh;
     PatternMatcherQueue pmq;
+    PatternMatcherQueue smsg_pmq[256];
 
     /* counters */
     uint32_t pkts;

src/stream-tcp-private.h

@@ -164,8 +164,10 @@ typedef struct TcpSession_ {
     TcpStream server;
     TcpStream client;
     void **aldata; /**< application level storage ptrs */
-    struct StreamMsg_ *smsg_head; /**< list of stream msgs (for detection inspection) */
-    struct StreamMsg_ *smsg_tail; /**< list of stream msgs (for detection inspection) */
+    struct StreamMsg_ *toserver_smsg_head; /**< list of stream msgs (for detection inspection) */
+    struct StreamMsg_ *toserver_smsg_tail; /**< list of stream msgs (for detection inspection) */
+    struct StreamMsg_ *toclient_smsg_head; /**< list of stream msgs (for detection inspection) */
+    struct StreamMsg_ *toclient_smsg_tail; /**< list of stream msgs (for detection inspection) */
 } TcpSession;
 
 #endif /* __STREAM_TCP_PRIVATE_H__ */

src/stream-tcp.c

@@ -174,6 +174,7 @@ void StreamTcpReturnStreamSegments (TcpStream *stream)
 void StreamTcpSessionClear(void *ssnptr)
 {
     SCEnter();
+    StreamMsg *smsg = NULL;
 
     TcpSession *ssn = (TcpSession *)ssnptr;
     if (ssn == NULL)
@@ -184,6 +185,31 @@ void StreamTcpSessionClear(void *ssnptr)
 
     AppLayerParserCleanupState(ssn);
 
+    /* if we have (a) smsg(s), return to the pool */
+    smsg = ssn->toserver_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toserver_smsg_head = NULL;
+
+    smsg = ssn->toclient_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toclient_smsg_head = NULL;
+
     memset(ssn, 0, sizeof(TcpSession));
     SCMutexLock(&ssn_pool_mutex);
     PoolReturn(ssn_pool, ssn);
@@ -209,6 +235,8 @@ static void StreamTcpSessionPktFree (Packet *p)
 {
     SCEnter();
 
+    StreamMsg *smsg = NULL;
+
     TcpSession *ssn = (TcpSession *)p->flow->protoctx;
     if (ssn == NULL)
         SCReturn;
@@ -216,6 +244,31 @@ static void StreamTcpSessionPktFree (Packet *p)
     StreamTcpReturnStreamSegments(&ssn->client);
     StreamTcpReturnStreamSegments(&ssn->server);
 
+    /* if we have (a) smsg(s), return to the pool */
+    smsg = ssn->toserver_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toserver_smsg_head = NULL;
+
+    smsg = ssn->toclient_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toclient_smsg_head = NULL;
+
     SCReturn;
 }
 
@@ -243,6 +296,8 @@ void *StreamTcpSessionPoolAlloc(void *null)
  *  \param s Void ptr to TcpSession memory */
 void StreamTcpSessionPoolFree(void *s)
 {
+    StreamMsg *smsg = NULL;
+
     if (s == NULL)
         return;
 
@@ -251,6 +306,31 @@ void StreamTcpSessionPoolFree(void *s)
     StreamTcpReturnStreamSegments(&ssn->client);
     StreamTcpReturnStreamSegments(&ssn->server);
 
+    /* if we have (a) smsg(s), return to the pool */
+    smsg = ssn->toserver_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toserver_smsg_head = NULL;
+
+    smsg = ssn->toclient_smsg_head;
+    while(smsg != NULL) {
+        StreamMsg *smsg_next = smsg->next;
+        SCLogDebug("returning smsg %p to pool", smsg);
+        smsg->next = NULL;
+        smsg->prev = NULL;
+        smsg->flow = NULL;
+        StreamMsgReturnToPool(smsg);
+        smsg = smsg_next;
+    }
+    ssn->toclient_smsg_head = NULL;
+
     StreamL7DataPtrFree(ssn);
     SCFree(ssn);
 

src/stream.c

@@ -34,8 +34,6 @@ static SCMutex stream_pool_memuse_mutex;
 static uint64_t stream_pool_memuse = 0;
 static uint64_t stream_pool_memcnt = 0;
 
-//static StreamMsgQueue stream_q;
-
 /* per queue setting */
 static uint16_t toserver_min_init_chunk_len = 0;
 static uint16_t toserver_min_chunk_len = 0;
@@ -131,22 +129,11 @@ void StreamMsgReturnToPool(StreamMsg *s) {
 /* Used by l7inspection to get msgs with data */
 StreamMsg *StreamMsgGetFromQueue(StreamMsgQueue *q)
 {
-//    SCMutexLock(&q->mutex_q);
-    if (q->len == 0) {
-        struct timespec cond_time;
-        cond_time.tv_sec = time(NULL) + 5;
-        cond_time.tv_nsec = 0;
-
-        /* if we have no stream msgs in queue, wait... for 5 seconds */
-//        SCCondTimedwait(&q->cond_q, &q->mutex_q, &cond_time);
-    }
     if (q->len > 0) {
         StreamMsg *s = StreamMsgDequeue(q);
-//        SCMutexUnlock(&q->mutex_q);
         return s;
     } else {
         /* return NULL if we have no stream msg. Should only happen on signals. */
-//        SCMutexUnlock(&q->mutex_q);
         return NULL;
     }
 }
@@ -154,16 +141,12 @@ StreamMsg *StreamMsgGetFromQueue(StreamMsgQueue *q)
 /* Used by stream reassembler to fill the queue for l7inspect reading */
 void StreamMsgPutInQueue(StreamMsgQueue *q, StreamMsg *s)
 {
-//    SCMutexLock(&q->mutex_q);
     StreamMsgEnqueue(q, s);
     SCLogDebug("q->len %" PRIu32 "", q->len);
-//    SCCondSignal(&q->cond_q);
-//    SCMutexUnlock(&q->mutex_q);
 }
 
 void StreamMsgQueuesInit(void) {
     SCMutexInit(&stream_pool_memuse_mutex, NULL);
-    //memset(&stream_q, 0, sizeof(stream_q));
 
     stream_msg_pool = PoolInit(5000,250,StreamMsgAlloc,NULL,StreamMsgFree);
     if (stream_msg_pool == NULL)
@@ -186,8 +169,6 @@ StreamMsgQueue *StreamMsgQueueGetNew(void) {
         return NULL;
 
     memset(smq, 0x00, sizeof(StreamMsgQueue));
-//    SCMutexInit(&smq->mutex_q, NULL);
-//    SCCondInit(&smq->cond_q, NULL);
     return smq;
 }
 

src/util-mpm.c

@@ -55,19 +55,23 @@ int PmqSetup(PatternMatcherQueue *pmq, uint32_t sig_maxid, uint32_t patmaxid) {
     memset(pmq, 0, sizeof(PatternMatcherQueue));
 
     if (patmaxid > 0) {
-        pmq->pattern_id_array = SCMalloc(patmaxid * sizeof(uint32_t));
+        pmq->pattern_id_array_size = patmaxid * sizeof(uint32_t);
+
+        pmq->pattern_id_array = SCMalloc(pmq->pattern_id_array_size);
         if (pmq->pattern_id_array == NULL) {
             SCReturnInt(-1);
         }
-        memset(pmq->pattern_id_array, 0, patmaxid * sizeof(uint32_t));
+        memset(pmq->pattern_id_array, 0, pmq->pattern_id_array_size);
         pmq->pattern_id_array_cnt = 0;
 
         /* lookup bitarray */
-        pmq->pattern_id_bitarray = SCMalloc((patmaxid / 8) + 1);
+        pmq->pattern_id_bitarray_size = (patmaxid / 8) + 1;
+
+        pmq->pattern_id_bitarray = SCMalloc(pmq->pattern_id_bitarray_size);
         if (pmq->pattern_id_bitarray == NULL) {
             SCReturnInt(-1);
         }
-        memset(pmq->pattern_id_bitarray, 0, (patmaxid / 8) + 1);
+        memset(pmq->pattern_id_bitarray, 0, pmq->pattern_id_bitarray_size);
 
         SCLogDebug("pmq->pattern_id_array %p, pmq->pattern_id_bitarray %p",
                 pmq->pattern_id_array, pmq->pattern_id_bitarray);
@@ -110,15 +114,44 @@ MpmVerifyMatch(MpmThreadCtx *thread_ctx, PatternMatcherQueue *pmq, uint32_t pati
     SCReturnInt(1);
 }
 
+/**
+ *  \brief Merge two pmq's bitarrays
+ *
+ *  \param src source pmq
+ *  \param dst destination pmq to merge into
+ */
+void PmqMerge(PatternMatcherQueue *src, PatternMatcherQueue *dst) {
+    uint32_t u;
+
+    if (src->pattern_id_array_cnt == 0)
+        return;
+
+    for (u = 0; u < src->pattern_id_bitarray_size && u < dst->pattern_id_bitarray_size; u++) {
+        dst->pattern_id_bitarray[u] |= src->pattern_id_bitarray[u];
+    }
+
+    /** \todo now set merged flag? */
+}
+
 /** \brief Reset a Pmq for reusage. Meant to be called after a single search.
  *  \param pmq Pattern matcher to be reset.
+ *  \todo memset is expensive, but we need it as we merge pmq's. We might use
+ *        a flag so we can clear pmq's the old way if we can.
  */
 void PmqReset(PatternMatcherQueue *pmq) {
+    if (pmq == NULL)
+        return;
+
+    memset(pmq->pattern_id_bitarray, 0, pmq->pattern_id_bitarray_size);
+    //memset(pmq->pattern_id_array, 0, pmq->pattern_id_array_size);
+    pmq->pattern_id_array_cnt = 0;
+/*
     uint32_t u;
     for (u = 0; u < pmq->pattern_id_array_cnt; u++) {
         pmq->pattern_id_bitarray[(pmq->pattern_id_array[u] / 8)] &= ~(1<<(pmq->pattern_id_array[u] % 8));
     }
     pmq->pattern_id_array_cnt = 0;
+*/
 }
 
 /** \brief Cleanup a Pmq

src/util-mpm.h

@@ -79,11 +79,14 @@ typedef struct MpmThreadCtx_ {
  *         thread has this and passes a pointer to it to the pattern matcher.
  *         The actual pattern matcher will fill the structure. */
 typedef struct PatternMatcherQueue_ {
-    uint32_t *pattern_id_array;     /** array with internal sig id's that had a
+    uint32_t *pattern_id_array;     /** array with pattern id's that had a
                                         pattern match. These will be inspected
                                         futher by the detection engine. */
     uint32_t pattern_id_array_cnt;
+    uint32_t pattern_id_array_size; /**< size in bytes */
+
     uint8_t *pattern_id_bitarray;   /** bitarray with pattern id matches */
+    uint32_t pattern_id_bitarray_size; /**< size in bytes */
 } PatternMatcherQueue;
 
 typedef struct MpmCtx_ {
@@ -142,6 +145,7 @@ typedef struct MpmTableElmt_ {
 MpmTableElmt mpm_table[MPM_TABLE_SIZE];
 
 int PmqSetup(PatternMatcherQueue *, uint32_t, uint32_t);
+void PmqMerge(PatternMatcherQueue *src, PatternMatcherQueue *dst);
 void PmqReset(PatternMatcherQueue *);
 void PmqCleanup(PatternMatcherQueue *);
 void PmqFree(PatternMatcherQueue *);