1
// Before: Traditional database-centric approach
2
int update_agent_status(agent_t *agent) {
3
    // Write to database
4
    wdb_update_agent(agent->id, agent->status);
5
    // External systems must poll database
6
    return 0;
7
}
8

9
// After: Real-time NATS streaming
10
int update_agent_status(agent_t *agent) {
11
    // Write to database (backward compatibility)
12
    wdb_update_agent(agent->id, agent->status);
13

14
    // Real-time NATS broadcast
15
    nats_publish_agent_event(agent, "status.change");
16

17
    // Instant notification to all subscribers
18
    return 0;
19
}

1
Agent → Manager → Database → API → External System
2
        (1-5s)     (100ms)    (5-30s polling)
3
Total Latency: 6-35 seconds

1
Agent → Manager → NATS → All Subscribers
2
        (1-5s)     (1ms)   (instant)
3
Total Latency: 1-5 seconds

1
// New NATS publishing in remoted daemon
2
typedef struct {
3
    natsConnection *conn;
4
    char *subject_prefix;
5
    bool enabled;
6
    pthread_mutex_t mutex;
7
} nats_context_t;
8

9
// Publish agent events without blocking main flow
10
int publish_agent_event(agent_t *agent, const char *event_type) {
11
    if (!nats_ctx.enabled) return 0;
12

13
    char subject[256];
14
    snprintf(subject, sizeof(subject), "%s.agent.%s.%s",
15
             nats_ctx.subject_prefix, agent->id, event_type);
16

17
    // Non-blocking publish
18
    natsConnection_PublishAsync(nats_ctx.conn, subject,
19
                               agent_data, data_len);
20
    return 0;
21
}

1
# Multiple Wazuh managers publishing to NATS cluster
2
wazuh-manager-1:
3
  nats_subjects:
4
    - wazuh.cluster.node1.events
5
    - wazuh.cluster.node1.alerts
6

7
wazuh-manager-2:
8
  nats_subjects:
9
    - wazuh.cluster.node2.events
10
    - wazuh.cluster.node2.alerts
11

12
# XDR platform subscribes to all
13
xdr-correlator:
14
  subscriptions:
15
    - wazuh.cluster.*.events
16
    - wazuh.cluster.*.alerts

1
# XDR Correlation Engine (Python example)
2
import asyncio
3
import nats
4
from nats.aio.client import Client as NATS
5

6
class XDRCorrelator:
7
    def __init__(self):
8
        self.nc = NATS()
9
        self.correlation_window = {}
10

11
    async def connect(self):
12
        await self.nc.connect("nats://wazuh-nats:4222")
13

14
        # Subscribe to all Wazuh events
15
        await self.nc.subscribe("wazuh.>", cb=self.process_event)
16

17
    async def process_event(self, msg):
18
        event = json.loads(msg.data.decode())
19

20
        # Real-time correlation logic
21
        if event['event_type'] == 'agent.status.change':
22
            await self.correlate_status_change(event)
23
        elif event['event_type'] == 'security.alert':
24
            await self.correlate_security_alert(event)
25

26
    async def correlate_status_change(self, event):
27
        agent_id = event['agent']['id']
28

29
        # Check for suspicious patterns
30
        if agent_id in self.correlation_window:
31
            recent_events = self.correlation_window[agent_id]
32

33
            # Detect rapid connect/disconnect pattern
34
            if self.detect_beacon_pattern(recent_events, event):
35
                await self.raise_xdr_alert("Potential C2 beacon detected", event)

1
// Wazuh manager publishes hunt telemetry
2
typedef struct {
3
    char query[1024];
4
    char agent_filter[256];
5
    time_t start_time;
6
    time_t end_time;
7
    int max_results;
8
} threat_hunt_query_t;
9

10
int execute_threat_hunt(threat_hunt_query_t *query) {
11
    // Execute hunt across agents
12
    hunt_results_t *results = run_distributed_query(query);
13

14
    // Stream results to NATS in real-time
15
    for (int i = 0; i < results->count; i++) {
16
        char subject[256];
17
        snprintf(subject, sizeof(subject),
18
                "wazuh.threathunt.%s.result", query->hunt_id);
19

20
        // Publish each result as it's found
21
        nats_publish(subject, results->items[i]);
22
    }
23

24
    return results->count;
25
}

1
// NATS-triggered response actions
2
void setup_response_subscriptions() {
3
    // Subscribe to response commands from XDR
4
    natsConnection_Subscribe(nats_conn,
5
        "wazuh.response.execute",
6
        response_handler, NULL);
7
}
8

9
void response_handler(natsConnection *nc, natsSubscription *sub,
10
                     natsMsg *msg, void *closure) {
11
    response_command_t cmd;
12
    parse_response_command(natsMsg_GetData(msg), &cmd);
13

14
    switch (cmd.action) {
15
        case ACTION_ISOLATE_AGENT:
16
            isolate_agent(cmd.agent_id);
17
            publish_response_status(cmd.correlation_id, "isolated");
18
            break;
19

20
        case ACTION_BLOCK_IP:
21
            add_firewall_rule(cmd.target_ip);
22
            publish_response_status(cmd.correlation_id, "blocked");
23
            break;
24

25
        case ACTION_KILL_PROCESS:
26
            remote_kill_process(cmd.agent_id, cmd.process_id);
27
            publish_response_status(cmd.correlation_id, "terminated");
28
            break;
29
    }
30
}

1
Traditional Wazuh (Database-centric):
2
- Event Processing: 5,000 EPS per manager
3
- Alert Latency: 5-10 seconds
4
- API Query Time: 100-500ms
5
- Horizontal Scaling: Complex cluster setup
6

7
NATS-Integrated Wazuh:
8
- Event Processing: 50,000+ EPS per manager
9
- Alert Latency: <100ms
10
- Streaming Updates: Real-time
11
- Horizontal Scaling: Simple pub/sub model

1
// Efficient NATS batching for high-volume events
2
typedef struct {
3
    natsMsg *messages[1000];
4
    int count;
5
    pthread_mutex_t mutex;
6
    time_t last_flush;
7
} batch_publisher_t;
8

9
void batch_publish_event(const char *subject, const void *data, int len) {
10
    pthread_mutex_lock(&batch.mutex);
11

12
    // Add to batch
13
    batch.messages[batch.count++] = natsMsg_Create(subject, data, len);
14

15
    // Flush if batch full or timeout
16
    if (batch.count >= 1000 ||
17
        time(NULL) - batch.last_flush > 1) {
18

19
        // Publish entire batch in one call
20
        for (int i = 0; i < batch.count; i++) {
21
            natsConnection_PublishMsg(nats_conn, batch.messages[i]);
22
            natsMsg_Destroy(batch.messages[i]);
23
        }
24

25
        batch.count = 0;
26
        batch.last_flush = time(NULL);
27
    }
28

29
    pthread_mutex_unlock(&batch.mutex);
30
}

1
// Secure NATS configuration with mTLS
2
int setup_secure_nats() {
3
    natsOptions *opts = NULL;
4
    natsOptions_Create(&opts);
5

6
    // mTLS authentication
7
    natsOptions_SetSecure(opts, true);
8
    natsOptions_LoadCertificatesChain(opts,
9
        "/var/ossec/etc/nats-client.crt",
10
        "/var/ossec/etc/nats-client.key");
11
    natsOptions_SetCATrustedCertificates(opts,
12
        "/var/ossec/etc/nats-ca.crt");
13

14
    // NKEY authentication for additional security
15
    natsOptions_SetNKey(opts,
16
        NKEY_SEED,
17
        sign_callback, NULL);
18

19
    // Connection security
20
    natsOptions_SetPingInterval(opts, 30000);  // 30s keepalive
21
    natsOptions_SetMaxPingsOut(opts, 3);       // Detect dead connections
22

23
    return natsConnection_Connect(&nats_conn, opts);
24
}

1
// Sign and encrypt sensitive messages
2
typedef struct {
3
    char signature[64];
4
    char nonce[32];
5
    time_t timestamp;
6
    char encrypted_payload[];
7
} secure_message_t;
8

9
int publish_secure_event(const char *subject, const void *data, int len) {
10
    secure_message_t *msg = calloc(1, sizeof(secure_message_t) + len + 16);
11

12
    // Generate nonce
13
    generate_random_bytes(msg->nonce, sizeof(msg->nonce));
14
    msg->timestamp = time(NULL);
15

16
    // Encrypt payload with AES-256-GCM
17
    int encrypted_len = aes_gcm_encrypt(data, len,
18
                                       encryption_key,
19
                                       msg->nonce,
20
                                       msg->encrypted_payload);
21

22
    // Sign the entire message
23
    generate_hmac_signature(msg,
24
                          sizeof(secure_message_t) + encrypted_len,
25
                          signing_key,
26
                          msg->signature);
27

28
    // Publish encrypted and signed message
29
    return natsConnection_Publish(nats_conn, subject,
30
                                 msg, sizeof(secure_message_t) + encrypted_len);
31
}

1
// Comprehensive audit logging for NATS operations
2
typedef struct {
3
    char event_id[64];
4
    char subject[256];
5
    char action[32];
6
    time_t timestamp;
7
    char source_ip[46];
8
    char user[64];
9
    int status;
10
    char error_msg[256];
11
} nats_audit_log_t;
12

13
void audit_nats_operation(const char *action, const char *subject, int status) {
14
    nats_audit_log_t audit = {0};
15

16
    generate_uuid(audit.event_id);
17
    strncpy(audit.action, action, sizeof(audit.action));
18
    strncpy(audit.subject, subject, sizeof(audit.subject));
19
    audit.timestamp = time(NULL);
20
    audit.status = status;
21

22
    // Log to secure audit file
23
    FILE *audit_file = fopen("/var/ossec/logs/nats_audit.log", "a");
24
    if (audit_file) {
25
        fprintf(audit_file, "%ld|%s|%s|%s|%d|%s\n",
26
                audit.timestamp, audit.event_id,
27
                audit.action, audit.subject,
28
                audit.status, audit.error_msg);
29
        fclose(audit_file);
30
    }
31

32
    // Also publish to audit stream
33
    publish_audit_event(&audit);
34
}

1
<!-- ossec.conf -->
2
<ossec_config>
3
  <nats>
4
    <enabled>yes</enabled>
5
    <server>nats://localhost:4222</server>
6
    <cluster>
7
      <server>nats://nats1:4222</server>
8
      <server>nats://nats2:4222</server>
9
      <server>nats://nats3:4222</server>
10
    </cluster>
11
    <subject_prefix>wazuh.prod</subject_prefix>
12
    <publish>
13
      <agent_events>yes</agent_events>
14
      <security_alerts>yes</security_alerts>
15
      <system_events>yes</system_events>
16
      <audit_logs>yes</audit_logs>
17
    </publish>
18
    <batch_size>100</batch_size>
19
    <flush_interval>1000</flush_interval>
20
  </nats>
21
</ossec_config>

1
<nats>
2
  <security>
3
    <tls>
4
      <enabled>yes</enabled>
5
      <cert>/var/ossec/etc/nats/client.crt</cert>
6
      <key>/var/ossec/etc/nats/client.key</key>
7
      <ca>/var/ossec/etc/nats/ca.crt</ca>
8
      <verify>yes</verify>
9
    </tls>
10
    <authentication>
11
      <nkey>SUAGY5YM...</nkey>
12
      <jwt>/var/ossec/etc/nats/jwt.token</jwt>
13
    </authentication>
14
    <encryption>
15
      <enabled>yes</enabled>
16
      <algorithm>AES-256-GCM</algorithm>
17
      <key_file>/var/ossec/etc/nats/encryption.key</key_file>
18
    </encryption>
19
  </security>
20
</nats>

1
// Expose NATS metrics for monitoring
2
typedef struct {
3
    uint64_t messages_published;
4
    uint64_t messages_failed;
5
    uint64_t bytes_sent;
6
    uint64_t reconnects;
7
    double avg_latency_ms;
8
    time_t last_error;
9
    char last_error_msg[256];
10
} nats_metrics_t;
11

12
void expose_nats_metrics() {
13
    nats_metrics_t metrics = collect_nats_metrics();
14

15
    // Publish to monitoring subject
16
    char subject[256];
17
    snprintf(subject, sizeof(subject),
18
            "%s.metrics.nats", nats_config.subject_prefix);
19

20
    cJSON *json = cJSON_CreateObject();
21
    cJSON_AddNumberToObject(json, "messages_published", metrics.messages_published);
22
    cJSON_AddNumberToObject(json, "messages_failed", metrics.messages_failed);
23
    cJSON_AddNumberToObject(json, "bytes_sent", metrics.bytes_sent);
24
    cJSON_AddNumberToObject(json, "reconnects", metrics.reconnects);
25
    cJSON_AddNumberToObject(json, "avg_latency_ms", metrics.avg_latency_ms);
26

27
    char *json_str = cJSON_Print(json);
28
    natsConnection_Publish(nats_conn, subject, json_str, strlen(json_str));
29

30
    free(json_str);
31
    cJSON_Delete(json);
32
}