Fix timeout and performance issues in Indexer E2E requests

deps/wazuh_testing/wazuh_testing/end_to_end/indexer_api.py

@@ -22,7 +22,7 @@
 
 
 def get_indexer_values(host_manager: HostManager, credentials: dict = {'user': 'admin', 'password': 'changeme'},
- index: str = 'wazuh-alerts*', greater_than_timestamp=None) -> Dict:
+ index: str = 'wazuh-alerts*', greater_than_timestamp=None, agent: str = '') -> Dict:
  """
  Get values from the Wazuh Indexer API.
 
@@ -49,26 +49,37 @@ def get_indexer_values(host_manager: HostManager, credentials: dict = {'user': '
  }
  }
 
- if greater_than_timestamp:
+ if greater_than_timestamp and agent:
+ query = {
+ "bool": {
+ "must": [
+ {"range": {"@timestamp": {"gte": f"{greater_than_timestamp}"}}},
+ {"match": {"agent.name": f"{agent}"}}
+ ]
+ }
+ }
+
+ data['query'] = query
+ elif greater_than_timestamp:
  query = {
  "bool": {
  "must": [
- {"match_all": {}},
  {"range": {"@timestamp": {"gte": f"{greater_than_timestamp}"}}}
  ]
  }
  }
 
- sort = [
- {
- "@timestamp": {
- "order": "desc"
+ data['query'] = query
+ elif agent:
+ query = {
+ "bool": {
+ "must": [
+ {"match": {"agent.name": f"{agent}"}}
+ ]
  }
- }
- ]
+ }
 
  data['query'] = query
- data['sort'] = sort
 
  param = {
  'pretty': 'true',

deps/wazuh_testing/wazuh_testing/end_to_end/monitoring.py

@@ -26,7 +26,8 @@
 from wazuh_testing.tools.system import HostManager
 
 
-def monitoring_events_multihost(host_manager: HostManager, monitoring_data: Dict, ignore_error: bool = False) -> Dict:
+def monitoring_events_multihost(host_manager: HostManager, monitoring_data: Dict, ignore_error: bool = True,
+ scan_interval: int = 5) -> Dict:
  """
  Monitor events on multiple hosts concurrently.
 
@@ -61,16 +62,17 @@ def monitoring_events_multihost(host_manager: HostManager, monitoring_data: Dict
  }
  }
  """
- def monitoring_event(host_manager: HostManager, host: str, monitoring_elements: List[Dict], scan_interval: int = 20,
- ignore_error: bool = False) -> Dict:
+ def monitoring_event(host_manager: HostManager, host: str, monitoring_elements: List[Dict],
+ ignore_error: bool = True,
+ scan_interval: int = 5) -> Dict:
  """
  Monitor the specified elements on a host.
 
  Args:
  host_manager (HostManager): Host Manager to handle the environment
  host (str): The target host.
  monitoring_elements(List): A list of dictionaries containing regex, timeout, and file.
- ignore_error: If True, ignore errors and continue monitoring.
+ ignore_error: If True, ignore TimeoutError and return the result.
 
  Raises:
  TimeoutError: If no match is found within the specified timeout.
@@ -95,10 +97,15 @@ def filter_events_by_timestamp(match_events: List) -> List:
  timestamp_format = "%Y/%m/%d %H:%M:%S"
  timestamp_format_parameter = "%Y-%m-%dT%H:%M:%S.%f"
 
- timestamp_datetime = datetime.strptime(timestamp_str, timestamp_format)
- greater_than_timestamp_formatted = datetime.strptime(greater_than_timestamp, timestamp_format_parameter)
+ try:
+ timestamp_datetime = datetime.strptime(timestamp_str, timestamp_format)
+ greater_than_timestamp_formatted = datetime.strptime(greater_than_timestamp,
+ timestamp_format_parameter)
+ except ValueError:
+ raise ValueError(f"Timestamp format not supported: {timestamp_str}."
+ 'Do the regex includes the timestamp?')
 
- if timestamp_datetime >=  greater_than_timestamp_formatted:
+ if timestamp_datetime >= greater_than_timestamp_formatted:
  match_that_fit_timestamp.append(match)
 
  return match_that_fit_timestamp
@@ -112,11 +119,12 @@ def filter_events_by_timestamp(match_events: List) -> List:
  element['n_iterations'], \
  element.get('greater_than_timestamp', None)
  current_timeout = 0
- regex_match = None
+ regex_match = False
 
  while current_timeout < timeout:
  file_content = host_manager.get_file_content(host, monitoring_file)
  match_regex = re.findall(regex, file_content)
+
  if greater_than_timestamp:
  match_that_fit_timestamp = filter_events_by_timestamp(match_regex)
  else:
@@ -129,7 +137,7 @@ def filter_events_by_timestamp(match_events: List) -> List:
 
  sleep(scan_interval)
 
- current_timeout += scan_interval
+ current_timeout = current_timeout + scan_interval
 
  if not regex_match:
  elements_not_found.append(element)
@@ -150,15 +158,12 @@ def filter_events_by_timestamp(match_events: List) -> List:
  with ThreadPoolExecutor() as executor:
  futures = []
  for host, data in monitoring_data.items():
- futures.append(executor.submit(monitoring_event, host_manager, host, data, ignore_error))
+ futures.append(executor.submit(monitoring_event, host_manager, host, data, ignore_error, scan_interval))
 
  results = {}
  for future in as_completed(futures):
- try:
- result = future.result()
- results.update(result)
- except Exception as e:
- logging.error(f"An error occurred: {e}")
+ result = future.result()
+ results.update(result)
 
  logging.info(f"Monitoring results: {results}")
 

deps/wazuh_testing/wazuh_testing/end_to_end/vulnerability_detector.py

@@ -131,6 +131,34 @@ def get_alerts_by_agent(alerts, regex) -> Dict:
  return alerts_vuln_by_agent
 
 
+def parse_vulnerability_detector_alerts(alerts) -> Dict:
+ """
+ Get specific alerts by agent.
+
+ Args:
+ alerts (list): List of alerts.
+ regex (str): Regular expression to match the alerts.
+
+ Returns:
+ dict: Dictionary containing the alerts by agent.
+ """
+ vulnerability_detector_alerts = {}
+ vulnerability_detector_alerts['affected'] = []
+ vulnerability_detector_alerts['mitigated'] = []
+
+ regex_affected = 'CVE.*? affects.*"?'
+ rege_mitigated = "The .* that affected .* was solved due to a package removal"
+
+ # Parse affected vuln alerts
+ for alert in alerts:
+ if re.match(regex_affected, alert['_source']['rule']['description']):
+ vulnerability_detector_alerts['affected'].append(alert)
+ elif re.match(rege_mitigated, alert['_source']['rule']['description']):
+ vulnerability_detector_alerts['mitigated'].append(alert)
+
+ return vulnerability_detector_alerts
+
+
 def get_indexed_vulnerabilities_by_agent(indexed_vulnerabilities) -> Dict:
  """Get indexed vulnerabilities by agent.