Python: normalize OpenAPI allowlist paths before prefix match

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

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()

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)