From: Davis McPherson -X (davmcphe - XORIANT CORPORATION at Cisco) <davmcphe@cisco.com>
Date: Tue, 1 Oct 2024 20:44:40 +0000 (+0000)
Subject: Pull request #4463: stream_tcp: implement flush on asymmetric flows in IDS mode when... 
X-Git-Tag: 3.4.0.0~13
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=087c1d6b77ea139ee4370f3c690fd144491c670e;p=thirdparty%2Fsnort3.git

Pull request #4463: stream_tcp: implement flush on asymmetric flows in IDS mode when queued bytes exceeds configured threshold

Merge in SNORT/snort3 from ~DAVMCPHE/snort3:ids_asymmetric_support to master

Squashed commit of the following:

commit 97bdb1f6e267a42f2f83dc62444a9c0bf97170ad
Author: davis mcpherson <davmcphe@cisco.com>
Date:   Thu Sep 5 09:30:58 2024 -0400

    stream_tcp: implement flush on asymmetric flows in IDS mode when queued bytes exceeds configure threshold
---

diff --git a/src/stream/tcp/tcp_module.cc b/src/stream/tcp/tcp_module.cc
index 0bfe2f7c9..894ffe553 100644
--- a/src/stream/tcp/tcp_module.cc
+++ b/src/stream/tcp/tcp_module.cc
@@ -238,6 +238,10 @@ static const Parameter s_params[] =
     { "idle_timeout", Parameter::PT_INT, "1:max31", "3600",
       "session deletion on idle " },
 
+    { "asymmetric_ids_flush_threshold", Parameter::PT_INT, "1:max31", "65535",
+      "max bytes queued on asymmetric flow before flush in IDS mode" },
+
+
     { nullptr, Parameter::PT_MAX, nullptr, nullptr, nullptr }
 };
 
@@ -331,6 +335,9 @@ bool StreamTcpModule::set(const char*, Value& v, SnortConfig*)
     else if ( v.is("max_segments") )
         config->max_queued_segs = v.get_uint32();
 
+    else if ( v.is("asymmetric_ids_flush_threshold") )
+        config->asymmetric_ids_flush_threshold = v.get_uint32();
+
     else if ( v.is("max_window") )
         config->max_window = v.get_uint32();
 
diff --git a/src/stream/tcp/tcp_reassembler.cc b/src/stream/tcp/tcp_reassembler.cc
index 03d117dc2..e7255ec57 100644
--- a/src/stream/tcp/tcp_reassembler.cc
+++ b/src/stream/tcp/tcp_reassembler.cc
@@ -595,7 +595,7 @@ bool TcpReassembler::final_flush_on_fin(int32_t flush_amt, Packet *p, FinSeqNumS
         && !p->flow->searching_for_service();
 }
 
-bool  TcpReassembler::flush_on_asymmetric_flow(uint32_t flushed, snort::Packet *p)
+bool TcpReassembler::asymmetric_flow_flushed(uint32_t flushed, snort::Packet *p)
 {
     bool asymmetric = flushed && seglist.seg_count && !p->flow->two_way_traffic() && !p->ptrs.tcph->is_syn();
     if ( asymmetric )
diff --git a/src/stream/tcp/tcp_reassembler.h b/src/stream/tcp/tcp_reassembler.h
index c7c03a960..66fed13e8 100644
--- a/src/stream/tcp/tcp_reassembler.h
+++ b/src/stream/tcp/tcp_reassembler.h
@@ -63,6 +63,7 @@ public:
 
     virtual int eval_flush_policy_on_ack(snort::Packet*) = 0;
     virtual int eval_flush_policy_on_data(snort::Packet*) = 0;
+    virtual int eval_asymmetric_flush(snort::Packet*) = 0;
     virtual int flush_stream(snort::Packet*, uint32_t dir, bool final_flush = false) = 0;
     void flush_queued_segments(snort::Flow* flow, bool clear, snort::Packet* = nullptr);
     void finish_and_final_flush(snort::Flow* flow, bool clear, snort::Packet*);
@@ -124,7 +125,7 @@ protected:
     void check_first_segment_hole();
     uint32_t perform_partial_flush(snort::Packet*, uint32_t flushed = 0);
     bool final_flush_on_fin(int32_t flush_amt, snort::Packet*, FinSeqNumStatus);
-    bool flush_on_asymmetric_flow(uint32_t flushed, snort::Packet *p);
+    bool asymmetric_flow_flushed(uint32_t flushed, snort::Packet *p);
 
     ProtocolAwareFlusher paf;
     TcpStreamTracker& tracker;
@@ -154,6 +155,9 @@ public:
     int eval_flush_policy_on_data(snort::Packet*) override
     { return 0; }
 
+    int eval_asymmetric_flush(snort::Packet*) override
+    { return 0; }
+
     int flush_stream(snort::Packet*, uint32_t, bool) override
     { return 0; }
 
diff --git a/src/stream/tcp/tcp_reassembler_ids.cc b/src/stream/tcp/tcp_reassembler_ids.cc
index 43d7c2329..124b855a0 100644
--- a/src/stream/tcp/tcp_reassembler_ids.cc
+++ b/src/stream/tcp/tcp_reassembler_ids.cc
@@ -231,25 +231,39 @@ int TcpReassemblerIds::eval_flush_policy_on_data(Packet* p)
     if ( tracker.is_retransmit_of_held_packet(p) )
         flushed = perform_partial_flush(p, flushed);
 
-    if ( flush_on_asymmetric_flow(flushed, p) )
+    if ( !p->flow->two_way_traffic() and
+        seglist.get_seg_bytes_total() > seglist.session->tcp_config->asymmetric_ids_flush_threshold )
     {
-        purge_to_seq(seglist.head->start_seq() + flushed);
-        tracker.r_win_base = seglist.seglist_base_seq;
-        tcpStats.flush_on_asymmetric_flow++;
+        seglist.skip_holes();
+        flushed += eval_asymmetric_flush(p);
     }
 
     return flushed;
 }
 
+int TcpReassemblerIds::eval_asymmetric_flush(snort::Packet* p)
+{
+    // asymmetric flush in IDS mode.. advance r_win_base to end of in-order data
+    tracker.r_win_base = tracker.rcv_nxt;
+
+    uint32_t flushed = eval_flush_policy_on_ack(p);
+    if ( flushed )
+        tcpStats.flush_on_asymmetric_flow++;
+
+    return flushed;
+}
+
 int TcpReassemblerIds::flush_stream(Packet* p, uint32_t dir, bool final_flush)
 {
-    if ( seglist.session->flow->two_way_traffic()
-        or (tracker.get_tcp_state() == TcpStreamTracker::TCP_MID_STREAM_RECV) )
-    {
-        uint32_t bytes = get_q_footprint();  // num bytes in post-ack mode
-        if ( bytes )
-            return flush_to_seq(bytes, p, dir);
-    }
+    uint32_t bytes = 0;
+
+    if ( seglist.session->flow->two_way_traffic() )
+         bytes = get_q_footprint();
+    else
+        bytes = get_q_sequenced();
+
+    if ( bytes )
+        return flush_to_seq(bytes, p, dir);
 
     if ( final_flush )
         return do_zero_byte_flush(p, dir);
diff --git a/src/stream/tcp/tcp_reassembler_ids.h b/src/stream/tcp/tcp_reassembler_ids.h
index 8d8b643ee..4ee533d55 100644
--- a/src/stream/tcp/tcp_reassembler_ids.h
+++ b/src/stream/tcp/tcp_reassembler_ids.h
@@ -47,6 +47,7 @@ public:
 
     int eval_flush_policy_on_ack(snort::Packet*) override;
     int eval_flush_policy_on_data(snort::Packet*) override;
+    int eval_asymmetric_flush(snort::Packet*) override;
     int flush_stream(snort::Packet*, uint32_t dir, bool final_flush = false) override;
 
     FlushPolicy get_flush_policy() const override
diff --git a/src/stream/tcp/tcp_reassembler_ips.cc b/src/stream/tcp/tcp_reassembler_ips.cc
index 3b474baa3..f0a513d76 100644
--- a/src/stream/tcp/tcp_reassembler_ips.cc
+++ b/src/stream/tcp/tcp_reassembler_ips.cc
@@ -153,7 +153,7 @@ int TcpReassemblerIps::eval_flush_policy_on_data(Packet* p)
     if ( tracker.is_retransmit_of_held_packet(p) )
         flushed = perform_partial_flush(p, flushed);
 
-    if ( flush_on_asymmetric_flow(flushed, p) )
+    if ( asymmetric_flow_flushed(flushed, p) )
     {
         purge_to_seq(seglist.head->start_seq() + flushed);
         tracker.r_win_base = seglist.seglist_base_seq;
@@ -163,6 +163,11 @@ int TcpReassemblerIps::eval_flush_policy_on_data(Packet* p)
     return flushed;
 }
 
+int TcpReassemblerIps::eval_asymmetric_flush(snort::Packet* p)
+{
+    return eval_flush_policy_on_data(p);
+}
+
 int TcpReassemblerIps::flush_stream(Packet* p, uint32_t dir, bool final_flush)
 {
     if ( seglist.session->flow->two_way_traffic()
diff --git a/src/stream/tcp/tcp_reassembler_ips.h b/src/stream/tcp/tcp_reassembler_ips.h
index bf21acf8e..3a81efd1e 100644
--- a/src/stream/tcp/tcp_reassembler_ips.h
+++ b/src/stream/tcp/tcp_reassembler_ips.h
@@ -46,6 +46,8 @@ public:
 
     int eval_flush_policy_on_ack(snort::Packet*) override;
     int eval_flush_policy_on_data(snort::Packet*) override;
+    int eval_asymmetric_flush(snort::Packet*) override;
+
     int flush_stream(snort::Packet*, uint32_t dir, bool final_flush = false) override;
 
     FlushPolicy get_flush_policy() const override
diff --git a/src/stream/tcp/tcp_reassembly_segments.cc b/src/stream/tcp/tcp_reassembly_segments.cc
index b39486991..0a3e0469f 100644
--- a/src/stream/tcp/tcp_reassembly_segments.cc
+++ b/src/stream/tcp/tcp_reassembly_segments.cc
@@ -362,7 +362,7 @@ void TcpReassemblySegments::purge_segments_left_of_hole(const TcpSegmentNode* en
     tracker->set_order(TcpStreamTracker::OUT_OF_SEQUENCE);
 
     if (PacketTracer::is_active())
-        PacketTracer::log("Stream: Skipped %u packets before seglist hole)\n", packets_skipped);
+        PacketTracer::log("Stream: Skipped %u packets before seglist hole\n", packets_skipped);
 }
 
 void TcpReassemblySegments::advance_rcv_nxt(TcpSegmentNode *tsn)
diff --git a/src/stream/tcp/tcp_session.cc b/src/stream/tcp/tcp_session.cc
index 2bfffd190..9ed5f7276 100644
--- a/src/stream/tcp/tcp_session.cc
+++ b/src/stream/tcp/tcp_session.cc
@@ -612,9 +612,9 @@ bool TcpSession::check_reassembly_queue_thresholds(TcpSegmentDescriptor& tsd, Tc
             tcpStats.exceeded_max_bytes++;
             bool ret_val = true;
 
-            // if inline and this is an asymmetric flow then skip over any seglist holes
+            // if this is an asymmetric flow then skip over any seglist holes
             // and flush to free up seglist space
-            if ( tsd.is_ips_policy_inline()  && !tsd.get_pkt()->flow->two_way_traffic() )
+            if ( !tsd.get_pkt()->flow->two_way_traffic() )
             {
                 space_left = listener->kickstart_asymmetric_flow(tsd, tcp_config->max_queued_bytes);
                 if ( space_left >= (int32_t)tsd.get_len() )
@@ -647,9 +647,9 @@ bool TcpSession::check_reassembly_queue_thresholds(TcpSegmentDescriptor& tsd, Tc
         {
             tcpStats.exceeded_max_segs++;
 
-            // if inline and this is an asymmetric flow then skip over any seglist holes
+            // if this is an asymmetric flow then skip over any seglist holes
             // and flush to free up seglist space
-            if ( tsd.is_ips_policy_inline() && !tsd.get_pkt()->flow->two_way_traffic() )
+            if ( !tsd.get_pkt()->flow->two_way_traffic() )
             {
                 listener->kickstart_asymmetric_flow(tsd, tcp_config->max_queued_bytes);
                 if ( listener->seglist.get_seg_count() + 1 <= tcp_config->max_queued_segs )
diff --git a/src/stream/tcp/tcp_state_syn_sent.cc b/src/stream/tcp/tcp_state_syn_sent.cc
index 4334cca5c..80b833c4e 100644
--- a/src/stream/tcp/tcp_state_syn_sent.cc
+++ b/src/stream/tcp/tcp_state_syn_sent.cc
@@ -151,6 +151,7 @@ bool TcpStateSynSent::fin_sent(TcpSegmentDescriptor& tsd, TcpStreamTracker& trk)
         }
         trk.set_tcp_state(TcpStreamTracker::TCP_FIN_WAIT1);
     }
+
     return true;
 }
 
diff --git a/src/stream/tcp/tcp_stream_config.cc b/src/stream/tcp/tcp_stream_config.cc
index 104e65273..4bf076547 100644
--- a/src/stream/tcp/tcp_stream_config.cc
+++ b/src/stream/tcp/tcp_stream_config.cc
@@ -51,7 +51,10 @@ void TcpStreamConfig::show() const
     str += std::to_string(max_queued_segs);
     str += " }";
     ConfigLogger::log_value("queue_limit", str.c_str());
-
+    str = "{ flush threshold = ";
+    str += std::to_string(asymmetric_ids_flush_threshold);
+    str += " }";
+    ConfigLogger::log_value("asymmetric_ids", str.c_str());
     ConfigLogger::log_flag("reassemble_async", !(flags & STREAM_CONFIG_NO_ASYNC_REASSEMBLY));
     ConfigLogger::log_limit("require_3whs", hs_timeout, -1, hs_timeout < 0 ? hs_timeout : -1);
     ConfigLogger::log_value("session_timeout", session_timeout);
diff --git a/src/stream/tcp/tcp_stream_config.h b/src/stream/tcp/tcp_stream_config.h
index 6e9459e0a..5aa25605c 100644
--- a/src/stream/tcp/tcp_stream_config.h
+++ b/src/stream/tcp/tcp_stream_config.h
@@ -63,6 +63,7 @@ public:
 
     uint32_t max_queued_bytes = 4194304;
     uint32_t max_queued_segs = 3072;
+    uint32_t asymmetric_ids_flush_threshold = 65535;
 
     uint32_t max_consec_small_segs = STREAM_DEFAULT_CONSEC_SMALL_SEGS;
     uint32_t max_consec_small_seg_size = STREAM_DEFAULT_MAX_SMALL_SEG_SIZE;
diff --git a/src/stream/tcp/tcp_stream_tracker.cc b/src/stream/tcp/tcp_stream_tracker.cc
index 279b9eaad..29d2584dc 100644
--- a/src/stream/tcp/tcp_stream_tracker.cc
+++ b/src/stream/tcp/tcp_stream_tracker.cc
@@ -126,9 +126,7 @@ int TcpStreamTracker::eval_flush_policy_on_ack(snort::Packet* p)
         oaitw_reassembler = nullptr;
     }
 
-    reassembler->eval_flush_policy_on_ack(p);
-
-    return 0;
+    return reassembler->eval_flush_policy_on_ack(p);
 }
 
 int TcpStreamTracker::eval_flush_policy_on_data(snort::Packet* p)
@@ -144,6 +142,19 @@ int TcpStreamTracker::eval_flush_policy_on_data(snort::Packet* p)
     return 0;
 }
 
+int TcpStreamTracker::eval_asymmetric_flush(snort::Packet* p)
+{
+    if( oaitw_reassembler )
+    {
+        delete oaitw_reassembler;
+        oaitw_reassembler = nullptr;
+    }
+
+    reassembler->eval_asymmetric_flush(p);
+
+    return 0;
+}
+
 TcpStreamTracker::TcpEvent TcpStreamTracker::set_tcp_event(const TcpSegmentDescriptor& tsd)
 {
     bool talker;
@@ -481,6 +492,10 @@ void TcpStreamTracker::fallback()
     //assert(server_side == to_server && server_side == !tracker.client_tracker);
 #endif
 
+    if (PacketTracer::is_active())
+        PacketTracer::log("Stream: %s tracker fallback to the Atom splitter.\n",
+            client_tracker ? "client" : "server");
+
     set_splitter(new AtomSplitter(!client_tracker));
     tcpStats.partial_fallbacks++;
 
@@ -872,7 +887,7 @@ int32_t TcpStreamTracker::kickstart_asymmetric_flow(const TcpSegmentDescriptor&
     else
         reassembler->reset_paf();
 
-    eval_flush_policy_on_data(tsd.get_pkt());
+    eval_asymmetric_flush(tsd.get_pkt());
 
     int32_t space_left = max_queued_bytes - seglist.get_seg_bytes_total();
 
diff --git a/src/stream/tcp/tcp_stream_tracker.h b/src/stream/tcp/tcp_stream_tracker.h
index b8850bfc5..a2096bb6f 100644
--- a/src/stream/tcp/tcp_stream_tracker.h
+++ b/src/stream/tcp/tcp_stream_tracker.h
@@ -97,6 +97,7 @@ public:
 
     int eval_flush_policy_on_ack(snort::Packet*);
     int eval_flush_policy_on_data(snort::Packet*);
+    int eval_asymmetric_flush(snort::Packet*);
     void update_stream_order(const TcpSegmentDescriptor&, bool aligned);
 
     void fallback();