From 57d946b7304d0427add3acf79e7f1932fe975a2b Mon Sep 17 00:00:00 2001 From: Jun <84921700+Amazingjun-j@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:15:21 +0800 Subject: [PATCH] fix(browser): drain live cleanup before releasing session (#6284) * fix(browser): drain live cleanup before releasing session * test(browser): cover cancellation precedence during cleanup * style(browser): separate cancellation regressions --- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/browser.py | 15 +- .../tests/test_browser_readonly_security.py | 135 ++++++++++++++++++ 3 files changed, 149 insertions(+), 3 deletions(-) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 8c95003d05..b8f1422882 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -233,7 +233,7 @@ the caller omits `model_name`. Suggestions check admission before their best-effort LLM error handler: explicit denies return 403 without invoking the model, while provider failures honor `authorization.fail_closed`. -`authz.py::resolve_route_permissions()` is shared by HTTP middleware, decorator-only auth, and Live Browser WebSocket admission. It evaluates registered permissions asynchronously as `resource="route"`, targeting full `resource:action` strings. HTTP caches decisions in `AuthContext`; provider errors follow `authorization.fail_closed` (decision errors are per-permission). Disabled authorization returns all permissions without a provider. Owner and admin checks remain independent. Live requires `threads:write` with `is_internal=False` after login/Origin checks but before acceptance, owner lookup, or session acquisition; denial closes 4403, unexpected setup errors close 4501, cancellation propagates. Checks are admission-only; restart Gateway to close old connections. Tests: `test_authorization_route_permissions.py`, `test_browser_readonly_security.py`, `test_auth.py`, `test_auth_middleware.py` in `tests/`. +`authz.py::resolve_route_permissions()` is shared by HTTP, decorator-only auth, and Live Browser WebSocket admission. It evaluates full `resource:action` route targets; provider errors follow `authorization.fail_closed`, while disabled authorization preserves the legacy permission set. Owner/admin checks stay separate. Live requires `threads:write` with `is_internal=False` after login/Origin checks but before acceptance or session acquisition; denial closes 4403 and unexpected setup errors close 4501. Admission is connection-scoped. On disconnect, Live drains owned `stop_screencast()` cleanup across cancellation before releasing the session lease, so cancellation cannot strand a manager pin and consume browser-session capacity. Tests: `test_authorization_route_permissions.py`, `test_browser_readonly_security.py`, `test_auth.py`, `test_auth_middleware.py`. Skill listing authorization mirrors the models pattern: `routers/skills.py` routes resolve `(provider, principal)` through `authz.py::resolve_skill_authorization()` — a thin sibling of `resolve_model_authorization`, both delegating to the shared `_resolve_route_scoped_authorization` core — and the shared `_filter_visible_skills` helper filters the user-scoped catalog (public + caller's custom skills) through `filter_resources(principal, "skill", ...)` by name. Three user-facing surfaces apply it: `GET /api/skills` (the frontend skill list / slash-command autocomplete), `GET /api/skills/custom`, and `GET /api/skills/{name}` — the detail endpoint returns the standard 404 for an invisible skill so it cannot become an existence oracle the filtered list closed (`get_model`'s 403 is an execution decision via `authorize("model", "use")`, which has no skills equivalent in this layer). Anonymous callers are unfiltered (mirroring `list_models`), and provider resolution/decision errors follow `authorization.fail_closed` (fail-closed → empty listing / 404, fail-open → full listing). Skill management endpoints (`install`/`upload`/`reload`/custom-skill CRUD) stay `require_admin_user`-gated, and runtime activation authorization is a separate layer. The built-in RBAC provider maps this to the per-role `skills` policy key. Tests: `tests/test_skills_listing_authorization.py`. diff --git a/backend/app/gateway/routers/browser.py b/backend/app/gateway/routers/browser.py index 87aeec036c..b290dd3af2 100644 --- a/backend/app/gateway/routers/browser.py +++ b/backend/app/gateway/routers/browser.py @@ -11,6 +11,7 @@ from app.gateway.authz import Permissions, require_permission, resolve_route_per from app.gateway.browser_capability import browser_capability from deerflow.config.paths import get_paths from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user +from deerflow.utils.file_io import await_drained from deerflow.utils.thread_id import ThreadId logger = logging.getLogger(__name__) @@ -476,6 +477,7 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: input_task: asyncio.Task | None = None reader_task: asyncio.Task | None = None poll_task: asyncio.Task | None = None + cancellation: asyncio.CancelledError | None = None try: # Seed the live page from the latest browser_view URL. A thread can have # a stale browser session from an earlier panel/live attempt; if that @@ -500,6 +502,8 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: await reader_task except WebSocketDisconnect: pass + except asyncio.CancelledError as exc: + cancellation = exc except Exception as exc: logger.exception("browser stream error: thread_id=%s err=%s", thread_id, exc) finally: @@ -510,7 +514,14 @@ async def browser_stream(websocket: WebSocket, thread_id: ThreadId) -> None: reader_task.cancel() if poll_task is not None: poll_task.cancel() - with contextlib.suppress(Exception): - await session.stop_screencast(_on_frame) + try: + await await_drained(session.stop_screencast(_on_frame)) + except asyncio.CancelledError as exc: + if cancellation is None: + cancellation = exc + except Exception: + pass session_lease.__exit__(None, None, None) reset_current_user(token) + if cancellation is not None: + raise cancellation diff --git a/backend/tests/test_browser_readonly_security.py b/backend/tests/test_browser_readonly_security.py index ba8b625d9f..2fbd27c093 100644 --- a/backend/tests/test_browser_readonly_security.py +++ b/backend/tests/test_browser_readonly_security.py @@ -223,3 +223,138 @@ async def test_browser_stream_rechecks_policy_on_new_connection(monkeypatch): await browser.browser_stream(websocket, "test-thread") websocket.close.assert_awaited_once_with(code=4403) negotiate.assert_awaited_once() # Second connection never gets this far. + + +@pytest.mark.asyncio +async def test_browser_stream_cancellation_during_disconnect_cleanup_releases_session_lease(monkeypatch): + config = AuthorizationConfig(enabled=False) + monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config) + user = get_auth_disabled_user() + monkeypatch.setattr(browser, "_authenticate_ws", AsyncMock(return_value=user)) + monkeypatch.setattr(browser, "_browser_tools_enabled", lambda: True) + monkeypatch.setattr("deerflow.config.get_app_config", lambda: SimpleNamespace(get_tool_config=lambda _: None)) + monkeypatch.setattr(browser, "_negotiate_browser_frame_format", AsyncMock(return_value=False)) + + stop_entered = asyncio.Event() + stop_release = asyncio.Event() + lease_released = asyncio.Event() + + async def stop_screencast(_on_frame): + stop_entered.set() + await stop_release.wait() + + session = SimpleNamespace( + start_screencast=AsyncMock(), + stop_screencast=stop_screencast, + current_url=AsyncMock(return_value=None), + tabs=AsyncMock(return_value=[]), + dispatch_input=AsyncMock(), + ) + + @contextmanager + def acquire_session(_thread_id, **_kwargs): + try: + yield session + finally: + lease_released.set() + + monkeypatch.setattr( + "deerflow.community.browser_automation.get_browser_session_manager", + lambda: SimpleNamespace(acquire_session=acquire_session), + ) + + websocket = MagicMock() + websocket.headers = {} + websocket.query_params = {} + websocket.send_text = AsyncMock() + websocket.close = AsyncMock() + websocket.receive_text = AsyncMock(side_effect=WebSocketDisconnect()) + websocket.app.state.thread_store.get = AsyncMock(return_value={"user_id": user.id}) + + stream_task = asyncio.create_task(browser.browser_stream(websocket, "test-thread")) + await asyncio.wait_for(stop_entered.wait(), timeout=1) + + stream_task.cancel("gateway shutdown") + await asyncio.sleep(0) + + assert not stream_task.done(), "Live cleanup returned before its owned screencast teardown drained" + assert not lease_released.is_set() + + stream_task.cancel("forced shutdown") + await asyncio.sleep(0) + assert not stream_task.done(), "Repeated cancellation abandoned Live cleanup" + + stop_release.set() + with pytest.raises(asyncio.CancelledError, match="gateway shutdown"): + await stream_task + + assert lease_released.is_set() + + +@pytest.mark.asyncio +async def test_browser_stream_preserves_first_cancellation_across_cleanup_cancellation(monkeypatch): + config = AuthorizationConfig(enabled=False) + monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config) + user = get_auth_disabled_user() + monkeypatch.setattr(browser, "_authenticate_ws", AsyncMock(return_value=user)) + monkeypatch.setattr(browser, "_browser_tools_enabled", lambda: True) + monkeypatch.setattr("deerflow.config.get_app_config", lambda: SimpleNamespace(get_tool_config=lambda _: None)) + monkeypatch.setattr(browser, "_negotiate_browser_frame_format", AsyncMock(return_value=False)) + + start_entered = asyncio.Event() + stop_entered = asyncio.Event() + stop_release = asyncio.Event() + lease_released = asyncio.Event() + + async def start_screencast(_on_frame): + start_entered.set() + await asyncio.Future() + + async def stop_screencast(_on_frame): + stop_entered.set() + await stop_release.wait() + + session = SimpleNamespace( + start_screencast=start_screencast, + stop_screencast=stop_screencast, + current_url=AsyncMock(return_value=None), + tabs=AsyncMock(return_value=[]), + dispatch_input=AsyncMock(), + ) + + @contextmanager + def acquire_session(_thread_id, **_kwargs): + try: + yield session + finally: + lease_released.set() + + monkeypatch.setattr( + "deerflow.community.browser_automation.get_browser_session_manager", + lambda: SimpleNamespace(acquire_session=acquire_session), + ) + + websocket = MagicMock() + websocket.headers = {} + websocket.query_params = {} + websocket.send_text = AsyncMock() + websocket.close = AsyncMock() + websocket.receive_text = AsyncMock(side_effect=WebSocketDisconnect()) + websocket.app.state.thread_store.get = AsyncMock(return_value={"user_id": user.id}) + + stream_task = asyncio.create_task(browser.browser_stream(websocket, "test-thread")) + await asyncio.wait_for(start_entered.wait(), timeout=1) + + stream_task.cancel("gateway shutdown") + await asyncio.wait_for(stop_entered.wait(), timeout=1) + + stream_task.cancel("forced shutdown") + await asyncio.sleep(0) + assert not stream_task.done(), "Repeated cancellation abandoned Live cleanup" + assert not lease_released.is_set() + + stop_release.set() + with pytest.raises(asyncio.CancelledError, match="gateway shutdown"): + await stream_task + + assert lease_released.is_set()