Wazuh Manager-Side Keep-Alive Integration with NATS#

1
#include "nats_integration.h"
2
#include "agent_keepalive.h"
3

4
// Enhanced agent keep-alive management
5
typedef struct {
6
    int agent_id;
7
    char agent_name[OS_MAXSTR];
8
    char ip_address[IPSIZE];
9
    time_t last_keepalive;
10
    time_t next_keepalive;
11
    agent_cs_t status;
12
    int keepalive_interval;
13
    int missed_keepalives;
14
    bool nats_notification_sent;
15
} manager_agent_t;
16

17
// Manager-initiated keep-alive function
18
int manager_send_keepalive(manager_agent_t *agent) {
19
    char keepalive_msg[OS_MAXSTR];
20
    time_t current_time = time(NULL);
21

22
    // Create keep-alive message
23
    snprintf(keepalive_msg, sizeof(keepalive_msg),
24
             "#!-agent keepalive %s %ld", agent->agent_name, current_time);
25

26
    // Send keep-alive to agent
27
    int result = send_message_to_agent(agent->agent_id, keepalive_msg);
28

29
    if (result == 0) {
30
        agent->next_keepalive = current_time + agent->keepalive_interval;
31

32
        // Publish to NATS for XDR monitoring
33
        if (nats_config.enabled) {
34
            keepalive_event_t event = {
35
                .agent_id = agent->agent_id,
36
                .agent_name = agent->agent_name,
37
                .ip_address = agent->ip_address,
38
                .timestamp = current_time,
39
                .event_type = "keepalive_sent",
40
                .manager_node = Config.node_name,
41
                .interval = agent->keepalive_interval
42
            };
43

44
            nats_publish_keepalive_event(&event);
45
        }
46

47
        mdebug2("Keep-alive sent to agent %s (%d)",
48
                agent->agent_name, agent->agent_id);
49
    } else {
50
        // Handle failed keep-alive
51
        agent->missed_keepalives++;
52
        handle_keepalive_failure(agent);
53
    }
54

55
    return result;
56
}
57

58
// Enhanced agent status monitoring
59
void monitor_agent_status(manager_agent_t *agent) {
60
    time_t current_time = time(NULL);
61
    agent_cs_t previous_status = agent->status;
62

63
    // Check if keep-alive is overdue
64
    if (current_time > agent->next_keepalive + KEEPALIVE_TIMEOUT) {
65
        agent->status = AGENT_CS_DISCONNECTED;
66
        agent->missed_keepalives++;
67

68
        // Trigger status change notification
69
        if (previous_status != agent->status) {
70
            agent_status_change_event_t event = {
71
                .agent_id = agent->agent_id,
72
                .agent_name = agent->agent_name,
73
                .previous_status = previous_status,
74
                .current_status = agent->status,
75
                .ip_address = agent->ip_address,
76
                .timestamp = current_time,
77
                .reason = "keepalive_timeout",
78
                .missed_keepalives = agent->missed_keepalives
79
            };
80

81
            // Update database
82
            wdb_update_agent_status(agent->agent_id, agent->status);
83

84
            // Publish to NATS for XDR platform
85
            if (nats_config.enabled) {
86
                nats_publish_agent_status_change(&event);
87
            }
88

89
            mwarn("Agent %s (%d) marked as disconnected due to keep-alive timeout",
90
                  agent->agent_name, agent->agent_id);
91
        }
92
    }
93
}

1
#include "keepalive_scheduler.h"
2

3
static manager_agent_t *agent_list = NULL;
4
static pthread_mutex_t agent_list_mutex = PTHREAD_MUTEX_INITIALIZER;
5
static pthread_t keepalive_thread;
6
static bool scheduler_running = false;
7

8
// Thread function for keep-alive management
9
void* keepalive_scheduler_thread(void *arg) {
10
    time_t current_time;
11
    manager_agent_t *current_agent;
12

13
    while (scheduler_running) {
14
        current_time = time(NULL);
15

16
        pthread_mutex_lock(&agent_list_mutex);
17

18
        for (current_agent = agent_list; current_agent; current_agent = current_agent->next) {
19
            // Check if it's time to send keep-alive
20
            if (current_time >= current_agent->next_keepalive) {
21
                manager_send_keepalive(current_agent);
22
            }
23

24
            // Monitor agent status
25
            monitor_agent_status(current_agent);
26
        }
27

28
        pthread_mutex_unlock(&agent_list_mutex);
29

30
        // Sleep for scheduler interval (default 30 seconds)
31
        sleep(KEEPALIVE_SCHEDULER_INTERVAL);
32
    }
33

34
    return NULL;
35
}
36

37
// Start the keep-alive scheduler
38
int start_keepalive_scheduler(void) {
39
    scheduler_running = true;
40

41
    if (pthread_create(&keepalive_thread, NULL, keepalive_scheduler_thread, NULL) != 0) {
42
        merror("Failed to create keep-alive scheduler thread");
43
        return -1;
44
    }
45

46
    minfo("Keep-alive scheduler started successfully");
47
    return 0;
48
}
49

50
// Add agent to scheduler
51
int add_agent_to_scheduler(int agent_id, const char *name, const char *ip) {
52
    manager_agent_t *new_agent = calloc(1, sizeof(manager_agent_t));
53
    if (!new_agent) {
54
        return -1;
55
    }
56

57
    new_agent->agent_id = agent_id;
58
    strncpy(new_agent->agent_name, name, sizeof(new_agent->agent_name) - 1);
59
    strncpy(new_agent->ip_address, ip, sizeof(new_agent->ip_address) - 1);
60
    new_agent->keepalive_interval = DEFAULT_KEEPALIVE_INTERVAL;
61
    new_agent->next_keepalive = time(NULL) + new_agent->keepalive_interval;
62
    new_agent->status = AGENT_CS_ACTIVE;
63
    new_agent->missed_keepalives = 0;
64

65
    pthread_mutex_lock(&agent_list_mutex);
66
    new_agent->next = agent_list;
67
    agent_list = new_agent;
68
    pthread_mutex_unlock(&agent_list_mutex);
69

70
    return 0;
71
}

1
#include "keepalive_events.h"
2

3
int nats_publish_keepalive_event(const keepalive_event_t *event) {
4
    if (!nats_conn || !event) {
5
        return -1;
6
    }
7

8
    pthread_mutex_lock(&nats_mutex);
9

10
    // Create secure JSON message
11
    cJSON *json = cJSON_CreateObject();
12
    cJSON *event_data = cJSON_CreateObject();
13

14
    // Core event data
15
    cJSON_AddStringToObject(event_data, "agent_id", event->agent_id);
16
    cJSON_AddStringToObject(event_data, "agent_name", event->agent_name);
17
    cJSON_AddStringToObject(event_data, "ip_address", event->ip_address);
18
    cJSON_AddNumberToObject(event_data, "timestamp", event->timestamp);
19
    cJSON_AddStringToObject(event_data, "event_type", event->event_type);
20
    cJSON_AddStringToObject(event_data, "manager_node", event->manager_node);
21
    cJSON_AddNumberToObject(event_data, "interval", event->interval);
22

23
    // Security metadata
24
    cJSON *security_meta = cJSON_CreateObject();
25
    cJSON_AddStringToObject(security_meta, "source", "wazuh-manager");
26
    cJSON_AddStringToObject(security_meta, "version", __ossec_version);
27
    cJSON_AddNumberToObject(security_meta, "sequence", get_message_sequence());
28

29
    cJSON_AddItemToObject(json, "event", event_data);
30
    cJSON_AddItemToObject(json, "security", security_meta);
31

32
    char *json_string = cJSON_Print(json);
33

34
    // Create subject with security classification
35
    char subject[512];
36
    snprintf(subject, sizeof(subject), "%s.manager.keepalive.%s",
37
             nats_config.subject_prefix, event->agent_id);
38

39
    // Publish with error handling
40
    natsStatus status = natsConnection_Publish(nats_conn, subject,
41
                                             json_string, strlen(json_string));
42

43
    // Audit log for security compliance
44
    if (status == NATS_OK) {
45
        minfo("Keep-alive event published for agent %s", event->agent_name);
46
    } else {
47
        merror("Failed to publish keep-alive event for agent %s: %s",
48
               event->agent_name, natsStatus_GetText(status));
49
    }
50

51
    // Cleanup
52
    free(json_string);
53
    cJSON_Delete(json);
54
    pthread_mutex_unlock(&nats_mutex);
55

56
    return (status == NATS_OK) ? 0 : -1;
57
}
58

59
int nats_publish_agent_status_change(const agent_status_change_event_t *event) {
60
    if (!nats_conn || !event) {
61
        return -1;
62
    }
63

64
    // Security-focused status change notification
65
    cJSON *json = cJSON_CreateObject();
66
    cJSON *status_data = cJSON_CreateObject();
67

68
    cJSON_AddStringToObject(status_data, "agent_id", event->agent_id);
69
    cJSON_AddStringToObject(status_data, "agent_name", event->agent_name);
70
    cJSON_AddNumberToObject(status_data, "previous_status", event->previous_status);
71
    cJSON_AddNumberToObject(status_data, "current_status", event->current_status);
72
    cJSON_AddStringToObject(status_data, "ip_address", event->ip_address);
73
    cJSON_AddNumberToObject(status_data, "timestamp", event->timestamp);
74
    cJSON_AddStringToObject(status_data, "reason", event->reason);
75
    cJSON_AddNumberToObject(status_data, "missed_keepalives", event->missed_keepalives);
76

77
    // Security alert level based on status change
78
    int alert_level = calculate_alert_level(event->previous_status,
79
                                          event->current_status,
80
                                          event->missed_keepalives);
81
    cJSON_AddNumberToObject(status_data, "alert_level", alert_level);
82

83
    cJSON_AddItemToObject(json, "status_change", status_data);
84

85
    char *json_string = cJSON_Print(json);
86

87
    // Publish to both general and alert-specific subjects
88
    char subject[512];
89
    snprintf(subject, sizeof(subject), "%s.agent.%s.status_change",
90
             nats_config.subject_prefix, event->agent_id);
91

92
    natsStatus status = natsConnection_Publish(nats_conn, subject,
93
                                             json_string, strlen(json_string));
94

95
    // Also publish to alert stream if high severity
96
    if (alert_level >= HIGH_ALERT_THRESHOLD) {
97
        char alert_subject[512];
98
        snprintf(alert_subject, sizeof(alert_subject), "%s.alerts.agent_disconnected",
99
                 nats_config.subject_prefix);
100
        natsConnection_Publish(nats_conn, alert_subject, json_string, strlen(json_string));
101
    }
102

103
    free(json_string);
104
    cJSON_Delete(json);
105

106
    return (status == NATS_OK) ? 0 : -1;
107
}

1
<ossec_config>
2
  <!-- Manager-side keep-alive configuration -->
3
  <remote>
4
    <connection>secure</connection>
5
    <port>1514</port>
6
    <protocol>tcp</protocol>
7

8
    <!-- New keep-alive settings -->
9
    <manager_keepalive>
10
      <enabled>yes</enabled>
11
      <interval>60</interval>  <!-- Send keep-alive every 60 seconds -->
12
      <timeout>180</timeout>   <!-- Consider agent dead after 180 seconds -->
13
      <max_missed>3</max_missed>
14
      <retry_interval>30</retry_interval>
15
    </manager_keepalive>
16
  </remote>
17

18
  <!-- NATS Integration for XDR Platform -->
19
  <nats>
20
    <enabled>yes</enabled>
21
    <server_url>nats://xdr-nats:4222</server_url>
22
    <credentials_file>/var/ossec/etc/nats.creds</credentials_file>
23
    <subject_prefix>wazuh.xdr</subject_prefix>
24
    <tls_cert>/var/ossec/etc/nats-client.crt</tls_cert>
25
    <tls_key>/var/ossec/etc/nats-client.key</tls_key>
26
    <max_reconnects>10</max_reconnects>
27
    <reconnect_delay>5</reconnect_delay>
28

29
    <!-- Security settings -->
30
    <encrypt_messages>yes</encrypt_messages>
31
    <message_signing>yes</message_signing>
32
    <audit_logging>yes</audit_logging>
33
  </nats>
34
</ossec_config>

1
{
2
  "event": {
3
    "agent_id": "001",
4
    "agent_name": "web-server-01",
5
    "ip_address": "192.168.1.100",
6
    "timestamp": 1716897300,
7
    "event_type": "keepalive_sent",
8
    "manager_node": "wazuh-manager-01",
9
    "interval": 60
10
  },
11
  "security": {
12
    "source": "wazuh-manager",
13
    "version": "4.7.0",
14
    "sequence": 12345
15
  }
16
}

1
{
2
  "status_change": {
3
    "agent_id": "001",
4
    "agent_name": "web-server-01",
5
    "previous_status": 0,
6
    "current_status": 1,
7
    "ip_address": "192.168.1.100",
8
    "timestamp": 1716897360,
9
    "reason": "keepalive_timeout",
10
    "missed_keepalives": 3,
11
    "alert_level": 8
12
  }
13
}