Migrate configuration to ossec.conf

framework/scripts/tests/test_wazuh_clusterd.py

@@ -140,10 +140,11 @@ def start(cls):
             return 'HAPHELPER_START'
 
 
-    async def gather(first, second, third):
+    async def gather(first, second):
         assert first == 'MASTER_START'
         assert second == 'LOCALSERVER_START'
-        assert third == 'HAPHELPER_START'
+        # FIXME: (20940) When write UT for the new components.
+        # assert third == 'HAPHELPER_START'
 
 
     wazuh_clusterd.cluster_utils = cluster_utils

framework/scripts/wazuh_clusterd.py

@@ -42,7 +42,7 @@ def set_logging(foreground_mode=False, debug_mode=0) -> WazuhLogger:
 
 def print_version():
     """Print Wazuh metadata."""
-    from wazuh.core.cluster import __version__, __author__, __wazuh_name__, __licence__
+    from wazuh.core.cluster import __author__, __licence__, __version__, __wazuh_name__
     print(f"\n{__wazuh_name__} {__version__} - {__author__}\n\n{__licence__}")
 
 
@@ -81,7 +81,7 @@ async def master_main(args: argparse.Namespace, cluster_config: dict, cluster_it
     logger : WazuhLogger
         Cluster logger.
     """
-    from wazuh.core.cluster import master, local_server
+    from wazuh.core.cluster import local_server, master
     from wazuh.core.cluster.hap_helper.hap_helper import HAPHelper
 
     cluster_utils.context_tag.set('Master')
@@ -96,7 +96,10 @@ async def master_main(args: argparse.Namespace, cluster_config: dict, cluster_it
                                                      concurrency_test=args.concurrency_test, node=my_server,
                                                      configuration=cluster_config, enable_ssl=args.ssl,
                                                      cluster_items=cluster_items)
-    await asyncio.gather(my_server.start(), my_local_server.start(), HAPHelper.start())
+    tasks = [my_server, my_local_server]
+    if not cluster_config.get(cluster_utils.HAPROXY_HELPER, {}).get(cluster_utils.HAPROXY_DISABLED, True):
+        tasks.append(HAPHelper)
+    await asyncio.gather(*[task.start() for task in tasks])
 
 
 #
@@ -116,8 +119,9 @@ async def worker_main(args: argparse.Namespace, cluster_config: dict, cluster_it
     logger : WazuhLogger
         Cluster logger.
     """
-    from wazuh.core.cluster import worker, local_server
     from concurrent.futures import ProcessPoolExecutor
+
+    from wazuh.core.cluster import local_server, worker
     cluster_utils.context_tag.set('Worker')
 
     # Pool is defined here so the child process is not recreated when the connection with master node is broken.
@@ -254,7 +258,7 @@ def main():
 
 if __name__ == '__main__':
     import wazuh.core.cluster.utils as cluster_utils
-    from wazuh.core import pyDaemonModule, common, configuration
+    from wazuh.core import common, configuration, pyDaemonModule
 
     cluster_items = cluster_utils.get_cluster_items()
     original_sig_handler = signal.signal(signal.SIGTERM, exit_handler)

framework/wazuh/core/cluster/cluster.py

@@ -16,23 +16,73 @@
 from os import listdir, path, remove, stat, walk
 from uuid import uuid4
 
+from jsonschema import ValidationError, validate
 from wazuh import WazuhError, WazuhException, WazuhInternalError
 from wazuh.core import common
+from wazuh.core.cluster.utils import (
+    AGENT_CHUNK_SIZE,
+    AGENT_RECONNECTION_STABILITY_TIME,
+    AGENT_RECONNECTION_TIME,
+    FREQUENCY,
+    HAPROXY_HELPER,
+    HAPROXY_PORT,
+    HAPROXY_PROTOCOL,
+    IMBALANCE_TOLERANCE,
+    REMOVE_DISCONNECTED_NODE_AFTER,
+    get_cluster_items,
+    read_config,
+)
 from wazuh.core.InputValidator import InputValidator
-from wazuh.core.cluster.utils import get_cluster_items, read_config
-from wazuh.core.utils import blake2b, mkdir_with_mode, get_utc_now, get_date_from_timestamp, to_relative_path
+from wazuh.core.utils import blake2b, get_date_from_timestamp, get_utc_now, mkdir_with_mode, to_relative_path
 
 logger = logging.getLogger('wazuh')
 
 # Separators used in compression/decompression functions to delimit files.
 FILE_SEP = '|@@//@@|'
 PATH_SEP = '|//@@//|'
+MIN_PORT = 1024
+MAX_PÖRT = 65535
 
 
 #
 # Cluster
 #
 
+def validate_haproxy_helper_config(config: dict):
+    """Validate the values of the give HAProxy helper configuration.
+
+    Parameters
+    ----------
+    config : dict
+        Configuration to validate.
+
+    Raises
+    ------
+    WazuhError(3004)
+        If there any invalid value.
+    """
+    SCHEMA = {
+        'type': 'object',
+        'properties': {
+            HAPROXY_PORT: {'type': 'integer', 'minimum': MIN_PORT, 'maximum': MAX_PÖRT},
+            HAPROXY_PROTOCOL: {'type': 'string', 'enum': ['http', 'https']},
+            FREQUENCY: {'type': 'integer', 'minimum': 10},
+            AGENT_RECONNECTION_STABILITY_TIME: {'type': 'integer', 'minimum': 10},
+            AGENT_CHUNK_SIZE: {'type': 'integer', 'minimum': 100},
+            AGENT_RECONNECTION_TIME: {'type': 'integer', 'minimum': 0},
+            IMBALANCE_TOLERANCE: {'type': 'number', 'minimum': 0, 'exclusiveMinimum': True, 'maximum': 1},
+            REMOVE_DISCONNECTED_NODE_AFTER: {'type': 'integer', 'minimum': 0},
+        },
+    }
+
+    try:
+        validate(config, SCHEMA)
+    except ValidationError as error:
+        raise WazuhError(
+            3004,
+            f'Invalid value for {error.path.pop()}. {error.message}'
+        )
+
 
 def check_cluster_config(config):
     """Verify that cluster configuration is correct.
@@ -71,8 +121,8 @@ def check_cluster_config(config):
     elif not isinstance(config['port'], int):
         raise WazuhError(3004, "Port has to be an integer.")
 
-    elif not 1024 < config['port'] < 65535:
-        raise WazuhError(3004, "Port must be higher than 1024 and lower than 65535.")
+    elif not MIN_PORT < config['port'] < MAX_PÖRT:
+        raise WazuhError(3004, f"Port must be higher than {MIN_PORT} and lower than {MAX_PÖRT}.")
 
     if len(config['nodes']) > 1:
         logger.warning(
@@ -84,6 +134,8 @@ def check_cluster_config(config):
     if len(invalid_elements) != 0:
         raise WazuhError(3004, f"Invalid elements in node fields: {', '.join(invalid_elements)}.")
 
+    validate_haproxy_helper_config(config.get(HAPROXY_HELPER, {}))
+
 
 def get_node():
     """Get dict with current active node information.
@@ -374,7 +426,7 @@ def compress_files(name, list_path, cluster_control_json=None, max_zip_size=None
             except zlib.error as e:
                 raise WazuhError(3001, str(e))
             except Exception as e:
-                result_logs['debug'][file].append(f"Exception raised: " + str(WazuhException(3001, str(e))))
+                result_logs['debug'][file].append("Exception raised: " + str(WazuhException(3001, str(e))))
                 update_cluster_control(file, cluster_control_json, exists=False)
 
         try: