fix: add recursive scanning to regex_filter and deny_filter (#5243)

* fix:add recursive scanning to regex_filter and deny_filter

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

* removed additional comments

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

* fix precommit

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

* correct recursion depth check in regex_filter and deny_filter plugins

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

* remove hardcoded Recursion parameter/ added try catch to catch recursion range error

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

* Added documentation for recursive behavior/feature

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>

---------

Signed-off-by: prakhar-singh1928 <prakhar.singh1928@ibm.com>
Merged-by: Jitesh Nair <jiteshnair@ibm.com>

Author: prakhar-singh1928

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-06-29T10:05:28Z",
+  "generated_at": "2026-06-18T08:57:41Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"

docs/docs/using/plugins/plugins.md

@@ -75,8 +75,8 @@ Plugins for filtering, validating, and controlling content.
 
 | Plugin | Type | Description |
 |--------|------|-------------|
-| [Deny Filter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/deny_filter) | Native | Deny list plugin for blocking specific content patterns |
-| [Regex Filter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/regex_filter) | Native | Search and replace plugin using regex patterns for content filtering |
+| [Deny Filter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/deny_filter) | Native | Deny list plugin for blocking specific content patterns; recursively scans nested dicts and lists so words are caught at any depth in the payload |
+| [Regex Filter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/regex_filter) | Native | Search and replace plugin using regex patterns for content filtering; recursively rewrites string values inside nested dicts and lists at any depth in the payload |
 | [Resource Filter](https://github.com/IBM/mcp-context-forge/tree/main/plugins/resource_filter) | Native | Resource filtering with max content size, protocol restrictions, blocked domains, and content pattern replacement |
 | [File Type Allowlist](https://github.com/IBM/mcp-context-forge/tree/main/plugins/file_type_allowlist) | Native | Allows only configured MIME types and file extensions for resources |
 | [Output Length Guard](https://github.com/IBM/mcp-context-forge/tree/main/plugins/output_length_guard) | Native | Guards tool outputs by length with block or truncate strategies |

docs/docs/using/rest-passthrough.md

@@ -185,11 +185,11 @@ Configure pre and post-request plugin processing:
 
 **Allowed Plugins:**
 
-- `deny_filter` - Block requests matching deny patterns
+- `deny_filter` - Block requests matching deny patterns; scans all string values recursively through nested dicts and lists
 - `rate_limit` - Rate limiting enforcement
 - `pii_filter` - PII detection and filtering
 - `response_shape` - Response transformation
-- `regex_filter` - Regex-based content filtering
+- `regex_filter` - Regex-based content filtering; applies search/replace patterns recursively through nested dicts and lists
 - `resource_filter` - Resource access control
 
 **Execution Order:**

plugins/deny_filter/deny.py

@@ -8,14 +8,17 @@
 This module loads configurations for plugins.
 """
 
+# Standard
+from typing import Any
+
 # Third-Party
 from pydantic import BaseModel
 
 # First-Party
 from cpex.framework import Plugin, PluginConfig, PluginContext, PluginViolation, PromptPrehookPayload, PromptPrehookResult
 from mcpgateway.services.logging_service import LoggingService
 
-# Initialize logging service first
+# Initialize logging service
 logging_service = LoggingService()
 logger = logging_service.get_logger(__name__)
 
@@ -30,6 +33,40 @@ class DenyListConfig(BaseModel):
     words: list[str]
 
 
+def _scan_for_denied_words(value: Any, deny_list: list[str], path: str = "") -> list[str]:
+    """Recursively scan for denied words in nested structures.
+
+    This function walks through nested dictionaries, lists, and strings,
+    checking all string values for denied words regardless of nesting depth.
+
+    Args:
+        value: The value to scan (can be str, dict, list, or other types).
+        deny_list: List of denied words to check for.
+        path: Current path in the structure (for error reporting).
+
+    Returns:
+        List of paths where denied words were found.
+
+    Raises:
+        RecursionError: Propagated naturally if the payload structure exceeds
+            Python's recursion limit (~950 levels on CPython).
+    """
+    violations: list[str] = []
+    if isinstance(value, str):
+        for word in deny_list:
+            if word.lower() in value.lower():
+                violations.append(f"{path}: contains '{word}'")
+    elif isinstance(value, dict):
+        for k, v in value.items():
+            new_path = f"{path}.{k}" if path else k
+            violations.extend(_scan_for_denied_words(v, deny_list, new_path))
+    elif isinstance(value, list):
+        for i, item in enumerate(value):
+            new_path = f"{path}[{i}]"
+            violations.extend(_scan_for_denied_words(item, deny_list, new_path))
+    return violations
+
+
 class DenyListPlugin(Plugin):
     """Example deny list plugin."""
 
@@ -56,16 +93,24 @@ async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginC
             The result of the plugin's analysis, including whether the prompt can proceed.
         """
         if payload.args:
-            for key in payload.args:
-                if any(word in payload.args[key] for word in self._deny_list):
-                    violation = PluginViolation(
-                        reason="Prompt not allowed",
-                        description="A deny word was found in the prompt",
-                        code="deny",
-                        details={},
-                    )
-                    logger.warning(f"Deny word detected in prompt argument '{key}'")
-                    return PromptPrehookResult(modified_payload=payload, violation=violation, continue_processing=False)
+            try:
+                violations = _scan_for_denied_words(payload.args, self._deny_list)
+            except RecursionError:
+                logger.error("deny_filter: RecursionError scanning prompt args — payload is pathologically nested; aborting hook")
+                raise
+            if violations:
+                violation = PluginViolation(
+                    reason="Prompt not allowed",
+                    description="Denied words found in prompt",
+                    code="deny",
+                    details={"violations": violations},
+                )
+                logger.warning(f"Denied words detected: {violations}")
+                return PromptPrehookResult(
+                    modified_payload=payload,
+                    violation=violation,
+                    continue_processing=False,
+                )
         return PromptPrehookResult(modified_payload=payload)
 
     async def shutdown(self) -> None:

plugins/regex_filter/search_replace.py

@@ -11,11 +11,12 @@
 # Standard
 import copy
 import re
+from typing import Any
 
 # Third-Party
 from pydantic import BaseModel
 
-# Third-Party
+# First-Party
 from cpex.framework import (
     Plugin,
     PluginConfig,
@@ -29,6 +30,11 @@
     ToolPreInvokePayload,
     ToolPreInvokeResult,
 )
+from mcpgateway.services.logging_service import LoggingService
+
+# Initialize logging service
+logging_service = LoggingService()
+logger = logging_service.get_logger(__name__)
 
 
 class SearchReplace(BaseModel):
@@ -53,6 +59,35 @@ class SearchReplaceConfig(BaseModel):
     words: list[SearchReplace]
 
 
+def _scan_and_replace_recursive(value: Any, patterns: list[tuple]) -> Any:
+    """Recursively scan and replace patterns in nested structures.
+
+    This function walks through nested dictionaries, lists, and strings,
+    applying regex patterns to all string values regardless of nesting depth.
+
+    Args:
+        value: The value to scan (can be str, dict, list, or other types).
+        patterns: List of (compiled_pattern, replacement) tuples.
+
+    Returns:
+        Modified value with patterns replaced in all nested strings.
+
+    Raises:
+        RecursionError: Propagated naturally if the payload structure exceeds
+            Python's recursion limit (~950 levels on CPython).
+    """
+    if isinstance(value, str):
+        result = value
+        for pattern, replacement in patterns:
+            result = pattern.sub(replacement, result)
+        return result
+    elif isinstance(value, dict):
+        return {k: _scan_and_replace_recursive(v, patterns) for k, v in value.items()}
+    elif isinstance(value, list):
+        return [_scan_and_replace_recursive(item, patterns) for item in value]
+    return value
+
+
 class SearchReplacePlugin(Plugin):
     """Example search replace plugin."""
 
@@ -85,11 +120,11 @@ async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginC
             The result of the plugin's analysis, including whether the prompt can proceed.
         """
         if payload.args:
-            modified_args = dict(payload.args)
-            for pattern, replacement in self.__patterns:
-                for key, value in modified_args.items():
-                    if isinstance(value, str):
-                        modified_args[key] = pattern.sub(replacement, value)
+            try:
+                modified_args = _scan_and_replace_recursive(payload.args, self.__patterns)
+            except RecursionError:
+                logger.error("regex_filter: RecursionError scanning prompt args — payload is pathologically nested; aborting hook")
+                raise
             payload = payload.model_copy(update={"args": modified_args})
         return PromptPrehookResult(modified_payload=payload)
 
@@ -103,7 +138,6 @@ async def prompt_post_fetch(self, payload: PromptPosthookPayload, context: Plugi
         Returns:
             The result of the plugin's analysis, including whether the prompt can proceed.
         """
-
         if payload.result.messages:
             modified_result = copy.deepcopy(payload.result)
             for index, message in enumerate(modified_result.messages):
@@ -123,11 +157,11 @@ async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginCo
             The result of the plugin's analysis, including whether the tool can proceed.
         """
         if payload.args:
-            modified_args = dict(payload.args)
-            for pattern, replacement in self.__patterns:
-                for key, value in modified_args.items():
-                    if isinstance(value, str):
-                        modified_args[key] = pattern.sub(replacement, value)
+            try:
+                modified_args = _scan_and_replace_recursive(payload.args, self.__patterns)
+            except RecursionError:
+                logger.error("regex_filter: RecursionError scanning tool args — payload is pathologically nested; aborting hook")
+                raise
             payload = payload.model_copy(update={"args": modified_args})
         return ToolPreInvokeResult(modified_payload=payload)
 
@@ -142,11 +176,11 @@ async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: Plugin
             The result of the plugin's analysis, including whether the tool result should proceed.
         """
         if payload.result and isinstance(payload.result, dict):
-            modified_result = dict(payload.result)
-            for pattern, replacement in self.__patterns:
-                for key, value in modified_result.items():
-                    if isinstance(value, str):
-                        modified_result[key] = pattern.sub(replacement, value)
+            try:
+                modified_result = _scan_and_replace_recursive(payload.result, self.__patterns)
+            except RecursionError:
+                logger.error("regex_filter: RecursionError scanning tool result — payload is pathologically nested; aborting hook")
+                raise
             payload = payload.model_copy(update={"result": modified_result})
         elif payload.result and isinstance(payload.result, str):
             result = payload.result