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

@@ -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

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(

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)