From d9e23b90c100d19b00270aea3cde0501d2354a6c Mon Sep 17 00:00:00 2001 From: Classic298 <27028174+Classic298@users.noreply.github.com> Date: Tue, 11 Aug 2026 03:17:18 +0200 Subject: [PATCH] refac: share one folder write-access check across chat folder_id paths (#28366) Chat creation and chat moves each carried their own copy of the same folder_id validation, resolving the folder and checking ownership and shared write access in slightly different ways. Both now call a single has_folder_write_access helper, which the chat-completions creation path uses as well, so ownership, inherited write grants and nonexistent or malformed ids behave identically everywhere a chat folder_id is set. The owner case also costs one query fewer than before. --- backend/open_webui/main.py | 9 +++++ backend/open_webui/routers/chats.py | 38 ++++++------------- .../utils/access_control/folders.py | 10 ++++- 3 files changed, 30 insertions(+), 27 deletions(-) diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index 7feca6db15..bde28c1af1 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -199,6 +199,7 @@ from open_webui.tasks import ( ) # Import from tasks.py from open_webui.utils import logger from open_webui.utils.access_control import has_permission +from open_webui.utils.access_control.folders import has_folder_write_access from open_webui.utils.actions import chat_action as chat_action_handler from open_webui.utils.asgi_middleware import ( AuthTokenMiddleware, @@ -1272,6 +1273,14 @@ async def chat_completion( if is_saved_chat_id(chat_id): if is_new_chat: + # The chat created below is persisted with this folder_id. + folder_id = metadata['folder_id'] + if folder_id is not None and not await has_folder_write_access(user.id, folder_id): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ERROR_MESSAGES.NOT_FOUND, + ) + # Build the full history upfront with ALL assistant placeholders user_message = metadata.get('user_message') or {} user_message_id = user_message.get('id') if user_message else None diff --git a/backend/open_webui/routers/chats.py b/backend/open_webui/routers/chats.py index bd2801bbe3..636cdd44ca 100644 --- a/backend/open_webui/routers/chats.py +++ b/backend/open_webui/routers/chats.py @@ -38,7 +38,7 @@ from open_webui.models.tags import TagModel, Tags from open_webui.socket.main import get_event_emitter from open_webui.tasks import has_active_tasks, stop_item_tasks from open_webui.utils.access_control import filter_allowed_access_grants, has_permission -from open_webui.utils.access_control.folders import has_folder_access +from open_webui.utils.access_control.folders import has_folder_access, has_folder_write_access from open_webui.utils.auth import bearer_security, get_admin_user, get_current_user, get_verified_user from open_webui.utils.chat_fork import build_fork_history from open_webui.utils.context_compaction import compact_chat_branch, get_chat_context_usage @@ -753,20 +753,11 @@ async def create_new_chat( user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), ): - # Reject a folder_id that doesn't belong to the caller. Without this the - # row is persisted with a dangling foreign reference — no read path - # surfaces it across users (all chat reads are user_id-filtered), but - # the row state is meaningless and downstream consumers shouldn't have - # to assume the column is clean. Also catches non-UUID / nonexistent IDs. - if form_data.folder_id is not None: - if not await Folders.get_folder_by_id_and_user_id(form_data.folder_id, user.id, db=db): - # Check shared folder write access - shared_folder = await Folders.get_folder_by_id(form_data.folder_id, db=db) - if not shared_folder or not await has_folder_access(user.id, shared_folder, 'write', db): - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=ERROR_MESSAGES.NOT_FOUND, - ) + if form_data.folder_id is not None and not await has_folder_write_access(user.id, form_data.folder_id, db=db): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ERROR_MESSAGES.NOT_FOUND, + ) try: chat = await Chats.insert_new_chat(str(uuid4()), user.id, form_data, db=db) @@ -2124,17 +2115,12 @@ async def update_chat_folder_id_by_id( ): chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) if chat: - # Same ownership check as the create path — reject foreign / dangling - # folder_id values. None is allowed (moves the chat out of any folder). - if form_data.folder_id is not None: - if not await Folders.get_folder_by_id_and_user_id(form_data.folder_id, user.id, db=db): - # Check shared folder write access - shared_folder = await Folders.get_folder_by_id(form_data.folder_id, db=db) - if not shared_folder or not await has_folder_access(user.id, shared_folder, 'write', db): - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=ERROR_MESSAGES.NOT_FOUND, - ) + # None is allowed: it moves the chat out of any folder. + if form_data.folder_id is not None and not await has_folder_write_access(user.id, form_data.folder_id, db=db): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=ERROR_MESSAGES.NOT_FOUND, + ) chat = await Chats.update_chat_folder_id_by_id_and_user_id(id, user.id, form_data.folder_id, db=db) await publish_event( diff --git a/backend/open_webui/utils/access_control/folders.py b/backend/open_webui/utils/access_control/folders.py index 7dcc1a3395..1eb32b660c 100644 --- a/backend/open_webui/utils/access_control/folders.py +++ b/backend/open_webui/utils/access_control/folders.py @@ -3,7 +3,7 @@ from open_webui.models.folders import FolderModel, Folders from sqlalchemy.ext.asyncio import AsyncSession -async def has_folder_access(user_id: str, folder: FolderModel, permission: str, db: AsyncSession) -> bool: +async def has_folder_access(user_id: str, folder: FolderModel, permission: str, db: AsyncSession | None) -> bool: """Check if user has access to folder directly or via ancestor inheritance.""" if folder.user_id == user_id: return True @@ -22,3 +22,11 @@ async def has_folder_access(user_id: str, folder: FolderModel, permission: str, if parent: return await has_folder_access(user_id, parent, permission, db) return False + + +async def has_folder_write_access(user_id: str, folder_id: str, db: AsyncSession | None = None) -> bool: + """Check write access on the folder with this id; False if no such folder exists.""" + folder = await Folders.get_folder_by_id(folder_id, db=db) + if not folder: + return False + return await has_folder_access(user_id, folder, 'write', db)