Authorize browser state through session-aware transports

news/+browser-session-transport.feature.md

@@ -0,0 +1 @@
+Add browser sessions using HttpOnly cookies and a `POST /_reflex/session` endpoint that issues a client token bound to the requesting session. Sessions scope HTTP and custom WebSocket state access; authenticated sockets can renew their cookie credentials without losing client state.

packages/reflex-base/news/+session-context-authorization.deprecation.md

@@ -0,0 +1 @@
+State access without an authorized session now warns by default and is rejected with `REFLEX_SESSION_TOKEN_MODE=enforce`. Trusted server-side fan-out can explicitly pass `SessionToken.SYSTEM`.

packages/reflex-base/news/+session-context-authorization.feature.md

@@ -0,0 +1 @@
+Add session ownership checks to event contexts, including chained events and streamed upload responses. See the [browser session guide](https://reflex.dev/docs/api-routes/sessions/) for rollout and deployment configuration.

packages/reflex-base/news/+upload-transaction-context.bugfix.md

@@ -0,0 +1 @@
+Give buffered uploads their own event transaction so canceling an upload cannot cancel unrelated clients' events.

packages/reflex-base/src/reflex_base/event/context.py

@@ -10,6 +10,11 @@
 
 from reflex_base import otel
 from reflex_base.context.base import BaseContext
+from reflex_base.environment import environment
+from reflex_base.session import SessionToken
+from reflex_base.utils import console
+from reflex_base.utils.compat import MISSING_TYPE
+from reflex_base.utils.exceptions import SessionAuthorizationError
 from reflex_base.utils.format import to_snake_case
 
 if TYPE_CHECKING:
@@ -90,6 +95,9 @@ class EventContext(BaseContext):
     # Identifies the client session.
     token: str
 
+    # Validated browser session, or SYSTEM for trusted server-side work.
+    session_token: SessionToken | None = dataclasses.field(default=None, repr=False)
+
     # Manages persistence of state across events.
     state_manager: StateManager = dataclasses.field(repr=False)
 
@@ -119,20 +127,55 @@ class EventContext(BaseContext):
     # event a handler yields resolves against the view that produced it.
     router_data: dict[str, Any] = dataclasses.field(default_factory=dict, repr=False)
 
+    def __post_init__(self) -> None:
+        """Validate the requested client state against the session.
+
+        Raises:
+            SessionAuthorizationError: If enforcement rejects the client token.
+        """
+        if not self.token or (
+            self.session_token is not None and self.session_token.authorizes(self.token)
+        ):
+            return
+        mode = environment.REFLEX_SESSION_TOKEN_MODE.get()
+        if mode == "off":
+            return
+        if mode == "enforce":
+            msg = "The session does not authorize this client token."
+            raise SessionAuthorizationError(msg)
+        console.deprecate(
+            feature_name="State access without an authorized session",
+            reason=(
+                "Pass the requesting session when accessing client state. "
+                "Trusted server-side work may explicitly use SessionToken.SYSTEM. "
+                "REFLEX_SESSION_TOKEN_MODE=enforce rejects this access."
+            ),
+            deprecation_version="0.10.1",
+            removal_version="1.0",
+        )
+
     def fork(
-        self, token: str | None = None, router_data: dict[str, Any] | None = None
+        self,
+        token: str | None = None,
+        router_data: dict[str, Any] | None = None,
+        *,
+        session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING,
     ) -> EventContext:
         """Return a new EventContext with the specified fields replaced.
 
         Args:
             token: The client token for the new context.
             router_data: The routing data for the new context, inherited from this context if empty.
+            session_token: The session to use, or omitted to inherit this context's session.
 
         Returns:
             A new EventContext with the specified fields replaced.
         """
         return type(self)(
-            token=token or self.token,
+            token=self.token if token is None else token,
+            session_token=self.session_token
+            if session_token is dataclasses.MISSING
+            else session_token,
             # Forks of a context without a txid are top-level events.
             parent_txid=self.txid or None,
             state_manager=self.state_manager,

packages/reflex-base/src/reflex_base/event/processor/event_processor.py

@@ -22,6 +22,8 @@
 from reflex_base.event.processor.future import EventFuture
 from reflex_base.event.processor.timeout import DrainTimeoutManager
 from reflex_base.registry import RegisteredEventHandler, RegistrationContext
+from reflex_base.session import SessionToken
+from reflex_base.utils.compat import MISSING_TYPE
 
 logger = logging.getLogger(__name__)
 
@@ -207,6 +209,7 @@ async def emit_event(token: str, *events: Event) -> None:
 
         self._root_context = EventContext(
             token="",
+            session_token=SessionToken.SYSTEM,
             # Belongs to no event: the events forked from it are top-level.
             txid="",
             parent_txid=None,
@@ -384,14 +387,20 @@ def _ensure_queue_task(self) -> asyncio.Queue[EventQueueEntry]:
         return self._queue
 
     async def enqueue(
-        self, token: str, event: Event, ev_ctx: EventContext | None = None
+        self,
+        token: str,
+        event: Event,
+        ev_ctx: EventContext | None = None,
+        *,
+        session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING,
     ) -> EventFuture:
         """Enqueue an event to be processed.
 
         Args:
             token: The client token associated with the event.
             event: The event to be enqueued.
             ev_ctx: The event context to use for this event.
+            session_token: The session authorizing the event, or omitted to inherit it.
 
         Returns:
             An EventFuture that resolves to the result of the associated task.
@@ -405,12 +414,16 @@ async def enqueue(
         if ev_ctx is None:
             try:
                 ev_ctx = EventContext.get().fork(
-                    token=token, router_data=event.router_data
+                    token=token,
+                    router_data=event.router_data,
+                    session_token=session_token,
                 )
             except LookupError as le:
                 if self._root_context is not None:
                     ev_ctx = self._root_context.fork(
-                        token=token, router_data=event.router_data
+                        token=token,
+                        router_data=event.router_data,
+                        session_token=session_token,
                     )
                 else:
                     msg = "Event processor is not running, call .start(...) first."
@@ -476,6 +489,8 @@ async def enqueue_stream_delta(
         self,
         token: str,
         event: Event,
+        *,
+        session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING,
     ) -> AsyncGenerator[Mapping[str, Any]]:
         """Enqueue an event and yield the deltas its chain emits for ``token``.
 
@@ -494,6 +509,7 @@ async def enqueue_stream_delta(
         Args:
             token: The client token associated with the event.
             event: The event to be enqueued.
+            session_token: The session authorizing the event, or omitted to inherit it.
 
         Yields:
             Deltas for the token emitted by the event handler and the events it chains.
@@ -518,17 +534,19 @@ async def _emit_delta_impl(
                 # go to the client the usual way.
                 await root.emit_delta_impl(token, delta)
 
+        try:
+            parent_context = EventContext.get()
+        except LookupError:
+            parent_context = root
         task_future = await self.enqueue(
             token,
             event,
-            # A fresh context rather than a copy of the root, which has no txid:
-            # this event needs one of its own, so the events it chains find its
-            # future and concurrent streams stay apart. As a top-level event,
-            # it has no parent.
-            ev_ctx=type(root)(
-                token=token,
-                state_manager=root.state_manager,
-                enqueue_impl=root.enqueue_impl,
+            ev_ctx=dataclasses.replace(
+                parent_context.fork(
+                    token=token,
+                    router_data=event.router_data,
+                    session_token=session_token,
+                ),
                 emit_delta_impl=_emit_delta_impl,
                 emit_event_impl=root.emit_event_impl,
                 # Like fork(): the handler span nests under the caller's span

packages/reflex-base/src/reflex_base/utils/exceptions.py

@@ -28,6 +28,10 @@ class ReflexRuntimeError(ReflexError, RuntimeError):
     """Custom RuntimeError for Reflex."""
 
 
+class SessionAuthorizationError(ReflexError, PermissionError):
+    """Raised when a browser session does not own the requested client state."""
+
+
 class UploadTypeError(ReflexError, TypeError):
     """Custom TypeError for upload related errors."""
 

packages/reflex-components-core/news/+session-upload-authorization.feature.md

@@ -0,0 +1 @@
+File uploads now carry the browser session cookie and reject client tokens belonging to another session when `REFLEX_SESSION_TOKEN_MODE=enforce` is enabled.

packages/reflex-components-core/src/reflex_components_core/core/_upload.py

@@ -18,6 +18,7 @@
 from typing import TYPE_CHECKING, Any, BinaryIO, Self, cast
 
 from python_multipart.multipart import MultipartParser, parse_options_header
+from reflex_base.event.context import EventContext
 from reflex_base.registry import RegistrationContext
 from reflex_base.utils import exceptions
 from reflex_base.utils.format import json_dumps
@@ -667,7 +668,9 @@ async def _ndjson_updates():
         if disconnect_seen:
             return
         # Enqueue the task on the main event loop, but emit deltas to the local queue.
-        async for delta in app.event_processor.enqueue_stream_delta(token, event):
+        async for delta in app.event_processor.enqueue_stream_delta(
+            token, event, session_token=request.scope.get("reflex.session")
+        ):
             yield json_dumps(StateUpdate(delta=delta)) + "\n"
 
     return DisconnectAwareStreamingResponse(
@@ -724,7 +727,9 @@ async def _start_handler(extra_args_raw: str | None) -> None:
                 handler_upload_param[0]: chunk_iter,
             },
         )
-        task_future = await app.event_processor.enqueue(token, event)
+        task_future = await app.event_processor.enqueue(
+            token, event, session_token=request.scope.get("reflex.session")
+        )
         chunk_iter.set_consumer_task(task_future)
 
     parser = _UploadChunkMultipartParser(
@@ -843,6 +848,14 @@ async def upload_file(request: Request):
         )
 
         token, handler_name = _require_upload_headers(request)
+        try:
+            dataclasses.replace(
+                EventContext.get(),
+                token=token,
+                session_token=request.scope.get("reflex.session"),
+            )
+        except exceptions.SessionAuthorizationError as error:
+            raise HTTPException(status_code=403, detail=str(error)) from error
         registered_event_handler = RegistrationContext.get().event_handlers.get(
             handler_name
         )

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
   "wrapt >=1.17.0,<2.6",
   "reflex-base >= 0.10.1.dev0",
   "reflex-components-code >= 0.10.0",
-  "reflex-components-core >= 0.10.0",
+  "reflex-components-core >= 0.10.1.dev0",
   "reflex-components-dataeditor >= 0.10.0",
   "reflex-components-gridjs >= 0.10.0",
   "reflex-components-lucide >= 1.1.0",