Add localfile wildcard support for windows IT

CHANGELOG.md

@@ -7,6 +7,10 @@ All notable changes to this project will be documented in this file.
 Wazuh commit: TBD \
 Release report: TBD
 
+## Added
+
+- Add Windows location wildcards tests ([#4263](https:/wazuh/wazuh-qa/pull/4263)) \- (Tests + Framework)
+
 ## [4.5.0] - TBD
 
 Wazuh commit: TBD \

deps/wazuh_testing/wazuh_testing/modules/logcollector/event_monitor.py

@@ -132,3 +132,40 @@ def check_ignore_restrict_message_not_found(message, regex, tag, prefix):
  with pytest.raises(TimeoutError):
  log_found = check_ignore_restrict_message(message=message, regex=regex, tag=tag, prefix=prefix)
  assert log_found is False, ERR_MSG_UNEXPECTED_IGNORE_EVENT
+
+
+def check_wildcard_pattern_expanded(file_path, location_regex, prefix, error_message=None, file_monitor=None,
+ timeout=T_10, escape=False):
+ """Create a callback to detect "New file that matches the '{file_path}' pattern: '(.*)'" line.
+ Args:
+ file_path (str): file path that is being monitored
+ location_regex (str): path configured in location tag
+ prefix (str): Daemon that generates the error log.
+ error_message (str): Error message.
+ file_monitor (FileMonitor): Log monitor.
+ timeout (int): Timeout to check the log.
+ escape (bool): Flag to escape special characters in the pattern.
+ Returns: True if the expected message has been found, False otherwise.
+ """
+ callback_msg = f".*New file that matches the '{location_regex}' pattern: '{file_path}'"
+
+ return check_logcollector_event(file_monitor=file_monitor, timeout=timeout, callback=callback_msg,
+ error_message=error_message, prefix=prefix, escape=escape)
+
+
+def check_win_wildcard_pattern_no_match(regex, prefix, error_message=None, file_monitor=None, timeout=T_10,
+ escape=False):
+ """Create a callback to detect "DEBUG: No file/folder that matches ..." line.
+ Args:
+ regex (str): regex pattern configured in location tag for monitoring
+ prefix (str): Daemon that generates the error log.
+ error_message (str): Error message.
+ file_monitor (FileMonitor): Log monitor.
+ timeout (int): Timeout to check the log.
+ escape (bool): Flag to escape special characters in the pattern.
+ Returns: True if the expected message has been found, False otherwise.
+ """
+ callback_msg = f".*expand_win32_wildcards.*DEBUG: No .* that matches {regex}"
+
+ return check_logcollector_event(file_monitor=file_monitor, timeout=timeout, callback=callback_msg,
+ error_message=error_message, prefix=prefix, escape=escape)

deps/wazuh_testing/wazuh_testing/tools/file.py

@@ -494,10 +494,10 @@ def recursive_directory_creation(path):
  if parent != '' and not os.path.exists(parent):
  split = os.path.split(parent)
  recursive_directory_creation(split[0])
- os.mkdir(parent)
+ os.mkdir(parent, mode=0o0777)
 
  if not os.path.exists(path):
- os.mkdir(path)
+ os.mkdir(path, mode=0o0777)
 
 
 def move_everything_from_one_directory_to_another(source_directory, destination_directory):

tests/integration/conftest.py

@@ -21,8 +21,8 @@
 from wazuh_testing.logcollector import create_file_structure, delete_file_structure
 from wazuh_testing.tools import (PREFIX, LOG_FILE_PATH, WAZUH_CONF, get_service, ALERT_FILE_PATH,
  WAZUH_LOCAL_INTERNAL_OPTIONS)
-from wazuh_testing.tools.configuration import get_minimal_configuration, get_wazuh_conf, write_wazuh_conf
-from wazuh_testing.tools.file import copy, recursive_directory_creation, remove_file, truncate_file, write_file
+from wazuh_testing.tools.file import (truncate_file, recursive_directory_creation, remove_file, copy, write_file,
+ delete_path_recursively)
 from wazuh_testing.tools.monitoring import FileMonitor, QueueMonitor, SocketController, close_sockets
 from wazuh_testing.tools.services import check_daemon_status, control_service, delete_dbs
 from wazuh_testing.tools.time import TimeMachine
@@ -1248,6 +1248,22 @@ def copy_file(source_path, destination_path):
  remove_file(file)
 
 
+@pytest.fixture()
+def create_files_in_folder(folder_path, file_list):
+ """Create a list of files, inside a given path. Deletes it at the end.
+ Args:
+ folder_path (str): folder path to create.
+ file_list (List): list of file names to create
+ """
+ recursive_directory_creation(folder_path)
+ for file in file_list:
+ write_file(os.path.join(folder_path, file))
+
+ yield
+
+ delete_path_recursively(folder_path)
+
+
 @pytest.fixture(scope='function')
 def create_file(new_file_path):
  """Create an empty file.
@@ -1266,18 +1282,18 @@ def create_file(new_file_path):
 def load_wazuh_basic_configuration():
  """Load a new basic configuration to the manager"""
  # Load ossec.conf with all disabled settings
- minimal_configuration = get_minimal_configuration()
+ minimal_configuration = conf.get_minimal_configuration()
 
  # Make a backup from current configuration
- backup_ossec_configuration = get_wazuh_conf()
+ backup_ossec_configuration = conf.get_wazuh_conf()
 
  # Write new configuration
- write_wazuh_conf(minimal_configuration)
+ conf.write_wazuh_conf(minimal_configuration)
 
  yield
 
  # Restore the ossec.conf backup
- write_wazuh_conf(backup_ossec_configuration)
+ conf.write_wazuh_conf(backup_ossec_configuration)
 
 
 @pytest.fixture(scope='function')

tests/integration/test_logcollector/test_configuration/test_basic_configuration_ignore_binaries.py

@@ -64,7 +64,9 @@
 from wazuh_testing.tools.services import control_service
 from wazuh_testing.tools.file import truncate_file
 import wazuh_testing.api as api
-from wazuh_testing.tools.monitoring import LOG_COLLECTOR_DETECTOR_PREFIX, WINDOWS_AGENT_DETECTOR_PREFIX, FileMonitor
+from wazuh_testing.tools.monitoring import FileMonitor
+from wazuh_testing.modules.logcollector import LOG_COLLECTOR_PREFIX, WINDOWS_AGENT_PREFIX
+from wazuh_testing.modules.logcollector.event_monitor import check_win_wildcard_pattern_no_match
 
 import subprocess as sb
 
@@ -145,9 +147,7 @@ def check_ignore_binaries_valid(cfg):
  wazuh_log_monitor = FileMonitor(LOG_FILE_PATH)
 
  if sys.platform == 'win32':
- log_callback = logcollector.callback_invalid_location_pattern(cfg['location'])
- wazuh_log_monitor.start(timeout=5, callback=log_callback,
- error_message=logcollector.GENERIC_CALLBACK_ERROR_INVALID_LOCATION)
+ check_win_wildcard_pattern_no_match(re.escape(cfg['location']), WINDOWS_AGENT_PREFIX, escape=False)
 
  if wazuh_component == 'wazuh-manager':
  real_configuration = cfg.copy()
@@ -167,22 +167,24 @@ def check_ignore_binaries_invalid(cfg):
  """
  wazuh_log_monitor = FileMonitor(LOG_FILE_PATH)
 
- log_callback = gc.callback_invalid_value('ignore_binaries', cfg['ignore_binaries'], prefix)
+ log_callback = gc.callback_invalid_value('ignore_binaries', cfg['ignore_binaries'], LOG_COLLECTOR_PREFIX)
  wazuh_log_monitor.start(timeout=5, callback=log_callback,
  error_message=gc.GENERIC_CALLBACK_ERROR_MESSAGE)
 
- log_callback = gc.callback_error_in_configuration('ERROR', prefix,
+ log_callback = gc.callback_error_in_configuration('ERROR', LOG_COLLECTOR_PREFIX,
  conf_path=f'{wazuh_configuration}')
  wazuh_log_monitor.start(timeout=5, callback=log_callback,
  error_message=gc.GENERIC_CALLBACK_ERROR_MESSAGE)
 
  if sys.platform != 'win32':
- log_callback = gc.callback_error_in_configuration('CRITICAL', prefix,
+ log_callback = gc.callback_error_in_configuration('CRITICAL', LOG_COLLECTOR_PREFIX,
  conf_path=f'{wazuh_configuration}')
  wazuh_log_monitor.start(timeout=5, callback=log_callback,
  error_message=gc.GENERIC_CALLBACK_ERROR_MESSAGE)
 
 
+# Test
+@pytest.mark.xfail(sys.platform == 'win32', reason="Flaky behavior in Windows agent. Blocked by Issue #4122")
 @pytest.mark.filterwarnings('ignore::urllib3.exceptions.InsecureRequestWarning')
 def test_ignore_binaries(get_configuration, configure_environment):
  '''

tests/integration/test_logcollector/test_location/data/configuration/configuration_win_location_wildcards.yaml

@@ -0,0 +1,29 @@
+- sections:
+ - section: localfile
+ elements:
+ - log_format:
+ value: syslog
+ - location:
+ value: LOCATION
+
+ - section: sca
+ elements:
+ - enabled:
+ value: 'no'
+
+ - section: rootcheck
+ elements:
+ - disabled:
+ value: 'yes'
+
+ - section: syscheck
+ elements:
+ - disabled:
+ value: 'yes'
+
+ - section: wodle
+ attributes:
+ - name: syscollector
+ elements:
+ - disabled:
+ value: 'yes'

tests/integration/test_logcollector/test_location/data/test_cases/cases_win_location_wildcards.yaml

@@ -0,0 +1,47 @@
+- name: test_single_asterisk_wildcard
+ description: Test location tag with one asterisk wildcard
+ configuration_parameters:
+ LOCATION: c:\testfol*\subfolder\test
+ metadata:
+ matches: true
+ location: c:\testfol*\subfolder\test
+
+- name: test_single_question_mark_wildcard
+ description: Test location tag with one question mark wildcard
+ configuration_parameters:
+ LOCATION: c:\testfolde?\subfolder\test
+ metadata:
+ location: c:\testfolde?\subfolder\test
+ matches: true
+
+- name: test_partial_words_with_asterisk_wildcards
+ description: Test location tag with words completed using asterisk wildcard
+ configuration_parameters:
+ LOCATION: c:\test*\sub*\t*
+ metadata:
+ location: c:\test*\sub*\t*
+ matches: true
+
+- name: test_mixed_wildcards
+ description: Test location tag with mixed asterisk and question mark wildcards
+ configuration_parameters:
+ LOCATION: c:\testf?lder\*\tes?
+ metadata:
+ location: c:\testf?lder\*\tes?
+ matches: true
+
+- name: test_mixed_wildcards_path_asterisk_only
+ description: Test location tag were filepath uses asterisk only
+ configuration_parameters:
+ LOCATION: c:\*\*\?es?
+ metadata:
+ location: c:\*\*\?es?
+ matches: true
+
+- name: test_invalid_wildcards_no_match
+ description: Test location tag where wildcards do not match a valid file
+ configuration_parameters:
+ LOCATION: c:\testfolder\subfolder\tes?.log
+ metadata:
+ matches: false
+ location: c:\testfolder\subfolder\tes?.log

tests/integration/test_logcollector/test_location/test_location.py

@@ -69,8 +69,8 @@
 pytestmark = pytest.mark.tier(level=0)
 
 # Configuration
-test_data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data', 'configuration')
-configurations_path = os.path.join(test_data_path, 'wazuh_location.yaml')
+test_data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data', 'configuration_template')
+configurations_path = os.path.join(test_data_path, 'configuration_location.yaml')
 local_internal_options = {'logcollector.debug': '2'}
 
 temp_dir = tempfile.gettempdir()
@@ -216,8 +216,9 @@ def location_file_date():
  file_structure = ast.literal_eval(file_structure_string_real_paths)
 
 
-def test_location(location_file_date, get_files_list, create_file_structure_module, get_configuration, configure_environment,
- configure_local_internal_options_module, file_monitoring, restart_logcollector):
+def test_location(location_file_date, get_files_list, create_file_structure_module, get_configuration,
+ configure_environment, configure_local_internal_options_module, file_monitoring,
+ restart_logcollector):
  '''
  description: Check if the 'wazuh-logcollector' monitors the log files specified in the 'location' tag.
  For this purpose, the test will create a testing log file, configure a 'localfile' section
@@ -281,7 +282,7 @@ def test_location(location_file_date, get_files_list, create_file_structure_modu
  if file_type == 'single_file':
  log_callback = logcollector.callback_analyzing_file(file_location)
  log_monitor.start(timeout=logcollector.LOG_COLLECTOR_GLOBAL_TIMEOUT, callback=log_callback,
- error_message=f"The expected 'Analyzing file {file_location}' message has not been produced")
+ error_message=f"The expected 'Analyzing file {file_location}' message was not found")
  elif file_type == 'wildcard_file':
  pattern = get_configuration['metadata']['location']
  log_callback = logcollector.callback_match_pattern_file(pattern, file_location)

tests/integration/test_logcollector/test_location/test_location_exclude.py

@@ -67,8 +67,8 @@
 pytestmark = pytest.mark.tier(level=0)
 
 # Configuration
-test_data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data', 'configuration')
-configurations_path = os.path.join(test_data_path, 'wazuh_location.yaml')
+test_data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data', 'configuration_template')
+configurations_path = os.path.join(test_data_path, 'configuration_location.yaml')
 
 temp_dir = tempfile.gettempdir()