mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-16 10:23:14 +00:00
Merge pull request #18480 from hamzaq453/fix/openapi-parameter-names-security
Fix: Remove exec() usage and handle invalid OpenAPI parameter names
This commit is contained in:
YutaSaito
@@ -7,12 +7,12 @@ from pathlib import PurePosixPath
|
||||
from typing import Any, Dict, Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
|
||||
from litellm.proxy._experimental.mcp_server.tool_registry import (
|
||||
global_mcp_tool_registry,
|
||||
)
|
||||
from litellm.types.llms.custom_http import httpxSpecialProvider
|
||||
|
||||
# Store the base URL and headers globally
|
||||
BASE_URL = ""
|
||||
@@ -42,79 +42,6 @@ def _sanitize_path_parameter_value(param_value: Any, param_name: str) -> str:
|
||||
return quote(value_str, safe="")
|
||||
|
||||
|
||||
async def _invoke_openapi_operation(
|
||||
*,
|
||||
method: str,
|
||||
base_url: str,
|
||||
path: str,
|
||||
headers: Dict[str, str],
|
||||
path_params: list,
|
||||
query_params: list,
|
||||
body_params: list,
|
||||
provided_params: Dict[str, Any],
|
||||
) -> str:
|
||||
"""Execute the OpenAPI operation using provided parameters."""
|
||||
|
||||
url = base_url + path
|
||||
|
||||
# Replace path parameters with sanitized values
|
||||
for param_name in path_params:
|
||||
param_value = provided_params.get(param_name, "")
|
||||
if param_value:
|
||||
try:
|
||||
safe_value = _sanitize_path_parameter_value(param_value, param_name)
|
||||
except ValueError as exc:
|
||||
return "Invalid path parameter: " + str(exc)
|
||||
url = url.replace("{" + param_name + "}", safe_value)
|
||||
|
||||
# Build query params
|
||||
params: Dict[str, Any] = {}
|
||||
for param_name in query_params:
|
||||
param_value = provided_params.get(param_name, "")
|
||||
if param_value:
|
||||
params[param_name] = param_value
|
||||
|
||||
# Build request body
|
||||
json_body = None
|
||||
if body_params:
|
||||
body_value = provided_params.get("body", {})
|
||||
if isinstance(body_value, dict):
|
||||
json_body = body_value
|
||||
elif body_value:
|
||||
import json as json_module
|
||||
|
||||
try:
|
||||
json_body = (
|
||||
json_module.loads(body_value)
|
||||
if isinstance(body_value, str)
|
||||
else {"data": body_value}
|
||||
)
|
||||
except Exception:
|
||||
json_body = {"data": body_value}
|
||||
|
||||
client = get_async_httpx_client(llm_provider=httpxSpecialProvider.MCP)
|
||||
|
||||
method_lower = method.lower()
|
||||
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 f"Unsupported HTTP method: {method}"
|
||||
|
||||
return response.text
|
||||
|
||||
|
||||
def load_openapi_spec(filepath: str) -> Dict[str, Any]:
|
||||
"""Load OpenAPI specification from JSON file."""
|
||||
with open(filepath, "r") as f:
|
||||
@@ -210,55 +137,107 @@ 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 accessed
|
||||
directly via **kwargs, avoiding syntax errors from invalid Python identifiers.
|
||||
|
||||
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 = {}
|
||||
|
||||
path_params, query_params, body_params = extract_parameters(operation)
|
||||
all_params = path_params + query_params + body_params
|
||||
original_method = method.lower()
|
||||
|
||||
# Build function signature dynamically
|
||||
if all_params:
|
||||
params_str = ", ".join(f"{p}: str = ''" for p in all_params)
|
||||
else:
|
||||
params_str = ""
|
||||
async def tool_function(**kwargs: Any) -> str:
|
||||
"""
|
||||
Dynamically generated tool function.
|
||||
|
||||
# Create the function code as a string
|
||||
func_code = f'''
|
||||
async def tool_function({params_str}) -> str:
|
||||
"""Dynamically generated tool function."""
|
||||
provided_params = {{}}
|
||||
for param_name in {all_params}:
|
||||
provided_params[param_name] = locals().get(param_name, "")
|
||||
return await _invoke_openapi_operation(
|
||||
method="{method}",
|
||||
base_url=base_url,
|
||||
path=path,
|
||||
headers=headers,
|
||||
path_params={path_params},
|
||||
query_params={query_params},
|
||||
body_params={body_params},
|
||||
provided_params=provided_params,
|
||||
)
|
||||
'''
|
||||
Accepts keyword arguments where keys are the original OpenAPI parameter names.
|
||||
The function safely handles parameter names that aren't valid Python identifiers
|
||||
by using **kwargs instead of named parameters.
|
||||
"""
|
||||
# Build URL from base_url and path
|
||||
url = base_url + path
|
||||
|
||||
# Execute the function code to create the actual function
|
||||
local_vars = {
|
||||
"headers": headers,
|
||||
"base_url": base_url,
|
||||
"path": path,
|
||||
"method": method,
|
||||
"_invoke_openapi_operation": _invoke_openapi_operation,
|
||||
}
|
||||
exec(func_code, local_vars)
|
||||
# Replace path parameters using original names from OpenAPI spec
|
||||
# Apply path traversal validation and URL encoding
|
||||
for param_name in path_params:
|
||||
param_value = kwargs.get(param_name, "")
|
||||
if param_value:
|
||||
try:
|
||||
# Sanitize and encode path parameter to prevent traversal attacks
|
||||
safe_value = _sanitize_path_parameter_value(param_value, param_name)
|
||||
except ValueError as exc:
|
||||
return "Invalid path parameter: " + str(exc)
|
||||
# Replace {param_name} or {{param_name}} in URL
|
||||
url = url.replace("{" + param_name + "}", safe_value)
|
||||
url = url.replace("{{" + param_name + "}}", safe_value)
|
||||
|
||||
return local_vars["tool_function"]
|
||||
# Build query params using original parameter names
|
||||
params: Dict[str, Any] = {}
|
||||
for param_name in query_params:
|
||||
param_value = kwargs.get(param_name, "")
|
||||
if param_value:
|
||||
# Use original parameter name in query string (as expected by API)
|
||||
params[param_name] = param_value
|
||||
|
||||
# Build request body
|
||||
json_body: Optional[Dict[str, Any]] = None
|
||||
if body_params:
|
||||
# Try "body" first (most common), then check all body param names
|
||||
body_value = kwargs.get("body", {})
|
||||
if not body_value:
|
||||
for param_name in body_params:
|
||||
body_value = kwargs.get(param_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):
|
||||
|
||||
+491
-70
@@ -1,101 +1,522 @@
|
||||
"""Tests for OpenAPI to MCP generator path handling."""
|
||||
"""
|
||||
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
|
||||
4. Path traversal attacks are prevented
|
||||
5. Path parameters are properly URL encoded
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import (
|
||||
create_tool_function,
|
||||
build_input_schema,
|
||||
extract_parameters,
|
||||
)
|
||||
|
||||
|
||||
class _DummyResponse:
|
||||
def __init__(self, text: str = "ok"):
|
||||
self.text = text
|
||||
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"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
class _DummyAsyncClient:
|
||||
"""Minimal async client stub that records requests."""
|
||||
func = create_tool_function(
|
||||
path="/repos/{repository-id}",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
last_instance = None
|
||||
# Should not raise SyntaxError
|
||||
assert callable(func)
|
||||
assert func.__name__ == "tool_function"
|
||||
|
||||
def __init__(self):
|
||||
self.requests = []
|
||||
_DummyAsyncClient.last_instance = self
|
||||
# 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
|
||||
)
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
result = await func(**{"repository-id": "test-repo"})
|
||||
assert result == '{"id": "123"}'
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
# 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]
|
||||
)
|
||||
|
||||
async def get(self, url, params=None, headers=None):
|
||||
self.requests.append(("get", url, params, headers))
|
||||
return _DummyResponse("dummy-response")
|
||||
@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",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_reject_path_traversal_inputs(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.get_async_httpx_client",
|
||||
lambda *_, **__: _DummyAsyncClient(),
|
||||
)
|
||||
_DummyAsyncClient.last_instance = None
|
||||
assert callable(func)
|
||||
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "filename",
|
||||
"in": "path",
|
||||
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,
|
||||
"schema": {"type": "string"},
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
tool_function = create_tool_function(
|
||||
path="/files/{filename}",
|
||||
method="GET",
|
||||
operation=operation,
|
||||
base_url="https://example.com",
|
||||
)
|
||||
func = create_tool_function(
|
||||
path="/create",
|
||||
method="post",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
response = await tool_function(filename="../admin")
|
||||
assert callable(func)
|
||||
|
||||
assert "Invalid path parameter" in response
|
||||
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"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_encode_and_request_safe_path_parameters(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.get_async_httpx_client",
|
||||
lambda *_, **__: _DummyAsyncClient(),
|
||||
)
|
||||
_DummyAsyncClient.last_instance = None
|
||||
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
|
||||
assert call_args[1]["json"] == {"name": "test"}
|
||||
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "filename",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
@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"},
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
tool_function = create_tool_function(
|
||||
path="/files/{filename}",
|
||||
method="GET",
|
||||
operation=operation,
|
||||
base_url="https://example.com",
|
||||
)
|
||||
func = create_tool_function(
|
||||
path="/repos/{repository-id}",
|
||||
method=method,
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
response = await tool_function(filename="report 2024.json")
|
||||
assert callable(func)
|
||||
|
||||
assert response == "dummy-response"
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "success"
|
||||
|
||||
dummy_client = _DummyAsyncClient.last_instance
|
||||
assert dummy_client is not None
|
||||
method, url, params, headers = dummy_client.requests[0]
|
||||
assert method == "get"
|
||||
assert url == "https://example.com/files/report%202024.json"
|
||||
assert params == {}
|
||||
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"])
|
||||
|
||||
|
||||
class TestPathSecurity:
|
||||
"""Test path traversal security and URL encoding."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_reject_path_traversal_inputs(self):
|
||||
"""Test that path traversal attacks (../admin) are rejected."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "filename",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
tool_function = create_tool_function(
|
||||
path="/files/{filename}",
|
||||
method="GET",
|
||||
operation=operation,
|
||||
base_url="https://example.com",
|
||||
)
|
||||
|
||||
response = await tool_function(**{"filename": "../admin"})
|
||||
|
||||
assert "Invalid path parameter" in response
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_encode_and_request_safe_path_parameters(self):
|
||||
"""Test that path parameters are properly URL encoded."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "filename",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
tool_function = create_tool_function(
|
||||
path="/files/{filename}",
|
||||
method="GET",
|
||||
operation=operation,
|
||||
base_url="https://example.com",
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "dummy-response"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
response = await tool_function(**{"filename": "report 2024.json"})
|
||||
|
||||
assert response == "dummy-response"
|
||||
|
||||
# Verify URL was properly encoded
|
||||
call_args = mock_client.return_value.__aenter__.return_value.get.call_args
|
||||
url = call_args[0][0]
|
||||
assert url == "https://example.com/files/report%202024.json"
|
||||
|
||||
Reference in New Issue
Block a user