app-layer: add 'incomplete' return logic

Allow app-layer parsers to indicate how much data they need
before being called again.

rust/src/parser.rs

@@ -48,6 +48,13 @@ impl AppLayerResult {
             needed: 0,
         };
     }
+    pub fn incomplete(consumed: u32, needed: u32) -> AppLayerResult {
+        return AppLayerResult {
+            status: 1,
+            consumed: consumed,
+            needed: needed,
+        };
+    }
 }
 
 /// Rust parser declaration

src/app-layer-parser.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2013 Open Information Security Foundation
+/* Copyright (C) 2007-2020 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
@@ -1178,6 +1178,8 @@ void AppLayerParserSetTxDetectFlags(uint8_t ipproto, AppProto alproto, void *tx,
 
 /***** General *****/
 
+/** \retval int -1 in case of unrecoverable error. App-layer tracking stops for this flow.
+ *  \retval int 0 ok */
 int AppLayerParserParse(ThreadVars *tv, AppLayerParserThreadCtx *alp_tctx, Flow *f, AppProto alproto,
                         uint8_t flags, const uint8_t *input, uint32_t input_len)
 {
@@ -1189,6 +1191,8 @@ int AppLayerParserParse(ThreadVars *tv, AppLayerParserThreadCtx *alp_tctx, Flow
     AppLayerParserProtoCtx *p = &alp_ctx.ctxs[f->protomap][alproto];
     void *alstate = NULL;
     uint64_t p_tx_cnt = 0;
+    uint32_t consumed = input_len;
+    const int direction = (flags & STREAM_TOSERVER) ? 0 : 1;
 
     /* we don't have the parser registered for this protocol */
     if (p->StateAlloc == NULL)
@@ -1233,13 +1237,39 @@ int AppLayerParserParse(ThreadVars *tv, AppLayerParserThreadCtx *alp_tctx, Flow
     /* invoke the recursive parser, but only on data. We may get empty msgs on EOF */
     if (input_len > 0 || (flags & STREAM_EOF)) {
         /* invoke the parser */
-        AppLayerResult res = p->Parser[(flags & STREAM_TOSERVER) ? 0 : 1](f, alstate, pstate,
+        AppLayerResult res = p->Parser[direction](f, alstate, pstate,
                 input, input_len,
                 alp_tctx->alproto_local_storage[f->protomap][alproto],
                 flags);
         if (res.status < 0)
         {
             goto error;
+        } else if (res.status > 0) {
+            if (f->proto == IPPROTO_TCP && f->protoctx != NULL) {
+                TcpSession *ssn = f->protoctx;
+                SCLogDebug("direction %d/%s", direction,
+                        (flags & STREAM_TOSERVER) ? "toserver" : "toclient");
+                BUG_ON(res.consumed > input_len);
+                if (direction == 0) {
+                    /* parser told us how much data it needs on top of what it
+                     * consumed. So we need tell stream engine how much we need
+                     * before the next call */
+                    ssn->client.data_required = res.needed;
+                    SCLogDebug("setting data_required %u", ssn->client.data_required);
+                } else {
+                    /* parser told us how much data it needs on top of what it
+                     * consumed. So we need tell stream engine how much we need
+                     * before the next call */
+                    ssn->server.data_required = res.needed;
+                    SCLogDebug("setting data_required %u", ssn->server.data_required);
+                }
+            } else {
+                /* incomplete is only supported for TCP */
+                BUG_ON(f->proto != IPPROTO_TCP);
+            }
+            BUG_ON(res.needed + res.consumed < input_len);
+            BUG_ON(res.needed == 0);
+            consumed = res.consumed;
         }
     }
 
@@ -1296,6 +1326,12 @@ int AppLayerParserParse(ThreadVars *tv, AppLayerParserThreadCtx *alp_tctx, Flow
         AppLayerParserStreamTruncated(f->proto, alproto, alstate, flags);
 
  end:
+    /* update app progress */
+    if (f->proto == IPPROTO_TCP && f->protoctx != NULL) {
+        TcpSession *ssn = f->protoctx;
+        StreamTcpUpdateAppLayerProgress(ssn, direction, consumed);
+    }
+
     SCReturnInt(0);
  error:
     /* Set the no app layer inspection flag for both

src/app-layer-parser.h

@@ -53,6 +53,7 @@
 
 #define APP_LAYER_OK (AppLayerResult) { 0, 0, 0 }
 #define APP_LAYER_ERROR (AppLayerResult) { -1, 0, 0 }
+#define APP_LAYER_INCOMPLETE(c,n) (AppLayerResult) { 1, (c), (n) }
 
 int AppLayerParserProtoIsRegistered(uint8_t ipproto, AppProto alproto);
 

src/app-layer.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2011 Open Information Security Foundation
+/* Copyright (C) 2007-2020 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
@@ -448,8 +448,6 @@ static int TCPProtoDetect(ThreadVars *tv,
         PACKET_PROFILING_APP_END(app_tctx, f->alproto);
         if (r < 0)
             goto failure;
-        (*stream)->app_progress_rel += data_len;
-
     } else {
         /* if the ssn is midstream, we may end up with a case where the
          * start of an HTTP request is missing. We won't detect HTTP based
@@ -518,9 +516,6 @@ static int TCPProtoDetect(ThreadVars *tv,
                             f->alproto, flags,
                             data, data_len);
                     PACKET_PROFILING_APP_END(app_tctx, f->alproto);
-                    if (r >= 0) {
-                        (*stream)->app_progress_rel += data_len;
-                    }
 
                     AppLayerDecoderEventsSetEventRaw(&p->app_layer_events,
                             APPLAYER_DETECT_PROTOCOL_ONLY_ONE_DIRECTION);
@@ -602,7 +597,6 @@ int AppLayerHandleTCPData(ThreadVars *tv, TcpReassemblyThreadCtx *ra_ctx,
                 flags, data, data_len);
         PACKET_PROFILING_APP_END(app_tctx, f->alproto);
         /* ignore parser result for gap */
-        (*stream)->app_progress_rel += data_len;
         goto end;
     }
 
@@ -660,9 +654,6 @@ int AppLayerHandleTCPData(ThreadVars *tv, TcpReassemblyThreadCtx *ra_ctx,
             r = AppLayerParserParse(tv, app_tctx->alp_tctx, f, f->alproto,
                                     flags, data, data_len);
             PACKET_PROFILING_APP_END(app_tctx, f->alproto);
-            if (r >= 0) {
-                (*stream)->app_progress_rel += data_len;
-            }
         }
     }
 

src/stream-tcp-private.h

@@ -117,6 +117,7 @@ typedef struct TcpStream_ {
 
     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 */
+    uint32_t data_required;         /**< data required from STREAM_APP_PROGRESS before calling app-layer again */
 
     StreamingBuffer sb;
     struct TCPSEG seg_tree;         /**< red black tree of TCP segments. Data is stored in TcpStream::sb */

src/stream-tcp-reassemble.c

@@ -1038,6 +1038,14 @@ static int ReassembleUpdateAppLayer (ThreadVars *tv,
             /* AppLayerHandleTCPData has likely updated progress. */
             app_progress = STREAM_APP_PROGRESS(*stream);
 
+            /* a GAP also consumes 'data required'. TODO perhaps we can use
+             * this to skip post GAP data until the start of a next record. */
+            if ((*stream)->data_required > mydata_len) {
+                (*stream)->data_required -= mydata_len;
+            } else {
+                (*stream)->data_required = 0;
+            }
+
             if (r < 0)
                 return 0;
 
@@ -1083,6 +1091,13 @@ static int ReassembleUpdateAppLayer (ThreadVars *tv,
             }
         }
     }
+    if ((p->flags & PKT_PSEUDO_STREAM_END) == 0 || ssn->state < TCP_CLOSED) {
+        if (mydata_len < (*stream)->data_required) {
+            SCLogDebug("mydata_len %u data_required %u", mydata_len, (*stream)->data_required);
+            SCReturnInt(0);
+        }
+    }
+    (*stream)->data_required = 0;
 
     /* update the app-layer */
     (void)AppLayerHandleTCPData(tv, ra_ctx, p, p->flow, ssn, stream,

src/stream-tcp.c

@@ -5849,6 +5849,21 @@ invalid:
     SCReturnInt(-1);
 }
 
+/** \brief update reassembly progress
+
+ * \param ssn TCP Session
+ * \param direction direction to set the flag in: 0 toserver, 1 toclient
+ */
+void StreamTcpUpdateAppLayerProgress(TcpSession *ssn, char direction,
+        const uint32_t progress)
+{
+    if (direction) {
+        ssn->server.app_progress_rel += progress;
+    } else {
+        ssn->client.app_progress_rel += progress;
+    }
+}
+
 /** \brief disable reassembly
 
  *  Disable app layer and set raw inspect to no longer accept new data.

src/stream-tcp.h

@@ -193,5 +193,8 @@ int StreamTcpInlineMode(void);
 
 int TcpSessionPacketSsnReuse(const Packet *p, const Flow *f, const void *tcp_ssn);
 
+void StreamTcpUpdateAppLayerProgress(TcpSession *ssn, char direction,
+        const uint32_t progress);
+
 #endif /* __STREAM_TCP_H__ */