mikebridge commented on code in PR #45554: URL: https://github.com/apache/superset/pull/45554#discussion_r4391859784
########## superset/semantic_layers/transport.py: ########## @@ -0,0 +1,376 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Host-owned provider HTTP policy with shared deadline enforcement.""" + +from __future__ import annotations + +import heapq +import logging +import os +import socket +import time +from collections.abc import Callable, Mapping +from contextvars import Token +from dataclasses import dataclass, field +from io import BytesIO +from threading import Condition, Lock, Thread +from typing import Any, TYPE_CHECKING +from urllib.parse import SplitResult, urljoin, urlsplit + +import requests +from flask import current_app, has_app_context +from superset_core.semantic_layers.transport import ( + ProviderRequestError, + ProviderRequestFailure as Failure, + ProviderResponse, + ProviderTransport, + RequestBudget, +) +from urllib3.exceptions import HTTPError + +from superset.utils.network import ( + ConnectionDeadline, + NonPublicAddressError, + PinnedHTTPAdapter, + PROVIDER_DEADLINE, +) + +if TYPE_CHECKING: + from requests.sessions import _Settings + +logger: logging.Logger = logging.getLogger(__name__) +MAX_REDIRECTS: int = 3 +_CHUNK_BYTES: int = 65536 + + +class HostProviderTransport(ProviderTransport): + """Fix operator policy at construction; keep budgets and sockets call-local.""" + + def __init__(self, *, allow_internal_hosts: bool = False) -> None: + self._allow_internal: bool = allow_internal_hosts + + def request( + self, + provider: str, + method: str, + url: str, + *, + budget: RequestBudget, + headers: Mapping[str, str] | None = None, + params: Mapping[str, Any] | None = None, + json: Any = None, + consume: Callable[[int], None] | None = None, + ) -> ProviderResponse: + # Do not coerce arbitrary objects while handling a URL-bearing failure. + provider = provider if isinstance(provider, str) else "Provider" + bound: _Deadline = _Deadline(budget.deadline) + token: Token[ConnectionDeadline | None] = PROVIDER_DEADLINE.set(bound) + try: + bound.remaining() # No parsing, resolution or socket work after expiry. + self._validated_url(url) + with _DirectSession(allow_environment_ca=self._allow_internal) as session: + adapter: PinnedHTTPAdapter = PinnedHTTPAdapter( + public_only=not self._allow_internal + ) + session.mount("https://";, adapter) + session.mount("http://";, adapter) + hop: int + for hop in range(MAX_REDIRECTS + 1): + with session.request( + method, + url, + headers={**(headers or {}), "Accept-Encoding": "identity"}, + params=params, + json=json, + timeout=bound.remaining(), + stream=True, + allow_redirects=False, + ) as response: + target: str | None = ( + self._next_redirect(method, response) + if hop < MAX_REDIRECTS + else None + ) + if target is not None: + url, params = target, None + continue # Context closes the unread redirect body. + if not 200 <= response.status_code < 300: + raise ProviderRequestError( + Failure.HTTP_STATUS, status=response.status_code + ) + if ( + response.headers.get("Content-Encoding", "identity").lower() + != "identity" + ): + raise ProviderRequestError(Failure.UNSUPPORTED_ENCODING) + body: bytes = self._read_body( + response, bound, budget.max_bytes, consume + ) + result: ProviderResponse = ProviderResponse( + response.status_code, response.headers, body, response.url + ) + bound.remaining() + return result + raise AssertionError("Redirect loop must return or fail") + except ( + requests.RequestException, + HTTPError, + OSError, + ProviderRequestError, + ) as exc: + failure: ProviderRequestError = self._failure_for(exc, budget.deadline) + error: ProviderRequestError = ProviderRequestError( + failure.failure, status=failure.status, provider=provider + ) + logger.warning( + "%s request failed: category=%s status=%s", + error.provider, + error.failure.value, + error.status, + ) + raise error from None + finally: + PROVIDER_DEADLINE.reset(token) + bound.close() + + def _validated_url(self, url: str) -> SplitResult: + try: + parts: SplitResult = urlsplit(url) + parts.port # noqa: B018 Validate malformed ports before any I/O. + except ValueError: + raise ProviderRequestError(Failure.INVALID_URL) from None + if not parts.hostname: + raise ProviderRequestError(Failure.INVALID_URL) + if parts.scheme not in ( + {"https", "http"} if self._allow_internal else {"https"} + ): + raise ProviderRequestError(Failure.REFUSED) + return parts + + def _next_redirect(self, method: str, response: requests.Response) -> str | None: + if ( + method.upper() != "GET" + or response.status_code not in {301, 302, 303, 307, 308} + or not response.headers.get("Location") + ): + return None + before: SplitResult = self._validated_url(response.url) + try: + target: str = urljoin(response.url, response.headers["Location"]) + after: SplitResult = urlsplit(target) + after.port # noqa: B018 + except ValueError: + raise ProviderRequestError(Failure.INVALID_URL) from None + + def origin(parts: SplitResult) -> tuple[str, str | None, int]: + return ( + parts.scheme, + parts.hostname, + parts.port + if parts.port is not None + else (443 if parts.scheme == "https" else 80), + ) + + return target if origin(before) == origin(after) else None Review Comment: Confirmed on this branch. The transport comes from the parent PR (#45553), whose head now has the fix, and I merged it in 3e31fc8fc8. The new regression test on the prepared destination catches the backslash and userinfo cases before the merge and passes after it. ########## tests/unit_tests/semantic_layers/provider_boundary_test.py: ########## @@ -0,0 +1,462 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Trusted readiness and safe failure handling at real host entry points.""" + +import logging +from collections.abc import Callable +from functools import partial +from typing import cast +from unittest.mock import MagicMock + +import pytest +from flask.testing import FlaskClient +from pydantic import BaseModel +from pytest_mock import MockerFixture +from superset_core.semantic_layers import ( + SemanticBackendError, + SemanticCredentialReadiness, +) +from superset_core.semantic_layers.errors import SemanticQueryRejectedError +from superset_core.semantic_layers.transport import ( + ProviderRequestError, + ProviderRequestFailure, +) +from werkzeug.test import TestResponse + +from superset.constants import PASSWORD_MASK +from superset.exceptions import QueryObjectValidationError +from superset.semantic_layers.exceptions import ( + execute_semantic_query, + normalize_provider_request_errors, + SemanticLayerExecutionError, + SemanticLayerQueryRejectedError, +) +from superset.semantic_layers.masking import credential_readiness, mask_configuration +from superset.utils import json + + +class PreviewConfiguration(BaseModel): + """Deliberately discard unknown fields like a real partial provider model.""" + + account: str | None = None + password: str | None = None + + [email protected]( + "app", [{"FEATURE_FLAGS": {"SEMANTIC_LAYERS": True}}], indirect=True +) [email protected]("writer", [False, True]) [email protected]("masked", [False, True]) +def test_preview_readiness_uses_raw_input_not_client_assertion( + client: FlaskClient, + full_api_access: None, + mocker: MockerFixture, + writer: bool, + masked: bool, +) -> None: + """Normalize before dropped fields, without looking up saved credentials.""" + provider: MagicMock = MagicMock() + provider.configuration_class = PreviewConfiguration + provider.get_configuration_schema.return_value = {"type": "object"} + mocker.patch.dict("superset.semantic_layers.api.registry", {"test": provider}) + mocker.patch( + "superset.semantic_layers.api.security_manager.can_access", + side_effect=lambda permission, view: permission != "can_write" or writer, + ) + lookup: MagicMock = mocker.patch( + "superset.semantic_layers.api.SemanticLayerDAO.find_by_uuid" + ) + response: TestResponse = client.post( + "/api/v1/semantic_layer/schema/configuration", + json={ + "type": "test", + "readiness": "resolved", + "configuration": { + "account": "fixture", + "password": "fixture-password", + # Pydantic drops this unknown subtree; readiness must be computed first. + "nested": [{"password": PASSWORD_MASK if masked else "fixture"}], + "readiness": "resolved", + }, + }, + ) + assert response.status_code == 200 + assert provider.get_configuration_schema.call_args.kwargs["readiness"] is ( + SemanticCredentialReadiness.RESOLVED + if writer and not masked + else SemanticCredentialReadiness.UNRESOLVED + ) + lookup.assert_not_called() + if not writer: + assert provider.get_configuration_schema.call_args.args == (None,) + + [email protected]( + "app", [{"FEATURE_FLAGS": {"SEMANTIC_LAYERS": True}}], indirect=True +) +def test_configuration_fallback_is_explicitly_unresolved( + client: FlaskClient, + full_api_access: None, + mocker: MockerFixture, +) -> None: + """A failed enrichment retries only a static unresolved schema.""" + provider: MagicMock = MagicMock() + provider.configuration_class = PreviewConfiguration + provider.get_configuration_schema.side_effect = [ + ValueError("fixture error"), + {"type": "object"}, + ] + mocker.patch.dict("superset.semantic_layers.api.registry", {"test": provider}) + response: TestResponse = client.post( + "/api/v1/semantic_layer/schema/configuration", + json={ + "type": "test", + "configuration": {"account": "fixture"}, + }, + ) + assert response.status_code == 200 + assert provider.get_configuration_schema.call_args.args == (None,) + assert ( + provider.get_configuration_schema.call_args.kwargs["readiness"] + is SemanticCredentialReadiness.UNRESOLVED + ) + + [email protected]( + "app", [{"FEATURE_FLAGS": {"SEMANTIC_LAYERS": True}}], indirect=True +) [email protected]("masked", [False, True]) +def test_runtime_readiness_uses_saved_raw_configuration( + client: FlaskClient, + full_api_access: None, + mocker: MockerFixture, + masked: bool, +) -> None: + """Client runtime data cannot override readiness of the saved connection.""" + layer: MagicMock = MagicMock(type="test") + layer.configuration = json.dumps( + {"password": PASSWORD_MASK if masked else "fixture"} + ) + layer.implementation.configuration = PreviewConfiguration(account="fixture") + provider: MagicMock = MagicMock() + provider.get_runtime_schema.return_value = {"type": "object"} + mocker.patch( + "superset.semantic_layers.api.SemanticLayerDAO.find_by_uuid", return_value=layer + ) + mocker.patch.dict("superset.semantic_layers.api.registry", {"test": provider}) + response: TestResponse = client.post( + "/api/v1/semantic_layer/00000000-0000-0000-0000-000000000001/schema/runtime", + json={"readiness": "resolved", "runtime_data": {"readiness": "resolved"}}, + ) + assert response.status_code == 200 + layer.raise_for_access.assert_called_once() + assert provider.get_runtime_schema.call_args.kwargs["readiness"] is ( + SemanticCredentialReadiness.UNRESOLVED + if masked + else SemanticCredentialReadiness.RESOLVED + ) + + +def test_masking_schema_is_explicitly_unresolved(mocker: MockerFixture) -> None: + """Masking asks for a static schema with unresolved readiness.""" + provider: MagicMock = MagicMock() + provider.get_configuration_schema.return_value = { + "properties": {"password": {"writeOnly": True}} + } + mocker.patch.dict("superset.semantic_layers.masking.registry", {"test": provider}) + assert mask_configuration("test", {"password": "fixture"}) == { + "password": PASSWORD_MASK + } + assert ( + provider.get_configuration_schema.call_args.kwargs["readiness"] + is SemanticCredentialReadiness.UNRESOLVED + ) + + [email protected]("status", [401, 403, 502]) +def test_transport_keeps_safe_text_and_suppresses_chain( + app_context: None, status: int +) -> None: + """Typed transport diagnostics survive without exporting an exception chain.""" + error: ProviderRequestError = ProviderRequestError( + ProviderRequestFailure.HTTP_STATUS, status=status, provider="MetricFlow" + ) + error.__cause__ = ValueError("private-cause") + dispatcher: MagicMock = MagicMock(side_effect=error) + captured: pytest.ExceptionInfo[QueryObjectValidationError] + with pytest.raises(QueryObjectValidationError) as captured: + execute_semantic_query(dispatcher) + assert str(captured.value) == str(error) + assert captured.value.__cause__ is None + assert captured.value.__suppress_context__ + + [email protected]("kind", ["backend", "unexpected", "refusal"]) +def test_execution_does_not_log_sensitive_cause_or_context( + app_context: None, + caplog: pytest.LogCaptureFixture, + kind: str, +) -> None: + """Typed and unknown provider failures never export diagnostic chains.""" + error: Exception = { + "backend": SemanticBackendError(), + "unexpected": RuntimeError("private-error"), + "refusal": SemanticQueryRejectedError("INVALID_QUERY"), + }[kind] + error.__cause__ = ValueError("private-cause") + error.__context__ = ValueError("private-context") + dispatcher: MagicMock = MagicMock(side_effect=error) + captured: pytest.ExceptionInfo[ + SemanticLayerExecutionError | SemanticLayerQueryRejectedError + ] + with ( + caplog.at_level(logging.ERROR), Review Comment: Good catch. The typed-error test now captures DEBUG on `superset.semantic_layers`. An injected backend traceback passes the old test and fails the revised one. Unknown failures keep their intentional server-side DEBUG diagnostics, covered by the separate existing test. Updated in 3e31fc8fc8. ########## superset-core/src/superset_core/semantic_layers/layer.py: ########## @@ -122,10 +124,14 @@ def from_configuration( *, cache_token: str, context: SemanticExecutionContext, + transport: ProviderTransport, Review Comment: Added the required keyword-only `transport` argument to the `from_configuration` example in 3e31fc8fc8, matching the SDK signature. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
