sadpandajoe commented on code in PR #45553: URL: https://github.com/apache/superset/pull/45553#discussion_r4397215540
########## tests/unit_tests/semantic_layers/transport_test.py: ########## @@ -0,0 +1,1818 @@ +# 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.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: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: requests.Response = requests.Response() + response.status_code = 503 + response.url = "https://fixture.test/secret"; + response.raw = MagicMock(spec=HTTPResponse) + response.raw.read.side_effect = AssertionError("503 raw body was read") + response.raw.read1.side_effect = AssertionError("503 raw body was read") + + def body_access(*args: Any, **kwargs: Any) -> bytes: + raise AssertionError("503 body accessor was used") + + monkeypatch.setattr(requests.Response, "content", property(body_access)) + monkeypatch.setattr(requests.Response, "text", property(body_access)) + monkeypatch.setattr(response, "iter_content", body_access) + 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() + assert f"{expected} request failed: category=http_status status=503" in caplog.text + assert str(raised.value) == f"{expected} request failed with HTTP 503." + assert "secret" not in caplog.text + assert "category=http_status status=503" in caplog.text + + +def test_tls_handshake_stall_is_bounded() -> None: + from threading import Event + + done: Event = Event() + listener: socket.socket = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen() + + def serve() -> None: + connection: socket.socket + connection, _ = listener.accept() + with connection: + done.wait(2) + + worker: Thread = Thread(target=serve, daemon=True) + worker.start() + try: + with _internal_hosts(True): + start: float = time.monotonic() + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "fixture", + "GET", + f"https://127.0.0.1:{listener.getsockname()[1]}/", + timeout=0.15, + ) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + assert time.monotonic() - start < 1.15 + finally: + done.set() + listener.close() + worker.join(2) + + +def test_connect_boundary_stall_uses_real_socket_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Block real socket I/O at connect without a platform-specific SYN blackhole.""" + left: socket.socket + right: socket.socket + left, right = socket.socketpair() + + class ConnectingSocket: + def setsockopt(self, level: int, option: int, value: int | bytes) -> None: + assert (level, option, value) == (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + + def settimeout(self, seconds: float) -> None: + left.settimeout(seconds) + + def connect(self, address: tuple[str, int]) -> None: + left.recv(1) # Real blocking I/O, governed by production's connect timeout. + + def close(self) -> None: + left.close() + + monkeypatch.setattr( + network.socket, "socket", lambda *args, **kwargs: ConnectingSocket() + ) + try: + with _internal_hosts(True): + start: float = time.monotonic() + with pytest.raises(ProviderRequestError) as raised: + provider_request( + "fixture", "GET", "http://127.0.0.1:12345/";, timeout=0.15 + ) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + assert time.monotonic() - start < 1.15 + finally: + left.close() + right.close() + + +def test_ten_concurrent_deadlines_release_descriptors() -> None: + from http.server import ThreadingHTTPServer + from threading import Event + + import psutil + + done: Event = Event() + + class Stall(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 + self.send_response(200) + self.send_header("Content-Length", "10") + self.end_headers() + self.wfile.write(b"x") + self.wfile.flush() + done.wait(2) + + def log_message(self, *args: Any) -> None: + pass + + before: int = psutil.Process().num_fds() + + class ConcurrentServer(ThreadingHTTPServer): + request_queue_size: int = 32 + + server: ThreadingHTTPServer = ConcurrentServer(("127.0.0.1", 0), Stall) + worker: Thread = Thread( + target=lambda: server.serve_forever(poll_interval=0.02), daemon=True + ) + worker.start() + + def call(index: int) -> tuple[float, float]: + budget: float = 0.15 + index * 0.02 + start: float = time.monotonic() + transport: semantic_layer_http.HostProviderTransport = ( + semantic_layer_http.HostProviderTransport(allow_internal_hosts=True) + ) + with pytest.raises(ProviderRequestError) as raised: + transport.request( + "fixture", + "GET", + f"http://127.0.0.1:{server.server_port}/";, + budget=RequestBudget(start + budget, 100), + ) + assert raised.value.failure == ProviderRequestFailure.TIMEOUT + return time.monotonic() - start, budget + + try: + with ThreadPoolExecutor(10) as pool: + results: list[tuple[float, float]] = list(pool.map(call, range(10))) + assert all(elapsed < budget + 1 for elapsed, budget in results) + finally: + done.set() + server.shutdown() + server.server_close() + worker.join(2) + deadline: float = time.monotonic() + 1 + while psutil.Process().num_fds() > before and time.monotonic() < deadline: + time.sleep(0.01) + assert psutil.Process().num_fds() <= before + assert not semantic_layer_http._scheduler()._tokens + + +def test_scheduler_registration_after_expiry_and_close_are_safe() -> None: + scheduler: semantic_layer_http.DeadlineScheduler = semantic_layer_http._scheduler() + bound: semantic_layer_http._Deadline = semantic_layer_http._Deadline( + time.monotonic() - 1 + ) + left: socket.socket + right: socket.socket + left, right = socket.socketpair() + try: + scheduler.register(bound, left) + right.settimeout(0.2) + assert right.recv(1) == b"" + scheduler.close(bound) + scheduler.close(bound) + assert not bound.sockets + finally: + left.close() + right.close() + + +def test_policy_is_default_off_and_isolated() -> None: + from superset import config + + assert config.SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS is False + app: Flask = Flask(__name__) + app.config.update( + SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS=True, + DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS=False, + ALERT_REPORTS_WEBHOOK_ALLOW_INTERNAL_HOSTS=False, + ) + with app.app_context(): + transport: ProviderTransport = semantic_layer_http.build_provider_transport() + assert app.config["DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS"] is False + assert app.config["ALERT_REPORTS_WEBHOOK_ALLOW_INTERNAL_HOSTS"] is False + with _server() as (port, _): + url: str = f"http://127.0.0.1:{port}/"; + assert ( + transport.request( + "fixture", "GET", url, budget=RequestBudget(time.monotonic() + 1, 100) + ).status + == 200 + ) + with pytest.raises(ProviderRequestError): + semantic_layer_http.build_provider_transport().request( + "fixture", "GET", url, budget=RequestBudget(time.monotonic() + 1, 100) + ) + + +def test_consumer_failure_is_not_mislabeled_as_invalid_url( + public_tls: tuple[ssl.SSLContext, list[str]], +) -> None: + def consume(size: int) -> None: + raise ValueError("caller budget exhausted") + + with _server(context=public_tls[0]) as (port, _): + with pytest.raises(ValueError, match="caller budget exhausted"): + _body(f"https://public.example.test:{port}/";, consume=consume) + + +def test_cross_scheme_redirect_returns_status_without_reading_body( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from unittest.mock import MagicMock + + response: MagicMock = MagicMock( Review Comment: This blocked-redirect test still uses a `MagicMock` response, so asserting `raw.read1` was not called does not cover `response.content`, `response.text` or `iter_content()`. If the GET-redirect branch of `_next_redirect` started reading the body before refusing the cross-scheme `Location`, this test would still pass. The `HTTP_STATUS` test now uses a real `Response` with tripwire accessors; could this case do the same so a body read on a refused redirect fails the test? ########## tests/unit_tests/semantic_layers/transport_test.py: ########## @@ -0,0 +1,1818 @@ +# 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.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), Review Comment: The redirect matrix covers default-port equivalence (`:443`) but never a same-host redirect to a different non-default port. If the port were dropped from the `origin()` comparison in `_next_redirect`, every case here would still pass, and the request (with the provider's headers) would be forwarded to another service on the same hostname. Could a case such as `https://provider.example:8443/next` from the port-443 start URL be added, asserting it is not followed and only one send happens? ########## superset/semantic_layers/transport.py: ########## @@ -0,0 +1,409 @@ +# 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( + method, + url, + headers={**(headers or {}), "Accept-Encoding": "identity"}, + params=params, + json=json, + timeout=bound.remaining(), + stream=True, + allow_redirects=False, + ) as response: + target: str | None = ( + self._next_redirect(method, response) + if hop < MAX_REDIRECTS + else None + ) + if target is not None: + url, params = target, None + continue # Context closes the unread redirect body. + if not 200 <= response.status_code < 300: + raise ProviderRequestError( + Failure.HTTP_STATUS, status=response.status_code + ) + if ( + response.headers.get("Content-Encoding", "identity").lower() + != "identity" + ): + raise ProviderRequestError(Failure.UNSUPPORTED_ENCODING) + body: bytes = self._read_body( + response, bound, budget.max_bytes, consume + ) + result: ProviderResponse = ProviderResponse( + response.status_code, response.headers, body, response.url + ) + bound.remaining() + return result + raise AssertionError("Redirect loop must return or fail") + except ( + requests.RequestException, + HTTPError, + OSError, + ProviderRequestError, + ) as exc: + failure: ProviderRequestError = self._failure_for(exc, budget.deadline) + error: ProviderRequestError = ProviderRequestError( + failure.failure, status=failure.status, provider=provider + ) + logger.warning( + "%s request failed: category=%s status=%s", + error.provider, + error.failure.value, + error.status, + ) + raise error from None + finally: + PROVIDER_DEADLINE.reset(token) + bound.close() + + @staticmethod + def _check_url_characters(url: str) -> None: + """Refuse characters URL parsers can discard or interpret differently.""" + if any( + char == "\\" or char.isspace() or ord(char) < 32 or ord(char) == 127 + for char in url + ): + raise ProviderRequestError(Failure.INVALID_URL) + + @classmethod + def _prepared_url(cls, url: str) -> SplitResult: + """Compare destinations using the same URL preparation as Requests.""" + cls._check_url_characters(url) + try: + prepared: requests.PreparedRequest = requests.PreparedRequest() + prepared.prepare_url(url, None) + if prepared.url is None: + raise ProviderRequestError(Failure.INVALID_URL) + parts: SplitResult = urlsplit(prepared.url) + parts.port # noqa: B018 Validate malformed ports before any I/O. + except (ValueError, requests.RequestException): + raise ProviderRequestError(Failure.INVALID_URL) from None + if not parts.hostname: + raise ProviderRequestError(Failure.INVALID_URL) + return parts + + def _validated_url(self, url: str) -> SplitResult: + parts: SplitResult = self._prepared_url(url) + if parts.username is not None or parts.password is not None: + raise ProviderRequestError(Failure.INVALID_URL) + if parts.scheme not in ( + {"https", "http"} if self._allow_internal else {"https"} + ): + raise ProviderRequestError(Failure.REFUSED) + return parts + + def _next_redirect(self, method: str, response: requests.Response) -> str | None: + if ( + method.upper() != "GET" + or response.status_code not in {301, 302, 303, 307, 308} + or not response.headers.get("Location") + ): + return None + before: SplitResult = self._validated_url(response.url) + self._check_url_characters(response.headers["Location"]) + try: + target: str = urljoin(response.url, response.headers["Location"]) + after: SplitResult = self._prepared_url(target) + if after.username is not None or after.password is not None: + raise ProviderRequestError(Failure.INVALID_URL) + except ValueError: + raise ProviderRequestError(Failure.INVALID_URL) from None + + def origin(parts: SplitResult) -> tuple[str, str | None, int]: + return ( + parts.scheme, + parts.hostname, + parts.port + if parts.port is not None + else (443 if parts.scheme == "https" else 80), + ) + + return after.geturl() if origin(before) == origin(after) else None + + @staticmethod + def _read_body( + response: requests.Response, + bound: _Deadline, + cap: int, + consume: Callable[[int], None] | None, + ) -> bytes: + body: BytesIO = BytesIO() + while True: + bound.remaining() + chunk: bytes = response.raw.read1(_CHUNK_BYTES, decode_content=False) + bound.remaining() # Shutdown can look like EOF; refuse partial bytes. + if not chunk: + return body.getvalue() + if consume is not None: + consume(len(chunk)) + if body.tell() + len(chunk) > cap: + raise ProviderRequestError(Failure.TOO_LARGE) + body.write(chunk) + + @staticmethod + def _failure_for(exc: Exception, deadline: float) -> ProviderRequestError: + if time.monotonic() >= deadline or isinstance( + exc, (TimeoutError, requests.Timeout) + ): + return ProviderRequestError(Failure.TIMEOUT) + if isinstance(exc, ProviderRequestError): + return exc + pending: list[BaseException] = [exc] + seen: set[int] = set() + while pending: + cause: BaseException = pending.pop() + if id(cause) in seen: + continue + seen.add(id(cause)) + if isinstance(cause, TimeoutError): + return ProviderRequestError(Failure.TIMEOUT) + if isinstance(cause, NonPublicAddressError): + return ProviderRequestError(Failure.REFUSED) + pending.extend( + item + for item in ( + *cause.args, + cause.__cause__, + getattr(cause, "reason", None), + ) + if isinstance(item, BaseException) + ) + return ProviderRequestError( + Failure.INVALID_URL if isinstance(exc, ValueError) else Failure.UNREACHABLE + ) + + +def build_provider_transport() -> ProviderTransport: + """Fail closed outside app context; read only the provider-specific setting.""" + return HostProviderTransport( + allow_internal_hosts=has_app_context() + and current_app.config.get("SEMANTIC_LAYER_ALLOW_INTERNAL_HOSTS", False) is True + ) + + +class _DirectSession(requests.Session): + def __init__(self, *, allow_environment_ca: bool = False) -> None: + super().__init__() + self.trust_env = False # Never discover credentials from netrc. + self._allow_environment_ca: bool = allow_environment_ca + + def merge_environment_settings(self, *args: Any, **kwargs: Any) -> _Settings: + settings: _Settings = super().merge_environment_settings(*args, **kwargs) + if self._allow_environment_ca and settings["verify"] is True: + settings["verify"] = ( + os.environ.get("REQUESTS_CA_BUNDLE") + or os.environ.get("CURL_CA_BUNDLE") + or settings["verify"] + ) + settings["proxies"] = {} + return settings + + def get_redirect_target(self, resp: requests.Response) -> None: + # Requests otherwise reads redirect bodies despite allow_redirects=False. + return None + + +@dataclass(eq=False) +class _Deadline: + at: float + sockets: list[socket.socket] = field(default_factory=list) + closed: bool = False + expired: bool = False + + def remaining(self) -> float: + seconds: float = self.at - time.monotonic() + if seconds <= 0: + raise TimeoutError + return seconds + + def watch(self, sock: socket.socket) -> None: + _scheduler().register(self, sock) + + def close(self) -> None: + # A request rejected before connecting need not start a scheduler. + if _SCHEDULER is not None: + _SCHEDULER.close(self) + + +class DeadlineScheduler: + """One lazy worker interrupts I/O through owned duplicate descriptors.""" + + def __init__(self) -> None: + self._condition: Condition = Condition() + self._heap: list[tuple[float, int, _Deadline]] = [] + self._tokens: set[_Deadline] = set() + self._worker: Thread = Thread( + target=self._run, name="provider-deadlines", daemon=True + ) + self._worker.start() + + def register(self, bound: _Deadline, sock: socket.socket) -> None: + with self._condition: + if bound.closed: + return + bound.sockets.append(sock.dup()) Review Comment: Registering a duplicate descriptor here means the redirect response's own close no longer closes the TCP connection: the `.dup()` copy stays open until `bound.close()` at the end of the whole request. Consider a provider that answers `302 Location: /final` on a keep-alive connection and is single-threaded, so it sits waiting for the next request on hop one. The transport opens a second connection for `/final`, the provider never accepts it, and the call runs to its deadline as `TIMEOUT` instead of completing. Should the previous hop's duplicate be closed and dropped from `bound.sockets` when the transport abandons that connection? -- 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]
