diff --git a/src/fastmcp/utilities/openapi/director.py b/src/fastmcp/utilities/openapi/director.py index 58e941ba7..537fdffdc 100644 --- a/src/fastmcp/utilities/openapi/director.py +++ b/src/fastmcp/utilities/openapi/director.py @@ -1,7 +1,7 @@ """Request director using openapi-core for stateless HTTP request building.""" from typing import Any -from urllib.parse import urljoin +from urllib.parse import quote, urljoin import httpx from jsonschema_path import SchemaPath @@ -205,12 +205,14 @@ class RequestDirector: Returns: Complete URL with path parameters substituted """ - # Substitute path parameters + # Substitute path parameters with URL-encoding to prevent + # path traversal and SSRF via crafted parameter values url_path = path_template for param_name, param_value in path_params.items(): placeholder = f"{{{param_name}}}" if placeholder in url_path: - url_path = url_path.replace(placeholder, str(param_value)) + safe_value = quote(str(param_value), safe="").replace(".", "%2E") + url_path = url_path.replace(placeholder, safe_value) # Combine with base URL return urljoin(base_url.rstrip("/") + "/", url_path.lstrip("/")) diff --git a/tests/utilities/openapi/test_director.py b/tests/utilities/openapi/test_director.py index 849c91ef4..edad3eab2 100644 --- a/tests/utilities/openapi/test_director.py +++ b/tests/utilities/openapi/test_director.py @@ -1,5 +1,7 @@ """Unit tests for RequestDirector.""" +from urllib.parse import unquote + import pytest from jsonschema_path import SchemaPath @@ -460,3 +462,130 @@ class TestRequestDirectorIntegration: assert request.method == "GET" assert str(request.url).startswith("https://api.example.com/search") + + +class TestPathTraversalPrevention: + """Test that path parameter values are URL-encoded to prevent SSRF/path traversal.""" + + @pytest.fixture + def director(self, basic_openapi_30_spec): + spec = SchemaPath.from_dict(basic_openapi_30_spec) + return RequestDirector(spec) + + @pytest.fixture + def path_route(self): + return HTTPRoute( + path="/api/v1/users/{id}/profile", + method="GET", + operation_id="get_user_profile", + parameters=[ + ParameterInfo( + name="id", + location="path", + required=True, + schema={"type": "string"}, + ) + ], + flat_param_schema={ + "type": "object", + "properties": {"id": {"type": "string"}}, + "required": ["id"], + }, + parameter_map={"id": {"location": "path", "openapi_name": "id"}}, + ) + + @pytest.mark.parametrize( + "malicious_id", + [ + "../../../admin/delete-all?", + "../../secret", + "../../../etc/passwd", + "foo/../../../admin", + "..%2F..%2Fadmin", + "..%2f..%2fadmin", + ], + ) + def test_path_traversal_encoded(self, director, path_route, malicious_id: str): + request = director.build( + path_route, {"id": malicious_id}, "https://api.example.com" + ) + url = str(request.url) + assert "/admin" not in url + assert "/secret" not in url + assert "/etc/passwd" not in url + assert url.startswith("https://api.example.com/api/v1/users/") + + def test_slash_in_param_is_encoded(self, director, path_route): + request = director.build(path_route, {"id": "a/b"}, "https://api.example.com") + url = str(request.url) + assert "/a/b/" not in url + assert "a%2Fb" in url + + def test_dot_dot_slash_is_encoded(self, director, path_route): + request = director.build( + path_route, {"id": "../admin"}, "https://api.example.com" + ) + url = str(request.url) + assert "%2E%2E%2Fadmin" in url or "%2e%2e%2fadmin" in url + assert url.startswith("https://api.example.com/api/v1/users/") + + def test_question_mark_encoded(self, director, path_route): + request = director.build( + path_route, {"id": "foo?bar=baz"}, "https://api.example.com" + ) + url = str(request.url) + assert "foo%3Fbar%3Dbaz" in url or "foo%3fbar%3dbaz" in url + + def test_hash_encoded(self, director, path_route): + request = director.build( + path_route, {"id": "foo#fragment"}, "https://api.example.com" + ) + url = str(request.url) + assert "foo%23fragment" in url + + def test_normal_values_still_work(self, director, path_route): + request = director.build( + path_route, {"id": "user-123"}, "https://api.example.com" + ) + assert ( + str(request.url) == "https://api.example.com/api/v1/users/user-123/profile" + ) + + def test_dotted_values_encode_dots(self, director, path_route): + """Dots are encoded to prevent path normalization by urljoin.""" + request = director.build( + path_route, {"id": "v1.2.3"}, "https://api.example.com" + ) + url = str(request.url) + assert "v1%2E2%2E3" in url + assert url.startswith("https://api.example.com/api/v1/users/") + + def test_numeric_values_still_work(self, director, path_route): + request = director.build(path_route, {"id": 42}, "https://api.example.com") + assert str(request.url) == "https://api.example.com/api/v1/users/42/profile" + + def test_bare_single_dot_encoded(self, director, path_route): + """Bare '.' must be encoded so urljoin doesn't normalize it away.""" + request = director.build(path_route, {"id": "."}, "https://api.example.com") + url = str(request.url) + assert "%2E" in url + assert url.startswith("https://api.example.com/api/v1/users/") + + def test_bare_dotdot_encoded(self, director, path_route): + """Bare '..' must be encoded so urljoin doesn't resolve it as traversal.""" + request = director.build(path_route, {"id": ".."}, "https://api.example.com") + url = str(request.url) + assert "%2E%2E" in url or "%2e%2e" in url + assert url.startswith("https://api.example.com/api/v1/users/") + + def test_double_encoded_traversal(self, director, path_route): + request = director.build( + path_route, + {"id": "..%2F..%2Fadmin"}, + "https://api.example.com", + ) + url = str(request.url) + decoded = unquote(unquote(url)) + # Verify traversal didn't escape the users/ prefix + assert decoded.startswith("https://api.example.com/api/v1/users/") + assert url.startswith("https://api.example.com/api/v1/users/")