Adding methods for rules and decoders

src/wazuh_qa_framework/global_variables/daemons.py

@@ -18,7 +18,7 @@
 WAZUH_DB_DAEMON = 'wazuh-db'
 AGENT_DAEMON = 'wazuh-agentd'
 
-WAZUH_ANGENT_WINDOWS_SERVICE_NAME = 'WazuhSvc'
+WAZUH_AGENT_WINDOWS_SERVICE_NAME = 'WazuhSvc'
 
 
 WAZUH_AGENT_DAEMONS = [LOGCOLLECTOR_DAEMON, SYSCHECK_DAEMON, EXEC_DAEMON, MODULES_DAEMON, AGENT_DAEMON]

src/wazuh_qa_framework/system/wazuh_handler.py

@@ -7,7 +7,7 @@
 from multiprocessing.pool import ThreadPool
 
 from wazuh_qa_framework.generic_modules.logging.base_logger import BaseLogger
-from wazuh_qa_framework.global_variables.daemons import WAZUH_ANGENT_WINDOWS_SERVICE_NAME
+from wazuh_qa_framework.global_variables.daemons import WAZUH_AGENT_WINDOWS_SERVICE_NAME
 from wazuh_qa_framework.system.host_manager import HostManager
 
 
@@ -518,7 +518,7 @@ def restart_agent(self, host):
  host (str): Hostname
  """
  self.logger.debug(f'Restarting agent {host}')
- service_name = WAZUH_ANGENT_WINDOWS_SERVICE_NAME if self.is_windows(host) else 'wazuh-agent'
+ service_name = WAZUH_AGENT_WINDOWS_SERVICE_NAME if self.is_windows(host) else 'wazuh-agent'
  if self.is_agent(host):
  self.control_service(host, service_name, 'restarted')
  self.logger.debug(f'Agent {host} restarted successfully')
@@ -575,7 +575,7 @@ def stop_agent(self, host):
  host (str): Hostname
  """
  self.logger.debug(f'Stopping agent {host}')
- service_name = WAZUH_ANGENT_WINDOWS_SERVICE_NAME if is_windows(host) else 'wazuh-agent'
+ service_name = WAZUH_AGENT_WINDOWS_SERVICE_NAME if self.is_windows(host) else 'wazuh-agent'
  if self.is_agent(host):
  self.control_service(host, service_name, 'stopped')
  self.logger.debug(f'Agent {host} stopped successfully')
@@ -632,7 +632,7 @@ def start_agent(self, host):
  host (str): Hostname
  """
  self.logger.debug(f'Starting agent {host}')
- service_name = WAZUH_ANGENT_WINDOWS_SERVICE_NAME if is_windows(host) else 'wazuh-agent'
+ service_name = WAZUH_AGENT_WINDOWS_SERVICE_NAME if self.is_windows(host) else 'wazuh-agent'
  if self.is_agent(host):
  self.control_service(host, service_name, 'started')
  self.logger.debug(f'Agent {host} started successfully')
@@ -727,11 +727,11 @@ def stop_environment(self, parallel=True):
  self.pool.map(self.stop_agent, agent_list)
  else:
  self.logger.info(message='Stopping environment: Managers')
- for manager in get_managers():
+ for manager in manager_list:
  self.stop_manager(manager)
 
  self.logger.info(message='Stopping environment: Agents')
- for agent in get_agents():
+ for agent in agent_list:
  self.stop_agent(agent)
 
  self.logger.info('Stopping environment')
@@ -754,11 +754,11 @@ def start_environment(self, parallel=True):
  self.pool.map(self.start_agent, agent_list)
  else:
  self.logger.info(message='Starting environment: Managers')
- for manager in get_managers():
+ for manager in manager_list:
  self.start_manager(manager)
 
  self.logger.info(message='Starting environment: Agents')
- for agent in get_agents():
+ for agent in agent_list:
  self.start_agent(agent)
 
  self.logger.info('Environment started successfully')
@@ -843,3 +843,88 @@ def is_manager(self, host):
  bool: True if host is manager
  """
  return host in self.get_managers()
+
+ def create_rule(self, host, new_rules_filepath, rules_filename, overwrite=True):
+ """Create new rules replaces an existing rule file or adds rules to an existing file from a file.
+
+ Args:
+ host (str): Host name.
+ new_rules_filepath (str): New rules filepath.
+ rules_filename (str): Host rules filename.
+ overwrite (bool): Replace the rules. Defaults to True.
+ """
+ try:
+ with open(new_rules_filepath, 'r') as file:
+ new_rules = file.read()
+
+ files_information = self.find_files(host, os.path.dirname(rules_filename), '*')
+ file_exist = False
+ for file_information in files_information:
+ if rules_filename in file_information['path']:
+ file_exist = True
+ break
+
+ if file_exist:
+ if overwrite:
+ self.logger.info(message=f'Changing {rules_filename} to {new_rules_filepath}')
+ else:
+ current_rules = self.get_file_content(host, rules_filename)
+ index_rule = current_rules.rfind("</rule>")
+ if index_rule != -1:
+ new_rules = current_rules[:index_rule] + '</rule>\n' + new_rules + '\n</group>'
+ self.logger.info(message=f'Adding rule from {new_rules_filepath} to {rules_filename}')
+
+ self.modify_file_content(host, rules_filename, new_rules)
+ self.logger.info(message=f'Rules succefully updated')
+
+ else:
+ if overwrite:
+ self.logger.info(message=f'{rules_filename} does not exist to be overwritten')
+ else:
+ self.create_file(host, rules_filename, f"<group>\n{new_rules}\n</group>")
+ self.logger.info(message=f'Rules succefully added into a new rule file named {rules_filename}')
+
+ except FileNotFoundError:
+ self.logger.error(message=f'The file {new_rules_filepath} does not exist.')
+
+ def create_decoder(self, host, new_decoder_filepath, decoder_filename, overwrite=True):
+ """Create new decoder replaces an existing decoder file or adds decoders to an existing file from a file.
+
+ Args:
+ host (str): Host name.
+ new_decoder_filepath (str): New decoder filepath.
+ decoder_filename (str): Host decoder filename.
+ overwrite (bool): Replace the decoder file. Defaults to True.
+ """
+ try:
+ with open(new_decoder_filepath, 'r') as file:
+ new_decoders = file.read()
+
+ files_information = self.find_files(host, os.path.dirname(decoder_filename), '*')
+ file_exist = False
+ for file_information in files_information:
+ if decoder_filename in file_information['path']:
+ file_exist = True
+ break
+
+ if file_exist:
+ if overwrite:
+ self.logger.info(message=f'Changing {decoder_filename} to {new_decoder_filepath}')
+ else:
+ current_decoders = self.get_file_content(host, decoder_filename)
+ index_decoder = current_decoders.rfind("</decoder>")
+ if index_decoder != -1:
+ new_decoders = current_decoders[:index_decoder] + '</decoder>\n' + new_decoders
+ self.logger.info(message=f'Adding decoder from {new_decoder_filepath} to {decoder_filename}')
+ self.modify_file_content(host, decoder_filename, new_decoders)
+ self.logger.info(message=f'Decoders succefully updated')
+
+ else:
+ if overwrite:
+ self.logger.info(message=f'{decoder_filename} does not exist to be overwritten')
+ else:
+ self.create_file(host, decoder_filename, new_decoders)
+ self.logger.info(message=f'Decoders succefully added into a new file named {decoder_filename}')
+
+ except FileNotFoundError:
+ self.logger.error(message=f'The file {new_decoder_filepath} does not exist.')