QU3B1M
commented
4 months ago Update report
Created a small library to handle WazuhAPI requests and responses with the purpose of simplifying the API testing for current and future tests. The library contains three files and for now is included inside the tests helpers, later, if its approved it could be configured to be an installable library.
Files:
-
endpoints.py- Lists all the WazuhAPI endpoints used by the library```python # - Wazuh API endpoints # -- Security -- SECURITY = '/security' # Users SECURITY_USER = f'{SECURITY}/user' SECURITY_USERS = f'{SECURITY}/users' # Authentication SECURITY_AUTHENTICATE = f'{SECURITY_USER}/authenticate' # Configuration SECURITY_CONFIG = f'{SECURITY}/config' # -- Groups -- GROUPS = '/groups' # -- Agents -- AGENTS = '/agents' AGENTS_GROUP = f'{AGENTS}/group' ``` -
exceptions.py- Defines exceptions to be raised by the library corresponding to failing status codes```python from requests import RequestException class BadRequest(RequestException): """A bad request error occurred.""" class Unauthorized(RequestException): """An unauthorized error occurred.""" class Forbidden(RequestException): """A forbidden error occurred.""" class NotFound(RequestException): """A not found error occurred.""" class MethodNotAllowed(RequestException): """A method not allowed error occurred.""" class TooManyRequests(RequestException): """A request limit exceeded error occurred.""" class InternalServerError(RequestException): """An internal server error occurred.""" class ServiceUnavailable(RequestException): """A service unavailable error occurred.""" class GatewayTimeout(RequestException): """A gateway timeout error occurred.""" responses_errors = { 400: BadRequest, 401: Unauthorized, 403: Forbidden, 404: NotFound, 405: MethodNotAllowed, 429: TooManyRequests, 500: InternalServerError, 501: ServiceUnavailable, 503: GatewayTimeout } ``` -
api.py- The main file of the library where the class WazuhAPI resides.```python import requests from datetime import datetime, timedelta from .exceptions import responses_errors from . import endpoints class WazuhAPI: def __init__(self, user: str, password: str, host: str = 'localhost', port: int = 55000) -> None: self.user = user self.password = password self.host = host self.port = port # Token default values self.token = None self.token_lifetime = 900 self.token_expiration = None # Create a requests session and disable the warnings self.session = requests.Session() requests.packages.urllib3.disable_warnings() # Authenticate and save the token value and expiration self.authenticate() # Security def authenticate(self) -> str: endpoint = self._get_complete_url(endpoints.SECURITY_AUTHENTICATE) credentials = (self.user, self.password) # _send_request is not used here because of the auth parameter. response = self.session.get(endpoint, auth=credentials, verify=False) if response.status_code in responses_errors.keys(): print(f'Authentication error: {response.content}') raise responses_errors[response.status_code] self.token_expiration = datetime.now() + timedelta(seconds=self.token_lifetime) self.token = response.json()['data']['token'] def extend_token_life(self, timeout: int = 99999999) -> dict: endpoint = self._get_complete_url(endpoints.SECURITY_CONFIG) payload = {"auth_token_exp_timeout": timeout, "rbac_mode": "white"} response = self._send_request('put', endpoint, payload=payload) self.token_lifetime = timeout return response # Agents def add_agent(self, name: str, ip: str) -> dict: endpoint = self._get_complete_url(endpoints.AGENTS) payload = {'name': name, 'ip': ip} return self._send_request('post', endpoint, payload=payload) def get_agent(self, agent_id: str) -> dict: endpoint = self._get_complete_url(endpoints.AGENTS) params = {'agents_list': [agent_id]} response = self._send_request('get', endpoint, query_params=params) return response[0] if response else {} def get_agents(self, **kwargs: dict) -> list[dict]: endpoint = self._get_complete_url(endpoints.AGENTS) return self._send_request('get', endpoint, query_params=kwargs) def delete_agent(self, agent_id: str) -> dict: endpoint = self._get_complete_url(endpoints.AGENTS) params = {'agents_list': [agent_id], 'status': 'all'} return self._send_request('delete', endpoint, query_params=params) def delete_agents(self, agents_list: list, **kwargs: dict) -> dict: endpoint = self._get_complete_url(endpoints.AGENTS) params = {**kwargs, 'agents_list': agents_list} return self._send_request('delete', endpoint, query_params=params) # --- INTERNAL METHODS --- def _send_request(self, method: str, endpoint: str, payload: dict = None, query_params: dict = {}) -> dict: if not self.token: self.authenticate() elif self.token_expiration <= datetime.now(): self.authenticate() # Set the headers and send the request headers = {'Authorization': f'Bearer {self.token}'} query_params['pretty'] = 'true' response = self.session.request( method, endpoint, data=payload, headers=headers, params=query_params, verify=False) # Check if the response is an error if response.status_code in responses_errors.keys(): print(f'Failing request to: {endpoint}\nError: {response.content}') raise responses_errors[response.status_code] return response.json().get('data', {}).get('affected_items', {}) def _get_complete_url(self, endpoint) -> str: if endpoint.startswith('/'): endpoint = endpoint[1:] return f'https://{self.host}:{self.port}/{endpoint}' ```
fcaffieri
pro-akim
rauldpm
Description
This issue aims to improve and complete the wazuh-agent tests generated in the first iterations of DTT1
Tasks