refac

backend/open_webui/models/notes.py

@@ -9,7 +9,7 @@ from open_webui.models.groups import Groups
 from open_webui.models.users import User, UserModel, UserResponse, Users
 from open_webui.utils.json_codec import JSONCodec
 from pydantic import BaseModel, ConfigDict, Field, field_validator
-from sqlalchemy import JSON, BigInteger, Boolean, Column, ForeignKey, Text, delete, func, or_, select, update
+from sqlalchemy import JSON, BigInteger, Boolean, Column, ForeignKey, Text, cast, delete, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 ####################
@@ -336,6 +336,27 @@ class NoteTable:
             note = result.scalars().first()
             return await self._to_note_model(note, db=db) if note else None
 
+    async def get_note_ids_by_file_id(self, file_id: str, owner_id: str, db: AsyncSession | None = None) -> list[str]:
+        """Find current file attachments in notes owned by the file owner."""
+        async with get_async_db_context(db) as db:
+            result = await db.execute(
+                select(Note.id, Note.data).filter(
+                    Note.user_id == owner_id,
+                    cast(Note.data, Text).like(f'%{file_id}%'),
+                )
+            )
+            # The text filter only narrows candidates; authorization needs an exact attachment.
+            return [
+                note_id
+                for note_id, data in result.all()
+                if isinstance(data, dict)
+                and isinstance(data.get('files'), list)
+                and any(
+                    isinstance(item, dict) and item.get('type') == 'file' and item.get('id') == file_id
+                    for item in data['files']
+                )
+            ]
+
     async def update_note_by_id(
         self, id: str, form_data: NoteUpdateForm, db: Optional[AsyncSession] = None
     ) -> Optional[NoteModel]:

backend/open_webui/utils/access_control/files.py

@@ -8,6 +8,7 @@ from open_webui.models.folders import FolderModel
 from open_webui.models.groups import Groups
 from open_webui.models.knowledge import Knowledges
 from open_webui.models.models import Models
+from open_webui.models.notes import Notes
 from open_webui.models.users import UserModel, Users
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -29,6 +30,7 @@ async def has_access_to_file(
     - Shared workspace models that attach the file directly
     - Channels the user is a member of
     - Shared chats
+    - Shared notes whose owner owns the attached file (read only)
 
     NOTE: This does NOT check direct file ownership — callers should check
     file.user_id == user.id separately before calling this.
@@ -84,6 +86,19 @@ async def has_access_to_file(
         if accessible_ids:
             return True
 
+    # Note attachment JSON is user-controlled, so only the file owner's notes can grant access.
+    if access_type == 'read':
+        note_ids = await Notes.get_note_ids_by_file_id(file.id, owner_id=file.user_id, db=db)
+        if note_ids and await AccessGrants.get_accessible_resource_ids(
+            user_id=user.id,
+            resource_type='note',
+            resource_ids=note_ids,
+            permission='read',
+            user_group_ids=user_group_ids,
+            db=db,
+        ):
+            return True
+
     # Check if the file is directly attached to a shared workspace model (per the ownership
     # note above, model write is conferred only for files the model owner owns).
     model_owners = await Models.get_model_owner_ids_by_file_id(file.id, db=db, include_background=access_type == 'read')
@@ -144,8 +159,6 @@ async def get_accessible_folder_files(
                 accessible.append(entry)
         elif entry_type == 'note':
             # Owner has no self-grant (notes are private by default), so check ownership too.
-            from open_webui.models.notes import Notes
-
             note = await Notes.get_note_by_id(entry_id, db=db)
             if note and (
                 note.user_id == user.id