mikebridge commented on code in PR #45553:
URL: https://github.com/apache/superset/pull/45553#discussion_r4395342081
##########
superset/utils/network.py:
##########
@@ -260,3 +265,171 @@ def is_host_up(host: str) -> bool:
return False
return output == 0
+
+
+class ConnectionDeadline(Protocol):
+ """A caller-owned deadline; socket code does not choose request policy."""
+
+ def remaining(self) -> float: ...
+ def watch(self, sock: socket.socket) -> None: ...
+
+
+PROVIDER_DEADLINE: ContextVar[ConnectionDeadline | None] = ContextVar(
+ "provider_connection_deadline", default=None
+)
+_RESOLVER: ThreadPoolExecutor | None = None
+_RESOLVER_LOCK: Lock = Lock()
Review Comment:
Confirmed with subprocess tests that import before patching: both startup
paths stalled before the fix. DNS submission now happens outside the native
lock, and the scheduler creates its lock after patching and is published only
after it has started. The preload and fork regressions pass in 33bc325518.
##########
tests/unit_tests/semantic_layers/transport_test.py:
##########
@@ -0,0 +1,1698 @@
+# 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 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 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.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: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)
+
+
+def test_bounded_lookup_and_connect_attempts_share_the_deadline(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ # Two public answers that both stall on connect, after a slow lookup.
+ answers: list[Any] = [
+ (2, 1, 6, "", ("192.0.2.10", 443)),
+ (2, 1, 6, "", ("192.0.2.11", 443)),
+ ]
+
+ def slow_lookup(*args: Any, **kwargs: Any) -> list[Any]:
+ time.sleep(0.2)
+ return answers
+
+ waits: list[float] = []
+
+ def stalled_connect(address: Any, timeout: Any, options: Any) -> Any:
+ waits.append(float(timeout))
+ time.sleep(float(timeout))
+ raise socket.timeout("timed out")
+
+ real_resolve: Any = socket.getaddrinfo
+
+ def scoped_resolve(host: str, *args: Any, **kwargs: Any) -> Any:
+ if host == "stalled.example.test":
+ return slow_lookup(host, *args, **kwargs)
+ return real_resolve(host, *args, **kwargs)
+
+ monkeypatch.setattr(network.socket, "getaddrinfo", scoped_resolve)
+ monkeypatch.setattr(network, "is_safe_ip", lambda address: True)
+ started: float = time.monotonic()
+ with patch("superset.utils.network._connect_address",
side_effect=stalled_connect):
+ with pytest.raises(ProviderRequestError) as raised:
+ _body("https://stalled.example.test/q";, seconds=0.5)
+ assert raised.value.failure == ProviderRequestFailure.TIMEOUT
+ elapsed: float = time.monotonic() - started
+ # Each attempt waits only for what is left; separate full waits take 1.2 s.
+ assert elapsed < 0.75, (elapsed, waits)
+
+
+def test_bounded_internal_requests_ignore_environment_proxies(
+ public_tls: tuple[ssl.SSLContext, list[str]], monkeypatch:
pytest.MonkeyPatch
+) -> None:
+ """A proxy would own the socket, so a bounded request connects directly."""
+ monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:9";)
+ monkeypatch.delenv("NO_PROXY", raising=False)
+ monkeypatch.delenv("no_proxy", raising=False)
+ payload: bytes = b'{"data": {"ok": true}}'
+ with (
+ _server({"/q": (200, {}, payload)}, public_tls[0]) as (
+ port,
+ seen,
+ ),
+ _internal_hosts(True),
+ ):
+ assert _body(f"https://public.example.test:{port}/q";) == payload
+ assert seen == [f"POST /q public.example.test:{port}"]
+
+
+def test_former_unbounded_internal_requests_ignore_environment_proxies(
+ public_tls: tuple[ssl.SSLContext, list[str]], monkeypatch:
pytest.MonkeyPatch
+) -> None:
+ """The SDK has no unbounded mode; Q7 requires direct connections for all
calls."""
+ monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:9";)
+ monkeypatch.delenv("NO_PROXY", raising=False)
+ monkeypatch.delenv("no_proxy", raising=False)
+ with (
+ _server({"/q": (200, {}, b"{}")}, public_tls[0]) as (port, seen),
+ _internal_hosts(True),
+ ):
+ assert (
+ provider_request(
+ "MetricFlow",
+ "GET",
+ f"https://public.example.test:{port}/q";,
+ timeout=2.0,
+ ).json()
+ == {}
+ )
+ assert seen == [f"GET /q public.example.test:{port}"]
+
+
[email protected]("allow_internal", [False, True])
[email protected]("bounded", [False, True])
[email protected]("failure", ["lookup", "empty"])
+def test_dns_failure_is_not_an_address_policy_refusal(
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+ allow_internal: bool,
+ failure: str,
+ bounded: bool,
+) -> None:
+ def resolve(*args: Any, **kwargs: Any) -> list[Any]:
+ if failure == "lookup":
+ raise socket.gaierror(socket.EAI_NONAME, "private DNS detail")
+ return []
+
+ real_resolve: Any = socket.getaddrinfo
+
+ def scoped_resolve(host: str, *args: Any, **kwargs: Any) -> Any:
+ if host == "missing.example.test":
+ return resolve(host, *args, **kwargs)
+ return real_resolve(host, *args, **kwargs)
+
+ monkeypatch.setattr(network.socket, "getaddrinfo", scoped_resolve)
+ monkeypatch.setenv("NO_PROXY", "*")
+ monkeypatch.setenv("no_proxy", "*")
+
+ def call() -> None:
+ if bounded:
+ _body("https://missing.example.test/q";)
+ else:
+ provider_request(
+ "MetricFlow", "POST", "https://missing.example.test/q";,
timeout=1
+ )
+
+ with _internal_hosts(allow_internal), pytest.raises(ProviderRequestError)
as raised:
+ call()
+ assert str(raised.value) == "MetricFlow endpoint could not be reached."
+ assert "unreachable" in caplog.text
+ assert "refused" not in caplog.text
+ assert "private DNS detail" not in caplog.text
+
+
[email protected](
+ "size,cap,success",
+ [(16, 16, True), (17, 16, False), (65537, 65537, True), (65537, 65536,
False)],
+)
+def test_byte_cap_boundaries_and_chunk_accounting(
+ public_tls: tuple[ssl.SSLContext, list[str]], size: int, cap: int,
success: bool
+) -> None:
+ consumed: list[int] = []
+ with _server({"/q": (200, {}, b"x" * size)}, public_tls[0]) as (port, _):
+ if success:
+ assert (
+ len(
+ _body(
+ f"https://public.example.test:{port}/q";,
+ max_bytes=cap,
+ consume=consumed.append,
+ )
+ )
+ == size
+ )
+ else:
+ with pytest.raises(ProviderRequestError) as raised:
+ _body(
+ f"https://public.example.test:{port}/q";,
+ max_bytes=cap,
+ consume=consumed.append,
+ )
+ assert raised.value.failure == ProviderRequestFailure.TOO_LARGE
+ assert sum(consumed) == size
+ assert max(consumed) <= 65536
+
+
[email protected](
+ "provider,expected",
+ [
+ ("UNTRUSTED_LABEL", "UNTRUSTED_LABEL"),
+ ("Cube\nforged=https://secret.test/";, "Provider"),
+ ],
+)
+def test_error_body_is_unread(
+ provider: str,
+ expected: str,
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+) -> None:
+ from unittest.mock import MagicMock
+
+ response: MagicMock = MagicMock(
+ status_code=503, url="https://fixture.test/secret";, headers={}
+ )
+ response.__enter__.return_value = response
+ monkeypatch.setattr(
+ semantic_layer_http._DirectSession, "request", lambda *args, **kwargs:
response
+ )
+ with pytest.raises(ProviderRequestError) as raised:
+ provider_request(provider, "GET", response.url, timeout=1)
+ assert raised.value.failure == ProviderRequestFailure.HTTP_STATUS
+ response.raw.read1.assert_not_called()
Review Comment:
Good catch. The test now uses a real `Response` whose body accessors are
tripwires. Reading the content makes it fail, while the normal path still
raises the `HTTP_STATUS` 503 error without reading the body. Updated in
33bc325518.
--
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]
