Spaces:
Running
Running
Download tests/test_fastmcp_openapi_security.py from Leon4gr45/openoperator: direct link, hf CLI and curl.
- Browser
- Download file 2.04 kB
-
https://huggingface.co/spaces/Leon4gr45/openoperator/resolve/main/tests/test_fastmcp_openapi_security.py
- Command line
-
hf download hf://spaces/Leon4gr45/openoperator/tests/test_fastmcp_openapi_security.py
-
curl -L -o test_fastmcp_openapi_security.py https://huggingface.co/spaces/Leon4gr45/openoperator/resolve/main/tests/test_fastmcp_openapi_security.py
2.04 kB
| 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"}}, | |
| } | |
| } | |
| }, | |
| } | |
| 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") | |