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 9081d3f2ae2d..f53fdd0ffa38 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,51 @@ 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 (len(resolved) > 1 or resolved[0] != ""): + resolved.pop() + # Protect only the leading root segment; internal empty segments can be popped. + continue + resolved.append(segment) + # 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 == base_path: - 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 == normalized_base_path: + continue + + base_path_with_slash = ( + normalized_base_path if normalized_base_path.endswith("/") else f"{normalized_base_path}/" + ) + if not normalized_url_path.startswith(base_path_with_slash): + return False - base_path_with_slash = base_path if base_path.endswith("/") else f"{base_path}/" - return url_path.startswith(base_path_with_slash) + return True diff --git a/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py b/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py index 838cf245844c..57c370828f1e 100644 --- a/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py +++ b/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py @@ -387,6 +387,47 @@ async def test_run_operation_enforces_allowed_base_url_path_case_before_authenti runner.http_client.request.assert_not_awaited() +@pytest.mark.parametrize( + ("allowed_base_url", "url", "is_allowed"), + [ + ("https://api.example.com/api", "https://api.example.com/api/.", True), + ("https://api.example.com/api", "https://api.example.com/api/users/..", True), + ("https://api.example.com/api/", "https://api.example.com/api/.", False), + ("https://api.example.com/api/", "https://api.example.com/api/users/..", False), + ("https://api.example.com/api/", "https://api.example.com/api//.", True), + ("https://api.example.com/api//", "https://api.example.com/api//.", False), + ], +) +async def test_run_operation_enforces_allowed_base_url_with_wire_targets(allowed_base_url, url, is_allowed): + runner = OpenApiRunner( + {}, + server_url_validation_options=ServerUrlValidationOptions(allowed_base_urls=[allowed_base_url]), + ) + operation = MagicMock() + operation.method = "GET" + operation.build_headers.return_value = {} + operation.responses = OrderedDict() + runner.build_operation_url = MagicMock(return_value=url) + runner.build_operation_payload = MagicMock(return_value=(None, None)) + response = MagicMock() + response.text = "response text" + runner.http_client = AsyncMock() + runner.http_client.headers = {} + runner.http_client.request = AsyncMock(return_value=response) + runner.auth_callback = AsyncMock(return_value={"Authorization": "******"}) + + if is_allowed: + result = await runner.run_operation(operation) + assert result == "response text" + runner.auth_callback.assert_awaited_once_with() + runner.http_client.request.assert_awaited_once() + else: + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await runner.run_operation(operation) + runner.auth_callback.assert_not_awaited() + runner.http_client.request.assert_not_awaited() + + async def test_run_operation_blocks_disallowed_url_before_request(): runner = OpenApiRunner({}) operation = MagicMock() 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 67e43493e78f..9bfea3792723 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 @@ -3,10 +3,12 @@ import ipaddress import socket +import httpx import pytest from semantic_kernel.connectors.openapi_plugin.server_url_validator import ( ServerUrlValidationOptions, + _normalize_path, validate_server_url, ) from semantic_kernel.exceptions import FunctionExecutionException @@ -256,3 +258,187 @@ 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", + "https://api.example.com/api//../../admin", + "https://api.example.com/api//%2e%2e/%2E%2E/admin", + "https://api.example.com/api//.%2e/%2E./admin", + "https://api.example.com/api///../../../admin", + "https://api.example.com/api//../../..", + "https://api.example.com/api/%2f../../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//../users", + "https://api.example.com/api//%2e%2e/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/..", + "https://api.example.com/api//..", + ], +) +async def test_validate_server_url_rejects_terminal_dot_segments_under_trailing_slash_allowlist(url): + """httpx sends terminal dot segments as bare paths, so an /api/ allowlist must reject them.""" + 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//.", + "https://api.example.com/api//../", + ], +) +async def test_validate_server_url_allows_terminal_dot_segments_normalizing_to_trailing_slash(url): + """httpx sends /api//. and /api//../ as /api/, matching an /api/ allowlist.""" + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api/"]) + + assert await validate_server_url(url, options) == [] + + +@pytest.mark.parametrize( + "url", + [ + "https://api.example.com/api/.", + "https://api.example.com/api/users/..", + "https://api.example.com/api//.", + "https://api.example.com/api//..", + ], +) +async def test_validate_server_url_allows_terminal_dot_segments_under_bare_allowlist(url): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"]) + + assert await validate_server_url(url, options) == [] + + +async def test_validate_server_url_rejects_terminal_dot_segment_under_double_slash_allowlist(): + """httpx sends /api//. as /api/, so an /api// allowlist must reject it.""" + 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/api//.", 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) + + +async def test_validate_server_url_terminal_dot_allowlist_allows_bare_path(): + """An allowlist ending in /api/. normalizes to bare /api in httpx, so it allows bare /api.""" + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api/."]) + + assert await validate_server_url("https://api.example.com/api", options) == [] + + +@pytest.mark.parametrize("decode_percent_encoded", [False, True]) +@pytest.mark.parametrize( + ("path", "expected"), + [ + ("/api//../../admin", "/admin"), + ("/api///../../../admin", "/admin"), + ("/api//..", "/api"), + ("/api//../", "/api/"), + ("/api//.", "/api/"), + ("/api/.", "/api"), + ("/api/users/..", "/api"), + ("/api//users", "/api//users"), + ("/api///users/", "/api///users/"), + ("/api//../../..", "/"), + ("//../..", "/"), + ("/../../admin", "/admin"), + ], +) +def test_normalize_path_pops_internal_empty_segments_but_preserves_root(path, expected, decode_percent_encoded): + assert _normalize_path(path, decode_percent_encoded) == expected + + +@pytest.mark.parametrize( + "path", + [ + "/api/.", + "/api/users/..", + "/api//.", + "/api//..", + "/api//../", + "/api//users", + "/api///users/", + "/api//../../admin", + "/api///../../../admin", + "/api/../admin", + "/api/./users", + "/..", + "/../..", + "//../..", + "/api//../../..", + ], +) +def test_normalize_path_matches_httpx_wire_path(path): + httpx_wire_path = httpx.URL(f"https://api.example.com{path}").raw_path.decode("ascii").partition("?")[0] + assert _normalize_path(path, decode_percent_encoded=False) == httpx_wire_path + + +async def test_validate_server_url_preserves_empty_segment_in_allowlist(): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api//"]) + + assert await validate_server_url("https://api.example.com/api//users", options) == [] + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url("https://api.example.com/api/users", options)