Wazuh MCP Server: Bridging SIEM and AI for Next-Generation Security Operations#

1
graph TB
2
    subgraph "Security Operations Platform"
3
        LLM[LLM/AI Model] <--> MCP[Wazuh MCP Server]
4
        MCP <--> WM[Wazuh Manager]
5
        WM --> A1[Agent 001]
6
        WM --> A2[Agent 002]
7
        WM --> A3[Agent 003]
8

9
        subgraph "MCP Components"
10
            MCP --> AUTH[JWT Authentication]
11
            MCP --> POOL[Connection Pool]
12
            MCP --> TOOLS[Tool Registry]
13
            MCP --> SSE[SSE Transport]
14
        end
15

16
        subgraph "Security Data"
17
            WM --> PROC[Processes]
18
            WM --> NET[Network Ports]
19
            WM --> PKG[Packages]
20
            WM --> SCAT[SCA Results]
21
        end
22
    end

1
# Available Security Tools
2
WAZUH_MCP_TOOLS = {
3
    'AuthenticateTool': 'Secure authentication with Wazuh API',
4
    'GetAgentsTool': 'Retrieve and analyze agent status',
5
    'GetAgentPortsTool': 'Network port analysis',
6
    'GetAgentPackagesTool': 'Installed software inventory',
7
    'GetAgentProcessesTool': 'Running process analysis',
8
    'ListRulesTool': 'Security rule exploration',
9
    'GetRuleFileContentTool': 'Rule content inspection',
10
    'GetRuleFilesTool': 'Rule file management',
11
    'GetAgentSCATool': 'Security Configuration Assessment'
12
}

1
# Configuration Options
2
wazuh:
3
  url: "https://wazuh-manager:55000"
4
  username: "admin"
5
  password: "${WAZUH_PASSWORD}"
6
  ssl_verify: true
7
  timeout: 30
8
  max_retries: 3
9

10
server:
11
  host: "0.0.0.0"
12
  port: 8000
13
  log_level: "INFO"
14

15
security:
16
  read_only_mode: false
17
  allowed_tools: ["GetAgents", "GetAgentPorts"]
18
  disabled_tools: []

1
from langchain_openai import ChatOpenAI
2
from langchain.agents import create_openai_tools_agent
3
from mcp_langchain import MultiServerMCPClient
4

5
# Initialize MCP client
6
mcp_client = MultiServerMCPClient({
7
    "wazuh-mcp-server": {
8
        "transport": "sse",
9
        "url": "http://127.0.0.1:8000/sse/",
10
        "headers": {
11
            "Authorization": f"Bearer {JWT_TOKEN}"
12
        }
13
    }
14
})
15

16
# Create AI agent with Wazuh tools
17
llm = ChatOpenAI(model="gpt-4", temperature=0)
18
tools = await mcp_client.get_tools()
19
agent = create_openai_tools_agent(llm, tools)
20

21
# Natural language security queries
22
async def analyze_security():
23
    response = await agent.ainvoke({
24
        "input": "Show me all Windows agents with suspicious PowerShell processes"
25
    })
26
    return response

1
# Traditional approach: Complex API calls
2
response = requests.get(
3
    f"{WAZUH_URL}/agents",
4
    headers={"Authorization": f"Bearer {token}"}
5
)
6
agents = response.json()["data"]["items"]
7
for agent in agents:
8
    if agent["os"]["platform"] == "windows":
9
        processes = requests.get(
10
            f"{WAZUH_URL}/syscollector/{agent['id']}/processes",
11
            headers=headers
12
        )
13
        # Complex filtering logic...
14

15
# MCP approach: Natural language
16
"Find all Windows agents running processes with encoded PowerShell commands"

1
# AI-powered investigation workflow
2
investigation_queries = [
3
    "List all agents that communicated with IP 192.168.1.100 in the last 24 hours",
4
    "Show me processes spawned by those agents during that timeframe",
5
    "Check if any of those processes have unusual network connections",
6
    "Identify any persistence mechanisms established"
7
]
8

9
for query in investigation_queries:
10
    result = await agent.ainvoke({"input": query})
11
    analyze_findings(result)

1
# Automated compliance checks
2
compliance_check = """
3
For all production servers:
4
1. Verify antivirus is installed and running
5
2. Check if all critical patches are applied
6
3. Ensure logging is enabled and functional
7
4. Validate firewall configurations
8
Generate a compliance report with any deviations
9
"""
10

11
report = await agent.ainvoke({"input": compliance_check})

1
sequenceDiagram
2
    participant Client
3
    participant MCP Server
4
    participant Wazuh API
5

6
    Client->>MCP Server: Request with credentials
7
    MCP Server->>Wazuh API: Authenticate
8
    Wazuh API-->>MCP Server: JWT Token
9
    MCP Server-->>Client: Session established
10

11
    loop Every request
12
        Client->>MCP Server: API call with token
13
        MCP Server->>MCP Server: Validate token
14
        alt Token valid
15
            MCP Server->>Wazuh API: Forward request
16
            Wazuh API-->>MCP Server: Response
17
            MCP Server-->>Client: Processed data
18
        else Token expired
19
            MCP Server->>Wazuh API: Refresh token
20
            Wazuh API-->>MCP Server: New token
21
            MCP Server->>Wazuh API: Retry request
22
        end
23
    end

1
# Optimized connection configuration
2
CONNECTION_POOL = {
3
    'max_connections': 20,
4
    'max_keepalive_connections': 10,
5
    'keepalive_expiry': 300,  # 5 minutes
6
    'http2': True,
7
    'retries': {
8
        'max_attempts': 3,
9
        'backoff_factor': 0.5
10
    }
11
}

1
# Batch operations for efficiency
2
async def batch_agent_analysis(agent_ids):
3
    tasks = []
4
    for agent_id in agent_ids:
5
        tasks.append(
6
            agent.ainvoke({
7
                "input": f"Analyze security posture of agent {agent_id}"
8
            })
9
        )
10
    results = await asyncio.gather(*tasks)
11
    return consolidate_results(results)

1
# Integrate with security orchestration
2
class WazuhMCPConnector:
3
    def __init__(self, mcp_url):
4
        self.client = MultiServerMCPClient({
5
            "wazuh": {"url": mcp_url, "transport": "sse"}
6
        })
7

8
    async def investigate_alert(self, alert_data):
9
        # AI-driven investigation
10
        context = f"Investigate alert: {alert_data}"
11
        findings = await self.client.query(context)
12

13
        # Automated response
14
        if findings['severity'] == 'critical':
15
            await self.isolate_host(findings['agent_id'])
16
            await self.create_incident_ticket(findings)
17

18
        return findings

1
# Automated threat hunting workflow
2
class ThreatHunter:
3
    def __init__(self, mcp_client):
4
        self.mcp = mcp_client
5
        self.hunting_queries = self.load_hunting_playbook()
6

7
    async def hunt(self):
8
        while True:
9
            for query in self.hunting_queries:
10
                result = await self.mcp.query(query)
11
                if self.is_suspicious(result):
12
                    await self.deep_dive_investigation(result)
13

14
            await asyncio.sleep(3600)  # Hunt every hour

1
# AI-powered alert enrichment
2
async def enrich_alert(alert):
3
    enrichment_prompt = f"""
4
    Alert: {alert['rule']['description']}
5
    Agent: {alert['agent']['name']}
6

7
    Please provide:
8
    1. Threat context and potential impact
9
    2. Related IOCs to investigate
10
    3. Recommended response actions
11
    4. Similar historical incidents
12
    """
13

14
    enriched_data = await mcp_agent.ainvoke({
15
        "input": enrichment_prompt
16
    })
17

18
    return merge_enrichment(alert, enriched_data)

1
# Prometheus metrics for MCP server
2
from prometheus_client import Counter, Histogram, Gauge
3

4
mcp_requests_total = Counter(
5
    'mcp_requests_total',
6
    'Total MCP requests',
7
    ['method', 'tool', 'status']
8
)
9

10
mcp_request_duration = Histogram(
11
    'mcp_request_duration_seconds',
12
    'MCP request duration',
13
    ['method', 'tool']
14
)
15

16
wazuh_agents_active = Gauge(
17
    'wazuh_agents_active',
18
    'Number of active Wazuh agents'
19
)

1
# Structured logging setup
2
import structlog
3

4
logger = structlog.get_logger()
5

6
logger.info(
7
    "mcp_request",
8
    tool="GetAgentProcesses",
9
    agent_id="001",
10
    user="analyst1",
11
    duration=0.234,
12
    status="success"
13
)

1
# Implement rate limiting
2
from aiohttp import ClientSession
3
from aiolimiter import AsyncLimiter
4

5
rate_limiter = AsyncLimiter(10, 1)  # 10 requests per second
6

7
async def rate_limited_request(url):
8
    async with rate_limiter:
9
        async with ClientSession() as session:
10
            return await session.get(url)