feat(semantic-layer): inject a bounded provider HTTP transport

UPDATING.md

@@ -24,6 +24,16 @@ assists people when migrating to a new version.
 
 ## Next
 
+- **Semantic-layer SDK transport argument.** Provider implementations must accept
+  the required keyword-only `transport: ProviderTransport` in `from_configuration`,
+  `get_configuration_schema`, and `get_runtime_schema`. Superset supplies it even
+  for static schema discovery. Store it on the layer and pass it to views and HTTP
+  helpers; do not construct HTTP clients for provider endpoints. Every request now
+  requires an absolute monotonic deadline and a response byte cap. The default
+  host policy allows public HTTPS destinations only; the provider-specific
+  `SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS = True` setting also allows HTTP and internal
+  destinations. Both policies ignore environment proxies and preserve budgets and
+  TLS hostname verification. See the [provider transport guide](docs/developer_docs/extensions/semantic-provider-transport.md).
 - Semantic-view filters with an empty IN or NOT IN selection no longer fail with a
   query validation error. They follow SQL semantics where the filter sits: an
   empty IN matches no rows (grouped charts return no rows; an ungrouped total

docs/developer_docs/extensions/extension-points/semantic-layers.md

@@ -37,6 +37,7 @@ from typing import Any
 from superset_core.semantic_layers.context import SemanticCredentials, SemanticExecutionContext
 from superset_core.semantic_layers.decorators import semantic_layer
 from superset_core.semantic_layers.layer import SemanticLayer
+from superset_core.semantic_layers.transport import ProviderTransport
 
 from my_extension.config import MyConfig
 from my_extension.view import MySemanticView
@@ -52,26 +53,33 @@ class MySemanticLayer(SemanticLayer[MyConfig, MySemanticView]):
 
     def __init__(
         self, configuration: MyConfig, *, cache_token: str,
-        context: SemanticExecutionContext,
+        context: SemanticExecutionContext, transport: ProviderTransport,
     ) -> None:
         self.configuration: MyConfig = configuration
         self.cache_token: str = cache_token
         self.context: SemanticExecutionContext = context
+        self.transport: ProviderTransport = transport
 
     @classmethod
     def from_configuration(
         cls, configuration: dict[str, Any], *, cache_token: str,
-        context: SemanticExecutionContext, credentials: SemanticCredentials | None = None,
+        context: SemanticExecutionContext, transport: ProviderTransport,
+        credentials: SemanticCredentials | None = None,
     ) -> "MySemanticLayer":
         config: MyConfig = MyConfig.model_validate(configuration)
-        return cls(config, cache_token=cache_token, context=context)
+        return cls(config, cache_token=cache_token, context=context, transport=transport)
 
     @classmethod
-    def get_configuration_schema(cls, configuration=None) -> dict:
+    def get_configuration_schema(
+        cls, configuration: MyConfig | None = None, *, transport: ProviderTransport,
+    ) -> dict[str, Any]:
         return MyConfig.model_json_schema()
 
     @classmethod
-    def get_runtime_schema(cls, configuration=None, runtime_data=None) -> dict:
+    def get_runtime_schema(
+        cls, configuration: MyConfig, runtime_data: dict[str, Any] | None = None,
+        *, transport: ProviderTransport,
+    ) -> dict[str, Any]:
         return {"type": "object", "properties": {}}
 
     def get_semantic_views(self, runtime_configuration: dict) -> set[MySemanticView]:
@@ -96,7 +104,8 @@ Store the parsed configuration instance on the writable `self.configuration`
 attribute, using the provider's configuration type. Assigning a raw dictionary is
 a type error by design; a read-only `@property` is not supported.
 The host reads this attribute to request the runtime schema.
-`get_configuration_schema()` and `get_configuration_schema(None)` must
+`get_configuration_schema(transport=transport)` and
+`get_configuration_schema(None, transport=transport)` must
 return a static schema without I/O or a constructed provider: the host calls
 this form before credentials exist and when identifying secret fields.
 
@@ -131,10 +140,18 @@ Import decorated provider modules **inside** a test or fixture that requests
 
 ```python
 def test_configuration(semantic_layer_registration: None) -> None:
+    from unittest.mock import create_autospec, MagicMock
+
+    from superset_core.semantic_layers.transport import ProviderTransport
+
     from my_extension.layer import MySemanticLayer
 
-    schema: dict[str, object] = MySemanticLayer.get_configuration_schema()
+    transport: MagicMock = create_autospec(ProviderTransport, instance=True)
+    schema: dict[str, object] = MySemanticLayer.get_configuration_schema(
+        transport=transport,
+    )
     assert schema["type"] == "object"
+    transport.request.assert_not_called()
 ```
 
 The fixture substitutes only the SDK registration decorator and restores it after
@@ -150,6 +167,14 @@ explicitly refused, not claimed as a full provider implementation.
 See [Testing semantic providers](../../semantic-layer-provider-testing.md) for the
 reusable SDK conformance suite and its provider-owned fixtures.
 
+## HTTP transport
+
+Provider construction and both schema-discovery entry points also receive a required
+keyword-only `transport: ProviderTransport`. It is a separate host capability, not
+a field of the execution context or a component of binding identity. See the
+[provider transport guide](../semantic-provider-transport.md) for budgets, destination
+policy, errors, and a test fake.
+
 ## Execution context
 
 The unreleased SDK uses one `SemanticLayer.from_configuration` factory with

docs/developer_docs/extensions/semantic-provider-transport.md

@@ -0,0 +1,160 @@
+---
+title: Semantic provider HTTP transport
+---
+
+<!--
+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.
+-->
+
+# Provider HTTP transport
+
+Superset supplies a `ProviderTransport` to semantic-layer providers. This keeps
+HTTP policy in the host and lets an extension use the same SDK on different hosts.
+Providers import only `superset_core` for this capability, never host HTTP helpers.
+
+## Accept the capability
+
+The following `SemanticLayer` entry points require keyword-only `transport=`:
+
+- `from_configuration(configuration, *, cache_token, context, transport, credentials=None)`
+- `get_configuration_schema(configuration=None, *, transport)`
+- `get_runtime_schema(configuration, runtime_data=None, *, transport)`
+
+Keep it on the constructed layer and pass it to views and discovery helpers.
+It is separate from `SemanticExecutionContext` and does not change memo or cache
+identity. Static schema discovery still receives the capability; providers must
+not perform remote discovery when no validated configuration was supplied.
+
+## Set budgets for a logical operation
+
+```python
+import time
+
+from superset_core.semantic_layers.transport import (
+    ProviderResponse, ProviderTransport, RequestBudget,
+)
+
+
+def fetch_catalog(transport: ProviderTransport, endpoint: str) -> ProviderResponse:
+    budget: RequestBudget = RequestBudget(
+        deadline=time.monotonic() + 10.0, max_bytes=4 * 1024 * 1024,
+    )
+    return transport.request("Example", "GET", endpoint, budget=budget)
+```
+
+`deadline` is an absolute monotonic time on this process, not a wall-clock timestamp.
+Reuse that deadline across pages, redirects and polling attempts. Each call's
+`max_bytes` limits its response body. An optional `consume(size)` callback receives
+chunk sizes before the per-call cap check, so a provider can account for a shared
+operation budget. Use a provider-specific exception when that shared budget is
+exhausted.
+
+DNS, connect, TLS, headers and body reads all share the deadline. A timed-out or
+truncated response is never returned as success. Resolution uses at most eight
+shared workers and at most eight outstanding submissions. At capacity, new
+lookups fail immediately with `TIMEOUT`. A caller stops waiting at its deadline,
+but its submission retains capacity until the worker completes it. An OS resolver
+already running cannot be cancelled; this bound does not recover stalled workers. Connected sockets are interrupted by one shared,
+lazily started scheduler. Neither component starts at module import.
+
+## Responses and errors
+
+`ProviderResponse` owns complete bounded body bytes, case-insensitive immutable
+headers, the status, and the final URL. Its `json()` uses the standard library;
+JSON decoding errors remain provider errors, not transport errors.
+
+The host returns only 2xx responses. `ProviderRequestError.failure` identifies
+`INVALID_URL`, `REFUSED`, `UNREACHABLE`, `TIMEOUT`, `TOO_LARGE`,
+`UNSUPPORTED_ENCODING`, or `HTTP_STATUS`. For `HTTP_STATUS`, `status` contains the
+status code. These are transport faults, not deliberate query refusals. Never
+log URLs, credentials, query values or response payloads when handling them.
+Errors and logs include the provider display name. Pass a constant label such as
+`MetricFlow` or `dbt Semantic Layer`, never endpoint or configuration values.
+Labels must start with an ASCII letter or digit and contain only ASCII letters,
+digits, spaces, underscores or hyphens, up to 64 characters. Other labels fall
+back to `Provider`; valid labels have trailing spaces removed. Non-string labels
+also use that fallback, without converting the supplied object to a string.
+
+The transport sends `Accept-Encoding: identity` and refuses other response
+encodings. Non-2xx bodies are not read. Logs contain only the sanitized provider name, status and failure
+category. GET may follow at most three redirects within the same scheme, hostname
+and effective port. Other redirects fail with `HTTP_STATUS` without reading their
+bodies. Redirects do not reset the deadline. No automatic retries are made.
+
+## Destination policy
+
+URLs containing userinfo (`https://user:password@host/`) are refused as
+`INVALID_URL`, both on the initial request and on redirects. Supply credentials
+through the request headers instead.
+
+By default, only HTTPS endpoints whose DNS answers are all public are allowed.
+The host applies its shared `is_safe_ip` policy, including special IPv6 ranges.
+It connects directly to a checked numeric address without resolving again, while
+preserving the original hostname for TLS SNI and certificate verification.
+
+Operators can set `SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS = True` in
+`superset_config.py` to allow private addresses and HTTP for these providers.
+This setting does not change OAuth or webhook policy. It never disables budgets,
+TLS verification or redirect restrictions. Both settings ignore environment proxy
+variables and never discover credentials from `.netrc`. Configure a reachable
+provider endpoint rather than relying on a proxy.
+In allow-internal mode, `REQUESTS_CA_BUNDLE` (or `CURL_CA_BUNDLE` as fallback)
+can supply the trusted CA bundle for private TLS endpoints. Certificate and
+hostname verification remain enabled. The default policy ignores both variables.
+There is no per-user opt-out; trusted operators own the setting.
+
+## Use a fake in provider tests
+
+A fake must honor the same 2xx-or-error contract. A minimal fixed-response fake:
+
+```python
+import time
+from collections.abc import Callable, Mapping
+from typing import Any
+
+from superset_core.semantic_layers.transport import (
+    ProviderRequestError, ProviderRequestFailure, ProviderResponse,
+    ProviderTransport, RequestBudget,
+)
+
+
+class FakeTransport(ProviderTransport):
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str]] = []
+
+    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:
+        if time.monotonic() >= budget.deadline:
+            raise ProviderRequestError(ProviderRequestFailure.TIMEOUT)
+        self.calls.append((method, url))
+        body: bytes = b'{"metrics": []}'
+        if consume is not None:
+            consume(len(body))
+        if len(body) > budget.max_bytes:
+            raise ProviderRequestError(ProviderRequestFailure.TOO_LARGE)
+        if time.monotonic() >= budget.deadline:
+            raise ProviderRequestError(ProviderRequestFailure.TIMEOUT)
+        return ProviderResponse(200, {"Content-Type": "application/json"}, body, url)
+```
+
+Add fixture-specific failures by raising `ProviderRequestError`, rather than
+returning a non-2xx `ProviderResponse`. Keep credentials out of recorded calls.

docs/developer_docs/semantic-layer-cache-clear.md

@@ -52,7 +52,8 @@ Providers implement one factory, invoked before eager discovery:
 @classmethod
 def from_configuration(
     cls, configuration: dict[str, Any], *, cache_token: str,
-    context: SemanticExecutionContext, credentials: SemanticCredentials | None = None,
+    context: SemanticExecutionContext, transport: ProviderTransport,
+    credentials: SemanticCredentials | None = None,
 ) -> SemanticLayer:
     ...
 ```

docs/sidebarTutorials.js

@@ -87,6 +87,7 @@ const sidebars = {
             'extensions/extension-points/semantic-layers',
           ],
         },
+        'extensions/semantic-provider-transport',
         'extensions/development',
         'extensions/deployment',
         'extensions/mcp',

superset-core/src/superset_core/semantic_layers/__init__.py

@@ -16,3 +16,19 @@
 # under the License.
 
 """Semantic layer contracts for extension authors."""
+
+from superset_core.semantic_layers.transport import (
+    ProviderRequestError,
+    ProviderRequestFailure,
+    ProviderResponse,
+    ProviderTransport,
+    RequestBudget,
+)
+
+__all__: list[str] = [
+    "ProviderRequestError",
+    "ProviderRequestFailure",
+    "ProviderResponse",
+    "ProviderTransport",
+    "RequestBudget",
+]

superset-core/src/superset_core/semantic_layers/layer.py

@@ -29,6 +29,7 @@
     SemanticCredentials,
     SemanticExecutionContext,
 )
+from superset_core.semantic_layers.transport import ProviderTransport
 from superset_core.semantic_layers.view import SemanticView
 
 ConfigT = TypeVar("ConfigT", bound=BaseModel)
@@ -122,10 +123,14 @@ def from_configuration(
         *,
         cache_token: str,
         context: SemanticExecutionContext,
+        transport: ProviderTransport,
         credentials: SemanticCredentials | None = None,
     ) -> SemanticLayer[ConfigT, SemanticViewT]:
         """Bind this layer and its views to one host execution identity.
 
+        The host supplies transport separately from execution identity.
+        Use it for provider HTTP; no implicit or unbounded client is available.
+
         Under CONNECTION, results and catalogs must not depend on the context.
         Store the supplied context and cache_token; partition provider caches on
         cache_token, never secrets. Non-connection credentials arrive exclusively
@@ -140,11 +145,14 @@ def from_configuration(
     def get_configuration_schema(
         cls,
         configuration: ConfigT | None = None,
+        *,
+        transport: ProviderTransport,
     ) -> dict[str, Any]:
         """
         Get the JSON schema for the configuration needed to add the semantic layer.
 
-        Calling with no arguments (or ``None``) must return a static schema
+        Omitting configuration (or passing ``None``), with the required
+        ``transport`` keyword, must return a static schema
         without I/O: no connections, network discovery or credential lookup.
         The host uses this form before configuration exists, including for
         secret masking. It must not require a constructed provider instance.
@@ -174,6 +182,8 @@ def get_runtime_schema(
         cls,
         configuration: ConfigT,
         runtime_data: dict[str, Any] | None = None,
+        *,
+        transport: ProviderTransport,
     ) -> dict[str, Any]:
         """
         Get the JSON schema for the runtime parameters needed to load semantic views.

superset-core/src/superset_core/semantic_layers/testing.py

@@ -25,6 +25,7 @@
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import replace
 from typing import NoReturn, TypeVar
+from unittest.mock import Mock
 
 import pyarrow as pa
 import pytest
@@ -37,6 +38,7 @@
     SemanticResultCompletenessError,
 )
 from superset_core.semantic_layers.layer import SemanticLayer
+from superset_core.semantic_layers.transport import ProviderTransport
 from superset_core.semantic_layers.types import (
     AdhocExpression,
     AggregationType,
@@ -485,6 +487,8 @@ def no_io(*args: object, **kwargs: object) -> NoReturn:
         attempted.append("schema I/O")
         raise AssertionError("Configuration schema attempted I/O")
 
+    transport: Mock = Mock(spec=ProviderTransport)
+    transport.request.side_effect = no_io
     patch: pytest.MonkeyPatch
     with pytest.MonkeyPatch.context() as patch:
         patch.setattr(sys, "meta_path", [NoHost(), *sys.meta_path])
@@ -501,6 +505,8 @@ def no_io(*args: object, **kwargs: object) -> NoReturn:
         patch.setattr(socket, "getaddrinfo", no_io)
         patch.setattr(socket.socket, "connect", no_io)
         patch.setattr(socket.socket, "connect_ex", no_io)
-        assert isinstance(provider.get_configuration_schema(), dict)
-        assert isinstance(provider.get_configuration_schema(None), dict)
+        assert isinstance(provider.get_configuration_schema(transport=transport), dict)
+        assert isinstance(
+            provider.get_configuration_schema(None, transport=transport), dict
+        )
     assert not attempted, attempted