fix: get_image_base64_from_file_id

backend/open_webui/utils/files.py

@@ -22,6 +22,7 @@ from open_webui.models.chats import Chats
 from open_webui.models.files import Files
 from open_webui.retrieval.web.utils import validate_url
 from open_webui.routers.files import upload_file_handler
+from open_webui.utils.access_control.files import has_access_to_file
 from open_webui.routers.images import (
     get_image_data,
     upload_image,
@@ -50,7 +51,7 @@ _IMAGE_MIME_FALLBACK = {
 }
 
 
-async def get_image_base64_from_url(url: str) -> Optional[str]:
+async def get_image_base64_from_url(url: str, user=None) -> Optional[str]:
     try:
         if url.startswith('http'):
             # Validate URL to prevent SSRF attacks against local/private networks.
@@ -70,25 +71,9 @@ async def get_image_base64_from_url(url: str) -> Optional[str]:
                 content_type = response.headers.get('Content-Type', 'image/png')
                 return f'data:{content_type};base64,{encoded_string}'
         else:
-            file = await Files.get_file_by_id(url)
-
-            if not file:
-                return None
-
-            file_path = await asyncio.to_thread(Storage.get_file, file.path)
-            file_path = Path(file_path)
-
-            if file_path.is_file():
-                with open(file_path, 'rb') as image_file:
-                    encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
-                    content_type = mimetypes.guess_type(file_path.name)[0] or (file.meta or {}).get('content_type')
-                    if not content_type and ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK:
-                        content_type = _IMAGE_MIME_FALLBACK.get(file_path.suffix.lower())
-                    if not content_type:
-                        return None
-                    return f'data:{content_type};base64,{encoded_string}'
-            else:
-                return None
+            # Non-URL string — treat as file_id. Delegate to the canonical
+            # file-ID resolver which enforces ownership/access checks.
+            return await get_image_base64_from_file_id(url, user=user)
 
     except Exception as e:
         return None
@@ -194,11 +179,25 @@ async def get_file_url_from_base64(request, base64_file_string, metadata, user):
     return None
 
 
-async def get_image_base64_from_file_id(id: str) -> Optional[str]:
+async def get_image_base64_from_file_id(id: str, user=None) -> Optional[str]:
     file = await Files.get_file_by_id(id)
     if not file:
         return None
 
+    # Gate file-by-id resolution by ownership to prevent exfiltration.
+    # A caller could place another user's file_id in an image_url field;
+    # without this check the server reads the file from disk, inlines it
+    # base64 into the LLM request, and the content leaks via OCR/describe.
+    # Owner, admin, and explicit read-grant holders are allowed.
+    if user is None:
+        return None
+    if (
+        file.user_id != user.id
+        and user.role != 'admin'
+        and not await has_access_to_file(file.id, 'read', user)
+    ):
+        return None
+
     try:
         file_path = await asyncio.to_thread(Storage.get_file, file.path)
         file_path = Path(file_path)