Applayer to flow fixes and cleanups.

src/app-layer-detect-proto.c

@@ -145,15 +145,20 @@ static void AlpProtoFreeSignature(AlpProtoSignature *s) {
  *  \param s signature
  *  \param buf pointer to buffer
  *  \param buflen length of the buffer
+ *  \param ip_proto packet's ip_proto
  *
  *  \retval proto the detected proto or ALPROTO_UNKNOWN if no match
  */
 static uint16_t AlpProtoMatchSignature(AlpProtoSignature *s, uint8_t *buf,
-        uint16_t buflen)
+        uint16_t buflen, uint16_t ip_proto)
 {
     SCEnter();
     uint16_t proto = ALPROTO_UNKNOWN;
 
+    if (s->ip_proto != ip_proto) {
+        goto end;
+    }
+
     if (s->co->offset > buflen) {
         SCLogDebug("s->co->offset (%"PRIu16") > buflen (%"PRIu16")",
                 s->co->offset, buflen);
@@ -499,9 +504,8 @@ uint16_t AppLayerDetectGetProto(AlpProtoDetectCtx *ctx, AlpProtoDetectThreadCtx
     uint8_t s_cnt = 1;
 
     while (proto == ALPROTO_UNKNOWN && s != NULL) {
-        /* TCP or UPD? */
+        proto = AlpProtoMatchSignature(s, buf, buflen, ipproto);
-        if (s->ip_proto == ipproto)
+
-            proto = AlpProtoMatchSignature(s, buf, buflen);
         s = s->map_next;
         if (s == NULL && s_cnt < tdir->pmq.pattern_id_array_cnt) {
             patid = tdir->pmq.pattern_id_array[s_cnt];

src/app-layer-parser.c

@@ -460,6 +460,21 @@ int AlpParseFieldByDelimiter(AppLayerParserResult *output, AppLayerParserState *
     SCReturnInt(0);
 }
 
+/** app layer id counter */
+static uint8_t al_module_id = 0;
+
+/** \brief Get a unique app layer id
+ */
+uint8_t AppLayerRegisterModule(void) {
+    uint8_t id = al_module_id;
+    al_module_id++;
+    return id;
+}
+
+uint8_t AppLayerGetStorageSize(void) {
+    return al_module_id;
+}
+
 /** \brief Get the Parsers id for storing the parser state.
  *
  * \retval Parser subsys id
@@ -563,7 +578,7 @@ int AppLayerRegisterProto(char *name, uint8_t proto, uint8_t flags,
     }
 
     if (al_proto_table[proto].storage_id == 0) {
-        al_proto_table[proto].storage_id = StreamL7RegisterModule();
+        al_proto_table[proto].storage_id = AppLayerRegisterModule();
     }
 
     SCLogDebug("registered %p at proto %" PRIu32 " flags %02X, al_proto_table "
@@ -864,7 +879,7 @@ int AppLayerParse(Flow *f, uint8_t proto, uint8_t flags, uint8_t *input,
             goto error;
     }
 
-    /* set the packets to no inspection and reassembly for the TLS sessions */
+    /* set the packets to no inspection and reassembly if required */
     if (parser_state->flags & APP_LAYER_PARSER_NO_INSPECTION) {
         FlowSetNoPayloadInspectionFlag(f);
         FlowSetSessionNoApplayerInspectionFlag(f);
@@ -1109,7 +1124,7 @@ void RegisterAppLayerParsers(void)
     memset(&al_proto_table, 0, sizeof(al_proto_table));
     memset(&al_parser_table, 0, sizeof(al_parser_table));
 
-    app_layer_sid = StreamL7RegisterModule();
+    app_layer_sid = AppLayerRegisterModule();
 
     /** setup result pool
      * \todo Per thread pool */

src/app-layer-parser.h

@@ -171,8 +171,12 @@ int AppLayerTransactionGetBaseId(Flow *f);
 
 void AppLayerParserRegisterTests(void);
 
-#include "stream-tcp-private.h"
 void AppLayerParserCleanupState(Flow *);
 
+
+uint8_t AppLayerRegisterModule(void);
+uint8_t AppLayerGetStorageSize(void);
+
+
 #endif /* __APP_LAYER_PARSER_H__ */
 

src/app-layer.c

@@ -44,8 +44,7 @@ uint16_t AppLayerGetProtoFromPacket(Packet *p) {
         SCReturnUInt(ALPROTO_UNKNOWN);
     }
 
-    TcpSession *ssn = (TcpSession *)p->flow->protoctx;
+    if (p->flow->aldata == NULL) {
-    if (ssn == NULL && p->flow->aldata == NULL) {
         SCReturnUInt(ALPROTO_UNKNOWN);
     }
 
@@ -65,8 +64,7 @@ void *AppLayerGetProtoStateFromPacket(Packet *p) {
         SCReturnPtr(NULL, "void");
     }
 
-    TcpSession *ssn = (TcpSession *)p->flow->protoctx;
+    if (p->flow->aldata == NULL) {
-    if (ssn == NULL && p->flow->aldata == NULL) {
         SCReturnPtr(NULL, "void");
     }
 
@@ -85,12 +83,13 @@ void *AppLayerGetProtoStateFromPacket(Packet *p) {
 void *AppLayerGetProtoStateFromFlow(Flow *f) {
     SCEnter();
 
-    if (f == NULL)
+    if (f == NULL) {
         SCReturnPtr(NULL, "void");
+    }
 
-    TcpSession *ssn = (TcpSession *)f->protoctx;
+    if (f->aldata == NULL) {
-    if (ssn == NULL || f->aldata == NULL)
         SCReturnPtr(NULL, "void");
+    }
 
     SCLogDebug("f->alproto %"PRIu16"", f->alproto);
 
@@ -259,7 +258,8 @@ int AppLayerHandleMsg(AlpProtoDetectThreadCtx *dp_ctx, StreamMsg *smsg)
  *  If the protocol is yet unknown, the proto detection code is run first.
  *
  *  \param dp_ctx Thread app layer detect context
- *  \param smsg Stream message
+ *  \param f unlocked flow
+ *  \param p UDP packet
  *
  *  \retval 0 ok
  *  \retval -1 error
@@ -271,12 +271,15 @@ int AppLayerHandleUdp(AlpProtoDetectThreadCtx *dp_ctx, Flow *f, Packet *p)
     uint16_t alproto = ALPROTO_UNKNOWN;
     int r = 0;
 
-    if (f == NULL)
+    if (f == NULL) {
-        return r;
+        SCReturnInt(r);
+    }
+
+    SCMutexLock(&f->m);
 
     alproto = f->alproto;
 
-    if (FlowGetPacketDirection(f,p) == TOSERVER) {
+    if (p->flowflags & FLOW_PKT_TOSERVER) {
         f->alflags |= FLOW_AL_STREAM_TOSERVER;
     } else {
         f->alflags |= FLOW_AL_STREAM_TOCLIENT;
@@ -327,6 +330,7 @@ int AppLayerHandleUdp(AlpProtoDetectThreadCtx *dp_ctx, Flow *f, Packet *p)
         }
     }
 
+    SCMutexUnlock(&f->m);
     SCReturnInt(r);
 }
 

src/decode-udp.c

@@ -77,7 +77,11 @@ void DecodeUDP(ThreadVars *tv, DecodeThreadVars *dtv, Packet *p, uint8_t *pkt, u
 
     /* Flow is an integral part of us */
     FlowHandlePacket(tv, p);
-    AppLayerHandleUdp(&dtv->udp_dp_ctx, p->flow, p);
+
+    /* handle the app layer part of the UDP packet payload */
+    if (p->flow != NULL) {
+        AppLayerHandleUdp(&dtv->udp_dp_ctx, p->flow, p);
+    }
 
     return;
 }

src/decode.c

@@ -223,6 +223,7 @@ void AddressDebugPrint(Address *a) {
     }
 }
 
+/** \brief Alloc and setup DecodeThreadVars */
 DecodeThreadVars *DecodeThreadVarsAlloc() {
 
     DecodeThreadVars *dtv = NULL;
@@ -231,6 +232,8 @@ DecodeThreadVars *DecodeThreadVarsAlloc() {
         return NULL;
 
     memset(dtv, 0, sizeof(DecodeThreadVars));
+
+    /* initialize UDP app layer code */
     AlpProtoFinalize2Thread(&dtv->udp_dp_ctx);
 
     return dtv;

src/detect-engine-state.c

@@ -250,22 +250,24 @@ int DeStateDetectStartDetection(ThreadVars *tv, DetectEngineThreadCtx *det_ctx,
     int match = 0;
     int r = 0;
 
-    if (alstate != NULL) {
+    if (alstate == NULL) {
-        for ( ; sm != NULL; sm = sm->next) {
+        SCReturnInt(0);
-            SCLogDebug("sm %p, sm->next %p", sm, sm->next);
+    }
-
+
-            if (sigmatch_table[sm->type].AppLayerMatch != NULL &&
+    for ( ; sm != NULL; sm = sm->next) {
-                    alproto == sigmatch_table[sm->type].alproto &&
+        SCLogDebug("sm %p, sm->next %p", sm, sm->next);
-                    alstate != NULL)
+
-            {
+        if (sigmatch_table[sm->type].AppLayerMatch != NULL &&
-                match = sigmatch_table[sm->type].AppLayerMatch(tv, det_ctx, f, flags, alstate, s, sm);
+                alproto == sigmatch_table[sm->type].alproto)
-                if (match == 0) {
+        {
-                    break;
+            match = sigmatch_table[sm->type].AppLayerMatch(tv, det_ctx, f,
-                } else if (sm->next == NULL) {
+                    flags, alstate, s, sm);
-                    r = 1;
+            if (match == 0) {
-                    sm = NULL; /* set to NULL as we have a match */
+                break;
-                    break;
+            } else if (sm->next == NULL) {
-                }
+                r = 1;
+                sm = NULL; /* set to NULL as we have a match */
+                break;
             }
         }
     }

src/flow-util.h

@@ -47,7 +47,7 @@
         (f)->sgh_toserver = NULL; \
         (f)->sgh_toclient = NULL; \
         (f)->aldata = NULL; \
-        (f)->alflags = FLOW_AL_PROTO_UNKNOWN; \
+        (f)->alflags = 0; \
         (f)->alproto = 0; \
     } while (0)
 
@@ -76,7 +76,7 @@
         FlowL7DataPtrFree(f); \
         SCFree((f)->aldata); \
         (f)->aldata = NULL; \
-        (f)->alflags = FLOW_AL_PROTO_UNKNOWN; \
+        (f)->alflags = 0; \
         (f)->alproto = 0; \
     } while(0)
 
@@ -92,7 +92,7 @@
         FlowL7DataPtrFree(f); \
         SCFree((f)->aldata); \
         (f)->aldata = NULL; \
-        (f)->alflags = FLOW_AL_PROTO_UNKNOWN; \
+        (f)->alflags = 0; \
         (f)->alproto = 0; \
     } while(0)
 

src/flow.c

@@ -54,6 +54,7 @@
 #include "detect.h"
 #include "detect-engine-state.h"
 #include "stream.h"
+
 #include "app-layer-parser.h"
 
 #define FLOW_DEFAULT_EMERGENCY_RECOVERY 30
@@ -92,7 +93,7 @@ void FlowL7DataPtrInit(Flow *f) {
     if (f->aldata != NULL)
         return;
 
-    uint32_t size = (uint32_t)(sizeof (void *) * StreamL7GetStorageSize());
+    uint32_t size = (uint32_t)(sizeof (void *) * AppLayerGetStorageSize());
 
 /////////XXXPR pass to flow memcap   if (StreamTcpCheckMemcap(size) == 0)
 /////////XXXPR pass to flow memcap        return;
@@ -102,7 +103,7 @@ void FlowL7DataPtrInit(Flow *f) {
       //  StreamTcpIncrMemuse(size);
 
         uint8_t u;
-        for (u = 0; u < StreamL7GetStorageSize(); u++) {
+        for (u = 0; u < AppLayerGetStorageSize(); u++) {
             f->aldata[u] = NULL;
         }
     }

src/stream-tcp-reassemble.h

@@ -25,6 +25,7 @@
 #ifndef __STREAM_TCP_REASSEMBLE_H__
 #define __STREAM_TCP_REASSEMBLE_H__
 
+#include "stream-tcp-private.h"
 #include "stream.h"
 #include "app-layer-detect-proto.h"
 

src/stream-tcp.h

@@ -25,6 +25,8 @@
 #ifndef __STREAM_TCP_H__
 #define __STREAM_TCP_H__
 
+#include "stream-tcp-private.h"
+
 #define COUNTER_STREAMTCP_STREAMS 1
 
 #define STREAM_VERBOSE    FALSE

src/stream.c

@@ -185,11 +185,6 @@ StreamMsgQueue *StreamMsgQueueGetByPort(uint16_t port) {
     return NULL;//&stream_q;
 }
 
-/* XXX hack */
-void StreamMsgSignalQueueHack(void) {
-    //SCCondSignal(&stream_q.cond_q);
-}
-
 void StreamMsgQueueSetMinInitChunkLen(uint8_t dir, uint16_t len) {
     if (dir == FLOW_PKT_TOSERVER) {
         toserver_min_init_chunk_len = len;
@@ -222,16 +217,3 @@ uint16_t StreamMsgQueueGetMinChunkLen(uint8_t dir) {
     }
 }
 
-/* StreamL7RegisterModule
- */
-static uint8_t l7_module_id = 0;
-uint8_t StreamL7RegisterModule(void) {
-    uint8_t id = l7_module_id;
-    l7_module_id++;
-    return id;
-}
-
-uint8_t StreamL7GetStorageSize(void) {
-    return l7_module_id;
-}
-

src/stream.h

@@ -90,10 +90,5 @@ void StreamMsgQueueSetMinChunkLen(uint8_t dir, uint16_t len);
 uint16_t StreamMsgQueueGetMinInitChunkLen(uint8_t);
 uint16_t StreamMsgQueueGetMinChunkLen(uint8_t);
 
-void StreamMsgSignalQueueHack(void);
-
-uint8_t StreamL7RegisterModule(void);
-uint8_t StreamL7GetStorageSize(void);
-
 #endif /* __STREAM_H__ */
 

src/tm-threads.c

@@ -1109,9 +1109,6 @@ void TmThreadKillThreads(void) {
             TmThreadsSetFlag(tv, THV_KILL);
             SCLogDebug("told thread %s to stop", tv->name);
 
-            /* XXX hack */
-            StreamMsgSignalQueueHack();
-
             if (tv->inq != NULL) {
                 int i;