feat: ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS

Co-Authored-By: Classic298 <27028174+Classic298@users.noreply.github.com>

backend/open_webui/config.py

@@ -696,8 +696,12 @@ def load_oauth_providers():
 
     if configured_providers and not OPENID_PROVIDER_URL.value:
         provider_list = ", ".join(configured_providers)
-        log.warning(f"⚠️  OAuth providers configured ({provider_list}) but OPENID_PROVIDER_URL not set - logout will not work!")
+        log.warning(
-        log.warning(f"Set OPENID_PROVIDER_URL to your OAuth provider's OpenID Connect discovery endpoint to fix logout functionality.")
+            f"⚠️  OAuth providers configured ({provider_list}) but OPENID_PROVIDER_URL not set - logout will not work!"
+        )
+        log.warning(
+            f"Set OPENID_PROVIDER_URL to your OAuth provider's OpenID Connect discovery endpoint to fix logout functionality."
+        )
 
 
 load_oauth_providers()
@@ -1328,6 +1332,10 @@ WEBHOOK_URL = PersistentConfig(
 
 ENABLE_ADMIN_EXPORT = os.environ.get("ENABLE_ADMIN_EXPORT", "True").lower() == "true"
 
+ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS = (
+    os.environ.get("ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS", "True").lower() == "true"
+)
+
 ENABLE_ADMIN_CHAT_ACCESS = (
     os.environ.get("ENABLE_ADMIN_CHAT_ACCESS", "True").lower() == "true"
 )

backend/open_webui/routers/knowledge.py

@@ -25,6 +25,7 @@ from open_webui.utils.access_control import has_access, has_permission
 
 
 from open_webui.env import SRC_LOG_LEVELS
+from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS
 from open_webui.models.models import Models, ModelForm
 
 
@@ -42,7 +43,7 @@ router = APIRouter()
 async def get_knowledge(user=Depends(get_verified_user)):
     knowledge_bases = []
 
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         knowledge_bases = Knowledges.get_knowledge_bases()
     else:
         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "read")
@@ -90,7 +91,7 @@ async def get_knowledge(user=Depends(get_verified_user)):
 async def get_knowledge_list(user=Depends(get_verified_user)):
     knowledge_bases = []
 
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         knowledge_bases = Knowledges.get_knowledge_bases()
     else:
         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "write")

backend/open_webui/routers/models.py

@@ -15,7 +15,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status
 
 from open_webui.utils.auth import get_admin_user, get_verified_user
 from open_webui.utils.access_control import has_access, has_permission
-
+from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS
 
 router = APIRouter()
 
@@ -27,7 +27,7 @@ router = APIRouter()
 
 @router.get("/", response_model=list[ModelUserResponse])
 async def get_models(id: Optional[str] = None, user=Depends(get_verified_user)):
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         return Models.get_models()
     else:
         return Models.get_models_by_user_id(user.id)

backend/open_webui/routers/prompts.py

@@ -1,4 +1,5 @@
 from typing import Optional
+from fastapi import APIRouter, Depends, HTTPException, status, Request
 
 from open_webui.models.prompts import (
     PromptForm,
@@ -7,9 +8,9 @@ from open_webui.models.prompts import (
     Prompts,
 )
 from open_webui.constants import ERROR_MESSAGES
-from fastapi import APIRouter, Depends, HTTPException, status, Request
 from open_webui.utils.auth import get_admin_user, get_verified_user
 from open_webui.utils.access_control import has_access, has_permission
+from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS
 
 router = APIRouter()
 
@@ -20,7 +21,7 @@ router = APIRouter()
 
 @router.get("/", response_model=list[PromptModel])
 async def get_prompts(user=Depends(get_verified_user)):
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         prompts = Prompts.get_prompts()
     else:
         prompts = Prompts.get_prompts_by_user_id(user.id, "read")
@@ -30,7 +31,7 @@ async def get_prompts(user=Depends(get_verified_user)):
 
 @router.get("/list", response_model=list[PromptUserResponse])
 async def get_prompt_list(user=Depends(get_verified_user)):
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         prompts = Prompts.get_prompts()
     else:
         prompts = Prompts.get_prompts_by_user_id(user.id, "write")

backend/open_webui/routers/tools.py

@@ -5,6 +5,8 @@ import time
 import re
 import aiohttp
 from pydantic import BaseModel, HttpUrl
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+
 
 from open_webui.models.tools import (
     ToolForm,
@@ -14,16 +16,15 @@ from open_webui.models.tools import (
     Tools,
 )
 from open_webui.utils.plugin import load_tool_module_by_id, replace_imports
-from open_webui.config import CACHE_DIR
-from open_webui.constants import ERROR_MESSAGES
-from fastapi import APIRouter, Depends, HTTPException, Request, status
 from open_webui.utils.tools import get_tool_specs
 from open_webui.utils.auth import get_admin_user, get_verified_user
 from open_webui.utils.access_control import has_access, has_permission
-from open_webui.env import SRC_LOG_LEVELS
-
 from open_webui.utils.tools import get_tool_servers_data
 
+from open_webui.env import SRC_LOG_LEVELS
+from open_webui.config import CACHE_DIR, ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS
+from open_webui.constants import ERROR_MESSAGES
+
 
 log = logging.getLogger(__name__)
 log.setLevel(SRC_LOG_LEVELS["MAIN"])
@@ -74,14 +75,16 @@ async def get_tools(request: Request, user=Depends(get_verified_user)):
             )
         )
 
-    if user.role != "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
+        # Admin can see all tools
+        return tools
+    else:
         tools = [
             tool
             for tool in tools
             if tool.user_id == user.id
             or has_access(user.id, "read", tool.access_control)
         ]
-
         return tools
 
 
@@ -92,7 +95,7 @@ async def get_tools(request: Request, user=Depends(get_verified_user)):
 
 @router.get("/list", response_model=list[ToolUserResponse])
 async def get_tool_list(user=Depends(get_verified_user)):
-    if user.role == "admin":
+    if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS:
         tools = Tools.get_tools()
     else:
         tools = Tools.get_tools_by_user_id(user.id, "write")