From dd026f3413099df9e9caee7374abd632adfd0f06 Mon Sep 17 00:00:00 2001 From: Eduard van Valkenburg Date: Thu, 9 Jul 2026 18:54:48 +0200 Subject: [PATCH] Encode OpenAPI server variable values (#14146) ## Summary - validate enum-constrained OpenAPI server variables before substitution - percent-encode substituted server variable values, including defaults - add regression tests for invalid enum values and reserved-character encoding ## Testing - uv run pytest tests/unit/connectors/openapi_plugin/test_sk_openapi.py -q --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../models/rest_api_operation.py | 12 ++- .../openapi_plugin/test_sk_openapi.py | 79 +++++++++++++++++++ 2 files changed, 87 insertions(+), 4 deletions(-) diff --git a/python/semantic_kernel/connectors/openapi_plugin/models/rest_api_operation.py b/python/semantic_kernel/connectors/openapi_plugin/models/rest_api_operation.py index 4905a7c210..cc137083b4 100644 --- a/python/semantic_kernel/connectors/openapi_plugin/models/rest_api_operation.py +++ b/python/semantic_kernel/connectors/openapi_plugin/models/rest_api_operation.py @@ -287,18 +287,22 @@ class RestApiOperation: # Substitute server variables if available for variable_name, variable_def in server_variables.items(): argument_name = variable_def.get("argument_name", variable_name) + allowed_values = variable_def.get("enum") if argument_name in arguments: - value = arguments[argument_name] - server_url_string = server_url_string.replace(f"{{{variable_name}}}", str(value)) + value = str(arguments[argument_name]) elif "default" in variable_def and variable_def["default"] is not None: # Use the default value if no argument is provided - value = variable_def["default"] - server_url_string = server_url_string.replace(f"{{{variable_name}}}", str(value)) + value = str(variable_def["default"]) else: # Raise an exception if no value is available raise FunctionExecutionException( f"No argument provided for the '{variable_name}' server variable of the operation '{self.id}'." ) + if allowed_values is not None and value not in allowed_values: + raise FunctionExecutionException( + f"Value '{value}' for server variable '{variable_name}' is not one of the allowed values." + ) + server_url_string = server_url_string.replace(f"{{{variable_name}}}", quote(value, safe="")) elif self.server_url: server_url_string = self.server_url elif api_host_url is not None: diff --git a/python/tests/unit/connectors/openapi_plugin/test_sk_openapi.py b/python/tests/unit/connectors/openapi_plugin/test_sk_openapi.py index 2037913145..58f79f9850 100644 --- a/python/tests/unit/connectors/openapi_plugin/test_sk_openapi.py +++ b/python/tests/unit/connectors/openapi_plugin/test_sk_openapi.py @@ -359,6 +359,74 @@ def test_get_server_url_with_servers_coerces_variable_argument_to_string(): assert operation.get_server_url(arguments=arguments) == expected_url +def test_get_server_url_with_server_variable_enum_rejects_invalid_argument(): + operation = RestApiOperation( + id="test", + method="GET", + servers=[ + { + "url": "https://{region}.api.vendor.example/v1", + "variables": {"region": {"default": "us", "enum": ["us", "eu"]}}, + } + ], + path="/resource/{id}", + ) + arguments = {"region": "external.example/"} + with pytest.raises(FunctionExecutionException, match="server variable 'region' is not one of the allowed values"): + operation.get_server_url(arguments=arguments) + + +def test_get_server_url_with_server_variable_enum_allows_valid_argument(): + operation = RestApiOperation( + id="test", + method="GET", + servers=[ + { + "url": "https://{region}.api.vendor.example/v1", + "variables": {"region": {"default": "us", "enum": ["us", "eu"]}}, + } + ], + path="/resource/{id}", + ) + arguments = {"region": "eu"} + expected_url = "https://eu.api.vendor.example/v1/" + assert operation.get_server_url(arguments=arguments) == expected_url + + +def test_get_server_url_with_server_variable_enum_rejects_invalid_default(): + operation = RestApiOperation( + id="test", + method="GET", + servers=[ + { + "url": "https://{region}.api.vendor.example/v1", + "variables": {"region": {"default": "external.example/", "enum": ["us", "eu"]}}, + } + ], + path="/resource/{id}", + ) + with pytest.raises(FunctionExecutionException, match="server variable 'region' is not one of the allowed values"): + operation.get_server_url() + + +def test_get_server_url_with_server_variable_encodes_reserved_characters(): + operation = RestApiOperation( + id="test", + method="GET", + servers=[ + { + "url": "https://{region}.api.vendor.example/v1", + "variables": {"region": {"default": "us"}}, + } + ], + path="/resource/{id}", + ) + arguments = {"region": "external.example/"} + result = operation.get_server_url(arguments=arguments) + assert result == "https://external.example%2F.api.vendor.example/v1/" + assert urlparse(result).hostname != "external.example" + + def test_get_server_url_with_servers_and_default_variable(): operation = RestApiOperation( id="test", @@ -381,6 +449,17 @@ def test_get_server_url_with_servers_coerces_default_variable_to_string(): assert operation.get_server_url() == expected_url +def test_get_server_url_with_servers_encodes_default_variable(): + operation = RestApiOperation( + id="test", + method="GET", + servers=[{"url": "https://example.com/{version}", "variables": {"version": {"default": "v1/beta"}}}], + path="/resource/{id}", + ) + expected_url = "https://example.com/v1%2Fbeta/" + assert operation.get_server_url() == expected_url + + def test_get_server_url_with_override(): operation = RestApiOperation( id="test",