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 <[email protected]>
This commit is contained in:
Eduard van Valkenburg
2026-07-09 16:54:48 +00:00
committed by GitHub
co-authored by Copilot
parent 008d40acb4
commit dd026f3413
2 changed files with 87 additions and 4 deletions
@@ -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:
@@ -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",