Merge pull request #4026 from plotly/feat/platform-streaming-env

Add DASH_SECRET_KEY and DASH_SHARED_STORAGE platform hooks

Author: T4rk1n

.ai/ARCHITECTURE.md

@@ -752,6 +752,8 @@ Special handling for Colab:
 | `DASH_PRUNE_ERRORS` | Simplify tracebacks |
 | `HOST` | Server host |
 | `PORT` | Server port |
+| `DASH_SECRET_KEY` | Signing secret for page, background and stream tokens when `server.secret_key` is unset |
+| `DASH_SHARED_STORAGE` | Shared-storage backend when `shared_storage=` is not passed (see Shared Storage) |
 
 ## Stores and Client-Side State
 
@@ -997,6 +999,20 @@ election only reaches processes in the same network + filesystem namespace.
   This works at 1 pod and fragments silently once it scales — use
   `RedisSharedStorage` (one Redis shared by all pods) instead.
 
+A hosting platform can switch the backend without editing the app through
+`DASH_SHARED_STORAGE` (`_shared_storage/_env.py`), read only when the app did
+not pass `shared_storage=` (the default is a sentinel, so an explicit argument,
+`None` included, always wins): `local`, `none`, `diskcache:///abs/path`, or a
+`redis://` / `rediss://` URL. A `key_prefix` query parameter on the Redis URL
+(`redis://host/0?key_prefix=myapp`) sets `RedisSharedStorage(key_prefix=)` and
+is stripped before the URL reaches redis-py; the diskcache path is
+percent-decoded. `cluster://` is reserved and raises; anything else raises
+`InvalidConfig` at construction, without echoing the value (it can hold
+credentials). The value becomes a zero-argument
+factory, so nothing is built or connected until `app.shared_storage` is first
+read, but a missing extra (`dash[redis]`, `dash[diskcache]`) fails at
+construction with the backend's own `ImportError`.
+
 ### Custom and Out-of-Tree Backends
 
 `BaseSharedStorage` is the stable, public extension point. A backend — shipped
@@ -1035,6 +1051,7 @@ same way as a built-in: `Dash(shared_storage=PostgresSharedStorage(...))`.
 | `_shared_storage/_polling.py` | `PollingSubscription`: shared poll-loop subscription for the diskcache/Redis backends |
 | `_shared_storage/_transport.py` | Length-prefixed, token-gated socket transport (local backend) |
 | `_shared_storage/_codec.py` | msgspec msgpack codec (data-only) |
+| `_shared_storage/_env.py` | `DASH_SHARED_STORAGE` parsing |
 | `dash.py` | `shared_storage` constructor arg + lazy `app.shared_storage` property |
 | `_callback_context.py` | `dash.ctx.shared_storage` accessor |
 
@@ -1405,8 +1422,15 @@ The connection id is never chosen by the client: every stream request rides on
 `?endId=`, the server-signed per-page-load token, and the backend derives the
 id from it (`get_stream_connection_id`), answering 403 when it is missing or
 forged -- otherwise a client could read or inject into another page's topic.
-Across worker processes every worker must resolve the same signing secret
-(`secret_key`).
+Across worker processes every worker must resolve the same signing secret.
+`_get_signing_secret` resolves, in order: `server.secret_key`, then
+`DASH_SECRET_KEY` (for Dash's signing only, never copied onto
+`server.secret_key`, so Flask sessions are untouched), then a secret persisted
+in the background-callback store, then a per-process random one. With the last,
+tokens only verify on the worker that issued them: stream requests 403 on the
+other workers, and the first failure in each process logs a warning pointing at
+`DASH_SECRET_KEY` (`_warn_unverified_stream_token`). A request with no token at
+all is not logged.
 
 Each run of streams (from the first stream after idle until none is in flight)
 also carries `&downlinkId=`, picked fresh by the client, and the connection id

CHANGELOG.md

@@ -5,6 +5,8 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [#4026](https://github.com/plotly/dash/pull/4026) Add `DASH_SECRET_KEY` so a hosting platform can give every worker and pod the same signing secret for page, background and stream tokens without editing the app (`server.secret_key` still wins and is not changed).
+- [#4026](https://github.com/plotly/dash/pull/4026) Add `DASH_SHARED_STORAGE` to pick the shared-storage backend when the app does not pass `shared_storage=`: `local`, `none`, `diskcache:///abs/path`, or a `redis://` / `rediss://` URL.
 - [#4024](https://github.com/plotly/dash/pull/4024) Add support for social media meta tags in regular Dash apps using `app.title`, `app.description`, and `app.image`.
 - [#3976](https://github.com/plotly/dash/pull/3976) Add a new `scrollToTop` prop to `dcc.Link` to control whether the page scrolls to the top after client-side navigation. It defaults to `True` to preserve the existing behavior. Fixes [#3974](https://github.com/plotly/dash/issues/3974).
 - [#3947](https://github.com/plotly/dash/pull/3947) Make `plotly-cloud` a default install dependency of Dash instead of an optional extra, so the `plotly` CLI and Dash's cloud integration work out of the box. The `dash[cloud]` extra is kept for backward compatibility.
@@ -22,6 +24,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 - [#3646](https://github.com/plotly/dash/pull/3646) Remove React 16 support (`16.14.0` is no longer an accepted value for `REACT_VERSION` / `_set_react_version`).
 
 ### Changed
+- [#4026](https://github.com/plotly/dash/pull/4026) Log a warning, once per process, when a streaming request is refused because its stream token failed verification, which on multi-worker deployments usually means the workers do not share a signing secret.
 - [#3987](https://github.com/plotly/dash/pull/3987) Forward FastAPI reload scope options (`reload_dirs`, `reload_excludes`, and `reload_includes`) to Uvicorn when reloading.
 - [#3986](https://github.com/plotly/dash/pull/3986) Adjust `_run_before_hooks` in the `fastapi` backend to honor a response returned by a `before_request` function, matching the `flask` backend's behavior.
 

dash/_callback.py

@@ -3,6 +3,7 @@
 import inspect
 import logging
 import re
+import threading
 import warnings
 from functools import wraps
 from typing import Callable, Optional, Any, List, Tuple, Union, Dict, TypeVar, cast
@@ -465,11 +466,14 @@ def get_request_end_id(secret: bytes):
     request; this verifies the signature and returns the underlying end_id so
     background handles can be checked against it.
     """
+    return _callback_signing.unsign(
+        secret, _callback_signing.END_SCOPE, _request_end_token()
+    )
+
+
+def _request_end_token():
     adapter = get_app().backend.request_adapter()
-    if not adapter:
-        return None
-    token = adapter.args.get("endId")
-    return _callback_signing.unsign(secret, _callback_signing.END_SCOPE, token)
+    return adapter.args.get("endId") if adapter else None
 
 
 def get_stream_connection_id() -> "str | None":
@@ -483,9 +487,9 @@ def get_stream_connection_id() -> "str | None":
     forged token yields ``None``, and the backend refuses the request (403).
 
     ``end_id`` is signed with the server secret, so across worker processes every
-    worker must resolve the same secret: set a ``secret_key`` on the server, or
-    cross-worker stream requests will not verify. Single-process apps are fine
-    with no configuration.
+    worker must resolve the same secret: set a ``secret_key`` on the server or
+    the ``DASH_SECRET_KEY`` environment variable, or cross-worker stream requests
+    will not verify. Single-process apps are fine with no configuration.
 
     The renderer also sends a ``downlinkId`` it picks fresh for each run of
     streams, giving every run its own topic (``<end_id>:<downlinkId>``). A run
@@ -494,6 +498,8 @@ def get_stream_connection_id() -> "str | None":
     """
     end_id = get_request_end_id(_get_signing_secret())
     if end_id is None:
+        if _request_end_token():
+            _warn_unverified_stream_token()
         return None
     downlink_id = get_app().backend.request_adapter().args.get("downlinkId")
     if not downlink_id:
@@ -504,6 +510,24 @@ def get_stream_connection_id() -> "str | None":
 
 
 _DOWNLINK_ID_RE = re.compile(r"[A-Za-z0-9_-]{1,64}")
+_stream_token_warning_lock = threading.Lock()
+_stream_token_warned = False
+
+
+def _warn_unverified_stream_token():
+    # Once per process: a secret mismatch fails every cross-worker request.
+    global _stream_token_warned  # pylint: disable=global-statement
+    with _stream_token_warning_lock:
+        if _stream_token_warned:
+            return
+        _stream_token_warned = True
+    get_app().logger.warning(
+        "A streaming request was refused (403): its stream token failed "
+        "verification. On multi-worker or multi-pod deployments this usually "
+        "means the workers do not share a signing secret (or the server "
+        "restarted since the page loaded); set server.secret_key or the "
+        "DASH_SECRET_KEY environment variable."
+    )
 
 
 def _get_signing_secret() -> bytes:

dash/_configs.py

@@ -35,6 +35,8 @@ def load_dash_env_vars():
                 "DASH_COMPRESS",
                 "DASH_MCP_ENABLED",
                 "DASH_MCP_PATH",
+                "DASH_SECRET_KEY",
+                "DASH_SHARED_STORAGE",
                 "HOST",
                 "PORT",
             )

dash/_shared_storage/_env.py

@@ -0,0 +1,75 @@
+"""Pick a shared-storage backend from ``DASH_SHARED_STORAGE``.
+
+Lets a hosting platform switch backends without editing the app's ``Dash(...)``
+call. Only used when the app did not pass ``shared_storage=``.
+"""
+
+import functools
+from typing import Any, Dict, Optional
+from urllib.parse import parse_qsl, unquote, urlencode, urlparse
+
+from ..exceptions import InvalidConfig
+from .diskcache import DiskcacheSharedStorage, _require_diskcache
+from .local import LocalSharedStorage
+from .redis import RedisSharedStorage, _require_redis
+
+ENV_VAR = "DASH_SHARED_STORAGE"
+
+
+def _invalid(reason: str) -> InvalidConfig:
+    # The value can hold credentials, so it stays out of the message.
+    return InvalidConfig(
+        f"{ENV_VAR} is not valid: {reason}. Use 'local', 'none', "
+        "'diskcache:///absolute/path', or a redis:// or rediss:// URL."
+    )
+
+
+def _redis_from_url(raw: str) -> Any:
+    parsed = urlparse(raw)
+    query = parse_qsl(parsed.query, keep_blank_values=True)
+    kwargs: Dict[str, Any] = {}
+    rest = []
+    for key, value in query:
+        if key == "key_prefix":
+            if not value:
+                raise _invalid("key_prefix cannot be empty")
+            kwargs["key_prefix"] = value
+        else:
+            rest.append((key, value))
+    url = parsed._replace(query=urlencode(rest)).geturl()
+    return functools.partial(RedisSharedStorage, url=url, **kwargs)
+
+
+def storage_from_env(value: Optional[str]) -> Any:
+    """Turn a ``DASH_SHARED_STORAGE`` value into a ``shared_storage`` argument.
+
+    Returns ``None`` (disabled), or a zero-argument callable that builds the
+    backend. Nothing is built or connected here, so startup stays lazy; missing
+    optional dependencies still fail now, with the backend's own error.
+    """
+    raw = (value or "").strip()
+    lowered = raw.lower()
+    if lowered in ("", "local"):
+        return LocalSharedStorage
+    if lowered == "none":
+        return None
+
+    try:
+        scheme = urlparse(raw).scheme.lower()
+    except ValueError:
+        raise _invalid("malformed URL") from None
+    if scheme in ("redis", "rediss"):
+        _require_redis()
+        return _redis_from_url(raw)
+    if scheme == "diskcache":
+        parsed = urlparse(raw)
+        if parsed.netloc or not parsed.path.startswith("/"):
+            raise _invalid("diskcache needs an absolute path (three slashes)")
+        _require_diskcache()
+        return functools.partial(DiskcacheSharedStorage, directory=unquote(parsed.path))
+    if scheme == "cluster":
+        raise InvalidConfig(
+            f"{ENV_VAR}: the cluster:// backend is not supported in this "
+            "version of Dash."
+        )
+    raise _invalid("unknown backend")

dash/dash.py

@@ -48,7 +48,12 @@
 )
 from .backends import get_backend
 from .version import __version__
-from ._configs import get_combined_config, pathname_configs, pages_folder_config
+from ._configs import (
+    get_combined_config,
+    load_dash_env_vars,
+    pathname_configs,
+    pages_folder_config,
+)
 from ._utils import (
     AttributeDict,
     format_tag,
@@ -76,11 +81,8 @@
 from . import backends
 
 from ._get_app import with_app_context, with_app_context_factory
-from ._shared_storage import (
-    BaseSharedStorage,
-    LocalSharedStorage,
-    SharedStorageError,
-)
+from ._shared_storage import BaseSharedStorage, SharedStorageError
+from ._shared_storage._env import storage_from_env
 from ._grouping import map_grouping, grouping_len, update_args_group
 from ._obsolete import ObsoleteChecker
 from ._callback_context import callback_context
@@ -154,6 +156,9 @@
 _ID_DUMMY = "_pages_dummy"
 
 _UNINITIALIZED = object()  # Sentinel for tracking init_app state
+# Default for ``shared_storage`` so "not passed" (DASH_SHARED_STORAGE may pick
+# the backend) can be told apart from an explicit argument.
+_SHARED_STORAGE_DEFAULT: Any = object()
 
 DASH_VERSION_URL = "https://dash-version.plotly.com:8080/current_version"
 
@@ -470,6 +475,20 @@ class Dash(ObsoleteChecker):
         takes a thread for milliseconds. ASGI backends (Quart, FastAPI) keep
         one open connection per browser instead and ignore this.
     :type stream_poll_interval: int
+
+    :param shared_storage: Backend for ``dash.ctx.shared_storage`` and the
+        streaming-callback transport: a ``BaseSharedStorage`` subclass or
+        instance, or ``None`` to disable. Default ``LocalSharedStorage`` (one
+        machine or pod). When not passed, the ``DASH_SHARED_STORAGE``
+        environment variable can choose it: ``local``, ``none``,
+        ``diskcache:///absolute/path``, or a ``redis://`` / ``rediss://`` URL.
+        An explicit argument, including ``None``, always wins.
+
+        Stream requests are signed per page load, so with several workers or
+        pods they all need the same signing secret: set ``server.secret_key``
+        or the ``DASH_SECRET_KEY`` environment variable. ``DASH_SECRET_KEY`` is
+        used for Dash's own signing only and does not set ``server.secret_key``.
+    :type shared_storage: BaseSharedStorage subclass or instance, or None
     """
 
     _plotlyjs_url: str
@@ -535,7 +554,7 @@ def __init__(  # pylint: disable=too-many-statements, too-many-branches
         stream_poll_interval: int = 100,
         shared_storage: Optional[
             Union[Type[BaseSharedStorage], BaseSharedStorage]
-        ] = LocalSharedStorage,
+        ] = _SHARED_STORAGE_DEFAULT,
         enable_mcp: Optional[bool] = None,
         mcp_path: Optional[str] = None,
         **obsolete,
@@ -724,6 +743,10 @@ def __init__(  # pylint: disable=too-many-statements, too-many-branches
         # access so it costs nothing until used and never binds in a gunicorn
         # preload master or the Flask reloader parent -- only in the worker that
         # actually touches it.
+        if shared_storage is _SHARED_STORAGE_DEFAULT:
+            shared_storage = storage_from_env(
+                load_dash_env_vars().get("DASH_SHARED_STORAGE")
+            )
         self._shared_storage_arg = shared_storage
         self._shared_storage_instance: Optional[BaseSharedStorage] = None
         self._shared_storage_lock = threading.Lock()
@@ -753,9 +776,8 @@ def __init__(  # pylint: disable=too-many-statements, too-many-branches
         self._pages_lock = threading.Lock()
         self._pages_async_lock: Optional[asyncio.Lock] = None
 
-        # Secret used to sign background-callback handles (see _callback_signing).
-        # Prefer the Flask/Quart secret_key (shared across workers when the
-        # operator sets one); otherwise fall back to a per-process random secret.
+        # Fallback signing secret, used when neither server.secret_key nor
+        # DASH_SECRET_KEY is set. See _get_signing_secret.
         self._generated_signing_secret: Optional[bytes] = None
 
         if server:
@@ -1022,7 +1044,7 @@ def shared_storage(self) -> BaseSharedStorage:
             with self._shared_storage_lock:
                 if self._shared_storage_instance is None:
                     storage = self._shared_storage_arg
-                    if isinstance(storage, type):
+                    if isinstance(storage, (type, functools.partial)):
                         storage = storage()
                     storage.start()
                     self._shared_storage_instance = storage
@@ -1071,21 +1093,29 @@ def serve_layout(self):
         )
 
     def _get_signing_secret(self) -> bytes:
-        """Return the secret used to sign background-callback handles.
+        """Return the secret used to sign page tokens (``end_id``), background
+        callback handles and stream connections.
 
         Resolution order:
 
         1. The server's ``secret_key`` if set (shared across workers when the
            operator configures one, e.g. for Flask-Login).
-        2. Otherwise a random secret persisted in the background-callback result
+        2. ``DASH_SECRET_KEY`` from the environment, so a hosting platform can
+           give every worker and pod the same key without editing the app. Used
+           for Dash's own signing only, never assigned to ``server.secret_key``
+           (that would change Flask session behavior).
+        3. Otherwise a random secret persisted in the background-callback result
            store, so every worker reads back the same value. This is exactly as
            shared as the callback results themselves, so it works cross-worker
            whenever the deployment is set up for multi-worker background
            callbacks (an explicitly shared cache / broker).
-        3. Finally, if no background manager is available, a per-process random
-           secret (there are no background handles to verify in that case).
+        4. Finally, if no background manager is available, a per-process random
+           secret. Stream tokens then only verify on the worker that issued
+           them, so multi-worker streaming apps need 1 or 2.
         """
-        key = getattr(self.server, "secret_key", None)
+        key = getattr(self.server, "secret_key", None) or load_dash_env_vars().get(
+            "DASH_SECRET_KEY"
+        )
         if key:
             return key.encode("utf-8") if isinstance(key, str) else key
         if self._generated_signing_secret is None: