fix(cache): dedupe annotation-layer data cache keys across users

superset/common/query_context_processor.py

@@ -45,6 +45,7 @@
 from superset.exceptions import (
     QueryObjectValidationError,
     SupersetException,
+    SupersetSecurityException,
 )
 from superset.explorables.base import Explorable
 from superset.extensions import cache_manager, security_manager
@@ -60,7 +61,6 @@
     get_column_name,
     get_column_names_from_columns,
     get_column_names_from_metrics,
-    get_user_id,
     is_adhoc_column,
     is_adhoc_metric,
 )
@@ -428,7 +428,7 @@ def query_cache_key(self, query_obj: QueryObject, **kwargs: Any) -> str | None:
         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.
+        # key must also bind the annotation sources' security scope.
         if query_obj and query_obj.annotation_layers:
             kwargs["annotation_context"] = self._annotation_cache_context(query_obj)
 
@@ -447,29 +447,98 @@ def query_cache_key(self, query_obj: QueryObject, **kwargs: Any) -> str | None:
 
     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.
+        Cache-key material binding annotation data to its security *scope* so
+        users with the same access share a cache entry and users with a
+        different scope -- or no access -- never read each other's data.
+
+        Annotation payloads are fetched under the requesting user's permissions
+        and stored on the same cache entry as the dataframe, so the key binds
+        the inputs that determine what a user may see:
+
+        * NATIVE layers: the ``can_read`` permission on ``Annotation``, the only
+          user-dependent dimension of these global records.
+        * Chart-backed (``line``/``table``) layers: see
+          :meth:`_annotation_source_scope`.
         """
-        source_rls: dict[str, list[str] | None] = {}
+        context: dict[str, Any] = {}
+
+        if any(
+            layer.get("sourceType") == "NATIVE" for layer in query_obj.annotation_layers
+        ):
+            context["annotation_read"] = security_manager.can_access(
+                "can_read", "Annotation"
+            )
+
+        source_scope: dict[str, Any] = {}
         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
+            source_scope[str(layer_value)] = self._annotation_source_scope(layer_value)
+        if source_scope:
+            context["source_scope"] = source_scope
+
+        return context
+
+    def _annotation_source_scope(self, layer_value: Any) -> dict[str, Any]:
+        """
+        Access and data-identity cache-key material for one chart-backed
+        annotation layer.
+
+        ``access`` keeps a user denied the referenced chart from reading an
+        authorized user's cached payload. ``data_key`` is the annotation chart's
+        own query cache key, capturing the datasource version, RLS clauses, and
+        per-user Jinja/virtual-dataset RLS material.
+        """
+        chart = ChartDAO.find_by_id(layer_value) if layer_value is not None else None
+        datasource = chart.datasource if chart else None
+        if chart is None or datasource is None:
+            return {"access": None, "data_key": None}
+
+        try:
+            annotation_query_context = chart.get_query_context()
+            if annotation_query_context is None:
+                # No saved query context to key on: the fetch itself fails for
+                # every user, so fall back to the datasource's access + RLS
+                # identity.
+                return {
+                    "access": security_manager.can_access_datasource(datasource),
+                    "data_key": security_manager.get_rls_cache_key(datasource),
+                }
+            # Bind the *same* authorization the fetch performs, not just
+            # ``can_access_datasource``: get_viz_annotation_data validates the
+            # annotation chart's query context, whose access check also honors
+            # promiscuous-chart-access (VIEWER_PROMISCUOUS_MODE) and guest-token
+            # scopes -- branches ``can_access_datasource`` skips. Keying on the
+            # narrower check would let a promiscuous-granted viewer collapse onto
+            # a truly-denied user's entry and read their cached payload.
+            try:
+                security_manager.raise_for_access(
+                    query_context=annotation_query_context
+                )
+                access: Any = True
+            except SupersetSecurityException:
+                access = False
+            data_key: Any = [
+                annotation_query_context.query_cache_key(query_object)
+                for query_object in annotation_query_context.queries
+            ]
+        except SupersetException:
+            # The annotation fetch raises these same errors and persists
+            # nothing, so a fallback key never stores real data; fail closed so
+            # this scope can't silently dedupe onto a successfully-derived one.
+            # Other errors propagate rather than weakening the key.
+            logger.warning(
+                "Could not derive annotation cache key for chart %s; "
+                "falling back to a fail-closed scope",
+                layer_value,
+                exc_info=True,
             )
-        return {"user_id": get_user_id(), "source_rls": source_rls}
+            return {
+                "access": False,
+                "data_key": security_manager.get_rls_cache_key(datasource),
+            }
+        return {"access": access, "data_key": data_key}
 
     def get_query_result(self, query_object: QueryObject) -> QueryResult:
         """

superset/datasource/api.py

@@ -48,6 +48,7 @@
 from superset.semantic_layers.mapper import SUPPORTED_FILTER_OPERATORS
 from superset.superset_typing import FlaskResponse
 from superset.utils import json
+from superset.utils.cache import set_data_cache_if_within_size
 from superset.utils.core import (
     apply_max_row_limit,
     DatasourceType,
@@ -281,7 +282,7 @@ def get_column_values(
             # Every distinct search term is its own key, so a few users typing
             # would otherwise pin one entry per keystroke for the full timeout.
             timeout = min(timeout, SEARCH_CACHE_TIMEOUT)
-        cache_manager.data_cache.set(cache_key, payload, timeout=timeout)
+        set_data_cache_if_within_size(cache_key, payload, timeout=timeout)
         logger.debug(
             "column-values cache MISS: uid=%s col=%s", datasource.uid, column_name
         )
@@ -575,7 +576,7 @@ def compatible(self, datasource_type: str, datasource_id: int) -> FlaskResponse:
         timeout = datasource.cache_timeout or app.config.get(
             "CACHE_DEFAULT_TIMEOUT", 300
         )
-        cache_manager.data_cache.set(cache_key, result, timeout=timeout)
+        set_data_cache_if_within_size(cache_key, result, timeout=timeout)
 
         return self.response(200, result=result)
 

superset/sql/execution/executor.py

@@ -1023,11 +1023,10 @@ def _store_in_cache(
             "total_execution_time_ms": result.total_execution_time_ms,
         }
 
-        cache_manager.data_cache.set(
-            cache_key,
-            cached_data,
-            timeout=timeout,
-        )
+        # Apply the same size cap as the chart-data path.
+        from superset.utils.cache import set_data_cache_if_within_size
+
+        set_data_cache_if_within_size(cache_key, cached_data, timeout=timeout)
 
     def _connection_carries_user_identity(self) -> bool:
         """

superset/utils/cache.py

@@ -54,6 +54,50 @@ def generate_cache_key(values_dict: dict[str, Any], key_prefix: str = "") -> str
     return cache_key
 
 
+def oversized_data_cache_value(cache_key: str, cache_value: Any) -> bool:
+    """Whether ``cache_value`` exceeds ``DATA_CACHE_MAX_VALUE_SIZE``.
+
+    Shared size guard so DATA-cache writers that store raw
+    (non-``QueryCacheManager``) payloads -- and so can't use
+    :func:`set_and_log_cache`, which wraps the value -- can also skip oversized
+    entries that would flood the cache backend. Returns ``False`` (never blocks)
+    when the cap is disabled (``None``), avoiding serialization overhead.
+    """
+    max_value_size = app.config.get("DATA_CACHE_MAX_VALUE_SIZE")
+    if max_value_size is None:
+        return False
+    value_size = len(pickle.dumps(cache_value, protocol=pickle.HIGHEST_PROTOCOL))
+    if value_size > max_value_size:
+        logger.warning(
+            "Skipping cache set for key %s: serialized value size %d bytes "
+            "exceeds DATA_CACHE_MAX_VALUE_SIZE (%d bytes)",
+            cache_key,
+            value_size,
+            max_value_size,
+        )
+        app.config["STATS_LOGGER"].incr("skip_cache_value_too_large")
+        return True
+    return False
+
+
+def set_data_cache_if_within_size(
+    cache_key: str, cache_value: Any, timeout: int | None = None
+) -> bool:
+    """Write to the DATA cache unless the value exceeds the size cap.
+
+    Wraps ``data_cache.set`` for writers that store raw payloads outside the
+    ``QueryCacheManager`` contract (and so can't use :func:`set_and_log_cache`).
+
+    :returns: whether the value was persisted.
+    """
+    if oversized_data_cache_value(cache_key, cache_value):
+        return False
+    return (
+        cache_manager.data_cache.set(cache_key, cache_value, timeout=timeout)
+        is not False
+    )
+
+
 def set_and_log_cache(
     cache_instance: Cache,
     cache_key: str,
@@ -88,24 +132,10 @@ def set_and_log_cache(
         )
         value = {**cache_value, "dttm": dttm}
 
-        # Skip caching results that are too large to protect the cache backend
-        # (e.g. Redis/Memcached) from being flooded by huge result sets. The chart
-        # still renders; the value is simply not cached, causing a re-query on the
-        # next load instead of a cache hit. Disabled when DATA_CACHE_MAX_VALUE_SIZE
-        # is None (the default), in which case no serialization overhead is incurred.
-        max_value_size = app.config.get("DATA_CACHE_MAX_VALUE_SIZE")
-        if max_value_size is not None:
-            value_size = len(pickle.dumps(value, protocol=pickle.HIGHEST_PROTOCOL))
-            if value_size > max_value_size:
-                logger.warning(
-                    "Skipping cache set for key %s: serialized value size %d bytes "
-                    "exceeds DATA_CACHE_MAX_VALUE_SIZE (%d bytes)",
-                    cache_key,
-                    value_size,
-                    max_value_size,
-                )
-                app.config["STATS_LOGGER"].incr("skip_cache_value_too_large")
-                return False
+        # Skip oversized results to protect the cache backend; the chart still
+        # renders and simply re-queries on the next load.
+        if oversized_data_cache_value(cache_key, value):
+            return False
 
         # Flask-Caching's set() returns bool | None: cachelib backends can report
         # a failed write by returning False without raising, while some backends