flow/stream: 'wrong thread' as stream event & counter

Set event at most once per flow, for the first 'wrong' packet.

Add 'tcp.pkt_on_wrong_thread' counter. This is incremented for each
'wrong' packet. Note that the first packet for a flow determines
what thread is 'correct'.

rules/stream-events.rules

@@ -88,6 +88,8 @@ alert tcp any any -> any any (msg:"SURICATA STREAM reassembly segment before bas
 alert tcp any any -> any any (msg:"SURICATA STREAM Packet is retransmission"; stream-event:pkt_retransmission; flowint:tcp.retransmission.count,+,1; noalert; classtype:protocol-command-decode; sid:2210053; rev:1;)
 # rule to alert if a stream has excessive retransmissions
 alert tcp any any -> any any (msg:"SURICATA STREAM excessive retransmissions"; flowbits:isnotset,tcp.retransmission.alerted; flowint:tcp.retransmission.count,>=,10; flowbits:set,tcp.retransmission.alerted; classtype:protocol-command-decode; sid:2210054; rev:1;)
+# Packet on wrong thread. Fires at most once per flow.
+alert tcp any any -> any any (msg:"SURICATA STREAM pkt seen on wrong thread"; stream-event:wrong_thread; sid:2210059; rev:1;)
 
-# next sid 2210059
+# next sid 2210060
 

src/decode-events.c

@@ -240,6 +240,7 @@ const struct DecodeEvents_ DEvents[] = {
     { "stream.pkt_bad_window_update", STREAM_PKT_BAD_WINDOW_UPDATE, },
 
     { "stream.suspected_rst_inject", STREAM_SUSPECTED_RST_INJECT, },
+    { "stream.wrong_thread", STREAM_WRONG_THREAD, },
 
     { "stream.reassembly_segment_before_base_seq", STREAM_REASSEMBLY_SEGMENT_BEFORE_BASE_SEQ, },
     { "stream.reassembly_no_segment", STREAM_REASSEMBLY_NO_SEGMENT, },

src/decode-events.h

@@ -250,6 +250,7 @@ enum {
     STREAM_PKT_BAD_WINDOW_UPDATE,
 
     STREAM_SUSPECTED_RST_INJECT,
+    STREAM_WRONG_THREAD,
 
     STREAM_REASSEMBLY_SEGMENT_BEFORE_BASE_SEQ,
     STREAM_REASSEMBLY_NO_SEGMENT,

src/flow.h

@@ -100,6 +100,8 @@ typedef struct AppLayerParserState_ AppLayerParserState;
 /** Indicate that alproto detection for flow should be done again */
 #define FLOW_CHANGE_PROTO               BIT_U32(24)
 
+#define FLOW_WRONG_THREAD               BIT_U32(25)
+
 /* File flags */
 
 /** no magic on files in this flow */

src/output-json-flow.c

@@ -274,6 +274,8 @@ static void JsonFlowLogJSON(JsonFlowLogThread *aft, json_t *js, Flow *f)
             json_string(reason));
 
     json_object_set_new(hjs, "alerted", json_boolean(FlowHasAlerts(f)));
+    if (f->flags & FLOW_WRONG_THREAD)
+        json_object_set_new(hjs, "wrong_thread", json_true());
 
     json_object_set_new(js, "flow", hjs);
 

src/stream-tcp.c

@@ -4679,10 +4679,15 @@ int StreamTcpPacket (ThreadVars *tv, Packet *p, StreamTcpThread *stt,
     /* assign the thread id to the flow */
     if (unlikely(p->flow->thread_id == 0)) {
         p->flow->thread_id = (FlowThreadId)tv->id;
-#ifdef DEBUG
     } else if (unlikely((FlowThreadId)tv->id != p->flow->thread_id)) {
         SCLogDebug("wrong thread: flow has %u, we are %d", p->flow->thread_id, tv->id);
-#endif
+        if (p->pkt_src == PKT_SRC_WIRE) {
+            StatsIncr(tv, stt->counter_tcp_wrong_thread);
+            if ((p->flow->flags & FLOW_WRONG_THREAD) == 0) {
+                p->flow->flags |= FLOW_WRONG_THREAD;
+                StreamTcpSetEvent(p, STREAM_WRONG_THREAD);
+            }
+        }
     }
 
     TcpSession *ssn = (TcpSession *)p->flow->protoctx;
@@ -5144,6 +5149,7 @@ TmEcode StreamTcpThreadInit(ThreadVars *tv, void *initdata, void **data)
     stt->counter_tcp_synack = StatsRegisterCounter("tcp.synack", tv);
     stt->counter_tcp_rst = StatsRegisterCounter("tcp.rst", tv);
     stt->counter_tcp_midstream_pickups = StatsRegisterCounter("tcp.midstream_pickups", tv);
+    stt->counter_tcp_wrong_thread = StatsRegisterCounter("tcp.pkt_on_wrong_thread", tv);
 
     /* init reassembly ctx */
     stt->ra_ctx = StreamTcpReassembleInitThreadCtx(tv);

src/stream-tcp.h

@@ -96,6 +96,8 @@ typedef struct StreamTcpThread_ {
     uint16_t counter_tcp_rst;
     /** midstream pickups */
     uint16_t counter_tcp_midstream_pickups;
+    /** wrong thread */
+    uint16_t counter_tcp_wrong_thread;
 
     /** tcp reassembly thread data */
     TcpReassemblyThreadCtx *ra_ctx;