Vulnerability Detector E2E Tests

deps/wazuh_testing/setup.py

@@ -32,6 +32,7 @@
  'data/all_disabled_ossec.conf',
  'data/syscollector_parsed_packages.json',
  'tools/migration_tool/delta_schema.json',
+ 'end_to_end/vulnerability_detector_packages/vuln_packages.json',
  'tools/migration_tool/CVE_JSON_5.0_bundled.json'
 ]
 

deps/wazuh_testing/wazuh_testing/end_to_end/__init__.py

@@ -11,6 +11,22 @@
 
 fetched_alerts_json_path = os.path.join(gettempdir(), 'alerts.json')
 
+base_path = {
+ 'linux': '/var/ossec',
+ 'windows': r'C:\Program Files (x86)\ossec-agent',
+ 'macos': '/Library/Ossec'
+}
+configuration_filepath_os = {
+ 'linux': os.path.join(base_path['linux'], 'etc', 'ossec.conf'),
+ 'windows': os.path.join(base_path['windows'], 'ossec.conf'),
+ 'macos': os.path.join(base_path['macos'], 'etc', 'ossec.conf')
+}
+logs_filepath_os = {
+ 'linux': os.path.join(base_path['linux'], 'logs', 'ossec.log'),
+ 'windows': os.path.join(base_path['windows'], 'ossec.log'),
+ 'macos': os.path.join(base_path['macos'], 'logs', 'ossec.log')
+}
+
 
 @retry(Exception, attempts=3, delay=5)
 def get_alert_indexer_api(query, credentials, ip_address, index='wazuh-alerts-4.x-*'):

deps/wazuh_testing/wazuh_testing/end_to_end/configuration.py

@@ -0,0 +1,204 @@
+"""
+Module for change configurations of remote hosts.
+----------------------------------------
+
+This module provides functions for configuring and managing remote host
+configurations using the HostManager class and related tools.
+
+Functions:
+ - backup_configurations: Backup configurations for all hosts in the specified host manager.
+ - restore_configuration: Restore configurations for all hosts in the specified host manager.
+ - configure_host: Configure a specific host.
+ - configure_environment: Configure the environment for all hosts in the specified host manager.
+
+
+Copyright (C) 2015, Wazuh Inc.
+Created by Wazuh, Inc. <[email protected]>.
+This program is a free software; you can redistribute it and/or modify it under the terms of GPLv2
+"""
+import xml.dom.minidom
+import logging
+
+from multiprocessing.pool import ThreadPool
+from typing import Dict, List
+
+from wazuh_testing.end_to_end import configuration_filepath_os
+from wazuh_testing.tools.configuration import set_section_wazuh_conf
+from wazuh_testing.tools.system import HostManager
+
+
+def backup_configurations(host_manager: HostManager) -> Dict[str, str]:
+ """
+ Backup configurations for all hosts in the specified host manager.
+
+ Args:
+ host_manager: An instance of the HostManager class containing information about hosts.
+
+ Returns:
+ dict: A dictionary mapping host names to their configurations.
+
+ Example of returned dictionary:
+ {
+ 'manager': '<ossec_config>...</ossec_config>',
+ 'agent1': ...
+ }
+ """
+ logging.info("Backing up configurations")
+ backup_configurations = {}
+ for host in host_manager.get_group_hosts('all'):
+ host_os_name = host_manager.get_host_variables(host)['os_name']
+ configuration_filepath = configuration_filepath_os[host_os_name]
+
+ backup_configurations[host] = host_manager.get_file_content(str(host),
+ configuration_filepath)
+ logging.info("Configurations backed up")
+ return backup_configurations
+
+
+def restore_configuration(host_manager: HostManager, configuration: Dict[str, List]) -> None:
+ """
+ Restore configurations for all hosts in the specified host manager.
+
+ Args:
+ host_manager: An instance of the HostManager class containing information about hosts.
+ configuration: A dictionary mapping host names to their configurations.
+
+ Example of configuration dictionary:
+ {
+ 'manager': '<ossec_config>...</ossec_config>',
+ 'agent1': ...
+ }
+ """
+ logging.info("Restoring configurations")
+ for host in host_manager.get_group_hosts('all'):
+ host_os_name = host_manager.get_host_variables(host)['os_name']
+ configuration_filepath = configuration_filepath_os[host_os_name]
+
+ host_manager.modify_file_content(host, configuration_filepath, configuration[host])
+ logging.info("Configurations restored")
+
+
+def configure_host(host: str, host_configuration: Dict[str, Dict], host_manager: HostManager) -> None:
+ """
+ Configure a specific host.
+
+ Args:
+ host: The name of the host to be configured.
+ host_configuration: Role of the configured host for the host. Check below for example.
+ host_manager: An instance of the HostManager class containing information about hosts.
+
+ Note: The host_configuration dictionary must contain a list of sections and elements to be configured. The sections
+ not included in the dictionary will not be modified maintaining the current configuration.
+
+
+ Example of host_configuration dictionary:
+ {
+ "manager1":[
+ {
+ "sections":[
+ {
+ "section":"vulnerability-detection",
+ "elements":[
+ {
+ "enabled":{
+ "value":"yes"
+ }
+ },
+ {
+ "index-status":{
+ "value":"yes"
+ }
+ },
+ {
+ "feed-update-interval":{
+ "value":"2h"
+ }
+ }
+ ]
+ },
+ ],
+ "metadata":{}
+ }
+ ],
+ }
+ """
+ logging.info(f"Configuring host {host}")
+
+ host_os = host_manager.get_host_variables(host)['os_name']
+ config_file_path = configuration_filepath_os[host_os]
+
+ host_config = host_configuration.get(host)
+
+ if not host_config:
+ raise TypeError(f"Host {host} configuration does not include a valid role (manager or agent):"
+ f"{host_configuration}")
+
+ current_config = host_manager.get_file_content(str(host), config_file_path)
+
+ # Extract the sections from the first element of host_config
+
+ sections = host_config[0].get('sections')
+
+ # Combine the current hos configuration and the desired configuration
+ new_config_unformatted = set_section_wazuh_conf(sections, current_config.split("\n"))
+
+ # Format new configuration
+ new_config_formatted_xml = xml.dom.minidom.parseString(''.join(new_config_unformatted))
+
+ # Get rid of the first no expected XML version line
+ new_config_formatted_xml = new_config_formatted_xml.toprettyxml().split("\n")[1:]
+
+ final_configuration = "\n".join(new_config_formatted_xml)
+
+ host_manager.modify_file_content(str(host), config_file_path, final_configuration)
+
+ logging.info(f"Host {host} configured")
+
+
+def configure_environment(host_manager: HostManager, configurations: Dict[str, List]) -> None:
+ """
+ Configure the environment for all hosts in the specified host manager.
+
+ Args:
+ host_manager: An instance of the HostManager class containing information about hosts.
+ configurations: A dictionary mapping host roles to their configuration details.
+
+ Example of host_configurations dictionary:
+ {
+ "manager1":[
+ {
+ "sections":[
+ {
+ "section":"vulnerability-detection",
+ "elements":[
+ {
+ "enabled":{
+ "value":"yes"
+ }
+ },
+ {
+ "index-status":{
+ "value":"yes"
+ }
+ },
+ {
+ "feed-update-interval":{
+ "value":"2h"
+ }
+ }
+ ]
+ },
+ ],
+ "metadata":{}
+ }
+ ],
+ }
+ """
+ logging.info("Configuring environment")
+ configure_environment_parallel_map = [(host, configurations) for host in host_manager.get_group_hosts('all')]
+
+ with ThreadPool() as pool:
+ pool.starmap(configure_host,
+ [(host, config, host_manager) for host, config in configure_environment_parallel_map])
+
+ logging.info("Environment configured")

deps/wazuh_testing/wazuh_testing/end_to_end/indexer_api.py

@@ -0,0 +1,83 @@
+"""
+Wazuh API Indexer Module.
+-----------------------------------
+
+This module provides functions to interact with the Wazuh Indexer API.
+
+Functions:
+ - get_indexer_values: Retrieves values from the Indexer API.
+
+Copyright (C) 2015, Wazuh Inc.
+Created by Wazuh, Inc. <[email protected]>.
+This program is a free software; you can redistribute it and/or modify it under the terms of GPLv2
+"""
+import requests
+import logging
+from typing import Dict
+
+from wazuh_testing.tools.system import HostManager
+
+
+STATE_INDEX_NAME = 'wazuh-vulnerabilities-states'
+
+
+def get_indexer_values(host_manager: HostManager, credentials: dict = {'user': 'admin', 'password': 'changeme'},
+ index: str = 'wazuh-alerts*', greater_than_timestamp=None) -> Dict:
+ """
+ Get values from the Wazuh Indexer API.
+
+ Args:
+ host_manager: An instance of the HostManager class containing information about hosts.
+ credentials (Optional): A dictionary containing the Indexer credentials. Defaults to
+ {'user': 'admin', 'password': 'changeme'}.
+ index (Optional): The Indexer index name. Defaults to 'wazuh-alerts*'.
+ greater_than_timestamp (Optional): The timestamp to filter the results. Defaults to None.
+
+ Returns:
+ Dict: A dictionary containing the values retrieved from the Indexer API.
+ """
+ logging.info(f"Getting values from the Indexer API for index {index}")
+
+ url = f"https://{host_manager.get_master_ip()}:9200/{index}/_search"
+ headers = {
+ 'Content-Type': 'application/json',
+ }
+
+ data = {
+ "query": {
+ "match_all": {}
+ }
+ }
+
+ if greater_than_timestamp:
+ query = {
+ "bool": {
+ "must": [
+ {"match_all": {}},
+ {"range": {"@timestamp": {"gte": f"{greater_than_timestamp}"}}}
+ ]
+ }
+ }
+
+ sort = [
+ {
+ "@timestamp": {
+ "order": "desc"
+ }
+ }
+ ]
+
+ data['query'] = query
+ data['sort'] = sort
+
+ param = {
+ 'pretty': 'true',
+ 'size': 10000,
+ }
+
+ response = requests.get(url=url, params=param, verify=False,
+ auth=requests.auth.HTTPBasicAuth(credentials['user'], credentials['password']),
+ headers=headers,
+ json=data)
+
+ return response.json()

deps/wazuh_testing/wazuh_testing/end_to_end/logs.py

@@ -0,0 +1,61 @@
+"""
+Logs management module for remote hosts.
+---------------------------------------
+
+Description:
+ This module provides functions for truncating logs and alerts for Wazuh agents and managers.
+
+Functions:
+ - truncate_remote_host_group_files: Truncate the specified files in all the host of a group
+ - get_hosts_logs: Get the logs from the specified host group
+
+
+Copyright (C) 2015, Wazuh Inc.
+Created by Wazuh, Inc. <[email protected]>.
+This program is a free software; you can redistribute it and/or modify it under the terms of GPLv2
+"""
+from typing import Dict
+
+from wazuh_testing import ALERTS_JSON_PATH
+from wazuh_testing.end_to_end import logs_filepath_os
+from wazuh_testing.tools.system import HostManager
+
+
+def truncate_remote_host_group_files(host_manager: HostManager, host_group: str,
+ file_to_truncate: str = 'logs') -> None:
+ """
+ Truncate log or alert files on remote hosts in a specified host group.
+
+ Parameters:
+ - host_manager (HostManager): An instance of the HostManager class for managing remote hosts.
+ - host_group (str): The name of the host group where the files will be truncated.
+ - file_to_truncate (str, optional): The type of file to truncate. Default is 'logs'.
+ Possible values are 'logs' for log files or 'alerts' for alert files.
+ """
+ for host in host_manager.get_group_hosts(host_group):
+ if file_to_truncate == 'logs':
+ host_os_name = host_manager.get_host_variables(host)['os_name']
+ log_file_path = logs_filepath_os[host_os_name]
+ elif file_to_truncate == 'alerts':
+ log_file_path = ALERTS_JSON_PATH
+ else:
+ log_file_path = file_to_truncate
+
+ host_manager.truncate_file(host, log_file_path)
+
+
+def get_hosts_logs(host_manager: HostManager, host_group: str = 'all') -> Dict[str, str]:
+ """
+ Get the logs from the specified host group.
+
+ Parameters:
+ - host_manager (HostManager): An instance of the HostManager class for managing remote hosts.
+ - host_group (str, optional): The name of the host group where the files will be truncated.
+ Default is 'all'.
+ """
+ host_logs = {}
+ for host in host_manager.get_group_hosts(host_group):
+ host_os_name = host_manager.get_host_variables(host)['os_name']
+ host_logs[host] = host_manager.get_file_content(host, logs_filepath_os[host_os_name])
+
+ return host_logs