test(extensions): assert storage API responses are marked no-store

Stored values are scoped to the requesting user, so the cache-control
header on ExtensionStorageRestApi.response() needs coverage to catch a
silent regression.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset/commands/database/oauth2.py

@@ -20,6 +20,7 @@ from functools import partial
 from typing import cast
 from uuid import UUID
 
+from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.database.exceptions import DatabaseNotFoundError
 from superset.daos.database import DatabaseUserOAuth2TokensDAO
@@ -29,6 +30,7 @@ from superset.exceptions import OAuth2Error
 from superset.key_value.types import JsonKeyValueCodec, KeyValueResource
 from superset.models.core import Database, DatabaseUserOAuth2Tokens
 from superset.superset_typing import OAuth2State
+from superset.utils.core import get_user_id
 from superset.utils.decorators import on_error, transaction
 from superset.utils.oauth2 import decode_oauth2_state
 
@@ -102,6 +104,14 @@ class OAuth2StoreTokenCommand(BaseCommand):
 
         self._state = decode_oauth2_state(self._parameters["state"])
 
+        # Bind the callback to the current session: require an authenticated,
+        # non-guest user whose id matches the one carried in the state.
+        user_id = get_user_id()
+        if user_id is None or security_manager.is_guest_user():
+            raise OAuth2Error("The OAuth2 callback requires an authenticated user")
+        if user_id != self._state["user_id"]:
+            raise OAuth2Error("The OAuth2 state belongs to a different user")
+
         if database := DatabaseUserOAuth2TokensDAO.get_database(
             self._state["database_id"]
         ):

superset/commands/dataset/importers/v1/utils.py

@@ -15,9 +15,12 @@
 # specific language governing permissions and limitations
 # under the License.
 import gzip
+import ipaddress
 import logging
 import os
 import re
+import socket
+from http.client import HTTPConnection, HTTPResponse, HTTPSConnection
 from typing import Any
 from urllib import request
 from urllib.parse import urljoin, urlparse
@@ -47,7 +50,7 @@ from superset.models.helpers import ChildMultipleResultsFound
 from superset.sql.parse import Table
 from superset.utils import json
 from superset.utils.core import get_user
-from superset.utils.network import is_safe_host
+from superset.utils.network import is_safe_host, is_safe_ip
 
 logger = logging.getLogger(__name__)
 
@@ -76,6 +79,47 @@ class _ValidatingRedirectHandler(HTTPRedirectHandler):
         return super().redirect_request(req, fp, code, msg, headers, newurl)
 
 
+def _raise_for_unsafe_peer(sock: socket.socket) -> None:
+    """
+    Validate that an established connection's actual peer is publicly
+    routable, so the address reached matches the policy applied to the host.
+    """
+    peer = sock.getpeername()[0]
+    if not is_safe_ip(ipaddress.ip_address(peer)):
+        raise DatasetForbiddenDataURI()
+
+
+class _PeerValidatingHTTPConnection(HTTPConnection):
+    """HTTP connection that validates the peer address on connect."""
+
+    def connect(self) -> None:
+        super().connect()
+        _raise_for_unsafe_peer(self.sock)
+
+
+class _PeerValidatingHTTPSConnection(HTTPSConnection):
+    """HTTPS connection that validates the peer address after the handshake."""
+
+    def connect(self) -> None:
+        super().connect()
+        _raise_for_unsafe_peer(self.sock)
+
+
+class _PeerValidatingHTTPHandler(request.HTTPHandler):
+    """Opens HTTP connections through the peer-validating connection class."""
+
+    def http_open(self, req: request.Request) -> HTTPResponse:
+        return self.do_open(_PeerValidatingHTTPConnection, req)
+
+
+class _PeerValidatingHTTPSHandler(request.HTTPSHandler):
+    """Opens HTTPS connections through the peer-validating connection class."""
+
+    def https_open(self, req: request.Request) -> HTTPResponse:
+        context = self._context  # type: ignore[attr-defined]
+        return self.do_open(_PeerValidatingHTTPSConnection, req, context=context)
+
+
 CHUNKSIZE = 512
 VARCHAR = re.compile(r"VARCHAR\((\d+)\)", re.IGNORECASE)
 
@@ -581,7 +625,17 @@ def load_data(data_uri: str, dataset: SqlaTable, database: Database) -> None:
 
     validate_data_uri(data_uri)
     logger.info("Downloading data from %s", data_uri)
-    opener = request.build_opener(_ValidatingRedirectHandler)
+    handlers: list[request.BaseHandler | type[request.BaseHandler]] = [
+        _ValidatingRedirectHandler
+    ]
+    if not app.config["DATASET_IMPORT_ALLOW_INTERNAL_DATA_URLS"]:
+        # Also enforce the policy at the socket layer: re-check the peer of
+        # every connection, including each redirect hop. Disable proxies so the
+        # connection is made directly to the destination and the peer check
+        # validates the destination address rather than a proxy's.
+        handlers.append(request.ProxyHandler({}))
+        handlers.extend([_PeerValidatingHTTPHandler, _PeerValidatingHTTPSHandler])
+    opener = request.build_opener(*handlers)
     data = opener.open(data_uri)  # pylint: disable=consider-using-with  # noqa: S310
     if data_uri.endswith(".gz"):
         data = gzip.open(data)

superset/common/query_context_processor.py

@@ -58,6 +58,7 @@ from superset.utils.core import (
     get_column_name,
     get_column_names_from_columns,
     get_column_names_from_metrics,
+    get_user_id,
     is_adhoc_column,
     is_adhoc_metric,
 )
@@ -270,6 +271,11 @@ class QueryContextProcessor:
         datasource = self._qc_datasource
         extra_cache_keys = datasource.get_extra_cache_keys(query_obj.to_dict())
 
+        # Annotation data is cached on the same entry as the dataframe, so the
+        # key must also bind the annotation sources' security context.
+        if query_obj and query_obj.annotation_layers:
+            kwargs["annotation_context"] = self._annotation_cache_context(query_obj)
+
         cache_key = (
             query_obj.cache_key(
                 datasource=datasource.uid,
@@ -283,6 +289,32 @@ class QueryContextProcessor:
         )
         return cache_key
 
+    def _annotation_cache_context(self, query_obj: QueryObject) -> dict[str, Any]:
+        """
+        Cache-key material binding cached annotation data to its security
+        context.
+
+        Annotation payloads are fetched per requesting user and stored on the
+        same cache entry as the dataframe, so the key also binds the requesting
+        user and, for chart-backed layers, the RLS clauses of the referenced
+        chart's datasource.
+        """
+        source_rls: dict[str, list[str] | None] = {}
+        for layer in query_obj.annotation_layers:
+            if layer.get("sourceType") not in ("line", "table"):
+                continue
+            layer_value = layer.get("value")
+            chart = (
+                ChartDAO.find_by_id(layer_value) if layer_value is not None else None
+            )
+            annotation_datasource = chart.datasource if chart else None
+            source_rls[str(layer.get("value"))] = (
+                security_manager.get_rls_cache_key(annotation_datasource)
+                if annotation_datasource
+                else None
+            )
+        return {"user_id": get_user_id(), "source_rls": source_rls}
+
     def get_query_result(self, query_object: QueryObject) -> QueryResult:
         """
         Returns a pandas dataframe based on the query object.
@@ -636,6 +668,11 @@ class QueryContextProcessor:
             if layer["sourceType"] == "NATIVE"
         ]
         layer_ids = [layer["value"] for layer in annotation_layers]
+        # Enforce the annotation read permission before returning layer records.
+        if layer_ids and not security_manager.can_access("can_read", "Annotation"):
+            raise QueryObjectValidationError(
+                _("You don't have access to annotation layers")
+            )
         layer_objects = {
             layer_object.id: layer_object
             for layer_object in AnnotationLayerDAO.find_by_ids(layer_ids)
@@ -645,6 +682,15 @@ class QueryContextProcessor:
         for layer in annotation_layers:
             layer_id = layer["value"]
             layer_name = layer["name"]
+            # A request may reference a layer id that does not exist; treat it
+            # as a validation error rather than failing on the missing key.
+            if (layer_object := layer_objects.get(layer_id)) is None:
+                raise QueryObjectValidationError(
+                    _(
+                        "Annotation layer with ID %(layer_id)s was not found",
+                        layer_id=layer_id,
+                    )
+                )
             columns = [
                 "start_dttm",
                 "end_dttm",
@@ -652,7 +698,6 @@ class QueryContextProcessor:
                 "long_descr",
                 "json_metadata",
             ]
-            layer_object = layer_objects[layer_id]
             records = [
                 {column: getattr(annotation, column) for column in columns}
                 for annotation in layer_object.annotation

superset/extensions/cache_middleware.py

@@ -26,8 +26,12 @@ if TYPE_CHECKING:
 # Matches only the static asset endpoint:
 # /api/v1/extensions/<publisher>/<name>/<path:file>, where the file portion may
 # contain nested segments (worker / WASM / chunk subfolders).
-# Does not match the list (/), get (/<publisher>/<name>), or info (/_info) endpoints.
-_ASSET_PATH_RE: re.Pattern[str] = re.compile(r"^/api/v1/extensions/[^/]+/[^/]+/.+$")
+# Does not match the list (/), get (/<publisher>/<name>), or info (/_info)
+# endpoints, nor the per-user storage endpoints under
+# /<publisher>/<name>/storage/, whose responses must keep ``Vary: Cookie``.
+_ASSET_PATH_RE: re.Pattern[str] = re.compile(
+    r"^/api/v1/extensions/[^/]+/[^/]+/(?!storage/).+$"
+)
 
 
 class ExtensionCacheMiddleware:

superset/extensions/storage/api.py

@@ -92,10 +92,16 @@ class ExtensionStorageRestApi(BaseApi):
     route_base = "/api/v1/extensions"
 
     def response(self, status_code: int, **kwargs: Any) -> Response:
-        """Helper method to create JSON responses."""
+        """Helper method to create JSON responses.
+
+        Stored values are scoped to the requesting user, so responses are
+        marked non-cacheable.
+        """
         from flask import jsonify
 
-        return jsonify(kwargs), status_code
+        response = jsonify(kwargs)
+        response.cache_control.no_store = True
+        return response, status_code
 
     def response_404(self, message: str = "Not found") -> Response:
         """Helper method to create 404 responses."""

superset/jinja_context.py

@@ -1267,6 +1267,34 @@ def get_dataset_id_from_context(metric_key: str) -> int:
     raise SupersetTemplateException(exc_message)
 
 
+def guest_user_can_access_dataset(dataset: SqlaTable) -> bool:
+    """
+    Whether the current guest (embedded) user may read the given dataset.
+
+    Guest access is granted per dashboard, so the dataset must back at least
+    one chart on a dashboard the guest token covers; a ``datasets`` allowlist
+    on the token further restricts the reachable IDs.
+
+    :param dataset: a dataset resolved without the DAO base filter.
+    :returns: whether the guest user may read the dataset.
+    """
+    guest_user = security_manager.get_current_guest_user_if_guest()
+    if not guest_user:
+        return False
+
+    allowed_datasets: list[int] | None = guest_user.guest_token.get("datasets")
+    if allowed_datasets is not None and (
+        not isinstance(allowed_datasets, list) or dataset.id not in allowed_datasets
+    ):
+        return False
+
+    return any(
+        security_manager.has_guest_access(dashboard)
+        for slc in dataset.slices
+        for dashboard in slc.dashboards
+    )
+
+
 def metric_macro(
     env: Environment,
     context: dict[str, Any],
@@ -1289,8 +1317,9 @@ def metric_macro(
     if not dataset_id:
         dataset_id = get_dataset_id_from_context(metric_key)
 
-    # Embedded user access is validated at the dashboard level, so we bypass
-    # the regular DAO filter for them
+    # Embedded (guest) user access is validated at the dashboard level, so the
+    # regular DAO filter is bypassed for them and dashboard-level scope is
+    # enforced explicitly below.
     dataset = DatasetDAO.find_by_id(
         dataset_id,
         skip_base_filter=security_manager.is_guest_user(),
@@ -1298,6 +1327,11 @@ def metric_macro(
     if not dataset:
         raise DatasetNotFoundError(f"Dataset ID {dataset_id} not found.")
 
+    # With the base filter skipped, scope a guest to datasets reachable through
+    # a dashboard their token grants; reuse the not-found error for consistency.
+    if security_manager.is_guest_user() and not guest_user_can_access_dataset(dataset):
+        raise DatasetNotFoundError(f"Dataset ID {dataset_id} not found.")
+
     metrics: dict[str, str] = {
         metric.metric_name: metric.expression for metric in dataset.metrics
     }

superset/security/manager.py

@@ -753,15 +753,24 @@ def _native_filter_query_modified(
     query: Any, allowed_columns: set[str], allowed_metrics: set[str]
 ) -> bool:
     """Whether a single query in a native-filter request reads beyond its targets."""
-    # Columns and group-by may only reference target column(s); adhoc (free-form
-    # SQL) columns cannot be validated, so reject them.
-    for key in ("columns", "groupby"):
+    # Columns, group-by, and series columns may only reference target column(s);
+    # adhoc (free-form SQL) columns cannot be validated, so reject them.
+    for key in ("columns", "groupby", "series_columns"):
         for col in getattr(query, key, None) or []:
             if not isinstance(col, str) or col not in allowed_columns:
                 return True
     for metric in getattr(query, "metrics", None) or []:
         if not _native_filter_term_allowed(metric, allowed_columns, allowed_metrics):
             return True
+    # A series-limit metric ranks the top-N groups in the inner query, so it is
+    # a value-returning term and is validated like a metric. ``QueryObject``
+    # renames the deprecated ``timeseries_limit_metric`` payload key onto this
+    # attribute, so both spellings are covered.
+    series_limit_metric = getattr(query, "series_limit_metric", None)
+    if series_limit_metric and not _native_filter_term_allowed(
+        series_limit_metric, allowed_columns, allowed_metrics
+    ):
+        return True
     # order-by entries are ``(expression, asc)`` pairs.
     for order in getattr(query, "orderby", None) or []:
         expr = order[0] if isinstance(order, (list, tuple)) and order else order
@@ -783,8 +792,9 @@ def _native_filter_request_modified(query_context: "QueryContext") -> bool:
     A native filter may only read the column(s) it targets on the dashboard it
     belongs to. The request is treated as modified (and therefore rejected for
     guest users) when it cannot be tied to a native filter on the requesting
-    dashboard, or when any value-returning term (column, group-by, metric, or
-    order-by) references something other than a target column, a simple
+    dashboard, or when any value-returning term (column, group-by, series
+    column, metric, series-limit metric, or order-by) references something
+    other than a target column, a simple
     aggregate over a target column, or the filter's configured sort metric.
     Free-form SQL terms and saved metrics other than the configured sort metric
     are rejected. Row-restricting clauses (``filter``/``extras``) are not
@@ -4263,6 +4273,15 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
                                                 child_slice_id=slice_id,
                                                 parent_slice=parent_slc,
                                             )
+                                            # Bind the request to the child
+                                            # chart's own datasource, mirroring
+                                            # the direct-chart leg above.
+                                            and (
+                                                child_slc := self.session.query(Slice)
+                                                .filter(Slice.id == slice_id)
+                                                .one_or_none()
+                                            )
+                                            and child_slc.datasource == datasource
                                         )
                                     )
                                 )

superset/tasks/context.py

@@ -128,7 +128,12 @@ class TaskContext(CoreTaskContext):
         """
         from superset.daos.tasks import TaskDAO
 
-        fresh_task = TaskDAO.find_one_or_none(uuid=self._task_uuid)
+        # Internal executor path: load the running task itself, bypassing the
+        # user-visibility base filter (which governs the REST API and may run
+        # inside a request context with no logged-in user).
+        fresh_task = TaskDAO.find_one_or_none(
+            uuid=self._task_uuid, skip_base_filter=True
+        )
         if not fresh_task:
             raise ValueError(f"Task {self._task_uuid} not found")
 

superset/tasks/decorators.py

@@ -378,7 +378,7 @@ class TaskWrapper(Generic[P]):
                 task.uuid,
             )
             # Return task in current state (caller can check status)
-            refreshed = TaskDAO.find_one_or_none(uuid=task.uuid)
+            refreshed = TaskDAO.find_one_or_none(uuid=task.uuid, skip_base_filter=True)
             return refreshed if refreshed else task
 
     def _execute_inline(
@@ -422,7 +422,7 @@ class TaskWrapper(Generic[P]):
                 set_ended_at=True,
             ).run()
             # Refresh to get updated task
-            refreshed = TaskDAO.find_one_or_none(uuid=task.uuid)
+            refreshed = TaskDAO.find_one_or_none(uuid=task.uuid, skip_base_filter=True)
             return refreshed if refreshed else task
 
         # Atomic transition: PENDING → IN_PROGRESS (set started_at for duration
@@ -441,7 +441,7 @@ class TaskWrapper(Generic[P]):
                 self.name,
                 task_uuid,
             )
-            refreshed = TaskDAO.find_one_or_none(uuid=task_uuid)
+            refreshed = TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
             return refreshed if refreshed else task
 
         # Update cached status (no DB read needed - we just wrote IN_PROGRESS)
@@ -520,7 +520,7 @@ class TaskWrapper(Generic[P]):
                     )
 
             # Refresh once at end to return current state
-            final_task = TaskDAO.find_one_or_none(uuid=task_uuid)
+            final_task = TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
             return final_task if final_task else task
 
         except Exception as ex:
@@ -542,7 +542,7 @@ class TaskWrapper(Generic[P]):
             )
 
             # Refresh once at end to return current state
-            final_task = TaskDAO.find_one_or_none(uuid=task_uuid)
+            final_task = TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
             return final_task if final_task else task
 
         finally:
@@ -552,7 +552,9 @@ class TaskWrapper(Generic[P]):
             # Publish completion notification for any waiters
             # Use final_task if set by try/except, otherwise refresh (fallback)
             if final_task is None:
-                final_task = TaskDAO.find_one_or_none(uuid=task_uuid)
+                final_task = TaskDAO.find_one_or_none(
+                    uuid=task_uuid, skip_base_filter=True
+                )
             if final_task and final_task.status in TERMINAL_STATES:
                 TaskManager.publish_completion(task_uuid, final_task.status)
 

superset/tasks/filters.py

@@ -37,16 +37,22 @@ class TaskFilter(BaseFilter):  # pylint: disable=too-few-public-methods
 
     def apply(self, query: Query, value: Any) -> Query:
         """Apply the filter to the query."""
-        from sqlalchemy import and_, select
+        from flask import has_request_context
+        from sqlalchemy import and_, false, select
 
         from superset import security_manager
         from superset.models.task_subscribers import TaskSubscriber
         from superset.models.tasks import Task
 
-        # If user is admin or no user_id, return unfiltered query.
-        # This typically applies to background tasks and system operations
         user_id = get_user_id()
-        if not user_id or security_manager.is_admin():
+        if not user_id:
+            # Within a request, a principal without a user id gets no tasks;
+            # background jobs run outside a request context and are unfiltered.
+            if has_request_context():
+                return query.filter(false())
+            return query
+
+        if security_manager.is_admin():
             return query
 
         is_subscribed = (

superset/tasks/manager.py

@@ -260,8 +260,10 @@ class TaskManager:
         def get_task() -> "Task | None":
             if app:
                 with app.app_context():
-                    return TaskDAO.find_one_or_none(uuid=task_uuid)
-            return TaskDAO.find_one_or_none(uuid=task_uuid)
+                    return TaskDAO.find_one_or_none(
+                        uuid=task_uuid, skip_base_filter=True
+                    )
+            return TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
 
         # Check current state first
         task = get_task()
@@ -477,7 +479,7 @@ class TaskManager:
         """
         from superset.daos.tasks import TaskDAO
 
-        task = TaskDAO.find_one_or_none(uuid=task_uuid)
+        task = TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
         return task is not None and task.status in ABORT_STATES
 
     @classmethod

superset/tasks/scheduler.py

@@ -311,7 +311,11 @@ def execute_task(  # noqa: C901
     # Convert string UUID to native UUID (Celery deserializes as string)
     native_uuid = UUID(task_uuid)
 
-    task = TaskDAO.find_one_or_none(uuid=native_uuid)
+    # Internal executor path: load the task being run regardless of the
+    # request principal (the user-visibility base filter applies to the REST
+    # API, not to the trusted task runner, which may run inside a request
+    # context with no logged-in user, e.g. eager/synchronous execution).
+    task = TaskDAO.find_one_or_none(uuid=native_uuid, skip_base_filter=True)
     if not task:
         logger.error("Task %s not found in metastore", task_uuid)
         return {"status": "error", "message": "Task not found"}
@@ -346,7 +350,7 @@ def execute_task(  # noqa: C901
             task_type,
             task_uuid,
         )
-        refreshed = TaskDAO.find_one_or_none(uuid=native_uuid)
+        refreshed = TaskDAO.find_one_or_none(uuid=native_uuid, skip_base_filter=True)
         return {
             "status": refreshed.status if refreshed else "unknown",
             "task_uuid": task_uuid,
@@ -489,7 +493,7 @@ def execute_task(  # noqa: C901
                 )
 
         # Refresh to get final status for return value and completion notification
-        refreshed = TaskDAO.find_one_or_none(uuid=native_uuid)
+        refreshed = TaskDAO.find_one_or_none(uuid=native_uuid, skip_base_filter=True)
         final_status = refreshed.status if refreshed else "unknown"
 
         # Publish completion notification for any waiters (e.g., sync callers)

superset/utils/link_redirect.py

@@ -140,11 +140,17 @@ def is_safe_redirect_url(url: str) -> bool:
     # following a Location header).
     stripped = _URL_STRIPPED_CONTROL_CHARS.sub("", url.strip())
 
-    # Block protocol-relative URLs
-    if stripped.startswith("//") or stripped.startswith("\\\\"):
+    # WHATWG URL parsers treat backslashes as forward slashes in special
+    # schemes, while urllib does not. Normalize backslashes to slashes before
+    # every structural check, mirroring Django's
+    # ``url_has_allowed_host_and_scheme``.
+    normalized = stripped.replace("\\", "/")
+
+    # Block protocol-relative URLs (any leading mix of slash and backslash)
+    if normalized.startswith("//"):
         return False
 
-    parsed = urlparse(stripped)
+    parsed = urlparse(normalized)
 
     # Relative paths are safe
     if not parsed.scheme and not parsed.netloc:

superset/utils/network.py

@@ -44,6 +44,19 @@ PORT_TIMEOUT = 5
 PING_TIMEOUT = 5
 
 
+def is_safe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """
+    Return True if a single IP address is public and globally routable.
+
+    IPv4-mapped IPv6 addresses (e.g. ``::ffff:127.0.0.1``) are unwrapped so
+    they are checked against the IPv4 unsafe networks rather than bypassing
+    them.
+    """
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
+        ip = ip.ipv4_mapped
+    return ip.is_global and not any(ip in net for net in _SSRF_UNSAFE_NETWORKS)
+
+
 def is_safe_host(host: str) -> bool:
     """
     Return True if ``host`` resolves exclusively to public, globally-routable
@@ -52,6 +65,10 @@ def is_safe_host(host: str) -> bool:
     Returns False if any resolved address falls within a private, loopback,
     link-local, or otherwise non-routable range.  An unresolvable host also
     returns False.
+
+    Name resolution here is independent of the resolution performed when a
+    connection is later opened, so callers that go on to fetch from ``host``
+    should also validate the connected peer address (see ``is_safe_ip``).
     """
     try:
         results = socket.getaddrinfo(host, None)
@@ -64,11 +81,7 @@ def is_safe_host(host: str) -> bool:
             ip = ipaddress.ip_address(sockaddr[0])
         except ValueError:
             return False
-        # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1) so they
-        # are checked against the IPv4 unsafe networks rather than bypassing.
-        if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
-            ip = ip.ipv4_mapped
-        if not ip.is_global or any(ip in net for net in _SSRF_UNSAFE_NETWORKS):
+        if not is_safe_ip(ip):
             return False
     return True
 

tests/unit_tests/commands/databases/oauth2_test.py

@@ -69,6 +69,7 @@ def test_validate_success(
     mock_parameters: OAuth2ProviderResponseSchema,
 ) -> None:
     mocker.patch("superset.utils.oauth2.decode_oauth2_state", return_value=mock_state)
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=1)
     mocker.patch.object(
         DatabaseUserOAuth2TokensDAO,
         "get_database",
@@ -90,6 +91,7 @@ def test_validate_database_not_found(
         "superset.utils.oauth2.decode_oauth2_state",
         return_value={"database_id": 999},
     )
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=1)
     mocker.patch.object(DatabaseUserOAuth2TokensDAO, "get_database", return_value=None)
 
     command = OAuth2StoreTokenCommand(mock_parameters)
@@ -115,6 +117,7 @@ def test_run_success(
         "get_database",
         return_value=mock_database,
     )
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=1)
     mocker.patch.object(
         DatabaseUserOAuth2TokensDAO,
         "find_one_or_none",
@@ -146,6 +149,7 @@ def test_run_existing_token(
         "get_database",
         return_value=mock_database,
     )
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=1)
     existing_token = MagicMock()
     mocker.patch.object(
         DatabaseUserOAuth2TokensDAO,
@@ -166,3 +170,23 @@ def test_run_existing_token(
     assert result == "new_token"
     mock_delete.assert_called_once_with([existing_token])
     mock_create.assert_called_once()
+
+
+def test_validate_rejects_state_not_bound_to_session(
+    mocker: MockerFixture,
+    mock_parameters: OAuth2ProviderResponseSchema,
+) -> None:
+    """
+    The callback must only store tokens for the user who initiated the
+    dance: a state minted for another user, or presented without an
+    authenticated session, is rejected before any token exchange.
+    """
+    command = OAuth2StoreTokenCommand(mock_parameters)
+
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=2)
+    with pytest.raises(OAuth2Error):
+        command.validate()
+
+    mocker.patch("superset.commands.database.oauth2.get_user_id", return_value=None)
+    with pytest.raises(OAuth2Error):
+        command.validate()

tests/unit_tests/common/test_query_context_processor.py

@@ -27,6 +27,7 @@ from superset.common.chart_data import ChartDataResultFormat, ChartDataResultTyp
 from superset.common.chart_data_timing import QueryDataResult, QueryTiming
 from superset.common.db_query_status import QueryStatus
 from superset.common.query_context_processor import QueryContextProcessor
+from superset.exceptions import QueryObjectValidationError
 from superset.utils.core import GenericDataType
 from superset.utils.date_parser import get_past_or_future
 
@@ -98,6 +99,25 @@ def processor(mock_query_context):
     return processor
 
 
+def test_query_cache_key_binds_annotation_data_to_requesting_user(processor):
+    """The cache key for annotated queries must differ per requesting user."""
+    query_obj = MagicMock()
+    query_obj.annotation_layers = [{"sourceType": "NATIVE", "name": "a", "value": 1}]
+    with (
+        patch(
+            "superset.common.query_context_processor.get_user_id",
+            side_effect=[1, 2],
+        ),
+        patch("superset.common.query_context_processor.security_manager"),
+    ):
+        processor.query_cache_key(query_obj)
+        processor.query_cache_key(query_obj)
+    contexts = [
+        call.kwargs["annotation_context"] for call in query_obj.cache_key.call_args_list
+    ]
+    assert contexts[0] != contexts[1]
+
+
 def test_get_data_table_like(processor, mock_query_context):
     df = pd.DataFrame({"col1": [1, 2, 3], "col2": ["a", "b", "c"]})
     coltypes = [GenericDataType.NUMERIC, GenericDataType.STRING]
@@ -2377,3 +2397,26 @@ def test_relative_offset_preserves_inner_bounds(
     # for #40501. Without the fix, inner_from/to_dttm == shifted dates.
     assert captured[0]["inner_from_dttm"] == pd.Timestamp("2026-05-01")
     assert captured[0]["inner_to_dttm"] == pd.Timestamp("2026-05-28")
+
+
+def test_get_native_annotation_data_requires_annotation_read_access():
+    """Native annotation layers are only served to users who can read them."""
+    query_obj = MagicMock()
+    query_obj.annotation_layers = [{"sourceType": "NATIVE", "name": "a", "value": 1}]
+    with (
+        patch(
+            "superset.common.query_context_processor.security_manager"
+        ) as security_manager_mock,
+        patch(
+            "superset.common.query_context_processor.AnnotationLayerDAO.find_by_ids",
+            return_value=[],
+        ) as find_by_ids_mock,
+    ):
+        # ``can_access`` is synchronous; force a plain Mock so the patched
+        # manager doesn't hand back a truthy coroutine that slips past the
+        # ``not can_access(...)`` guard.
+        security_manager_mock.can_access = MagicMock(return_value=False)
+        with pytest.raises(QueryObjectValidationError):
+            QueryContextProcessor.get_native_annotation_data(query_obj)
+    security_manager_mock.can_access.assert_called_once_with("can_read", "Annotation")
+    find_by_ids_mock.assert_not_called()

tests/unit_tests/daos/test_tasks.py

@@ -19,6 +19,7 @@ from collections.abc import Iterator
 from uuid import UUID
 
 import pytest
+from pytest_mock import MockerFixture
 from sqlalchemy.orm.session import Session
 from superset_core.tasks.types import TaskProperties, TaskScope, TaskStatus
 
@@ -395,9 +396,15 @@ def test_remove_subscriber_not_subscribed(session_with_task: Session) -> None:
     assert result is None
 
 
-def test_get_status(session_with_task: Session) -> None:
+def test_get_status(session_with_task: Session, mocker: MockerFixture) -> None:
     """Test get_status returns status string when task found by UUID"""
     from superset.daos.tasks import TaskDAO
+    from superset.models.task_subscribers import TaskSubscriber
+
+    # get_status enforces the TaskFilter, so the polling user must be
+    # authenticated and subscribed to see the task.
+    mocker.patch("superset.tasks.filters.get_user_id", return_value=TEST_USER_ID)
+    mocker.patch("superset.security_manager.is_admin", return_value=False)
 
     task = create_task(
         session_with_task,
@@ -405,6 +412,8 @@ def test_get_status(session_with_task: Session) -> None:
         task_key="status-task",
         status=TaskStatus.IN_PROGRESS,
     )
+    session_with_task.add(TaskSubscriber(task_id=task.id, user_id=TEST_USER_ID))
+    session_with_task.flush()
 
     result = TaskDAO.get_status(task.uuid)
 

tests/unit_tests/databases/api_test.py

@@ -700,6 +700,10 @@ def test_oauth2_happy_path(
         return_value=None,
     )
 
+    mocker.patch(
+        "superset.commands.database.oauth2.get_user_id",
+        return_value=1,
+    )
     state: OAuth2State = {
         "user_id": 1,
         "database_id": 1,
@@ -776,6 +780,10 @@ def test_oauth2_permissions(
         return_value=None,
     )
 
+    mocker.patch(
+        "superset.commands.database.oauth2.get_user_id",
+        return_value=1,
+    )
     state: OAuth2State = {
         "user_id": 1,
         "database_id": 1,
@@ -857,6 +865,10 @@ def test_oauth2_multiple_tokens(
         return_value=None,
     )
 
+    mocker.patch(
+        "superset.commands.database.oauth2.get_user_id",
+        return_value=1,
+    )
     state: OAuth2State = {
         "user_id": 1,
         "database_id": 1,

tests/unit_tests/datasets/commands/importers/v1/import_test.py

@@ -2239,3 +2239,28 @@ def test_import_restore_blocked_by_active_twin_at_incoming_identity(
     assert "another active dataset" in str(excinfo.value)
     # Check-before-mutate: the failed import leaves the row soft-deleted.
     assert existing.deleted_at is not None
+
+
+def test_peer_validating_connection_blocks_rebound_peer() -> None:
+    """
+    The import fetch validates the connected peer address, so a hostname that
+    passes ``is_safe_host`` and then re-resolves to an internal address (DNS
+    rebinding) is rejected before any request bytes are sent.
+    """
+    from http.client import HTTPConnection
+    from unittest.mock import MagicMock, patch
+
+    from superset.commands.dataset.exceptions import DatasetForbiddenDataURI
+    from superset.commands.dataset.importers.v1.utils 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(DatasetForbiddenDataURI):
+            conn.connect()

tests/unit_tests/extensions/storage/test_api.py

@@ -78,6 +78,31 @@ def test_ephemeral_get_delegates_to_dao(
         )
 
 
+@patch("superset.extensions.storage.api.ExtensionEphemeralDAO")
+@patch("superset.extensions.storage.utils.get_extensions")
+def test_ephemeral_get_response_is_marked_no_store(
+    mock_get_ext: MagicMock, mock_dao: MagicMock, app: Flask
+) -> None:
+    """Stored values are scoped to the requesting user, so responses built via
+    `response()` must never be cached (e.g. by a shared/CDN cache)."""
+    mock_get_ext.return_value = {"acme.dashboard": MagicMock()}
+    Babel(app)
+    app.appbuilder = MagicMock()
+    app.appbuilder.sm.is_item_public.return_value = True
+    mock_dao.get_raw.return_value = (get_codec("json").encode({"data": 42}), "json")
+
+    with app.test_request_context(
+        "/api/v1/extensions/acme/dashboard/storage/ephemeral/my-key"
+    ):
+        g.user = MagicMock(id=7)
+
+        body, _status_code = ExtensionStorageRestApi().get_ephemeral(
+            "acme", "dashboard", "my-key"
+        )
+
+        assert body.cache_control.no_store is True
+
+
 @patch("superset.extensions.storage.api.ExtensionEphemeralDAO")
 @patch("superset.extensions.storage.utils.get_extensions")
 def test_ephemeral_get_returns_none_when_entry_missing(

tests/unit_tests/extensions/test_cache_middleware.py

@@ -103,6 +103,17 @@ def test_unrelated_path_is_not_intercepted() -> None:
     assert headers == upstream
 
 
+def test_storage_endpoints_are_not_intercepted() -> None:
+    """Per-user storage responses must keep Vary: Cookie for shared caches."""
+    upstream = [("Vary", "Accept-Encoding, Cookie")]
+    for path in (
+        "/api/v1/extensions/acme/my-ext/storage/ephemeral/some-key",
+        "/api/v1/extensions/acme/my-ext/storage/persistent/some-key",
+    ):
+        headers = call_middleware(path, upstream)
+        assert headers == upstream
+
+
 # --- Vary stripping logic ---
 
 

tests/unit_tests/jinja_context_test.py

@@ -1096,6 +1096,34 @@ def test_metric_macro_with_dataset_id(mocker: MockerFixture) -> None:
     mock_get_form_data.assert_not_called()
 
 
+def test_metric_macro_guest_user_dataset_out_of_scope(mocker: MockerFixture) -> None:
+    """
+    Test that ``metric_macro`` denies a guest user a dataset that is not
+    reachable through any dashboard their guest token grants.
+    """
+    mocker.patch("superset.security_manager.is_guest_user", return_value=True)
+    guest_user = mocker.MagicMock()
+    guest_user.guest_token = {}
+    mocker.patch(
+        "superset.security_manager.get_current_guest_user_if_guest",
+        return_value=guest_user,
+    )
+    DatasetDAO = mocker.patch("superset.daos.dataset.DatasetDAO")  # noqa: N806
+    DatasetDAO.find_by_id.return_value = SqlaTable(
+        id=1,
+        table_name="test_dataset",
+        metrics=[
+            SqlMetric(metric_name="count", expression="COUNT(*)"),
+        ],
+        database=Database(database_name="my_database", sqlalchemy_uri="sqlite://"),
+        schema="my_schema",
+        sql=None,
+    )
+    env = SandboxedEnvironment(undefined=DebugUndefined)
+    with pytest.raises(DatasetNotFoundError):
+        metric_macro(env, {}, "count", 1)
+
+
 def test_metric_macro_recursive(mocker: MockerFixture) -> None:
     """
     Test the ``metric_macro`` when the definition is recursive.
@@ -1732,6 +1760,13 @@ def test_metric_macro_embedded_user_skips_base_filter(mocker: MockerFixture) ->
     mock_is_guest_user = mocker.patch("superset.security_manager.is_guest_user")
     mock_is_guest_user.return_value = True
 
+    # Dashboard-level guest scope is asserted separately; here the dataset is
+    # in scope so the test can focus on the base-filter bypass.
+    mocker.patch(
+        "superset.jinja_context.guest_user_can_access_dataset",
+        return_value=True,
+    )
+
     DatasetDAO = mocker.patch("superset.daos.dataset.DatasetDAO")  # noqa: N806
     DatasetDAO.find_by_id.return_value = SqlaTable(
         table_name="test_dataset",

tests/unit_tests/security/manager_test.py

@@ -141,6 +141,69 @@ def test_raise_for_access_guest_user_ok_subset(
     sm.raise_for_access(query_context=query_context)
 
 
+def test_raise_for_access_guest_user_deck_multi_child_requires_child_datasource(
+    mocker: MockerFixture,
+    app_context: None,
+) -> None:
+    """
+    The deck.gl multi-layer child leg must bind the requested datasource to
+    the child chart: a valid parent/child pair does not authorize querying
+    an arbitrary dataset.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "is_guest_user", return_value=True)
+    mocker.patch.object(sm, "can_access", return_value=False)
+    mocker.patch.object(sm, "can_access_schema", return_value=False)
+    mocker.patch.object(sm, "is_editor", return_value=False)
+    mocker.patch.object(sm, "can_access_dashboard", return_value=True)
+    mocker.patch.object(sm, "get_current_guest_user_if_guest", return_value=None)
+    mocker.patch(
+        "superset.is_feature_enabled",
+        side_effect=lambda feature: feature == "EMBEDDED_SUPERSET",
+    )
+    mocker.patch(
+        "superset.security.manager.query_context_modified",
+        return_value=False,
+    )
+
+    child_datasource = mocker.MagicMock()
+    other_datasource = mocker.MagicMock()
+
+    parent_slc = mocker.MagicMock()
+    parent_slc.params = json.dumps({"viz_type": "deck_multi", "deck_slices": [42]})
+    child_slc = mocker.MagicMock()
+    child_slc.datasource = child_datasource
+
+    dashboard = mocker.MagicMock()
+    dashboard.slices = [parent_slc]
+
+    query_mock = mocker.patch.object(sm.session, "query")
+    query_mock.return_value.filter.return_value.one_or_none.side_effect = [
+        dashboard,
+        parent_slc,
+        child_slc,
+        dashboard,
+        parent_slc,
+        child_slc,
+    ]
+
+    query_context = mocker.MagicMock()
+    query_context.form_data = {
+        "dashboardId": 10,
+        "slice_id": 42,
+        "parent_slice_id": 41,
+    }
+
+    # Requesting the child's own datasource is allowed.
+    query_context.datasource = child_datasource
+    sm.raise_for_access(query_context=query_context)
+
+    # The same chart context with any other datasource is rejected.
+    query_context.datasource = other_datasource
+    with pytest.raises(SupersetSecurityException):
+        sm.raise_for_access(query_context=query_context)
+
+
 def test_raise_for_access_guest_user_tampered_id(
     mocker: MockerFixture,
     app_context: None,
@@ -1459,6 +1522,32 @@ def test_query_context_modified_native_filter_arbitrary_saved_metric_blocked(
     assert query_context_modified(qc)
 
 
+def test_query_context_modified_native_filter_series_limit_terms_blocked(
+    mocker: MockerFixture,
+) -> None:
+    """A series-limit metric or series column beyond the target is modified."""
+    query = SimpleNamespace(
+        columns=["region"],
+        metrics=[],
+        groupby=[],
+        series_columns=["region"],
+        series_limit=5,
+        series_limit_metric={
+            "expressionType": "SIMPLE",
+            "column": {"column_name": "salary"},
+            "aggregate": "MAX",
+        },
+    )
+    qc = _native_filter_ctx(mocker, [query])
+    assert query_context_modified(qc)
+
+    query = SimpleNamespace(
+        columns=["region"], metrics=[], groupby=[], series_columns=["ssn"]
+    )
+    qc = _native_filter_ctx(mocker, [query])
+    assert query_context_modified(qc)
+
+
 def test_query_context_modified_native_filter_orderby_arbitrary_column_blocked(
     mocker: MockerFixture,
 ) -> None:

tests/unit_tests/tasks/test_filters.py

@@ -0,0 +1,44 @@
+# 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.
+from unittest.mock import MagicMock
+
+from flask import current_app
+from pytest_mock import MockerFixture
+from sqlalchemy import false
+
+from superset.tasks.filters import TaskFilter
+
+
+def test_task_filter_fails_closed_for_request_without_user_id(
+    mocker: MockerFixture,
+    app_context: None,
+) -> None:
+    """
+    A request-bound principal without a user id (anonymous or guest user)
+    must not receive the unfiltered task list.
+    """
+    mocker.patch("superset.tasks.filters.get_user_id", return_value=None)
+    task_filter = TaskFilter("id", MagicMock())
+    query = MagicMock()
+
+    with current_app.test_request_context("/api/v1/task/"):
+        filtered = task_filter.apply(query, None)
+
+    assert filtered is not query
+    query.filter.assert_called_once()
+    (predicate,) = query.filter.call_args.args
+    assert str(predicate) == str(false())

tests/unit_tests/utils/test_link_redirect.py

@@ -170,3 +170,21 @@ def test_safe_path_with_tab_in_internal_segment(app: Flask) -> None:
     """A tab inside a regular path segment is still a relative URL after
     stripping; it must not flip the result to safe-then-unsafe."""
     assert is_safe_redirect_url("/dashboard/1?from=tab%09inside")
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "/\\evil.com",  # slash-backslash
+        "\\/evil.com",  # backslash-slash
+        "\\\\evil.com",  # double backslash
+        "/\\/evil.com",  # slash-backslash-slash
+        "/%09/\\evil.com",  # browser-stripped TAB then slash-backslash
+        "https:/\\evil.com",  # backslash inside an absolute URL
+    ],
+)
+def test_unsafe_backslash_protocol_relative(app: Flask, url: str) -> None:
+    """WHATWG URL parsers treat backslashes as forward slashes in special
+    schemes, so any leading mix of slash and backslash is navigated as a
+    protocol-relative URL and must be rejected."""
+    assert not is_safe_redirect_url(url)