flow timeout cleanup and fix

Flow timeout code worked by luck when checking if a flow still needed
reassembly for app layer inspection or logging. It would check for a
part of raw reassembly (smsg list) to determine if detection was
needed. In this case it would also process app layer cleanup,
including logging.

Introduced AppLayerTransactionGetActive which returns the lowest tx_id
in a direction that still needs some work.

FlowForceReassemblyNeedReassmbly now uses it to determine if the
applayer still needs work.

Converted FlowForceReassemblyForHash to use the checking function
FlowForceReassemblyNeedReassmbly as well, so that checking if a flow
needs work is now unified.

diff --git a/src/app-layer-parser.c b/src/app-layer-parser.c
index b9f98836cb03ded04059cbebff754151e03bd9fa..4fb9bf3b37c8cfdaa8fcc651ee964d059e13376d 100644 (file)
--- a/src/app-layer-parser.c
+++ b/src/app-layer-parser.c
@@ -1220,6 +1220,23 @@ error:
     SCReturnInt(-1);
 }
 
+/**
+ *  \brief Get 'active' tx id, meaning the lowest id that still need work.
+ *
+ *  \retval id tx id
+ */
+uint64_t AppLayerTransactionGetActive(Flow *f, uint8_t flags) {
+    AppLayerProto *p = &al_proto_table[f->alproto];
+    uint64_t log_id = ((AppLayerParserStateStore *)f->alparser)->log_id;
+    uint64_t inspect_id = ((AppLayerParserStateStore *)f->alparser)->
+        inspect_id[flags & STREAM_TOSERVER ? 0 : 1];
+    if (p->logger == TRUE) {
+        return (log_id < inspect_id) ? log_id : inspect_id;
+    } else {
+        return inspect_id;
+    }
+}
+
 void AppLayerTransactionUpdateLogId(Flow *f)
 {
     DEBUG_ASSERT_FLOW_LOCKED(f);

diff --git a/src/app-layer-parser.h b/src/app-layer-parser.h
index 9b2e270b349145661f9b559c5e0e3d7c0172d334..1b0a0321949375f35dc53fc27176d66d7d0ec499 100644 (file)
--- a/src/app-layer-parser.h
+++ b/src/app-layer-parser.h
@@ -344,6 +344,7 @@ void AppLayerTransactionUpdateInspectId(Flow *f, uint8_t direction);
  */
 uint64_t AppLayerTransactionGetInspectId(Flow *f, uint8_t flags);
 
+uint64_t AppLayerTransactionGetActive(Flow *f, uint8_t flags);
 
 
 void AppLayerSetEOF(Flow *);

diff --git a/src/flow-manager.c b/src/flow-manager.c
index 86869d7346af79f1390c12aae8d925f3efe1d777..f158fda85c44a388b7b6a9b6997f0a565b71f7d7 100644 (file)
--- a/src/flow-manager.c
+++ b/src/flow-manager.c
@@ -228,7 +228,8 @@ static int FlowManagerFlowTimedOut(Flow *f, struct timeval *ts) {
     }
 
     int server = 0, client = 0;
-    if (FlowForceReassemblyNeedReassmbly(f, &server, &client) == 1) {
+    if (!(f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) &&
+            FlowForceReassemblyNeedReassmbly(f, &server, &client) == 1) {
         FlowForceReassemblyForFlowV2(f, server, client);
         return 0;
     }

diff --git a/src/flow-timeout.c b/src/flow-timeout.c
index a3de06acca3d12b2e169a2a72a0d29c98b3ffd03..ac9d5f2d174605e9bb6240377d916e8c4f87c79d 100644 (file)
--- a/src/flow-timeout.c
+++ b/src/flow-timeout.c
@@ -292,15 +292,16 @@ static inline Packet *FlowForceReassemblyPseudoPacketGet(int direction,
 int FlowForceReassemblyNeedReassmbly(Flow *f, int *server, int *client) {
     TcpSession *ssn;
 
-    /* looks like we have no flows in this queue */
-    if (f == NULL || (f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE)) {
-        return 0;
+    if (f == NULL) {
+        *server = *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NONE;
+        SCReturnInt(0);
     }
 
     /* Get the tcp session for the flow */
     ssn = (TcpSession *)f->protoctx;
     if (ssn == NULL) {
-        return 0;
+        *server = *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NONE;
+        SCReturnInt(0);
     }
 
     *client = StreamNeedsReassembly(ssn, 0);
@@ -317,13 +318,29 @@ int FlowForceReassemblyNeedReassmbly(Flow *f, int *server, int *client) {
             *server = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
     }
 
+    /* if app layer still needs some love, push through */
+    if (f->alproto != ALPROTO_UNKNOWN && f->alstate != NULL &&
+        AppLayerAlprotoSupportsTxs(f->alproto))
+    {
+        uint64_t total_txs = AppLayerGetTxCnt(f->alproto, f->alstate);
+
+        if (AppLayerTransactionGetActive(f, STREAM_TOCLIENT) < total_txs) {
+            if (*server != STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY)
+                *server = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
+        }
+        if (AppLayerTransactionGetActive(f, STREAM_TOSERVER) < total_txs) {
+            if (*client != STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY)
+                *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
+        }
+    }
+
     /* nothing to do */
     if (*client == STREAM_HAS_UNPROCESSED_SEGMENTS_NONE &&
         *server == STREAM_HAS_UNPROCESSED_SEGMENTS_NONE) {
-        return 0;
+        SCReturnInt(0);
     }
 
-    return 1;
+    SCReturnInt(1);
 }
 
 /**
@@ -528,8 +545,10 @@ static inline void FlowForceReassemblyForHash(void)
                 continue;
             }
 
+            (void)FlowForceReassemblyNeedReassmbly(f, &server_ok, &client_ok);
+
             /* ah ah!  We have some unattended toserver segments */
-            if ((client_ok = StreamNeedsReassembly(ssn, 0)) == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
+            if (client_ok == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
                 StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->client.last_ack = (ssn->client.seg_list_tail->seq +
@@ -547,7 +566,7 @@ static inline void FlowForceReassemblyForHash(void)
                 }
             }
             /* oh oh!  We have some unattended toclient segments */
-            if ((server_ok = StreamNeedsReassembly(ssn, 1)) == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
+            if (server_ok == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
                 StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->server.last_ack = (ssn->server.seg_list_tail->seq +