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