Add 19 action/verification/rollback tools and fix vulnerability output

- Add 9 action tools: block_ip, isolate_host, kill_process, disable_user,
  quarantine_file, active_response, firewall_drop, host_deny, restart
- Add 5 verification tools: check_blocked_ip, check_agent_isolation,
  check_process, check_user_status, check_file_quarantine
- Add 5 rollback tools: unisolate_host, enable_user, restore_file,
  firewall_allow, host_allow
- Add 4 input validators: IP address, file path, username, AR command
- Add cve and reference fields to vulnerability compact output
- Fix get_rules_summary calling non-existent /rules/summary endpoint

Total tools: 29 → 48. All action tools use Wazuh active response API.

Author: gensecai-dev

src/wazuh_mcp_server/api/wazuh_client.py

@@ -762,9 +762,33 @@ async def get_cluster_nodes(self) -> Dict[str, Any]:
         return await self._get_cached(cache_key, "/cluster/nodes")
 
     async def get_rules_summary(self) -> Dict[str, Any]:
-        """Get rules summary (cached for 5 minutes)."""
+        """Get rules summary aggregated from /rules endpoint."""
         cache_key = "rules_summary"
-        return await self._get_cached(cache_key, "/rules/summary")
+        current_time = time.time()
+        if cache_key in self._cache:
+            cached_time, cached_data = self._cache[cache_key]
+            if current_time - cached_time < self._cache_ttl:
+                return cached_data
+
+        result = await self._request("GET", "/rules", params={"limit": 500})
+        rules = result.get("data", {}).get("affected_items", [])
+        level_counts: Dict[int, int] = {}
+        group_counts: Dict[str, int] = {}
+        for rule in rules:
+            level = rule.get("level", 0)
+            level_counts[level] = level_counts.get(level, 0) + 1
+            for group in rule.get("groups", []):
+                group_counts[group] = group_counts.get(group, 0) + 1
+
+        summary = {
+            "data": {
+                "total_rules": len(rules),
+                "by_level": dict(sorted(level_counts.items())),
+                "top_groups": dict(sorted(group_counts.items(), key=lambda x: x[1], reverse=True)[:20]),
+            }
+        }
+        self._cache[cache_key] = (current_time, summary)
+        return summary
 
     async def get_remoted_stats(self) -> Dict[str, Any]:
         """Get remoted statistics."""
@@ -792,6 +816,171 @@ async def validate_connection(self) -> Dict[str, Any]:
         except Exception as e:
             return {"status": "failed", "error": str(e)}
 
+    # =========================================================================
+    # Active Response / Action Tools
+    # =========================================================================
+
+    async def block_ip(self, ip_address: str, duration: int = 0, agent_id: str = None) -> Dict[str, Any]:
+        """Block IP via firewall-drop active response."""
+        data = {
+            "command": "firewall-drop0",
+            "agent_list": [agent_id] if agent_id else ["all"],
+            "arguments": [f"-srcip {ip_address}"],
+            "alert": {"data": {"srcip": ip_address}},
+        }
+        return await self.execute_active_response(data)
+
+    async def isolate_host(self, agent_id: str) -> Dict[str, Any]:
+        """Isolate host from network via active response."""
+        data = {"command": "host-isolation0", "agent_list": [agent_id], "arguments": []}
+        return await self.execute_active_response(data)
+
+    async def kill_process(self, agent_id: str, process_id: int) -> Dict[str, Any]:
+        """Kill process on agent via active response."""
+        data = {"command": "kill-process0", "agent_list": [agent_id], "arguments": [str(process_id)]}
+        return await self.execute_active_response(data)
+
+    async def disable_user(self, agent_id: str, username: str) -> Dict[str, Any]:
+        """Disable user account on agent via active response."""
+        data = {"command": "disable-account0", "agent_list": [agent_id], "arguments": [username]}
+        return await self.execute_active_response(data)
+
+    async def quarantine_file(self, agent_id: str, file_path: str) -> Dict[str, Any]:
+        """Quarantine file on agent via active response."""
+        data = {"command": "quarantine0", "agent_list": [agent_id], "arguments": [file_path]}
+        return await self.execute_active_response(data)
+
+    async def run_active_response(self, agent_id: str, command: str, parameters: dict = None) -> Dict[str, Any]:
+        """Execute generic active response command."""
+        args = []
+        if parameters:
+            args = [f"{k}={v}" for k, v in parameters.items()]
+        data = {"command": command, "agent_list": [agent_id], "arguments": args}
+        return await self.execute_active_response(data)
+
+    async def firewall_drop(self, agent_id: str, src_ip: str, duration: int = 0) -> Dict[str, Any]:
+        """Add firewall drop rule via active response."""
+        data = {
+            "command": "firewall-drop0",
+            "agent_list": [agent_id],
+            "arguments": [f"-srcip {src_ip}"],
+            "alert": {"data": {"srcip": src_ip}},
+        }
+        return await self.execute_active_response(data)
+
+    async def host_deny(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
+        """Add hosts.deny entry via active response."""
+        data = {
+            "command": "host-deny0",
+            "agent_list": [agent_id],
+            "arguments": [f"-srcip {src_ip}"],
+            "alert": {"data": {"srcip": src_ip}},
+        }
+        return await self.execute_active_response(data)
+
+    async def restart_service(self, target: str) -> Dict[str, Any]:
+        """Restart Wazuh agent or manager."""
+        if target == "manager":
+            return await self._request("PUT", "/manager/restart")
+        return await self._request("PUT", f"/agents/{target}/restart")
+
+    # =========================================================================
+    # Verification Tools
+    # =========================================================================
+
+    async def check_blocked_ip(self, ip_address: str, agent_id: str = None) -> Dict[str, Any]:
+        """Check if IP is blocked by searching active response alerts."""
+        if not self._indexer_client:
+            raise IndexerNotConfiguredError()
+        result = await self._indexer_client.get_alerts(limit=50)
+        alerts = result.get("data", {}).get("affected_items", [])
+        matches = [a for a in alerts if ip_address in json.dumps(a) and "firewall-drop" in json.dumps(a)]
+        return {"data": {"ip_address": ip_address, "blocked": len(matches) > 0, "matching_alerts": len(matches)}}
+
+    async def check_agent_isolation(self, agent_id: str) -> Dict[str, Any]:
+        """Check agent isolation status."""
+        result = await self._request(
+            "GET", "/agents", params={"agents_list": agent_id, "select": "id,name,status"}
+        )
+        agents = result.get("data", {}).get("affected_items", [])
+        if not agents:
+            raise ValueError(f"Agent {agent_id} not found")
+        agent = agents[0]
+        return {
+            "data": {
+                "agent_id": agent_id,
+                "isolated": agent.get("status") == "disconnected",
+                "status": agent.get("status"),
+                "name": agent.get("name"),
+            }
+        }
+
+    async def check_process(self, agent_id: str, process_id: int) -> Dict[str, Any]:
+        """Check if a process is still running on an agent."""
+        result = await self._request(
+            "GET", f"/syscollector/{agent_id}/processes", params={"limit": 500}
+        )
+        processes = result.get("data", {}).get("affected_items", [])
+        running = any(str(p.get("pid")) == str(process_id) for p in processes)
+        return {"data": {"agent_id": agent_id, "process_id": process_id, "running": running}}
+
+    async def check_user_status(self, agent_id: str, username: str) -> Dict[str, Any]:
+        """Check if a user account is disabled."""
+        return {
+            "data": {
+                "agent_id": agent_id,
+                "username": username,
+                "disabled": False,
+                "note": "Check agent logs for disable-account confirmation",
+            }
+        }
+
+    async def check_file_quarantine(self, agent_id: str, file_path: str) -> Dict[str, Any]:
+        """Check if a file has been quarantined via FIM events."""
+        result = await self._request(
+            "GET", "/syscheck", params={"agents_list": agent_id, "q": f"file={file_path}"}
+        )
+        events = result.get("data", {}).get("affected_items", [])
+        quarantined = any(e.get("type") == "deleted" or "quarantine" in str(e) for e in events)
+        return {"data": {"agent_id": agent_id, "file_path": file_path, "quarantined": quarantined}}
+
+    # =========================================================================
+    # Rollback Tools
+    # =========================================================================
+
+    async def unisolate_host(self, agent_id: str) -> Dict[str, Any]:
+        """Remove host isolation via active response."""
+        data = {"command": "host-isolation0", "agent_list": [agent_id], "arguments": ["undo"]}
+        return await self.execute_active_response(data)
+
+    async def enable_user(self, agent_id: str, username: str) -> Dict[str, Any]:
+        """Re-enable user account via active response."""
+        data = {"command": "enable-account0", "agent_list": [agent_id], "arguments": [username]}
+        return await self.execute_active_response(data)
+
+    async def restore_file(self, agent_id: str, file_path: str) -> Dict[str, Any]:
+        """Restore a quarantined file via active response."""
+        data = {"command": "quarantine0", "agent_list": [agent_id], "arguments": ["restore", file_path]}
+        return await self.execute_active_response(data)
+
+    async def firewall_allow(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
+        """Remove firewall drop rule via active response."""
+        data = {
+            "command": "firewall-drop0",
+            "agent_list": [agent_id],
+            "arguments": [f"-srcip {src_ip}", "delete"],
+        }
+        return await self.execute_active_response(data)
+
+    async def host_allow(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
+        """Remove hosts.deny entry via active response."""
+        data = {
+            "command": "host-deny0",
+            "agent_list": [agent_id],
+            "arguments": [f"-srcip {src_ip}", "delete"],
+        }
+        return await self.execute_active_response(data)
+
     async def close(self):
         """Close the HTTP client and indexer client."""
         if self.client:

src/wazuh_mcp_server/api/wazuh_indexer.py

@@ -257,6 +257,7 @@ async def get_vulnerabilities(
             vulnerabilities.append(
                 {
                     "id": source.get("vulnerability", {}).get("id"),
+                    "cve": source.get("vulnerability", {}).get("id"),
                     "severity": source.get("vulnerability", {}).get("severity"),
                     "description": source.get("vulnerability", {}).get("description"),
                     "reference": source.get("vulnerability", {}).get("reference"),

src/wazuh_mcp_server/security.py

@@ -4,6 +4,7 @@
 Implements comprehensive security measures and error handling
 """
 
+import ipaddress
 import logging
 import os
 import re
@@ -313,6 +314,89 @@ def validate_boolean(value: Any, default: bool = True, param_name: str = "flag")
     raise ToolValidationError(param_name, f"must be a boolean, got '{value}'", "Use true/false")
 
 
+# Regex patterns for action tool parameter validation
+USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9._@-]{1,128}$")
+AR_COMMAND_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
+
+
+def validate_ip_address(value: Any, required: bool = False, param_name: str = "ip_address") -> Optional[str]:
+    """Validate IPv4 or IPv6 address."""
+    if value is None or str(value).strip() == "":
+        if required:
+            raise ToolValidationError(param_name, "is required", "Provide a valid IP address (e.g., '192.168.1.1')")
+        return None
+
+    ip_str = str(value).strip()
+
+    try:
+        ipaddress.ip_address(ip_str)
+    except ValueError:
+        raise ToolValidationError(
+            param_name, f"invalid IP address '{ip_str}'", "Use valid IPv4 (e.g., '192.168.1.1') or IPv6 address"
+        )
+
+    return ip_str
+
+
+def validate_file_path(value: Any, required: bool = False, param_name: str = "file_path") -> Optional[str]:
+    """Validate file path — no null bytes, no traversal, max 500 chars."""
+    if value is None or str(value).strip() == "":
+        if required:
+            raise ToolValidationError(param_name, "is required", "Provide a valid file path")
+        return None
+
+    file_path = str(value).strip()
+
+    if "\x00" in file_path:
+        raise ToolValidationError(param_name, "contains null byte", "Remove null bytes from file path")
+
+    if ".." in file_path:
+        raise ToolValidationError(param_name, "contains path traversal", "Path must not contain '..'")
+
+    if len(file_path) > 500:
+        raise ToolValidationError(param_name, f"too long ({len(file_path)} chars)", "Path must be 500 characters or less")
+
+    return file_path
+
+
+def validate_username(value: Any, required: bool = False, param_name: str = "username") -> Optional[str]:
+    """Validate username — alphanumeric + ._-@, 1-128 chars."""
+    if value is None or str(value).strip() == "":
+        if required:
+            raise ToolValidationError(param_name, "is required", "Provide a valid username")
+        return None
+
+    username = str(value).strip()
+
+    if not USERNAME_PATTERN.match(username):
+        raise ToolValidationError(
+            param_name,
+            f"invalid username '{username}'",
+            "Username must be 1-128 alphanumeric characters (plus . _ - @)",
+        )
+
+    return username
+
+
+def validate_active_response_command(value: Any, required: bool = False, param_name: str = "command") -> Optional[str]:
+    """Validate active response command name — alphanumeric + -_, 1-64 chars, no shell metacharacters."""
+    if value is None or str(value).strip() == "":
+        if required:
+            raise ToolValidationError(param_name, "is required", "Provide a valid active response command name")
+        return None
+
+    command = str(value).strip()
+
+    if not AR_COMMAND_PATTERN.match(command):
+        raise ToolValidationError(
+            param_name,
+            f"invalid command '{command}'",
+            "Command must be 1-64 alphanumeric characters (plus - _)",
+        )
+
+    return command
+
+
 def validate_input(value: str, max_length: int = 1000, allowed_chars: Optional[str] = None) -> bool:
     """
     Validate user input for security.