detect: improve stateful detection

Now that MPM runs when the TX progress is right, stateful detection
operates differently.

Changes:

1. raw stream inspection is now also an inspect engine

   Since this engine doesn't take the transactions into account, it
   could potentially run multiple times on the same data. To avoid
   this, basic result caching is in place.

2. the engines are sorted by progress, but the 'MPM' engine is first
   even if the progress is higher

   If MPM flags a rule to be inspected, the inspect engine for that
   buffer runs first. If this step fails, the rule is no longer
   evaluated. No state is stored.

src/detect-engine-payload.c

@@ -33,6 +33,7 @@
 #include "detect-parse.h"
 #include "detect-engine-content-inspection.h"
 #include "detect-engine-prefilter.h"
+#include "detect-engine-state.h"
 
 #include "stream.h"
 #include "stream-tcp.h"
@@ -232,6 +233,93 @@ int DetectEngineInspectStreamPayload(DetectEngineCtx *de_ctx,
     return r;
 }
 
+struct StreamContentInspectEngineData {
+    DetectEngineCtx *de_ctx;
+    DetectEngineThreadCtx *det_ctx;
+    const Signature *s;
+    const SigMatchData *smd;
+    Flow *f;
+};
+
+static int StreamContentInspectEngineFunc(void *cb_data, const uint8_t *data, const uint32_t data_len)
+{
+    SCEnter();
+    int r = 0;
+    struct StreamContentInspectEngineData *smd = cb_data;
+#ifdef DEBUG
+    smd->det_ctx->stream_persig_cnt++;
+    smd->det_ctx->stream_persig_size += data_len;
+#endif
+    smd->det_ctx->buffer_offset = 0;
+    smd->det_ctx->discontinue_matching = 0;
+    smd->det_ctx->inspection_recursion_counter = 0;
+
+    r = DetectEngineContentInspection(smd->de_ctx, smd->det_ctx,
+            smd->s, smd->smd,
+            smd->f, (uint8_t *)data, data_len, 0,
+            DETECT_ENGINE_CONTENT_INSPECTION_MODE_STREAM, NULL);
+    if (r == 1) {
+        SCReturnInt(1);
+    }
+
+    SCReturnInt(0);
+}
+
+/**
+ *  \brief inspect engine for stateful rules
+ *
+ *  Caches results as it may be called multiple times if we inspect
+ *  multiple transactions in one packet.
+ *
+ *  Returns "can't match" if depth is reached.
+ */
+int DetectEngineInspectStream(ThreadVars *tv,
+        DetectEngineCtx *de_ctx, DetectEngineThreadCtx *det_ctx,
+        const Signature *s, const SigMatchData *smd,
+        Flow *f, uint8_t flags, void *alstate, void *txv, uint64_t tx_id)
+{
+    Packet *p = det_ctx->p; /* TODO: get rid of this HACK */
+
+    if (det_ctx->stream_already_inspected)
+        return det_ctx->stream_last_result;
+
+    uint64_t unused;
+    struct StreamContentInspectEngineData inspect_data = { de_ctx, det_ctx, s, smd, f };
+    int match = StreamReassembleRaw(f->protoctx, p,
+            StreamContentInspectEngineFunc, &inspect_data,
+            &unused);
+
+    bool is_last = false;
+    TcpSession *ssn = f->protoctx;
+    if (flags & STREAM_TOSERVER) {
+        TcpStream *stream = &ssn->client;
+        if (stream->flags & STREAMTCP_STREAM_FLAG_DEPTH_REACHED)
+            is_last = true;
+    } else {
+        TcpStream *stream = &ssn->server;
+        if (stream->flags & STREAMTCP_STREAM_FLAG_DEPTH_REACHED)
+            is_last = true;
+    }
+
+    SCLogDebug("%s ran stream for sid %u on packet %"PRIu64" and we %s",
+            is_last? "LAST:" : "normal:", s->id, p->pcap_cnt,
+            match ? "matched" : "didn't match");
+    det_ctx->stream_already_inspected = true;
+
+    if (match) {
+        det_ctx->stream_last_result = DETECT_ENGINE_INSPECT_SIG_MATCH;
+        return DETECT_ENGINE_INSPECT_SIG_MATCH;
+    } else {
+        if (is_last) {
+            det_ctx->stream_last_result = DETECT_ENGINE_INSPECT_SIG_CANT_MATCH;
+            //SCLogNotice("last, so DETECT_ENGINE_INSPECT_SIG_CANT_MATCH");
+            return DETECT_ENGINE_INSPECT_SIG_CANT_MATCH;
+        }
+        /* TODO maybe we can set 'CANT_MATCH' for EOF too? */
+        det_ctx->stream_last_result = DETECT_ENGINE_INSPECT_SIG_NO_MATCH;
+        return DETECT_ENGINE_INSPECT_SIG_NO_MATCH;
+    }
+}
 
 #ifdef UNITTESTS
 

src/detect-engine-payload.h

@@ -32,6 +32,10 @@ int DetectEngineInspectPacketPayload(DetectEngineCtx *,
 int DetectEngineInspectStreamPayload(DetectEngineCtx *,
         DetectEngineThreadCtx *, const Signature *, Flow *,
         Packet *);
+int DetectEngineInspectStream(ThreadVars *tv,
+        DetectEngineCtx *de_ctx, DetectEngineThreadCtx *det_ctx,
+        const Signature *s, const SigMatchData *smd,
+        Flow *f, uint8_t flags, void *alstate, void *txv, uint64_t tx_id);
 
 void PayloadRegisterTests(void);
 

src/detect-engine-state.c

@@ -364,7 +364,7 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
                                 const Signature *s, Packet *p, Flow *f, uint8_t flags,
                                 AppProto alproto)
 {
-    SCLogDebug("rule %u", s->id);
+    SCLogDebug("rule %u/%u", s->id, s->num);
 
     /* TX based matches (inspect engines) */
     if (unlikely(!AppLayerParserProtocolSupportsTxs(f->proto, alproto))) {
@@ -381,13 +381,11 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
     int alert_cnt = 0;
     uint8_t direction = (flags & STREAM_TOSERVER) ? 0 : 1;
     int check_before_add = 0;
-    uint64_t tx_id = 0;
-    uint64_t total_txs = 0;
 
     /* if continue detection already inspected this rule for this tx,
      * continue with the first not-inspected tx */
     uint8_t offset = det_ctx->de_state_sig_array[s->num] & 0xef;
-    tx_id = AppLayerParserGetTransactionInspectId(f->alparser, flags);
+    uint64_t tx_id = AppLayerParserGetTransactionInspectId(f->alparser, flags);
     if (offset > 0) {
         SCLogDebug("using stored_tx_id %u instead of %u", (uint)tx_id+offset, (uint)tx_id);
         tx_id += offset;
@@ -396,11 +394,12 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
         check_before_add = 1;
     }
 
-    total_txs = AppLayerParserGetTxCnt(f->proto, alproto, alstate);
+    uint64_t total_txs = AppLayerParserGetTxCnt(f->proto, alproto, alstate);
     SCLogDebug("total_txs %"PRIu64, total_txs);
 
     SCLogDebug("starting: start tx %u, packet %u", (uint)tx_id, (uint)p->pcap_cnt);
 
+    det_ctx->stream_already_inspected = false;
     for (; tx_id < total_txs; tx_id++) {
         int total_matches = 0;
         void *tx = AppLayerParserGetTx(f->proto, alproto, alstate, tx_id);
@@ -409,6 +408,20 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
             continue;
         det_ctx->tx_id = tx_id;
         det_ctx->tx_id_set = 1;
+        det_ctx->p = p;
+
+        /* see if we need to consider the next tx in our decision to add
+         * a sig to the 'no inspect array'. */
+        int next_tx_no_progress = 0;
+        if (!TxIsLast(tx_id, total_txs)) {
+            void *next_tx = AppLayerParserGetTx(f->proto, alproto, alstate, tx_id+1);
+            if (next_tx != NULL) {
+                int c = AppLayerParserGetStateProgress(f->proto, alproto, next_tx, flags);
+                if (c == 0) {
+                    next_tx_no_progress = 1;
+                }
+            }
+        }
 
         DetectEngineAppInspectionEngine *engine = s->app_inspect;
         SCLogDebug("engine %p", engine);
@@ -421,6 +434,12 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
                 int match = engine->Callback(tv, de_ctx, det_ctx,
                         s, engine->smd, f, flags, alstate, tx, tx_id);
                 SCLogDebug("engine %p match %d", engine, match);
+                if ((match == DETECT_ENGINE_INSPECT_SIG_NO_MATCH || match == DETECT_ENGINE_INSPECT_SIG_CANT_MATCH)
+                        && (engine->mpm)) {
+                    SCLogDebug("MPM and not matching, so skip the whole TX");
+                    // TODO
+                    goto try_next;
+                } else
                 if (match == DETECT_ENGINE_INSPECT_SIG_MATCH) {
                     inspect_flags |= BIT_U32(engine->id);
                     engine = engine->next;
@@ -464,46 +483,31 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
          * we store the state so that ContinueDetection knows about it */
         int tx_is_done = (AppLayerParserGetStateProgress(f->proto, alproto, tx, flags) >=
                 AppLayerParserGetStateProgressCompletionStatus(alproto, flags));
-        /* see if we need to consider the next tx in our decision to add
-         * a sig to the 'no inspect array'. */
-        int next_tx_no_progress = 0;
-        if (!TxIsLast(tx_id, total_txs)) {
-            void *next_tx = AppLayerParserGetTx(f->proto, alproto, alstate, tx_id+1);
-            if (next_tx != NULL) {
-                int c = AppLayerParserGetStateProgress(f->proto, alproto, next_tx, flags);
-                if (c == 0) {
-                    next_tx_no_progress = 1;
-                }
-            }
-        }
 
         SCLogDebug("tx %u, packet %u, rule %u, alert_cnt %u, last tx %d, tx_is_done %d, next_tx_no_progress %d",
                 (uint)tx_id, (uint)p->pcap_cnt, s->num, alert_cnt,
                 TxIsLast(tx_id, total_txs), tx_is_done, next_tx_no_progress);
 
-        /* if we have something to store (partial match or file store info),
-         * then we do it now. */
-        if (inspect_flags != 0) {
-            if (!(TxIsLast(tx_id, total_txs)) || !tx_is_done) {
-                if (engine == NULL || inspect_flags & DE_STATE_FLAG_SIG_CANT_MATCH) {
-                    inspect_flags |= DE_STATE_FLAG_FULL_INSPECT;
-                }
-
-                /* store */
-                StoreStateTx(det_ctx, f, flags, tx_id, tx,
-                        s, smd, inspect_flags, file_no_match, check_before_add);
-            } else {
-                StoreStateTxFileOnly(det_ctx, f, flags, tx_id, tx, file_no_match);
+        /* store our state */
+        if (!(TxIsLast(tx_id, total_txs)) || !tx_is_done) {
+            if (engine == NULL || inspect_flags & DE_STATE_FLAG_SIG_CANT_MATCH) {
+                inspect_flags |= DE_STATE_FLAG_FULL_INSPECT;
             }
+
+            /* store */
+            StoreStateTx(det_ctx, f, flags, tx_id, tx,
+                    s, smd, inspect_flags, file_no_match, check_before_add);
         } else {
-            SCLogDebug("no state to store");
+            StoreStateTxFileOnly(det_ctx, f, flags, tx_id, tx, file_no_match);
         }
+    try_next:
         if (next_tx_no_progress)
             break;
     } /* for */
 
     det_ctx->tx_id = 0;
     det_ctx->tx_id_set = 0;
+    det_ctx->p = NULL;
     return alert_cnt ? 1:0;
 }
 
@@ -517,6 +521,7 @@ static int DoInspectItem(ThreadVars *tv,
     const int next_tx_no_progress)                // tx after current is still dormant
 {
     Signature *s = de_ctx->sig_array[item->sid];
+    det_ctx->stream_already_inspected = false;
 
     SCLogDebug("file_no_match %u, sid %u", *file_no_match, s->id);
 
@@ -624,6 +629,7 @@ static int DoInspectItem(ThreadVars *tv,
 
     det_ctx->tx_id = inspect_tx_id;
     det_ctx->tx_id_set = 1;
+    det_ctx->p = p;
     SCLogDebug("inspecting: tx %"PRIu64" packet %"PRIu64, inspect_tx_id, p->pcap_cnt);
 
     uint8_t direction = (flags & STREAM_TOSERVER) ? 0 : 1;
@@ -803,6 +809,7 @@ void DeStateDetectContinueDetection(ThreadVars *tv, DetectEngineCtx *de_ctx,
     }
 
 end:
+    det_ctx->p = NULL;
     det_ctx->tx_id = 0;
     det_ctx->tx_id_set = 0;
     return;

src/detect-engine.c

@@ -50,7 +50,7 @@
 
 #include "detect-engine.h"
 #include "detect-engine-state.h"
-
+#include "detect-engine-payload.h"
 #include "detect-byte-extract.h"
 #include "detect-content.h"
 #include "detect-uricontent.h"
@@ -110,7 +110,8 @@ void DetectAppLayerInspectEngineRegister(const char *name,
 
     if ((alproto >= ALPROTO_FAILED) ||
         (!(dir == SIG_FLAG_TOSERVER || dir == SIG_FLAG_TOCLIENT)) ||
-        (sm_list < DETECT_SM_LIST_MATCH) ||
+        (sm_list < DETECT_SM_LIST_MATCH) || (sm_list >= SHRT_MAX) ||
+        (progress < 0 || progress >= SHRT_MAX) ||
         (Callback == NULL))
     {
         SCLogError(SC_ERR_INVALID_ARGUMENTS, "Invalid arguments");
@@ -147,12 +148,62 @@ void DetectAppLayerInspectEngineRegister(const char *name,
     }
 }
 
+/** \internal
+ *  \brief append the stream inspection
+ *
+ *  If stream inspection is MPM, then prepend it.
+ */
+static void AppendStreamInspectEngine(Signature *s, SigMatchData *stream, int direction, uint32_t id)
+{
+    bool prepend = false;
+
+    DetectEngineAppInspectionEngine *new_engine = SCCalloc(1, sizeof(DetectEngineAppInspectionEngine));
+    if (unlikely(new_engine == NULL)) {
+        exit(EXIT_FAILURE);
+    }
+    if (SigMatchListSMBelongsTo(s, s->init_data->mpm_sm) == DETECT_SM_LIST_PMATCH) {
+        SCLogDebug("stream is mpm");
+        prepend = true;
+        new_engine->mpm = true;
+    }
+    new_engine->alproto = ALPROTO_UNKNOWN; /* all */
+    new_engine->dir = direction;
+    new_engine->sm_list = DETECT_SM_LIST_PMATCH;
+    new_engine->smd = stream;
+    new_engine->Callback = DetectEngineInspectStream;
+    new_engine->progress = 0;
+
+    /* append */
+    if (s->app_inspect == NULL) {
+        s->app_inspect = new_engine;
+        new_engine->id = DE_STATE_FLAG_BASE; /* id is used as flag in stateful detect */
+    } else if (prepend) {
+        new_engine->next = s->app_inspect;
+        s->app_inspect = new_engine;
+        new_engine->id = id;
+
+    } else {
+        DetectEngineAppInspectionEngine *a = s->app_inspect;
+        while (a->next != NULL) {
+            a = a->next;
+        }
+
+        a->next = new_engine;
+        new_engine->id = id;
+    }
+    SCLogDebug("sid %u: engine %p/%u added", s->id, new_engine, new_engine->id);
+}
+
 int DetectEngineAppInspectionEngine2Signature(Signature *s)
 {
     const int nlists = DetectBufferTypeMaxId();
     SigMatchData *ptrs[nlists];
     memset(&ptrs, 0, (nlists * sizeof(SigMatchData *)));
 
+    const int mpm_list = s->init_data->mpm_sm ?
+        SigMatchListSMBelongsTo(s, s->init_data->mpm_sm) :
+        -1;
+
     /* convert lists to SigMatchData arrays */
     int i = 0;
     for (i = DETECT_SM_LIST_DYNAMIC_START; i < nlists; i++) {
@@ -162,8 +213,12 @@ int DetectEngineAppInspectionEngine2Signature(Signature *s)
         ptrs[i] = SigMatchList2DataArray(s->init_data->smlists[i]);
     }
 
+    bool head_is_mpm = false;
+    uint32_t last_id = DE_STATE_FLAG_BASE;
     DetectEngineAppInspectionEngine *t = g_app_inspect_engines;
     while (t != NULL) {
+        bool prepend = false;
+
         if (ptrs[t->sm_list] == NULL)
             goto next;
         if (t->alproto == ALPROTO_UNKNOWN) {
@@ -178,28 +233,47 @@ int DetectEngineAppInspectionEngine2Signature(Signature *s)
             if (t->dir == 0)
                 goto next;
         }
-
         DetectEngineAppInspectionEngine *new_engine = SCCalloc(1, sizeof(DetectEngineAppInspectionEngine));
         if (unlikely(new_engine == NULL)) {
             exit(EXIT_FAILURE);
         }
+        if (mpm_list == t->sm_list) {
+            SCLogDebug("%s is mpm", DetectBufferTypeGetNameById(t->sm_list));
+            prepend = true;
+            head_is_mpm = true;
+            new_engine->mpm = true;
+        }
+
         new_engine->alproto = t->alproto;
         new_engine->dir = t->dir;
         new_engine->sm_list = t->sm_list;
         new_engine->smd = ptrs[new_engine->sm_list];
         new_engine->Callback = t->Callback;
+        new_engine->progress = t->progress;
 
         if (s->app_inspect == NULL) {
             s->app_inspect = new_engine;
-            new_engine->id = DE_STATE_FLAG_BASE; /* id is used as flag in stateful detect */
+            last_id = new_engine->id = DE_STATE_FLAG_BASE; /* id is used as flag in stateful detect */
+
+        /* prepend engine if forced or if our engine has a lower progress. */
+        } else if (prepend || (!head_is_mpm && s->app_inspect->progress > new_engine->progress)) {
+            new_engine->next = s->app_inspect;
+            s->app_inspect = new_engine;
+            new_engine->id = ++last_id;
+
         } else {
             DetectEngineAppInspectionEngine *a = s->app_inspect;
             while (a->next != NULL) {
+                if (a->next && a->next->progress > new_engine->progress) {
+                    break;
+                }
+
                 a = a->next;
             }
 
+            new_engine->next = a->next;
             a->next = new_engine;
-            new_engine->id = a->id + 1;
+            new_engine->id = ++last_id;
         }
         SCLogDebug("sid %u: engine %p/%u added", s->id, new_engine, new_engine->id);
 
@@ -208,6 +282,30 @@ next:
         t = t->next;
     }
 
+    if ((s->flags & SIG_FLAG_STATE_MATCH) && s->init_data->smlists[DETECT_SM_LIST_PMATCH] != NULL) {
+        /* if engine is added multiple times, we pass it the same list */
+        SigMatchData *stream = SigMatchList2DataArray(s->init_data->smlists[DETECT_SM_LIST_PMATCH]);
+        BUG_ON(stream == NULL);
+        if (s->flags & SIG_FLAG_TOSERVER && !(s->flags & SIG_FLAG_TOCLIENT)) {
+            AppendStreamInspectEngine(s, stream, 0, last_id + 1);
+        } else if (s->flags & SIG_FLAG_TOCLIENT && !(s->flags & SIG_FLAG_TOSERVER)) {
+            AppendStreamInspectEngine(s, stream, 1, last_id + 1);
+        } else {
+            AppendStreamInspectEngine(s, stream, 0, last_id + 1);
+            AppendStreamInspectEngine(s, stream, 1, last_id + 1);
+        }
+    }
+
+#ifdef DEBUG
+    DetectEngineAppInspectionEngine *iter = s->app_inspect;
+    while (iter) {
+        SCLogNotice("%u: engine %s id %u progress %d %s", s->id,
+                DetectBufferTypeGetNameById(iter->sm_list), iter->id,
+                iter->progress,
+                iter->sm_list == mpm_list ? "MPM":"");
+        iter = iter->next;
+    }
+#endif
     return 0;
 }
 

src/detect.c

@@ -1272,7 +1272,7 @@ void SigMatchSignatures(ThreadVars *th_v, DetectEngineCtx *de_ctx, DetectEngineT
 
         /* 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->sm_arrays[DETECT_SM_LIST_PMATCH] != NULL) {
+        if (!(sflags & SIG_FLAG_STATE_MATCH) && s->sm_arrays[DETECT_SM_LIST_PMATCH] != NULL) {
             KEYWORD_PROFILING_SET_LIST(det_ctx, DETECT_SM_LIST_PMATCH);
             /* if we have stream msgs, inspect against those first,
              * but not for a "dsize" signature */

src/detect.h

@@ -334,7 +334,8 @@ typedef struct DetectEngineAppInspectionEngine_ {
     AppProto alproto;
     uint8_t dir;
     uint8_t id;     /**< per sig id used in state keeping */
-    int16_t sm_list;
+    uint16_t mpm:1;
+    uint16_t sm_list:15;
     int16_t progress;
 
     /* \retval 0 No match.  Don't discontinue matching yet.  We need more data.
@@ -835,6 +836,9 @@ typedef struct DetectEngineThreadCtx_ {
     uint16_t tx_id_set;
     /** ID of the transaction currently being inspected. */
     uint64_t tx_id;
+    Packet *p;
+    bool stream_already_inspected;
+    int stream_last_result;
 
     SC_ATOMIC_DECLARE(int, so_far_used_by_detect);