Wazuh Agent Status to NATS Integration - POC Document#

Core Dependencies (Rust)#

1
[dependencies]
2
# NATS client
3
async-nats = "0.35"
4

5
# Database access
6
rusqlite = { version = "0.31", features = ["bundled"] }
7
tokio-rusqlite = "0.5"
8

9
# Async runtime
10
tokio = { version = "1.0", features = ["full"] }
11

12
# Serialization
13
serde = { version = "1.0", features = ["derive"] }
14
serde_json = "1.0"
15

16
# Error handling
17
anyhow = "1.0"
18
thiserror = "1.0"
19

20
# Logging
21
tracing = "0.1"
22
tracing-subscriber = "0.3"
23

24
# Configuration
25
config = "0.14"
26
clap = { version = "4.0", features = ["derive"] }
27

28
# File system monitoring
29
notify = "6.0"
30

31
# Cryptography (for message integrity)
32
ring = "0.17"
33
base64 = "0.22"
34

35
# Time handling
36
chrono = { version = "0.4", features = ["serde"] }

Agent Status Event Structure#

1
#[derive(Debug, Clone, Serialize, Deserialize)]
2
pub struct AgentStatusEvent {
3
    pub agent_id: String,
4
    pub agent_name: String,
5
    pub previous_status: AgentStatus,
6
    pub current_status: AgentStatus,
7
    pub ip_address: Option<String>,
8
    pub last_keepalive: Option<DateTime<Utc>>,
9
    pub manager_node: String,
10
    pub timestamp: DateTime<Utc>,
11
    pub event_id: String,
12
    pub signature: Option<String>, // For integrity verification
13
}
14

15
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
16
pub enum AgentStatus {
17
    Active,
18
    Disconnected,
19
    NeverConnected,
20
    Pending,
21
    Unknown,
22
}

1. Database Monitor (src/database/monitor.rs)#

1
pub struct WazuhDbMonitor {
2
    db_path: PathBuf,
3
    poll_interval: Duration,
4
    last_check: HashMap<String, AgentStatusSnapshot>,
5
}
6

7
impl WazuhDbMonitor {
8
    pub async fn new(db_path: impl Into<PathBuf>) -> Result<Self>;
9
    pub async fn start_monitoring(&mut self) -> Result<Receiver<AgentStatusEvent>>;
10
    pub async fn get_current_agents(&self) -> Result<Vec<AgentInfo>>;
11
    pub async fn detect_changes(&mut self) -> Result<Vec<AgentStatusEvent>>;
12
}

Key Implementation Points:

• Query /var/ossec/queue/db/global.db for agent table
• Implement SQLite row-level change detection
• Handle database locks and connection pooling
• Monitor for new agent registrations

2. NATS Publisher (src/nats/publisher.rs)#

1
pub struct NatsPublisher {
2
    client: async_nats::Client,
3
    subject_prefix: String,
4
    encryption_key: Option<Vec<u8>>,
5
}
6

7
impl NatsPublisher {
8
    pub async fn new(config: NatsConfig) -> Result<Self>;
9
    pub async fn publish_agent_event(&self, event: &AgentStatusEvent) -> Result<()>;
10
    pub async fn publish_batch(&self, events: Vec<AgentStatusEvent>) -> Result<()>;
11
    pub fn create_subject(&self, agent_id: &str, event_type: &str) -> String;
12
}

NATS Subject Structure:

• wazuh.agent.{agent_id}.status - Individual agent status changes
• wazuh.agent.{agent_id}.connected - Agent connection events
• wazuh.agent.{agent_id}.disconnected - Agent disconnection events
• wazuh.security.agent_events - Aggregated security events
• wazuh.manager.{node_id}.heartbeat - Manager health status

3. Configuration Management (src/config/mod.rs)#

1
#[derive(Debug, Deserialize)]
2
pub struct Config {
3
    pub wazuh: WazuhConfig,
4
    pub nats: NatsConfig,
5
    pub security: SecurityConfig,
6
    pub monitoring: MonitoringConfig,
7
}
8

9
#[derive(Debug, Deserialize)]
10
pub struct NatsConfig {
11
    pub url: String,
12
    pub credentials_file: Option<PathBuf>,
13
    pub tls_cert: Option<PathBuf>,
14
    pub tls_key: Option<PathBuf>,
15
    pub subject_prefix: String,
16
    pub max_reconnects: usize,
17
    pub reconnect_delay: Duration,
18
}

4. Security Implementation (src/security/mod.rs)#

1
pub struct MessageSigner {
2
    private_key: Vec<u8>,
3
}
4

5
impl MessageSigner {
6
    pub fn new(key_path: impl AsRef<Path>) -> Result<Self>;
7
    pub fn sign_message(&self, message: &str) -> Result<String>;
8
    pub fn verify_signature(&self, message: &str, signature: &str) -> Result<bool>;
9
}
10

11
pub struct EncryptionHandler {
12
    key: Vec<u8>,
13
}
14

15
impl EncryptionHandler {
16
    pub fn encrypt(&self, data: &[u8]) -> Result<Vec<u8>>;
17
    pub fn decrypt(&self, encrypted_data: &[u8]) -> Result<Vec<u8>>;
18
}

Phase 1: Database Integration (3-4 hours)#

1. SQLite Connection Management

   • Implement connection pooling for /var/ossec/queue/db/global.db
   • Handle database locks and read-only access
   • Create agent status query functions

2. Change Detection Algorithm

   • Implement efficient diff mechanism for agent status
   • Handle agent registration/deregistration events
   • Create timestamp-based change tracking

Phase 2: NATS Integration (2-3 hours)#

1. NATS Client Setup

   • Configure connection with authentication (JWT/NKeys)
   • Implement TLS encryption for transport security
   • Handle connection failures and auto-reconnection

2. Message Publishing

   • Design subject hierarchy for different event types
   • Implement message serialization (JSON/Protocol Buffers)
   • Add message deduplication mechanisms

Phase 3: Security & Reliability (2-3 hours)#

1. Message Integrity

   • Implement HMAC signing for message authenticity
   • Add optional message encryption for sensitive data
   • Create message versioning scheme

2. Error Handling & Resilience

   • Implement exponential backoff for retries
   • Add circuit breaker for external dependencies
   • Create comprehensive logging and metrics

Phase 4: Testing & Documentation (1-2 hours)#

1. Unit Tests

   • Mock SQLite database interactions
   • Test NATS message publishing scenarios
   • Validate security implementations

2. Integration Tests

   • Test with actual Wazuh manager setup
   • Verify NATS message delivery
   • Performance testing under load

Agent Status Change Event#

1
{
2
  "agent_id": "001",
3
  "agent_name": "web-server-01",
4
  "previous_status": "Active",
5
  "current_status": "Disconnected",
6
  "ip_address": "192.168.1.100",
7
  "last_keepalive": "2025-05-28T10:30:45Z",
8
  "manager_node": "wazuh-manager-01",
9
  "timestamp": "2025-05-28T10:35:00Z",
10
  "event_id": "evt_001_1716897300_status_change",
11
  "signature": "abc123def456..."
12
}

Batch Status Update#

1
{
2
  "batch_id": "batch_1716897300",
3
  "manager_node": "wazuh-manager-01",
4
  "timestamp": "2025-05-28T10:35:00Z",
5
  "events": [
6
    {
7
      /* individual agent events */
8
    }
9
  ],
10
  "summary": {
11
    "total_agents": 150,
12
    "active_agents": 145,
13
    "disconnected_agents": 5,
14
    "changes_detected": 3
15
  }
16
}

Security Requirements#

• Filesystem Permissions: Read access to Wazuh databases
• Network Security: TLS encryption for NATS connections
• Authentication: NATS JWT/NKeys or credentials file
• Message Integrity: HMAC signatures for all published events

Performance Targets#

• Latency: < 100ms from status change to NATS publish
• Throughput: Handle 1000+ agents with 5-second polling
• Resource Usage: < 50MB memory, < 1% CPU under normal load
• Reliability: 99.9% message delivery success rate

Monitoring & Observability#

• Metrics: Agent count, message publish rate, error rates
• Logging: Structured logs with correlation IDs
• Health Checks: Database connectivity, NATS connection status
• Alerting: Failed message delivery, database connection issues

Technical Implementation (40%)#

• Code quality and Rust best practices
• Proper error handling and edge cases
• Security implementation completeness
• Performance and resource efficiency

Architecture & Design (30%)#

• System design and component separation
• Scalability considerations
• Security architecture decisions
• Database integration approach

Testing & Documentation (20%)#

• Unit and integration test coverage
• Code documentation quality
• Configuration examples and setup guides
• Error handling scenarios

Security & Reliability (10%)#

• Message integrity implementation
• Connection security measures
• Failure recovery mechanisms
• Input validation and sanitization

Advanced Features (Optional)#

• Implement message compression for high-throughput scenarios
• Add support for multiple Wazuh manager nodes
• Create a web dashboard for monitoring service health

Security Enhancements (Optional)#

• Implement message rate limiting per agent
• Add anomaly detection for unusual status patterns
• Create audit logging for all published events

Operational Excellence (Optional)#

• Add Prometheus metrics endpoint
• Implement graceful shutdown with message queue draining
• Create Docker containerization with security hardening