Security hardening: fix session fixation, origin validation, command injection, and 12 more issues

- Prevent session fixation by always generating server-side UUIDs for new sessions
- Harden origin validation to exact match only (remove wildcard suffix and localhost substring matching)
- Add command injection prevention for all active response arguments via _sanitize_ar_argument
- Fix auth token scopes: empty list [] now correctly denies access (None = full access for legacy)
- Fix circuit breaker tripping on user input errors (narrow to connection/server exceptions)
- Fix retry logic: let httpx 5xx/ConnectError/TimeoutException propagate to tenacity instead of wrapping
- Fix SSE ACTIVE_CONNECTIONS double-decrement by moving dec into generator finally block
- Replace regex IP validation with ipaddress.ip_address() for proper IPv4/IPv6 support
- Fix SanitizingLogFilter crash when log record args is a dict
- Replace Redis KEYS with SCAN for O(log N) iteration instead of O(N) blocking
- Fix indexer _search retry: let server errors propagate for tenacity
- Improve check_agent_isolation and check_user_status to use alert history
- Add level parameter format validation and group_by parameter whitelist
- Bound auth token storage (10k max), OAuth stores, and rate limiter memory
- Remove Redis URL from log messages to prevent credential leakage
- Add 21 new tests covering all fixes (54 total)
- Bump version to 4.0.9

Author: alokemajumder

CHANGELOG.md

@@ -5,6 +5,39 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.0.9] - 2026-03-02
+
+### Security
+- **Session fixation prevention**: Server now always generates UUIDs for new sessions; client-provided session IDs are only used to look up existing sessions
+- **Origin validation hardened**: Removed insecure wildcard suffix matching (`*.example.com`) and overly-permissive localhost substring matching; only exact match and explicit `*` wildcard are allowed
+- **Command injection prevention**: All active response and rollback methods now sanitize arguments, blocking shell metacharacters (`;`, `|`, `` ` ``, `$`, etc.)
+- **Auth token scopes**: Empty scopes list (`[]`) now correctly denies all access; previously returned True like `None` (full access)
+- **Bounded token storage**: Auth token store evicts oldest entries when exceeding 10,000 tokens
+- **OAuth bounded stores**: Authorization codes (1,000 max), access/refresh tokens (5,000 max), and client registrations (1,000 max) are now bounded to prevent memory exhaustion
+- **Rate limiter bounded memory**: Added `MAX_TRACKED_CLIENTS = 10,000` with automatic cleanup of stale entries
+- **Redis URL logging**: Removed Redis URLs (which may contain passwords) from log messages
+
+### Fixed
+- **Circuit breaker tripping on user errors**: Narrowed `expected_exception` from `Exception` to specific connection/server error types so `ValueError` doesn't trip the circuit
+- **Retry logic defeated by exception wrapping**: 5xx `HTTPStatusError`, `ConnectError`, and `TimeoutException` now propagate directly to tenacity instead of being wrapped
+- **SSE ACTIVE_CONNECTIONS double-decrement**: Moved decrement into SSE generator `finally` block with `track_connection` flag to prevent gauge going negative
+- **IPv6 validation**: Replaced regex-based IP validation with `ipaddress.ip_address()` for proper IPv4 and IPv6 support
+- **SanitizingLogFilter dict args**: Fixed crash when log record args is a dict instead of a tuple
+- **Redis KEYS command**: Replaced `KEYS` (O(N) blocking) with `SCAN` (cursor-based iteration) in `RedisSessionStore`
+- **Indexer retry logic**: `_search()` now lets 5xx, ConnectError, and TimeoutException propagate for tenacity retry
+- **`check_agent_isolation`**: Now checks alert history for isolation evidence instead of using disconnected status as a proxy
+- **`check_user_status`**: Now searches active response alert history instead of returning hardcoded data
+
+### Added
+- `_sanitize_ar_argument()` static method on `WazuhClient` for input sanitization of active response commands
+- `group_by` parameter validation with whitelist of allowed fields
+- `level` parameter format validation (must match `^[0-9]{1,2}\+?$`)
+- 21 new test cases covering all audit v2 fixes (54 total tests)
+
+### Changed
+- `CircuitBreakerConfig.expected_exception` now accepts `Union[Type[Exception], Tuple[Type[Exception], ...]]`
+- `WazuhClient` circuit breaker uses `(ConnectionError, httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError)` instead of `Exception`
+
 ## [4.0.8] - 2026-02-26
 
 ### Fixed

README.md

@@ -7,7 +7,7 @@
 
 **Production-ready MCP server connecting AI assistants to Wazuh SIEM.**
 
-> **Version 4.0.8** | Wazuh 4.8.0 - 4.14.3 | [Full Changelog](CHANGELOG.md)
+> **Version 4.0.9** | Wazuh 4.8.0 - 4.14.3 | [Full Changelog](CHANGELOG.md)
 
 ---
 

docs/api/README.md

@@ -1,6 +1,6 @@
 # FastMCP Tools API Reference
 
-Complete reference for all 48 tools available in Wazuh MCP Server v4.0.8.
+Complete reference for all 48 tools available in Wazuh MCP Server v4.0.9.
 
 ## 🛠️ Tool Categories
 
@@ -141,7 +141,7 @@ All tools return JSON responses with consistent structure:
   "metadata": {
     "query_time": "2024-01-01T12:00:00Z",
     "api_source": "wazuh_server",
-    "version": "4.0.8"
+    "version": "4.0.9"
   }
 }
 ```

docs/configuration.md

@@ -1,6 +1,6 @@
 # Configuration Guide
 
-Complete configuration reference for Wazuh MCP Server v4.0.8.
+Complete configuration reference for Wazuh MCP Server v4.0.9.
 
 ## 📋 Configuration Overview
 
@@ -128,7 +128,7 @@ CACHE_MAX_SIZE=1000                      # Maximum cache entries
 # Transport Settings
 MCP_TRANSPORT=streamable-http            # Transport type (streamable-http)
 MCP_SERVER_NAME=Wazuh MCP Server         # Server name
-MCP_SERVER_VERSION=4.0.8                 # Server version
+MCP_SERVER_VERSION=4.0.9                 # Server version
 
 # Tool Configuration
 ENABLE_SECURITY_TOOLS=true               # Enable security analysis tools

docs/security/README.md

@@ -1,6 +1,6 @@
 # Security Configuration Guide
 
-Comprehensive security hardening guide for Wazuh MCP Server v4.0.8 production deployments.
+Comprehensive security hardening guide for Wazuh MCP Server v4.0.9 production deployments.
 
 ## 🔒 Security Overview
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "wazuh-mcp-server"
-version = "4.0.8"
+version = "4.0.9"
 description = "Production-grade MCP remote server for Wazuh SIEM integration with SSE transport"
 readme = "README.md"
 license = {text = "MIT"}

src/wazuh_mcp_server/__init__.py

@@ -4,5 +4,5 @@
 through the Model Context Protocol (MCP), enabling natural language security operations.
 """
 
-__version__ = "4.0.8"
+__version__ = "4.0.9"
 __all__ = ["__version__"]

src/wazuh_mcp_server/api/wazuh_client.py

@@ -59,8 +59,13 @@ def __init__(self, config: WazuhConfig):
         self._cache_ttl = 300  # 5 minutes for static data
         self._cache_max_size = 100
 
-        # Circuit breaker for API resilience
-        circuit_config = CircuitBreakerConfig(failure_threshold=5, recovery_timeout=60, expected_exception=Exception)
+        # Circuit breaker for API resilience — only trip on connection/server errors,
+        # not on user-input errors (ValueError) which shouldn't degrade the circuit
+        circuit_config = CircuitBreakerConfig(
+            failure_threshold=5,
+            recovery_timeout=60,
+            expected_exception=(ConnectionError, httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError),
+        )
         self._circuit_breaker = CircuitBreaker(circuit_config)
 
         # Initialize Wazuh Indexer client if configured (required for Wazuh 4.8.0+)
@@ -421,15 +426,23 @@ async def _execute_request(self, method: str, endpoint: str, **kwargs) -> Dict[s
                     return response.json()
                 except (json.JSONDecodeError, ValueError):
                     raise ValueError(f"Invalid JSON response from Wazuh API after re-auth: {endpoint}")
+            elif e.response.status_code >= 500:
+                # Server errors: let them propagate as httpx exceptions so tenacity
+                # retry logic can see them and retry (via _is_retryable)
+                logger.error(f"Wazuh API server error: {e.response.status_code} - {e.response.text}")
+                raise
             else:
+                # Client errors (4xx except 401): not retryable, wrap as ValueError
                 logger.error(f"Wazuh API request failed: {e.response.status_code} - {e.response.text}")
                 raise ValueError(f"Wazuh API request failed: {e.response.status_code} - {e.response.text}")
         except httpx.ConnectError:
+            # Let connection errors propagate for retry logic
             logger.error(f"Lost connection to Wazuh server at {self.config.wazuh_host}")
-            raise ConnectionError(f"Lost connection to Wazuh server at {self.config.wazuh_host}")
+            raise
         except httpx.TimeoutException:
+            # Let timeout errors propagate for retry logic
             logger.error("Request timeout to Wazuh server")
-            raise ConnectionError("Request timeout to Wazuh server")
+            raise
 
     async def get_manager_info(self) -> Dict[str, Any]:
         """Get Wazuh manager information (cached for 5 minutes)."""
@@ -837,8 +850,27 @@ async def validate_connection(self) -> Dict[str, Any]:
     # Active Response / Action Tools
     # =========================================================================
 
+    @staticmethod
+    def _sanitize_ar_argument(value: str, param_name: str) -> str:
+        """Sanitize active response argument to prevent command injection.
+
+        Active response arguments are passed to shell commands on agents.
+        Only allow safe characters to prevent shell metacharacter injection.
+        """
+        import re
+
+        # Strip leading/trailing whitespace
+        value = value.strip()
+        if not value:
+            raise ValueError(f"{param_name} cannot be empty")
+        # Block shell metacharacters and control chars
+        if re.search(r'[;&|`$(){}\[\]<>!\\\'"\n\r\t]', value):
+            raise ValueError(f"{param_name} contains invalid characters")
+        return value
+
     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."""
+        ip_address = self._sanitize_ar_argument(ip_address, "ip_address")
         data = {
             "command": "firewall-drop0",
             "agent_list": [agent_id] if agent_id else ["all"],
@@ -854,29 +886,32 @@ async def isolate_host(self, agent_id: str) -> Dict[str, Any]:
 
     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)]}
+        data = {"command": "kill-process0", "agent_list": [agent_id], "arguments": [str(int(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."""
+        username = self._sanitize_ar_argument(username, "username")
         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."""
+        file_path = self._sanitize_ar_argument(file_path, "file_path")
         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()]
+            args = [self._sanitize_ar_argument(f"{k}={v}", f"parameter:{k}") 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."""
+        src_ip = self._sanitize_ar_argument(src_ip, "src_ip")
         data = {
             "command": "firewall-drop0",
             "agent_list": [agent_id],
@@ -887,6 +922,7 @@ async def firewall_drop(self, agent_id: str, src_ip: str, duration: int = 0) ->
 
     async def host_deny(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
         """Add hosts.deny entry via active response."""
+        src_ip = self._sanitize_ar_argument(src_ip, "src_ip")
         data = {
             "command": "host-deny0",
             "agent_list": [agent_id],
@@ -915,18 +951,33 @@ async def check_blocked_ip(self, ip_address: str, agent_id: str = None) -> Dict[
         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."""
+        """Check agent isolation status by examining agent connectivity and alert history."""
         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]
+        status = agent.get("status")
+        # Check alert history for isolation commands if indexer is available
+        isolation_confirmed = False
+        if self._indexer_client and status == "disconnected":
+            try:
+                alerts = await self._indexer_client.get_alerts(limit=20)
+                items = alerts.get("data", {}).get("affected_items", [])
+                isolation_confirmed = any(
+                    _dict_contains_text(a, "host-isolation") and _dict_contains_text(a, agent_id) for a in items
+                )
+            except Exception:
+                pass
         return {
             "data": {
                 "agent_id": agent_id,
-                "isolated": agent.get("status") == "disconnected",
-                "status": agent.get("status"),
+                "status": status,
+                "possibly_isolated": status == "disconnected",
+                "isolation_confirmed": isolation_confirmed,
                 "name": agent.get("name"),
+                "note": "A disconnected agent may be isolated or simply offline. "
+                "Check isolation_confirmed for active response evidence.",
             }
         }
 
@@ -938,13 +989,32 @@ async def check_process(self, agent_id: str, process_id: int) -> Dict[str, Any]:
         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."""
+        """Check user account status by searching active response alerts and agent logs."""
+        # Search for disable-account active response alerts
+        disable_evidence = False
+        enable_evidence = False
+        if self._indexer_client:
+            try:
+                alerts = await self._indexer_client.get_alerts(limit=50)
+                items = alerts.get("data", {}).get("affected_items", [])
+                for alert in items:
+                    if _dict_contains_text(alert, username) and _dict_contains_text(alert, agent_id):
+                        if _dict_contains_text(alert, "disable-account"):
+                            disable_evidence = True
+                        if _dict_contains_text(alert, "enable-account"):
+                            enable_evidence = True
+            except Exception:
+                pass
+        # Most recent action takes precedence
+        likely_disabled = disable_evidence and not enable_evidence
         return {
             "data": {
                 "agent_id": agent_id,
                 "username": username,
-                "disabled": False,
-                "note": "Check agent logs for disable-account confirmation",
+                "likely_disabled": likely_disabled,
+                "disable_action_found": disable_evidence,
+                "enable_action_found": enable_evidence,
+                "note": "Status based on active response alert history. " "Verify on the host for definitive status.",
             }
         }
 
@@ -966,16 +1036,19 @@ async def unisolate_host(self, agent_id: str) -> Dict[str, Any]:
 
     async def enable_user(self, agent_id: str, username: str) -> Dict[str, Any]:
         """Re-enable user account via active response."""
+        username = self._sanitize_ar_argument(username, "username")
         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."""
+        file_path = self._sanitize_ar_argument(file_path, "file_path")
         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."""
+        src_ip = self._sanitize_ar_argument(src_ip, "src_ip")
         data = {
             "command": "firewall-drop0",
             "agent_list": [agent_id],
@@ -985,6 +1058,7 @@ async def firewall_allow(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
 
     async def host_allow(self, agent_id: str, src_ip: str) -> Dict[str, Any]:
         """Remove hosts.deny entry via active response."""
+        src_ip = self._sanitize_ar_argument(src_ip, "src_ip")
         data = {
             "command": "host-deny0",
             "agent_list": [agent_id],

src/wazuh_mcp_server/api/wazuh_indexer.py

@@ -92,7 +92,9 @@ async def _ensure_initialized(self):
     @retry(
         stop=stop_after_attempt(3),
         wait=wait_exponential(multiplier=1, min=1, max=10),
-        retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
+        retry=retry_if_exception_type(
+            (httpx.RequestError, httpx.HTTPStatusError, httpx.ConnectError, httpx.TimeoutException)
+        ),
         reraise=True,
     )
     async def _search(
@@ -127,11 +129,16 @@ async def _search(
 
         except httpx.HTTPStatusError as e:
             logger.error(f"Indexer search failed: {e.response.status_code} - {e.response.text}")
+            if e.response.status_code >= 500:
+                # Let server errors propagate so tenacity retry can see them
+                raise
             raise ValueError(f"Indexer query failed: {e.response.status_code}")
         except httpx.ConnectError:
-            raise ConnectionError(f"Cannot connect to Wazuh Indexer at {self.host}:{self.port}")
+            # Let connection errors propagate for retry
+            raise
         except httpx.TimeoutException:
-            raise ConnectionError(f"Timeout connecting to Wazuh Indexer at {self.host}:{self.port}")
+            # Let timeout errors propagate for retry
+            raise
 
     async def get_alerts(
         self,

src/wazuh_mcp_server/auth.py

@@ -40,8 +40,10 @@ def is_valid(self) -> bool:
 
     def has_scope(self, scope: str) -> bool:
         """Check if token has specific scope."""
+        if self.scopes is None:
+            return True  # None means scopes not configured (legacy tokens)
         if not self.scopes:
-            return True  # No scopes means full access
+            return False  # Empty list means no permissions
         return scope in self.scopes
 
 
@@ -207,6 +209,15 @@ def create_token(self, api_key: str) -> Optional[str]:
             metadata={"api_key_name": key_obj.name, **key_obj.metadata},
         )
 
+        # Bound token storage to prevent memory growth
+        if len(self.tokens) > 10000:
+            self.cleanup_expired()
+        if len(self.tokens) > 10000:
+            # Still too many — evict oldest tokens
+            sorted_tokens = sorted(self.tokens.items(), key=lambda t: t[1].created_at)
+            for old_token, _ in sorted_tokens[: len(sorted_tokens) - 5000]:
+                self.tokens.pop(old_token, None)
+
         self.tokens[token] = token_obj
         return token