From b33f1ec2b215bd97d800a98004e6aeaa5d8d21ce Mon Sep 17 00:00:00 2001 From: hamzaq453 Date: Sun, 28 Dec 2025 19:28:11 +0500 Subject: [PATCH] Fix: Remove exec() usage and handle invalid OpenAPI parameter names - Add to_safe_identifier() to convert any parameter name to valid Python identifier - Refactor create_tool_function() to use closure with **kwargs instead of exec() - Handle edge cases: hyphens, dots, leading digits, Python keywords, special chars - Add comprehensive test suite covering all edge cases - Fixes #18471: OpenAPI MCP server crashes on invalid parameter names - Security: Eliminates arbitrary code execution risk from untrusted OpenAPI specs --- .../mcp_server/openapi_to_mcp_generator.py | 234 +++++--- .../test_openapi_to_mcp_generator.py | 528 ++++++++++++++++++ 2 files changed, 695 insertions(+), 67 deletions(-) create mode 100644 tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py diff --git a/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py b/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py index 72288f8e67..dc5d0ce73a 100644 --- a/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py +++ b/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py @@ -3,6 +3,8 @@ This module is used to generate MCP tools from OpenAPI specs. """ import json +import keyword +import re from typing import Any, Dict, Optional import httpx @@ -17,6 +19,64 @@ BASE_URL = "" HEADERS: Dict[str, str] = {} +def to_safe_identifier(name: str) -> str: + """ + Convert an OpenAPI parameter name to a safe Python identifier. + + This function ensures that any parameter name from an OpenAPI spec can be + used as a Python function parameter without causing syntax errors or + security issues. It handles: + - Hyphens, dots, and other special characters + - Leading digits + - Python keywords + - Special characters like $, @, etc. + + Args: + name: The original parameter name from the OpenAPI spec + + Returns: + A valid Python identifier that can be used in function signatures + + Examples: + >>> to_safe_identifier("repository-id") + 'repository_id' + >>> to_safe_identifier("2fa-code") + '_2fa_code' + >>> to_safe_identifier("user.name") + 'user_name' + >>> to_safe_identifier("$filter") + '_filter' + >>> to_safe_identifier("class") + 'class_' + """ + if not name: + return "_empty_" + + # Start with underscore if first char is not a letter + # Replace all non-alphanumeric chars (except underscore) with underscore + # Collapse multiple underscores + safe = re.sub(r'[^a-zA-Z0-9_]', '_', name) + safe = re.sub(r'_+', '_', safe) # Collapse multiple underscores + + # If starts with digit, prefix with underscore + if safe and safe[0].isdigit(): + safe = '_' + safe + + # If empty after sanitization, use a default + if not safe: + safe = '_param_' + + # If it's a Python keyword, append underscore + if keyword.iskeyword(safe): + safe = safe + '_' + + # Ensure it doesn't start with a digit (shouldn't happen after above, but double-check) + if safe and safe[0].isdigit(): + safe = '_' + safe + + return safe + + def load_openapi_spec(filepath: str) -> Dict[str, Any]: """Load OpenAPI specification from JSON file.""" with open(filepath, "r") as f: @@ -112,12 +172,20 @@ def create_tool_function( ): """Create a tool function for an OpenAPI operation. + This function creates an async tool function that can be called with + keyword arguments. Parameter names from the OpenAPI spec are safely + mapped to valid Python identifiers to avoid syntax errors and security + issues. + Args: path: API endpoint path method: HTTP method (get, post, put, delete, patch) operation: OpenAPI operation object base_url: Base URL for the API headers: Optional headers to include in requests (e.g., authentication) + + Returns: + An async function that accepts **kwargs and makes the HTTP request """ if headers is None: headers = {} @@ -125,77 +193,109 @@ def create_tool_function( path_params, query_params, body_params = extract_parameters(operation) all_params = path_params + query_params + body_params - # Build function signature dynamically - if all_params: - params_str = ", ".join(f"{p}: str = ''" for p in all_params) - else: - params_str = "" - - # Create the function code as a string - func_code = f''' -async def tool_function({params_str}) -> str: - """Dynamically generated tool function.""" - url = base_url + path + # Create mapping from original parameter names to safe identifiers + # This allows us to accept kwargs with original names but use safe names internally + param_name_map: Dict[str, str] = {} + safe_to_original_map: Dict[str, str] = {} - # Replace path parameters - path_param_names = {path_params} - for param_name in path_param_names: - param_value = locals().get(param_name, "") - if param_value: - url = url.replace("{{" + param_name + "}}", str(param_value)) - - # Build query params - query_param_names = {query_params} - params = {{}} - for param_name in query_param_names: - param_value = locals().get(param_name, "") - if param_value: - params[param_name] = param_value - - # Build request body - body_param_names = {body_params} - json_body = None - if body_param_names: - body_value = locals().get("body", {{}}) - if isinstance(body_value, dict): - json_body = body_value - elif body_value: - # If it's a string, try to parse as JSON - import json as json_module - try: - json_body = json_module.loads(body_value) if isinstance(body_value, str) else {{"data": body_value}} - except: - json_body = {{"data": body_value}} - - # Make HTTP request - async with httpx.AsyncClient() as client: - if "{method.lower()}" == "get": - response = await client.get(url, params=params, headers=headers) - elif "{method.lower()}" == "post": - response = await client.post(url, params=params, json=json_body, headers=headers) - elif "{method.lower()}" == "put": - response = await client.put(url, params=params, json=json_body, headers=headers) - elif "{method.lower()}" == "delete": - response = await client.delete(url, params=params, headers=headers) - elif "{method.lower()}" == "patch": - response = await client.patch(url, params=params, json=json_body, headers=headers) - else: - return "Unsupported HTTP method: {method}" + for orig_name in all_params: + safe_name = to_safe_identifier(orig_name) + # Handle collisions: if safe name already exists, append a counter + counter = 1 + original_safe = safe_name + while safe_name in safe_to_original_map: + safe_name = f"{original_safe}_{counter}" + counter += 1 - return response.text -''' + param_name_map[orig_name] = safe_name + safe_to_original_map[safe_name] = orig_name - # Execute the function code to create the actual function - local_vars = { - "httpx": httpx, - "headers": headers, - "base_url": base_url, - "path": path, - "method": method, - } - exec(func_code, local_vars) + # Store original parameter lists for use in the closure + original_path_params = path_params + original_query_params = query_params + original_body_params = body_params + original_method = method.lower() - return local_vars["tool_function"] + async def tool_function(**kwargs: Any) -> str: + """ + Dynamically generated tool function. + + Accepts keyword arguments where keys are the original OpenAPI parameter names. + The function safely handles parameter names that aren't valid Python identifiers. + """ + # Build URL from base_url and path + url = base_url + path + + # Replace path parameters using original names from OpenAPI spec + for orig_param_name in original_path_params: + # Try to get value using original name first, then safe name + param_value = kwargs.get(orig_param_name, "") + if not param_value and orig_param_name in param_name_map: + safe_name = param_name_map[orig_param_name] + param_value = kwargs.get(safe_name, "") + + if param_value: + # Replace {param_name} or {{param_name}} in URL + url = url.replace("{" + orig_param_name + "}", str(param_value)) + url = url.replace("{{" + orig_param_name + "}}", str(param_value)) + + # Build query params using original parameter names + params: Dict[str, Any] = {} + for orig_param_name in original_query_params: + # Try to get value using original name first, then safe name + param_value = kwargs.get(orig_param_name, "") + if not param_value and orig_param_name in param_name_map: + safe_name = param_name_map[orig_param_name] + param_value = kwargs.get(safe_name, "") + + if param_value: + # Use original parameter name in query string (as expected by API) + params[orig_param_name] = param_value + + # Build request body + json_body: Optional[Dict[str, Any]] = None + if original_body_params: + # Try "body" first (most common), then check all body param names + body_value = kwargs.get("body", {}) + if not body_value: + for orig_param_name in original_body_params: + body_value = kwargs.get(orig_param_name, {}) + if body_value: + break + # Also try safe name + if orig_param_name in param_name_map: + safe_name = param_name_map[orig_param_name] + body_value = kwargs.get(safe_name, {}) + if body_value: + break + + if isinstance(body_value, dict): + json_body = body_value + elif body_value: + # If it's a string, try to parse as JSON + try: + json_body = json.loads(body_value) if isinstance(body_value, str) else {"data": body_value} + except (json.JSONDecodeError, TypeError): + json_body = {"data": body_value} + + # Make HTTP request + async with httpx.AsyncClient() as client: + if original_method == "get": + response = await client.get(url, params=params, headers=headers) + elif original_method == "post": + response = await client.post(url, params=params, json=json_body, headers=headers) + elif original_method == "put": + response = await client.put(url, params=params, json=json_body, headers=headers) + elif original_method == "delete": + response = await client.delete(url, params=params, headers=headers) + elif original_method == "patch": + response = await client.patch(url, params=params, json=json_body, headers=headers) + else: + return f"Unsupported HTTP method: {original_method}" + + return response.text + + return tool_function def register_tools_from_openapi(spec: Dict[str, Any], base_url: str): diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py new file mode 100644 index 0000000000..fa28dc3498 --- /dev/null +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py @@ -0,0 +1,528 @@ +""" +Tests for OpenAPI to MCP generator, focusing on security and edge cases. + +This test suite ensures that: +1. Parameter names with invalid Python identifiers are handled safely +2. No exec() is used (security) +3. All edge cases (hyphens, dots, keywords, special chars) work correctly +""" + +import json +import pytest +from unittest.mock import AsyncMock, patch +from typing import Dict, Any + +from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import ( + to_safe_identifier, + create_tool_function, + build_input_schema, + extract_parameters, +) + + +class TestToSafeIdentifier: + """Test the to_safe_identifier function for various edge cases.""" + + def test_hyphen_in_name(self): + """Test parameter names with hyphens.""" + assert to_safe_identifier("repository-id") == "repository_id" + assert to_safe_identifier("user-name") == "user_name" + assert to_safe_identifier("api-key") == "api_key" + + def test_leading_digit(self): + """Test parameter names starting with digits.""" + assert to_safe_identifier("2fa-code") == "_2fa_code" + assert to_safe_identifier("123abc") == "_123abc" + assert to_safe_identifier("0test") == "_0test" + + def test_dots_in_name(self): + """Test parameter names with dots.""" + assert to_safe_identifier("user.name") == "user_name" + assert to_safe_identifier("config.value") == "config_value" + assert to_safe_identifier("api.v2") == "api_v2" + + def test_dollar_sign(self): + """Test parameter names with dollar signs (OData style).""" + assert to_safe_identifier("$filter") == "_filter" + assert to_safe_identifier("$context") == "_context" + assert to_safe_identifier("$select") == "_select" + + def test_python_keywords(self): + """Test Python keywords are handled.""" + assert to_safe_identifier("class") == "class_" + assert to_safe_identifier("from") == "from_" + assert to_safe_identifier("not") == "not_" + assert to_safe_identifier("def") == "def_" + assert to_safe_identifier("import") == "import_" + + def test_special_characters(self): + """Test various special characters.""" + assert to_safe_identifier("user@domain") == "user_domain" + assert to_safe_identifier("test#hash") == "test_hash" + assert to_safe_identifier("path/to/resource") == "path_to_resource" + assert to_safe_identifier("param+value") == "param_value" + + def test_multiple_special_chars(self): + """Test names with multiple special characters.""" + assert to_safe_identifier("user-name.email@domain") == "user_name_email_domain" + assert to_safe_identifier("$filter.value") == "_filter_value" + + def test_already_valid_identifier(self): + """Test that valid identifiers remain unchanged (except keywords).""" + assert to_safe_identifier("valid_name") == "valid_name" + assert to_safe_identifier("validName123") == "validName123" + assert to_safe_identifier("_private") == "_private" + + def test_empty_string(self): + """Test empty string handling.""" + assert to_safe_identifier("") == "_empty_" + + def test_only_special_chars(self): + """Test names that are only special characters.""" + result = to_safe_identifier("---") + assert result.startswith("_") + assert len(result) > 0 + + def test_collision_handling(self): + """Test that similar names produce different safe identifiers.""" + # These should produce different results + name1 = to_safe_identifier("user-name") + name2 = to_safe_identifier("user_name") + # They might be the same after sanitization, which is acceptable + # The important thing is they're both valid identifiers + + +class TestCreateToolFunction: + """Test create_tool_function with various parameter name edge cases.""" + + @pytest.mark.asyncio + async def test_hyphenated_path_parameter(self): + """Test function with hyphenated path parameter (e.g., repository-id).""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/repos/{repository-id}", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + # Should not raise SyntaxError + assert callable(func) + assert func.__name__ == "tool_function" + + # Test calling with original parameter name + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = '{"id": "123"}' + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"repository-id": "test-repo"}) + assert result == '{"id": "123"}' + + # Verify URL was constructed correctly + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert "repository-id" in str(call_args[0][0]) or "test-repo" in str(call_args[0][0]) + + @pytest.mark.asyncio + async def test_leading_digit_parameter(self): + """Test function with parameter starting with digit (e.g., 2fa-code).""" + operation = { + "parameters": [ + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/verify", + method="post", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "verified" + mock_client.return_value.__aenter__.return_value.post = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"2fa-code": "123456"}) + assert result == "verified" + + # Verify query parameter was included + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + assert call_args[1]["params"]["2fa-code"] == "123456" + + @pytest.mark.asyncio + async def test_dot_in_parameter_name(self): + """Test function with dot in parameter name (e.g., user.name).""" + operation = { + "parameters": [ + { + "name": "user.name", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/search", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "found" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"user.name": "john.doe"}) + assert result == "found" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["user.name"] == "john.doe" + + @pytest.mark.asyncio + async def test_dollar_sign_parameter(self): + """Test function with dollar sign parameter (OData style, e.g., $filter).""" + operation = { + "parameters": [ + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/entities", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "[]" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"$filter": "name eq 'test'"}) + assert result == "[]" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["$filter"] == "name eq 'test'" + + @pytest.mark.asyncio + async def test_python_keyword_parameter(self): + """Test function with Python keyword as parameter name (e.g., class).""" + operation = { + "parameters": [ + { + "name": "class", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/items", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "items" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"class": "premium"}) + assert result == "items" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["class"] == "premium" + + @pytest.mark.asyncio + async def test_multiple_problematic_parameters(self): + """Test function with multiple problematic parameter names.""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + ] + } + + func = create_tool_function( + path="/repos/{repository-id}", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "success" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func( + **{ + "repository-id": "test-repo", + "2fa-code": "123", + "$filter": "active", + } + ) + assert result == "success" + + @pytest.mark.asyncio + async def test_request_body_parameter(self): + """Test function with request body parameter.""" + operation = { + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"name": {"type": "string"}}, + } + } + }, + } + } + + func = create_tool_function( + path="/create", + method="post", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "created" + mock_client.return_value.__aenter__.return_value.post = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"body": {"name": "test"}}) + assert result == "created" + + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + assert call_args[1]["json"] == {"name": "test"} + + @pytest.mark.asyncio + async def test_no_parameters(self): + """Test function with no parameters.""" + operation = {} + + func = create_tool_function( + path="/health", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "ok" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func() + assert result == "ok" + + @pytest.mark.asyncio + async def test_all_http_methods(self): + """Test all supported HTTP methods.""" + methods = ["get", "post", "put", "delete", "patch"] + + for method in methods: + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/repos/{repository-id}", + method=method, + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "success" + + client_method = getattr( + mock_client.return_value.__aenter__.return_value, method + ) + client_method.return_value = mock_response + client_method = AsyncMock(return_value=mock_response) + setattr( + mock_client.return_value.__aenter__.return_value, + method, + client_method, + ) + + result = await func(**{"repository-id": "test"}) + assert result == "success" + + def test_no_exec_usage(self): + """Verify that create_tool_function does not use exec().""" + import ast + import inspect + + # Get the source code of create_tool_function + source = inspect.getsource(create_tool_function) + + # Parse the AST + tree = ast.parse(source) + + # Check for exec() calls + exec_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name) and node.func.id == "exec": + exec_calls.append(node) + + # Should have no exec() calls + assert len(exec_calls) == 0, "create_tool_function should not use exec()" + + +class TestBuildInputSchema: + """Test that build_input_schema preserves original parameter names.""" + + def test_original_parameter_names_preserved(self): + """Test that original parameter names are preserved in input schema.""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + ] + } + + schema = build_input_schema(operation) + + # Original names should be in the schema + assert "repository-id" in schema["properties"] + assert "2fa-code" in schema["properties"] + assert "$filter" in schema["properties"] + + # Required should include original names + assert "repository-id" in schema["required"] + + +class TestExtractParameters: + """Test parameter extraction from OpenAPI operations.""" + + def test_extract_path_query_body_params(self): + """Test extraction of different parameter types.""" + operation = { + "parameters": [ + {"name": "repo-id", "in": "path"}, + {"name": "filter", "in": "query"}, + {"name": "data", "in": "body"}, + ], + "requestBody": { + "content": {"application/json": {"schema": {"type": "object"}}} + }, + } + + path_params, query_params, body_params = extract_parameters(operation) + + assert "repo-id" in path_params + assert "filter" in query_params + assert "data" in body_params + assert "body" in body_params # From requestBody + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) +