sadpandajoe commented on code in PR #45554:
URL: https://github.com/apache/superset/pull/45554#discussion_r4383703155
##########
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:
Making `transport` a required argument of `from_configuration` also breaks
the factory example in `docs/developer_docs/semantic-layer-cache-clear.md` (the
"Provider SDK contract" section), which still shows the signature without it. A
provider author copying that snippet gets `TypeError: unexpected keyword
argument 'transport'` when the host constructs the layer. Should that example
be updated alongside this signature change?
##########
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:
A redirect `Location` such as `https://attacker.example\@provider.example/`
passes this same-origin check, because `urlsplit()` reads the host as
`provider.example` (everything before the last `@` is treated as userinfo).
urllib3 ends the authority at the backslash and connects to `attacker.example`.
The loop then repeats the request with the original `headers`, so the
provider's `Authorization`/API-key header would be sent to the other host, and
`PinnedHTTPAdapter` still allows it because it is a public address. Could the
redirect target be compared using the same parser that `requests` uses to
connect (for example `urllib3.util.parse_url`), or rejected when it contains a
backslash or userinfo?
##########
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:
This test only captures records at `logging.ERROR`, so the `backend` case
can't catch a regression in the separate `except SemanticBackendError` branch
of `execute_semantic_query`. If that branch started logging the cause with
`exc_info` at DEBUG, `private-cause` would reach the server logs and this test
would still pass; the DEBUG-level tests only exercise `log_provider_failure`
and the metadata fallback. Could this use `caplog.at_level(logging.DEBUG,
logger="superset.semantic_layers")` so the `not in caplog.text` and `exc_info
is None` assertions cover every level?
--
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]
