fix(db-engine-specs): close OAuth2 token-exchange redirect/DNS-rebinding gap; tighten semantic layer secret-reveal check

The OAuth2 endpoint host check validated the hostname up front but the
actual POST could still be redirected to an internal target by the
remote server, or reach a different address than validated via DNS
rebinding. Route token requests through a peer-validating requester
and stop following redirects. Also make the scheme check unconditional
(previously skippable via DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS) and
guard urlparse against a bare ValueError on malformed input.

Separately, the semantic layer secret-reveal guard used dict.get(key)
to detect unchanged fields, which can't distinguish a key that's
absent from one explicitly stored as None -- a new None-valued key
could slip through as "unchanged" alongside a revealed secret.

superset/commands/semantic_layer/update.py

@@ -43,6 +43,10 @@ from superset.utils.decorators import on_error, transaction
 
 logger = logging.getLogger(__name__)
 
+# Sentinel distinguishing "key absent" from "key present with value None"
+# when reading the stored configuration -- dict.get's own default can't.
+_MISSING = object()
+
 
 def _unmask_configuration(
     existing_raw_configuration: str | None,
@@ -63,6 +67,23 @@ def _unmask_configuration(
     write-only, since a client only ever sends the sentinel back for a value
     it previously received masked (including a value masked by the
     fail-closed fallback).
+
+    A masked field is only ever restored, though, when every OTHER
+    submitted key is unchanged from what's stored -- i.e. this is a pure
+    "reveal what I was shown masked" round-trip, not an edit that also
+    changes some other connector field. Without that check, an editor
+    (entitled to edit this connection, but not to see its real secret --
+    that's the entire reason GET/list mask it) could reveal a masked value
+    while simultaneously changing a destination-relevant field in the same
+    request, poisoning the stored configuration: the very next legitimate
+    call through this layer (``POST /<uuid>/schema/runtime`` always uses
+    the stored, now-poisoned configuration) would send the real secret to
+    wherever that field now points. Semantic layer connector schemas are
+    pluggable and defined outside this repo (see
+    ``superset/core/api/core_api_injection.py``), so unlike the analogous
+    database-connection fix there's no fixed "destination fields" list to
+    narrow this to -- any other field changing at all is treated as unsafe
+    to combine with a secret reveal.
     """
     try:
         existing_configuration = (
@@ -71,6 +92,26 @@ def _unmask_configuration(
     except (TypeError, ValueError):
         existing_configuration = {}
 
+    masked_keys = {
+        key
+        for key, value in new_configuration.items()
+        if value == PASSWORD_MASK and key in existing_configuration
+    }
+    # `.get(key)` alone can't tell "key absent from storage" apart from "key
+    # present and stored as None" -- both return None -- so a newly
+    # introduced key with an explicit None value would be misread as
+    # unchanged and let a masked secret slip through alongside it. A
+    # sentinel default makes that distinction explicit.
+    if masked_keys and any(
+        key not in masked_keys and existing_configuration.get(key, _MISSING) != value
+        for key, value in new_configuration.items()
+    ):
+        raise SemanticLayerInvalidError(
+            "This update changes the configuration while reusing a stored "
+            "secret value (a masked field). Provide the real value for any "
+            "masked field to confirm a configuration change."
+        )
+
     return {
         key: (
             existing_configuration[key]

superset/config.py

@@ -2732,6 +2732,15 @@ DATABASE_OAUTH2_JWT_ALGORITHM = "HS256"
 # Timeout when fetching access and refresh tokens.
 DATABASE_OAUTH2_TIMEOUT = timedelta(seconds=30)
 
+# When True, the OAuth2 authorization/token endpoint URIs configured for a
+# database (either via DATABASE_OAUTH2_CLIENTS or, per-connection, via a
+# database's own encrypted_extra.oauth2_client_info) are permitted to target
+# hosts in private/internal IP ranges (RFC-1918, loopback, link-local).
+# Intended for deployments with a legitimately internal identity provider.
+# Leave False (the default) in any deployment where untrusted users can
+# create or edit database connections.
+DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS: bool = False
+
 # Enable/disable CSP warning
 CONTENT_SECURITY_POLICY_WARNING = True
 

superset/db_engine_specs/base.py

@@ -35,11 +35,10 @@ from typing import (
     TypedDict,
     Union,
 )
-from urllib.parse import urlencode, urljoin
+from urllib.parse import urlencode, urljoin, urlparse
 from uuid import UUID, uuid4
 
 import pandas as pd
-import requests
 from apispec import APISpec
 from apispec.ext.marshmallow import MarshmallowPlugin
 from deprecation import deprecated
@@ -95,7 +94,12 @@ from superset.utils import core as utils, json
 from superset.utils.core import ColumnSpec, GenericDataType, QuerySource
 from superset.utils.hashing import hash_from_str
 from superset.utils.json import redact_sensitive, reveal_sensitive
-from superset.utils.network import is_hostname_valid, is_port_open
+from superset.utils.network import (
+    get_ssrf_safe_requester,
+    is_hostname_valid,
+    is_port_open,
+    is_safe_host,
+)
 from superset.utils.oauth2 import (
     encode_oauth2_state,
     generate_code_challenge,
@@ -901,6 +905,47 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
 
         return config
 
+    @staticmethod
+    def _validate_oauth2_endpoint_host(uri: str) -> None:
+        """
+        Validate an OAuth2 authorization/token endpoint URI before it's used.
+
+        ``config["authorization_request_uri"]``/``config["token_request_uri"]``
+        can come from a database's own ``encrypted_extra.oauth2_client_info``
+        (editable by anyone with ``can_write`` on Database, not just the
+        deployment operator). The authorization URI is handed to the user's
+        browser as a redirect target; the token URI is POSTed to directly by
+        this server, carrying the connection's ``client_secret`` in the
+        request body. Neither is otherwise validated, so an attacker with
+        write access to one database's config could point either at an
+        internal host, exfiltrating the client secret (token URI) or using
+        Superset as an open redirect into the internal network (authorization
+        URI) -- and since the connection is typically shared, this is
+        exercised by every user who goes through that database's OAuth2 flow,
+        not just the one who configured it.
+
+        Operators with a legitimately internal IdP can opt out via
+        ``DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS`` -- but that flag only
+        widens which *hosts* are acceptable, not which URI *schemes* are;
+        a non-http(s) scheme is refused unconditionally.
+        """
+        try:
+            parsed = urlparse(uri)
+        except ValueError as ex:
+            # e.g. an unmatched IPv6 bracket -- urlparse raises rather than
+            # returning an unusable result.
+            raise OAuth2Error("Invalid OAuth2 endpoint URI") from ex
+
+        if parsed.scheme not in ("http", "https"):
+            raise OAuth2Error("Invalid OAuth2 endpoint URI")
+
+        if app.config["DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS"]:
+            return
+
+        if not parsed.hostname or not is_safe_host(parsed.hostname):
+            logger.warning("OAuth2 endpoint refused: target host is not allowed")
+            raise OAuth2Error("Invalid OAuth2 endpoint URI")
+
     @classmethod
     def get_oauth2_authorization_uri(
         cls,
@@ -916,6 +961,7 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         (e.g., Google's prompt=consent).
         """
         uri = config["authorization_request_uri"]
+        cls._validate_oauth2_endpoint_host(uri)
         params: dict[str, str] = {
             "scope": config["scope"],
             "response_type": "code",
@@ -947,6 +993,7 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         """
         timeout = app.config["DATABASE_OAUTH2_TIMEOUT"].total_seconds()
         uri = config["token_request_uri"]
+        cls._validate_oauth2_endpoint_host(uri)
         req_body: dict[str, str] = {
             "code": code,
             "client_id": config["id"],
@@ -959,10 +1006,21 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         if code_verifier:
             req_body["code_verifier"] = code_verifier
 
+        # `_validate_oauth2_endpoint_host` only checked the hostname; a
+        # server at that (safe) host could still respond with a 30x
+        # redirecting the actual request to an internal target, or a
+        # low-TTL DNS record could resolve differently by the time this
+        # connects (DNS rebinding). Don't follow redirects, and re-validate
+        # the address actually connected to.
+        requester = get_ssrf_safe_requester(
+            allow_unsafe_hosts=app.config["DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS"]
+        )
         response = (
-            requests.post(uri, data=req_body, timeout=timeout)
+            requester.post(uri, data=req_body, timeout=timeout, allow_redirects=False)
             if config["request_content_type"] == "data"
-            else requests.post(uri, json=req_body, timeout=timeout)
+            else requester.post(
+                uri, json=req_body, timeout=timeout, allow_redirects=False
+            )
         )
         response.raise_for_status()
         return response.json()
@@ -978,16 +1036,26 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         """
         timeout = app.config["DATABASE_OAUTH2_TIMEOUT"].total_seconds()
         uri = config["token_request_uri"]
+        cls._validate_oauth2_endpoint_host(uri)
         req_body = {
             "client_id": config["id"],
             "client_secret": config["secret"],
             "refresh_token": refresh_token,
             "grant_type": "refresh_token",
         }
+        # See the matching comment in ``get_oauth2_token``: the hostname
+        # check above doesn't protect against a 30x redirect to an internal
+        # target or DNS rebinding, so route through the peer-validating
+        # requester and refuse to follow redirects.
+        requester = get_ssrf_safe_requester(
+            allow_unsafe_hosts=app.config["DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS"]
+        )
         response = (
-            requests.post(uri, data=req_body, timeout=timeout)
+            requester.post(uri, data=req_body, timeout=timeout, allow_redirects=False)
             if config["request_content_type"] == "data"
-            else requests.post(uri, json=req_body, timeout=timeout)
+            else requester.post(
+                uri, json=req_body, timeout=timeout, allow_redirects=False
+            )
         )
         if response.status_code in (400, 401, 403):
             try:

superset/db_engine_specs/gsheets.py

@@ -197,6 +197,7 @@ class GSheetsEngineSpec(ShillelaghEngineSpec):
         from superset.utils.oauth2 import encode_oauth2_state, generate_code_challenge
 
         uri = config["authorization_request_uri"]
+        cls._validate_oauth2_endpoint_host(uri)
         params: dict[str, str] = {
             "scope": config["scope"],
             "response_type": "code",

superset/db_engine_specs/snowflake.py

@@ -373,6 +373,7 @@ class SnowflakeEngineSpec(PostgresBaseEngineSpec):
         Return URI for initial OAuth2 request.
         """
         uri = config["authorization_request_uri"]
+        cls._validate_oauth2_endpoint_host(uri)
         # When calling the Snowflake OAuth authorization endpoint for a custom client,
         # specify only the query parameters documented in the URL below.
         # Adding unsupported parameters

superset/security/manager.py

@@ -5332,10 +5332,10 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             audience = audience()
         return audience
 
-    @staticmethod
-    def validate_guest_token_resources(resources: GuestTokenResources) -> None:
+    def validate_guest_token_resources(self, resources: GuestTokenResources) -> None:
         # pylint: disable=import-outside-toplevel
         from superset.commands.dashboard.embedded.exceptions import (
+            EmbeddedDashboardAccessDeniedError,
             EmbeddedDashboardNotFoundError,
         )
         from superset.daos.dashboard import EmbeddedDashboardDAO
@@ -5349,11 +5349,24 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
                     embedded = EmbeddedDashboardDAO.find_by_id(str(resource["id"]))
                     if not embedded:
                         raise EmbeddedDashboardNotFoundError()
+                    dashboard = embedded.dashboard
                 elif not dashboard.embedded:
                     # A raw dashboard id must still reference an embedded dashboard;
                     # otherwise a guest token could be scoped to a non-embedded one.
                     raise EmbeddedDashboardNotFoundError()
 
+                # The caller minting the token must themselves be entitled to
+                # the dashboard being scoped. `grant_guest_token` is a
+                # coarse, instance-wide permission -- without this check, an
+                # operator who narrows it to a non-Admin role (a realistic
+                # "embedding backend service" grant) would let that
+                # principal mint a fully valid guest token for *any*
+                # embedded dashboard, not just ones they have access to.
+                try:
+                    self.raise_for_access(dashboard=dashboard)
+                except SupersetSecurityException as ex:
+                    raise EmbeddedDashboardAccessDeniedError() from ex
+
     def create_guest_access_token(
         self,
         user: GuestTokenUser,

superset/utils/network.py

@@ -18,6 +18,12 @@ import ipaddress
 import platform
 import socket
 import subprocess
+from typing import Any
+
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.connection import HTTPConnection, HTTPSConnection
+from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
 
 # Networks that must never be reached via user-supplied hostnames.
 # Includes loopback, RFC-1918 private ranges, link-local (covers cloud
@@ -44,6 +50,107 @@ PORT_TIMEOUT = 5
 PING_TIMEOUT = 5
 
 
+class SSRFProtectionError(ConnectionError):
+    """
+    Raised when an outbound request's connected peer is not a public,
+    globally-routable address. Subclasses ``ConnectionError`` so it's
+    already covered by any caller that catches connection failures broadly
+    (and, when raised from within an active ``requests``/``urllib3``
+    connection attempt, surfaces to callers as a
+    ``requests.exceptions.ConnectionError``, since ``requests`` wraps
+    whatever a connection class raises during ``connect()``).
+    """
+
+
+def _raise_for_unsafe_peer(conn: HTTPConnection) -> None:
+    """
+    Validate that a connection's actual peer is publicly routable.
+
+    An upfront ``is_safe_host`` check resolves and validates the hostname
+    once, ahead of time; the connection opened here is resolved
+    independently and may reach a different address (DNS rebinding via a
+    low-TTL record), so the check has to be repeated against the address
+    actually connected to.
+    """
+    sock = conn.sock
+    if sock is None:
+        return
+    peer = sock.getpeername()[0]
+    if not is_safe_ip(ipaddress.ip_address(peer)):
+        raise SSRFProtectionError("Request target host is not allowed.")
+
+
+class _PeerValidatingHTTPConnection(HTTPConnection):
+    """HTTP connection that validates the peer address on connect."""
+
+    def connect(self) -> None:
+        super().connect()
+        _raise_for_unsafe_peer(self)
+
+
+class _PeerValidatingHTTPSConnection(HTTPSConnection):
+    """HTTPS connection that validates the peer address after the handshake."""
+
+    def connect(self) -> None:
+        super().connect()
+        _raise_for_unsafe_peer(self)
+
+
+class _PeerValidatingHTTPConnectionPool(HTTPConnectionPool):
+    ConnectionCls = _PeerValidatingHTTPConnection
+
+
+class _PeerValidatingHTTPSConnectionPool(HTTPSConnectionPool):
+    ConnectionCls = _PeerValidatingHTTPSConnection
+
+
+class PeerValidatingHTTPAdapter(HTTPAdapter):
+    """
+    Transport adapter that routes requests through connection classes which
+    validate the connected peer address, closing the TOCTOU window between
+    an upfront ``is_safe_host`` check and the connection ``send()`` actually
+    opens.
+
+    Mirrors the peer-validation approach already used for webhook alert/
+    report dispatch (``superset.reports.notifications.webhook``) and
+    dataset-import data URIs
+    (``superset.commands.dataset.importers.v1.utils``); factored out here so
+    other outbound-request call sites (e.g. OAuth2 token/authorization
+    endpoints) can reuse it instead of re-implementing it.
+    """
+
+    def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:
+        super().init_poolmanager(*args, **kwargs)
+        # Assign a new dict rather than mutating the manager's dict in
+        # place -- the attribute otherwise aliases urllib3's module-global
+        # default scheme-to-pool-class mapping.
+        self.poolmanager.pool_classes_by_scheme = {
+            "http": _PeerValidatingHTTPConnectionPool,
+            "https": _PeerValidatingHTTPSConnectionPool,
+        }
+
+
+def get_ssrf_safe_requester(allow_unsafe_hosts: bool = False) -> Any:
+    """
+    Return a ``requests``-compatible object (the ``requests`` module, or a
+    ``Session``) for making an outbound request to a host that isn't fully
+    operator-controlled.
+
+    The returned session's transport re-validates the actually-connected
+    peer address on every request, closing the TOCTOU window an upfront
+    ``is_safe_host`` check alone leaves open. Pass ``allow_unsafe_hosts=True``
+    for deployments that intentionally target internal hosts (the caller is
+    still expected to have made that an explicit, documented opt-in).
+    """
+    if allow_unsafe_hosts:
+        return requests
+    session = requests.Session()
+    adapter = PeerValidatingHTTPAdapter()
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    return session
+
+
 def is_safe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
     """
     Return True if a single IP address is public and globally routable.

superset/views/utils.py

@@ -44,6 +44,7 @@ from superset.common.db_query_status import QueryStatus
 from superset.exceptions import (
     SerializationError,
     SupersetException,
+    SupersetSecurityException,
 )
 from superset.extensions import security_manager
 from superset.legacy import update_time_range
@@ -504,6 +505,17 @@ def get_dashboard_extra_filters(
     ):
         return []
 
+    # Does the caller actually have access to this dashboard? A chart can
+    # legitimately be reused across multiple dashboards, so passing chart
+    # membership above isn't an entitlement check -- without this, a
+    # principal who owns/can access some chart also embedded on a
+    # dashboard they have no access to could pull that dashboard's default
+    # filter configuration into their own request.
+    try:
+        security_manager.raise_for_access(dashboard=dashboard)
+    except SupersetSecurityException:
+        return []
+
     with contextlib.suppress(json.JSONDecodeError):
         json_metadata = json.loads(dashboard.json_metadata)
         native_filters = [

tests/unit_tests/commands/semantic_layer/update_test.py

@@ -528,6 +528,69 @@ def test_unmask_configuration_missing_existing_key() -> None:
     assert result == {"account": "test", "password": PASSWORD_MASK}
 
 
+def test_unmask_configuration_rejects_secret_reveal_with_changed_field() -> None:
+    """
+    A masked field must not be revealed in the same update that also
+    changes some other configuration field. An editor is entitled to edit
+    this connection, but not to see its real secret (that's the entire
+    reason the read path masks it) -- revealing it while also changing a
+    potentially destination-relevant field would poison the stored
+    configuration with the real secret attached to attacker-controlled
+    config, silently leaking it on the next legitimate use of this layer.
+    """
+    with pytest.raises(SemanticLayerInvalidError):
+        _unmask_configuration(
+            '{"account": "prod-account", "password": "hunter2"}',
+            {"account": "attacker-account", "password": PASSWORD_MASK},
+        )
+
+
+def test_unmask_configuration_rejects_secret_reveal_with_new_none_valued_key() -> None:
+    """
+    A newly introduced key with an explicit ``None`` value must be treated
+    as a configuration change, the same as any other new/changed key --
+    ``dict.get(key)`` alone can't distinguish "key absent from storage" from
+    "key present and stored as None" (both return None), which would let
+    this slip through as "unchanged" and reveal the masked secret alongside
+    it.
+    """
+    with pytest.raises(SemanticLayerInvalidError):
+        _unmask_configuration(
+            '{"account": "prod-account", "password": "hunter2"}',
+            {
+                "account": "prod-account",
+                "password": PASSWORD_MASK,
+                "proxy_host": None,
+            },
+        )
+
+
+def test_unmask_configuration_allows_fresh_secret_with_changed_field() -> None:
+    """
+    A deliberate configuration change is still possible when a genuinely
+    fresh (non-masked) secret is supplied alongside it.
+    """
+    result = _unmask_configuration(
+        '{"account": "prod-account", "password": "hunter2"}',
+        {"account": "new-account", "password": "fresh-secret"},
+    )
+
+    assert result == {"account": "new-account", "password": "fresh-secret"}
+
+
+def test_unmask_configuration_allows_unrelated_field_addition_with_no_mask() -> None:
+    """
+    Adding/changing fields with no masked value present at all is
+    unaffected -- there's no secret being reused, so nothing to protect.
+    """
+    result = _unmask_configuration(
+        '{"account": "prod-account"}',
+        {"account": "new-account", "extra_option": "value"},
+    )
+
+    assert result == {"account": "new-account", "extra_option": "value"}
+
+
 def test_update_semantic_layer_preserves_masked_secret_end_to_end(
     mocker: MockerFixture,
 ) -> None:

tests/unit_tests/db_engine_specs/test_base.py

@@ -39,7 +39,7 @@ from superset.db_engine_specs.base import (
     convert_inspector_columns,
 )
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
-from superset.exceptions import OAuth2RedirectError
+from superset.exceptions import OAuth2Error, OAuth2RedirectError
 from superset.sql.parse import Table
 from superset.superset_typing import (
     OAuth2ClientConfig,
@@ -953,6 +953,18 @@ def test_extract_errors_no_match_falls_back(mocker: MockerFixture) -> None:
     assert result == [expected]
 
 
+@pytest.fixture(autouse=True)
+def _mock_safe_oauth2_host(mocker: MockerFixture) -> None:
+    """
+    OAuth2 endpoint URIs are now validated via ``is_safe_host`` (real DNS
+    resolution) before use. The test fixtures below use non-resolving
+    example hostnames, so mock it the same way test_impala.py mocks it for
+    its own SSRF check; SSRF-rejection behavior itself is covered by
+    dedicated tests further down that override this per-test.
+    """
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
+
+
 def test_get_oauth2_authorization_uri_standard_params(mocker: MockerFixture) -> None:
     """
     Test that BaseEngineSpec.get_oauth2_authorization_uri uses standard OAuth 2.0
@@ -1049,7 +1061,7 @@ def test_get_oauth2_token_without_pkce(mocker: MockerFixture) -> None:
     """
     from superset.db_engine_specs.base import BaseEngineSpec
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.json.return_value = {
         "access_token": "test-access-token",  # noqa: S105
         "expires_in": 3600,
@@ -1082,7 +1094,7 @@ def test_get_oauth2_token_with_pkce(mocker: MockerFixture) -> None:
     from superset.db_engine_specs.base import BaseEngineSpec
     from superset.utils.oauth2 import generate_code_verifier
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.json.return_value = {
         "access_token": "test-access-token",  # noqa: S105
         "expires_in": 3600,
@@ -1166,7 +1178,7 @@ def test_get_oauth2_token_additional_params(mocker: MockerFixture) -> None:
             "audience": "https://api.example.com",
         }
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.json.return_value = {
         "access_token": "test-access-token",  # noqa: S105
         "expires_in": 3600,
@@ -1203,7 +1215,7 @@ def test_get_oauth2_fresh_token_success(mocker: MockerFixture) -> None:
     """
     from superset.db_engine_specs.base import BaseEngineSpec
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.status_code = 200
     mock_post.return_value.json.return_value = {
         "access_token": "new-access-token",
@@ -1234,7 +1246,7 @@ def test_get_oauth2_fresh_token_raises_on_invalid_grant(
     from superset.db_engine_specs.base import BaseEngineSpec
     from superset.exceptions import OAuth2TokenRefreshError
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.status_code = 400
     mock_post.return_value.json.return_value = {"error": "invalid_grant"}
 
@@ -1268,7 +1280,7 @@ def test_get_oauth2_fresh_token_preserves_token_on_ambiguous_error(
 
     from superset.db_engine_specs.base import BaseEngineSpec
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.status_code = status_code
     mock_post.return_value.json.return_value = {"error": error}
     mock_post.return_value.raise_for_status.side_effect = HTTPError()
@@ -1297,7 +1309,7 @@ def test_get_oauth2_fresh_token_raises_on_server_error(mocker: MockerFixture) ->
 
     from superset.db_engine_specs.base import BaseEngineSpec
 
-    mock_post = mocker.patch("superset.db_engine_specs.base.requests.post")
+    mock_post = _mock_requester(mocker).return_value.post
     mock_post.return_value.status_code = 500
     mock_post.return_value.raise_for_status.side_effect = HTTPError("500 Server Error")
 
@@ -1315,6 +1327,220 @@ def test_get_oauth2_fresh_token_raises_on_server_error(mocker: MockerFixture) ->
         BaseEngineSpec.get_oauth2_fresh_token(config, "refresh-token")
 
 
+def _mock_requester(mocker: MockerFixture) -> Any:
+    """
+    Patch ``get_ssrf_safe_requester`` where ``base.py`` looks it up and
+    return the mock, so callers can configure ``.return_value.post`` and/or
+    assert whether (and how) a requester was obtained at all.
+    """
+    return mocker.patch("superset.db_engine_specs.base.get_ssrf_safe_requester")
+
+
+def _oauth2_config_targeting(uri: str) -> OAuth2ClientConfig:
+    return {
+        "id": "client-id",
+        "secret": "client-secret",
+        "scope": "read write",
+        "redirect_uri": "http://localhost:8088/api/v1/database/oauth2/",
+        "authorization_request_uri": uri,
+        "token_request_uri": uri,
+        "request_content_type": "json",
+    }
+
+
+def test_get_oauth2_token_rejects_unsafe_host(mocker: MockerFixture) -> None:
+    """
+    ``token_request_uri`` can come from a database's own
+    ``encrypted_extra.oauth2_client_info`` (editable by anyone with
+    ``can_write`` on Database), and is POSTed to directly by this server
+    carrying the connection's client_secret. An internal/private target
+    must be refused rather than silently reaching it.
+    """
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=False)
+    mock_get_requester = _mock_requester(mocker)
+
+    config = _oauth2_config_targeting("http://169.254.169.254/latest/meta-data/")
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_token(config, "code")
+
+    mock_get_requester.assert_not_called()
+
+
+def test_get_oauth2_fresh_token_rejects_unsafe_host(mocker: MockerFixture) -> None:
+    """
+    Same protection as ``get_oauth2_token``, for the refresh-token exchange.
+    """
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=False)
+    mock_get_requester = _mock_requester(mocker)
+
+    config = _oauth2_config_targeting("http://10.0.0.5/token")
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_fresh_token(config, "refresh-token")
+
+    mock_get_requester.assert_not_called()
+
+
+def test_get_oauth2_authorization_uri_rejects_unsafe_host(
+    mocker: MockerFixture,
+) -> None:
+    """
+    ``authorization_request_uri`` is handed to the user's browser as a
+    redirect target; an internal host would turn Superset into an open
+    redirect into the internal network.
+    """
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=False)
+
+    config = _oauth2_config_targeting("http://192.168.1.1/authorize")
+    state: OAuth2State = {
+        "database_id": 1,
+        "user_id": 1,
+        "default_redirect_uri": "http://localhost:8088/api/v1/oauth2/",
+        "tab_id": "1234",
+    }
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_authorization_uri(config, state)
+
+
+def test_oauth2_endpoint_rejects_non_http_scheme(mocker: MockerFixture) -> None:
+    """
+    A non-http(s) scheme is refused outright, before any host resolution.
+    """
+    is_safe_host = mocker.patch("superset.db_engine_specs.base.is_safe_host")
+    mock_get_requester = _mock_requester(mocker)
+
+    config = _oauth2_config_targeting("file:///etc/passwd")
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_token(config, "code")
+
+    is_safe_host.assert_not_called()
+    mock_get_requester.assert_not_called()
+
+
+def test_oauth2_endpoint_allows_internal_host_when_configured(
+    mocker: MockerFixture,
+) -> None:
+    """
+    Operators with a legitimately internal IdP can opt out via
+    DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS -- the host is not even checked
+    once that's set.
+    """
+    mocker.patch.dict(
+        "superset.db_engine_specs.base.app.config",
+        {"DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS": True},
+    )
+    is_safe_host = mocker.patch("superset.db_engine_specs.base.is_safe_host")
+    mock_get_requester = _mock_requester(mocker)
+    mock_post = mock_get_requester.return_value.post
+    mock_post.return_value.json.return_value = {
+        "access_token": "access-token",
+        "expires_in": 3600,
+    }
+
+    config = _oauth2_config_targeting("http://10.0.0.5/token")
+
+    BaseEngineSpec.get_oauth2_token(config, "code")
+
+    is_safe_host.assert_not_called()
+    mock_get_requester.assert_called_once_with(allow_unsafe_hosts=True)
+    mock_post.assert_called_once()
+
+
+def test_oauth2_endpoint_scheme_check_applies_even_with_internal_hosts_allowed(
+    mocker: MockerFixture,
+) -> None:
+    """
+    DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS widens which *hosts* are acceptable,
+    not which URI *schemes* are -- a non-http(s) scheme (e.g. ``file://``)
+    must still be refused even when that flag is set.
+    """
+    mocker.patch.dict(
+        "superset.db_engine_specs.base.app.config",
+        {"DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS": True},
+    )
+    is_safe_host = mocker.patch("superset.db_engine_specs.base.is_safe_host")
+    mock_get_requester = _mock_requester(mocker)
+
+    config = _oauth2_config_targeting("file:///etc/passwd")
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_token(config, "code")
+
+    is_safe_host.assert_not_called()
+    mock_get_requester.assert_not_called()
+
+
+def test_oauth2_endpoint_malformed_uri_raises_oauth2_error(
+    mocker: MockerFixture,
+) -> None:
+    """
+    ``urlparse`` raises a bare ``ValueError`` (not caught anywhere upstream
+    of ``get_oauth2_authorization_uri``) for malformed IPv6 bracket syntax.
+    That must surface as ``OAuth2Error`` rather than an uncaught ValueError.
+    """
+    is_safe_host = mocker.patch("superset.db_engine_specs.base.is_safe_host")
+
+    config = _oauth2_config_targeting("http://[::1/authorize")
+    state: OAuth2State = {
+        "database_id": 1,
+        "user_id": 1,
+        "default_redirect_uri": "http://localhost:8088/api/v1/oauth2/",
+        "tab_id": "1234",
+    }
+
+    with pytest.raises(OAuth2Error):
+        BaseEngineSpec.get_oauth2_authorization_uri(config, state)
+
+    is_safe_host.assert_not_called()
+
+
+def test_get_oauth2_token_does_not_follow_redirects(mocker: MockerFixture) -> None:
+    """
+    A hostname check alone doesn't stop a server at that (safe) host from
+    responding with a 30x that redirects the actual request -- carrying
+    ``client_secret`` -- to an internal target. The request must be made
+    with ``allow_redirects=False``.
+    """
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
+    mock_get_requester = _mock_requester(mocker)
+    mock_post = mock_get_requester.return_value.post
+    mock_post.return_value.json.return_value = {
+        "access_token": "access-token",
+        "expires_in": 3600,
+    }
+
+    config = _oauth2_config_targeting("https://oauth.example.com/token")
+
+    BaseEngineSpec.get_oauth2_token(config, "code")
+
+    mock_get_requester.assert_called_once_with(allow_unsafe_hosts=False)
+    assert mock_post.call_args.kwargs["allow_redirects"] is False
+
+
+def test_get_oauth2_fresh_token_does_not_follow_redirects(
+    mocker: MockerFixture,
+) -> None:
+    """Same protection as ``get_oauth2_token``, for the refresh-token exchange."""
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
+    mock_get_requester = _mock_requester(mocker)
+    mock_post = mock_get_requester.return_value.post
+    mock_post.return_value.status_code = 200
+    mock_post.return_value.json.return_value = {
+        "access_token": "access-token",
+        "expires_in": 3600,
+    }
+
+    config = _oauth2_config_targeting("https://oauth.example.com/token")
+
+    BaseEngineSpec.get_oauth2_fresh_token(config, "refresh-token")
+
+    mock_get_requester.assert_called_once_with(allow_unsafe_hosts=False)
+    assert mock_post.call_args.kwargs["allow_redirects"] is False
+
+
 def test_start_oauth2_dance_uses_config_redirect_uri(mocker: MockerFixture) -> None:
     """
     Test that start_oauth2_dance uses DATABASE_OAUTH2_REDIRECT_URI config if set.
@@ -1327,6 +1553,7 @@ def test_start_oauth2_dance_uses_config_redirect_uri(mocker: MockerFixture) -> N
             "DATABASE_OAUTH2_REDIRECT_URI": custom_redirect_uri,
             "SECRET_KEY": "test-secret-key",
             "DATABASE_OAUTH2_JWT_ALGORITHM": "HS256",
+            "DATABASE_OAUTH2_ALLOW_INTERNAL_HOSTS": True,
         },
     )
     mocker.patch("superset.daos.key_value.KeyValueDAO")

tests/unit_tests/db_engine_specs/test_databricks.py

@@ -829,7 +829,10 @@ def test_get_oauth2_token_native(
     """
     Test `get_oauth2_token` for Native engine.
     """
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -857,6 +860,7 @@ def test_get_oauth2_token_native(
             "grant_type": "authorization_code",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 
@@ -867,7 +871,10 @@ def test_get_oauth2_token_python(
     """
     Test `get_oauth2_token` for Python Connector engine.
     """
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -895,6 +902,7 @@ def test_get_oauth2_token_python(
             "grant_type": "authorization_code",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 
@@ -905,7 +913,10 @@ def test_get_oauth2_fresh_token_native(
     """
     Test `get_oauth2_fresh_token` for Native engine.
     """
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "new-access-token",
         "expires_in": 3600,
@@ -932,6 +943,7 @@ def test_get_oauth2_fresh_token_native(
             "grant_type": "refresh_token",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 
@@ -989,6 +1001,10 @@ def test_get_oauth2_authorization_uri_derives_from_workspace_host(
     database = mocker.MagicMock()
     database.url_object.host = host
     mocker.patch("superset.db.session.get", return_value=database)
+    # is_safe_host does live DNS resolution; whether these fixture hosts
+    # happen to resolve depends on real-world DNS state outside test
+    # control, so pin it rather than relying on that.
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
 
     url = spec.get_oauth2_authorization_uri(
         _unresolved_oauth2_config(), _oauth2_state()
@@ -1064,7 +1080,10 @@ def test_get_oauth2_fresh_token_python(
     """
     Test `get_oauth2_fresh_token` for Python Connector engine.
     """
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "new-access-token",
         "expires_in": 3600,
@@ -1091,6 +1110,7 @@ def test_get_oauth2_fresh_token_python(
             "grant_type": "refresh_token",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 

tests/unit_tests/db_engine_specs/test_databricks_multi_cloud.py

@@ -91,6 +91,10 @@ def test_get_oauth2_authorization_uri_uses_workspace_host(
         "superset.db.session.get",
         return_value=_mock_database(mocker, host),
     )
+    # is_safe_host does live DNS resolution; whether these fixture hosts
+    # happen to resolve depends on real-world DNS state outside test
+    # control, so pin it rather than relying on that.
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
 
     state: OAuth2State = {
         "database_id": 1,

tests/unit_tests/db_engine_specs/test_gsheets.py

@@ -703,7 +703,10 @@ def test_get_oauth2_token(
     """
     from superset.db_engine_specs.gsheets import GSheetsEngineSpec
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -729,6 +732,7 @@ def test_get_oauth2_token(
             "grant_type": "authorization_code",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 
@@ -741,7 +745,10 @@ def test_get_oauth2_fresh_token(
     """
     from superset.db_engine_specs.gsheets import GSheetsEngineSpec
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -766,6 +773,7 @@ def test_get_oauth2_fresh_token(
             "grant_type": "refresh_token",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 
@@ -900,7 +908,10 @@ def test_get_oauth2_fresh_token_success(
     """
     from superset.db_engine_specs.gsheets import GSheetsEngineSpec
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "new-access-token",
         "expires_in": 3600,
@@ -924,7 +935,10 @@ def test_get_oauth2_fresh_token_invalid_grant(
     """
     from superset.db_engine_specs.gsheets import GSheetsEngineSpec
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().status_code = 400
     requests.post().text = (
         '{"error": "invalid_grant",'
@@ -951,7 +965,10 @@ def test_get_oauth2_fresh_token_other_http_error(
     http_error = HTTPError()
     http_error.response = mock_response
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().raise_for_status.side_effect = http_error
 
     with pytest.raises(HTTPError):

tests/unit_tests/db_engine_specs/test_snowflake.py

@@ -542,7 +542,11 @@ def test_get_oauth2_token(
     """
     from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
 
-    requests: mock.MagicMock = mocker.patch("superset.db_engine_specs.base.requests")
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
+    mock_get_requester: mock.MagicMock = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -568,6 +572,7 @@ def test_get_oauth2_token(
             "grant_type": "authorization_code",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 

tests/unit_tests/db_engine_specs/test_trino.py

@@ -1109,7 +1109,11 @@ def test_get_oauth2_token(
     """
     from superset.db_engine_specs.trino import TrinoEngineSpec
 
-    requests = mocker.patch("superset.db_engine_specs.base.requests")
+    mocker.patch("superset.db_engine_specs.base.is_safe_host", return_value=True)
+    mock_get_requester = mocker.patch(
+        "superset.db_engine_specs.base.get_ssrf_safe_requester"
+    )
+    requests = mock_get_requester.return_value
     requests.post().json.return_value = {
         "access_token": "access-token",
         "expires_in": 3600,
@@ -1135,6 +1139,7 @@ def test_get_oauth2_token(
             "grant_type": "authorization_code",
         },
         timeout=30.0,
+        allow_redirects=False,
     )
 
 

tests/unit_tests/security/manager_test.py

@@ -3985,18 +3985,94 @@ def test_validate_guest_token_resources_rejects_non_embedded_int_id(
 def test_validate_guest_token_resources_accepts_embedded_int_id(
     app_context: None, mocker: MockerFixture
 ) -> None:
-    """A raw int id for an embedded dashboard is accepted."""
+    """A raw int id for an embedded dashboard is accepted when the caller
+    minting the token is entitled to it."""
     from superset.security.guest_token import GuestTokenResourceType
 
     sm = SupersetSecurityManager(appbuilder)
     embedded_dash = MagicMock()
     embedded_dash.embedded = [MagicMock()]  # embedded
     mocker.patch("superset.models.dashboard.Dashboard.get", return_value=embedded_dash)
+    raise_for_access = mocker.patch.object(sm, "raise_for_access")
 
     sm.validate_guest_token_resources(
         [{"type": GuestTokenResourceType.DASHBOARD, "id": 5}]
     )
 
+    raise_for_access.assert_called_once_with(dashboard=embedded_dash)
+
+
+def test_validate_guest_token_resources_rejects_unauthorized_dashboard(
+    app_context: None, mocker: MockerFixture
+) -> None:
+    """
+    Minting a guest token for a dashboard the calling principal is not
+    themselves entitled to must be refused. Without this, a non-Admin role
+    granted only the coarse `can_grant_guest_token` permission (a realistic
+    "embedding backend service" grant, since SECURITY.md treats an
+    operator-narrowed permission as shifting the boundary, not redefining
+    the model) could mint a valid guest token scoped to *any* embedded
+    dashboard in the instance, not just ones it can see.
+    """
+    from superset.commands.dashboard.embedded.exceptions import (
+        EmbeddedDashboardAccessDeniedError,
+    )
+    from superset.exceptions import SupersetSecurityException
+    from superset.security.guest_token import GuestTokenResourceType
+
+    sm = SupersetSecurityManager(appbuilder)
+    embedded_dash = MagicMock()
+    embedded_dash.embedded = [MagicMock()]  # embedded
+    mocker.patch("superset.models.dashboard.Dashboard.get", return_value=embedded_dash)
+    mocker.patch.object(
+        sm,
+        "raise_for_access",
+        side_effect=SupersetSecurityException(mocker.MagicMock()),
+    )
+
+    with pytest.raises(EmbeddedDashboardAccessDeniedError):
+        sm.validate_guest_token_resources(
+            [{"type": GuestTokenResourceType.DASHBOARD, "id": 5}]
+        )
+
+
+def test_validate_guest_token_resources_checks_access_via_embedded_dao_fallback(
+    app_context: None, mocker: MockerFixture
+) -> None:
+    """
+    The same access check applies on the EmbeddedDashboardDAO lookup path
+    (a resource id that isn't a plain dashboard id -- e.g. the embedded
+    config's own uuid), not just the `Dashboard.get` path.
+    """
+    from superset.exceptions import SupersetSecurityException
+    from superset.security.guest_token import GuestTokenResourceType
+
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch("superset.models.dashboard.Dashboard.get", return_value=None)
+    target_dashboard = MagicMock()
+    embedded = MagicMock()
+    embedded.dashboard = target_dashboard
+    mocker.patch(
+        "superset.daos.dashboard.EmbeddedDashboardDAO.find_by_id",
+        return_value=embedded,
+    )
+    raise_for_access = mocker.patch.object(
+        sm,
+        "raise_for_access",
+        side_effect=SupersetSecurityException(mocker.MagicMock()),
+    )
+
+    from superset.commands.dashboard.embedded.exceptions import (
+        EmbeddedDashboardAccessDeniedError,
+    )
+
+    with pytest.raises(EmbeddedDashboardAccessDeniedError):
+        sm.validate_guest_token_resources(
+            [{"type": GuestTokenResourceType.DASHBOARD, "id": "some-uuid"}]
+        )
+
+    raise_for_access.assert_called_once_with(dashboard=target_dashboard)
+
 
 def test_is_editor_query_owner(mocker: MockerFixture, app_context: None) -> None:
     """

tests/unit_tests/utils/test_network.py

@@ -15,11 +15,17 @@
 # specific language governing permissions and limitations
 # under the License.
 import ipaddress
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 
-from superset.utils.network import is_safe_host, is_safe_ip
+from superset.utils.network import (
+    get_ssrf_safe_requester,
+    is_safe_host,
+    is_safe_ip,
+    PeerValidatingHTTPAdapter,
+    SSRFProtectionError,
+)
 
 
 @pytest.mark.parametrize(
@@ -164,3 +170,64 @@ def test_is_safe_host_rejects_cgnat_range() -> None:
         return_value=[(None, None, None, None, ("100.100.100.200", 0))],
     ):
         assert is_safe_host("cgnat-host") is False
+
+
+def test_peer_validating_connection_blocks_rebound_peer() -> None:
+    """
+    A hostname that passes ``is_safe_host`` at validation time and then
+    re-resolves to an internal address by the time the connection is opened
+    (DNS rebinding) must be rejected before any request bytes are sent --
+    mirrors the equivalent test for webhook dispatch and dataset-import
+    data-URI fetches, which use the same pattern.
+    """
+    from urllib3.connection import HTTPConnection
+
+    from superset.utils.network import _PeerValidatingHTTPConnection
+
+    sock = MagicMock()
+    sock.getpeername.return_value = ("169.254.169.254", 80)
+
+    with patch.object(
+        HTTPConnection, "connect", lambda self: setattr(self, "sock", sock)
+    ):
+        conn = _PeerValidatingHTTPConnection("rebinder.example.com")
+        with pytest.raises(SSRFProtectionError):
+            conn.connect()
+
+
+def test_peer_validating_connection_allows_public_peer() -> None:
+    """A connection whose actual peer resolves to a public address is
+    allowed through unmodified."""
+    from urllib3.connection import HTTPConnection
+
+    from superset.utils.network import _PeerValidatingHTTPConnection
+
+    sock = MagicMock()
+    sock.getpeername.return_value = ("93.184.216.34", 80)  # example.com, public
+
+    with patch.object(
+        HTTPConnection, "connect", lambda self: setattr(self, "sock", sock)
+    ):
+        conn = _PeerValidatingHTTPConnection("example.com")
+        conn.connect()  # should not raise
+
+
+def test_get_ssrf_safe_requester_returns_plain_requests_when_allowed() -> None:
+    """
+    With ``allow_unsafe_hosts=True`` (an explicit, documented operator
+    opt-in), the plain ``requests`` module is returned -- no peer pinning.
+    """
+    import requests
+
+    assert get_ssrf_safe_requester(allow_unsafe_hosts=True) is requests
+
+
+def test_get_ssrf_safe_requester_pins_peer_by_default() -> None:
+    """
+    By default, ``get_ssrf_safe_requester`` returns a session whose adapters
+    validate the connected peer address rather than the plain ``requests``
+    module.
+    """
+    requester = get_ssrf_safe_requester()
+    assert isinstance(requester.get_adapter("http://x/"), PeerValidatingHTTPAdapter)
+    assert isinstance(requester.get_adapter("https://x/"), PeerValidatingHTTPAdapter)

tests/unit_tests/views/test_utils.py

@@ -110,7 +110,10 @@ def test_get_dashboard_extra_filters_includes_native_filter_defaults(
     db.session.add_all([chart, dashboard])
     db.session.flush()
 
-    with patch("superset.charts.data.dashboard_filter_context._check_dashboard_access"):
+    with (
+        patch("superset.charts.data.dashboard_filter_context._check_dashboard_access"),
+        patch("superset.views.utils.security_manager.raise_for_access"),
+    ):
         extra_filters = get_dashboard_extra_filters(chart.id, dashboard.id)
 
     assert extra_filters == [{"col": "region", "op": "IN", "val": ["APAC"]}]
@@ -123,6 +126,7 @@ def test_get_dashboard_extra_filters_includes_native_filter_defaults(
 
     with (
         patch("superset.charts.data.dashboard_filter_context._check_dashboard_access"),
+        patch("superset.views.utils.security_manager.raise_for_access"),
         patch(
             "superset.views.utils.build_extra_filters",
             return_value=[legacy_filter],
@@ -134,3 +138,66 @@ def test_get_dashboard_extra_filters_includes_native_filter_defaults(
         legacy_filter,
         {"col": "region", "op": "IN", "val": ["APAC"]},
     ]
+
+
+def test_get_dashboard_extra_filters_denies_unauthorized_dashboard(
+    session: Session,
+) -> None:
+    """
+    A chart can legitimately be reused across multiple dashboards, so
+    chart-membership on a dashboard is not an entitlement check for that
+    dashboard. A caller who can access the chart but not the dashboard it's
+    also placed on must not have that dashboard's filter configuration
+    pulled into their request.
+    """
+    Dashboard.metadata.create_all(session.get_bind())
+
+    dataset = SqlaTable(
+        table_name="unauthorized_dash_table",
+        database=Database(
+            database_name="unauthorized_dash_db", sqlalchemy_uri="sqlite://"
+        ),
+    )
+    db.session.add(dataset)
+    db.session.flush()
+
+    chart = Slice(
+        slice_name="shared_chart",
+        datasource_id=dataset.id,
+        datasource_type="table",
+    )
+    dashboard = Dashboard(
+        dashboard_title="dashboard_caller_cant_access",
+        slices=[chart],
+        published=True,
+        json_metadata=json.dumps(
+            {
+                "default_filters": json.dumps(
+                    {"legacy-filter": {"country": ["Brazil"]}}
+                ),
+                "filter_scopes": {},
+            }
+        ),
+        position_json="{}",
+    )
+    db.session.add_all([chart, dashboard])
+    db.session.flush()
+
+    from unittest.mock import MagicMock
+
+    from superset.exceptions import SupersetSecurityException
+
+    with (
+        patch("superset.charts.data.dashboard_filter_context._check_dashboard_access"),
+        patch(
+            "superset.views.utils.security_manager.raise_for_access",
+            side_effect=SupersetSecurityException(MagicMock()),
+        ),
+        patch(
+            "superset.views.utils.build_extra_filters",
+            return_value=[{"col": "country", "op": "in", "val": ["Brazil"]}],
+        ),
+    ):
+        extra_filters = get_dashboard_extra_filters(chart.id, dashboard.id)
+
+    assert extra_filters == []