Wazuh Core Integration with NATS - POC Document#

1
// New NATS integration headers
2
#include "nats_integration.h"
3
#include "agent_events.h"
4

5
// Modified function: handle_agent_connection()
6
int handle_agent_connection(agent *ag, char *msg) {
7
    int result = 0;
8
    agent_status_t previous_status = ag->status;
9

10
    // Existing Wazuh logic
11
    result = process_agent_message(ag, msg);
12

13
    // NEW: NATS integration
14
    if (nats_config.enabled && ag->status != previous_status) {
15
        agent_status_event_t event = {
16
            .agent_id = ag->id,
17
            .agent_name = ag->name,
18
            .previous_status = previous_status,
19
            .current_status = ag->status,
20
            .ip_address = ag->ip,
21
            .timestamp = time(NULL),
22
            .manager_node = Config.node_name
23
        };
24

25
        nats_publish_agent_status(&event);
26
    }
27

28
    return result;
29
}

1
// Modified function: wdb_update_agent_keepalive()
2
int wdb_update_agent_keepalive(int agent_id, agent_cs_t status,
3
                              const char *sync_status, int *sock) {
4
    int result = 0;
5
    agent_cs_t previous_status = AGENT_CS_UNKNOWN;
6

7
    // Get previous status before update
8
    previous_status = wdb_get_agent_status(agent_id);
9

10
    // Existing Wazuh database update logic
11
    result = wdb_execute_agent_keepalive_update(agent_id, status, sync_status, sock);
12

13
    // NEW: NATS integration for database-level status changes
14
    if (nats_config.enabled && result == 0 && previous_status != status) {
15
        db_status_event_t event = {
16
            .agent_id = agent_id,
17
            .previous_status = previous_status,
18
            .current_status = status,
19
            .sync_status = sync_status,
20
            .timestamp = time(NULL),
21
            .source = "wazuh-db"
22
        };
23

24
        nats_publish_db_status(&event);
25
    }
26

27
    return result;
28
}

1
#ifndef NATS_CLIENT_H
2
#define NATS_CLIENT_H
3

4
#include "shared.h"
5
#include "nats/nats.h"
6

7
typedef struct {
8
    bool enabled;
9
    char *server_url;
10
    char *credentials_file;
11
    char *subject_prefix;
12
    char *tls_cert;
13
    char *tls_key;
14
    int max_reconnects;
15
    int reconnect_delay;
16
    bool encrypt_messages;
17
    char *encryption_key;
18
} nats_config_t;
19

20
typedef struct {
21
    char agent_id[16];
22
    char agent_name[256];
23
    agent_cs_t previous_status;
24
    agent_cs_t current_status;
25
    char ip_address[46];
26
    time_t timestamp;
27
    char manager_node[256];
28
    char event_id[64];
29
} agent_status_event_t;
30

31
// Function declarations
32
int nats_init(const nats_config_t *config);
33
int nats_cleanup(void);
34
int nats_publish_agent_status(const agent_status_event_t *event);
35
int nats_publish_security_alert(const alert_data_t *alert);
36
int nats_publish_raw_message(const char *subject, const char *data);
37
bool nats_is_connected(void);
38

39
#endif /* NATS_CLIENT_H */

1
#include "nats_client.h"
2
#include "cJSON.h"
3

4
static natsConnection *nats_conn = NULL;
5
static nats_config_t nats_config;
6
static pthread_mutex_t nats_mutex = PTHREAD_MUTEX_INITIALIZER;
7

8
int nats_init(const nats_config_t *config) {
9
    natsStatus status = NATS_OK;
10
    natsOptions *opts = NULL;
11

12
    if (!config || !config->enabled) {
13
        return 0; // NATS disabled
14
    }
15

16
    memcpy(&nats_config, config, sizeof(nats_config_t));
17

18
    // Create NATS options
19
    status = natsOptions_Create(&opts);
20
    if (status != NATS_OK) {
21
        merror("Failed to create NATS options: %s", natsStatus_GetText(status));
22
        return -1;
23
    }
24

25
    // Configure TLS if specified
26
    if (nats_config.tls_cert && nats_config.tls_key) {
27
        status = natsOptions_SetSecure(opts, true);
28
        if (status == NATS_OK) {
29
            status = natsOptions_LoadCertificatesChain(opts,
30
                nats_config.tls_cert, nats_config.tls_key);
31
        }
32
    }
33

34
    // Configure credentials if specified
35
    if (nats_config.credentials_file) {
36
        status = natsOptions_SetUserCredentialsFromFiles(opts,
37
            nats_config.credentials_file, NULL);
38
    }
39

40
    // Set reconnection parameters
41
    natsOptions_SetMaxReconnect(opts, nats_config.max_reconnects);
42
    natsOptions_SetReconnectWait(opts, nats_config.reconnect_delay * 1000);
43

44
    // Connect to NATS
45
    status = natsConnection_Connect(&nats_conn, opts);
46
    if (status != NATS_OK) {
47
        merror("Failed to connect to NATS: %s", natsStatus_GetText(status));
48
        natsOptions_Destroy(opts);
49
        return -1;
50
    }
51

52
    natsOptions_Destroy(opts);
53
    minfo("NATS integration initialized successfully");
54
    return 0;
55
}
56

57
int nats_publish_agent_status(const agent_status_event_t *event) {
58
    if (!nats_conn || !event) {
59
        return -1;
60
    }
61

62
    pthread_mutex_lock(&nats_mutex);
63

64
    // Create JSON message
65
    cJSON *json = cJSON_CreateObject();
66
    cJSON *agent_id = cJSON_CreateString(event->agent_id);
67
    cJSON *agent_name = cJSON_CreateString(event->agent_name);
68
    cJSON *prev_status = cJSON_CreateNumber(event->previous_status);
69
    cJSON *curr_status = cJSON_CreateNumber(event->current_status);
70
    cJSON *ip_addr = cJSON_CreateString(event->ip_address);
71
    cJSON *timestamp = cJSON_CreateNumber(event->timestamp);
72
    cJSON *manager = cJSON_CreateString(event->manager_node);
73

74
    cJSON_AddItemToObject(json, "agent_id", agent_id);
75
    cJSON_AddItemToObject(json, "agent_name", agent_name);
76
    cJSON_AddItemToObject(json, "previous_status", prev_status);
77
    cJSON_AddItemToObject(json, "current_status", curr_status);
78
    cJSON_AddItemToObject(json, "ip_address", ip_addr);
79
    cJSON_AddItemToObject(json, "timestamp", timestamp);
80
    cJSON_AddItemToObject(json, "manager_node", manager);
81

82
    char *json_string = cJSON_Print(json);
83

84
    // Create subject
85
    char subject[512];
86
    snprintf(subject, sizeof(subject), "%s.agent.%s.status",
87
             nats_config.subject_prefix, event->agent_id);
88

89
    // Publish message
90
    natsStatus status = natsConnection_Publish(nats_conn, subject,
91
                                             json_string, strlen(json_string));
92

93
    // Cleanup
94
    free(json_string);
95
    cJSON_Delete(json);
96
    pthread_mutex_unlock(&nats_mutex);
97

98
    if (status != NATS_OK) {
99
        merror("Failed to publish NATS message: %s", natsStatus_GetText(status));
100
        return -1;
101
    }
102

103
    return 0;
104
}

1
#include "nats_client.h"
2
#include "config.h"
3

4
int Read_NATS(XML_NODE node, void *configp, __attribute__((unused)) void *mailp) {
5
    nats_config_t *nats_config = (nats_config_t *)configp;
6

7
    // Initialize defaults
8
    nats_config->enabled = false;
9
    nats_config->max_reconnects = 10;
10
    nats_config->reconnect_delay = 2;
11
    nats_config->encrypt_messages = false;
12

13
    if (!node) {
14
        return 0;
15
    }
16

17
    for (int i = 0; node[i]; i++) {
18
        if (!node[i]->element) {
19
            merror(XML_ELEMNULL);
20
            return OS_INVALID;
21
        } else if (!node[i]->content) {
22
            merror(XML_VALUENULL, node[i]->element);
23
            return OS_INVALID;
24
        }
25

26
        if (strcmp(node[i]->element, "enabled") == 0) {
27
            nats_config->enabled = (strcmp(node[i]->content, "yes") == 0);
28
        } else if (strcmp(node[i]->element, "server_url") == 0) {
29
            os_strdup(node[i]->content, nats_config->server_url);
30
        } else if (strcmp(node[i]->element, "credentials_file") == 0) {
31
            os_strdup(node[i]->content, nats_config->credentials_file);
32
        } else if (strcmp(node[i]->element, "subject_prefix") == 0) {
33
            os_strdup(node[i]->content, nats_config->subject_prefix);
34
        } else if (strcmp(node[i]->element, "tls_cert") == 0) {
35
            os_strdup(node[i]->content, nats_config->tls_cert);
36
        } else if (strcmp(node[i]->element, "tls_key") == 0) {
37
            os_strdup(node[i]->content, nats_config->tls_key);
38
        } else if (strcmp(node[i]->element, "max_reconnects") == 0) {
39
            nats_config->max_reconnects = atoi(node[i]->content);
40
        } else if (strcmp(node[i]->element, "reconnect_delay") == 0) {
41
            nats_config->reconnect_delay = atoi(node[i]->content);
42
        } else if (strcmp(node[i]->element, "encrypt_messages") == 0) {
43
            nats_config->encrypt_messages = (strcmp(node[i]->content, "yes") == 0);
44
        } else {
45
            merror(XML_INVELEM, node[i]->element);
46
            return OS_INVALID;
47
        }
48
    }
49

50
    return 0;
51
}