diff --git a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py index ed8258fce36a..40bd728d9aa8 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py +++ b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py @@ -2,7 +2,7 @@ import ipaddress from typing import Any -from urllib.parse import ParseResult, urlparse +from urllib.parse import ParseResult, unquote, urlparse from pydantic import Field @@ -113,12 +113,53 @@ def _effective_port(url: ParseResult) -> int | None: return None +def _normalize_path(path: str, decode_percent_encoded: bool = False) -> str: + """Normalize a URL path the same way httpx does before routing. + + Applies RFC 3986 §5.2.4 remove_dot_segments. Empty segments are preserved. + Paths that would climb above the root are clamped at ``/``. + """ + if decode_percent_encoded: + path = unquote(path) + + # remove_dot_segments (RFC 3986 §5.2.4), preserving empty segments like httpx. + segments = path.split("/") + resolved: list[str] = [] + for segment in segments: + if segment == ".": + continue + if segment == "..": + if resolved and resolved[-1] != "": + resolved.pop() + # else: clamp at root (leading empty from absolute path stays) + continue + resolved.append(segment) + if segments[-1] in (".", ".."): + resolved.append("") + # Absolute paths keep a leading empty segment from the initial "/". + if not resolved: + return "/" + normalized = "/".join(resolved) + if path.startswith("/") and not normalized.startswith("/"): + normalized = "/" + normalized + return normalized or "/" + + def _matches_path_prefix(url_path: str, base_path: str) -> bool: url_path = url_path or "/" base_path = base_path or "/" - if url_path.lower() == base_path.lower(): - return True + for decode_percent_encoded in (False, True): + normalized_url_path = _normalize_path(url_path, decode_percent_encoded) + normalized_base_path = _normalize_path(base_path, decode_percent_encoded) + + if normalized_url_path.lower() == normalized_base_path.lower(): + continue + + base_path_with_slash = ( + normalized_base_path if normalized_base_path.endswith("/") else f"{normalized_base_path}/" + ) + if not normalized_url_path.lower().startswith(base_path_with_slash.lower()): + return False - base_path_with_slash = base_path if base_path.endswith("/") else f"{base_path}/" - return url_path.lower().startswith(base_path_with_slash.lower()) + return True diff --git a/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py b/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py index e6d6f900e6ec..8063e755cf41 100644 --- a/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py +++ b/python/tests/unit/connectors/openapi_plugin/test_server_url_validator.py @@ -241,3 +241,76 @@ async def test_validate_server_url_returns_no_addresses_for_allowed_base_url(): async def test_validate_server_url_returns_no_addresses_when_private_access_is_allowed(): options = ServerUrlValidationOptions(allow_private_network_access=True) assert await validate_server_url("https://internal.example/api", options) == [] + + +@pytest.mark.parametrize( + "url", + [ + "https://api.example.com/api/../admin", + "https://api.example.com/api/%2e%2e/admin", + "https://api.example.com/api/%2E%2E/admin", + "https://api.example.com/api/foo/../../admin", + ], +) +async def test_validate_server_url_rejects_path_prefix_escape(url): + """Dot-segment escapes must not satisfy an /api allowlist prefix.""" + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"]) + + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url(url, options) + + +@pytest.mark.parametrize( + "url", + [ + "https://api.example.com/api/./users", + "https://api.example.com/api/users", + "https://api.example.com/api/", + "https://api.example.com/api", + ], +) +async def test_validate_server_url_allows_normalized_paths_under_prefix(url): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"]) + + assert await validate_server_url(url, options) == [] + + +async def test_validate_server_url_still_rejects_path_that_only_shares_prefix_chars(): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"]) + + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url("https://api.example.com/apiusers", options) + + +async def test_validate_server_url_rejects_path_that_only_matches_after_decoding(): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"]) + + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url("https://api.example.com/admin/%2e%2e/api", options) + + +@pytest.mark.parametrize( + "url", + [ + "https://api.example.com/api/.", + "https://api.example.com/api/users/..", + ], +) +async def test_validate_server_url_preserves_trailing_slash_after_normalization(url): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api/"]) + + assert await validate_server_url(url, options) == [] + + +@pytest.mark.parametrize( + "allowed_base_url", + [ + "https://api.example.com/api/.", + "https://api.example.com/api/", + ], +) +async def test_validate_server_url_trailing_slash_allowlist_does_not_allow_bare_path(allowed_base_url): + options = ServerUrlValidationOptions(allowed_base_urls=[allowed_base_url]) + + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url("https://api.example.com/api", options)