Wazuh API Connection#
The central component of the library, the WazuhApiConnection
class, abstracts the communication with the Wazuh API. It automates the authentication process and ensures that a valid
authorization token is available and applied as authorization header for every request. Additionally it limits the amount
of requests that may be filed against the Wazuh API in order to prevent error responses, that would occur when rate
limitation would be exceeded.
The example below shows how an instance of the WazuhApiConnection class can be created.
WazuhApiConnection object#from wazuh_api_kit import WazuhApiConnection
connection: WazuhApiConnection = WazuhApiConnection(
wazuh_api_url="https://example.wazuh.api:55000",
wazuh_api_username="api-user",
wazuh_api_password="MySup3rS3cRetP455w0rD",
)
The following table describes additional arguments that may be passed to configure the WazuhApiConnection’s behavior.
Argument |
Type |
Default value |
Description |
|---|---|---|---|
|
|
|
Instructs the |
|
|
|
Setting this option to |
|
|
|
TTL of authorization tokens in seconds. A reauthentication will be triggered when the token’s age exceeds this value or on encountering an |
|
|
Defaults to 12 actions per 3 seconds which are roughly 240 possible requests per minute. |
Optional rate limitation. |
Requests#
The request method may be used to file
requests against the Wazuh API. It will enforce the rate limit and ensure the authorization token’s maximum age is
not exceeded for each request as described in the diagram below. With every re-authentication the cached Wazuh API
info will be renewed as well.
Should the Wazuh API unexpectedly return an HTTP 401 Unauthorized response for your request, the
WazuhApiConnection will react by performing a
re-authentication once.
from http import HTTPMethod
from wazuh_api_kit import WazuhApiConnection
from wazuh_api_kit.model import WazuhResponse
connection: WazuhApiConnection = WazuhApiConnection(...)
api_info: WazuhResponse = connection.request(
method=HTTPMethod.GET,
path="/"
)
print(api_info.data["api_version"])
Should the Wazuh API respond with an HTTP status code other than 200, a
WazuhApiError will be thrown. This exception class
contains all the responded error information.
The request method will return with a
WazuhResponse for successful responses.
In general the Wazuh API responds with JSON objects in the following format:
{
"data": { },
"message": "A response message",
"error": 0
}
The only exception to this format is the GET / endpoint (Get API Info), which’s response contains the API Info directly.
Such an edge case’s response will be wrapped in the WazuhResponse’s
data attribute.
Bulk Responses#
Some endpoints such as GET /agents (List agents) respond with multiple entries in the following format:
{
"affected_items": [],
"total_affected_items": 0,
"total_failed_items": 0,
"failed_items": []
}
For convenience such a response’s data may be wrapped in an
WazuhResponse object.
Bellow Example illustrates how the response data of the GET /agents may be wrapped and accessed.
from http import HTTPMethod
from wazuh_api_kit import WazuhApiConnection
from wazuh_api_kit.model import WazuhResponse, WazuhBulkResponse
connection: WazuhApiConnection = WazuhApiConnection(...)
raw_agents_response: WazuhResponse = connection.request(
method=HTTPMethod.GET,
path="/agents",
parameters={"status": "active"},
)
agents: WazuhBulkResponse = WazuhBulkResponse(
response_data=raw_agents_response.data,
)
for agent_dict in agents.affected_items:
print(f"{agent_dict['id']}: {agent_dict['version']}")
000: Wazuh v4.14.4
001: Wazuh v4.14.4
Additionally a mapping function for the affected_items and failed_items may be provided and
the list types may be specified through the generic type annotation [AffectedT, FailedT] of the class.
Bellow example shows how the affected_items of the GET /agents endpoint may be mapped to a string.
from http import HTTPMethod
from wazuh_api_kit import WazuhApiConnection
from wazuh_api_kit.model import WazuhResponse, WazuhBulkResponse
connection: WazuhApiConnection = WazuhApiConnection(...)
raw_agents_response: WazuhResponse = connection.request(
method=HTTPMethod.GET,
path="/agents",
parameters={"status": "active"},
)
agents: WazuhBulkResponse[str, any] = WazuhBulkResponse[str, any](
# Maps the affected_items dictionary elements to strings
affected_map=lambda agent_dict: f"{agent_dict['id']}: {agent_dict['version']}",
response_data=raw_agents_response.data,
)
for agent in agents.affected_items:
print(agent)
000: Wazuh v4.14.4
001: Wazuh v4.14.4
Streams#
Instead of requesting multiple pages, GET endpoints, which respond with a “bulk response”, may be
requested for all their data, through the get_stream
method. This method will request pages and yield the individual items of the pages affected_items until all items have
been fetched.
Bellow example fetches all Wazuh Agents via stream.
from typing import Iterator
from wazuh_api_kit import WazuhApiConnection
connection: WazuhApiConnection = WazuhApiConnection(...)
agents: Iterator[dict[str:any]] = connection.get_stream(
path="/agents",
parameters={"status": "active"},
page_size=500,
)
for agent_dict in agents:
print(f"{agent_dict['id']}: {agent_dict['version']}")
000: Wazuh v4.14.4
001: Wazuh v4.14.4
For convenience the through the stream requested items may be mapped. This can be achieved through the specification of
a mapping function for the mapping argument, as shown in the following example.
from typing import Iterator
from wazuh_api_kit import WazuhApiConnection
connection: WazuhApiConnection = WazuhApiConnection(
wazuh_api_url="https://localhost:55000",
wazuh_api_username="wazuh",
wazuh_api_password="wazuh",
wazuh_api_insecure=True,
)
agents: Iterator[str] = connection.get_stream(
path="/agents",
parameters={"status": "active"},
page_size=500,
mapping=lambda agent_dict: f"{agent_dict['id']}: {agent_dict['version']}"
)
for agent in agents:
print(agent)
000: Wazuh v4.14.4
001: Wazuh v4.14.4