add: Add EPS limit test #2947

deps/wazuh_testing/wazuh_testing/modules/eps/__init__.py

@@ -0,0 +1,17 @@
+import os
+import json
+from datetime import datetime, timedelta
+from copy import deepcopy
+
+from wazuh_testing.tools.time import parse_date_time_format
+
+
+# Timeouts
+T_5 = 5
+T_10 = 10
+T_15 = 15
+T_20 = 20
+T_60 = 60
+
+ANALYSISD_PREFIX = r'.*wazuh-analysisd*'
+MAILD_PREFIX = r'.*wazuh-maild*'

deps/wazuh_testing/wazuh_testing/modules/eps/event_monitor.py

@@ -0,0 +1,109 @@
+import re
+from datetime import datetime
+
+from wazuh_testing.modules import eps as eps
+from wazuh_testing.tools import LOG_FILE_PATH, ANALYSISD_STATE, ALERT_LOGS_PATH
+from wazuh_testing.tools.monitoring import FileMonitor
+
+
+def make_analysisd_callback(pattern, prefix=eps.ANALYSISD_PREFIX):
+ """Create a callback function from a text pattern.
+
+ It already contains the vulnerability-detector prefix.
+
+ Args:
+ pattern (str): String to match on the log.
+ prefix (str): regular expression used as prefix before the pattern.
+
+ Returns:
+ lambda: function that returns if there's a match in the file
+
+ Examples:
+ >>> callback_bionic_update_started = make_vuln_callback("Starting Ubuntu Bionic database update")
+ """
+ pattern = r'\s+'.join(pattern.split())
+ regex = re.compile(r'{}{}'.format(prefix, pattern))
+
+ return lambda line: regex.match(line) is not None
+
+
+def check_analysisd_event(file_monitor=None, callback='', error_message=None, update_position=True,
+ timeout=eps.T_60, prefix=eps.ANALYSISD_PREFIX, accum_results=1,
+ file_to_monitor=LOG_FILE_PATH):
+ """Check if a vulnerability event occurs
+
+ Args:
+ file_monitor (FileMonitor): FileMonitor object to monitor the file content.
+ callback (str): log regex to check in Wazuh log
+ error_message (str): error message to show in case of expected event does not occur
+ update_position (boolean): filter configuration parameter to search in Wazuh log
+ timeout (str): timeout to check the event in Wazuh log
+ prefix (str): log pattern regex
+ accum_results (int): Accumulation of matches.
+ """
+ file_monitor = FileMonitor(file_to_monitor) if file_monitor is None else file_monitor
+ error_message = f"Could not find this event in {file_to_monitor}: {callback}" if error_message is None else \
+ error_message
+
+ file_monitor.start(timeout=timeout, update_position=update_position, accum_results=accum_results,
+ callback=make_analysisd_callback(callback, prefix), error_message=error_message)
+
+
+def check_eps_disabled():
+ """Check if the eps module is disabled"""
+ check_analysisd_event(callback=fr'.*INFO: EPS limit disabled.*', timeout=eps.T_10)
+
+
+def check_eps_enabled(maximun, timeframe):
+ """Check if the eps module is enable"""
+ check_analysisd_event(callback=fr".*INFO: EPS limit enabled, EPS: '{maximun}', timeframe: '{timeframe}'",
+ timeout=eps.T_10)
+
+
+def check_configuration_error():
+ """Check the configuration error event in ossec.log"""
+ check_analysisd_event(timeout=eps.T_60, callback=r".* \(\d+\): Configuration error at.*",
+ error_message="Could not find the event 'Configuration error at 'etc/ossec.conf' "
+ 'in ossec.log', prefix=eps.MAILD_PREFIX)
+
+
+def get_words_from_file(words, filename):
+ """Get the words from file
+
+ Args:
+ wordss (str): Word to find in the file
+ Returns:
+ str: Line that match in file
+ """
+ with open(filename, 'r') as file:
+ for index, line in enumerate(file):
+ # search string
+ if words in line:
+ return line
+
+
+def get_analysisd_state(word):
+ """Get the value of word in wazuh-analysisd.state
+
+ Args:
+ word (str): Word to find in the file
+ """
+ line = get_words_from_file(word, ANALYSISD_STATE)
+ return float(line.split("\'")[1::2][0])
+
+
+def get_alert_timestamp(start_log, end_log):
+ """Get the timestamp of the alert if exist in the alerts.log file between two string
+
+ Args:
+ start_log (str): Start message to find
+ end_log (str): End message to find
+ """
+ with open(ALERT_LOGS_PATH, 'r') as file:
+ str_file = file.read()
+ index1 = str_file.find(end_log)
+ index2 = str_file[0: index1].rfind(start_log)
+ str_alert = str_file[index2: index1]
+ timestamp = str_alert[str_alert.find(start_log) + len(start_log):str_alert.find(': ')]
+
+ return datetime.fromtimestamp(float(timestamp)).strftime('%Y-%m-%d %H:%M:%S')

deps/wazuh_testing/wazuh_testing/tools/__init__.py

@@ -77,6 +77,9 @@
  else:
  HOSTS_FILE_PATH = os.path.join('/', 'etc', 'hosts')
  GLOBAL_DB_PATH = os.path.join(WAZUH_PATH, 'queue', 'db', 'global.db')
+ ANALYSISD_STATE = os.path.join(WAZUH_PATH, 'var', 'run', 'wazuh-analysisd.state')
+ SIMULATE_AGENT = os.path.join('deps','wazuh_testing','wazuh_testing','scripts','simulate_agents.py')
+ WAZUH_INTERNAL_OPTIONS = os.path.join(WAZUH_PATH, 'etc', 'internal_options.conf')
 
  try:
  import grp
@@ -126,6 +129,7 @@ def get_service():
 CLIENT_CUSTOM_CERT_PATH = os.path.join(_data_path, 'sslmanager.cert')
 
 WAZUH_LOGS_PATH = os.path.join(WAZUH_PATH, 'logs')
+ALERT_PATH = os.path.join(WAZUH_LOGS_PATH, 'alerts')
 ALERT_FILE_PATH = os.path.join(WAZUH_LOGS_PATH, 'alerts', 'alerts.json')
 ALERT_LOGS_PATH = os.path.join(WAZUH_LOGS_PATH, 'alerts', 'alerts.log')
 CLUSTER_LOGS_PATH = os.path.join(WAZUH_LOGS_PATH, 'cluster.log')

deps/wazuh_testing/wazuh_testing/tools/agent_simulator.py

@@ -1017,6 +1017,7 @@ class Logcollector:
  def __init__(self):
  self.logcollector_tag = 'syslog'
  self.logcollector_mq = 'x'
+ self.message_counter = 0
 
  def generate_event(self):
  """Generate logcollector event
@@ -1026,8 +1027,10 @@ def generate_event(self):
  """
  log = 'Mar 24 10:12:36 centos8 sshd[12249]: Invalid user random_user from 172.17.1.1 port 56550'
 
- message = f"{self.logcollector_mq}:{self.logcollector_tag}:{log}"
+ message_counter_info = f"Message number: {self.message_counter}"
+ message = f"{self.logcollector_mq}:{self.logcollector_tag}:{log}:{message_counter_info}"
 
+ self.message_counter = self.message_counter + 1
  return message
 
 

deps/wazuh_testing/wazuh_testing/tools/configuration.py

@@ -742,3 +742,21 @@ def update_configuration_template(configurations, old_values, new_values):
  configurations_to_update = configurations_to_update.replace(old_value, new_value)
 
  return json.loads(configurations_to_update)
+
+
+def get_simulate_agent_configuration(data_file_path):
+ """Load simulate agent configuration file.
+
+ Args:
+ data_file_path (str): Test case template file path.
+
+ Returns:
+ dict: Configurations names.
+ """
+ configuration_file = file.read_yaml(data_file_path)
+ configuration_parameters = {}
+
+ for test_case in configuration_file:
+ configuration_parameters.update(test_case['configuration_parameters'])
+
+ return configuration_parameters

requirements.txt

@@ -21,14 +21,16 @@ py~=1.10.0
 pycryptodome>=3.9.8
 pyOpenSSL==19.1.0
 pytest-html==3.1.1
-pytest==6.2.5
+pytest==6.2.2 ; python_version <= "3.9"
+pytest==7.1.2 ; python_version >= "3.10"
 pyyaml==5.4
 requests==2.23.0
 scipy>=1.0; platform_system == "Linux" or platform_system == "Darwin" or platform_system=='Windows'
 seaborn>=0.11.1; platform_system == "Linux" or platform_system == "Darwin" or platform_system=='Windows'
 setuptools~=56.0.0
 testinfra==5.0.0
-jq>=1.1.2; platform_system == "Linux" or platform_system == "Darwin"
+jq==1.1.2 ; (platform_system == "Linux" or platform_system == "Darwin") and python_version <= "3.9"
+jq==1.2.2 ; python_version >= "3.10"
 cryptography==3.3.2; platform_system == "Linux" or platform_system == "Darwin" or platform_system=='Windows'
 urllib3
 numpydoc>=1.1.0