Spaces:
Running
Running
File size: 2,037 Bytes
be6c5ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | import sys
from pathlib import Path
import httpx
import pytest
from fastmcp.server.providers.openapi import OpenAPIProvider
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
OPENAPI_SPEC = {
"openapi": "3.1.0",
"info": {"title": "FastMCP security regression", "version": "1.0.0"},
"paths": {
"/api/v1/users/{id}/profile": {
"get": {
"operationId": "get_user_profile",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {"200": {"description": "ok"}},
}
}
},
}
@pytest.mark.asyncio
async def test_openapi_provider_percent_encodes_path_parameters():
captured = {}
async def handler(request: httpx.Request) -> httpx.Response:
captured["path"] = request.url.path
captured["raw_path"] = request.url.raw_path.decode("ascii")
captured["authorization"] = request.headers.get("authorization")
return httpx.Response(200, json={"ok": True})
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(
base_url="http://backend.local/",
headers={"Authorization": "Bearer admin_secret"},
transport=transport,
) as client:
provider = OpenAPIProvider(openapi_spec=OPENAPI_SPEC, client=client)
tool = await provider.get_tool("get_user_profile")
assert tool is not None
result = await tool.run({"id": "../../../admin/delete-all?"})
assert result.structured_content == {"ok": True}
assert captured["authorization"] == "Bearer admin_secret"
assert captured["path"].startswith("/api/v1/users/")
assert captured["raw_path"].startswith("/api/v1/users/%2E%2E%2F")
assert captured["raw_path"].endswith("/profile")
|