mikebridge commented on code in PR #45553: URL: https://github.com/apache/superset/pull/45553#discussion_r4391858617
########## superset/semantic_layers/transport.py: ########## @@ -0,0 +1,398 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Host-owned provider HTTP policy with shared deadline enforcement.""" + +from __future__ import annotations + +import heapq +import logging +import os +import socket +import time +from collections.abc import Callable, Mapping +from contextvars import Token +from dataclasses import dataclass, field +from io import BytesIO +from threading import Condition, Lock, Thread +from typing import Any, TYPE_CHECKING +from urllib.parse import SplitResult, urljoin, urlsplit + +import requests +from flask import current_app, has_app_context +from superset_core.semantic_layers.transport import ( + ProviderRequestError, + ProviderRequestFailure as Failure, + ProviderResponse, + ProviderTransport, + RequestBudget, +) +from urllib3.exceptions import HTTPError + +from superset.utils.network import ( + ConnectionDeadline, + NonPublicAddressError, + PinnedHTTPAdapter, + PROVIDER_DEADLINE, +) + +if TYPE_CHECKING: + from requests.sessions import _Settings + +logger: logging.Logger = logging.getLogger(__name__) +MAX_REDIRECTS: int = 3 +_CHUNK_BYTES: int = 65536 + + +class HostProviderTransport(ProviderTransport): + """Fix operator policy at construction; keep budgets and sockets call-local.""" + + def __init__(self, *, allow_internal_hosts: bool = False) -> None: + self._allow_internal: bool = allow_internal_hosts + + def request( + self, + provider: str, + method: str, + url: str, + *, + budget: RequestBudget, + headers: Mapping[str, str] | None = None, + params: Mapping[str, Any] | None = None, + json: Any = None, + consume: Callable[[int], None] | None = None, + ) -> ProviderResponse: + # Do not coerce arbitrary objects while handling a URL-bearing failure. + provider = provider if isinstance(provider, str) else "Provider" + bound: _Deadline = _Deadline(budget.deadline) + token: Token[ConnectionDeadline | None] = PROVIDER_DEADLINE.set(bound) + try: + bound.remaining() # No parsing, resolution or socket work after expiry. + self._validated_url(url) + with _DirectSession(allow_environment_ca=self._allow_internal) as session: + adapter: PinnedHTTPAdapter = PinnedHTTPAdapter( + public_only=not self._allow_internal + ) + session.mount("https://";, adapter) + session.mount("http://";, adapter) + hop: int + for hop in range(MAX_REDIRECTS + 1): + with session.request( + method, + url, + headers={**(headers or {}), "Accept-Encoding": "identity"}, + params=params, + json=json, + timeout=bound.remaining(), + stream=True, + allow_redirects=False, + ) as response: + target: str | None = ( + self._next_redirect(method, response) + if hop < MAX_REDIRECTS + else None + ) + if target is not None: + url, params = target, None + continue # Context closes the unread redirect body. + if not 200 <= response.status_code < 300: + raise ProviderRequestError( + Failure.HTTP_STATUS, status=response.status_code + ) + if ( + response.headers.get("Content-Encoding", "identity").lower() + != "identity" + ): + raise ProviderRequestError(Failure.UNSUPPORTED_ENCODING) + body: bytes = self._read_body( + response, bound, budget.max_bytes, consume + ) + result: ProviderResponse = ProviderResponse( + response.status_code, response.headers, body, response.url + ) + bound.remaining() + return result + raise AssertionError("Redirect loop must return or fail") + except ( + requests.RequestException, + HTTPError, + OSError, + ProviderRequestError, + ) as exc: + failure: ProviderRequestError = self._failure_for(exc, budget.deadline) + error: ProviderRequestError = ProviderRequestError( + failure.failure, status=failure.status, provider=provider + ) + logger.warning( + "%s request failed: category=%s status=%s", + error.provider, + error.failure.value, + error.status, + ) + raise error from None + finally: + PROVIDER_DEADLINE.reset(token) + bound.close() + + @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.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: Review Comment: Good catch. The initial URL now follows the redirect policy: a username or password in the URL is rejected as `INVALID_URL` before anything is sent, since credentials belong in headers. Added first-hop coverage in e791fe98e8. ########## tests/unit_tests/semantic_layers/transport_test.py: ########## @@ -0,0 +1,1654 @@ +# 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) + + original: Any = requests.Session.merge_environment_settings + + def trust_test_ca(self: requests.Session, *args: Any) -> dict[str, Any]: + settings: dict[str, Any] = original(self, *args) + if get_ident() != owner: + return settings + settings["verify"] = str(cert) Review Comment: Good catch. The fixture now installs its test CA without rewriting `verify`. The hostname-mismatch and untrusted-certificate cases both reject the connection, and temporarily disabling verification in production code makes both tests fail. Fixed in e791fe98e8. ########## superset/utils/network.py: ########## @@ -260,3 +265,161 @@ 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() + + +class NonPublicAddressError(OSError): + """Resolution included an address refused by the shared destination rules.""" + + +def _resolve_addresses( + host: str, port: int, timeout: float, *, public_only: bool +) -> list[str]: + """Bound waiting and pin the complete answer set before any connection.""" + global _RESOLVER # pylint: disable=global-statement + if timeout <= 0: + raise TimeoutError + with _RESOLVER_LOCK: + if _RESOLVER is None: + _RESOLVER = ThreadPoolExecutor(8, "provider-dns") + lookup: Future[Any] = _RESOLVER.submit( + socket.getaddrinfo, host, port, socket.AF_UNSPEC, socket.SOCK_STREAM + ) + try: + answers: Any = lookup.result(timeout=timeout) + finally: + lookup.cancel() # Queued work is disposable; running system DNS cannot stop. Review Comment: Good catch. DNS admission now bounds outstanding submissions as well as workers. A timed-out request keeps its slot until the submitted lookup actually finishes, and a saturated pool fails immediately instead of queueing. A blocked-worker test covers queued work and slot recovery afterwards, in e791fe98e8. -- 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]
