mikebridge commented on code in PR #45553: URL: https://github.com/apache/superset/pull/45553#discussion_r4414870131
########## superset/semantic_layers/transport.py: ########## @@ -0,0 +1,421 @@ +# 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 threading +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( Review Comment: Good catch, fixed in b54ff69fb2. The transport now filters `urllib3.connectionpool` records only while a provider request is active. A DEBUG loopback test fails without the filter (the query token shows up in the log) and checks that unrelated concurrent HTTP still logs. ########## tests/unit_tests/utils/test_network.py: ########## @@ -270,3 +270,194 @@ def test_public_embedded_address_does_not_allow_6to4() -> None: assert address.sixtofour == ipaddress.IPv4Address("8.8.8.8") assert is_safe_ip(ipaddress.IPv4Address("8.8.8.8")) assert not is_safe_ip(address) + + +def test_public_resolution_deduplicates_and_rejects_any_unsafe_answer() -> None: + from superset.utils.network import NonPublicAddressError, resolve_public_addresses + + answers: list[tuple[int, int, int, str, tuple[str, int]]] = [ + (2, 1, 6, "", ("8.8.8.8", 443)) + ] * 2 + with patch("superset.utils.network.socket.getaddrinfo", return_value=answers): + assert resolve_public_addresses("fixture.test", 443, timeout=1) == ["8.8.8.8"] + answers.append((2, 1, 6, "", ("127.0.0.1", 443))) + with ( + patch("superset.utils.network.socket.getaddrinfo", return_value=answers), + pytest.raises(NonPublicAddressError), + ): + resolve_public_addresses("fixture.test", 443, timeout=1) + + +def test_pinned_connection_does_not_repeat_dns_lookup() -> None: + from contextvars import Token + + from superset.utils.network import ( + _PinnedHTTPSConnection, + ConnectionDeadline, + PROVIDER_DEADLINE, + ) + + bound: MagicMock = MagicMock() + bound.remaining.return_value = 1 + connection: _PinnedHTTPSConnection = _PinnedHTTPSConnection("fixture.test", 443) + assert connection._dns_host == "fixture.test" + assert callable(connection._new_conn) + with ( + patch( + "superset.utils.network.socket.getaddrinfo", + side_effect=[ + [(2, 1, 6, "", ("8.8.8.8", 443))], + [(2, 1, 6, "", ("127.0.0.1", 443))], + ], + ) as resolve, + patch("superset.utils.network._connect_address") as connect, + ): + token: Token[ConnectionDeadline | None] = PROVIDER_DEADLINE.set(bound) + try: + assert connection._new_conn() is connect.return_value + finally: + PROVIDER_DEADLINE.reset(token) + resolve.assert_called_once() + connect.assert_called_once_with(("8.8.8.8", 443), 1, connection.socket_options) + bound.watch.assert_called_once_with(connect.return_value) + assert ( + connection.host == "fixture.test" + ) # Original hostname remains available to TLS. + + +def test_failed_socket_registration_closes_connected_socket() -> None: + from contextvars import Token + + from urllib3.exceptions import NewConnectionError + + from superset.utils.network import ( + _PinnedHTTPSConnection, + ConnectionDeadline, + PROVIDER_DEADLINE, + ) + + bound: MagicMock = MagicMock() + bound.remaining.return_value = 1 + bound.watch.side_effect = OSError("descriptor limit") + connection: _PinnedHTTPSConnection = _PinnedHTTPSConnection("fixture.test", 443) + with ( + patch("superset.utils.network._resolve_addresses", return_value=["8.8.8.8"]), + patch("superset.utils.network._connect_address") as connect, + ): + token: Token[ConnectionDeadline | None] = PROVIDER_DEADLINE.set(bound) + try: + with pytest.raises(NewConnectionError): + connection._new_conn() + finally: + PROVIDER_DEADLINE.reset(token) + connect.return_value.close.assert_called_once() + + +def test_empty_resolution_fails_without_connecting() -> None: + from superset.utils.network import resolve_public_addresses + + with ( + patch("superset.utils.network.socket.getaddrinfo", return_value=[]), + patch("superset.utils.network._connect_address") as connect, + ): + with pytest.raises(OSError, match="No destination address"): + resolve_public_addresses("fixture.test", 443, timeout=1) + connect.assert_not_called() + + [email protected]("option_fails", [False, True]) +def test_pinned_connect_applies_socket_options_before_connect( Review Comment: Added in b54ff69fb2: a mocked socket factory connects to a public IPv6 address, and the test asserts `AF_INET6` and the numeric connect target. An always-`AF_INET` mutant passes the 67 existing network tests and fails the new one. ########## tests/unit_tests/semantic_layers/transport_test.py: ########## @@ -0,0 +1,1933 @@ +# 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. + +"""Semantic-layer provider HTTP requests reach only allowed destinations.""" + +from __future__ import annotations + +import importlib.util +import ipaddress +import os +import socket +import ssl +import subprocess +import sys +import textwrap +import time +import traceback +from collections.abc import Callable, Iterator +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from http.server import BaseHTTPRequestHandler, HTTPServer +from importlib.machinery import ModuleSpec +from io import BytesIO +from pathlib import Path +from threading import get_ident, Lock, Thread +from types import ModuleType +from typing import Any, cast, TypeAlias +from unittest.mock import Mock, patch +from urllib.parse import parse_qs, urlsplit + +import pytest +import requests +import urllib3.util.connection +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID +from flask import Flask +from superset_core.semantic_layers.transport import ( + ProviderRequestError, + ProviderRequestFailure, + ProviderResponse, + ProviderTransport, + RequestBudget, +) +from urllib3.response import HTTPResponse + +from superset.semantic_layers import transport as semantic_layer_http +from superset.utils import json, network + +# All tests run in their own pytest process; unrelated background traffic is +# explicitly exercised by the reference cases below. + + +def provider_request( + provider: str, method: str, url: str, *, timeout: float, **kwargs: Any +) -> ProviderResponse: + return semantic_layer_http.build_provider_transport().request( + provider, + method, + url, + budget=RequestBudget(time.monotonic() + timeout, 4 * 1024 * 1024), + **kwargs, + ) + + +def provider_body( + provider: str, + method: str, + url: str, + *, + deadline: float, + max_bytes: int, + timeout: tuple[float, float], + **kwargs: Any, +) -> bytes: + return ( + semantic_layer_http.build_provider_transport() + .request( + provider, method, url, budget=RequestBudget(deadline, max_bytes), **kwargs + ) + .body + ) + + [email protected](autouse=True) +def loopback_connections_only(monkeypatch: pytest.MonkeyPatch) -> None: + """Make accidental outbound traffic fail the test before a socket connects.""" + original: Any = socket.socket.connect + + def connect(sock: socket.socket, address: Any) -> None: + assert isinstance(address, tuple) + assert ipaddress.ip_address(address[0]).is_loopback + original(sock, address) + + monkeypatch.setattr(socket.socket, "connect", connect) + + +Route: TypeAlias = tuple[int, dict[str, str], bytes] +OK: Route = (200, {}, b'{"ok": true}') + + +class _Handler(BaseHTTPRequestHandler): + routes: dict[str, Route] = {} + seen: list[str] = [] + + def _respond(self) -> None: + self.seen.append(f"{self.command} {self.path} {self.headers['Host']}") + status: int + headers: dict[str, str] + body: bytes + status, headers, body = self.routes.get(self.path, OK) + self.send_response(status) + for key, value in headers.items(): + self.send_header(key, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + self._respond() + + def do_POST(self) -> None: # noqa: N802 + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self._respond() + + def log_message(self, *args: Any) -> None: + pass + + +@contextmanager +def _server( + routes: dict[str, Route] | None = None, context: ssl.SSLContext | None = None +) -> Iterator[tuple[int, list[str]]]: + seen: list[str] = [] + handler: type[_Handler] = type( + "Handler", (_Handler,), {"routes": routes or {}, "seen": seen} + ) + server: HTTPServer = HTTPServer(("127.0.0.1", 0), handler) + if context is not None: + server.socket = context.wrap_socket(server.socket, server_side=True) + thread: Thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server.server_address[1], seen + finally: + server.shutdown() + server.server_close() + + +@contextmanager +def _dns_answers(hostname: str, answers: list[Any]) -> Iterator[None]: + # Bounded requests resolve on provider-dns workers, so scope DNS by hostname. + original: Any = socket.getaddrinfo + + def resolve(host: str, *args: Any, **kwargs: Any) -> Any: + if host == hostname: + return answers + return original(host, *args, **kwargs) + + with patch.object(socket, "getaddrinfo", side_effect=resolve): + yield + + +def _background_connection() -> None: + """Complete real loopback traffic while a test's network patches are live.""" + with _server() as (port, _), ThreadPoolExecutor(1) as worker: + connection: socket.socket = worker.submit( + urllib3.util.connection.create_connection, ("127.0.0.1", port), timeout=1 + ).result(timeout=2) + connection.close() + + +def _background_request() -> None: + with _server() as (port, seen), ThreadPoolExecutor(1) as worker: + with requests.Session() as session: + session.trust_env = False + response: requests.Response = worker.submit( + session.get, f"http://127.0.0.1:{port}/background";, timeout=1 + ).result(timeout=2) + assert response.status_code == 200 + assert seen == [f"GET /background 127.0.0.1:{port}"] + + +def _certificate(directory: Path, name: str) -> tuple[Path, Path]: + """Write a self-signed certificate for ``name``; return (cert, key) paths.""" + key: ec.EllipticCurvePrivateKey = ec.generate_private_key(ec.SECP256R1()) + subject: x509.Name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)]) + now: datetime = datetime.now(timezone.utc) + certificate: x509.Certificate = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName(name)]), False) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), True) + .sign(key, hashes.SHA256()) + ) + cert_path: Path = directory / f"{name}.crt" + key_path: Path = directory / f"{name}.key" + cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) + key_path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + return cert_path, key_path + + +@contextmanager +def _internal_hosts(allowed: bool) -> Iterator[None]: + app: Flask = Flask(__name__) + app.config.update( + WORKSPACE_NAME="semantic-layer-http-test", + SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS=allowed, + ) + with app.app_context(): + yield + + [email protected]("url", ["http://example.com/api";, "ftp://example.com/x";]) +def test_https_guard_ignores_background_requests(url: str) -> None: + with patch("superset.semantic_layers.transport._DirectSession.request") as send: + _background_request() + with pytest.raises(ProviderRequestError, match="not allowed"): + provider_request("Cube", "GET", url, timeout=1) + send.assert_not_called() + + [email protected]("host", ["127.0.0.1", "localhost"]) +def test_internal_address_is_refused_before_connecting(host: str) -> None: + listener: socket.socket = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen() + listener.setblocking(False) + port: int = listener.getsockname()[1] + try: + with pytest.raises(ProviderRequestError) as raised: + provider_request("Cube", "GET", f"https://{host}:{port}/x";, timeout=2) + with pytest.raises(BlockingIOError): + listener.accept() # Nothing ever connected. + finally: + listener.close() + assert host not in str(raised.value) + assert str(port) not in str(raised.value) + + +def test_any_internal_answer_refuses_the_host() -> None: + answers: list[tuple[Any, ...]] = [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 443)), + ] + with ( + _dns_answers("mixed.example.com", answers), + patch("superset.utils.network._connect_address") as connect, + ): + _background_connection() + with pytest.raises(ProviderRequestError): + provider_request("Cube", "GET", "https://mixed.example.com/x";, timeout=1) + connect.assert_not_called() + + [email protected]("address", ["64:ff9b::a9fe:a9fe", "64:ff9b::a00:5"]) +def test_translated_ipv6_answers_are_checked_as_ipv4(address: str) -> None: + answers: list[tuple[Any, ...]] = [ + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", (address, 443, 0, 0)), + ] + with ( + _dns_answers("nat64.example.com", answers), + patch("superset.utils.network._connect_address") as connect, + ): + _background_connection() + with pytest.raises(ProviderRequestError): + provider_request("Cube", "GET", "https://nat64.example.com/x";, timeout=1) + connect.assert_not_called() + + +def test_checked_connection_ignores_background_connections() -> None: + answers: list[tuple[Any, ...]] = [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)), + ] + real_resolve: Any = socket.getaddrinfo + resolved: list[str] = [] + + def resolve(host: str, *args: Any, **kwargs: Any) -> Any: + # socket.getaddrinfo is process-global: answer and record only this + # test's host, and leave lookups from unrelated threads untouched. + if host != "public.example.com": + return real_resolve(host, *args, **kwargs) + resolved.append(host) + return answers + + with ( + patch.object(network.socket, "getaddrinfo", side_effect=resolve), + patch( + "superset.utils.network._connect_address", + side_effect=OSError("stop"), + ) as connect, + ): + _background_connection() + # Deterministically model an unrelated lookup while the global hook is live. + assert socket.getaddrinfo("localhost", 443) == real_resolve("localhost", 443) + with pytest.raises(ProviderRequestError): + provider_request("Cube", "GET", "https://public.example.com/x";, timeout=1) + _background_connection() + assert resolved == ["public.example.com"] + assert connect.call_args.args[0] == ("93.184.216.34", 443) + + [email protected] +def public_tls( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> tuple[ssl.SSLContext, list[str]]: + """Treat ``*.example.test`` as public, served on 127.0.0.1 with a trusted cert.""" + cert: Path + key: Path + cert, key = _certificate(tmp_path, "public.example.test") + context: ssl.SSLContext = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.minimum_version = ssl.TLSVersion.TLSv1_2 + context.load_cert_chain(cert, key) + resolved: list[str] = [] + real_resolve: Any = socket.getaddrinfo + + owner: int = get_ident() + + def resolve(host: str, *args: Any, **kwargs: Any) -> Any: + # Names also resolve on provider-dns workers; IP-literal connects belong + # to the caller only. Do not record another thread's loopback traffic. + if host.endswith(".example.test") or ( + host == "127.0.0.1" and get_ident() == owner + ): + resolved.append(host) + local: str = "127.0.0.1" if host.endswith(".example.test") else host + return real_resolve(local, *args, **kwargs) + + monkeypatch.setattr(network.socket, "getaddrinfo", resolve) + monkeypatch.setattr(network, "is_safe_ip", lambda address: True) + monkeypatch.setattr(requests.adapters, "DEFAULT_CA_BUNDLE_PATH", str(cert)) + return context, resolved + + +def test_public_tls_fixture_preserves_verification_choice( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + """Trust installation must not override a caller disabling verification.""" + session: semantic_layer_http._DirectSession + with semantic_layer_http._DirectSession() as session: + settings: semantic_layer_http._Settings = session.merge_environment_settings( + "https://public.example.test/";, {}, False, False, None + ) + assert settings["verify"] is False + + +def test_public_tls_ignores_background_loopback_lookups( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + resolved: list[str] = public_tls[1] + assert socket.getaddrinfo("localhost", 443) + with ThreadPoolExecutor(1) as worker: + assert worker.submit(socket.getaddrinfo, "127.0.0.1", 443).result(timeout=2) + assert resolved == [] + assert socket.getaddrinfo("public.example.test", 443) + assert socket.getaddrinfo("public.example.test", 443) + assert resolved == ["public.example.test", "public.example.test"] + + +def test_default_path_verifies_the_configured_hostname( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + context: ssl.SSLContext + resolved: list[str] + context, resolved = public_tls + with _server(context=context) as (port, seen): + response: ProviderResponse = provider_request( + "Cube", "GET", f"https://public.example.test:{port}/meta";, timeout=2 + ) + with pytest.raises(ProviderRequestError, match="could not be reached"): + # The certificate names public.example.test, not this host. + provider_request( + "Cube", "GET", f"https://other.example.test:{port}/meta";, timeout=2 + ) + assert response.json() == {"ok": True} + assert seen == [f"GET /meta public.example.test:{port}"] + # Each hostname is resolved once; urllib3 then connects to the IP literal. + names: list[str] = [host for host in resolved if not host[0].isdigit()] + assert names == ["public.example.test", "other.example.test"] + assert set(resolved) - set(names) == set() # Numeric connects do not resolve again. + + +def test_default_path_rejects_an_untrusted_certificate( + public_tls: tuple[ssl.SSLContext, list[str]], tmp_path: Path +) -> None: + """A matching hostname alone does not establish certificate trust.""" + untrusted: Path = tmp_path / "untrusted" + untrusted.mkdir() + cert: Path + key: Path + cert, key = _certificate(untrusted, "public.example.test") + context: ssl.SSLContext = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.minimum_version = ssl.TLSVersion.TLSv1_2 + context.load_cert_chain(cert, key) + port: int + seen: list[str] + with _server(context=context) as (port, seen): + with pytest.raises(ProviderRequestError, match="could not be reached"): + provider_request( + "fixture", "GET", f"https://public.example.test:{port}/";, timeout=2 + ) + assert seen == [] + + +def test_default_path_follows_same_origin_get_redirects_only( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + context: ssl.SSLContext = public_tls[0] + routes: dict[str, Route] = { + "/accounts": (308, {"Location": "/accounts/"}, b""), + "/away": (302, {"Location": "https://other.example.test/x"}, b""), + } + with _server(routes, context) as (port, seen): + base: str = f"https://public.example.test:{port}"; + response: ProviderResponse = provider_request( + "MetricFlow", "GET", f"{base}/accounts", timeout=2 + ) + with pytest.raises(ProviderRequestError, match="HTTP 302"): + provider_request("MetricFlow", "GET", f"{base}/away", timeout=2) + assert response.json() == {"ok": True} + assert [entry.split()[1] for entry in seen] == ["/accounts", "/accounts/", "/away"] + + [email protected]( + "location,followed", + [ + ("https://evil.example\\@provider.example/";, False), + ("https://evil.example%[email protected]/";, False), + ("https://attacker:[email protected]/x";, False), + ("https://@provider.example/x";, True), + ("https://:[email protected]/x";, False), + ("https://[email protected]/";, False), + ("https://PROVIDER.EXAMPLE/next";, True), + ("https://provider.example./next";, False), + ("https://provider.example:443/next";, True), + ("https://provider.example:8443/next";, False), + ("https://provider.example:0/next";, True), + ("http://provider.example/next";, False), + ("/next", True), + ("/next\t", False), + ("/next\n", False), + ("/next\x00", False), + ("/next\x7f", False), + ("/next path", False), + ], +) +def test_redirect_checks_prepared_destination( + monkeypatch: pytest.MonkeyPatch, location: str, followed: bool +) -> None: + """Exercise Requests preparation before capturing adapter-bound credentials.""" + sent: list[requests.PreparedRequest] = [] + + def send( + adapter: network.PinnedHTTPAdapter, + request: requests.PreparedRequest, + **kwargs: Any, + ) -> requests.Response: + sent.append(request) + response: requests.Response = requests.Response() + response.status_code = 302 if len(sent) == 1 else 200 + assert request.url is not None + response.url = request.url + response.request = request + response.raw = HTTPResponse(body=BytesIO(b"ok"), preload_content=False) + if len(sent) == 1: + response.headers["Location"] = location + return response + + monkeypatch.setattr(network.PinnedHTTPAdapter, "send", send) + if followed: + assert ( + provider_request( + "fixture", + "GET", + "https://provider.example/start";, + timeout=2, + headers={"Authorization": "Bearer private", "X-API-Key": "private"}, + ).status + == 200 + ) + else: + with pytest.raises(ProviderRequestError): + provider_request( + "fixture", + "GET", + "https://provider.example/start";, + timeout=2, + headers={"Authorization": "Bearer private", "X-API-Key": "private"}, + ) + assert len(sent) == (2 if followed else 1) + assert all(urlsplit(request.url).hostname == "provider.example" for request in sent) + assert all(request.headers["X-API-Key"] == "private" for request in sent) + assert all(request.headers["Authorization"] == "Bearer private" for request in sent) + + [email protected]("suffix", ["\\@provider.example/", "/path\n", "/a b"]) +def test_ambiguous_first_hop_is_rejected_before_adapter( + monkeypatch: pytest.MonkeyPatch, suffix: str +) -> None: + """First-hop validation must not accept syntax Requests interprets differently.""" + send: Mock = Mock(side_effect=AssertionError("Invalid URL reached adapter")) + monkeypatch.setattr(network.PinnedHTTPAdapter, "send", send) + with pytest.raises(ProviderRequestError) as raised: + provider_request("fixture", "GET", f"https://evil.example{suffix}";, timeout=2) + assert raised.value.failure == ProviderRequestFailure.INVALID_URL + send.assert_not_called() + + +def test_hostless_prepared_url_is_invalid_before_adapter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Non-HTTP preparation can succeed without yielding a network hostname.""" + send: Mock = Mock(side_effect=AssertionError("Invalid URL reached adapter")) + monkeypatch.setattr(network.PinnedHTTPAdapter, "send", send) + raised: pytest.ExceptionInfo[ProviderRequestError] + with pytest.raises(ProviderRequestError) as raised: + provider_request("fixture", "GET", "mailto:[email protected]";, timeout=2) + assert raised.value.failure == ProviderRequestFailure.INVALID_URL + send.assert_not_called() + + +def test_missing_prepared_url_is_invalid_before_parsing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A missing result from preparation must not be passed to the URL parser.""" + prepare: Mock = Mock(return_value=None) + parse: Mock = Mock(wraps=urlsplit) + send: Mock = Mock(side_effect=AssertionError("Invalid URL reached adapter")) + monkeypatch.setattr(requests.PreparedRequest, "prepare_url", prepare) + monkeypatch.setattr(semantic_layer_http, "urlsplit", parse) + monkeypatch.setattr(network.PinnedHTTPAdapter, "send", send) + raised: pytest.ExceptionInfo[ProviderRequestError] + with pytest.raises(ProviderRequestError) as raised: + provider_request("fixture", "GET", "https://provider.example/";, timeout=2) + assert raised.value.failure == ProviderRequestFailure.INVALID_URL + prepare.assert_called_once_with("https://provider.example/";, None) + parse.assert_not_called() + send.assert_not_called() + + [email protected]("userinfo", ["user:pass", "user", ":pass"]) +def test_first_hop_userinfo_is_refused_before_sending( + monkeypatch: pytest.MonkeyPatch, userinfo: str +) -> None: + """Provider credentials belong in headers, not URL authority.""" + send: Mock = Mock(side_effect=AssertionError("Userinfo reached adapter")) + monkeypatch.setattr(network.PinnedHTTPAdapter, "send", send) + raised: pytest.ExceptionInfo[ProviderRequestError] + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "fixture", "GET", f"https://{userinfo}@provider.example/";, timeout=2 + ) + assert raised.value.failure == ProviderRequestFailure.INVALID_URL + send.assert_not_called() + + +def test_flag_allows_local_http_development_endpoints() -> None: + with _server() as (port, seen), _internal_hosts(True): + response: ProviderResponse = provider_request( + "Cube", "GET", f"http://localhost:{port}/meta";, timeout=2 + ) + assert response.json() == {"ok": True} + assert seen == [f"GET /meta localhost:{port}"] + + +def test_flag_off_keeps_local_endpoints_refused() -> None: + with _server() as (port, seen), _internal_hosts(False): + with pytest.raises(ProviderRequestError) as raised: + provider_request("Cube", "GET", f"https://localhost:{port}/m";, timeout=2) + assert seen == [] + assert str(raised.value) == ("Cube endpoint is not allowed.") + + +def test_post_redirects_are_not_followed() -> None: + routes: dict[str, Route] = {"/graphql": (307, {"Location": "/other"}, b"")} + with _server(routes) as (port, seen), _internal_hosts(True): + with pytest.raises(ProviderRequestError, match=r"^DJ request failed"): + provider_request( + "DJ", "POST", f"http://127.0.0.1:{port}/graphql";, json={}, timeout=2 + ) + assert [entry.split()[:2] for entry in seen] == [["POST", "/graphql"]] + + +def test_redirect_chains_are_bounded() -> None: + routes: dict[str, Route] = {"/loop": (302, {"Location": "/loop"}, b"")} + with _server(routes) as (port, seen), _internal_hosts(True): + with pytest.raises(ProviderRequestError, match="HTTP 302"): + provider_request("DJ", "GET", f"http://127.0.0.1:{port}/loop";, timeout=2) + assert len(seen) == semantic_layer_http.MAX_REDIRECTS + 1 + + +def test_error_message_carries_provider_and_status_only() -> None: + routes: dict[str, Route] = {"/a": (500, {}, b"internal detail")} + with _server(routes) as (port, _), _internal_hosts(True): + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "MetricFlow", "GET", f"http://127.0.0.1:{port}/a";, timeout=2 + ) + assert str(raised.value) == "MetricFlow request failed with HTTP 500." + + +def test_transport_failure_hides_the_destination( + caplog: pytest.LogCaptureFixture, +) -> None: + with patch( + "superset.semantic_layers.transport._DirectSession.request", + side_effect=OSError("refused"), + ): + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "DJ", + "GET", + "https://dj.internal.example.com:8443/x?token=t&password=pass";, + timeout=1, + ) + assert "dj.internal" not in str(raised.value) + assert "8443" not in str(raised.value) + assert "dj.internal" not in caplog.text + assert "unreachable" in caplog.text + assert "pass" not in caplog.text + assert "token" not in caplog.text + + [email protected]("allowed", [False, True]) +def test_failure_logs_never_carry_url_secrets( + caplog: pytest.LogCaptureFixture, allowed: bool +) -> None: + listener: socket.socket = socket.socket() + listener.bind(("127.0.0.1", 0)) + port: int = listener.getsockname()[1] + listener.close() # Nothing listens: the connection is refused or blocked. + url: str = f"https://127.0.0.1:{port}/x?token=SECRET_TOKEN&password=SECRET_PW"; + with _internal_hosts(allowed), caplog.at_level("WARNING"): + with pytest.raises(ProviderRequestError): + provider_request("Cube", "GET", url, timeout=2) + assert caplog.records + assert "SECRET" not in caplog.text + assert "unreachable" in caplog.text or "refused" in caplog.text + + [email protected]( + "location", + [ + "https://public.example.test:NOT_A_PORT/";, + "https://[public.example.test/";, + ], +) +def test_malformed_redirects_raise_the_safe_error( + public_tls: tuple[ssl.SSLContext, list[str]], + caplog: pytest.LogCaptureFixture, + location: str, +) -> None: + routes: dict[str, Route] = {"/a": (302, {"Location": location}, b"")} + with _server(routes, public_tls[0]) as (port, seen), caplog.at_level("WARNING"): + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "Cube", "GET", f"https://public.example.test:{port}/a";, timeout=2 + ) + assert "NOT_A_PORT" not in str(raised.value) + assert "NOT_A_PORT" not in caplog.text + assert len(seen) == 1 + + [email protected]( + "url", ["https://[dj.example.com/x";, "https://dj.example.com:NOT_A_PORT/x";] +) +def test_malformed_endpoint_urls_raise_the_safe_error(url: str) -> None: + with patch("superset.semantic_layers.transport._DirectSession.request") as send: + _background_request() + with pytest.raises(ProviderRequestError) as raised: + provider_request("DJ", "GET", url, timeout=1) + send.assert_not_called() + assert "NOT_A_PORT" not in str(raised.value) + + +# --------------------------------------------------------------------------- # +# Bounded bodies: the same checks, plus a cumulative deadline and byte cap +# --------------------------------------------------------------------------- # +@contextmanager +def _dripping_server( + context: ssl.SSLContext, size: int, interval: float, stall_after: int = -1 +) -> Iterator[int]: + """Serve ``size`` bytes one at a time, ``interval`` seconds apart. + + After ``stall_after`` bytes, stop sending for longer than any test waits. + """ + + class Drip(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self.send_response(200) + self.send_header("Content-Length", str(size)) + self.end_headers() + try: + for sent in range(size): + if sent == stall_after: + time.sleep(5) + self.wfile.write(b"x") + self.wfile.flush() + time.sleep(interval) + except OSError: + pass # The client gave up, as intended. + + def log_message(self, *args: Any) -> None: + pass + + server: HTTPServer = HTTPServer(("127.0.0.1", 0), Drip) + server.socket = context.wrap_socket(server.socket, server_side=True) + thread: Thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server.server_address[1] + finally: + server.shutdown() + server.server_close() + + +def _body( + url: str, *, seconds: float = 2.0, max_bytes: int = 4096, **kwargs: Any +) -> bytes: + return provider_body( + "MetricFlow", + "POST", + url, + deadline=time.monotonic() + seconds, + max_bytes=max_bytes, + timeout=(5.0, 30.0), + json={"query": "{ ok }"}, + **kwargs, + ) + + +def test_bounded_body_is_read_through_the_checked_connection( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + payload: bytes = b'{"data": {"ok": true}}' + consumed: list[int] = [] + with _server({"/q": (200, {}, payload)}, public_tls[0]) as (port, seen): + body: bytes = _body( + f"https://public.example.test:{port}/q";, consume=consumed.append + ) + assert body == payload + assert sum(consumed) == len(payload) + # One lookup of the hostname; the connection then uses the checked address. + assert public_tls[1].count("public.example.test") == 1 + assert seen == [f"POST /q public.example.test:{port}"] + + +def test_bounded_body_is_refused_past_its_byte_cap( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + consumed: list[int] = [] + with _server({"/q": (200, {}, b"x" * 3000)}, public_tls[0]) as (port, _): + with pytest.raises(ProviderRequestError, match="byte budget"): + _body( + f"https://public.example.test:{port}/q";, + max_bytes=1024, + consume=consumed.append, + ) + assert 1024 < sum(consumed) <= 3000 + + +def test_bounded_body_stops_at_the_deadline_while_dripping( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + started: float = time.monotonic() + with _dripping_server(public_tls[0], size=200, interval=0.05) as port: + with pytest.raises(ProviderRequestError) as raised: + _body(f"https://public.example.test:{port}/q";, seconds=0.5) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + elapsed: float = time.monotonic() - started # Before server shutdown. + # Per-read timeouts alone (30 s) would let the drip run for ten seconds. + assert elapsed < 1.5 + + +def test_bounded_read_after_progress_waits_only_for_the_time_left( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + # Bytes arrive for ~0.8 s, then stop. The first read's timeout was the whole + # 1.2 s budget, so only re-arming it per read ends this near the deadline. + started: float = time.monotonic() + with _dripping_server(public_tls[0], 100, 0.05, stall_after=16) as port: + with pytest.raises(ProviderRequestError, match="time budget") as raised: + _body(f"https://public.example.test:{port}/q";, seconds=1.2) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + elapsed: float = time.monotonic() - started # Before server shutdown. + assert elapsed < 1.7 + + [email protected]("allow_internal", [False, True]) +def test_bounded_lookup_waits_no_longer_than_the_deadline( + monkeypatch: pytest.MonkeyPatch, allow_internal: bool +) -> None: + def stuck(*args: Any, **kwargs: Any) -> Any: + time.sleep(1.5) + raise OSError("simulated stuck system resolver") + + real_resolve: Any = socket.getaddrinfo + + def scoped_resolve(host: str, *args: Any, **kwargs: Any) -> Any: + if host == "slow-dns.example.test": + return stuck(host, *args, **kwargs) + return real_resolve(host, *args, **kwargs) + + monkeypatch.setattr(network.socket, "getaddrinfo", scoped_resolve) + started: float = time.monotonic() + with ( + _internal_hosts(allow_internal), + pytest.raises(ProviderRequestError, match="time budget") as raised, + ): + _body("https://slow-dns.example.test/q";, seconds=0.3) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + assert time.monotonic() - started < 1.0 + + +def test_bounded_body_refuses_internal_destinations() -> None: + answers: list[Any] = [(2, 1, 6, "", ("169.254.169.254", 443))] + with ( + _dns_answers("metadata.example.test", answers), + patch("superset.utils.network._connect_address") as connect, + ): + _background_connection() + with pytest.raises(ProviderRequestError, match="not allowed"): + _body("https://metadata.example.test/q";) + connect.assert_not_called() + + +def test_bounded_error_bodies_are_not_read_or_logged( + public_tls: tuple[ssl.SSLContext, list[str]], caplog: pytest.LogCaptureFixture +) -> None: + routes: dict[str, Route] = {"/q": (500, {}, b"internal detail")} + with _server(routes, public_tls[0]) as (port, _), caplog.at_level("WARNING"): + with pytest.raises(ProviderRequestError) as raised: + _body(f"https://public.example.test:{port}/q";) + assert str(raised.value) == "MetricFlow request failed with HTTP 500." + assert "internal detail" not in caplog.text + + +def test_bounded_body_refuses_encoded_responses( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + routes: dict[str, Route] = {"/q": (200, {"Content-Encoding": "gzip"}, b"\x1f\x8b")} + with _server(routes, public_tls[0]) as (port, _): + with pytest.raises(ProviderRequestError, match="encoding is not supported"): + _body(f"https://public.example.test:{port}/q";) + + +def test_bounded_post_redirects_are_not_followed( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + routes: dict[str, Route] = {"/q": (307, {"Location": "/elsewhere"}, b"")} + with _server(routes, public_tls[0]) as (port, seen): + with pytest.raises(ProviderRequestError, match="HTTP 307"): + _body(f"https://public.example.test:{port}/q";) + assert [entry.split()[1] for entry in seen] == ["/q"] + + +def test_bounded_body_refuses_an_expired_deadline() -> None: + with patch("superset.semantic_layers.transport._DirectSession.request") as send: + _background_request() + with pytest.raises(ProviderRequestError, match="time budget") as raised: + _body("https://public.example.test/q";, seconds=-1) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + send.assert_not_called() + + [email protected]("size", [0, 2 * 1024 * 1024]) +def test_bounded_redirect_bodies_are_never_read( + public_tls: tuple[ssl.SSLContext, list[str]], size: int +) -> None: + # Requests pre-reads a redirect's body even with allow_redirects=False. + observed: list[int] = [] + original: Any = vars(requests.Response)["content"].fget + + def content(response: requests.Response) -> bytes: + body: bytes = original(response) + # Response.content is process-global too; ignore other requests. + if response.url == url: + observed.append(len(body)) + return body + + routes: dict[str, Route] = {"/q": (307, {"Location": "/other"}, b"x" * size)} + with _server(routes, public_tls[0]) as (port, seen): + url: str = f"https://public.example.test:{port}/q"; + with patch.object(requests.Response, "content", property(content)): + unrelated: requests.Response = requests.Response() + unrelated.url = "https://unrelated.example.test/"; + unrelated._content = b"unrelated" + assert unrelated.content == b"unrelated" + assert observed == [] + owned: requests.Response = requests.Response() + owned.url = url + owned._content = b"owned" + assert owned.content == b"owned" + assert observed == [len(b"owned")] + observed.clear() + with pytest.raises(ProviderRequestError, match="HTTP 307"): + _body(url, max_bytes=1024) + assert len(seen) == 1 + assert observed == [] + + +@contextmanager +def _framing_server(context: ssl.SSLContext, mode: str) -> Iterator[int]: + """Drip HTTP framing bytes (a header or chunk trailer), not body bytes.""" + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + self.rfile.read(int(self.headers.get("Content-Length", 0))) + prefix: bytes = { + "headers": b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nX-Test: ", + "chunk-trailers": b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked" + b"\r\n\r\n0\r\nX-Trailer: ", + # No length or chunking: the body ends when the connection does. + "close-delimited": b"HTTP/1.0 200 OK\r\n\r\n", + }[mode] + try: + self.wfile.write(prefix) + self.wfile.flush() + for _ in range(14): + self.wfile.write(b"x") + self.wfile.flush() + time.sleep(0.04) + self.wfile.write(b"\r\n\r\n") + self.wfile.flush() + except OSError: + pass + + def log_message(self, *args: Any) -> None: + pass + + server: HTTPServer = HTTPServer(("127.0.0.1", 0), Handler) + server.socket = context.wrap_socket(server.socket, server_side=True) + thread: Thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server.server_address[1] + finally: + server.shutdown() + server.server_close() + thread.join() + + [email protected]("mode", ["headers", "chunk-trailers", "close-delimited"]) [email protected]("allow_internal", [False, True]) +def test_bounded_framing_obeys_the_absolute_deadline( + public_tls: tuple[ssl.SSLContext, list[str]], mode: str, allow_internal: bool +) -> None: + # Allowing internal hosts relaxes destinations only, never the deadline. + with _framing_server(public_tls[0], mode) as port, _internal_hosts(allow_internal): + started: float = time.monotonic() + with pytest.raises(ProviderRequestError) as raised: + _body(f"https://public.example.test:{port}/q";, seconds=0.15) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + elapsed: float = time.monotonic() - started + assert elapsed < 0.40, (mode, allow_internal, elapsed) Review Comment: Reworked in b54ff69fb2. The server now signals once it has sent the selected framing prefix, and only then does the test arm the final deadline window and release the drip. The tight elapsed-time assertion is gone; only a generous 10 s bound on the live I/O remains. Turning off deadline socket watching makes the test fail. ########## UPDATING.md: ########## @@ -24,6 +24,17 @@ assists people when migrating to a new version. ## Next +- **Semantic-layer SDK transport argument.** Provider implementations must accept Review Comment: Updated in b54ff69fb2: the cache-clear guide factory example now includes the required keyword-only `transport: ProviderTransport` argument, 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]
