stream: introduce min inspect depth logic

Some rules need to inspect both raw stream data and higher level
buffers together. When this higher level buffer is a streaming
buffer itself, the risk of mismatch exists.

This patch allows an app-layer parser to set a 'min inspect depth'.
The value is used by the stream engine to keep at least this
depth worth of data, so that the detection engine can request
all of it for inspection.

For rules that have the SIG_FLAG_FLUSH flag set, data is inspected
not from offset raw_progress, but from raw_progress minus
min_inspect_depth.

At this time this is only used for sigs that have their fast_pattern
in a HTTP body and have raw stream match as well.

src/app-layer-htp.c

@@ -695,6 +695,10 @@ static int Setup(Flow *f, HtpState *hstate)
 
     htp_connp_open(hstate->connp, NULL, f->sp, NULL, f->dp, &f->startts);
 
+    StreamTcpReassemblySetMinInspectDepth(f->protoctx, STREAM_TOSERVER,
+            htp_cfg_rec->request.inspect_min_size);
+    StreamTcpReassemblySetMinInspectDepth(f->protoctx, STREAM_TOCLIENT,
+            htp_cfg_rec->response.inspect_min_size);
     return 0;
 error:
     return -1;

src/detect-engine-payload.c

@@ -82,7 +82,8 @@ static void PrefilterPktStream(DetectEngineThreadCtx *det_ctx,
         struct StreamMpmData stream_mpm_data = { det_ctx, mpm_ctx };
         StreamReassembleRaw(p->flow->protoctx, p,
                 StreamMpmFunc, &stream_mpm_data,
-                &det_ctx->raw_stream_progress);
+                &det_ctx->raw_stream_progress,
+                false /* mpm doesn't use min inspect depth */);
         SCLogDebug("POST det_ctx->raw_stream_progress %"PRIu64,
                 det_ctx->raw_stream_progress);
     } else {
@@ -268,7 +269,7 @@ int DetectEngineInspectStreamPayload(DetectEngineCtx *de_ctx,
     struct StreamContentInspectData inspect_data = { de_ctx, det_ctx, s, f };
     int r = StreamReassembleRaw(f->protoctx, p,
             StreamContentInspectFunc, &inspect_data,
-            &unused);
+            &unused, ((s->flags & SIG_FLAG_FLUSH) != 0));
     return r;
 }
 
@@ -338,7 +339,7 @@ int DetectEngineInspectStream(ThreadVars *tv,
     struct StreamContentInspectEngineData inspect_data = { de_ctx, det_ctx, s, smd, f };
     int match = StreamReassembleRaw(f->protoctx, p,
             StreamContentInspectEngineFunc, &inspect_data,
-            &unused);
+            &unused, ((s->flags & SIG_FLAG_FLUSH) != 0));
 
     bool is_last = false;
     if (flags & STREAM_TOSERVER) {

src/stream-tcp-list.c

@@ -698,6 +698,19 @@ static inline uint64_t GetLeftEdge(TcpSession *ssn, TcpStream *stream)
                 raw_progress -= (uint64_t)chunk_size;
             }
         }
+
+        /* apply min inspect depth: if it is set we need to keep data
+         * before the raw progress. */
+        if (use_app && stream->min_inspect_depth) {
+            if (raw_progress < stream->min_inspect_depth)
+                raw_progress = 0;
+            else
+                raw_progress -= stream->min_inspect_depth;
+
+            SCLogDebug("stream->min_inspect_depth %u, raw_progress %"PRIu64,
+                    stream->min_inspect_depth, raw_progress);
+        }
+
         if (use_app) {
             left_edge = MIN(STREAM_APP_PROGRESS(stream), raw_progress);
             SCLogDebug("left_edge %"PRIu64", using both app:%"PRIu64", raw:%"PRIu64,

src/stream-tcp-private.h

@@ -115,6 +115,9 @@ typedef struct TcpStream_ {
     uint32_t raw_progress_rel;      /**< raw reassembly progress relative to STREAM_BASE_OFFSET */
     uint32_t log_progress_rel;      /**< streaming logger progress relative to STREAM_BASE_OFFSET */
 
+    uint32_t min_inspect_depth;     /**< min inspect size set by the app layer, to make sure enough data
+                                     *   remains available for inspection together with app layer buffers */
+
     StreamingBuffer sb;
     struct TCPSEG seg_tree;         /**< red black tree of TCP segments. Data is stored in TcpStream::sb */
     uint32_t segs_right_edge;

src/stream-tcp-reassemble.c

@@ -1526,11 +1526,17 @@ static int StreamReassembleRawInline(TcpSession *ssn, const Packet *p,
  *  \param[in] progress_in progress to work from
  *  \param[out] progress_out absolute progress value of the data this
  *                           call handled.
+ *  \param eof we're wrapping up so inspect all data we have, incl unACKd
+ *  \param respect_inspect_depth use Stream::min_inspect_depth if set
+ *
+ *  `respect_inspect_depth` is used to avoid useless inspection of too
+ *  much data.
  */
 static int StreamReassembleRawDo(TcpSession *ssn, TcpStream *stream,
                         StreamReassembleRawFunc Callback, void *cb_data,
                         const uint64_t progress_in,
-                        uint64_t *progress_out, bool eof)
+                        uint64_t *progress_out, bool eof,
+                        bool respect_inspect_depth)
 {
     SCEnter();
     int r = 0;
@@ -1539,6 +1545,27 @@ static int StreamReassembleRawDo(TcpSession *ssn, TcpStream *stream,
     uint64_t progress = progress_in;
     uint64_t last_ack_abs = STREAM_BASE_OFFSET(stream); /* absolute right edge of ack'd data */
 
+    /* if the app layer triggered a flush, and we're supposed to
+     * use a minimal inspect depth, we actually take the app progress
+     * as that is the right edge of the data. Then we take the window
+     * of 'min_inspect_depth' before that. */
+    if (respect_inspect_depth &&
+        (stream->flags & STREAMTCP_STREAM_FLAG_TRIGGER_RAW)
+        && stream->min_inspect_depth)
+    {
+        progress = STREAM_APP_PROGRESS(stream);
+        if (stream->min_inspect_depth >= progress) {
+            progress = 0;
+        } else {
+            progress -= stream->min_inspect_depth;
+        }
+        SCLogDebug("applied min inspect depth due to STREAMTCP_STREAM_FLAG_TRIGGER_RAW: progress %"PRIu64, progress);
+
+        SCLogDebug("stream app %"PRIu64", raw %"PRIu64, STREAM_APP_PROGRESS(stream), STREAM_RAW_PROGRESS(stream));
+    }
+
+    SCLogDebug("progress %"PRIu64", min inspect depth %u %s", progress, stream->min_inspect_depth, stream->flags & STREAMTCP_STREAM_FLAG_TRIGGER_RAW ? "STREAMTCP_STREAM_FLAG_TRIGGER_RAW":"(no trigger)");
+
     /* get window of data that is acked */
     if (STREAM_LASTACK_GT_BASESEQ(stream)) {
         SCLogDebug("last_ack %u, base_seq %u", stream->last_ack, stream->base_seq);
@@ -1626,7 +1653,7 @@ end:
 
 int StreamReassembleRaw(TcpSession *ssn, const Packet *p,
                         StreamReassembleRawFunc Callback, void *cb_data,
-                        uint64_t *progress_out)
+                        uint64_t *progress_out, bool respect_inspect_depth)
 {
     /* handle inline seperately as the logic is very different */
     if (StreamTcpInlineMode() == TRUE) {
@@ -1649,7 +1676,7 @@ int StreamReassembleRaw(TcpSession *ssn, const Packet *p,
 
     return StreamReassembleRawDo(ssn, stream, Callback, cb_data,
             STREAM_RAW_PROGRESS(stream), progress_out,
-            (p->flags & PKT_PSEUDO_STREAM_END));
+            (p->flags & PKT_PSEUDO_STREAM_END), respect_inspect_depth);
 }
 
 int StreamReassembleLog(TcpSession *ssn, TcpStream *stream,
@@ -1661,7 +1688,7 @@ int StreamReassembleLog(TcpSession *ssn, TcpStream *stream,
         return 0;
 
     return StreamReassembleRawDo(ssn, stream, Callback, cb_data,
-            progress_in, progress_out, eof);
+            progress_in, progress_out, eof, false);
 }
 
 /** \internal
@@ -1819,6 +1846,23 @@ void StreamTcpReassembleTriggerRawReassembly(TcpSession *ssn, int direction)
     }
 }
 
+void StreamTcpReassemblySetMinInspectDepth(TcpSession *ssn, int direction, uint32_t depth)
+{
+#ifdef DEBUG
+    BUG_ON(ssn == NULL);
+#endif
+
+    if (ssn != NULL) {
+        if (direction == STREAM_TOSERVER) {
+            ssn->client.min_inspect_depth = depth;
+            SCLogDebug("ssn %p: set client.min_inspect_depth to %u", ssn, depth);
+        } else {
+            ssn->server.min_inspect_depth = depth;
+            SCLogDebug("ssn %p: set server.min_inspect_depth to %u", ssn, depth);
+        }
+    }
+}
+
 #ifdef UNITTESTS
 /** unit tests and it's support functions below */
 

src/stream-tcp-reassemble.h

@@ -128,6 +128,7 @@ int StreamTcpCheckStreamContents(uint8_t *, uint16_t , TcpStream *);
 #endif
 
 bool StreamReassembleRawHasDataReady(TcpSession *ssn, Packet *p);
+void StreamTcpReassemblySetMinInspectDepth(TcpSession *ssn, int direction, uint32_t depth);
 
 static inline bool STREAM_LASTACK_GT_BASESEQ(const TcpStream *stream)
 {

src/stream-tcp.h

@@ -132,7 +132,8 @@ int StreamReassembleLog(TcpSession *ssn, TcpStream *stream,
         uint64_t progress_in,
         uint64_t *progress_out, bool eof);
 int StreamReassembleRaw(TcpSession *ssn, const Packet *p,
-        StreamReassembleRawFunc Callback, void *cb_data, uint64_t *progress_out);
+        StreamReassembleRawFunc Callback, void *cb_data,
+        uint64_t *progress_out, bool respect_inspect_depth);
 void StreamReassembleRawUpdateProgress(TcpSession *ssn, Packet *p, uint64_t progress);
 
 void StreamTcpDetectLogFlush(ThreadVars *tv, StreamTcpThread *stt, Flow *f, Packet *p, PacketQueue *pq);

src/tests/stream-tcp-reassemble.c

@@ -54,7 +54,7 @@ static int TestReassembleRawValidate(TcpSession *ssn, Packet *p,
 {
     struct TestReassembleRawCallbackData cb = { data, data_len };
     uint64_t progress = 0;
-    int r = StreamReassembleRaw(ssn, p, TestReassembleRawCallback, &cb, &progress);
+    int r = StreamReassembleRaw(ssn, p, TestReassembleRawCallback, &cb, &progress, false);
     if (r == 1) {
         StreamReassembleRawUpdateProgress(ssn, p, progress);
     }