stream and application layer improvements

src/app-layer-detect-proto.c

@@ -54,8 +54,12 @@ typedef struct AlpProtoDetectDirection_ {
     uint32_t id;
     uint16_t map[ALP_DETECT_MAX];   /**< a mapping between condition id's and
                                          protocol */
-    uint16_t max_depth;             /**< max depth of all patterns, so we can
+    uint16_t max_len;              /**< max length of all patterns, so we can
                                          limit the search */
+    uint16_t min_len;              /**< min length of all patterns, so we can
+                                         tell the stream engine to feed data
+                                         to app layer as soon as it has min
+                                         size data */
 } AlpProtoDetectDirection;
 
 typedef struct AlpProtoDetectCtx_ {
@@ -86,6 +90,8 @@ void AlpProtoInit(AlpProtoDetectCtx *ctx) {
 
     ctx->toserver.id = 0;
     ctx->toclient.id = 0;
+    ctx->toclient.min_len = INSPECT_BYTES;
+    ctx->toserver.min_len = INSPECT_BYTES;
 }
 
 void AlpProtoTestDestroy(AlpProtoDetectCtx *ctx) {
@@ -131,8 +137,13 @@ void AlpProtoAdd(AlpProtoDetectCtx *ctx, uint16_t ip_proto, uint16_t al_proto, c
     dir->map[dir->id] = al_proto;
     dir->id++;
 
-    if (depth > dir->max_depth)
+    if (depth > dir->max_len)
-        dir->max_depth = depth;
+        dir->max_len = depth;
+
+    /* set the min_len for the stream engine to set the min smsg size for app
+       layer*/
+    if (depth < dir->min_len)
+        dir->min_len = depth;
 
     /* no longer need the cd */
     DetectContentFree(cd);
@@ -189,9 +200,10 @@ void AlpProtoFinalizeGlobal(AlpProtoDetectCtx *ctx) {
         exit(EXIT_FAILURE);
 #endif
 
-    /* tell the stream reassembler we only want chunks of size max_depth */
+    /* tell the stream reassembler, that initially we only want chunks of size
-    StreamMsgQueueSetMinInitChunkLen(FLOW_PKT_TOCLIENT, ctx->toclient.max_depth);
+       min_len */
-    StreamMsgQueueSetMinInitChunkLen(FLOW_PKT_TOSERVER, ctx->toserver.max_depth);
+    StreamMsgQueueSetMinInitChunkLen(FLOW_PKT_TOCLIENT, ctx->toclient.min_len);
+    StreamMsgQueueSetMinInitChunkLen(FLOW_PKT_TOSERVER, ctx->toserver.min_len);
 }
 
 void AppLayerDetectProtoThreadInit(void) {
@@ -305,8 +317,8 @@ uint16_t AppLayerDetectGetProto(AlpProtoDetectCtx *ctx, AlpProtoDetectThreadCtx
 
     /* see if we can limit the data we inspect */
     uint16_t searchlen = buflen;
-    if (searchlen > dir->max_depth)
+    if (searchlen > dir->max_len)
-        searchlen = dir->max_depth;
+        searchlen = dir->max_len;
 
     uint16_t proto = ALPROTO_UNKNOWN;
     uint32_t cnt = 0;
@@ -448,21 +460,24 @@ int AppLayerHandleMsg(AlpProtoDetectThreadCtx *dp_ctx, StreamMsg *smsg)
                 /* store the proto and setup the L7 data array */
                 StreamL7DataPtrInit(ssn);
                 ssn->alproto = alproto;
+                ssn->flags |= STREAMTCP_FLAG_APPPROTO_DETECTION_COMPLETED;
 
                 r = AppLayerParse(smsg->flow, alproto, smsg->flags,
                                smsg->data.data, smsg->data.data_len);
             } else {
-                SCLogDebug("ALPROTO_UNKNOWN flow %p", smsg->flow);
+                if (smsg->flags & STREAM_TOSERVER) {
-
+                    if (smsg->data.data_len >= alp_proto_ctx.toserver.max_len) {
-                TcpSession *ssn = smsg->flow->protoctx;
+                        ssn->flags |= STREAMTCP_FLAG_APPPROTO_DETECTION_COMPLETED;
-                if (ssn != NULL) {
+                        SCLogDebug("ALPROTO_UNKNOWN flow %p", smsg->flow);
-                    if (smsg->flags & STREAM_TOCLIENT) {
-                        StreamTcpSetSessionNoReassemblyFlag(ssn, 1);
-                    } else if (smsg->flags & STREAM_TOSERVER) {
                         StreamTcpSetSessionNoReassemblyFlag(ssn, 0);
                     }
+                } else if (smsg->flags & STREAM_TOCLIENT) {
+                    if (smsg->data.data_len >= alp_proto_ctx.toclient.max_len) {
+                        ssn->flags |= STREAMTCP_FLAG_APPPROTO_DETECTION_COMPLETED;
+                        SCLogDebug("ALPROTO_UNKNOWN flow %p", smsg->flow);
+                        StreamTcpSetSessionNoReassemblyFlag(ssn, 1);
+                    }
                 }
-
             }
         } else {
             SCLogDebug("stream data (len %" PRIu32 " (%" PRIu32 ")), alproto "

src/app-layer-htp.c

@@ -156,6 +156,12 @@ static int HTPHandleRequestData(Flow *f, void *htp_state,
 
     HtpState *hstate = (HtpState *)htp_state;
 
+    if (hstate->connp->in_status == STREAM_STATE_ERROR) {
+        SCLogError(SC_ERR_ALPARSER, "Inbound parser is in error state, no"
+                " need to feed data to libhtp");
+        SCReturnInt(-1);
+    }
+
     /* Unset the body inspection (the callback should
      * reactivate it if necessary) */
     hstate->flags &= ~HTP_FLAG_NEW_BODY_SET;
@@ -242,6 +248,12 @@ static int HTPHandleResponseData(Flow *f, void *htp_state,
 
     HtpState *hstate = (HtpState *)htp_state;
 
+    if (hstate->connp->out_status == STREAM_STATE_ERROR) {
+        SCLogError(SC_ERR_ALPARSER, "Outbound parser is in error state, no"
+                " need to feed data to libhtp");
+        SCReturnInt(-1);
+    }
+
     /* Unset the body inspection (the callback should
      * reactivate it if necessary) */
     hstate->flags &= ~HTP_FLAG_NEW_BODY_SET;

src/app-layer-parser.c

@@ -1077,11 +1077,6 @@ static int AppLayerParserTest01 (void)
         goto end;
     }
 
-    if (ssn.aldata != NULL) {
-        printf("App Layer state has not been cleared\n");
-        result = 0;
-        goto end;
-    }
 end:
     StreamL7DataPtrFree(&ssn);
     StreamTcpFreeConfig(TRUE);

src/stream-tcp-private.h

@@ -7,6 +7,7 @@ typedef struct TcpSegment_ {
     uint16_t payload_len; /* actual size of the payload */
     uint32_t seq;
     uint16_t pool_size; /* size of the memory */
+    uint8_t flags;
     struct TcpSegment_ *next;
     struct TcpSegment_ *prev;
 } TcpSegment;
@@ -30,6 +31,10 @@ typedef struct TcpStream_ {
     uint8_t os_policy; /**< target based OS policy used for reassembly and handling packets*/
     uint16_t flags;      /**< Flag specific to the stream e.g. Timestamp */
     TcpSegment *seg_list_tail;  /**< Last segment in the reassembled stream seg list*/
+    uint32_t tmp_ra_base_seq;   /**< Temporary reassembled seq, to be used until
+                                     app layer protocol has not been detected,
+                                     beacuse every smsg needs to contain all the
+                                     initial segments too */
 } TcpStream;
 
 /* from /usr/include/netinet/tcp.h */
@@ -49,20 +54,64 @@ enum
     TCP_CLOSED,
 };
 
-#define STREAMTCP_FLAG_MIDSTREAM                0x0001  /**< Flag for mid stream session*/
+#define STREAMTCP_FLAG_MIDSTREAM                0x0001  /**< Flag for mid stream
-#define STREAMTCP_FLAG_MIDSTREAM_ESTABLISHED    0x0002  /**< Flag for mid stream established session*/
+                                                             session*/
-#define STREAMTCP_FLAG_MIDSTREAM_SYNACK         0x0004  /**<Flag for mid session when syn/ack is received*/
+#define STREAMTCP_FLAG_MIDSTREAM_ESTABLISHED    0x0002  /**< Flag for mid stream
-#define STREAMTCP_FLAG_TIMESTAMP                0x0008  /**< Flag for TCP Timestamp option*/
+                                                             established
-#define STREAMTCP_FLAG_SERVER_WSCALE            0x0010  /**< Server supports wscale (even though it can be 0) */
+                                                             session*/
-#define STREAMTCP_FLAG_ZERO_TIMESTAMP           0x0020  /**< Flag to indicate the zero value of timestamp*/
+#define STREAMTCP_FLAG_MIDSTREAM_SYNACK         0x0004  /**< Flag for mid session
-#define STREAMTCP_FLAG_NOCLIENT_REASSEMBLY      0x0040  /**< Flag to avoid stream reassembly / application layer
+                                                             when syn/ack is
-                                                             inspection for the client stream.*/
+                                                             received*/
-#define STREAMTCP_FLAG_NOSERVER_REASSEMBLY      0x0080  /**< Flag to avoid stream reassembly / application layer
+#define STREAMTCP_FLAG_TIMESTAMP                0x0008  /**< Flag for TCP
-                                                             inspection for the server stream.*/
+                                                             Timestamp option*/
-#define STREAMTCP_FLAG_ASYNC                    0x0100  /**< Flag to indicate that the session is handling
+#define STREAMTCP_FLAG_SERVER_WSCALE            0x0010  /**< Server supports
-                                                             asynchronous stream.*/
+                                                             wscale (even though
-#define STREAMTCP_FLAG_4WHS                     0x0200  /**< Flag to indicate we're dealing with 4WHS:
+                                                             it can be 0) */
-                                                             SYN, SYN, SYN/ACK, ACK (http://www.breakingpointsystems.com/community/blog/tcp-portals-the-three-way-handshake-is-a-lie) */
+#define STREAMTCP_FLAG_ZERO_TIMESTAMP           0x0020  /**< Flag to indicate the
+                                                             zero value of
+                                                             timestamp*/
+#define STREAMTCP_FLAG_NOCLIENT_REASSEMBLY      0x0040  /**< Flag to avoid stream
+                                                             reassembly/app layer
+                                                             inspection for the
+                                                             client stream.*/
+#define STREAMTCP_FLAG_NOSERVER_REASSEMBLY      0x0080  /**< Flag to avoid stream
+                                                             reassembly / app layer
+                                                             inspection for the
+                                                             server stream.*/
+#define STREAMTCP_FLAG_ASYNC                    0x0100  /**< Flag to indicate
+                                                             that the session is
+                                                             handling asynchronous
+                                                             stream.*/
+#define STREAMTCP_FLAG_4WHS                     0x0200  /**< Flag to indicate
+                                                             we're dealing with
+                                                             4WHS: SYN, SYN,
+                                                             SYN/ACK, ACK
+ (http://www.breakingpointsystems.com/community/blog/tcp-portals-the-three-way-handshake-is-a-lie) */
+
+#define STREAMTCP_FLAG_APPPROTO_DETECTION_COMPLETED 0x0400  /**< Flag to indicate
+                                                             the app layer has
+                                                             detected the app
+                                                             layer protocol on
+                                                             the current
+                                                             TCP session */
+#define STREAMTCP_FLAG_PAUSE_TOSERVER_REASSEMBLY 0x0800 /**< Flag to pause stream
+                                                             reassembly / app layer
+                                                             inspection for the
+                                                             server stream.*/
+#define STREAMTCP_FLAG_PAUSE_TOCLIENT_REASSEMBLY 0x1000 /**< Flag to pause stream
+                                                             reassembly / app layer
+                                                             inspection for the
+                                                             client stream.*/
+
+#define SEGMENTTCP_FLAG_PROCESSED               0x01    /**< Flag to indicate
+                                                             that the current
+                                                             segment has been
+                                                             processed by the
+                                                             reassembly code and
+                                                             should be deleted
+                                                             after app layer
+                                                             protocol has been
+                                                             detected. */
 
 #define PAWS_24DAYS         2073600         /**< 24 days in seconds */
 

src/stream-tcp-reassemble.h

@@ -54,6 +54,8 @@ void StreamL7DataPtrFree(TcpSession *);
 
 void StreamTcpSetSessionNoReassemblyFlag (TcpSession *, char );
 void StreamTcpSetOSPolicy(TcpStream *, Packet *);
+void StreamTcpReassemblePause (TcpSession *, char );
+void StreamTcpReassembleUnPause (TcpSession *, char );
 
 #endif /* __STREAM_TCP_REASSEMBLE_H__ */
 

src/stream-tcp.c

@@ -2964,6 +2964,10 @@ static int StreamTcpTest02 (void) {
     StreamTcpThread stt;
     uint8_t payload[4];
     TCPHdr tcph;
+    TcpReassemblyThreadCtx ra_ctx;
+    StreamMsgQueue stream_q;
+    memset(&stream_q, 0, sizeof(StreamMsgQueue));
+    memset(&ra_ctx, 0, sizeof(TcpReassemblyThreadCtx));
     memset (&p, 0, sizeof(Packet));
     memset (&f, 0, sizeof(Flow));
     memset(&tv, 0, sizeof (ThreadVars));
@@ -2975,6 +2979,8 @@ static int StreamTcpTest02 (void) {
     p.tcph = &tcph;
     p.flowflags = FLOW_PKT_TOSERVER;
     int ret = 0;
+    ra_ctx.stream_q = &stream_q;
+    stt.ra_ctx = &ra_ctx;
 
     StreamTcpInitConfig(TRUE);
 

src/stream.c

@@ -144,7 +144,7 @@ void StreamMsgQueuesInit(void) {
 
     stream_msg_pool = PoolInit(5000,250,StreamMsgAlloc,NULL,StreamMsgFree);
     if (stream_msg_pool == NULL)
-        exit(1); /* XXX */
+        exit(EXIT_FAILURE); /* XXX */
 }
 
 void StreamMsgQueuesDeinit(char quiet) {

src/stream.h

@@ -13,6 +13,8 @@
 
 #define MSG_DATA_SIZE       512
 
+#define STREAMQUEUE_FLAG_INIT    0x01
+
 typedef struct StreamMsg_ {
     uint32_t id; /* unique stream id */
     uint8_t flags; /* msg flags */
@@ -42,6 +44,7 @@ typedef struct StreamMsgQueue_ {
     uint16_t len;
     SCMutex mutex_q;
     SCCondT cond_q;
+    uint8_t flags;
 #ifdef DBG_PERF
     uint16_t dbg_maxlen;
 #endif /* DBG_PERF */