From 43e8594b2348296bfaf4ba11e44ed331e60bd383 Mon Sep 17 00:00:00 2001 From: Tanmaya Panda Date: Mon, 3 Aug 2026 11:49:07 +0530 Subject: [PATCH] fix: enforce trusted-endpoint validation for all authentication methods (CWE-918) validate_endpoint() only invoked well_known_kusto_endpoints.validate_trusted_endpoint() when the token provider was a CloudInfoTokenProvider. BasicTokenProvider and CallbackTokenProvider derive from TokenProviderBase directly, so the flows built by with_aad_user_token_authentication, with_aad_application_token_authentication, with_token_provider and with_async_token_provider skipped host validation entirely while still sending 'Authorization: Bearer ' to whatever host the connection string named. The Java SDK (ClientImpl.validateEndpointAsync) validates unconditionally, so Python was the outlier. Removing the gate alone would have introduced a second problem: resolving the login endpoint calls CloudSettings.get_cloud_info_for_cluster, which issues a GET to the cluster itself. Validating after that call means an untrusted connection string still drives a request to an arbitrary host, leaving an SSRF primitive, and it would newly require the metadata endpoint on hosts that are trusted explicitly via add_trusted_hosts. The login endpoint is therefore now resolved lazily, and only once the hostname is known to appear in at least one cloud's allow list. Hosts that no login endpoint could ever make trusted are rejected without any network call, and hosts trusted via add_trusted_hosts or an override policy no longer need cloud metadata at all. Passing a plain string is still supported. Known limitation, pre-existing and unchanged in scope: the async client calls the synchronous validate_endpoint, so metadata resolution still blocks the event loop for allow-listed hosts. That needs an async CloudSettings path and is left as a follow-up. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 41fe4674-b54d-43ae-9bd2-6769c81623e4 --- .../azure/kusto/data/client_base.py | 25 +++--- .../kusto/data/kusto_trusted_endpoints.py | 22 +++-- .../tests/test_endpoint_validation.py | 89 +++++++++++++++++++ 3 files changed, 120 insertions(+), 16 deletions(-) create mode 100644 azure-kusto-data/tests/test_endpoint_validation.py diff --git a/azure-kusto-data/azure/kusto/data/client_base.py b/azure-kusto-data/azure/kusto/data/client_base.py index 418ecba1e..d9a2ccc1b 100644 --- a/azure-kusto-data/azure/kusto/data/client_base.py +++ b/azure-kusto-data/azure/kusto/data/client_base.py @@ -9,7 +9,6 @@ from requests import Response, Session from azure.kusto.data._cloud_settings import CloudSettings -from azure.kusto.data._token_providers import CloudInfoTokenProvider from .client_details import ClientDetails from .client_request_properties import ClientRequestProperties from .exceptions import KustoServiceError, KustoThrottlingError, KustoApiError @@ -78,16 +77,20 @@ def set_proxy(self, proxy_url: str): def validate_endpoint(self): if not self._endpoint_validated and self._aad_helper is not None: - if isinstance(self._aad_helper.token_provider, CloudInfoTokenProvider): - endpoint = CloudSettings.get_cloud_info_for_cluster( - self._kusto_cluster, - self._aad_helper.token_provider._proxy_dict, - self._session if isinstance(self._session, Session) else None, - ).login_endpoint - well_known_kusto_endpoints.validate_trusted_endpoint( - self._kusto_cluster, - endpoint, - ) + # Trusted-endpoint validation must run for every authentication method. Gating it on the + # token provider type let token-based and callback-based flows send the Authorization + # header to an arbitrary host named in the connection string. + # The login endpoint is resolved lazily because doing so contacts the cluster itself. + well_known_kusto_endpoints.validate_trusted_endpoint( + self._kusto_cluster, + lambda: ( + CloudSettings.get_cloud_info_for_cluster( + self._kusto_cluster, + self._aad_helper.token_provider._proxy_dict, + self._session if isinstance(self._session, Session) else None, + ).login_endpoint + ), + ) self._endpoint_validated = True @staticmethod diff --git a/azure-kusto-data/azure/kusto/data/kusto_trusted_endpoints.py b/azure-kusto-data/azure/kusto/data/kusto_trusted_endpoints.py index dd5e4bd27..d36adf7d7 100644 --- a/azure-kusto-data/azure/kusto/data/kusto_trusted_endpoints.py +++ b/azure-kusto-data/azure/kusto/data/kusto_trusted_endpoints.py @@ -1,5 +1,5 @@ import copy -from typing import List, Dict +from typing import Callable, List, Dict, Union from urllib.parse import urlparse from azure.kusto.data.helpers import get_string_tail_lower_case @@ -73,18 +73,30 @@ def add_trusted_hosts(self, rules, replace): self._additional_matcher = create_fast_suffix_matcher_from_existing(rules, None if replace else self._additional_matcher) - def validate_trusted_endpoint(self, endpoint: str, login_endpoint: str): + def validate_trusted_endpoint(self, endpoint: str, login_endpoint: Union[str, Callable[[], str]]): + """Validates that the endpoint is trusted. + + `login_endpoint` may be a callable, in which case it is only resolved if the built-in + per-cloud allow lists actually have to be consulted. Resolving it may require contacting + the cluster itself, so callers should pass a callable to avoid reaching out to hosts that + can never be trusted. + """ hostname = urlparse(endpoint).hostname self.validate_hostname_is_trusted(hostname if hostname is not None else endpoint, login_endpoint) - def validate_hostname_is_trusted(self, hostname: str, login_endpoint: str): + def validate_hostname_is_trusted(self, hostname: str, login_endpoint: Union[str, Callable[[], str]]): if _is_local_address(hostname): return if self._override_matcher is not None: if self._override_matcher(hostname): return - else: - matcher = self._matchers.get(login_endpoint.lower()) + elif any(matcher.is_match(hostname) for matcher in self._matchers.values()): + # Only resolve the login endpoint once the hostname is known to appear in at least one + # cloud's allow list. Resolving it may contact the cluster, so doing it first would let + # an untrusted connection string drive a request to an arbitrary host before it is + # rejected. + resolved_login_endpoint = login_endpoint() if callable(login_endpoint) else login_endpoint + matcher = self._matchers.get(resolved_login_endpoint.lower()) if matcher is not None and matcher.is_match(hostname): return diff --git a/azure-kusto-data/tests/test_endpoint_validation.py b/azure-kusto-data/tests/test_endpoint_validation.py new file mode 100644 index 000000000..2c6e2a8db --- /dev/null +++ b/azure-kusto-data/tests/test_endpoint_validation.py @@ -0,0 +1,89 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License +"""Trusted-endpoint validation must be enforced for every authentication method. + +Token-based and callback-based authentication used to bypass validation entirely, which let a +connection string send its Authorization header to an arbitrary host. +""" + +import asyncio +from unittest.mock import patch + +import pytest + +from azure.kusto.data import KustoClient, KustoConnectionStringBuilder +from azure.kusto.data.exceptions import KustoClientInvalidConnectionStringException +from azure.kusto.data.kusto_trusted_endpoints import MatchRule, well_known_kusto_endpoints + +UNTRUSTED_HOST = "https://kusto.attacker.example.com" +TRUSTED_HOST = "https://somecluster.kusto.windows.net" +TOKEN = "a token that must never leave the machine" + + +def _bypassing_auth_kcsbs(cluster: str): + """Connection strings whose token providers do not derive from CloudInfoTokenProvider.""" + return { + "user_token": KustoConnectionStringBuilder.with_aad_user_token_authentication(cluster, TOKEN), + "application_token": KustoConnectionStringBuilder.with_aad_application_token_authentication(cluster, TOKEN), + "token_provider": KustoConnectionStringBuilder.with_token_provider(cluster, lambda: TOKEN), + "async_token_provider": KustoConnectionStringBuilder.with_async_token_provider(cluster, lambda: asyncio.sleep(0, result=TOKEN)), + } + + +@pytest.fixture(params=["user_token", "application_token", "token_provider", "async_token_provider"]) +def bypassing_auth_name(request): + return request.param + + +class TestEndpointValidation: + def test_untrusted_host_is_rejected_for_token_based_auth(self, bypassing_auth_name): + kcsb = _bypassing_auth_kcsbs(UNTRUSTED_HOST)[bypassing_auth_name] + with KustoClient(kcsb) as client: + with pytest.raises(KustoClientInvalidConnectionStringException): + client.execute_query("PythonTest", "Deft") + + def test_untrusted_host_is_rejected_before_any_network_call(self, bypassing_auth_name): + """Validation must not contact the untrusted host, otherwise it is an SSRF primitive.""" + kcsb = _bypassing_auth_kcsbs(UNTRUSTED_HOST)[bypassing_auth_name] + with patch("requests.get") as mock_get, patch("requests.Session.get") as mock_session_get, patch("requests.Session.post") as mock_post: + with KustoClient(kcsb) as client: + with pytest.raises(KustoClientInvalidConnectionStringException): + client.execute_query("PythonTest", "Deft") + assert not mock_get.called + assert not mock_session_get.called + assert not mock_post.called + + def test_explicitly_trusted_host_needs_no_cloud_metadata(self): + """Hosts trusted via add_trusted_hosts must not require the metadata endpoint.""" + try: + well_known_kusto_endpoints.add_trusted_hosts([MatchRule("kusto.attacker.example.com", True)], False) + resolved = [] + + def resolver(): + resolved.append(True) + return "https://login.microsoftonline.com" + + well_known_kusto_endpoints.validate_trusted_endpoint(UNTRUSTED_HOST, resolver) + assert not resolved + finally: + well_known_kusto_endpoints.add_trusted_hosts(None, True) + + def test_login_endpoint_resolved_only_for_allow_listed_hosts(self): + resolved = [] + + def resolver(): + resolved.append(True) + return "https://login.microsoftonline.com" + + well_known_kusto_endpoints.validate_trusted_endpoint(TRUSTED_HOST, resolver) + assert resolved + + resolved.clear() + with pytest.raises(KustoClientInvalidConnectionStringException): + well_known_kusto_endpoints.validate_trusted_endpoint(UNTRUSTED_HOST, resolver) + assert not resolved + + def test_plain_string_login_endpoint_still_supported(self): + well_known_kusto_endpoints.validate_trusted_endpoint(TRUSTED_HOST, "https://login.microsoftonline.com") + with pytest.raises(KustoClientInvalidConnectionStringException): + well_known_kusto_endpoints.validate_trusted_endpoint(UNTRUSTED_HOST, "https://login.microsoftonline.com")