mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
fix: URL-encode path params to prevent SSRF/path traversal (GHSA-vv7q-7jx5-f767) (#3507)
* fix: URL-encode path params in OpenAPI provider to prevent SSRF/path traversal Co-authored-by: Claude <noreply@anthropic.com> * Exempt too-long from core-category requirement in triage * fix: also encode dots in path params to prevent bare .. traversal * fix: only encode .. (not all dots) to preserve valid dotted values * fix: encode all dots in path params to prevent single-dot normalization * fix: check decoded path stays within prefix in double-encoding test --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
parent
c861862aed
commit
40bdfb6b1d
2 changed files with 134 additions and 3 deletions
|
|
@ -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("/"))
|
||||
|
|
|
|||
|
|
@ -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/")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue