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..9081d3f2ae2d 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py +++ b/python/semantic_kernel/connectors/openapi_plugin/server_url_validator.py @@ -117,8 +117,8 @@ 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(): + if url_path == base_path: return True 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 url_path.startswith(base_path_with_slash) 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 b990ebfb834e..838cf245844c 100644 --- a/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py +++ b/python/tests/unit/connectors/openapi_plugin/test_openapi_runner.py @@ -6,6 +6,10 @@ import pytest from semantic_kernel.connectors.openapi_plugin.models.rest_api_operation import RestApiOperation +from semantic_kernel.connectors.openapi_plugin.models.rest_api_parameter import RestApiParameter +from semantic_kernel.connectors.openapi_plugin.models.rest_api_parameter_location import ( + RestApiParameterLocation, +) from semantic_kernel.connectors.openapi_plugin.models.rest_api_payload import RestApiPayload from semantic_kernel.connectors.openapi_plugin.openapi_manager import OpenApiRunner from semantic_kernel.connectors.openapi_plugin.server_url_validator import ServerUrlValidationOptions @@ -328,6 +332,61 @@ async def mock_request(*args, **kwargs): assert result == "response text" +@pytest.mark.parametrize( + ("method", "route", "is_allowed"), + [ + ("GET", "safe", True), + ("POST", "safe", True), + ("GET", "SAFE", False), + ("POST", "SAFE", False), + ], +) +async def test_run_operation_enforces_allowed_base_url_path_case_before_authentication(method, route, is_allowed): + runner = OpenApiRunner( + {}, + server_url_validation_options=ServerUrlValidationOptions( + allowed_base_urls=["https://api.example.com/api/safe"] + ), + ) + operation = RestApiOperation( + id="caseSensitiveOperation", + method=method, + servers=[{"url": "https://api.example.com"}], + path="/api/{route}", + params=[ + RestApiParameter( + name="route", + type="string", + location=RestApiParameterLocation.PATH, + is_required=True, + ) + ], + ) + 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, {"route": route}) + + assert result == "response text" + runner.auth_callback.assert_awaited_once_with() + runner.http_client.request.assert_awaited_once() + request = runner.http_client.request.await_args.kwargs + assert request["method"] == method + assert request["url"] == "https://api.example.com/api/safe" + assert request["headers"]["Authorization"] == "******" + else: + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await runner.run_operation(operation, {"route": route}) + + 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 e6d6f900e6ec..67e43493e78f 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 @@ -132,6 +132,21 @@ async def test_validate_server_url_rejects_when_allowed_base_urls_do_not_match() await validate_server_url("https://api.example.com/v2/orders", options) +@pytest.mark.parametrize("path", ["/api/safe", "/api/safe/orders"]) +async def test_validate_server_url_allows_matching_allowed_base_url_path_case(path): + options = ServerUrlValidationOptions(allowed_base_urls=["https://API.EXAMPLE.COM:443/api/safe"]) + + await validate_server_url(f"https://api.example.com{path}", options) + + +@pytest.mark.parametrize("path", ["/api/SAFE", "/api/SAFE/orders", "/api/private", "/api/safeX"]) +async def test_validate_server_url_rejects_non_matching_allowed_base_url_path(path): + options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api/safe"]) + + with pytest.raises(FunctionExecutionException, match="allowed base URLs"): + await validate_server_url(f"https://api.example.com{path}", options) + + @pytest.mark.parametrize("url", ["https://10.0.0.5/", "https://[fec0::1]/"]) async def test_validate_server_url_allows_private_network_access_after_scheme_gate(url): options = ServerUrlValidationOptions(allow_private_network_access=True)