From d8a70b156c0ae3d6861acb1503383b9382d18d0a Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Tue, 29 Sep 2026 17:31:58 -0700 Subject: [PATCH 1/5] Authorize browser state through session-aware HTTP and WebSocket transports --- news/+browser-session-transport.feature.md | 1 + ...ssion-context-authorization.deprecation.md | 1 + .../+session-context-authorization.feature.md | 1 + .../+upload-transaction-context.bugfix.md | 1 + .../src/reflex_base/event/context.py | 47 +- .../event/processor/event_processor.py | 40 +- .../src/reflex_base/utils/exceptions.py | 4 + .../+session-upload-authorization.feature.md | 1 + .../reflex_components_core/core/_upload.py | 17 +- reflex/app.py | 162 ++++- reflex/istate/shared.py | 2 + reflex/session.py | 326 ++++++++++ reflex/utils/token_manager.py | 23 +- tests/units/components/core/test_upload.py | 124 ++++ tests/units/istate/test_shared.py | 4 +- .../event/processor/test_event_processor.py | 20 + tests/units/reflex_base/event/test_context.py | 92 +++ tests/units/test_app.py | 287 ++++++++- tests/units/test_session.py | 561 ++++++++++++++++++ tests/units/utils/test_token_manager.py | 75 ++- 20 files changed, 1735 insertions(+), 54 deletions(-) create mode 100644 news/+browser-session-transport.feature.md create mode 100644 packages/reflex-base/news/+session-context-authorization.deprecation.md create mode 100644 packages/reflex-base/news/+session-context-authorization.feature.md create mode 100644 packages/reflex-base/news/+upload-transaction-context.bugfix.md create mode 100644 packages/reflex-components-core/news/+session-upload-authorization.feature.md create mode 100644 reflex/session.py create mode 100644 tests/units/test_session.py diff --git a/news/+browser-session-transport.feature.md b/news/+browser-session-transport.feature.md new file mode 100644 index 00000000000..5bf5abdece7 --- /dev/null +++ b/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 are validated for HTTP and WebSocket requests, and valid HTTP sessions refresh automatically. diff --git a/packages/reflex-base/news/+session-context-authorization.deprecation.md b/packages/reflex-base/news/+session-context-authorization.deprecation.md new file mode 100644 index 00000000000..79f49ce7b82 --- /dev/null +++ b/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`. diff --git a/packages/reflex-base/news/+session-context-authorization.feature.md b/packages/reflex-base/news/+session-context-authorization.feature.md new file mode 100644 index 00000000000..6e1169bc684 --- /dev/null +++ b/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. diff --git a/packages/reflex-base/news/+upload-transaction-context.bugfix.md b/packages/reflex-base/news/+upload-transaction-context.bugfix.md new file mode 100644 index 00000000000..842b18afc0e --- /dev/null +++ b/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. diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index 66a88ac3e27..d3ac01599ea 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/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.9.13", + 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, diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index 071655da946..14632463199 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/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,7 +387,12 @@ 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. @@ -392,6 +400,7 @@ async def enqueue( 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 = self._root_context 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 diff --git a/packages/reflex-base/src/reflex_base/utils/exceptions.py b/packages/reflex-base/src/reflex_base/utils/exceptions.py index 46d5fb863ea..7c932d9d655 100644 --- a/packages/reflex-base/src/reflex_base/utils/exceptions.py +++ b/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.""" diff --git a/packages/reflex-components-core/news/+session-upload-authorization.feature.md b/packages/reflex-components-core/news/+session-upload-authorization.feature.md new file mode 100644 index 00000000000..ef685f9dfb5 --- /dev/null +++ b/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. diff --git a/packages/reflex-components-core/src/reflex_components_core/core/_upload.py b/packages/reflex-components-core/src/reflex_components_core/core/_upload.py index 441d8143dcf..955f78e84d8 100644 --- a/packages/reflex-components-core/src/reflex_components_core/core/_upload.py +++ b/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 ) diff --git a/reflex/app.py b/reflex/app.py index 9bcb9892f59..715e042fbc1 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -49,8 +49,10 @@ from reflex_base.event.context import EventContext from reflex_base.event.processor import BaseStateEventProcessor, EventProcessor from reflex_base.registry import RegistrationContext +from reflex_base.session import SessionToken, SessionTokenManager from reflex_base.telemetry_context import CompileTrigger, TelemetryContext from reflex_base.utils import memo_paths +from reflex_base.utils.compat import MISSING_TYPE from reflex_base.utils.imports import ImportVar from reflex_base.utils.types import ASGIApp, Message, Receive, Scope, Send from reflex_base.vars.dep_tracking import is_dependency @@ -87,6 +89,12 @@ replace_brackets_with_keywords, verify_route_validity, ) +from reflex.session import ( + SessionMiddleware, + session_authorization_error, + session_endpoint, + trusted_session_origins, +) from reflex.state import BaseState, State, StateUpdate, all_base_state_classes from reflex.utils import ( codespaces, @@ -480,6 +488,10 @@ class App(MiddlewareMixin, LifespanMixin): # The backend API object. _api: Starlette | None = None + _session_token_manager: SessionTokenManager = dataclasses.field( + init=False, repr=False + ) + # The state class to use for the app. _state: type[BaseState] | None = None @@ -844,6 +856,10 @@ def __call__(self) -> ASGIApp: if isinstance(api_transformer, Starlette): # Mount the api to the starlette app. App._add_cors(api_transformer) + api_transformer.add_exception_handler( + exceptions.SessionAuthorizationError, + session_authorization_error, + ) api_transformer.mount("", asgi_app) asgi_app = api_transformer else: @@ -851,6 +867,11 @@ def __call__(self) -> ASGIApp: asgi_app = api_transformer(asgi_app) top_asgi_app = Starlette(lifespan=self._run_lifespan_tasks) + top_asgi_app.add_middleware( + SessionMiddleware, + manager=self._session_token_manager, + api_url=config.api_url, + ) # Make sure Reflex contexts are attached for each request. top_asgi_app.add_middleware(_ContextMiddleware, reflex_app=self) top_asgi_app.mount("", asgi_app) @@ -860,12 +881,25 @@ def __call__(self) -> ASGIApp: return top_asgi_app def _add_default_endpoints(self): - """Add default api endpoints (ping).""" + """Add the framework's HTTP endpoints.""" # To test the server. if not self._api: return config = get_config() + self._session_token_manager = SessionTokenManager( + config.app_name, + ttl=config.session_token_ttl, + refresh_interval=config.session_token_refresh_interval, + ) + self._api.add_exception_handler( + exceptions.SessionAuthorizationError, session_authorization_error + ) + self._api.add_route( + config.prepend_backend_path("/_reflex/session"), + self._issue_session, + methods=["POST"], + ) self._api.add_route( config.prepend_backend_path(str(constants.Endpoint.PING)), ping, @@ -877,6 +911,23 @@ def _add_default_endpoints(self): methods=["GET"], ) + async def _issue_session(self, request: Request) -> Response: + """Issue a browser session using the current public URLs. + + Args: + request: The incoming session request. + + Returns: + Session metadata and the HttpOnly cookie. + """ + config = get_config() + return await session_endpoint( + request, + manager=self._session_token_manager, + api_url=config.api_url, + allowed_origins=trusted_session_origins(config), + ) + def _add_optional_endpoints(self): """Add optional api endpoints (_upload).""" from reflex_components_core.core.upload import Upload, get_upload_dir @@ -1823,6 +1874,8 @@ def modify_state( token: str, background: bool = False, previous_dirty_vars: dict[str, set[str]] | None = None, + *, + session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING, ) -> contextlib.AbstractAsyncContextManager[BaseState]: ... @overload @@ -1831,6 +1884,8 @@ def modify_state( token: BaseStateToken, background: bool = False, previous_dirty_vars: dict[str, set[str]] | None = None, + *, + session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING, ) -> contextlib.AbstractAsyncContextManager[BaseState]: ... @contextlib.asynccontextmanager @@ -1839,6 +1894,8 @@ async def modify_state( token: BaseStateToken | str, background: bool = False, previous_dirty_vars: dict[str, set[str]] | None = None, + *, + session_token: SessionToken | MISSING_TYPE | None = dataclasses.MISSING, **context: Unpack[StateModificationContext], ) -> AsyncIterator[BaseState]: """Modify the state out of band. @@ -1847,6 +1904,7 @@ async def modify_state( token: The token to modify the state for. background: Whether the modification is happening in a background task. previous_dirty_vars: Vars that are considered dirty from a previous operation. + session_token: The session authorizing access, or omitted to inherit the current context. Yields: The state to modify. @@ -1869,7 +1927,9 @@ async def modify_state( # shared-state fan-out runs in a task that copied the triggering # event's context for a different client. No-op without an EventContext. try: - forked_context = EventContext.get().fork(token=token.ident) + forked_context = EventContext.get().fork( + token=token.ident, session_token=session_token + ) except LookupError: pass else: @@ -2129,6 +2189,9 @@ def __init__(self, namespace: str, app: App): # connect time instead of for every event on the connection. self._static_router_data: dict[str, dict[str, Any]] = {} + self._sessions: dict[str, SessionToken] = {} + self._session_refresh_at: dict[str, float] = {} + # Start time and count of the current process-wide client_error window. self._client_error_window_start = 0.0 self._client_error_window_count = 0 @@ -2199,14 +2262,33 @@ async def on_connect(self, sid: str, environ: dict, auth: Any = None): ) query_params = urllib.parse.parse_qs(environ.get("QUERY_STRING", "")) token_list = query_params.get("token", []) - if token_list: - # The boot event is the connection's first event: processing it - # records the new sid and token on the state under the state lock. - await self.link_token_to_sid( - sid, token_list[0], update_state=boot_event is None - ) + token = token_list[0] if token_list else "" + mode = environment.REFLEX_SESSION_TOKEN_MODE.get() + issued_session = None + if mode != "off": + manager = self.app._session_token_manager + session = environ.get("asgi.scope", {}).get("reflex.session") + issue_session = session is None + if session is None: + session = manager.create() + self._sessions[sid] = session + self._session_refresh_at[sid] = session.issued_at + manager.refresh_interval + if not token or (mode == "enforce" and not session.authorizes(token)): + token = manager.create_client_token(session) + await self.emit("new_token", token, to=sid) + if issue_session: + issued_session = session + if token: + # The boot event records the new sid and token under its state lock. + await self.link_token_to_sid(sid, token, update_state=boot_event is None) else: logger.warning(f"No token provided in connection for session {sid}") + if issued_session is not None: + await self.emit( + "session_token", + self.app._session_token_manager.encode(issued_session), + to=sid, + ) subprotocol = environ.get("HTTP_SEC_WEBSOCKET_PROTOCOL") if subprotocol and subprotocol != constants.Reflex.VERSION: @@ -2228,6 +2310,8 @@ async def on_connect(self, sid: str, environ: dict, auth: Any = None): # token link and connection data made above before the error # refuses the connect. self._static_router_data.pop(sid, None) + self._sessions.pop(sid, None) + self._session_refresh_at.pop(sid, None) if (linked_token := self.sid_to_token.get(sid)) is not None: await self._token_manager.disconnect_token(linked_token, sid) raise @@ -2283,6 +2367,8 @@ def on_disconnect(self, sid: str) -> asyncio.Task | None: otel.record_connection(-1) self._client_error_counts.pop(sid, None) self._static_router_data.pop(sid, None) + self._sessions.pop(sid, None) + self._session_refresh_at.pop(sid, None) # Get token before cleaning up disconnect_token = self.sid_to_token.get(sid) if disconnect_token: @@ -2351,6 +2437,13 @@ async def on_event(self, sid: str, data: Any): ) return + session = self._sessions.get(sid) + if session is not None and time.time() >= self._session_refresh_at[sid]: + self._session_refresh_at[sid] = ( + time.time() + self.app._session_token_manager.refresh_interval + ) + await self.emit("session_refresh", to=sid) + fields = data if isinstance(fields, str): @@ -2409,10 +2502,10 @@ async def on_event(self, sid: str, data: Any): else "404" ).removeprefix("/") if not otel.enabled: - await self.app.event_processor.enqueue(token, event) + await self.app.event_processor.enqueue(token, event, session_token=session) return with otel.remote_context(fields): - await self.app.event_processor.enqueue(token, event) + await self.app.event_processor.enqueue(token, event, session_token=session) async def on_ping(self, sid: str): """Event for testing the API endpoint. @@ -2517,7 +2610,22 @@ async def link_token_to_sid( without an extra load and save of the whole state tree. """ # Use TokenManager for duplicate detection and Redis support - new_token = await self._token_manager.link_token_to_sid(token, sid) + session = self._sessions.get(sid) + if session is not None: + # Validate before linking or loading any client state. + context = dataclasses.replace( + EventContext.get(), token=token, session_token=session + ) + with context: + new_token = await self._token_manager.link_token_to_sid( + token, + sid, + token_factory=functools.partial( + self.app._session_token_manager.create_client_token, session + ), + ) + else: + new_token = await self._token_manager.link_token_to_sid(token, sid) if new_token: # Duplicate detected, emit new token to client @@ -2525,15 +2633,23 @@ async def link_token_to_sid( # Update client state to apply new sid/token for running background tasks. if update_state and self.app._state is not None: - async with self.app.state_manager.modify_state( - BaseStateToken(ident=new_token or token, cls=self.app._state) - ) as state: - state.router_data[constants.RouteVar.SESSION_ID] = sid - # Record the identity the state was loaded under; duplicate-token - # handling can hand back a fresh one here. - state.router_data[constants.RouteVar.CLIENT_TOKEN] = new_token or token - # Rebuild from router_data to keep the session var in step with it. - if ( - session := SessionData.from_router_data(state.router_data) - ) != state.rx_router_session: - state.rx_router_session = session + context = EventContext.get().fork( + token=new_token or token, session_token=session + ) + with context: + async with self.app.state_manager.modify_state( + BaseStateToken(ident=new_token or token, cls=self.app._state) + ) as state: + state.router_data[constants.RouteVar.SESSION_ID] = sid + # Record the identity the state was loaded under; duplicate-token + # handling can hand back a fresh one here. + state.router_data[constants.RouteVar.CLIENT_TOKEN] = ( + new_token or token + ) + # Rebuild from router_data to keep the session var in step with it. + if ( + router_session := SessionData.from_router_data( + state.router_data + ) + ) != state.rx_router_session: + state.rx_router_session = router_session diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index a4d24de0657..7f2c1a22008 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -10,6 +10,7 @@ from reflex_base.constants import ROUTER_DATA, ROUTER_VARS from reflex_base.event import Event, get_hydrate_event from reflex_base.registry import RegistrationContext +from reflex_base.session import SessionToken from reflex_base.utils.exceptions import ReflexRuntimeError from reflex_base.vars.base import _owner_state @@ -69,6 +70,7 @@ async def _update_client(token: str): async with app.modify_state( BaseStateToken(ident=token, cls=state_type), previous_dirty_vars=previous_dirty_vars, + session_token=SessionToken.SYSTEM, ): pass diff --git a/reflex/session.py b/reflex/session.py new file mode 100644 index 00000000000..6763f1a401e --- /dev/null +++ b/reflex/session.py @@ -0,0 +1,326 @@ +"""HTTP and WebSocket transport for browser session credentials.""" + +from __future__ import annotations + +import dataclasses +import json +import logging +from contextlib import nullcontext +from ipaddress import ip_address +from typing import TYPE_CHECKING +from urllib.parse import urlsplit + +from reflex_base import constants +from reflex_base.environment import environment +from reflex_base.event.context import EventContext +from reflex_base.session import SessionToken, SessionTokenManager +from reflex_base.utils.exceptions import SessionAuthorizationError +from reflex_base.utils.types import ASGIApp, Message, Receive, Scope, Send +from starlette.requests import Request +from starlette.responses import JSONResponse + +if TYPE_CHECKING: + from reflex_base.config import Config + +logger = logging.getLogger(__name__) + + +def _origin(url: str) -> str: + """Extract a normalized HTTP origin from a configured URL. + + Args: + url: The configured frontend or API URL. + + Returns: + The URL's origin, or an empty string for non-HTTP URLs. + """ + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + return "" + host = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname + port = parsed.port + if port is not None and port != (443 if parsed.scheme == "https" else 80): + host += f":{port}" + return f"{parsed.scheme}://{host}" + + +def trusted_session_origins(config: Config) -> tuple[str, ...]: + """Get the explicit browser origins allowed to install session credentials. + + A wildcard CORS policy cannot authorize credential exchange: accepting a + session chosen by another origin would permit session fixation. + + Args: + config: The application configuration. + + Returns: + The trusted frontend, API, and explicitly configured CORS origins. + """ + urls = [config.api_url, config.deploy_url or "", *config.cors_allowed_origins] + if environment.REFLEX_ENV_MODE.get() == constants.Env.DEV: + port = config.frontend_port or constants.DefaultPorts.FRONTEND_PORT + urls.extend( + f"http://{host}:{port}" for host in ("localhost", "127.0.0.1", "[::1]") + ) + return tuple(dict.fromkeys(origin for url in urls if (origin := _origin(url)))) + + +def _secure_cookie(api_url: str) -> bool: + """Choose cookie security from the public API URL, including behind proxies. + + Args: + api_url: The configured externally reachable backend URL. + + Returns: + False only for development HTTP on a non-loopback host. + """ + url = urlsplit(api_url) + if url.scheme != "http" or environment.REFLEX_ENV_MODE.get() != constants.Env.DEV: + return True + hostname = url.hostname or "" + if hostname == "localhost" or hostname.endswith(".localhost"): + return True + try: + return ip_address(hostname).is_loopback + except ValueError: + return False + + +def _cookie_name(manager: SessionTokenManager, secure: bool) -> str: + """Get the app-specific cookie name for this transport. + + Args: + manager: The app's session manager. + secure: Whether the cookie uses the secure host prefix. + + Returns: + The cookie name. + """ + return f"__Host-{manager.cookie_name}" if secure else manager.cookie_name + + +def _session_cookie( + manager: SessionTokenManager, session: SessionToken, secure: bool +) -> bytes: + """Serialize a session cookie without requiring Python's Partitioned support. + + Args: + manager: The app's session manager. + session: The session to encode. + secure: Whether to use a secure, partitioned cookie. + + Returns: + The Set-Cookie header value. + """ + attributes = "SameSite=None; Secure; Partitioned" if secure else "SameSite=Lax" + return ( + f"{_cookie_name(manager, secure)}={manager.encode(session)}; " + f"Path=/; Max-Age={manager.ttl}; HttpOnly; {attributes}" + ).encode("ascii") + + +class SessionMiddleware: + """Validate sessions, hide their credentials, and scope HTTP state access.""" + + def __init__(self, app: ASGIApp, manager: SessionTokenManager, api_url: str): + """Wrap an ASGI app with browser session handling. + + Args: + app: The next ASGI application. + manager: The app's session manager. + api_url: The configured externally reachable backend URL. + """ + self.app = app + self.manager = manager + self.secure = _secure_cookie(api_url) + self.cookie_name = _cookie_name(manager, self.secure) + self._cookie_names = {manager.cookie_name, f"__Host-{manager.cookie_name}"} + if not self.secure: + logger.warning( + "Session cookies use insecure SameSite=Lax on this development HTTP " + "API URL. Use HTTPS for secure, cross-site session cookies." + ) + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + """Validate credentials before invoking HTTP or WebSocket handlers. + + Args: + scope: The incoming ASGI scope. + receive: The request message receiver. + send: The response message sender. + + Raises: + SessionAuthorizationError: If state access fails after a response started. + """ + if scope["type"] not in {"http", "websocket"}: + await self.app(scope, receive, send) + return + + credentials = [] + headers = [] + for name, value in scope.get("headers", []): + if name.lower() != b"cookie": + headers.append((name, value)) + continue + remaining = [] + for part in value.decode("latin-1").split(";"): + cookie_name, _, cookie_value = part.strip().partition("=") + if cookie_name in self._cookie_names: + if cookie_name == self.cookie_name: + credentials.append(cookie_value) + else: + remaining.append(part) + if remaining: + headers.append((name, ";".join(remaining).strip().encode("latin-1"))) + + scope["headers"] = headers + session = self.manager.decode(credentials[0]) if len(credentials) == 1 else None + scope["reflex.session"] = session + if scope["type"] == "websocket": + await self.app(scope, receive, send) + return + + response_started = False + + async def send_with_session(message: Message) -> None: + """Refresh valid sessions when the HTTP response has no session cookie. + + Args: + message: The outgoing ASGI message. + """ + nonlocal response_started + if message["type"] == "http.response.start": + response_started = True + response_headers = message.setdefault("headers", []) + if ( + session is not None + and self.manager.should_refresh(session) + and not any( + name.lower() == b"set-cookie" + and value.startswith(f"{self.cookie_name}=".encode("ascii")) + for name, value in response_headers + ) + ): + response_headers.append(( + b"set-cookie", + _session_cookie( + self.manager, self.manager.refresh(session), self.secure + ), + )) + await send(message) + + try: + context = dataclasses.replace( + EventContext.get(), token="", session_token=session + ) + except LookupError: + context = nullcontext() + with context: + try: + await self.app(scope, receive, send_with_session) + except SessionAuthorizationError as error: + if response_started: + raise + response = await session_authorization_error(Request(scope), error) + await response(scope, receive, send_with_session) + + +async def session_authorization_error( # noqa: RUF029 + _request: Request, _exception: Exception +) -> JSONResponse: + """Reject unauthorized state access without disclosing credential details. + + Args: + _request: The incoming request. + _exception: The authorization failure. + + Returns: + A no-store forbidden response. + """ + return JSONResponse( + {"detail": "The session does not authorize this client token."}, + status_code=403, + headers={"Cache-Control": "no-store"}, + ) + + +async def session_endpoint( + request: Request, + *, + manager: SessionTokenManager, + api_url: str, + allowed_origins: tuple[str, ...] = (), +) -> JSONResponse: + """Create or renew a browser session and return a bound client token. + + The JSON response never includes the session credential. Non-simple fetch + requests are required so browser cross-origin requests must pass CORS. + Installing a credential supplied by a WebSocket bootstrap additionally + requires an explicit trusted Origin to prevent session fixation. + + Args: + request: The incoming POST request, after SessionMiddleware. + manager: The app's session manager. + api_url: The configured externally reachable backend URL. + allowed_origins: Explicit browser origins trusted to exchange credentials. + + Returns: + A no-store response with session metadata or a request rejection. + """ + if request.headers.get("sec-fetch-mode") == "navigate" or ( + request.headers.get("content-type", "").split(";", 1)[0].strip().lower() + != "application/json" + and "reflex-client-token" not in request.headers + ): + return JSONResponse( + {"detail": "Use a JSON fetch request to establish a session."}, + status_code=403, + headers={"Cache-Control": "no-store"}, + ) + + try: + payload = json.loads(await request.body() or b"{}") + except (ValueError, UnicodeError): + payload = None + if not isinstance(payload, dict) or ( + "session_token" in payload and not isinstance(payload["session_token"], str) + ): + return JSONResponse( + {"detail": "Expected a JSON object with an optional session_token string."}, + status_code=400, + headers={"Cache-Control": "no-store"}, + ) + + session = request.scope.get("reflex.session") + if session is None and "session_token" in payload: + origin = request.headers.get("origin") + if origin is None or origin == "null" or origin not in allowed_origins: + return JSONResponse( + {"detail": "Session exchange requires a trusted Origin."}, + status_code=403, + headers={"Cache-Control": "no-store"}, + ) + session = manager.decode(payload["session_token"]) + if session is None: + return JSONResponse( + {"detail": "Invalid or expired session token."}, + status_code=401, + headers={"Cache-Control": "no-store"}, + ) + session = manager.refresh(session) if session is not None else manager.create() + client_token = request.headers.get("reflex-client-token", "") + if not session.authorizes(client_token): + client_token = manager.create_client_token(session) + response = JSONResponse( + { + "token_type": "cookie", + "expires_in": manager.ttl, + "client_token": client_token, + }, + headers={"Cache-Control": "no-store"}, + ) + response.raw_headers.append(( + b"set-cookie", + _session_cookie(manager, session, _secure_cookie(api_url)), + )) + return response diff --git a/reflex/utils/token_manager.py b/reflex/utils/token_manager.py index afb1685a5ad..8da1ac95515 100644 --- a/reflex/utils/token_manager.py +++ b/reflex/utils/token_manager.py @@ -97,12 +97,15 @@ async def is_token_connected(self, token: str) -> bool: return token in self.token_to_socket @abstractmethod - async def link_token_to_sid(self, token: str, sid: str) -> str | None: + async def link_token_to_sid( + self, token: str, sid: str, *, token_factory: Callable[[], str] = _get_new_token + ) -> str | None: """Link a token to a session ID. Args: token: The client token. sid: The Socket.IO session ID. + token_factory: Create a replacement token when a duplicate tab connects. Returns: New token if duplicate detected and new token generated, None otherwise. @@ -155,12 +158,15 @@ def __init__(self): """Initialize the local token manager.""" super().__init__() - async def link_token_to_sid(self, token: str, sid: str) -> str | None: + async def link_token_to_sid( + self, token: str, sid: str, *, token_factory: Callable[[], str] = _get_new_token + ) -> str | None: """Link a token to a session ID. Args: token: The client token. sid: The Socket.IO session ID. + token_factory: Create a replacement token when a duplicate tab connects. Returns: New token if duplicate detected and new token generated, None otherwise. @@ -169,7 +175,7 @@ async def link_token_to_sid(self, token: str, sid: str) -> str | None: if ( socket_record := self.token_to_socket.get(token) ) is not None and sid != socket_record.sid: - new_token = _get_new_token() + new_token = token_factory() self.token_to_socket[new_token] = SocketRecord( instance_id=self.instance_id, sid=sid ) @@ -312,12 +318,15 @@ def _ensure_socket_record_task(self) -> None: suppress_exceptions=[Exception], ) - async def link_token_to_sid(self, token: str, sid: str) -> str | None: + async def link_token_to_sid( + self, token: str, sid: str, *, token_factory: Callable[[], str] = _get_new_token + ) -> str | None: """Link a token to a session ID with Redis-based duplicate detection. Args: token: The client token. sid: The Socket.IO session ID. + token_factory: Create a replacement token when a duplicate tab connects. Returns: New token if duplicate detected and new token generated, None otherwise. @@ -338,12 +347,14 @@ async def link_token_to_sid(self, token: str, sid: str) -> str | None: token_exists_in_redis = await self.redis.exists(redis_key) except Exception as e: logger.error(f"Redis error checking token existence: {e}") - return await super().link_token_to_sid(token, sid) + return await super().link_token_to_sid( + token, sid, token_factory=token_factory + ) new_token = None if token_exists_in_redis: # Duplicate exists somewhere - generate new token - token = new_token = _get_new_token() + token = new_token = token_factory() # Store in local dicts socket_record = self.token_to_socket[token] = SocketRecord( diff --git a/tests/units/components/core/test_upload.py b/tests/units/components/core/test_upload.py index 1fd26e09135..55d9d483aab 100644 --- a/tests/units/components/core/test_upload.py +++ b/tests/units/components/core/test_upload.py @@ -3,9 +3,12 @@ import json from pathlib import Path from typing import Any, cast +from unittest.mock import AsyncMock, Mock import pytest from reflex_base.event import EventChain, EventHandler, EventSpec, parse_args_spec +from reflex_base.event.context import EventContext +from reflex_base.session import SessionToken, SessionTokenManager from reflex_base.vars import VarData from reflex_base.vars.base import LiteralVar, Var from reflex_components_core.core._upload import ( @@ -16,6 +19,7 @@ _sanitize_upload_filename, _upload_file_from_starlette, _UploadChunkMultipartParser, + upload, ) from reflex_components_core.core.upload import ( GhostUpload, @@ -30,6 +34,7 @@ from starlette.datastructures import UploadFile as StarletteUploadFile from starlette.exceptions import HTTPException from starlette.formparsers import MultiPartException +from starlette.requests import Request import reflex as rx from reflex import event @@ -91,6 +96,125 @@ async def chunk_drop_handler_missing_annotation(self, chunk_iter): """Invalid handler missing the UploadChunkIterator annotation.""" +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("mode", ["off", "warn", "enforce"]) +@pytest.mark.parametrize("authorization", ["owned", "foreign", "anonymous"]) +async def test_upload_session_authorization( + monkeypatch: pytest.MonkeyPatch, + streaming: bool, + mode: str, + authorization: str, +): + """Upload requests validate before body reads and propagate their own session. + + Args: + monkeypatch: Environment fixture. + streaming: Whether to dispatch a streaming upload handler. + mode: The authorization policy. + authorization: The request session's relationship to the client token. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", mode) + codec = SessionTokenManager("upload", secrets=(b"s" * 32,)) + session = codec.create() + token = codec.create_client_token(session) + request_session = { + "owned": session, + "foreign": codec.create(), + "anonymous": None, + }[authorization] + handler = ( + f"{StreamingUploadStateTest.get_full_name()}.chunk_drop_handler" + if streaming + else f"{UploadStateTest.get_full_name()}.upload_alias_handler" + ) + body = ( + b'--upload\r\nContent-Disposition: form-data; name="files"; filename="test.txt"\r\n' + b"Content-Type: text/plain\r\n\r\ncontent\r\n--upload--\r\n" + ) + receive = AsyncMock( + return_value={"type": "http.request", "body": body, "more_body": False} + ) + request = Request( + { + "type": "http", + "method": "POST", + "asgi": {"spec_version": "2.4"}, + "headers": [ + (b"reflex-client-token", token.encode()), + (b"reflex-event-handler", handler.encode()), + (b"content-type", b"multipart/form-data; boundary=upload"), + ], + "reflex.session": request_session, + }, + receive=receive, + ) + dispatched = [] + tasks = [] + + async def buffered(client_token, event, *, session_token): # noqa: RUF029 + """Record the request identity forwarded to the buffered event stream. + + Args: + client_token: The upload's client token. + event: The upload event. + session_token: The authorizing request session. + + Yields: + An empty state delta. + """ + dispatched.append((client_token, session_token)) + yield {} + + async def enqueue(client_token, event, *, session_token): # noqa: RUF029 + """Drain streamed chunks and record the dispatch context. + + Args: + client_token: The upload's client token. + event: The upload event. + session_token: The authorizing request session. + + Returns: + The background upload task. + """ + dispatched.append((client_token, session_token)) + + async def consume(): + """Drain all chunks so the multipart parser can finish.""" + async for _ in event.payload["chunk_iter"]: + pass + + task = asyncio.create_task(consume()) + tasks.append(task) + return task + + app = Mock( + event_processor=Mock( + enqueue=AsyncMock(side_effect=enqueue), enqueue_stream_delta=buffered + ) + ) + root = EventContext( + token="", + session_token=SessionToken.SYSTEM, + state_manager=Mock(), + enqueue_impl=AsyncMock(), + ) + with root: + if mode == "enforce" and authorization != "owned": + with pytest.raises(HTTPException) as error: + await upload(app)(request) + assert error.value.status_code == 403 + receive.assert_not_awaited() + app.event_processor.enqueue.assert_not_awaited() + assert dispatched == [] + else: + response = await upload(app)(request) + await response(request.scope, receive, AsyncMock()) + if tasks: + await asyncio.gather(*tasks) + assert dispatched == [(token, request_session)] + assert not app.state_manager.mock_calls + + def test_cancel_upload(): spec = cancel_upload("foo_id") assert isinstance(spec, EventSpec) diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index e7cb2e73010..0ba02aadf70 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, Mock, patch import pytest +from reflex_base.session import SessionToken import reflex as rx from reflex.istate.shared import ( @@ -58,7 +59,8 @@ def _mock_app(token_manager) -> tuple[Mock, list[str]]: modified_tokens: list[str] = [] @asynccontextmanager - async def modify_state(token, previous_dirty_vars=None): + async def modify_state(token, previous_dirty_vars=None, *, session_token): + assert session_token is SessionToken.SYSTEM modified_tokens.append(token.ident) yield Mock() diff --git a/tests/units/reflex_base/event/processor/test_event_processor.py b/tests/units/reflex_base/event/processor/test_event_processor.py index e7a389b4392..6a199372752 100644 --- a/tests/units/reflex_base/event/processor/test_event_processor.py +++ b/tests/units/reflex_base/event/processor/test_event_processor.py @@ -908,6 +908,26 @@ async def test_stream_delta_yields_single_delta(token: str): assert collected == [{"state": {"x": 1}}] +async def test_stream_delta_forks_ambient_transaction(mocker: MockerFixture): + """A buffered upload must not reuse the root transaction ID. + + Args: + mocker: The mock fixture. + """ + ep = EventProcessor(graceful_shutdown_timeout=2).configure() + enqueue = mocker.spy(EventProcessor, "enqueue") + async with ep: + parent = EventContext.get().fork(token="upload") + with parent: + event = Event.from_event_type(delta_event())[0] + assert [d async for d in ep.enqueue_stream_delta("upload", event)] + upload_context = enqueue.call_args.kwargs["ev_ctx"] + assert ep._root_context is not None + assert upload_context.txid != ep._root_context.txid + assert upload_context.txid != parent.txid + assert upload_context.parent_txid == parent.txid + + async def test_stream_delta_yields_multiple_deltas(token: str): """enqueue_stream_delta yields all deltas in order. diff --git a/tests/units/reflex_base/event/test_context.py b/tests/units/reflex_base/event/test_context.py index 46558b0bf04..1d1d12d3f86 100644 --- a/tests/units/reflex_base/event/test_context.py +++ b/tests/units/reflex_base/event/test_context.py @@ -3,7 +3,10 @@ import dataclasses from unittest import mock +import pytest from reflex_base.event.context import EventContext +from reflex_base.session import SessionToken, SessionTokenManager +from reflex_base.utils.exceptions import SessionAuthorizationError def test_fork_creates_child(mock_root_event_context: EventContext): @@ -98,3 +101,92 @@ async def test_emit_event_noop_when_no_impl(): emit_event_impl=None, ) await ctx.emit_event() + + +@pytest.mark.parametrize("operation", ["constructor", "fork", "replace"]) +@pytest.mark.parametrize("authorized", [True, False]) +def test_context_session_authorization( + mock_root_event_context: EventContext, + monkeypatch: pytest.MonkeyPatch, + operation: str, + authorized: bool, +): + """Every context creation path checks ownership. + + Args: + mock_root_event_context: The root context fixture. + monkeypatch: The environment patch fixture. + operation: The creation path under test. + authorized: Whether the client token belongs to the session. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + manager = SessionTokenManager("test", secrets=(b"a" * 32,)) + session = manager.create() + token = manager.create_client_token(session if authorized else manager.create()) + parent = dataclasses.replace(mock_root_event_context, session_token=session) + + def create_context() -> EventContext: + """Exercise the selected context construction path. + + Returns: + The constructed context. + """ + if operation == "fork": + return parent.fork(token=token) + if operation == "replace": + return dataclasses.replace(parent, token=token) + return EventContext( + token=token, + session_token=session, + state_manager=parent.state_manager, + enqueue_impl=parent.enqueue_impl, + ) + + if authorized: + assert create_context().session_token is session + else: + with pytest.raises(SessionAuthorizationError): + create_context() + + +@pytest.mark.parametrize("mode", ["off", "warn", "enforce"]) +def test_missing_session_modes( + mock_root_event_context: EventContext, + monkeypatch: pytest.MonkeyPatch, + mode: str, +): + """Missing sessions warn, pass, or fail according to the rollout mode. + + Args: + mock_root_event_context: The root context fixture. + monkeypatch: The environment patch fixture. + mode: The rollout mode. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", mode) + with mock.patch("reflex_base.event.context.console.deprecate") as warning: + if mode == "enforce": + with pytest.raises(SessionAuthorizationError): + dataclasses.replace( + mock_root_event_context, token="legacy", session_token=None + ) + else: + dataclasses.replace( + mock_root_event_context, token="legacy", session_token=None + ) + assert warning.call_count == (mode == "warn") + + +def test_system_session_and_explicit_fanout( + mock_root_event_context: EventContext, monkeypatch: pytest.MonkeyPatch +): + """Trusted fan-out explicitly overrides the ambient session. + + Args: + mock_root_event_context: The root context fixture. + monkeypatch: The environment patch fixture. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + parent = dataclasses.replace(mock_root_event_context, session_token=None) + child = parent.fork(token="other-client", session_token=SessionToken.SYSTEM) + assert child.session_token is SessionToken.SYSTEM + assert child.fork().session_token is SessionToken.SYSTEM diff --git a/tests/units/test_app.py b/tests/units/test_app.py index 6227420b31c..25b9a5cf5b8 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -4,6 +4,7 @@ import builtins import contextlib import contextvars +import dataclasses import functools import io import json @@ -40,6 +41,7 @@ from reflex_base.event.processor import BaseStateEventProcessor from reflex_base.plugins import CompileContext, CompilerHooks, PageContext, Plugin from reflex_base.registry import RegistrationContext +from reflex_base.session import SessionToken, SessionTokenManager from reflex_base.style import Style from reflex_base.utils import exceptions, format, memo_paths from reflex_base.utils.imports import ImportVar @@ -86,6 +88,7 @@ from reflex.istate.manager.token import BaseStateToken from reflex.istate.storage import Cookie, LocalStorage, SessionStorage from reflex.model import Model +from reflex.session import SessionMiddleware from reflex.state import ( BaseState, OnLoadInternalState, @@ -1508,6 +1511,7 @@ async def send(_message): @pytest.mark.asyncio +@pytest.mark.usefixtures("attached_mock_event_context") async def test_upload_file_raises_client_disconnect_when_stream_send_fails( token: str, ): @@ -1536,7 +1540,8 @@ async def form(): # noqa: RUF029 stream_closed = asyncio.Event() - async def enqueue_stream_delta(_token, _event): + async def enqueue_stream_delta(_token, _event, *, session_token): + assert session_token is request_mock.scope.get("reflex.session") try: yield {"state": {"ok": True}} await asyncio.Event().wait() @@ -1576,6 +1581,7 @@ async def send(message): @pytest.mark.asyncio +@pytest.mark.usefixtures("attached_mock_event_context") @pytest.mark.parametrize( "state", [FileUploadState, ChildFileUploadState, GrandChildFileUploadState], @@ -1619,6 +1625,7 @@ async def form(): # noqa: RUF029 @pytest.mark.asyncio +@pytest.mark.usefixtures("attached_mock_event_context") async def test_upload_file_unknown_handler_returns_400( token: str, ): @@ -1650,6 +1657,7 @@ async def test_upload_file_unknown_handler_returns_400( @pytest.mark.asyncio +@pytest.mark.usefixtures("attached_mock_event_context") @pytest.mark.parametrize( "state", [FileUploadState, ChildFileUploadState, GrandChildFileUploadState], @@ -4134,6 +4142,69 @@ async def test_modify_state_rebinds_event_context_to_token( EventContext.reset(actor_token) +@pytest.mark.parametrize( + "authorization", ["inherited", "explicit", "system", "anonymous", "foreign"] +) +async def test_modify_state_session_authorization( + app_with_processor: App, + monkeypatch: pytest.MonkeyPatch, + mocker: MockerFixture, + authorization: str, +): + """Out-of-band state access validates inherited and explicit sessions before reads. + + Args: + app_with_processor: The application fixture. + monkeypatch: Environment fixture. + mocker: Mock fixture. + authorization: How the caller supplies its session. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + app_with_processor._state_manager = StateManagerMemory() + app_with_processor._event_namespace = AsyncMock() + manager = SessionTokenManager("modify", secrets=(b"s" * 32,)) + session = manager.create() + token = manager.create_client_token(session) + assert app_with_processor._event_processor is not None + root = app_with_processor._event_processor._root_context + assert root is not None + root = dataclasses.replace( + root, + session_token=session if authorization == "inherited" else SessionToken.SYSTEM, + ) + kwargs = ( + {} + if authorization == "inherited" + else { + "session_token": { + "explicit": session, + "system": SessionToken.SYSTEM, + "anonymous": None, + "foreign": manager.create(), + }[authorization], + } + ) + modify = mocker.spy(app_with_processor.state_manager, "modify_state_with_links") + with root: + if authorization in {"anonymous", "foreign"}: + with pytest.raises(exceptions.SessionAuthorizationError): + async with app_with_processor.modify_state( + BaseStateToken(ident=token, cls=EmptyState), **kwargs + ): + pytest.fail("Unauthorized state access was permitted") + modify.assert_not_called() + else: + async with app_with_processor.modify_state( + BaseStateToken(ident=token, cls=EmptyState), **kwargs + ): + assert EventContext.get().session_token is ( + SessionToken.SYSTEM if authorization == "system" else session + ) + assert EventContext.get().token == token + modify.assert_called_once() + assert EventContext.get() is root + + def test_set_contexts_no_event_processor(isolated_context: contextvars.Context): """When event processor is None, EventContext should not be touched.""" @@ -4820,15 +4891,22 @@ def test_client_error_constants_match_frontend(): @pytest_asyncio.fixture -async def event_namespace_with_processor_mock() -> AsyncGenerator[EventNamespace, None]: +async def event_namespace_with_processor_mock( + attached_mock_event_context: EventContext, +) -> AsyncGenerator[EventNamespace, None]: """An EventNamespace whose app has a mocked event processor. + Args: + attached_mock_event_context: The active event context. + Yields: The EventNamespace instance. """ app = App() app._event_processor = Mock(enqueue=AsyncMock()) + app._session_token_manager = SessionTokenManager("test", secrets=(b"s" * 32,)) event_namespace = EventNamespace("/event", app) + event_namespace.emit = AsyncMock() yield event_namespace # The token manager is backed by redis when one is configured; drop the # tokens these tests link so they do not show up in another test's @@ -4884,6 +4962,10 @@ async def test_on_event_uses_connect_time_router_data( enqueue_mock = cast(AsyncMock, event_namespace.app.event_processor.enqueue) enqueue_mock.assert_called_once() + assert ( + enqueue_mock.call_args.kwargs["session_token"] + is event_namespace._sessions["sid1"] + ) enqueued_token, event = enqueue_mock.call_args[0] assert enqueued_token == token assert event.router_data[constants.RouteVar.CLIENT_TOKEN] == token @@ -5016,6 +5098,7 @@ async def test_on_event_falls_back_to_environ_without_connect( enqueue_mock = cast(AsyncMock, event_namespace.app.event_processor.enqueue) assert enqueue_mock.call_count == 2 for call in enqueue_mock.call_args_list: + assert call.kwargs["session_token"] is None _, event = call[0] assert event.router_data[constants.RouteVar.SESSION_ID] == "sid1" assert ( @@ -5023,6 +5106,199 @@ async def test_on_event_falls_back_to_environ_without_connect( ) +@pytest.mark.parametrize("mode", ["off", "warn", "enforce"]) +@pytest.mark.parametrize("cookie", ["missing", "invalid", "valid"]) +@pytest.mark.parametrize("client", ["missing", "bound", "foreign"]) +async def test_connect_session_authorization( + event_namespace_with_processor_mock: EventNamespace, + monkeypatch: pytest.MonkeyPatch, + mocker: MockerFixture, + mode: str, + cookie: str, + client: str, +): + """Connection tokens are replaced before state access when enforcement requires it. + + Args: + event_namespace_with_processor_mock: The connected namespace fixture. + monkeypatch: Environment fixture. + mocker: Mock fixture. + mode: The authorization policy. + cookie: The cookie validity. + client: The incoming client token's binding. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", mode) + namespace = event_namespace_with_processor_mock + manager = namespace.app._session_token_manager + browser_session = manager.create() + incoming = ( + "" + if client == "missing" + else manager.create_client_token( + browser_session if client == "bound" else manager.create() + ) + ) + state = Mock(router_data={}) + accesses = [] + + @contextlib.asynccontextmanager + async def modify_state(token): + """Check authorization at the first state access. + + Args: + token: The requested state token. + + Yields: + A minimal mutable state. + """ + context = EventContext.get() + assert context.token == token.ident + assert context.session_token is namespace._sessions.get("sid") + if mode == "enforce": + assert context.session_token is not None + assert context.session_token.authorizes(token.ident) + accesses.append(token.ident) + yield state + + mocker.patch.object(namespace.app.state_manager, "modify_state", modify_state) + scope = _connect_environ(incoming)["asgi.scope"] + scope["type"] = "websocket" + if cookie != "missing": + credential = manager.encode(browser_session) if cookie == "valid" else "invalid" + scope["headers"].append(( + b"cookie", + f"__Host-{manager.cookie_name}={credential}".encode(), + )) + + async def connect(scope, receive, send): + """Pass the validated handshake into Socket.IO.""" + await namespace.on_connect( + "sid", {"QUERY_STRING": f"token={incoming}", "asgi.scope": scope} + ) + + await SessionMiddleware(connect, manager, "https://api.test")( + scope, AsyncMock(), AsyncMock() + ) + emitted = { + call.args[0]: call.args[1] + for call in cast(AsyncMock, namespace.emit).call_args_list + } + replacement_expected = mode != "off" and ( + not incoming or (mode == "enforce" and (cookie != "valid" or client != "bound")) + ) + assert ("new_token" in emitted) is replacement_expected + assert ("session_token" in emitted) is (mode != "off" and cookie != "valid") + if mode != "off": + session = namespace._sessions["sid"] + if cookie == "valid": + assert session.id == browser_session.id + if "session_token" in emitted: + issued = manager.decode(emitted["session_token"]) + assert issued is not None + assert issued.id == session.id + else: + assert not namespace._sessions + token = emitted.get("new_token", incoming) + assert accesses == ([token] if token else []) + if token: + assert namespace.sid_to_token["sid"] == token + + +async def test_link_rejects_foreign_client_before_state_access( + event_namespace_with_processor_mock: EventNamespace, + monkeypatch: pytest.MonkeyPatch, + mocker: MockerFixture, +): + """Direct token linking must not load or register an unauthorized client. + + Args: + event_namespace_with_processor_mock: The namespace fixture. + monkeypatch: Environment fixture. + mocker: Mock fixture. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + namespace = event_namespace_with_processor_mock + manager = namespace.app._session_token_manager + namespace._sessions["sid"] = manager.create() + foreign_token = manager.create_client_token(manager.create()) + link = mocker.patch.object( + namespace._token_manager, "link_token_to_sid", new_callable=AsyncMock + ) + modify = mocker.patch.object(namespace.app.state_manager, "modify_state") + with pytest.raises(exceptions.SessionAuthorizationError): + await namespace.link_token_to_sid("sid", foreign_token) + link.assert_not_called() + modify.assert_not_called() + + +async def test_duplicate_client_stays_bound_to_session( + event_namespace_with_processor_mock: EventNamespace, + monkeypatch: pytest.MonkeyPatch, +): + """Duplicate-tab replacement tokens retain the requesting session's binding. + + Args: + event_namespace_with_processor_mock: The namespace fixture. + monkeypatch: Environment fixture. + """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + namespace = event_namespace_with_processor_mock + manager = namespace.app._session_token_manager + session = manager.create() + token = manager.create_client_token(session) + environ = _connect_environ(token) + environ["asgi.scope"]["reflex.session"] = session + await namespace.on_connect("sid1", environ) + await namespace.on_connect("sid2", environ) + replacement = namespace.sid_to_token["sid2"] + assert replacement != token + assert session.authorizes(replacement) + assert namespace.sid_to_token["sid1"] == token + cast(AsyncMock, namespace.emit).assert_awaited_once_with( + "new_token", replacement, to="sid2" + ) + + +async def test_session_refresh_notice_is_throttled( + event_namespace_with_processor_mock: EventNamespace, + mocker: MockerFixture, +): + """A connection requests cookie refresh at the interval, preserving event context. + + Args: + event_namespace_with_processor_mock: The namespace fixture. + mocker: Mock fixture. + """ + namespace = event_namespace_with_processor_mock + manager = namespace.app._session_token_manager + session = manager.create() + token = manager.create_client_token(session) + environ = _connect_environ(token) + environ["asgi.scope"]["reflex.session"] = session + await namespace.on_connect("sid", environ) + now = mocker.patch( + "reflex.app.time.time", + return_value=session.issued_at + manager.refresh_interval - 1, + ) + await namespace.on_event("sid", _client_event_payload()) + cast(AsyncMock, namespace.emit).assert_not_called() + now.return_value += 2 + await namespace.on_event("sid", _client_event_payload()) + await namespace.on_event("sid", _client_event_payload()) + cast(AsyncMock, namespace.emit).assert_awaited_once_with( + "session_refresh", to="sid" + ) + enqueue = cast(AsyncMock, namespace.app.event_processor.enqueue) + assert all( + call.kwargs["session_token"] is session for call in enqueue.call_args_list + ) + task = namespace.on_disconnect("sid") + if task is not None: + await task + assert "sid" not in namespace._sessions + assert "sid" not in namespace._session_refresh_at + + @pytest.mark.parametrize("compile_raises", [False, True]) def test_compile_releases_memo_naming_caches( mocker: MockerFixture, compile_raises: bool @@ -5280,7 +5556,8 @@ async def test_on_event_uses_frontend_traceparent(otel_exporter): } seen: list = [] - async def enqueue(token, event): + async def enqueue(token, event, *, session_token): + assert session_token is None await asyncio.sleep(0) seen.append(trace.get_current_span().get_span_context()) @@ -5399,10 +5676,14 @@ async def send(message): # noqa: RUF029 @pytest.mark.asyncio +@pytest.mark.usefixtures("attached_mock_event_context") async def test_connect_disconnect_counts_connections(otel_metrics): """Connect and disconnect adjust the open connection gauge.""" mock_app = unittest.mock.Mock() mock_app._state = None + mock_app._session_token_manager = SessionTokenManager( + "metrics", secrets=(b"s" * 32,) + ) ns = EventNamespace(namespace="/", app=mock_app) ns.emit = unittest.mock.AsyncMock() await ns.on_connect("sid1", {"QUERY_STRING": "token=t1"}) diff --git a/tests/units/test_session.py b/tests/units/test_session.py new file mode 100644 index 00000000000..fef110877c2 --- /dev/null +++ b/tests/units/test_session.py @@ -0,0 +1,561 @@ +"""Tests for browser session cookies, bootstrap requests, and request contexts.""" + +import asyncio +from functools import partial +from unittest.mock import AsyncMock + +import pytest +from pytest_mock import MockerFixture +from reflex_base.config import Config +from reflex_base.event.context import EventContext +from reflex_base.session import SessionToken, SessionTokenManager +from reflex_base.utils.exceptions import SessionAuthorizationError +from reflex_base.utils.types import Scope +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route +from starlette.testclient import TestClient + +from reflex.session import ( + SessionMiddleware, + session_authorization_error, + session_endpoint, + trusted_session_origins, +) + + +@pytest.fixture +def manager() -> SessionTokenManager: + """Create an isolated session codec. + + Returns: + A manager with a fixed test signing key. + """ + return SessionTokenManager("session_test", ttl=100, secrets=(b"s" * 32,)) + + +def _inspect(request: Request) -> JSONResponse: + """Expose non-secret request details for transport assertions. + + Args: + request: The request after middleware processing. + + Returns: + The cookie header and public session identifier. + """ + session = request.scope["reflex.session"] + return JSONResponse({ + "cookies": request.headers.getlist("cookie"), + "session": session.id if session else None, + }) + + +def _app( + manager: SessionTokenManager, + api_url: str = "https://api.test", + allowed_origins: tuple[str, ...] = ("https://api.test",), +) -> SessionMiddleware: + """Build the session routes without Reflex application state. + + Args: + manager: The session manager. + api_url: The configured public backend URL. + allowed_origins: Origins allowed to exchange a session credential. + + Returns: + A minimal ASGI app with session middleware. + """ + return SessionMiddleware( + Starlette( + routes=[ + Route("/", _inspect), + Route( + "/_reflex/session", + partial( + session_endpoint, + manager=manager, + api_url=api_url, + allowed_origins=allowed_origins, + ), + methods=["POST"], + ), + ] + ), + manager=manager, + api_url=api_url, + ) + + +def test_bootstrap_session(manager: SessionTokenManager): + """The bootstrap returns only a bound client token and an HttpOnly cookie. + + Args: + manager: The session manager. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post("/_reflex/session", json={}) + cookie = response.cookies[f"__Host-{manager.cookie_name}"] + session = manager.decode(cookie) + assert session is not None + assert response.json() == { + "token_type": "cookie", + "expires_in": 100, + "client_token": response.json()["client_token"], + } + assert session.authorizes(response.json()["client_token"]) + assert cookie not in response.text + assert response.headers["cache-control"] == "no-store" + attributes = set(response.headers["set-cookie"].split("; ")[1:]) + assert attributes == { + "Path=/", + "Max-Age=100", + "HttpOnly", + "SameSite=None", + "Secure", + "Partitioned", + } + inspected = client.get("/") + assert inspected.json() == {"cookies": [], "session": session.id} + assert "set-cookie" not in inspected.headers + + +def test_bootstrap_preserves_owned_client_token(manager: SessionTokenManager): + """Renewal preserves session identity and only echoes tokens it owns. + + Args: + manager: The session manager. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + first = client.post("/_reflex/session", json={}) + first_session = manager.decode(first.cookies[f"__Host-{manager.cookie_name}"]) + assert first_session is not None + token = first.json()["client_token"] + second = client.post("/_reflex/session", headers={"Reflex-Client-Token": token}) + assert second.json()["client_token"] == token + second_session = manager.decode(second.cookies[f"__Host-{manager.cookie_name}"]) + assert second_session is not None + assert second_session.id == first_session.id + + foreign_token = manager.create_client_token(manager.create()) + third = client.post( + "/_reflex/session", headers={"Reflex-Client-Token": foreign_token} + ) + assert third.json()["client_token"] != foreign_token + assert first_session.authorizes(third.json()["client_token"]) + + +def test_exchange_websocket_session(manager: SessionTokenManager): + """A trusted frontend can persist a session received over its WebSocket. + + Args: + manager: The session manager. + """ + session = manager.create() + token = manager.create_client_token(session) + credential = manager.encode(session) + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post( + "/_reflex/session", + json={"session_token": credential}, + headers={"Origin": "https://api.test", "Reflex-Client-Token": token}, + ) + assert response.status_code == 200 + exchanged = manager.decode(response.cookies[f"__Host-{manager.cookie_name}"]) + assert exchanged is not None + assert exchanged.id == session.id + assert response.json()["client_token"] == token + assert credential not in response.text + + +@pytest.mark.parametrize( + "origin", [None, "null", "https://evil.test", "*", "https://api.test.evil.test"] +) +def test_exchange_rejects_untrusted_origin( + manager: SessionTokenManager, origin: str | None +): + """An attacker cannot install a known session through wildcard CORS. + + Args: + manager: The session manager. + origin: An untrusted or missing request Origin. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post( + "/_reflex/session", + json={"session_token": manager.encode(manager.create())}, + headers={"Origin": origin} if origin else {}, + ) + assert response.status_code == 403 + assert "set-cookie" not in response.headers + + +def test_exchange_cookie_has_precedence(manager: SessionTokenManager): + """A body credential cannot replace an existing authenticated session. + + Args: + manager: The session manager. + """ + session = manager.create() + cookie_name = f"__Host-{manager.cookie_name}" + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post( + "/_reflex/session", + json={"session_token": manager.encode(manager.create())}, + headers={ + "Cookie": f"{cookie_name}={manager.encode(session)}", + "Origin": "https://evil.test", + }, + ) + assert response.status_code == 200 + refreshed = manager.decode(response.cookies[cookie_name]) + assert refreshed is not None + assert refreshed.id == session.id + + +@pytest.mark.parametrize( + "body", + [b"not-json", b"null", b"[]", b'{"session_token":null}', b'{"session_token":42}'], +) +def test_exchange_rejects_malformed_body(manager: SessionTokenManager, body: bytes): + """Malformed exchange payloads are rejected without creating a session. + + Args: + manager: The session manager. + body: The malformed JSON payload. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post( + "/_reflex/session", + content=body, + headers={"Content-Type": "application/json", "Origin": "https://api.test"}, + ) + assert response.status_code == 400 + assert "set-cookie" not in response.headers + + +@pytest.mark.parametrize("navigate", [False, True]) +def test_exchange_rejects_invalid_credential_or_navigation( + manager: SessionTokenManager, navigate: bool +): + """Only verified credentials sent by fetch can become cookies. + + Args: + manager: The session manager. + navigate: Whether the request is a browser navigation. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post( + "/_reflex/session", + json={ + "session_token": manager.encode(manager.create()) + if navigate + else "invalid" + }, + headers={ + "Origin": "https://api.test", + "Sec-Fetch-Mode": "navigate" if navigate else "cors", + }, + ) + assert response.status_code == (403 if navigate else 401) + assert "set-cookie" not in response.headers + + +@pytest.mark.parametrize("mode", ["dev", "prod"]) +def test_trusted_session_origins(monkeypatch: pytest.MonkeyPatch, mode: str): + """Exchange trusts explicit configured origins and development loopbacks. + + Args: + monkeypatch: Environment fixture. + mode: The runtime mode. + """ + monkeypatch.setenv("REFLEX_ENV_MODE", mode) + config = Config( + app_name="test", + api_url="https://API.test:443/backend/", + deploy_url="https://frontend.test/app/", + frontend_port=3456, + cors_allowed_origins=["*", "https://embed.test", "https://frontend.test"], + ) + origins = trusted_session_origins(config) + assert origins[:3] == ( + "https://api.test", + "https://frontend.test", + "https://embed.test", + ) + assert "*" not in origins + assert origins[3:] == ( + ("http://localhost:3456", "http://127.0.0.1:3456", "http://[::1]:3456") + if mode == "dev" + else () + ) + + +@pytest.mark.parametrize( + "headers", + [ + {}, + {"Content-Type": "text/plain"}, + {"Content-Type": "application/x-www-form-urlencoded"}, + {"Content-Type": "application/json", "Sec-Fetch-Mode": "navigate"}, + ], +) +def test_bootstrap_rejects_simple_requests(manager: SessionTokenManager, headers: dict): + """Forms and navigation cannot mint a session. + + Args: + manager: The session manager. + headers: Headers that do not identify a permitted fetch request. + """ + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.post("/_reflex/session", headers=headers) + assert response.status_code == 403 + assert response.headers["cache-control"] == "no-store" + assert "set-cookie" not in response.headers + assert client.get("/_reflex/session").status_code == 405 + + +@pytest.mark.parametrize("cookie", [None, "invalid"]) +def test_ordinary_requests_never_create_sessions( + manager: SessionTokenManager, mocker: MockerFixture, cookie: str | None +): + """Missing and malformed credentials remain unauthenticated. + + Args: + manager: The session manager. + mocker: Mock fixture. + cookie: An optional malformed cookie. + """ + create = mocker.spy(manager, "create") + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.get( + "/", + headers={"cookie": f"__Host-{manager.cookie_name}={cookie}"} + if cookie + else {}, + ) + assert response.json() == {"cookies": [], "session": None} + assert "set-cookie" not in response.headers + create.assert_not_called() + + +@pytest.mark.parametrize("duplicate", [False, True]) +def test_session_cookie_stripping(manager: SessionTokenManager, duplicate: bool): + """Split cookie headers retain unrelated cookies and reject duplicate sessions. + + Args: + manager: The session manager. + duplicate: Whether the expected cookie name appears twice. + """ + session = manager.create() + cookie = manager.encode(session) + cookie_name = f"__Host-{manager.cookie_name}" + headers = [ + ("cookie", f"auth=enterprise; {cookie_name}={cookie}; preference=dark"), + ("cookie", f"{manager.cookie_name}=invalid; other_app_session=keep"), + ] + if duplicate: + headers.append(("cookie", f"{cookie_name}={cookie}")) + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.get("/", headers=headers) + assert response.json() == { + "cookies": ["auth=enterprise; preference=dark", "other_app_session=keep"], + "session": None if duplicate else session.id, + } + + +def test_http_refresh_keeps_session_identity( + manager: SessionTokenManager, mocker: MockerFixture +): + """Expired refresh intervals renew credentials without losing client bindings. + + Args: + manager: The session manager. + mocker: Mock fixture. + """ + now = mocker.patch("reflex_base.session.time.time", return_value=1000) + session = manager.create() + token = manager.create_client_token(session) + cookie_name = f"__Host-{manager.cookie_name}" + cookie = manager.encode(session) + now.return_value = 1060 + with TestClient(_app(manager), base_url="https://api.test") as client: + response = client.get("/", headers={"cookie": f"{cookie_name}={cookie}"}) + refreshed = manager.decode(response.cookies[cookie_name]) + assert refreshed is not None + assert refreshed.id == session.id + assert refreshed.expires_at == 1160 + assert refreshed.authorizes(token) + bootstrap = client.post( + "/_reflex/session", json={}, headers={"cookie": f"{cookie_name}={cookie}"} + ) + assert len(bootstrap.headers.get_list("set-cookie")) == 1 + + +@pytest.mark.parametrize( + ("api_url", "mode", "secure"), + [ + ("https://api.test", "dev", True), + ("http://localhost:8000", "dev", True), + ("http://127.0.0.1:8000", "dev", True), + ("http://[::1]:8000", "dev", True), + ("http://192.168.1.5:8000", "dev", False), + ("http://api.test", "prod", True), + ], +) +def test_cookie_policy( + manager: SessionTokenManager, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + api_url: str, + mode: str, + secure: bool, +): + """Public API configuration controls cookies even behind HTTP proxies. + + Args: + manager: The session manager. + monkeypatch: Environment fixture. + caplog: Captured log fixture. + api_url: The externally reachable URL. + mode: The app runtime mode. + secure: Whether a secure cookie is expected. + """ + monkeypatch.setenv("REFLEX_ENV_MODE", mode) + with TestClient(_app(manager, api_url), base_url="http://internal-proxy") as client: + response = client.post("/_reflex/session", json={}) + cookie = response.headers["set-cookie"] + assert ("; Secure" in cookie) is secure + assert ("; Partitioned" in cookie) is secure + assert cookie.startswith("__Host-") is secure + assert f"SameSite={'None' if secure else 'Lax'}" in cookie + assert ("insecure SameSite=Lax" in caplog.text) is not secure + + +@pytest.mark.asyncio +async def test_websocket_validates_without_setting_cookie( + manager: SessionTokenManager, mocker: MockerFixture +): + """WebSocket handshakes expose the session without refreshing its credential. + + Args: + manager: The session manager. + mocker: Mock fixture. + """ + session = manager.create() + cookie = manager.encode(session) + app, receive, send = AsyncMock(), AsyncMock(), AsyncMock() + refresh = mocker.spy(manager, "refresh") + scope: Scope = { + "type": "websocket", + "headers": [(b"cookie", f"__Host-{manager.cookie_name}={cookie}".encode())], + } + await SessionMiddleware(app, manager, "https://api.test")(scope, receive, send) + assert scope["reflex.session"].id == session.id + assert scope["headers"] == [] + app.assert_awaited_once_with(scope, receive, send) + refresh.assert_not_called() + send.assert_not_called() + + +@pytest.mark.asyncio +async def test_http_context_isolation(manager: SessionTokenManager): + """Concurrent requests narrow the root context and restore it afterward. + + Args: + manager: The session manager. + """ + sessions = [manager.create(), manager.create()] + root = EventContext( + token="", + session_token=SessionToken.SYSTEM, + state_manager=AsyncMock(), + enqueue_impl=AsyncMock(), + ) + observed = [] + + async def inner(scope, receive, send): + """Observe the request context across task switches.""" + await asyncio.sleep(0) + context = EventContext.get() + assert context.token == "" + assert context.cached_states is not root.cached_states + assert context.session_token is scope["reflex.session"] + observed.append(context.session_token.id if context.session_token else None) + + middleware = SessionMiddleware(inner, manager, "https://api.test") + with root: + await asyncio.gather( + *[ + middleware( + { + "type": "http", + "headers": [ + ( + b"cookie", + f"__Host-{manager.cookie_name}={manager.encode(session)}".encode(), + ) + ], + }, + AsyncMock(), + AsyncMock(), + ) + for session in sessions + ], + middleware({"type": "http", "headers": []}, AsyncMock(), AsyncMock()), + ) + assert EventContext.get() is root + assert set(observed) == {None, *(session.id for session in sessions)} + + +@pytest.mark.asyncio +async def test_http_unauthorized_state_access_is_forbidden( + manager: SessionTokenManager, +): + """State authorization errors become 403 before a response starts. + + Args: + manager: The session manager. + """ + send = AsyncMock() + await SessionMiddleware( + AsyncMock(side_effect=SessionAuthorizationError("private diagnostics")), + manager, + "https://api.test", + )( + {"type": "http", "headers": []}, + AsyncMock(), + send, + ) + assert send.await_args_list[0].args[0]["status"] == 403 + assert b"private diagnostics" not in send.await_args_list[1].args[0]["body"] + + +def test_starlette_unauthorized_state_access_is_forbidden(manager: SessionTokenManager): + """Starlette handles authorization failures before its generic 500 handler. + + Args: + manager: The session manager. + """ + + def forbidden(request: Request) -> JSONResponse: + """Reject access to another session's client state. + + Args: + request: The incoming request. + + Raises: + SessionAuthorizationError: Always. + """ + msg = "private diagnostics" + raise SessionAuthorizationError(msg) + + app = Starlette( + routes=[Route("/", forbidden)], + exception_handlers={SessionAuthorizationError: session_authorization_error}, + ) + with TestClient(SessionMiddleware(app, manager, "https://api.test")) as client: + response = client.get("/") + assert response.status_code == 403 + assert response.headers["cache-control"] == "no-store" + assert "private diagnostics" not in response.text diff --git a/tests/units/utils/test_token_manager.py b/tests/units/utils/test_token_manager.py index af64cb86b7a..ee91033d125 100644 --- a/tests/units/utils/test_token_manager.py +++ b/tests/units/utils/test_token_manager.py @@ -5,9 +5,14 @@ import time from collections.abc import Callable, Generator from contextlib import asynccontextmanager +from functools import partial +from typing import Any from unittest.mock import AsyncMock, Mock, patch import pytest +from pytest_mock import MockerFixture +from reflex_base.event.context import EventContext +from reflex_base.session import SessionTokenManager from reflex import config from reflex.app import EventNamespace @@ -22,6 +27,45 @@ from tests.units.mock_redis import mock_redis +@pytest.mark.parametrize("backend", ["local", "redis", "redis-down"]) +async def test_duplicate_token_uses_session_bound_factory( + mocker: MockerFixture, backend: str +): + """Every duplicate path preserves the caller's session-bound token factory. + + Args: + mocker: Mock fixture. + backend: The token manager backend and availability. + """ + codec = SessionTokenManager("duplicates", secrets=(b"s" * 32,)) + session = codec.create() + token = codec.create_client_token(session) + if backend == "local": + manager = LocalTokenManager() + else: + redis = AsyncMock() + redis.get_connection_kwargs = Mock(return_value={"db": 0}) + redis.exists.return_value = True + if backend == "redis-down": + redis.exists.side_effect = ConnectionError("unavailable") + manager = RedisTokenManager(redis) + mocker.patch.object(manager, "_ensure_socket_record_task") + manager.token_to_socket[token] = SocketRecord( + instance_id=manager.instance_id, sid="sid1" + ) + manager.sid_to_token["sid1"] = token + factory = Mock(side_effect=partial(codec.create_client_token, session)) + + replacement = await manager.link_token_to_sid(token, "sid2", token_factory=factory) + + assert replacement is not None + assert replacement != token + assert session.authorizes(replacement) + assert not codec.create().authorizes(replacement) + assert manager.sid_to_token == {"sid1": token, "sid2": replacement} + factory.assert_called_once_with() + + class TestTokenManager: """Tests for the TokenManager factory.""" @@ -366,16 +410,19 @@ async def test_link_token_to_sid_redis_error_fallback(self, manager, mock_redis) """ token, sid = "token1", "sid1" mock_redis.exists.side_effect = Exception("Redis connection error") + token_factory = Mock(return_value="replacement-token") with patch.object( LocalTokenManager, "link_token_to_sid", new_callable=AsyncMock ) as mock_super: mock_super.return_value = None - result = await manager.link_token_to_sid(token, sid) + result = await manager.link_token_to_sid( + token, sid, token_factory=token_factory + ) assert result is None - mock_super.assert_called_once_with(token, sid) + mock_super.assert_called_once_with(token, sid, token_factory=token_factory) async def test_link_token_to_sid_redis_set_error_continues( self, manager, mock_redis @@ -738,7 +785,7 @@ def redis_url(): return redis_url -def query_string_for(token: str) -> dict[str, str]: +def query_string_for(token: str) -> dict[str, Any]: """Generate query string for given token. Args: @@ -747,12 +794,25 @@ def query_string_for(token: str) -> dict[str, str]: Returns: The generated query string. """ - return {"QUERY_STRING": f"token={token}"} + codec = SessionTokenManager("namespace", secrets=(b"s" * 32,)) + return { + "QUERY_STRING": f"token={token}", + "asgi.scope": {"reflex.session": codec.create()}, + } @pytest.fixture -def event_namespace_factory() -> Generator[Callable[[], EventNamespace], None, None]: - """Yields the EventNamespace factory function.""" +def event_namespace_factory( + attached_mock_event_context: EventContext, +) -> Generator[Callable[[], EventNamespace], None, None]: + """Create namespaces with validated browser sessions and an active context. + + Args: + attached_mock_event_context: The active event context. + + Yields: + The EventNamespace factory function. + """ namespace = config.get_config().get_event_namespace() created_objs = [] @@ -761,6 +821,9 @@ def new_event_namespace() -> EventNamespace: state.router_data = {} mock_app = Mock() + mock_app._session_token_manager = SessionTokenManager( + "namespace", secrets=(b"s" * 32,) + ) mock_app.state_manager.modify_state = Mock( return_value=AsyncMock(__aenter__=AsyncMock(return_value=state)) ) From 1eb1a60f4a6a08fac20114495fcd15e0887d501e Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Mon, 5 Oct 2026 14:10:34 -0700 Subject: [PATCH 2/5] Adapt connect hydration tests to session-aware fixtures --- tests/units/test_app.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/tests/units/test_app.py b/tests/units/test_app.py index 25b9a5cf5b8..5cb1783606a 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -4907,12 +4907,13 @@ async def event_namespace_with_processor_mock( app._session_token_manager = SessionTokenManager("test", secrets=(b"s" * 32,)) event_namespace = EventNamespace("/event", app) event_namespace.emit = AsyncMock() + token_manager = event_namespace._token_manager yield event_namespace # The token manager is backed by redis when one is configured; drop the # tokens these tests link so they do not show up in another test's # enumeration of the shared instance. Awaited rather than run in a fresh # loop via asyncio.run: the redis client is bound to the test's loop. - await event_namespace._token_manager.disconnect_all() + await token_manager.disconnect_all() def _connect_environ(token: str) -> dict[str, Any]: @@ -5336,7 +5337,7 @@ def fake_compile_app(*_args: Any, **_kwargs: Any) -> bool: @pytest.mark.asyncio async def test_on_connect_processes_boot_event_from_auth( - event_namespace: EventNamespace, + event_namespace_with_processor_mock: EventNamespace, ): """The hydrate event carried in the socket.io CONNECT packet is processed on connect. @@ -5345,8 +5346,9 @@ async def test_on_connect_processes_boot_event_from_auth( first; a connect without a boot event still does. Args: - event_namespace: The event namespace. + event_namespace_with_processor_mock: The event namespace with session support. """ + event_namespace = event_namespace_with_processor_mock event_namespace._token_manager = Mock() event_namespace._token_manager.link_token_to_sid = AsyncMock(return_value=None) event_namespace.on_event = AsyncMock() @@ -5360,7 +5362,7 @@ async def test_on_connect_processes_boot_event_from_auth( "new_sid", {"QUERY_STRING": "token=abc"}, {"event": boot_event} ) event_namespace._token_manager.link_token_to_sid.assert_awaited_once_with( - "abc", "new_sid" + "abc", "new_sid", token_factory=unittest.mock.ANY ) event_namespace.on_event.assert_awaited_once_with("new_sid", boot_event) modify_state.assert_not_called() @@ -5488,13 +5490,14 @@ async def test_on_connect_sends_no_mismatch_without_one( @pytest.mark.asyncio async def test_on_connect_unlinks_token_when_boot_event_fails( - event_namespace: EventNamespace, + event_namespace_with_processor_mock: EventNamespace, ): """A boot event that fails to process drops the sid/token link before refusing the connect. Args: - event_namespace: The event namespace. + event_namespace_with_processor_mock: The event namespace with session support. """ + event_namespace = event_namespace_with_processor_mock event_namespace._token_manager = Mock() event_namespace._token_manager.link_token_to_sid = AsyncMock(return_value=None) event_namespace._token_manager.disconnect_token = AsyncMock() @@ -5510,6 +5513,8 @@ async def test_on_connect_unlinks_token_when_boot_event_fails( ) # The connection-scoped router data cached for the refused sid goes too. assert "new_sid" not in event_namespace._static_router_data + assert "new_sid" not in event_namespace._sessions + assert "new_sid" not in event_namespace._session_refresh_at def test_call_app_wraps_with_otel_asgi_middleware(): From 05fa726556c96cf798a47051ed5796db6c8f8fe3 Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Mon, 5 Oct 2026 14:24:05 -0700 Subject: [PATCH 3/5] Use the validated root context for streamed event fallback --- .../src/reflex_base/event/processor/event_processor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py index 14632463199..9a7b673ca2d 100644 --- a/packages/reflex-base/src/reflex_base/event/processor/event_processor.py +++ b/packages/reflex-base/src/reflex_base/event/processor/event_processor.py @@ -537,7 +537,7 @@ async def _emit_delta_impl( try: parent_context = EventContext.get() except LookupError: - parent_context = self._root_context + parent_context = root task_future = await self.enqueue( token, event, From 9120624b655fd7084e817135448af4af6ef60669 Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Wed, 14 Oct 2026 02:16:59 -0700 Subject: [PATCH 4/5] Scope custom sockets and renew authenticated browser sessions --- news/+browser-session-transport.feature.md | 2 +- .../src/reflex_base/event/context.py | 2 +- pyproject.toml | 2 +- reflex/app.py | 21 ++++++++ reflex/session.py | 16 +++--- tests/units/test_app.py | 47 ++++++++++++++-- tests/units/test_session.py | 53 +++++++++++++++++-- 7 files changed, 127 insertions(+), 16 deletions(-) diff --git a/news/+browser-session-transport.feature.md b/news/+browser-session-transport.feature.md index 5bf5abdece7..5c3b5086298 100644 --- a/news/+browser-session-transport.feature.md +++ b/news/+browser-session-transport.feature.md @@ -1 +1 @@ -Add browser sessions using HttpOnly cookies and a `POST /_reflex/session` endpoint that issues a client token bound to the requesting session. Sessions are validated for HTTP and WebSocket requests, and valid HTTP sessions refresh automatically. +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. diff --git a/packages/reflex-base/src/reflex_base/event/context.py b/packages/reflex-base/src/reflex_base/event/context.py index d3ac01599ea..c61f2a59943 100644 --- a/packages/reflex-base/src/reflex_base/event/context.py +++ b/packages/reflex-base/src/reflex_base/event/context.py @@ -150,7 +150,7 @@ def __post_init__(self) -> None: "Trusted server-side work may explicitly use SessionToken.SYSTEM. " "REFLEX_SESSION_TOKEN_MODE=enforce rejects this access." ), - deprecation_version="0.9.13", + deprecation_version="0.10.1", removal_version="1.0", ) diff --git a/pyproject.toml b/pyproject.toml index 9520867e570..d997e427105 100644 --- a/pyproject.toml +++ b/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", diff --git a/reflex/app.py b/reflex/app.py index 715e042fbc1..ec53339ec28 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -18,6 +18,7 @@ import time import traceback import urllib.parse +import uuid from collections.abc import ( AsyncIterator, Callable, @@ -2265,6 +2266,9 @@ async def on_connect(self, sid: str, environ: dict, auth: Any = None): token = token_list[0] if token_list else "" mode = environment.REFLEX_SESSION_TOKEN_MODE.get() issued_session = None + if mode == "off" and not token: + token = str(uuid.uuid4()) + await self.emit("new_token", token, to=sid) if mode != "off": manager = self.app._session_token_manager session = environ.get("asgi.scope", {}).get("reflex.session") @@ -2316,6 +2320,23 @@ async def on_connect(self, sid: str, environ: dict, auth: Any = None): await self._token_manager.disconnect_token(linked_token, sid) raise + async def on_session_refresh(self, sid: str) -> None: + """Renew credentials for the session already authenticated by this socket. + + Connection authority survives cookie expiry, including suspended browser + timers. The credential is delivered only to the requesting connection. + + Args: + sid: The Socket.IO session id requesting renewal. + """ + if (session := self._sessions.get(sid)) is None: + return + manager = self.app._session_token_manager + renewed = manager.refresh(session) + self._sessions[sid] = renewed + self._session_refresh_at[sid] = renewed.issued_at + manager.refresh_interval + await self.emit("session_token", manager.encode(renewed), to=sid) + def _build_static_router_data(self, sid: str, environ: dict) -> dict[str, Any]: """Build the connection-scoped router_data entries for a socket. diff --git a/reflex/session.py b/reflex/session.py index 6763f1a401e..359e9badfc8 100644 --- a/reflex/session.py +++ b/reflex/session.py @@ -120,7 +120,7 @@ def _session_cookie( class SessionMiddleware: - """Validate sessions, hide their credentials, and scope HTTP state access.""" + """Validate sessions, hide their credentials, and scope request state access.""" def __init__(self, app: ASGIApp, manager: SessionTokenManager, api_url: str): """Wrap an ASGI app with browser session handling. @@ -176,10 +176,6 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: scope["headers"] = headers session = self.manager.decode(credentials[0]) if len(credentials) == 1 else None scope["reflex.session"] = session - if scope["type"] == "websocket": - await self.app(scope, receive, send) - return - response_started = False async def send_with_session(message: Message) -> None: @@ -217,8 +213,15 @@ async def send_with_session(message: Message) -> None: context = nullcontext() with context: try: - await self.app(scope, receive, send_with_session) + await self.app( + scope, + receive, + send if scope["type"] == "websocket" else send_with_session, + ) except SessionAuthorizationError as error: + if scope["type"] == "websocket": + await send({"type": "websocket.close", "code": 1008}) + return if response_started: raise response = await session_authorization_error(Request(scope), error) @@ -315,6 +318,7 @@ async def session_endpoint( { "token_type": "cookie", "expires_in": manager.ttl, + "refresh_in": manager.refresh_interval, "client_token": client_token, }, headers={"Cache-Control": "no-store"}, diff --git a/tests/units/test_app.py b/tests/units/test_app.py index 5cb1783606a..b57560b7c52 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -5184,8 +5184,8 @@ async def connect(scope, receive, send): call.args[0]: call.args[1] for call in cast(AsyncMock, namespace.emit).call_args_list } - replacement_expected = mode != "off" and ( - not incoming or (mode == "enforce" and (cookie != "valid" or client != "bound")) + replacement_expected = not incoming or ( + mode == "enforce" and (cookie != "valid" or client != "bound") ) assert ("new_token" in emitted) is replacement_expected assert ("session_token" in emitted) is (mode != "off" and cookie != "valid") @@ -5300,6 +5300,36 @@ async def test_session_refresh_notice_is_throttled( assert "sid" not in namespace._session_refresh_at +async def test_socket_renews_expired_cookie_without_changing_state_owner( + event_namespace_with_processor_mock, + mocker, +): + """A live connection can renew its own session after an idle browser's cookie expires.""" + namespace = event_namespace_with_processor_mock + manager = namespace.app._session_token_manager + session = manager.create() + token = manager.create_client_token(session) + environ = _connect_environ(token) + environ["asgi.scope"]["reflex.session"] = session + await namespace.on_connect("sid", environ) + mocker.patch("reflex_base.session.time.time", return_value=session.expires_at + 1) + assert manager.decode(manager.encode(session)) is None + await namespace.on_session_refresh("sid") + emitted = cast(AsyncMock, namespace.emit) + credential = emitted.call_args.args[1] + renewed = manager.decode(credential) + assert renewed is not None + assert renewed.id == session.id + assert renewed.authorizes(token) + assert namespace._sessions["sid"] == renewed + emitted.reset_mock() + task = namespace.on_disconnect("sid") + if task is not None: + await task + await namespace.on_session_refresh("sid") + emitted.assert_not_called() + + @pytest.mark.parametrize("compile_raises", [False, True]) def test_compile_releases_memo_naming_caches( mocker: MockerFixture, compile_raises: bool @@ -5402,14 +5432,18 @@ def _schema_checking_namespace( @pytest.mark.asyncio async def test_on_connect_reports_state_schema_mismatch( - event_namespace: EventNamespace, caplog: pytest.LogCaptureFixture + event_namespace: EventNamespace, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, ): """A frontend compiled for another state tree is told once, and still hydrates. Args: event_namespace: The event namespace. caplog: pytest fixture capturing backend log records. + monkeypatch: Environment fixture. """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "off") on_event, emit = _schema_checking_namespace(event_namespace, "backend-schema") boot_event = {"name": "state.hydrate_and_load", "payload": {}, "router_data": {}} @@ -5467,15 +5501,20 @@ async def test_on_connect_reports_state_schema_mismatch( pytest.param(None, id="no-auth"), ], ) +@pytest.mark.usefixtures("attached_mock_event_context") async def test_on_connect_sends_no_mismatch_without_one( - event_namespace: EventNamespace, auth: dict[str, Any] | None + event_namespace: EventNamespace, + auth: dict[str, Any] | None, + monkeypatch: pytest.MonkeyPatch, ): """A matching or missing frontend schema sends nothing to the client. Args: event_namespace: The event namespace. auth: The CONNECT auth payload. + monkeypatch: Environment fixture. """ + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "off") on_event, emit = _schema_checking_namespace(event_namespace, "same") state = Mock(router_data={}) event_namespace.app.state_manager.modify_state = Mock( diff --git a/tests/units/test_session.py b/tests/units/test_session.py index fef110877c2..8d70ca3808f 100644 --- a/tests/units/test_session.py +++ b/tests/units/test_session.py @@ -101,6 +101,7 @@ def test_bootstrap_session(manager: SessionTokenManager): assert response.json() == { "token_type": "cookie", "expires_in": 100, + "refresh_in": 50, "client_token": response.json()["client_token"], } assert session.authorizes(response.json()["client_token"]) @@ -459,11 +460,13 @@ async def test_websocket_validates_without_setting_cookie( @pytest.mark.asyncio -async def test_http_context_isolation(manager: SessionTokenManager): +@pytest.mark.parametrize("scope_type", ["http", "websocket"]) +async def test_request_context_isolation(manager: SessionTokenManager, scope_type: str): """Concurrent requests narrow the root context and restore it afterward. Args: manager: The session manager. + scope_type: The ASGI transport type. """ sessions = [manager.create(), manager.create()] root = EventContext( @@ -489,7 +492,7 @@ async def inner(scope, receive, send): *[ middleware( { - "type": "http", + "type": scope_type, "headers": [ ( b"cookie", @@ -502,12 +505,56 @@ async def inner(scope, receive, send): ) for session in sessions ], - middleware({"type": "http", "headers": []}, AsyncMock(), AsyncMock()), + middleware({"type": scope_type, "headers": []}, AsyncMock(), AsyncMock()), ) assert EventContext.get() is root assert set(observed) == {None, *(session.id for session in sessions)} +@pytest.mark.asyncio +@pytest.mark.parametrize("authorized", [False, True]) +async def test_custom_websocket_state_authority(manager, monkeypatch, authorized): + """Custom WebSocket state access uses the browser session instead of SYSTEM.""" + monkeypatch.setenv("REFLEX_SESSION_TOKEN_MODE", "enforce") + session = manager.create() + token = manager.create_client_token(session) + root = EventContext( + token="", + session_token=SessionToken.SYSTEM, + state_manager=AsyncMock(), + enqueue_impl=AsyncMock(), + ) + + async def inner(scope, receive, send): + """Access a client through the ambient request context.""" + with EventContext.get().fork(token=token): + await send({"type": "websocket.accept"}) + + headers = ( + [ + ( + b"cookie", + f"__Host-{manager.cookie_name}={manager.encode(session)}".encode(), + ) + ] + if authorized + else [] + ) + send = AsyncMock() + with root: + await SessionMiddleware(inner, manager, "https://api.test")( + {"type": "websocket", "headers": headers}, + AsyncMock(), + send, + ) + assert EventContext.get() is root + send.assert_awaited_once_with( + {"type": "websocket.accept"} + if authorized + else {"type": "websocket.close", "code": 1008} + ) + + @pytest.mark.asyncio async def test_http_unauthorized_state_access_is_forbidden( manager: SessionTokenManager, From ebed735dae423f429e3a2a8bc7e1f9caaa142c8a Mon Sep 17 00:00:00 2001 From: Masen Furer Date: Wed, 14 Oct 2026 02:41:04 -0700 Subject: [PATCH 5/5] Preserve the public database session helper when loading App --- reflex/app.py | 2 +- reflex/{session.py => session_middleware.py} | 0 tests/units/test_app.py | 2 +- tests/units/test_model.py | 14 ++++++++++++++ ...{test_session.py => test_session_middleware.py} | 2 +- 5 files changed, 17 insertions(+), 3 deletions(-) rename reflex/{session.py => session_middleware.py} (100%) rename tests/units/{test_session.py => test_session_middleware.py} (99%) diff --git a/reflex/app.py b/reflex/app.py index ec53339ec28..f832cdaeda9 100644 --- a/reflex/app.py +++ b/reflex/app.py @@ -90,7 +90,7 @@ replace_brackets_with_keywords, verify_route_validity, ) -from reflex.session import ( +from reflex.session_middleware import ( SessionMiddleware, session_authorization_error, session_endpoint, diff --git a/reflex/session.py b/reflex/session_middleware.py similarity index 100% rename from reflex/session.py rename to reflex/session_middleware.py diff --git a/tests/units/test_app.py b/tests/units/test_app.py index b57560b7c52..b7a590f800c 100644 --- a/tests/units/test_app.py +++ b/tests/units/test_app.py @@ -88,7 +88,7 @@ from reflex.istate.manager.token import BaseStateToken from reflex.istate.storage import Cookie, LocalStorage, SessionStorage from reflex.model import Model -from reflex.session import SessionMiddleware +from reflex.session_middleware import SessionMiddleware from reflex.state import ( BaseState, OnLoadInternalState, diff --git a/tests/units/test_model.py b/tests/units/test_model.py index 696d27fa618..15b06f0307f 100644 --- a/tests/units/test_model.py +++ b/tests/units/test_model.py @@ -32,6 +32,20 @@ sqlmodel = pytest.importorskip("sqlmodel") +@pytest.mark.parametrize("model_first", [False, True]) +def test_database_session_survives_app_import(model_first: bool): + """Loading browser session transports must not shadow the public DB helper. + + Args: + model_first: Whether to resolve the database helper before the App import. + """ + script = "import reflex as rx\n" + if model_first: + script += "assert callable(rx.session)\n" + script += "from reflex.app import App\nfrom reflex.model import session\nassert rx.session is session\n" + subprocess.run([sys.executable, "-c", script], check=True, capture_output=True) + + @pytest.mark.parametrize("version", ["0.0.45", "0.0.47"]) def test_db_extra_accepts_utc_sqlmodel(version: str): """The database extra must accept versions used to generate UTC migrations. diff --git a/tests/units/test_session.py b/tests/units/test_session_middleware.py similarity index 99% rename from tests/units/test_session.py rename to tests/units/test_session_middleware.py index 8d70ca3808f..5d65f2d8e54 100644 --- a/tests/units/test_session.py +++ b/tests/units/test_session_middleware.py @@ -17,7 +17,7 @@ from starlette.routing import Route from starlette.testclient import TestClient -from reflex.session import ( +from reflex.session_middleware import ( SessionMiddleware, session_authorization_error, session_endpoint,