Test input validation RPC (#567)

* Test input validation RPC

Signed-off-by: Mihai Criveti <[email protected]>

* Test input validation RPC

Signed-off-by: Mihai Criveti <[email protected]>

---------

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

.github/tools/update_dependencies.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 """
 Asynchronous Python dependency updater with persistent caching,
 comment preservation, and multiple configuration options.

Makefile

@@ -1265,7 +1265,7 @@ container-run-ssl-host: certs container-check-image
 	@echo "✅ Container started with TLS (host networking)"
 
 
-	
+
 
 container-push: container-check-image
 	@echo "📤 Preparing to push image..."
@@ -1580,7 +1580,7 @@ docker-run:
 
 docker-run-host:
 	@$(MAKE) container-run-host CONTAINER_RUNTIME=docker
-	
+
 docker-run-ssl:
 	@$(MAKE) container-run-ssl CONTAINER_RUNTIME=docker
 

docs/requirements.txt

@@ -87,4 +87,4 @@ weasyprint>=65.1
 webcolors>=24.11.1
 webencodings>=0.5.1
 zipp>=3.23.0
-zopfli>=0.2.3.post1
+zopfli>=0.2.3.post1

tests/security/test_rpc_api.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""RPC method validation test
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Mihai Criveti
+
+This script tests if alicious method names reach the tool lookup logic
+instead of being rejected at the validation layer.
+
+Usage:
+    python test_rpc_vulnerability_demo.py
+"""
+
+# Standard
+import json
+import os
+import sys
+import pytest
+
+try:
+    # Third-Party
+    import requests
+except ImportError:
+    print("Please install requests: pip install requests")
+    sys.exit(1)
+
+@pytest.mark.skip(reason="Disabled temporarily as this requires a live MCP Gateway instance")
+def test_rpc_vulnerability():
+    """Test the RPC endpoint with malicious method names."""
+
+    # Configuration
+    base_url = os.getenv("MCPGATEWAY_URL", "http://localhost:4444")
+    bearer_token = os.getenv("MCPGATEWAY_BEARER_TOKEN")
+
+    if not bearer_token:
+        print("Please set MCPGATEWAY_BEARER_TOKEN environment variable")
+        print("You can generate one with:")
+        print("  export MCPGATEWAY_BEARER_TOKEN=$(python3 -m mcpgateway.utils.create_jwt_token -u admin --secret my-test-key)")
+        sys.exit(1)
+
+    headers = {"Authorization": f"Bearer {bearer_token}", "Content-Type": "application/json"}
+
+    print("=" * 80)
+    print("RPC METHOD VALIDATION VULNERABILITY TEST")
+    print("=" * 80)
+    print(f"Testing against: {base_url}/rpc")
+    print()
+
+    # Test cases
+    test_cases = [
+        {"name": "XSS in method name", "payload": {"jsonrpc": "2.0", "method": "<script>alert(1)</script>", "id": 1}},
+        {"name": "SQL injection in method name", "payload": {"jsonrpc": "2.0", "method": "'; DROP TABLE users; --", "id": 2}},
+        {"name": "Command injection in method name", "payload": {"jsonrpc": "2.0", "method": "; cat /etc/passwd", "id": 3}},
+        {"name": "Path traversal in method name", "payload": {"jsonrpc": "2.0", "method": "../../../etc/passwd", "id": 4}},
+        {"name": "Valid method name (control)", "payload": {"jsonrpc": "2.0", "method": "tools_list", "id": 5}},
+    ]
+
+    for test in test_cases:
+        print(f"\nTest: {test['name']}")
+        print(f"Method: {test['payload']['method']}")
+        print("-" * 40)
+
+        try:
+            response = requests.post(f"{base_url}/rpc", json=test["payload"], headers=headers, timeout=5)
+
+            print(f"Status Code: {response.status_code}")
+
+            # Pretty print the response
+            try:
+                response_data = response.json()
+                print(f"Response: {json.dumps(response_data, indent=2)}")
+
+                # Check for the vulnerability signature
+                if response.status_code == 200 and "error" in response_data:
+                    error_data = response_data["error"].get("data", "")
+                    error_message = response_data["error"].get("message", "")
+
+                    # The vulnerability: malicious input appears in "Tool not found" error
+                    if "Tool not found:" in str(error_data) and test["payload"]["method"] in str(error_data):
+                        print("\n❌ VULNERABILITY DETECTED!")
+                        print("   The malicious method name reached the tool lookup logic.")
+                        print("   This indicates validation is happening AFTER processing.")
+                    elif test["payload"]["method"] in str(error_data) or test["payload"]["method"] in error_message:
+                        print("\n❌ SECURITY ISSUE: User input reflected in error message!")
+                    else:
+                        print("\n✅ Method appears to be properly rejected")
+
+                elif response.status_code in [400, 422]:
+                    print("\n✅ Method rejected at validation layer (good!)")
+                elif response.status_code == 200 and "result" in response_data:
+                    if test["name"] == "Valid method name (control)":
+                        print("\n✅ Valid method processed successfully")
+                    else:
+                        print("\n❌ CRITICAL: Malicious method was executed!")
+
+            except ValueError:
+                print(f"Raw Response: {response.text[:200]}...")
+
+        except requests.exceptions.RequestException as e:
+            print(f"Request failed: {e}")
+
+    print("\n" + "=" * 80)
+    print("SUMMARY")
+    print("=" * 80)
+    print("\nVulnerability Indicators:")
+    print("- Error message contains 'Tool not found: <malicious-input>'")
+    print("- HTTP 200 status with error instead of 422/400")
+    print("- User input reflected in error messages")
+    print("\nExpected Secure Behavior:")
+    print("- HTTP 422 or 400 for invalid method formats")
+    print("- Generic error message without user input")
+    print("- Validation before any processing/lookup")
+    print("=" * 80)
+
+
+if __name__ == "__main__":
+    test_rpc_vulnerability()