Merge pull request #1642 in SNORT/snort3 from ~KBHANDAN/snort3:ha to master

Squashed commit of the following:

commit 2ba9df6b36c5f614106d178f3ab2d56d399640e4
Author: Kaushal Bhandankar <kbhandan@cisco.com>
Date:   Tue Jun 11 07:49:28 2019 -0400

    flow: Fixes for DAQ-backed HA implementation

diff --git a/src/flow/ha.cc b/src/flow/ha.cc
index 6e25cf730dac401a47fde94eacac8d6526f702d4..e24a69369237a254a17bdf94898ed1cfd8be6767 100644 (file)
--- a/src/flow/ha.cc
+++ b/src/flow/ha.cc
@@ -34,6 +34,7 @@
 #include "flow.h"
 #include "flow_key.h"
 #include "ha_module.h"
+#include "session.h"
 
 using namespace snort;
 
@@ -180,6 +181,12 @@ bool FlowHAState::sync_interval_elapsed()
            ( pkt_time.tv_usec > next_update.tv_usec ) ) );
 }
 
+void FlowHAState::init_next_update()
+{
+    packet_gettimeofday(&next_update);
+    timeradd(&next_update, &min_session_lifetime, &next_update);
+}
+
 void FlowHAState::set_next_update()
 {
     timeradd(&next_update, &min_sync_interval, &next_update);
@@ -189,6 +196,7 @@ void FlowHAState::reset()
 {
     state = INITIAL_STATE;
     pending = NONE_PENDING;
+    init_next_update();
 }
 
 FlowHAClient::FlowHAClient(uint8_t length, bool session_client)
@@ -314,7 +322,7 @@ static uint16_t calculate_update_msg_content_length(Flow& flow, bool full)
         if ((i == 0) || full || flow.ha_state->check_pending(1 << (i - 1)))
         {
             assert(ha->client_map[i]);
-            length += (ha->client_map[i]->get_message_size() + sizeof(HAClientHeader));
+            length += (ha->client_map[i]->get_message_size(flow) + sizeof(HAClientHeader));
         }
     }
 
@@ -377,12 +385,16 @@ static void consume_ha_delete_message(HAMessage&, const FlowKey& key)
     Stream::delete_flow(&key);
 }
 
-static Flow* consume_ha_update_message(HAMessage& msg, const FlowKey& key)
+static Flow* consume_ha_update_message(HAMessage& msg, const FlowKey& key, Packet* p)
 {
     // flow will be nullptr if/when the session does not exist in the caches
+    bool no_flow_found = false;
     Flow* flow = Stream::get_flow(&key);
     if (!flow)
+    {
+        no_flow_found = true;
         ha_stats.update_msgs_recv_no_flow++;
+    }
 
     // pointer to one past the last byte in the message
     const uint8_t* content_end = msg.buffer + msg.buffer_length;
@@ -429,10 +441,15 @@ static Flow* consume_ha_update_message(HAMessage& msg, const FlowKey& key)
     if (msg.cursor == content_end)
         ha_stats.update_msgs_consumed++;
 
+    if( p && no_flow_found && flow && flow->session )
+    {
+        flow->session->setup(p);
+    }
+
     return flow;
 }
 
-static Flow* consume_ha_message(HAMessage& msg)
+static Flow* consume_ha_message(HAMessage& msg, Packet* p = nullptr)
 {
     ha_stats.msgs_recv++;
 
@@ -473,7 +490,7 @@ static Flow* consume_ha_message(HAMessage& msg)
         }
         case HA_UPDATE_EVENT:
         {
-            flow = consume_ha_update_message(msg, key);
+            flow = consume_ha_update_message(msg, key, p);
             ha_stats.update_msgs_recv++;
             break;
         }
@@ -635,7 +652,7 @@ Flow* HighAvailability::process_daq_import(Packet& p, FlowKey& key)
         if (p.daq_instance->ioctl(DIOCTL_GET_FLOW_HA_STATE, &fhs, sizeof(fhs)) == DAQ_SUCCESS)
         {
             HAMessage ha_msg(fhs.data, fhs.length);
-            flow = consume_ha_message(ha_msg);
+            flow = consume_ha_message(ha_msg, &p);
             ha_stats.daq_imports++;
             // Validate that the imported flow matches up with the given flow key.
             if (flow)

diff --git a/src/flow/ha.h b/src/flow/ha.h
index 47525d27940e78e4d5016d27f0bbe8f6dceaa838..229c8903e222116714833bad2da30ef5e9ca4667 100644 (file)
--- a/src/flow/ha.h
+++ b/src/flow/ha.h
@@ -72,6 +72,7 @@ public:
     bool check_any(uint8_t state);
     static void config_timers(struct timeval, struct timeval);
     bool sync_interval_elapsed();
+    void init_next_update();
     void set_next_update();
     void reset();
 
@@ -127,7 +128,7 @@ public:
     virtual bool consume(snort::Flow*&, const snort::FlowKey*, snort::HAMessage&, uint8_t size) = 0;
     virtual bool produce(snort::Flow&, snort::HAMessage&) = 0;
     virtual bool is_update_required(snort::Flow*) { return false; }
-    uint8_t get_message_size() { return max_length; }
+    virtual uint8_t get_message_size(Flow&) { return max_length; }
 
     FlowHAClientHandle handle;  // Actual handle for the instance
     uint8_t index;

diff --git a/src/flow/test/ha_test.cc b/src/flow/test/ha_test.cc
index 7e1f8baf0727fb3a660ba9a0b74a8dba5b3c5bfe..7c61c2a36d0791ad70b7669b2d71bfe5a0fc009d 100644 (file)
--- a/src/flow/test/ha_test.cc
+++ b/src/flow/test/ha_test.cc
@@ -498,7 +498,7 @@ TEST(high_availability_test, consume_error_truncated_client_hdr)
     HAMessage msg((uint8_t*) &chdr, sizeof(chdr) / 2);
     FlowKey key;
 
-    consume_ha_update_message(msg, key);
+    consume_ha_update_message(msg, key, &s_pkt);
     CHECK(ha_stats.update_msgs_consumed == 0);
     CHECK(ha_stats.truncated_msgs == 1);
 }
@@ -509,7 +509,7 @@ TEST(high_availability_test, consume_error_invalid_client_idx)
     HAMessage msg((uint8_t*) &chdr, sizeof(chdr));
     FlowKey key;
 
-    consume_ha_update_message(msg, key);
+    consume_ha_update_message(msg, key, &s_pkt);
     CHECK(ha_stats.update_msgs_consumed == 0);
     CHECK(ha_stats.unknown_client_idx == 1);
 }
@@ -524,7 +524,7 @@ TEST(high_availability_test, consume_error_truncated_client_msg)
     HAMessage msg((uint8_t*) &input, sizeof(input));
     FlowKey key;
 
-    consume_ha_update_message(msg, key);
+    consume_ha_update_message(msg, key, &s_pkt);
     CHECK(ha_stats.update_msgs_consumed == 0);
     CHECK(ha_stats.truncated_msgs == 1);
 }
@@ -539,7 +539,7 @@ TEST(high_availability_test, consume_error_client_consume)
     HAMessage msg((uint8_t*) &input, sizeof(input));
     FlowKey key;
 
-    consume_ha_update_message(msg, key);
+    consume_ha_update_message(msg, key, &s_pkt);
     CHECK(ha_stats.update_msgs_consumed == 0);
     CHECK(ha_stats.client_consume_errors == 1);
 }
@@ -549,7 +549,7 @@ TEST(high_availability_test, consume_error_truncated_msg_hdr)
     HAMessageHeader hdr = { };
     HAMessage msg((uint8_t*) &hdr, sizeof(hdr) / 2);
 
-    CHECK(consume_ha_message(msg) == nullptr);
+    CHECK(consume_ha_message(msg, &s_pkt) == nullptr);
     CHECK(ha_stats.truncated_msgs == 1);
 }
 
@@ -558,7 +558,7 @@ TEST(high_availability_test, consume_error_version_mismatch)
     HAMessageHeader hdr = { 0, HA_MESSAGE_VERSION + 1, 0, 0 };
     HAMessage msg((uint8_t*) &hdr, sizeof(hdr));
 
-    CHECK(consume_ha_message(msg) == nullptr);
+    CHECK(consume_ha_message(msg, &s_pkt) == nullptr);
     CHECK(ha_stats.msg_version_mismatch == 1);
 }
 
@@ -567,7 +567,7 @@ TEST(high_availability_test, consume_error_length_mismatch)
     HAMessageHeader hdr = { 0, HA_MESSAGE_VERSION, 0x42, 0 };
     HAMessage msg((uint8_t*) &hdr, sizeof(hdr));
 
-    CHECK(consume_ha_message(msg) == nullptr);
+    CHECK(consume_ha_message(msg, &s_pkt) == nullptr);
     CHECK(ha_stats.msg_length_mismatch == 1);
 }
 

diff --git a/src/network_inspectors/appid/appid_api.cc b/src/network_inspectors/appid/appid_api.cc
index c4c313a77708afe8df9e355d91075886099b3f48..d4afb331393f90ce41ad2abbfd5cef0eeddea088 100644 (file)
--- a/src/network_inspectors/appid/appid_api.cc
+++ b/src/network_inspectors/appid/appid_api.cc
@@ -25,8 +25,10 @@
 
 #include "appid_api.h"
 
+#include "managers/inspector_manager.h"
 #include "utils/util.h"
 
+#include "appid_module.h"
 #include "appid_session.h"
 #include "appid_session_api.h"
 #include "app_info_table.h"
@@ -121,10 +123,8 @@ uint32_t AppIdApi::produce_ha_state(Flow& flow, uint8_t* buf)
     return sizeof(*appHA);
 }
 
-// FIXIT-H last param AppIdSession ctor is appid inspector, we need that but no good way to get it
-// at the moment...code to allocate session ifdef'ed out until this is resolved...
-uint32_t AppIdApi::consume_ha_state(Flow& flow, const uint8_t* buf, uint8_t, IpProtocol /*proto*/,
-    SfIp* /*ip*/, uint16_t /*port*/)
+uint32_t AppIdApi::consume_ha_state(Flow& flow, const uint8_t* buf, uint8_t, IpProtocol proto,
+    SfIp* ip, uint16_t port)
 {
     const AppIdSessionHA* appHA = (const AppIdSessionHA*)buf;
     if (appHA->flags & APPID_HA_FLAGS_APP)
@@ -132,37 +132,39 @@ uint32_t AppIdApi::consume_ha_state(Flow& flow, const uint8_t* buf, uint8_t, IpP
         AppIdSession* asd =
             (AppIdSession*)(flow.get_flow_data(AppIdSession::inspector_id));
 
-#ifdef APPID_HA_SUPPORT_ENABLED
         if (!asd)
         {
-            asd = new AppIdSession(proto, ip, port, nullptr);
-            flow.set_flow_data(asd);
-            asd->service.set_id(appHA->appId[1]);
-            if ( asd->service.get_id() == APP_ID_FTP_CONTROL )
+            AppIdInspector* inspector = (AppIdInspector*) InspectorManager::get_inspector(MOD_NAME, true);
+            if(inspector)
             {
-                asd->set_session_flags(APPID_SESSION_CLIENT_DETECTED |
-                    APPID_SESSION_NOT_A_SERVICE | APPID_SESSION_SERVICE_DETECTED);
-                if ( !ServiceDiscovery::add_ftp_service_state(*asd) )
-                    asd->set_session_flags(APPID_SESSION_CONTINUE);
 
-                asd->service_disco_state = APPID_DISCO_STATE_STATEFUL;
-            }
-            else
-                asd->service_disco_state = APPID_DISCO_STATE_FINISHED;
-
-            asd->client_disco_state = APPID_DISCO_STATE_FINISHED;
+                asd = new AppIdSession(proto, ip, port, *inspector);
+                flow.set_flow_data(asd);
+                asd->service.set_id(appHA->appId[1]);
+                if ( asd->service.get_id() == APP_ID_FTP_CONTROL )
+                {
+                    asd->set_session_flags(APPID_SESSION_CLIENT_DETECTED |
+                            APPID_SESSION_NOT_A_SERVICE | APPID_SESSION_SERVICE_DETECTED);
+                    if ( !ServiceDiscovery::add_ftp_service_state(*asd) )
+                        asd->set_session_flags(APPID_SESSION_CONTINUE);
+
+                    asd->service_disco_state = APPID_DISCO_STATE_STATEFUL;
+                }
+                else
+                    asd->service_disco_state = APPID_DISCO_STATE_FINISHED;
+
+                asd->client_disco_state = APPID_DISCO_STATE_FINISHED;
 #ifdef ENABLE_APPID_THIRD_PARTY
-            if (asd->tpsession)
-                asd->tpsession->set_state(TP_STATE_HA);
+                if (asd->tpsession)
+                    asd->tpsession->set_state(TP_STATE_HA);
 #endif
+            }
         }
-#else
+
         if ( !asd )
         {
-            assert(false);
             return sizeof(*appHA);
         }
-#endif
 
         if( (appHA->flags & APPID_HA_FLAGS_TP_DONE) && asd->tpsession )
         {

diff --git a/src/network_inspectors/appid/test/appid_api_test.cc b/src/network_inspectors/appid/test/appid_api_test.cc
index 23dff4411989538f3451f04372a94d600fe4d4eb..685de0dc50e48b2949ce53c743755c873d6210a6 100644 (file)
--- a/src/network_inspectors/appid/test/appid_api_test.cc
+++ b/src/network_inspectors/appid/test/appid_api_test.cc
@@ -45,6 +45,14 @@
 
 using namespace snort;
 
+namespace snort
+{
+
+class Inspector* InspectorManager::get_inspector(char const*, bool, SnortConfig*)
+{ return nullptr; }
+
+}
+
 const char* AppInfoManager::get_app_name(AppId)
 {
     return test_app_name;
@@ -87,7 +95,6 @@ TEST(appid_api, get_application_id)
 }
 
 // FIXIT - enable this test when consume ha appid api call is fixed
-#ifdef APPID_HA_SUPPORT_ENABLED
 TEST(appid_api, produce_ha_state)
 {
     AppIdSessionHA appHA, cmp_buf;
@@ -96,11 +103,11 @@ TEST(appid_api, produce_ha_state)
     memset((void*)&cmp_buf, 0, sizeof(cmp_buf));
     mock_session->common.flow_type = APPID_FLOW_TYPE_IGNORE;
     mock_session->common.flags |= APPID_SESSION_SERVICE_DETECTED | APPID_SESSION_HTTP_SESSION;
-    uint32_t val = appid_api.produce_ha_state(flow, (uint8_t*)&appHA);
+    uint32_t val = appid_api.produce_ha_state(*flow, (uint8_t*)&appHA);
     CHECK_TRUE(val == sizeof(appHA));
     CHECK_TRUE(memcmp(&appHA, &cmp_buf, val) == 0);
     mock_session->common.flow_type = APPID_FLOW_TYPE_NORMAL;
-    val = appid_api.produce_ha_state(flow, (uint8_t*)&appHA);
+    val = appid_api.produce_ha_state(*flow, (uint8_t*)&appHA);
     CHECK_TRUE(val == sizeof(appHA));
     CHECK_TRUE(appHA.appId[0] == APPID_UT_ID);
     CHECK_TRUE(appHA.appId[1] == APPID_UT_ID + 1);
@@ -116,7 +123,10 @@ TEST(appid_api, produce_ha_state)
     mock_flow_data= nullptr;
     SfIp ip;
     ip.pton(AF_INET, "192.168.1.222");
-    appid_api.consume_ha_state(flow, (uint8_t*)&appHA, 0, IpProtocol::TCP, &ip, 1066);
+    val = appid_api.consume_ha_state(*flow, (uint8_t*)&appHA, 0, IpProtocol::TCP, &ip, 1066);
+    CHECK_TRUE(val == sizeof(appHA));
+    //FIXIT-H refactor below code to test AppId consume functionality
+    /*
     AppIdSession* session = (AppIdSession*)flow->get_flow_data(AppIdSession::inspector_id);
     CHECK_TRUE(session);
     CHECK_TRUE(session->get_tp_app_id() == appHA.appId[0]);
@@ -130,11 +140,15 @@ TEST(appid_api, produce_ha_state)
     CHECK_TRUE(session->service_disco_state == APPID_DISCO_STATE_FINISHED);
     CHECK_TRUE(session->client_disco_state == APPID_DISCO_STATE_FINISHED);
     delete session;
+    */
 
     // test logic when service app is ftp control
     appHA.appId[1] = APP_ID_FTP_CONTROL;
     mock_flow_data= nullptr;
-    appid_api.consume_ha_state(flow, (uint8_t*)&appHA, 0, IpProtocol::TCP, &ip, 1066);
+    val = appid_api.consume_ha_state(*flow, (uint8_t*)&appHA, 0, IpProtocol::TCP, &ip, 1066);
+    CHECK_TRUE(val == sizeof(appHA));
+    //FIXIT-H refactor below code to test AppId consume functionality
+    /*
     session = (AppIdSession*)flow->get_flow_data(AppIdSession::inspector_id);
     CHECK_TRUE(session);
     uint64_t flags = session->get_session_flags(APPID_SESSION_CLIENT_DETECTED |
@@ -145,8 +159,8 @@ TEST(appid_api, produce_ha_state)
     CHECK_TRUE(session->service_disco_state == APPID_DISCO_STATE_STATEFUL);
     CHECK_TRUE(session->client_disco_state == APPID_DISCO_STATE_FINISHED);
     delete session;
+    */
 }
-#endif
 
 TEST(appid_api, create_appid_session_api)
 {

diff --git a/src/stream/base/stream_ha.cc b/src/stream/base/stream_ha.cc
index ff7bc6b9b4bf5d632bc68ab447e20acf9db29d7a..5ab88cb530db704587a9a35fd79cb6a6c5ee6df0 100644 (file)
--- a/src/stream/base/stream_ha.cc
+++ b/src/stream/base/stream_ha.cc
@@ -146,6 +146,8 @@ bool StreamHAClient::produce(Flow& flow, HAMessage& msg)
     SessionHAContent* hac = (SessionHAContent*) msg.cursor;
 
     hac->ssn_state = flow.ssn_state;
+    hac->ssn_state.session_flags &= ~HA_IGNORED_SESSION_FLAGS;
+
     hac->flow_state = flow.flow_state;
     hac->flags = 0;
     if (!is_client_lower(flow))