feat(#2947): add test limitation

deps/wazuh_testing/wazuh_testing/scripts/syslog_simulator.py

@@ -1,36 +1,111 @@
+import socket
 import argparse
+import sys
+import logging
+import time
 
-from wazuh_testing.tools.syslog_simulator import Syslogger
 
+TCP = 'tcp'
+UDP = 'udp'
+DEFAULT_MESSAGE = 'Login failed: admin, test'
+LOGGER = logging.getLogger('syslog_simulator')
+TCP_LIMIT = 1000
+UDP_LIMIT = 200
 
-def main():
+
+def set_logging(debug=False):
+ LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)
+ handler = logging.StreamHandler(sys.stdout)
+ handler.setFormatter(logging.Formatter("%(asctime)s — %(levelname)s — %(message)s"))
+ LOGGER.addHandler(handler)
+
+
+def validate_parameters(parameters):
+ protocol_limit = TCP_LIMIT if parameters.protocol == TCP else UDP_LIMIT
+
+ if parameters.messages_number <= 0:
+ LOGGER.error(f"The number of messages parameter has to be greater than 0")
+ return sys.exit(1)
+
+ if parameters.eps > 0 and parameters.eps > protocol_limit:
+ LOGGER.error(f"You can't select eps greather than {protocol_limit}")
+ return sys.exit(1)
+
+
+def get_parameters():
  arg_parser = argparse.ArgumentParser()
 
- arg_parser.add_argument('-m', '--message', metavar='<message>', type=str, required=True,
- default='Login failed: admin, test message, Message number:',
- help="The syslog message", dest='message')
+ arg_parser.add_argument('-n', '--messages-number', metavar='<messages_number>', type=int,
+ help='Number of messages to send', required=True, default=0,
+ dest='messages_number')
+
+ arg_parser.add_argument('-m', '--message', metavar='<message>', type=str,
+ help='Message to send', required=False, default=DEFAULT_MESSAGE,
+ dest='message')
+
+ arg_parser.add_argument('-a', '--address', metavar='<address>', type=str,
+ help='Sender IP address', required=False, default='localhost',
+ dest='address')
 
- arg_parser.add_argument('-e', '--num-messages', metavar='<num_messages>', type=int,
- help='Set the amount of message to be sent.',
- required=False, default=0, dest='num_messages')
+ arg_parser.add_argument('-p', '--port', metavar='<port>', type=int,
+ help='Sender destination port', required=False, default=514,
+ dest='port')
 
- arg_parser.add_argument('-f', '--fixed-message-size', metavar='<fixed_message_size>', type=int, required=False,
- default=None, help='Size of all the agent modules messages (KB)', dest='fixed_message_size')
+ arg_parser.add_argument('--protocol', metavar='<protocol>', type=str,
+ help='Sender protocol', required=False, default='tcp', choices=['tcp', 'udp'],
+ dest='protocol')
 
- arg_parser.add_argument('-t', '--interval-burst-time', metavar='<interval_burst_time>', dest='interval_burst_time',
- type=int, required=False, default=0,
- help='Interval time in seconds for the messages burst')
+ arg_parser.add_argument('-e', '--eps', metavar='<eps>', type=int,
+ help='Event per second', required=False, default=-1, dest='eps')
 
- arg_parser.add_argument('-b', '--num-messages-per-burst', metavar='<num_messages_per_burst>',
- dest='num_messages_per_burst', type=int, required=False, default=0,
- help='Number of messages to send per burst')
+ arg_parser.add_argument('-d', '--debug', action='store_true', required=False, help='Activate debug logging')
 
- args = arg_parser.parse_args()
+ return arg_parser.parse_args()
 
- syslogger = Syslogger(args.message, args.num_messages, args.fixed_message_size, args.interval_burst_time,
- args.num_messages_per_burst)
 
- syslogger.start()
+def send_messages(message, num_messages, eps, address='locahost', port=514, protocol=TCP):
+ sent_messages = 0
+ custom_message = f"{message}\n" if message[-1] != '\n' not in message else message
+ protocol_limit = TCP_LIMIT if protocol == TCP else UDP_LIMIT
+ speed = eps if eps > 0 else protocol_limit
+
+ LOGGER.info(f"Sending {num_messages} to {address}:{port} via {protocol.upper()} ({speed}/s)")
+
+ # Create socket
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM if protocol == TCP else socket.SOCK_DGRAM)
+ if protocol == TCP:
+ sock.connect((address, port))
+
+ try:
+ # Get initial time
+ initial_batch_time = time.time()
+ start_batch_time = time.time()
+
+ # Send the specified number messages
+ while sent_messages < num_messages:
+ if protocol == TCP:
+ sock.send(custom_message.encode())
+ else:
+ sock.sendto(custom_message.encode(), (address, port))
+ sent_messages += 1
+
+ # Wait until next batch
+ if sent_messages % speed == 0:
+ time.sleep(1 - (time.time() - start_batch_time))
+ start_batch_time = time.time()
+
+ LOGGER.info(f"Sent {sent_messages} messages in {round(time.time() - initial_batch_time, 0)}s")
+ finally:
+ sock.close()
+
+
+def main():
+ parameters = get_parameters()
+ set_logging(parameters.debug)
+ validate_parameters(parameters)
+
+ send_messages(parameters.message, parameters.messages_number, parameters.eps, parameters.address, parameters.port,
+ parameters.protocol)
 
 
 if __name__ == "__main__":

deps/wazuh_testing/wazuh_testing/tools/run_simulator.py

@@ -23,13 +23,20 @@ def simulate_agent(param):
  -c {param['enable_logcollector_msg_number']} -g {param['message']}", shell=True)
 
 
-def syslog_simulator(param):
- """Function to run the script syslog_simulator.py
+def syslog_simulator(parameters):
+ """Run the syslog simulator tool.
 
  Args:
- param (dict): Dictionary with script parameters
+ parameters (dict): Script parameters.
  """
  python_executable = sys.executable
- subprocess.call(f"{python_executable} {SYSLOG_SIMULATOR} -m {param['message']} -e {param['num_messages']} \
- -f {param['msg_size']} -t {param['interval_burst_time']} -b {param['messages_per_burst']}",
- shell=True)
+ run_parameters = f"{python_executable} {SYSLOG_SIMULATOR} "
+ run_parameters += f"-a {parameters['address']} " if 'address' in parameters else ''
+ run_parameters += f"-e {parameters['eps']} " if 'eps' in parameters else ''
+ run_parameters += f"--protocol {parameters['protocol']} " if 'protocol' in parameters else ''
+ run_parameters += f"-n {parameters['messages_number']} " if 'messages_number' in parameters else ''
+ run_parameters += f"-m {parameters['message']} " if 'message' in parameters else ''
+ run_parameters = run_parameters.strip()
+
+ # Run the syslog simulator tool with custom parameters
+ subprocess.call(run_parameters, shell=True)

tests/integration/test_analysisd/test_eps/data/configuration_template/event_processing_test_module/configuration_limitation.yaml

@@ -0,0 +1,21 @@
+- sections:
+ - section: remote
+ elements:
+ - connection:
+ value: syslog
+ - port:
+ value: PORT
+ - protocol:
+ value: PROTOCOL
+ - allowed-ips:
+ value: 0.0.0.0/0
+ - section: global
+ elements:
+ - limits:
+ elements:
+ - eps:
+ elements:
+ - maximum:
+ value: MAXIMUM
+ - timeframe:
+ value: TIMEFRAME

tests/integration/test_analysisd/test_eps/data/test_cases/event_processing_test_module/cases_limitation.yaml

@@ -0,0 +1,16 @@
+- name: maximum 500 - timeframe 5
+ description: EPS limitation
+ configuration_parameters:
+ PORT: 514
+ PROTOCOL: tcp
+ MAXIMUM: '250'
+ TIMEFRAME: '10'
+ metadata:
+ maximum: 250
+ timeframe: 10
+ # syslog simulator parameters
+ address: 'localhost'
+ port: 514
+ protocol: 'tcp'
+ messages_number: 10000
+ eps: 1000

tests/integration/test_analysisd/test_eps/test_event_processing.py

@@ -0,0 +1,71 @@
+import os
+import pytest
+import threading
+import time
+from math import ceil
+
+from wazuh_testing.tools.configuration import load_configuration_template, get_test_cases_data
+from wazuh_testing.modules.analysisd import event_monitor as evm
+from wazuh_testing.tools.services import control_service
+from wazuh_testing.modules.analysisd import ANALYSISD_STATE_INTERNAL_DEFAULT
+from wazuh_testing.processes import check_if_daemons_are_running
+from wazuh_testing.tools.run_simulator import syslog_simulator
+from wazuh_testing.tools.thread_executor import ThreadExecutor
+
+# Reference paths
+TEST_DATA_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data')
+CONFIGURATIONS_PATH = os.path.join(TEST_DATA_PATH, 'configuration_template', 'event_processing_test_module')
+TEST_CASES_PATH = os.path.join(TEST_DATA_PATH, 'test_cases', 'event_processing_test_module')
+local_internal_options = {'wazuh_modules.debug': '2', 'monitord.rotate_log': '0',
+ 'analysisd.state_interval': '1'}
+
+# ------------------------------- TEST_LIMITATION ----------------------------------------------------------------------
+# Configuration and cases data
+t1_configurations_path = os.path.join(CONFIGURATIONS_PATH, 'configuration_limitation.yaml')
+t1_cases_path = os.path.join(TEST_CASES_PATH, 'cases_limitation.yaml')
+
+# Limitation test configurations (t1)
+t1_configuration_parameters, t1_configuration_metadata, t1_case_ids = get_test_cases_data(t1_cases_path)
+t1_configurations = load_configuration_template(t1_configurations_path, t1_configuration_parameters,
+ t1_configuration_metadata)
+
+@pytest.mark.tier(level=0)
+@pytest.mark.parametrize('configuration, metadata', zip(t1_configurations, t1_configuration_metadata), ids=t1_case_ids)
+def test_limitation(configuration, metadata, load_wazuh_basic_configuration,
+ set_wazuh_configuration, configure_local_internal_options_module,
+ truncate_monitored_files, restart_wazuh_daemon_function):
+ # Set syslog simulator parameters according to the use case data
+ syslog_simulator_parameters = {'address': metadata['address'], 'port': metadata['port'],
+ 'protocol': metadata['protocol'], 'eps': metadata['eps'],
+ 'messages_number': metadata['messages_number']}
+
+ # Run syslog simulator thread
+ syslog_simulator_thread = ThreadExecutor(syslog_simulator, {'parameters': syslog_simulator_parameters})
+ syslog_simulator_thread.start()
+
+ # Wait until syslog simulator is started
+ time.sleep(1)
+
+ # Get analysisd stats
+ analysisd_state = evm.get_analysisd_state()
+ events_received = int(analysisd_state['events_received'])
+ events_processed = int(analysisd_state['events_processed'])
+
+ # Check that wazuh-manager is processing syslog events
+ assert events_received > 0, '(0): No events are being received when it is expected'
+ assert events_processed > 0, 'No events are being processed when it is expected'
+
+ # Wait until the limitation period has expired
+ time.sleep(ceil((metadata['maximum'] * metadata['timeframe']) / metadata['eps']))
+
+ # Get analysisd stats in limitation stage
+ analysisd_state = evm.get_analysisd_state()
+ events_received = int(analysisd_state['events_received'])
+ events_processed = int(analysisd_state['events_processed'])
+
+ # Check that the wazuh-manager is receiving events but it is not processing them due to the limitation
+ assert events_received > 0, '(1): No events are being received when it is expected'
+ assert events_processed == 0, f"Events are being processed when the limit has been reached. {events_processed} != 0"
+
+ # Wait until syslog simulator ends
+ syslog_simulator_thread.join()