From: ARUNKUMAR KAYAMBU -X (akayambu - XORIANT CORPORATION at Cisco) <akayambu@cisco.com>
Date: Mon, 18 Mar 2024 15:07:10 +0000 (+0000)
Subject: Pull request #4238: Show conn details
X-Git-Tag: 3.1.83.0~6
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=4074913099927b2940c73aa24ac136fa4dec6b9c;p=thirdparty%2Fsnort3.git

Pull request #4238: Show conn details

Merge in SNORT/snort3 from ~AKAYAMBU/snort3:show_conn_details to master

Squashed commit of the following:

commit 5a7c785c52599c257ff8e2da88d4ec7e63858351
Author: RAGHURAAM CONJEEVARAM UDAYANAN -X (rconjeev - XORIANT CORPORATION at Cisco) <rconjeev@cisco.com>
Date:   Wed Nov 8 17:39:11 2023 +0530

    flow: add filter to dump flows
---

diff --git a/src/flow/CMakeLists.txt b/src/flow/CMakeLists.txt
index 1b3a448fc..d8f5ec9ca 100644
--- a/src/flow/CMakeLists.txt
+++ b/src/flow/CMakeLists.txt
@@ -30,6 +30,7 @@ add_library (flow OBJECT
     ha_module.h
     prune_stats.h
     stash_item.h
+    filter_flow_critera.h
 )
 
 install(FILES ${FLOW_INCLUDES}
diff --git a/src/flow/filter_flow_critera.h b/src/flow/filter_flow_critera.h
new file mode 100644
index 000000000..7995f1a38
--- /dev/null
+++ b/src/flow/filter_flow_critera.h
@@ -0,0 +1,36 @@
+//--------------------------------------------------------------------------
+// Copyright (C) 2014-2024 Cisco and/or its affiliates. All rights reserved.
+//
+// This program is free software; you can redistribute it and/or modify it
+// under the terms of the GNU General Public License Version 2 as published
+// by the Free Software Foundation.  You may not use, modify or distribute
+// this program under any other version of the GNU General Public License.
+//
+// This program is distributed in the hope that it will be useful, but
+// WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License along
+// with this program; if not, write to the Free Software Foundation, Inc.,
+// 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+//--------------------------------------------------------------------------
+
+#ifndef FILTER_FLOW_CRITERIA_H
+#define FILTER_FLOW_CRITERIA_H
+
+#include <string>
+#include "sfip/sf_ip.h"
+#include <framework/decode_data.h>
+
+struct FilterFlowCriteria
+{
+	PktType pkt_type;
+    snort::SfIp source_sfip;
+    snort::SfIp destination_sfip;
+	uint16_t source_port = 0;
+	uint16_t destination_port = 0;
+    snort::SfIp source_subnet_sfip;
+    snort::SfIp destination_subnet_sfip;
+};
+#endif
diff --git a/src/flow/flow.h b/src/flow/flow.h
index 3fd020c34..119000b77 100644
--- a/src/flow/flow.h
+++ b/src/flow/flow.h
@@ -486,11 +486,13 @@ public:  // FIXIT-M privatize if possible
     uint8_t outer_server_ttl = 0;
 
     uint8_t response_count = 0;
+    uint8_t dump_code = 0;
 
     struct
     {
         bool client_initiated : 1;  // Set if the first packet on the flow was from the side that is
                                     // currently considered to be the client
+        bool key_is_reversed : 1; // The _l members are the destinations
         bool app_direction_swapped : 1; // Packet direction swapped from application perspective
         bool disable_inspect : 1;
         bool trigger_detained_packet_event : 1;
diff --git a/src/flow/flow_cache.cc b/src/flow/flow_cache.cc
index 8623bec52..bee007c62 100644
--- a/src/flow/flow_cache.cc
+++ b/src/flow/flow_cache.cc
@@ -24,14 +24,23 @@
 
 #include "flow/flow_cache.h"
 
+#include <sstream>
+
+#include "control/control.h"
 #include "detection/detection_engine.h"
 #include "hash/hash_defs.h"
 #include "hash/zhash.h"
 #include "helpers/flag_context.h"
+#include "log/messages.h"
+#ifdef REG_TEST
+#include "main/analyzer.h"
+#endif
 #include "main/thread_config.h"
 #include "packet_io/active.h"
 #include "packet_tracer/packet_tracer.h"
 #include "stream/base/stream_module.h"
+#include "stream/tcp/tcp_stream_session.h"
+#include "stream/tcp/tcp_trace.h"
 #include "time/packet_time.h"
 #include "trace/trace_api.h"
 #include "utils/stats.h"
@@ -54,12 +63,142 @@ static const unsigned WDT_MASK = 7; // kick watchdog once for every 8 flows dele
 constexpr uint8_t MAX_PROTOCOLS = (uint8_t)to_utype(PktType::MAX) - 1; //removing PktType::NONE from count
 constexpr uint64_t max_skip_protos = (1ULL << MAX_PROTOCOLS) - 1;
 
+uint8_t DumpFlows::dump_code = 0;
+
+#ifndef REG_TEST
+DumpFlows::DumpFlows(unsigned count,  ControlConn* conn)
+#else
+DumpFlows::DumpFlows(unsigned count, ControlConn* conn, int resume)
+#endif
+    : snort::AnalyzerCommand(conn), dump_count(count)
+#ifdef REG_TEST
+    , resume(resume)
+#endif
+{
+    next.resize(ThreadConfig::get_instance_max());
+    ++dump_code;
+}
+
+bool DumpFlows::open_files(const std::string& base_name)
+{
+    dump_stream.resize(ThreadConfig::get_instance_max());
+    for (unsigned i = 0; i < ThreadConfig::get_instance_max(); ++i)
+    {
+        std::string file_name = base_name + std::to_string(i + 1);
+        dump_stream[i].open(file_name, std::fstream::out | std::fstream::trunc);
+        if (0 != (dump_stream[i].rdstate() & std::fstream::failbit))
+        {
+            LogRespond(ctrlcon, "Dump flows failed to open %s\n", file_name.c_str());
+            return false;
+        }
+    }
+    return true;
+}
+
+void DumpFlows::cidr2mask(const uint32_t cidr, uint32_t* mask) const
+{
+    size_t i;
+
+    for (i = cidr; i > 0;)
+    {
+        --i;
+        mask[i / 32] |= 0x00000001 << (i % 32);
+    }
+}
+
+bool DumpFlows::set_ip(std::string filter_ip, snort::SfIp& ip, snort::SfIp& subnet) const
+{
+    size_t slash_pos = filter_ip.find('/');
+    if (slash_pos != std::string::npos)
+    {
+        std::string ip_part = filter_ip.substr(0, slash_pos);
+        std::string subnet_part = filter_ip.substr(slash_pos+1);
+
+        if (ip_part.find(':') != std::string::npos)
+        {
+            //filter is IPV6
+            if (ip.pton(AF_INET6, ip_part.c_str()) != SFIP_SUCCESS)
+                return false;
+
+            if (subnet_part.find(':') == std::string::npos)
+            {
+                //IPV6 cidr
+                uint32_t cidr = std::stoi(subnet_part);
+                if (cidr > 128)
+                    return false;
+                uint32_t mask_v6[4]={0};
+                cidr2mask(std::stoi(subnet_part), mask_v6);
+                if (subnet.set(&mask_v6, AF_INET6) != SFIP_SUCCESS)
+                    return false;
+            }
+            else if (subnet_part.empty() || (subnet.pton(AF_INET6, subnet_part.c_str()) != SFIP_SUCCESS))
+                return false;
+            return true;
+        }
+        else if (ip_part.find('.') != std::string::npos)
+        {
+            //filter is  IPV4
+            if (ip.pton(AF_INET, ip_part.c_str()) != SFIP_SUCCESS)
+                return false;
+
+            if (subnet_part.find('.') == std::string::npos)
+            {
+                //IPV4 cidr
+                uint32_t cidr = std::stoi(subnet_part);
+                if (cidr > 32)
+                    return false;
+                uint32_t mask_v4[1]={0};
+                cidr2mask(std::stoi(subnet_part), mask_v4);
+                if (subnet.set(&mask_v4, AF_INET) != SFIP_SUCCESS)
+                    return false;
+            }
+            else if (subnet_part.empty())
+                return false;
+            else
+            {
+                //IPV4 netmask
+                if (subnet.pton(AF_INET, subnet_part.c_str()) != SFIP_SUCCESS)
+                    return false;
+            }
+            return true;
+        }
+        return false;
+    }
+    else
+    {
+        //No mask
+        if (filter_ip.find(':') != std::string::npos)
+        {
+            return ip.pton(AF_INET6, filter_ip.c_str()) == SFIP_SUCCESS;
+        }
+        else if (filter_ip.find('.') != std::string::npos)
+        {
+            return ip.pton(AF_INET, filter_ip.c_str()) == SFIP_SUCCESS;
+        }
+        else if (filter_ip.empty())
+            return true;
+    }
+    return false;
+}
+
+bool DumpFlows::execute(Analyzer&, void**)
+{
+    if (!flow_con)
+        return true;
+    unsigned id = get_instance_id();
+#ifdef REG_TEST
+    if (!next[id] && -1 != resume)
+        Analyzer::get_local_analyzer()->resume(resume);
+#endif
+    bool first = !next[id];
+    next[id] = 1;
+    return flow_con->dump_flows(dump_stream[id], dump_count, ffc, first, dump_code);
+}
+
 //-------------------------------------------------------------------------
 // FlowCache stuff
 //-------------------------------------------------------------------------
 
-extern THREAD_LOCAL const snort::Trace* stream_trace;
-
 FlowCache::FlowCache(const FlowCacheConfig& cfg) : config(cfg)
 {
     hash_table = new ZHash(config.max_flows, sizeof(FlowKey), MAX_PROTOCOLS, false);
@@ -604,6 +743,189 @@ unsigned FlowCache::purge()
     return retired;
 }
 
+std::string FlowCache::timeout_to_str(time_t t)
+{
+    std::stringstream out;
+    time_t hours = t / (60 * 60);
+    if (hours)
+    {
+        out << hours << "h";
+        t -= hours * (60 * 60);
+    }
+    time_t minutes = t / 60;
+    if (minutes || hours)
+    {
+        out << minutes << "m";
+        t -= minutes * 60;
+    }
+    if (t || !hours)
+        out << t << "s";
+    return out.str();
+}
+
+
+bool FlowCache::is_ip_match(const SfIp& flow_sfip, const SfIp& filter_sfip, const SfIp& filter_subnet_sfip) const
+{
+    //if address is empty
+    if (!filter_sfip.is_set())
+        return true;
+
+    //if no subnet mask
+    if (!filter_subnet_sfip.is_set())
+    {
+        return filter_sfip.fast_equals_raw(flow_sfip);
+    }
+    else
+    {
+        if (filter_sfip.get_family() != flow_sfip.get_family())
+            return false;
+        const uint64_t* filter_ptr = filter_sfip.get_ip64_ptr();
+        const uint64_t* flow_ptr = flow_sfip.get_ip64_ptr();
+        const uint64_t* subnet_sfip = filter_subnet_sfip.get_ip64_ptr();
+        return (filter_ptr[0] & subnet_sfip[0]) == (flow_ptr[0] & subnet_sfip[0]) && (filter_ptr[1] & subnet_sfip[1]) == (flow_ptr[1] & subnet_sfip[1]);
+    }
+}
+
+bool FlowCache::filter_flows(const Flow& flow, const FilterFlowCriteria& ffc) const
+{
+    const SfIp& flow_srcip = flow.flags.client_initiated ? flow.client_ip : flow.server_ip;
+    const SfIp& flow_dstip = flow.flags.client_initiated ? flow.server_ip : flow.client_ip;
+
+    return ((ffc.pkt_type == PktType::NONE || flow.pkt_type == ffc.pkt_type)
+            && is_ip_match(flow_srcip, ffc.source_sfip, ffc.source_subnet_sfip)
+            && is_ip_match(flow_dstip, ffc.destination_sfip, ffc.destination_subnet_sfip)
+            && (!ffc.source_port || ffc.source_port == (flow.flags.key_is_reversed ? flow.key->port_h : flow.key->port_l))
+            && (!ffc.destination_port || ffc.destination_port == (flow.flags.key_is_reversed ? flow.key->port_l : flow.key->port_h)));
+
+}
+
+void FlowCache::output_flow(std::fstream& stream, const Flow& flow, const struct timeval& now) const
+{
+    char src_ip[INET6_ADDRSTRLEN];
+    src_ip[0] = 0;
+    char dst_ip[INET6_ADDRSTRLEN];
+    dst_ip[0] = 0;
+    uint16_t src_port;
+    uint16_t dst_port;
+    if (flow.flags.key_is_reversed)
+    {
+        SfIp ip;
+        ip.set(flow.key->ip_h);
+        ip.ntop(src_ip, sizeof(src_ip));
+        ip.set(flow.key->ip_l);
+        ip.ntop(dst_ip, sizeof(dst_ip));
+        src_port = flow.key->port_h;
+        dst_port = flow.key->port_l;
+    }
+    else
+    {
+        SfIp ip;
+        ip.set(flow.key->ip_l);
+        ip.ntop(src_ip, sizeof(src_ip));
+        ip.set(flow.key->ip_h);
+        ip.ntop(dst_ip, sizeof(dst_ip));
+        src_port = flow.key->port_l;
+        dst_port = flow.key->port_h;
+    }
+    std::stringstream out;
+    std::stringstream proto;
+    switch ( flow.pkt_type )
+    {
+        case PktType::IP:
+            out << "IP " << flow.key->addressSpaceId << ": " << src_ip << " " << dst_ip;
+            break;
+
+        case PktType::ICMP:
+            out << "ICMP " << flow.key->addressSpaceId << ": " << src_ip << " type " << src_port << " "
+                << dst_ip;
+            break;
+
+        case PktType::TCP:
+            out << "TCP " << flow.key->addressSpaceId << ": " << src_ip << "/" << src_port << " "
+                << dst_ip << "/" << dst_port;
+            if (flow.session)
+            {
+                TcpStreamSession* tcp_session = static_cast<TcpStreamSession*>(flow.session);
+                proto << " state client " << stream_tcp_state_to_str(tcp_session->client)
+                    << " server " << stream_tcp_state_to_str(tcp_session->server);
+            }
+            break;
+
+        case PktType::UDP:
+            out << "UDP " << flow.key->addressSpaceId << ": "<< src_ip << "/" << src_port << " "
+                << dst_ip << "/" << dst_port;
+            break;
+
+        default:
+            assert(false);
+    }
+    out << " pkts/bytes client " << flow.flowstats.client_pkts << "/" << flow.flowstats.client_bytes
+        << " server " << flow.flowstats.server_pkts << "/" << flow.flowstats.server_bytes
+        << " idle " << (now.tv_sec - flow.last_data_seen) << "s, uptime "
+        << (now.tv_sec - flow.flowstats.start_time.tv_sec) << "s, timeout ";
+    std::string t = flow.is_hard_expiration() ?
+        timeout_to_str(flow.expire_time - now.tv_sec) :
+        timeout_to_str((flow.last_data_seen + config.proto[to_utype(flow.key->pkt_type)].nominal_timeout) - now.tv_sec);
+    out << t;
+    stream << out.str() << proto.str() << std::endl;
+}
+
+bool FlowCache::dump_flows(std::fstream& stream, unsigned count, const FilterFlowCriteria& ffc, bool first, uint8_t code) const
+{
+    struct timeval now;
+    packet_gettimeofday(&now);
+    unsigned i;
+
+    for(uint8_t proto_id = to_utype(PktType::NONE)+1; proto_id <= to_utype(PktType::ICMP); proto_id++)
+    {
+        if (first)
+        {
+            Flow* walk_flow = static_cast<Flow*>(hash_table->get_walk_user_data(proto_id));
+            if (!walk_flow)
+            {
+                //Return only if all the protocol caches are processed.
+                if (proto_id < to_utype(PktType::ICMP))
+                    continue;
+
+                return true;
+            }
+            walk_flow->dump_code = code;
+            bool matched_filter = filter_flows(*walk_flow, ffc);
+            if (matched_filter)
+                output_flow(stream, *walk_flow, now);
+            i = 1;
+
+        }
+        else
+            i = 0;
+        while (i < count)
+        {
+            Flow* walk_flow = static_cast<Flow*>(hash_table->get_next_walk_user_data(proto_id));
+
+            if (!walk_flow )
+            {
+                //Return only if all the protocol caches are processed.
+                if (proto_id < to_utype(PktType::ICMP))
+                    break;
+                return true;
+            }
+            if (walk_flow->dump_code != code)
+            {
+                walk_flow->dump_code = code;
+                bool matched_filter = filter_flows(*walk_flow, ffc);
+                if (matched_filter)
+                    output_flow(stream, *walk_flow, now);
+                ++i;
+            }
+#ifdef REG_TEST
+            else
+                LogMessage("dump_flows skipping already dumped flow\n");
+#endif
+        }
+    }
+    return false;
+}
+
 size_t FlowCache::uni_flows_size() const
 {
     return uni_flows ? uni_flows->get_count() : 0;
diff --git a/src/flow/flow_cache.h b/src/flow/flow_cache.h
index 98f475bb5..20deac882 100644
--- a/src/flow/flow_cache.h
+++ b/src/flow/flow_cache.h
@@ -26,13 +26,19 @@
 // Flows are stored in a ZHash instance by FlowKey.
 
 #include <ctime>
+#include <fstream>
+#include <mutex>
 #include <type_traits>
+#include <vector>
+#include <memory>
 
 #include "framework/counts.h"
+#include "main/analyzer_command.h"
 #include "main/thread.h"
 
 #include "flow_config.h"
 #include "prune_stats.h"
+#include "filter_flow_critera.h"
 
 namespace snort
 {
@@ -40,6 +46,37 @@ class Flow;
 struct FlowKey;
 }
 
+class DumpFlows : public snort::AnalyzerCommand
+{
+public:
+#ifndef REG_TEST
+    DumpFlows(unsigned count, ControlConn*);
+#else
+    DumpFlows(unsigned count, ControlConn*, int resume);
+#endif
+    ~DumpFlows() override = default;
+    bool open_files(const std::string& base_name);
+    void cidr2mask(const uint32_t cidr, uint32_t* mask) const;
+    bool set_ip(std::string filter_ip, snort::SfIp& ip, snort::SfIp& subnet) const;
+    bool execute(Analyzer&, void**) override;
+    const char* stringify() override
+    { return "DumpFlows"; }
+    void set_filter_criteria(const FilterFlowCriteria& filter_criteria)
+    {ffc = filter_criteria;}
+
+private:
+    //dump_code is to track if the flow is dumped only once per dump_flow command.
+    static uint8_t dump_code;
+    std::vector<std::fstream> dump_stream;
+    std::vector<unsigned> next;
+    unsigned dump_count;
+    FilterFlowCriteria ffc;
+#ifdef REG_TEST
+    int resume = -1;
+#endif
+};
+
+
 class FlowUniList;
 
 class FlowCache
@@ -62,6 +99,7 @@ public:
     unsigned timeout(unsigned num_flows, time_t cur_time);
     unsigned delete_flows(unsigned num_to_delete);
     unsigned prune_multiple(PruneReason, bool do_cleanup);
+    bool dump_flows(std::fstream&, unsigned count, const FilterFlowCriteria& ffc, bool first, uint8_t code) const;
 
     unsigned purge();
     unsigned get_count();
@@ -111,8 +149,11 @@ private:
     void remove(snort::Flow*);
     void retire(snort::Flow*);
     unsigned prune_unis(PktType);
-    unsigned delete_active_flows
-        (unsigned mode, unsigned num_to_delete, unsigned &deleted);
+    unsigned delete_active_flows(unsigned mode, unsigned num_to_delete, unsigned &deleted);
+    static std::string timeout_to_str(time_t);
+    bool is_ip_match(const snort::SfIp& flow_ip, const snort::SfIp& filter_ip, const snort::SfIp& subnet) const;
+    bool filter_flows(const snort::Flow&, const FilterFlowCriteria&) const;
+    void output_flow(std::fstream&, const snort::Flow&, const struct timeval&) const;
 
 private:
     static const unsigned cleanup_flows = 1;
diff --git a/src/flow/flow_control.cc b/src/flow/flow_control.cc
index 49b15b06e..3acc2202a 100644
--- a/src/flow/flow_control.cc
+++ b/src/flow/flow_control.cc
@@ -133,6 +133,9 @@ bool FlowControl::prune_one(PruneReason reason, bool do_cleanup)
 unsigned FlowControl::prune_multiple(PruneReason reason, bool do_cleanup)
 { return cache->prune_multiple(reason, do_cleanup); }
 
+bool FlowControl::dump_flows(std::fstream& stream, unsigned count, const FilterFlowCriteria& ffc, bool first, uint8_t code) const
+{ return cache->dump_flows(stream, count, ffc, first, code); }
+
 void FlowControl::timeout_flows(unsigned max, time_t cur_time)
 {
     cache->timeout(max, cur_time);
@@ -162,13 +165,14 @@ Flow* FlowControl::stale_flow_cleanup(FlowCache* cache, Flow* flow, Packet* p)
 // packet foo
 //-------------------------------------------------------------------------
 
-void FlowControl::set_key(FlowKey* key, Packet* p)
+bool FlowControl::set_key(FlowKey* key, Packet* p)
 {
     const ip::IpApi& ip_api = p->ptrs.ip_api;
     uint32_t mplsId;
     uint16_t vlanId;
     PktType type = p->type();
     IpProtocol ip_proto = p->get_ip_proto_next();
+    bool reversed;
 
     if ( p->proto_bits & PROTO_BIT__VLAN )
         vlanId = layer::get_vlan_layer(p)->vid();
@@ -182,19 +186,20 @@ void FlowControl::set_key(FlowKey* key, Packet* p)
 
     if ( (p->ptrs.decode_flags & DECODE_FRAG) )
     {
-        key->init(p->context->conf, type, ip_proto, ip_api.get_src(),
+        reversed = key->init(p->context->conf, type, ip_proto, ip_api.get_src(),
             ip_api.get_dst(), ip_api.id(), vlanId, mplsId, *p->pkth);
     }
     else if ( type == PktType::ICMP )
     {
-        key->init(p->context->conf, type, ip_proto, ip_api.get_src(), p->ptrs.icmph->type,
+        reversed = key->init(p->context->conf, type, ip_proto, ip_api.get_src(), p->ptrs.icmph->type,
             ip_api.get_dst(), 0, vlanId, mplsId, *p->pkth);
     }
     else
     {
-        key->init(p->context->conf, type, ip_proto, ip_api.get_src(), p->ptrs.sp,
+        reversed = key->init(p->context->conf, type, ip_proto, ip_api.get_src(), p->ptrs.sp,
             ip_api.get_dst(), p->ptrs.dp, vlanId, mplsId, *p->pkth);
     }
+    return reversed;
 }
 
 static bool is_bidirectional(const Flow* flow)
@@ -390,7 +395,7 @@ bool FlowControl::process(PktType type, Packet* p, bool* new_flow)
         return false;
 
     FlowKey key;
-    set_key(&key, p);
+    bool reversed = set_key(&key, p);
     Flow* flow = cache->find(&key);
 
     if (flow)
@@ -413,6 +418,11 @@ bool FlowControl::process(PktType type, Packet* p, bool* new_flow)
             if ( !flow )
                 return true;
 
+            if ( p->is_tcp() and p->ptrs.tcph->is_syn_ack() )
+                flow->flags.key_is_reversed = !reversed;
+            else
+                flow->flags.key_is_reversed = reversed;
+
             if ( new_flow )
                 *new_flow = true;
         }
diff --git a/src/flow/flow_control.h b/src/flow/flow_control.h
index 637978d69..b84c1abca 100644
--- a/src/flow/flow_control.h
+++ b/src/flow/flow_control.h
@@ -26,12 +26,14 @@
 // processed.  flows are pruned as needed to process new flows.
 
 #include <cstdint>
+#include <fstream>
 #include <vector>
 
 #include "flow/flow_config.h"
 #include "framework/counts.h"
 #include "framework/decode_data.h"
 #include "framework/inspector.h"
+#include "flow/flow_cache.h"
 
 namespace snort
 {
@@ -71,6 +73,8 @@ public:
     void check_expected_flow(snort::Flow*, snort::Packet*);
     unsigned prune_multiple(PruneReason, bool do_cleanup);
 
+    bool dump_flows(std::fstream&, unsigned count, const FilterFlowCriteria& ffc, bool first, uint8_t code) const;
+
     int add_expected_ignore(
         const snort::Packet* ctrlPkt, PktType, IpProtocol,
         const snort::SfIp *srcIP, uint16_t srcPort,
@@ -100,7 +104,7 @@ public:
     PegCount get_num_flows() const;
 
 private:
-    void set_key(snort::FlowKey*, snort::Packet*);
+    bool set_key(snort::FlowKey*, snort::Packet*);
     unsigned process(snort::Flow*, snort::Packet*, bool new_ha_flow);
     void update_stats(snort::Flow*, snort::Packet*);
 
diff --git a/src/flow/flow_key.cc b/src/flow/flow_key.cc
index 2f55806a4..732fe2e9b 100644
--- a/src/flow/flow_key.cc
+++ b/src/flow/flow_key.cc
@@ -355,7 +355,7 @@ bool FlowKey::init(
     flags.group_used = (ingress_group != DAQ_PKTHDR_UNKNOWN and egress_group != DAQ_PKTHDR_UNKNOWN);
     init_groups(ingress_group, egress_group, reversed);
 
-    return false;
+    return reversed;
 }
 
 bool FlowKey::init(
@@ -397,7 +397,7 @@ bool FlowKey::init(
     flags.group_used = ((pkt_hdr.flags & DAQ_PKT_FLAG_SIGNIFICANT_GROUPS) != 0);
     init_groups(pkt_hdr.ingress_group, pkt_hdr.egress_group, reversed);
 
-    return false;
+    return reversed;
 }
 
 //-------------------------------------------------------------------------
diff --git a/src/flow/test/flow_cache_test.cc b/src/flow/test/flow_cache_test.cc
index 85cdf800d..2dd950da4 100644
--- a/src/flow/test/flow_cache_test.cc
+++ b/src/flow/test/flow_cache_test.cc
@@ -26,6 +26,7 @@
 
 #include "flow/flow_control.h"
 
+#include "control/control.h"
 #include "detection/detection_engine.h"
 #include "flow/expect_cache.h"
 #include "flow/flow_cache.h"
@@ -53,8 +54,11 @@ THREAD_LOCAL bool Active::s_suspend = false;
 THREAD_LOCAL Active::ActiveSuspendReason Active::s_suspend_reason = Active::ASP_NONE;
 
 THREAD_LOCAL const Trace* stream_trace = nullptr;
+THREAD_LOCAL FlowControl* flow_con = nullptr;
 
 void Active::drop_packet(snort::Packet const*, bool) { }
+Analyzer* Analyzer::get_local_analyzer() { return nullptr; }
+void Analyzer::resume(unsigned long) { }
 void Active::set_drop_reason(char const*) { }
 ExpectCache::ExpectCache(uint32_t) { }
 ExpectCache::~ExpectCache() = default;
@@ -67,7 +71,16 @@ const SnortConfig* SnortConfig::get_conf() { return nullptr; }
 uint8_t TraceApi::get_constraints_generation() { return 0; }
 void TraceApi::filter(const Packet&) {}
 void ThreadConfig::preemptive_kick() {}
-
+SfIpRet SfIp::set(void const*) { return SFIP_SUCCESS; }
+SfIpRet SfIp::pton(const int, const char* ) { return SFIP_SUCCESS; }
+const char* SfIp::ntop(char* buf, int) const
+{ buf[0] = 0; return buf; }
+unsigned snort::get_instance_id() { return 0; }
+unsigned ThreadConfig::get_instance_max() { return 0; }
+bool ControlConn::respond(const char*, ...) { return true; }
+class TcpStreamTracker;
+const char* stream_tcp_state_to_str(const TcpStreamTracker&) { return "error"; }
+void LogMessage(const char*, ...) { }
 namespace snort
 {
 Flow::~Flow() = default;
@@ -78,6 +91,7 @@ void Flow::free_flow_data() { }
 void Flow::set_client_initiate(Packet*) { }
 void Flow::set_direction(Packet*) { }
 void Flow::set_mpls_layer_per_dir(Packet*) { }
+void packet_gettimeofday(struct timeval* tv) { tv = {}; }
 
 time_t packet_time() { return 0; }
 
diff --git a/src/flow/test/flow_control_test.cc b/src/flow/test/flow_control_test.cc
index b52c2f8b7..abcc797b3 100644
--- a/src/flow/test/flow_control_test.cc
+++ b/src/flow/test/flow_control_test.cc
@@ -77,6 +77,7 @@ size_t FlowCache::flows_size() const { return 0; }
 void Flow::init(PktType) { }
 const SnortConfig* SnortConfig::get_conf() { return nullptr; }
 void FlowCache::unlink_uni(Flow*) { }
+bool FlowCache::dump_flows(std::fstream&, unsigned, const FilterFlowCriteria&, bool, uint8_t) const { return false; }
 void Flow::set_client_initiate(Packet*) { }
 void Flow::set_direction(Packet*) { }
 void Flow::set_mpls_layer_per_dir(Packet*) { }
diff --git a/src/hash/hash_lru_cache.cc b/src/hash/hash_lru_cache.cc
index 3ebb5650e..43764bbc9 100644
--- a/src/hash/hash_lru_cache.cc
+++ b/src/hash/hash_lru_cache.cc
@@ -37,20 +37,13 @@ HashLruCache::HashLruCache()
 
 void HashLruCache::insert(HashNode* hnode)
 {
-    if ( head )
-    {
-        hnode->gprev = nullptr;
-        hnode->gnext = head;
+    hnode->gprev = nullptr;
+    hnode->gnext = head;
+    if (head)
         head->gprev = hnode;
-        head = hnode;
-    }
     else
-    {
-        hnode->gprev = nullptr;
-        hnode->gnext = nullptr;
-        head = hnode;
         tail = hnode;
-    }
+    head = hnode;
 }
 
 void HashLruCache::touch(HashNode* hnode)
@@ -58,6 +51,9 @@ void HashLruCache::touch(HashNode* hnode)
     if ( hnode == cursor )
         cursor = hnode->gprev;
 
+    if ( walk_cursor == hnode )
+        walk_cursor = hnode->gprev;
+
     if ( hnode != head )
     {
         remove_node(hnode);
@@ -70,6 +66,9 @@ void HashLruCache::remove_node(HashNode* hnode)
     if ( cursor == hnode )
         cursor = hnode->gprev;
 
+    if ( walk_cursor == hnode )
+        walk_cursor = hnode->gprev;
+
     if ( head == hnode )
     {
         head = head->gnext;
diff --git a/src/hash/hash_lru_cache.h b/src/hash/hash_lru_cache.h
index 9383692e4..c5ed0e88b 100644
--- a/src/hash/hash_lru_cache.h
+++ b/src/hash/hash_lru_cache.h
@@ -63,10 +63,27 @@ public:
         return hnode;
     }
 
+    snort::HashNode* get_walk_node()
+    {
+        if ( tail )
+            walk_cursor = tail->gprev;
+        return tail;
+    }
+
+    snort::HashNode* get_next_walk_node()
+    {
+        snort::HashNode* rnode = walk_cursor;
+        if ( walk_cursor )
+            walk_cursor = walk_cursor->gprev;
+        return rnode;
+    }
+
 private:
     snort::HashNode* head = nullptr;
     snort::HashNode* tail = nullptr;
     snort::HashNode* cursor = nullptr;
+    //walk_cursor is used to traverse from tail to head while dumping the flows.
+    snort::HashNode* walk_cursor = nullptr;
 };
 
 #endif
diff --git a/src/hash/xhash.cc b/src/hash/xhash.cc
index a1597809d..b4b206f4b 100644
--- a/src/hash/xhash.cc
+++ b/src/hash/xhash.cc
@@ -511,6 +511,18 @@ void* XHash::get_lru_user_data(uint8_t type)
     return lru_caches[type]->get_lru_user_data();
 }
 
+void* XHash::get_walk_user_data(uint8_t type)
+{
+    HashNode* walk_node = lru_caches[type]->get_walk_node();
+    return walk_node ? walk_node->data : nullptr;
+}
+
+void* XHash::get_next_walk_user_data(uint8_t type)
+{
+    HashNode* walk_node = lru_caches[type]->get_next_walk_node();
+    return walk_node ? walk_node->data : nullptr;
+}
+
 HashNode* XHash::release_lru_node(uint8_t type)
 {
     assert(type < num_lru_caches);
diff --git a/src/hash/xhash.h b/src/hash/xhash.h
index 5829499d5..6bcd1fb75 100644
--- a/src/hash/xhash.h
+++ b/src/hash/xhash.h
@@ -60,10 +60,12 @@ public:
     void* get_user_data();
     void* get_user_data(const void* key, uint8_t type = 0);
     void release(uint8_t type = 0);
-    int release_node(const void* key, uint8_t type = 0);
+    int release_node(const void* key, u_int8_t type = 0);
     int release_node(HashNode* node, uint8_t type = 0);
     void* get_mru_user_data(uint8_t type = 0);
     void* get_lru_user_data(uint8_t type = 0);
+    void* get_walk_user_data(uint8_t type = 0);
+    void* get_next_walk_user_data(uint8_t type = 0);
     bool delete_lru_node(uint8_t type = 0);
     void clear_hash();
     bool full() const { return !fhead; }
diff --git a/src/sfip/sf_ip.h b/src/sfip/sf_ip.h
index 0fd8ed7e1..bb31fb219 100644
--- a/src/sfip/sf_ip.h
+++ b/src/sfip/sf_ip.h
@@ -70,6 +70,7 @@ struct SO_PUBLIC SfIp
     uint32_t get_ip4_value() const;
     const uint32_t* get_ip4_ptr() const;
     const uint32_t* get_ip6_ptr() const;
+    const uint64_t* get_ip64_ptr() const;
     const uint32_t* get_ptr() const;
     bool is_set() const;
     bool is_ip6() const;
@@ -121,6 +122,7 @@ private:
         uint8_t ip8[16];
         uint16_t ip16[8];
         uint32_t ip32[4];
+        uint64_t ip64[2];
     };
     int16_t family;
 } __attribute__((__packed__));
@@ -159,6 +161,11 @@ inline const uint32_t* SfIp::get_ip6_ptr() const
     return ip32;
 }
 
+inline const uint64_t* SfIp::get_ip64_ptr() const
+{
+    return ip64;
+}
+
 inline const uint32_t* SfIp::get_ptr() const
 {
     if (is_ip4())
@@ -416,13 +423,9 @@ inline bool SfIp::fast_gt6(const SfIp& ip2) const
 
 inline bool SfIp::fast_eq6(const SfIp& ip2) const
 {
-    if (ip32[0] != ip2.ip32[0])
-        return false;
-    if (ip32[1] != ip2.ip32[1])
-        return false;
-    if (ip32[2] != ip2.ip32[2])
+    if (ip64[0] != ip2.ip64[0])
         return false;
-    if (ip32[3] != ip2.ip32[3])
+    if (ip64[1] != ip2.ip64[1])
         return false;
 
     return true;
diff --git a/src/stream/base/stream_module.cc b/src/stream/base/stream_module.cc
index c7a592731..1f52643a9 100644
--- a/src/stream/base/stream_module.cc
+++ b/src/stream/base/stream_module.cc
@@ -24,14 +24,20 @@
 
 #include "stream_module.h"
 
+#include "control/control.h"
 #include "detection/rules.h"
+#include "flow/flow_cache.h"
 #include "log/messages.h"
+#include "lua/lua.h"
+#include "main/analyzer_command.h"
 #include "main/snort.h"
 #include "main/snort_config.h"
+#include "managers/inspector_manager.h"
 #include "stream/flush_bucket.h"
 #include "stream/tcp/tcp_stream_tracker.h"
 #include "time/packet_time.h"
 #include "trace/trace.h"
+#include "flow/filter_flow_critera.h"
 
 using namespace snort;
 using namespace std;
@@ -132,6 +138,128 @@ const TraceOption* StreamModule::get_trace_options() const
 #endif
 }
 
+static int dump_flows(lua_State* L)
+{
+    ControlConn* ctrlcon = ControlConn::query_from_lua(L);
+    PktType proto_type = PktType::NONE;
+    Inspector* inspector = InspectorManager::get_inspector("stream", Module::GLOBAL, IT_STREAM);
+    if (!inspector)
+    {
+        LogRespond(ctrlcon, "Dump flows requires stream to be configured\n");
+        return -1;
+    }
+    const char* file_name = luaL_optstring(L, 1, nullptr);
+    if (!file_name)
+    {
+        LogRespond(ctrlcon, "Dump flows requires a file name\n");
+        return -1;
+    }
+    int count = luaL_optint(L, 2, 4);
+    if (0 >= count || 100 < count)
+    {
+        LogRespond(ctrlcon, "Dump flows requires a count value of 1-100\n");
+        return -1;
+    }
+    const char* protocol = luaL_optstring(L, 3, nullptr);
+    if (!protocol)
+    {
+        LogRespond(ctrlcon, "protocol must be a string or convertible to a string\n");
+        return -1;
+    }
+    
+    if (protocol[0] != '\0')
+    {
+        auto proto_it = protocol_to_type.find(protocol);
+        if (proto_it == protocol_to_type.end())
+        {
+            LogRespond(ctrlcon, "valid protocols are IP/TCP/UDP/ICMP\n");
+            return -1;
+        }
+        else
+            proto_type = proto_it->second;
+    }
+
+    std::string source_ip = luaL_optstring(L, 4, nullptr);
+    if (!source_ip.c_str())
+    {
+        LogRespond(ctrlcon, "source_ip must be a string or convertible to a string\n");
+        return -1;
+    }
+    std::string destination_ip= luaL_optstring(L, 5, nullptr);
+    if (!destination_ip.c_str())
+    {
+        LogRespond(ctrlcon, "destination_ip must be a string or convertible to a string\n");
+        return -1;
+    }
+    int source_port = luaL_optint(L, 6, -1);
+    if ( source_port<0 || source_port>65535 )
+    {
+        LogRespond(ctrlcon, "source_port must be between 0-65535\n");
+        return -1;
+    }
+    int destination_port = luaL_optint(L, 7, -1);
+    if ( destination_port<0 || destination_port>65535)
+    {
+        LogRespond(ctrlcon, "destination_port must be between 0-65535\n");
+        return -1;
+    }
+
+/*resume count is used to complete the command execution from
+uncompleted queue*/
+#ifdef REG_TEST
+    int resume = luaL_optint(L, 8, -1);
+#endif
+    DumpFlows* df = new DumpFlows(count, ctrlcon
+#ifdef REG_TEST
+        , resume
+#endif
+    );
+    SfIp src_ip,src_subnet;
+    if (!df->set_ip(source_ip, src_ip, src_subnet))
+    {
+        LogRespond(ctrlcon, "Invalid source ip\n");
+        delete df;
+        return -1;
+    }
+    SfIp dst_ip,dst_subnet;
+    if (!df->set_ip(destination_ip, dst_ip, dst_subnet))
+    {
+        LogRespond(ctrlcon, "Invalid destination ip\n");
+        delete df;
+        return -1;
+    }
+
+    FilterFlowCriteria ffc;
+    ffc.pkt_type = proto_type;
+    ffc.source_port = static_cast<uint16_t>(source_port);
+    ffc.destination_port = static_cast<uint16_t>(destination_port);
+    ffc.source_sfip=src_ip;
+    ffc.destination_sfip=dst_ip;
+    ffc.source_subnet_sfip=src_subnet;
+    ffc.destination_subnet_sfip=dst_subnet;
+    df->set_filter_criteria(ffc);
+
+    if (!df->open_files(file_name))
+    {
+        delete df;
+        return -1;
+    }
+
+    LogRespond(ctrlcon, "== dumping connections\n");
+    main_broadcast_command(df, ctrlcon);
+    return 0;
+}
+
+static const Command stream_cmds[] =
+{
+    { "dump_flows", dump_flows, nullptr, "dump the flow table" },
+
+    { nullptr, nullptr, nullptr, nullptr }
+};
+
+const snort::Command* StreamModule::get_commands() const
+{ return stream_cmds; }
+
 const PegInfo* StreamModule::get_pegs() const
 { return base_pegs; }
 
diff --git a/src/stream/base/stream_module.h b/src/stream/base/stream_module.h
index 3fb169b90..1af46bc33 100644
--- a/src/stream/base/stream_module.h
+++ b/src/stream/base/stream_module.h
@@ -21,8 +21,11 @@
 #ifndef STREAM_MODULE_H
 #define STREAM_MODULE_H
 
+#include <map>
+
 #include "flow/flow_config.h"
 #include "flow/flow_control.h"
+#include "framework/decode_data.h"
 #include "framework/module.h"
 #include "main/analyzer.h"
 #include "main/reload_tuner.h"
@@ -37,6 +40,15 @@ extern THREAD_LOCAL snort::ProfileStats s5PerfStats;
 extern THREAD_LOCAL class FlowControl* flow_con;
 extern THREAD_LOCAL const snort::Trace* stream_trace;
 
+static const std::map<std::string, PktType> protocol_to_type =
+{
+    {"TCP", PktType::TCP},
+    {"UDP", PktType::UDP},
+    {"IP", PktType::IP},
+    {"ICMP", PktType::ICMP},
+};
+
+
 #ifdef DEBUG_MSGS
 enum
 {
@@ -166,6 +178,7 @@ public:
     bool set(const char*, snort::Value&, snort::SnortConfig*) override;
     bool end(const char*, int, snort::SnortConfig*) override;
 
+    const snort::Command* get_commands() const override;
     const PegInfo* get_pegs() const override;
     PegCount* get_counts() const override;
     snort::ProfileStats* get_profile() const override;
diff --git a/src/stream/stream.cc b/src/stream/stream.cc
index ccc6c527c..9d73723ae 100644
--- a/src/stream/stream.cc
+++ b/src/stream/stream.cc
@@ -79,11 +79,14 @@ Flow* Stream::new_flow(const FlowKey* key)
 { return flow_con->new_flow(key); }
 
 void Stream::delete_flow(const FlowKey* key)
-{ flow_con->release_flow(key); }
+{
+    flow_con->release_flow(key);
+}
 
 void Stream::delete_flow(Flow* flow)
 {
-    flow_con->release_flow(flow, PruneReason::NONE);
+    if (flow_con)
+        flow_con->release_flow(flow, PruneReason::NONE);
 }
 
 //-------------------------------------------------------------------------
diff --git a/src/stream/tcp/tcp_trace.cc b/src/stream/tcp/tcp_trace.cc
index 780f53309..611e62993 100644
--- a/src/stream/tcp/tcp_trace.cc
+++ b/src/stream/tcp/tcp_trace.cc
@@ -32,18 +32,23 @@
 #include "tcp_session.h"
 #include "tcp_stream_tracker.h"
 
-#ifndef DEBUG_MSGS
-void S5TraceTCP(const TcpSegmentDescriptor&, const snort::Packet*) { }
-#else
-#define LCL(p, x) ((p).x() - (p).get_iss())
-#define RMT(p, x, q) ((p).x - (q).get_iss())
-
 static const char* const statext[] =
 {
     "LST", "SYS", "SYR", "EST", "MDS", "MDR", "FW1", "FW2", "CLW",
     "CLG", "LAK", "TWT", "CLD", "NON"
 };
 
+const char* stream_tcp_state_to_str(const TcpStreamTracker& t)
+{
+    return statext[t.get_tcp_state()];
+}
+
+#ifndef DEBUG_MSGS
+void S5TraceTCP(const TcpSegmentDescriptor&, const snort::Packet*) { }
+#else
+#define LCL(p, x) ((p).x() - (p).get_iss())
+#define RMT(p, x, q) ((p).x - (q).get_iss())
+
 static const char* const flushxt[] = { "IGN", "FPR", "PRE", "PRO", "PAF" };
 
 inline void TraceEvent(const TcpSegmentDescriptor& tsd, uint32_t txd, uint32_t rxd,
@@ -133,7 +138,7 @@ inline void TraceState(const TcpStreamTracker& a, const TcpStreamTracker& b, con
 
     debug_logf(stream_tcp_trace, TRACE_STATE, p,
         "      %s ST=%s      UA=%-4u NS=%-4u LW=%-5u RN=%-4u RW=%-4u ISS=%-4u IRS=%-4u\n",
-        s, statext[a.get_tcp_state()], ua, ns, a.get_snd_wnd( ),
+        s, stream_tcp_state_to_str(a), ua, ns, a.get_snd_wnd( ),
         RMT(a, rcv_nxt, b), RMT(a, r_win_base, b), a.get_iss(), a.get_irs());
 
     unsigned paf = a.is_splitter_paf() ? 2 : 0;
diff --git a/src/stream/tcp/tcp_trace.h b/src/stream/tcp/tcp_trace.h
index e1e68b08e..fddf362c6 100644
--- a/src/stream/tcp/tcp_trace.h
+++ b/src/stream/tcp/tcp_trace.h
@@ -29,6 +29,7 @@ struct Packet;
 }
 
 class TcpSegmentDescriptor;
+class TcpStreamTracker;
 
 enum
 {
@@ -38,5 +39,7 @@ enum
 
 void S5TraceTCP(const TcpSegmentDescriptor&, const snort::Packet*);
 
+const char* stream_tcp_state_to_str(const TcpStreamTracker&);
+
 #endif