Add support to TCP and UDP protocols

CMakeLists.txt

@@ -29,6 +29,7 @@ if (NOT DEFINED FILTER)
         CONTENT_INSPECTION
         BUFFER
         YARA
+        TEST
     )
 else (NOT DEFINED FILTER)
     set(
@@ -72,12 +73,15 @@ set(
 
 include_directories(
         toolkit
+        toolkit/thread-pool-cpp/include/
         samples/base
+        samples/base/network
         samples/
 	${HIREDIS_INCLUDE_DIRS}
 	${Boost_INCLUDE_DIRS}
 )
 
+add_compile_options(-Wno-unknown-pragmas)
 
 ####################
 #   CORE SOURCES   #
@@ -96,9 +100,22 @@ set(
         samples/base/ThreadGroup.cpp samples/base/ThreadGroup.hpp
         samples/base/AlertManager.cpp samples/base/AlertManager.hpp
 
-        samples/base/Server.cpp samples/base/Server.hpp
+        samples/base/DarwinPacket.cpp samples/base/DarwinPacket.hpp
+        samples/base/network/ANextFilterConnector.cpp samples/base/network/ANextFilterConnector.hpp
+        samples/base/network/UnixNextFilterConnector.cpp samples/base/network/UnixNextFilterConnector.hpp
+        samples/base/network/TcpNextFilterConnector.cpp samples/base/network/TcpNextFilterConnector.hpp
+        samples/base/network/UdpNextFilterConnector.cpp samples/base/network/UdpNextFilterConnector.hpp
+
+        samples/base/network/AServer.cpp samples/base/network/AServer.hpp
+        samples/base/network/UnixServer.cpp samples/base/network/UnixServer.hpp
+        samples/base/network/TcpServer.cpp samples/base/network/TcpServer.hpp
+        samples/base/network/UdpServer.cpp samples/base/network/UdpServer.hpp
         samples/base/Manager.cpp samples/base/Manager.hpp
-        samples/base/Session.cpp samples/base/Session.hpp
+        samples/base/network/ASession.cpp samples/base/network/ASession.hpp
+        samples/base/network/UnixSession.cpp samples/base/network/UnixSession.hpp
+        samples/base/network/TcpSession.cpp samples/base/network/TcpSession.hpp
+        samples/base/network/UdpSession.cpp samples/base/network/UdpSession.hpp
+        samples/base/ATask.cpp samples/base/ATask.hpp
 
         toolkit/Network.cpp toolkit/Network.hpp
         toolkit/Validators.cpp toolkit/Validators.hpp

manager/HeartBeat.py

@@ -7,22 +7,46 @@
 __doc__ = 'The heartbeat functions and main class'
 
 from os import kill, access, F_OK
-
+import socket
 
 class HeartBeat:
     """
     This class is in charge of the health check for the filters.
     """
 
     @staticmethod
-    def check_socket(file):
+    def check_unix_socket(file):
         try:
             if access(file, F_OK):
                 return True
             return False
         except Exception:
             return False
 
+    def check_network_socket(filter_network):
+        if filter_network['socket_type'] == 'UNIX':
+            return HeartBeat.check_unix_socket(filter_network['address_path'])
+
+        elif filter_network['socket_type'] == 'TCP':
+            # parse ip address field
+            splitted_address = filter_network['address_path'].rsplit(':', 1)
+            if '[' in splitted_address[0]: # ipv6 address
+                host_address = splitted_address[0][1:-1]
+            else:
+                host_address = splitted_address[0]
+            host_port = int(splitted_address[1])
+            #test connection to socket
+            if ':' in host_address: # ipv6
+                s = socket.socket(socket.AF_INET6)
+            else:
+                s = socket.socket(socket.AF_INET)
+            s.settimeout(2)
+            res = s.connect_ex((host_address,host_port))
+            s.close()
+            return res == 0
+        else: #No check for UDP
+            return True
+
     @staticmethod
     def check_pid_file(file):
         """

manager/Services.py

@@ -50,9 +50,10 @@ def start_all(self):
                     self.stop_one(filter, no_lock=True)
                     self.clean_one(filter, no_lock=True)
                 else:
-                    logger.debug("Linking UNIX sockets...")
-                    filter['status'] = psutil.STATUS_RUNNING
-                    call(['ln', '-s', filter['socket'], filter['socket_link']])
+                    if filter['network']['socket_type'] == 'UNIX':
+                        logger.debug("Linking UNIX sockets...")
+                        filter['status'] = psutil.STATUS_RUNNING
+                        call(['ln', '-s', filter['socket'], filter['socket_link']])
 
     def rotate_logs_all(self):
         """
@@ -89,6 +90,14 @@ def _build_cmd(filt):
 
         cmd = [filt['exec_path']]
 
+        # Flags MUST be before positional arguments as the parsing on HardenedBSD is not done on all the arguments
+        # On BSD getopt stops at the first argument which is not in the specified flags
+        if filt['network']['socket_type'] == 'UDP':
+            cmd.append('-u')
+
+        if filt['next_filter_network']['socket_type'] == 'UDP':
+            cmd.append('-v')
+
         try:
             if filt['log_level'] not in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "DEVELOPER"]:
                 logger.warning(
@@ -99,15 +108,15 @@ def _build_cmd(filt):
                 cmd.append(filt['log_level'])
         except KeyError:
             pass
-
+        
         cmd += [
             filt['name'],
-            filt['socket'],
+            filt['network']['address_path'],
             filt['config_file'],
             filt['monitoring'],
             filt['pid_file'],
             filt['output'],
-            filt['next_filter_unix_socket'],
+            filt['next_filter_network']['address_path'],
             str(filt['nb_thread']),
             str(filt['cache_size']),
             str(filt['threshold']),
@@ -369,6 +378,9 @@ def update(self, names, prefix, suffix):
                     prefix=prefix, suffix=suffix,
                     name=n, extension=new[n]['extension']
                 )
+                # TODO Handle TCP case?
+                if new[n]['network']['socket_type'] == 'UNIX':
+                    new[n]['network']['address_path'] = new[n]['socket']
 
                 new[n]['monitoring'] = '{prefix}/sockets{suffix}/{name}_mon{extension}.sock'.format(
                     prefix=prefix, suffix=suffix,
@@ -531,7 +543,7 @@ def _wait_process_ready(content):
                 status = "Process not running"
                 continue
 
-            if not HeartBeat.check_socket(content['monitoring']):
+            if not HeartBeat.check_unix_socket(content['monitoring']):
                 status = "Monitoring socket not created"
                 continue
 
@@ -550,7 +562,7 @@ def _wait_process_ready(content):
                 continue
 
             if "running" in resp:
-                if not HeartBeat.check_socket(content['socket']):
+                if not HeartBeat.check_network_socket(content['network']):
                     status = "Main socket not created"
                     continue
             else:
@@ -574,10 +586,10 @@ def hb_one(self, filter):
             if not HeartBeat.check_process(pid):
                 raise Exception('Process not running')
 
-            if not HeartBeat.check_socket(filter['socket']):
+            if not HeartBeat.check_network_socket(filter['network']):
                 raise Exception('Socket not accessible')
 
-            if not HeartBeat.check_socket(filter['monitoring']):
+            if not HeartBeat.check_unix_socket(filter['monitoring']):
                 raise Exception('Monitoring socket not accessible')
 
         except Exception as e:

manager/config.py

@@ -189,10 +189,35 @@ def set_defaults(validator, properties, instance, schema):
                         "enum": ["NONE", "RAW", "LOG", "PARSED"],
                         "default": "NONE"
                         },
-                    "next_filter": {"type": "string"},
+                    "next_filter": {
+                        "OneOf":[
+                            {"type": "string"},
+                            {"type": "object",
+                            "properties": {
+                                "socket_type":{
+                                    "type":"string",
+                                    "enum": ["NONE", "UNIX", "TCP", "UDP"],
+                                    "default":"NONE"
+                                    },
+                                "address_path": {"type": "string"}
+                                }
+                            }
+                        ]
+                    },
                     "threshold": {
                         "type": "integer",
                         "default": 100
+                        },
+                    "network": {
+                        "type": "object",
+                        "properties": {
+                            "socket_type":{
+                                "type":"string",
+                                "enum": ["UNIX", "TCP", "UDP"],
+                                "default":"UNIX"
+                                },
+                            "address_path": {"type": "string"}
+                            }
                         }
                 },
                 "required": ["name", "exec_path", "config_file"],
@@ -268,19 +293,34 @@ def complete_filters_conf(prefix, suffix):
         filter['failures'] = 0
         filter['extension'] = '.1'
         filter['pid_file'] = '{prefix}/run{suffix}/{filter}{extension}.pid'.format(prefix=prefix, suffix=suffix, filter=filter['name'], extension=filter['extension'])
-
-        if not filter['next_filter']:
-            filter['next_filter_unix_socket'] = 'no'
-        else:
-            filter['next_filter_unix_socket'] = '{prefix}/sockets{suffix}/{next_filter}.sock'.format(
-                prefix=prefix, suffix=suffix,
-                next_filter=filter['next_filter']
-            )
-
+
         filter['socket'] = '{prefix}/sockets{suffix}/{filter}{extension}.sock'.format(prefix=prefix, suffix=suffix, filter=filter['name'], extension=filter['extension'])
         filter['socket_link'] = '{prefix}/sockets{suffix}/{filter}.sock'.format(prefix=prefix, suffix=suffix, filter=filter['name'])
 
+        if 'network' not in filter:
+            filter['network'] = { "socket_type":"UNIX", "address_path":filter['socket'] }
+        elif filter['network']['socket_type'] == "UNIX":
+            filter['network']['address_path'] = filter['socket']
+
         filter['monitoring'] = '{prefix}/sockets{suffix}/{filter}_mon{extension}.sock'.format(
             prefix=prefix, suffix=suffix,
             filter=filter['name'], extension=filter['extension']
         )
+
+    # Next filter is setup with a second loop (we need network information already setup)
+    for _, filter in filters.items():
+        if 'next_filter' not in filter or filter['next_filter'] == '':
+            filter['next_filter_network'] = { "socket_type":"NONE", "address_path":"no" }
+        else:
+            if isinstance(filter['next_filter'], str):
+                #check other filters
+                modified = False
+                for _, other_filter in filters.items():
+                    if filter['next_filter'] == other_filter['name']:
+                        filter['next_filter_network'] = other_filter['network']
+                        modified = True
+                if not modified:
+                    raise ConfParseError("Filter '{}' had next_filter configured to '{}' but it was not found in the configuration"
+                                        .format(filter['name'], filter['next_filter']))
+            else:
+                filter['next_filter_network'] = filter['next_filter']

manager/manager.py

@@ -33,7 +33,7 @@ def create_dirs(dirs, prefix, suffix):
     for d in dirs:
         path = '{}/{}{}'.format(prefix, d, suffix)
         if not os.path.exists(path):
-            os.mkdir(path)
+            os.makedirs(path, exist_ok=True)
 
 # Argparse
 parser = argparse.ArgumentParser()

samples/base/AGenerator.cpp

@@ -13,6 +13,17 @@
 #include "AGenerator.hpp"
 #include "AlertManager.hpp"
 
+AGenerator::AGenerator(size_t nb_task_threads): _threadPool{GetThreadPoolOptions(nb_task_threads)}
+{
+    ;
+}
+
+tp::ThreadPoolOptions AGenerator::GetThreadPoolOptions(size_t nb_task_threads){
+    auto opts = tp::ThreadPoolOptions();
+    opts.setThreadCount(nb_task_threads);
+    return opts;
+}
+
 bool AGenerator::Configure(std::string const& configFile, const std::size_t cache_size) {
     DARWIN_LOGGER;
     DARWIN_LOG_DEBUG("AGenerator:: Configuring...");
@@ -116,3 +127,7 @@ bool AGenerator::ReadConfig(const std::string &configuration_file_path) {
     conf_file_stream.close();
     return true;
 }
+
+tp::ThreadPool& AGenerator::GetTaskThreadPool() {
+    return this->_threadPool;
+}

samples/base/AGenerator.hpp

@@ -11,25 +11,27 @@
 #include <iostream>
 #include <fstream>
 #include <string>
+#include <thread_pool.hpp>
 
-#include "Session.hpp"
+#include "ATask.hpp"
 #include "../toolkit/rapidjson/document.h"
 
 class AGenerator {
 public:
-    AGenerator() = default;
+    AGenerator(size_t nb_task_threads);
     virtual ~AGenerator() = default;
 
+
 // Methods to be implemented by the child
 public:
-    /// Create a new task.
     ///
-    /// \param socket The Session's socket.
-    /// \param manager The Session manager.
-    /// \return A pointer to a new session.
-    virtual darwin::session_ptr_t
-    CreateTask(boost::asio::local::stream_protocol::socket& socket,
-               darwin::Manager& manager) noexcept = 0;
+    /// \brief Create a Task object
+    /// 
+    /// \param s a shred pointer to the sessions creating the task
+    /// \return std::shared_ptr<darwin::ATask> a shared pointer to the created task
+    ///
+    virtual std::shared_ptr<darwin::ATask> CreateTask(darwin::session_ptr_t s) noexcept = 0;
+
     virtual bool ConfigureNetworkObject(boost::asio::io_context &context);
 
 protected:
@@ -59,6 +61,8 @@ class AGenerator {
     Configure(std::string const& configFile,
               const std::size_t cache_size) final;
 
+    virtual tp::ThreadPool& GetTaskThreadPool() final;
+
 private:
     /// Open and read the configuration file.
     /// Try to load the json format of the configuration.
@@ -76,7 +80,18 @@ class AGenerator {
     virtual bool ExtractCustomAlertingTags(const rapidjson::Document &configuration,
                                            std::string& tags);
 
+    ///
+    /// \brief Get the configuration object of the task thread pool
+    /// 
+    /// \param nb_task_threads number of workers (threads) to spawn for the tasks
+    /// \return tp::ThreadPoolOptions 
+    ///
+    static tp::ThreadPoolOptions GetThreadPoolOptions(size_t nb_task_threads);
+
 protected:
     std::shared_ptr<boost::compute::detail::lru_cache<xxh::hash64_t, unsigned int>> _cache; //!< The cache for already processed request
     std::mutex _cache_mutex;
+
+    tp::ThreadPool _threadPool;
+
 };