From d38b1f5dd54941cff0f8b10351af7fbeace4b472 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 28 Sep 2026 14:04:40 +0200 Subject: [PATCH 1/2] Python: Clarify HttpPlugin defaults --- .../core_plugins/http_plugin.py | 64 +++++++++++-------- .../unit/core_plugins/test_http_plugin.py | 44 +++++++++---- 2 files changed, 66 insertions(+), 42 deletions(-) diff --git a/python/semantic_kernel/core_plugins/http_plugin.py b/python/semantic_kernel/core_plugins/http_plugin.py index c461deccbf8c..e5961707665c 100644 --- a/python/semantic_kernel/core_plugins/http_plugin.py +++ b/python/semantic_kernel/core_plugins/http_plugin.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. import json +import logging from typing import Annotated, Any, ClassVar from urllib.parse import urlparse @@ -10,15 +11,20 @@ from semantic_kernel.functions.kernel_function_decorator import kernel_function from semantic_kernel.kernel_pydantic import KernelBaseModel +logger = logging.getLogger(__name__) + class HttpPlugin(KernelBaseModel): """A plugin that provides HTTP functionality. Usage: + # Without domain restrictions (backward compatible, logs a warning): + kernel.add_plugin(HttpPlugin(), "http") + # With allowed domains (recommended): kernel.add_plugin(HttpPlugin(allowed_domains=["example.com", "api.example.com"]), "http") - # Explicitly allow all domains (opt-in, less secure): + # Explicitly allow all domains, even when allowed_domains is also set: kernel.add_plugin(HttpPlugin(allow_all_domains=True), "http") Examples: @@ -28,44 +34,51 @@ class HttpPlugin(KernelBaseModel): {{http.deleteAsync $url}} Security: - - By default, all requests are blocked unless ``allowed_domains`` is provided - or ``allow_all_domains`` is set to True. + - When ``allowed_domains`` is not provided, requests to any HTTP or HTTPS + domain are allowed for backward compatibility and a warning is logged. - When ``allowed_domains`` is set and ``allow_all_domains`` is False, HTTP - redirects are disabled to prevent redirect-based domain bypass (SSRF). + redirects are disabled to prevent redirect-based domain bypass. - When ``allow_all_domains`` is True, redirects are allowed regardless of whether ``allowed_domains`` is also set. - Only ``http`` and ``https`` URL schemes are permitted. - - Only standard ports (80, 443) are permitted by default. Set ``allowed_ports`` - to permit additional ports. Port validation is skipped when - ``allow_all_domains`` is True. + - When domain restrictions are configured, only standard ports (80, 443) + are permitted by default. Set ``allowed_ports`` to permit additional ports. """ allowed_domains: set[str] | None = None - """Set of allowed domains to send requests to.""" + """Set of allowed domains to send requests to. If None, all domains are allowed.""" allow_all_domains: bool = False - """When True, requests to any domain are allowed. Must be explicitly set.""" + """When True, requests to any domain are allowed, overriding ``allowed_domains``.""" allowed_ports: set[int] | None = None - """Set of ports permitted for outbound requests. Defaults to ``{80, 443}`` when not set. + """Set of ports permitted for outbound requests when domain restrictions are configured. - Ignored when ``allow_all_domains`` is True. Set explicitly to permit non-standard ports - (e.g. ``allowed_ports={443, 8443}``). + Defaults to ``{80, 443}`` when not set. Set explicitly to permit non-standard + ports (e.g. ``allowed_ports={443, 8443}``). """ _ALLOWED_SCHEMES: ClassVar[frozenset[str]] = frozenset({"http", "https"}) _DEFAULT_SCHEME_PORTS: ClassVar[dict[str, int]] = {"http": 80, "https": 443} _DEFAULT_ALLOWED_PORTS: ClassVar[frozenset[int]] = frozenset({80, 443}) + def model_post_init(self, __context: Any) -> None: + """Warn when the plugin is configured without domain restrictions.""" + super().model_post_init(__context) + if self.allowed_domains is None and not self.allow_all_domains: + logger.warning( + "HttpPlugin was created without `allowed_domains`; requests to any HTTP or HTTPS domain are allowed. " + "Set `allowed_domains` to restrict outbound requests." + ) + @property def _allow_redirects(self) -> bool: """Whether HTTP redirects should be followed. - Redirects are only allowed when ``allow_all_domains`` is True. - When domain restrictions are configured, redirects are disabled - to prevent redirect-based SSRF bypass. + Redirects are allowed when domain restrictions are not configured or + ``allow_all_domains`` is True. """ - return self.allow_all_domains + return self.allow_all_domains or self.allowed_domains is None def _is_uri_allowed(self, url: str) -> bool: """Check if the URL's host and scheme are permitted. @@ -94,8 +107,8 @@ def _is_uri_allowed(self, url: str) -> bool: except ValueError: return False - # If allow_all_domains is set, skip the domain and port allow-list checks. - if self.allow_all_domains: + # Preserve unrestricted behavior when no domain allow-list is configured. + if self.allow_all_domains or self.allowed_domains is None: return True # Enforce the port allow-list (deny-by-default to non-standard ports). @@ -105,27 +118,22 @@ def _is_uri_allowed(self, url: str) -> bool: if port not in allowed_ports: return False - # If allowed_domains is set, check against it - if self.allowed_domains is not None: - return host.lower() in {domain.lower() for domain in self.allowed_domains} - - # Default: deny all - return False + return host.lower() in {domain.lower() for domain in self.allowed_domains} def _validate_url(self, url: str) -> None: """Validate the URL before sending a request. Always checks that the URL is non-empty, uses an allowed scheme, and has a - syntactically valid port. When ``allow_all_domains`` is False, additionally - enforces the port and domain allow-lists. + syntactically valid port. When domain restrictions are configured, + additionally enforces the port and domain allow-lists. Args: url: The URL to validate. Raises: FunctionExecutionException: If the URL is empty, uses a disallowed scheme, - has a malformed port, or (unless ``allow_all_domains`` is True) targets - a port or domain that is not allowed. + has a malformed port, or targets a configured port or domain that is + not allowed. """ if not url: raise FunctionExecutionException("url cannot be `None` or empty") diff --git a/python/tests/unit/core_plugins/test_http_plugin.py b/python/tests/unit/core_plugins/test_http_plugin.py index c0c8211b1d78..fb580f2e60f6 100644 --- a/python/tests/unit/core_plugins/test_http_plugin.py +++ b/python/tests/unit/core_plugins/test_http_plugin.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import logging from unittest.mock import patch import pytest @@ -250,25 +251,40 @@ async def test_allowed_domains_exact_subdomain_match(): assert plugin._is_uri_allowed("https://other.example.com/path") is False -# Security regression tests +# Request validation tests -async def test_default_constructor_denies_all(): - """Test that default HttpPlugin() denies all requests (issue 115285).""" - plugin = HttpPlugin() - assert plugin._is_uri_allowed("https://example.com/path") is False - assert plugin._is_uri_allowed("https://any-domain.com/path") is False +def test_default_constructor_allows_domains_and_logs_warning(caplog): + """Test that the default configuration remains unrestricted and logs a warning.""" + with caplog.at_level(logging.WARNING, logger="semantic_kernel.core_plugins.http_plugin"): + plugin = HttpPlugin() + assert plugin._is_uri_allowed("https://example.com/path") is True + assert plugin._is_uri_allowed("https://any-domain.com:8443/path") is True + assert "without `allowed_domains`" in caplog.text + assert "requests to any HTTP or HTTPS domain are allowed" in caplog.text + + +def test_allowed_domains_does_not_log_unrestricted_warning(caplog): + """Test that restricted configurations do not log the unrestricted warning.""" + with caplog.at_level(logging.WARNING, logger="semantic_kernel.core_plugins.http_plugin"): + HttpPlugin(allowed_domains={"example.com"}) + + assert "without `allowed_domains`" not in caplog.text + + +@patch("aiohttp.ClientSession.get") +async def test_default_constructor_allows_requests_and_redirects(mock_get): + """Test that the default configuration preserves request and redirect behavior.""" + mock_get.return_value.__aenter__.return_value.text.return_value = "OK" + mock_get.return_value.__aenter__.return_value.status = 200 -@pytest.mark.parametrize("method", ["get", "post", "put", "delete"]) -async def test_default_constructor_blocks_requests(method): - """Test that default HttpPlugin() blocks all HTTP methods (issue 115285).""" plugin = HttpPlugin() - with pytest.raises(FunctionExecutionException, match="Sending requests to the provided location is not allowed"): - if method in ["post", "put"]: - await getattr(plugin, method)(url="https://example.com/path", body={"key": "value"}) - else: - await getattr(plugin, method)(url="https://example.com/path") + response = await plugin.get("https://example.com/path") + + assert response == "OK" + _, kwargs = mock_get.call_args + assert kwargs["allow_redirects"] is True @patch("aiohttp.ClientSession.get") From e130fbb9356427dcb2c94b88a3677bb23b662cb8 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 28 Sep 2026 14:20:52 +0200 Subject: [PATCH 2/2] Python: Warn on default HttpPlugin configuration --- .../core_plugins/http_plugin.py | 60 ++++++++++--------- .../unit/core_plugins/test_http_plugin.py | 48 ++++++++------- 2 files changed, 59 insertions(+), 49 deletions(-) diff --git a/python/semantic_kernel/core_plugins/http_plugin.py b/python/semantic_kernel/core_plugins/http_plugin.py index e5961707665c..21759fbdf8d2 100644 --- a/python/semantic_kernel/core_plugins/http_plugin.py +++ b/python/semantic_kernel/core_plugins/http_plugin.py @@ -18,13 +18,10 @@ class HttpPlugin(KernelBaseModel): """A plugin that provides HTTP functionality. Usage: - # Without domain restrictions (backward compatible, logs a warning): - kernel.add_plugin(HttpPlugin(), "http") - # With allowed domains (recommended): kernel.add_plugin(HttpPlugin(allowed_domains=["example.com", "api.example.com"]), "http") - # Explicitly allow all domains, even when allowed_domains is also set: + # Explicitly allow all domains (opt-in, less secure): kernel.add_plugin(HttpPlugin(allow_all_domains=True), "http") Examples: @@ -34,28 +31,30 @@ class HttpPlugin(KernelBaseModel): {{http.deleteAsync $url}} Security: - - When ``allowed_domains`` is not provided, requests to any HTTP or HTTPS - domain are allowed for backward compatibility and a warning is logged. + - By default, all requests are blocked unless ``allowed_domains`` is provided + or ``allow_all_domains`` is set to True. A warning is logged when neither + option is configured. - When ``allowed_domains`` is set and ``allow_all_domains`` is False, HTTP - redirects are disabled to prevent redirect-based domain bypass. + redirects are disabled to prevent redirect-based domain bypass (SSRF). - When ``allow_all_domains`` is True, redirects are allowed regardless of whether ``allowed_domains`` is also set. - Only ``http`` and ``https`` URL schemes are permitted. - - When domain restrictions are configured, only standard ports (80, 443) - are permitted by default. Set ``allowed_ports`` to permit additional ports. + - Only standard ports (80, 443) are permitted by default. Set ``allowed_ports`` + to permit additional ports. Port validation is skipped when + ``allow_all_domains`` is True. """ allowed_domains: set[str] | None = None - """Set of allowed domains to send requests to. If None, all domains are allowed.""" + """Set of allowed domains to send requests to.""" allow_all_domains: bool = False - """When True, requests to any domain are allowed, overriding ``allowed_domains``.""" + """When True, requests to any domain are allowed. Must be explicitly set.""" allowed_ports: set[int] | None = None - """Set of ports permitted for outbound requests when domain restrictions are configured. + """Set of ports permitted for outbound requests. Defaults to ``{80, 443}`` when not set. - Defaults to ``{80, 443}`` when not set. Set explicitly to permit non-standard - ports (e.g. ``allowed_ports={443, 8443}``). + Ignored when ``allow_all_domains`` is True. Set explicitly to permit non-standard ports + (e.g. ``allowed_ports={443, 8443}``). """ _ALLOWED_SCHEMES: ClassVar[frozenset[str]] = frozenset({"http", "https"}) @@ -63,22 +62,24 @@ class HttpPlugin(KernelBaseModel): _DEFAULT_ALLOWED_PORTS: ClassVar[frozenset[int]] = frozenset({80, 443}) def model_post_init(self, __context: Any) -> None: - """Warn when the plugin is configured without domain restrictions.""" + """Warn when the default configuration blocks all requests.""" super().model_post_init(__context) if self.allowed_domains is None and not self.allow_all_domains: logger.warning( - "HttpPlugin was created without `allowed_domains`; requests to any HTTP or HTTPS domain are allowed. " - "Set `allowed_domains` to restrict outbound requests." + "HttpPlugin was created without `allowed_domains` and with `allow_all_domains=False`; " + "all HTTP requests will be blocked. Set `allowed_domains` or `allow_all_domains=True` " + "to enable requests." ) @property def _allow_redirects(self) -> bool: """Whether HTTP redirects should be followed. - Redirects are allowed when domain restrictions are not configured or - ``allow_all_domains`` is True. + Redirects are only allowed when ``allow_all_domains`` is True. + When domain restrictions are configured, redirects are disabled + to prevent redirect-based SSRF bypass. """ - return self.allow_all_domains or self.allowed_domains is None + return self.allow_all_domains def _is_uri_allowed(self, url: str) -> bool: """Check if the URL's host and scheme are permitted. @@ -107,8 +108,8 @@ def _is_uri_allowed(self, url: str) -> bool: except ValueError: return False - # Preserve unrestricted behavior when no domain allow-list is configured. - if self.allow_all_domains or self.allowed_domains is None: + # If allow_all_domains is set, skip the domain and port allow-list checks. + if self.allow_all_domains: return True # Enforce the port allow-list (deny-by-default to non-standard ports). @@ -118,22 +119,27 @@ def _is_uri_allowed(self, url: str) -> bool: if port not in allowed_ports: return False - return host.lower() in {domain.lower() for domain in self.allowed_domains} + # If allowed_domains is set, check against it + if self.allowed_domains is not None: + return host.lower() in {domain.lower() for domain in self.allowed_domains} + + # Default: deny all + return False def _validate_url(self, url: str) -> None: """Validate the URL before sending a request. Always checks that the URL is non-empty, uses an allowed scheme, and has a - syntactically valid port. When domain restrictions are configured, - additionally enforces the port and domain allow-lists. + syntactically valid port. When ``allow_all_domains`` is False, additionally + enforces the port and domain allow-lists. Args: url: The URL to validate. Raises: FunctionExecutionException: If the URL is empty, uses a disallowed scheme, - has a malformed port, or targets a configured port or domain that is - not allowed. + has a malformed port, or (unless ``allow_all_domains`` is True) targets + a port or domain that is not allowed. """ if not url: raise FunctionExecutionException("url cannot be `None` or empty") diff --git a/python/tests/unit/core_plugins/test_http_plugin.py b/python/tests/unit/core_plugins/test_http_plugin.py index fb580f2e60f6..a153d1755ac3 100644 --- a/python/tests/unit/core_plugins/test_http_plugin.py +++ b/python/tests/unit/core_plugins/test_http_plugin.py @@ -251,40 +251,44 @@ async def test_allowed_domains_exact_subdomain_match(): assert plugin._is_uri_allowed("https://other.example.com/path") is False -# Request validation tests +# Security regression tests -def test_default_constructor_allows_domains_and_logs_warning(caplog): - """Test that the default configuration remains unrestricted and logs a warning.""" +def test_default_constructor_denies_all_and_logs_warning(caplog): + """Test that default HttpPlugin() denies all requests and logs a warning.""" with caplog.at_level(logging.WARNING, logger="semantic_kernel.core_plugins.http_plugin"): plugin = HttpPlugin() - assert plugin._is_uri_allowed("https://example.com/path") is True - assert plugin._is_uri_allowed("https://any-domain.com:8443/path") is True + assert plugin._is_uri_allowed("https://example.com/path") is False + assert plugin._is_uri_allowed("https://any-domain.com/path") is False assert "without `allowed_domains`" in caplog.text - assert "requests to any HTTP or HTTPS domain are allowed" in caplog.text - - -def test_allowed_domains_does_not_log_unrestricted_warning(caplog): - """Test that restricted configurations do not log the unrestricted warning.""" + assert "all HTTP requests will be blocked" in caplog.text + + +@pytest.mark.parametrize( + "plugin_kwargs", + [ + {"allowed_domains": {"example.com"}}, + {"allow_all_domains": True}, + ], +) +def test_explicit_domain_configuration_does_not_log_default_warning(caplog, plugin_kwargs): + """Test that explicit domain configurations do not log the default warning.""" with caplog.at_level(logging.WARNING, logger="semantic_kernel.core_plugins.http_plugin"): - HttpPlugin(allowed_domains={"example.com"}) + HttpPlugin(**plugin_kwargs) assert "without `allowed_domains`" not in caplog.text -@patch("aiohttp.ClientSession.get") -async def test_default_constructor_allows_requests_and_redirects(mock_get): - """Test that the default configuration preserves request and redirect behavior.""" - mock_get.return_value.__aenter__.return_value.text.return_value = "OK" - mock_get.return_value.__aenter__.return_value.status = 200 - +@pytest.mark.parametrize("method", ["get", "post", "put", "delete"]) +async def test_default_constructor_blocks_requests(method): + """Test that default HttpPlugin() blocks all HTTP methods.""" plugin = HttpPlugin() - response = await plugin.get("https://example.com/path") - - assert response == "OK" - _, kwargs = mock_get.call_args - assert kwargs["allow_redirects"] is True + with pytest.raises(FunctionExecutionException, match="Sending requests to the provided location is not allowed"): + if method in ["post", "put"]: + await getattr(plugin, method)(url="https://example.com/path", body={"key": "value"}) + else: + await getattr(plugin, method)(url="https://example.com/path") @patch("aiohttp.ClientSession.get")