From: Victor Julien <victor@inliniac.net>
Date: Tue, 3 Dec 2013 13:28:09 +0000 (+0100)
Subject: flow timeout cleanup and fix
X-Git-Tag: suricata-2.0beta2~61
X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=3b3dce8328007e66f9f8b7070c712da76321852b;p=thirdparty%2Fsuricata.git

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 b9f98836cb..4fb9bf3b37 100644
--- 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 9b2e270b34..1b0a032194 100644
--- 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 86869d7346..f158fda85c 100644
--- 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 a3de06acca..ac9d5f2d17 100644
--- 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 +