refac: ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS renamed to BYPASS_ADMIN_ACCESS_CONTROL
This commit is contained in:
		
							parent
							
								
									02479425a5
								
							
						
					
					
						commit
						e6da38464b
					
				|  | @ -1358,6 +1358,14 @@ ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS = ( | ||||||
|     os.environ.get("ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS", "True").lower() == "true" |     os.environ.get("ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS", "True").lower() == "true" | ||||||
| ) | ) | ||||||
| 
 | 
 | ||||||
|  | BYPASS_ADMIN_ACCESS_CONTROL = ( | ||||||
|  |     os.environ.get( | ||||||
|  |         "BYPASS_ADMIN_ACCESS_CONTROL", | ||||||
|  |         os.environ.get("ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS", "True"), | ||||||
|  |     ).lower() | ||||||
|  |     == "true" | ||||||
|  | ) | ||||||
|  | 
 | ||||||
| ENABLE_ADMIN_CHAT_ACCESS = ( | ENABLE_ADMIN_CHAT_ACCESS = ( | ||||||
|     os.environ.get("ENABLE_ADMIN_CHAT_ACCESS", "True").lower() == "true" |     os.environ.get("ENABLE_ADMIN_CHAT_ACCESS", "True").lower() == "true" | ||||||
| ) | ) | ||||||
|  |  | ||||||
|  | @ -329,7 +329,7 @@ from open_webui.config import ( | ||||||
|     ENABLE_MESSAGE_RATING, |     ENABLE_MESSAGE_RATING, | ||||||
|     ENABLE_USER_WEBHOOKS, |     ENABLE_USER_WEBHOOKS, | ||||||
|     ENABLE_EVALUATION_ARENA_MODELS, |     ENABLE_EVALUATION_ARENA_MODELS, | ||||||
|     ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS, |     BYPASS_ADMIN_ACCESS_CONTROL, | ||||||
|     USER_PERMISSIONS, |     USER_PERMISSIONS, | ||||||
|     DEFAULT_USER_ROLE, |     DEFAULT_USER_ROLE, | ||||||
|     PENDING_USER_OVERLAY_CONTENT, |     PENDING_USER_OVERLAY_CONTENT, | ||||||
|  | @ -378,7 +378,7 @@ from open_webui.config import ( | ||||||
|     RESPONSE_WATERMARK, |     RESPONSE_WATERMARK, | ||||||
|     # Admin |     # Admin | ||||||
|     ENABLE_ADMIN_CHAT_ACCESS, |     ENABLE_ADMIN_CHAT_ACCESS, | ||||||
|     ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS, |     BYPASS_ADMIN_ACCESS_CONTROL, | ||||||
|     ENABLE_ADMIN_EXPORT, |     ENABLE_ADMIN_EXPORT, | ||||||
|     # Tasks |     # Tasks | ||||||
|     TASK_MODEL, |     TASK_MODEL, | ||||||
|  | @ -1290,7 +1290,7 @@ async def get_models( | ||||||
|             model_info = Models.get_model_by_id(model["id"]) |             model_info = Models.get_model_by_id(model["id"]) | ||||||
|             if model_info: |             if model_info: | ||||||
|                 if ( |                 if ( | ||||||
|                     (user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS) |                     (user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL) | ||||||
|                     or user.id == model_info.user_id |                     or user.id == model_info.user_id | ||||||
|                     or has_access( |                     or has_access( | ||||||
|                         user.id, type="read", access_control=model_info.access_control |                         user.id, type="read", access_control=model_info.access_control | ||||||
|  | @ -1338,7 +1338,7 @@ async def get_models( | ||||||
|     # Filter out models that the user does not have access to |     # Filter out models that the user does not have access to | ||||||
|     if ( |     if ( | ||||||
|         user.role == "user" |         user.role == "user" | ||||||
|         or (user.role == "admin" and not ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS) |         or (user.role == "admin" and not BYPASS_ADMIN_ACCESS_CONTROL) | ||||||
|     ) and not BYPASS_MODEL_ACCESS_CONTROL: |     ) and not BYPASS_MODEL_ACCESS_CONTROL: | ||||||
|         models = get_filtered_models(models, user) |         models = get_filtered_models(models, user) | ||||||
| 
 | 
 | ||||||
|  | @ -1411,7 +1411,7 @@ async def chat_completion( | ||||||
| 
 | 
 | ||||||
|             # Check if user has access to the model |             # Check if user has access to the model | ||||||
|             if not BYPASS_MODEL_ACCESS_CONTROL and ( |             if not BYPASS_MODEL_ACCESS_CONTROL and ( | ||||||
|                 user.role != "admin" or not ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS |                 user.role != "admin" or not BYPASS_ADMIN_ACCESS_CONTROL | ||||||
|             ): |             ): | ||||||
|                 try: |                 try: | ||||||
|                     check_model_access(user, model) |                     check_model_access(user, model) | ||||||
|  |  | ||||||
|  | @ -25,7 +25,7 @@ from open_webui.utils.access_control import has_access, has_permission | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
| from open_webui.env import SRC_LOG_LEVELS | from open_webui.env import SRC_LOG_LEVELS | ||||||
| from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS | from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL | ||||||
| from open_webui.models.models import Models, ModelForm | from open_webui.models.models import Models, ModelForm | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
|  | @ -43,7 +43,7 @@ router = APIRouter() | ||||||
| async def get_knowledge(user=Depends(get_verified_user)): | async def get_knowledge(user=Depends(get_verified_user)): | ||||||
|     knowledge_bases = [] |     knowledge_bases = [] | ||||||
| 
 | 
 | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         knowledge_bases = Knowledges.get_knowledge_bases() |         knowledge_bases = Knowledges.get_knowledge_bases() | ||||||
|     else: |     else: | ||||||
|         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "read") |         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "read") | ||||||
|  | @ -91,7 +91,7 @@ async def get_knowledge(user=Depends(get_verified_user)): | ||||||
| async def get_knowledge_list(user=Depends(get_verified_user)): | async def get_knowledge_list(user=Depends(get_verified_user)): | ||||||
|     knowledge_bases = [] |     knowledge_bases = [] | ||||||
| 
 | 
 | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         knowledge_bases = Knowledges.get_knowledge_bases() |         knowledge_bases = Knowledges.get_knowledge_bases() | ||||||
|     else: |     else: | ||||||
|         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "write") |         knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "write") | ||||||
|  |  | ||||||
|  | @ -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.auth import get_admin_user, get_verified_user | ||||||
| from open_webui.utils.access_control import has_access, has_permission | from open_webui.utils.access_control import has_access, has_permission | ||||||
| from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS | from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL | ||||||
| 
 | 
 | ||||||
| router = APIRouter() | router = APIRouter() | ||||||
| 
 | 
 | ||||||
|  | @ -27,7 +27,7 @@ router = APIRouter() | ||||||
| 
 | 
 | ||||||
| @router.get("/", response_model=list[ModelUserResponse]) | @router.get("/", response_model=list[ModelUserResponse]) | ||||||
| async def get_models(id: Optional[str] = None, user=Depends(get_verified_user)): | async def get_models(id: Optional[str] = None, user=Depends(get_verified_user)): | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         return Models.get_models() |         return Models.get_models() | ||||||
|     else: |     else: | ||||||
|         return Models.get_models_by_user_id(user.id) |         return Models.get_models_by_user_id(user.id) | ||||||
|  | @ -117,7 +117,7 @@ async def get_model_by_id(id: str, user=Depends(get_verified_user)): | ||||||
|     model = Models.get_model_by_id(id) |     model = Models.get_model_by_id(id) | ||||||
|     if model: |     if model: | ||||||
|         if ( |         if ( | ||||||
|             (user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS) |             (user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL) | ||||||
|             or model.user_id == user.id |             or model.user_id == user.id | ||||||
|             or has_access(user.id, "read", model.access_control) |             or has_access(user.id, "read", model.access_control) | ||||||
|         ): |         ): | ||||||
|  |  | ||||||
|  | @ -10,7 +10,7 @@ from open_webui.models.prompts import ( | ||||||
| from open_webui.constants import ERROR_MESSAGES | from open_webui.constants import ERROR_MESSAGES | ||||||
| from open_webui.utils.auth import get_admin_user, get_verified_user | 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.utils.access_control import has_access, has_permission | ||||||
| from open_webui.config import ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS | from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL | ||||||
| 
 | 
 | ||||||
| router = APIRouter() | router = APIRouter() | ||||||
| 
 | 
 | ||||||
|  | @ -21,7 +21,7 @@ router = APIRouter() | ||||||
| 
 | 
 | ||||||
| @router.get("/", response_model=list[PromptModel]) | @router.get("/", response_model=list[PromptModel]) | ||||||
| async def get_prompts(user=Depends(get_verified_user)): | async def get_prompts(user=Depends(get_verified_user)): | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         prompts = Prompts.get_prompts() |         prompts = Prompts.get_prompts() | ||||||
|     else: |     else: | ||||||
|         prompts = Prompts.get_prompts_by_user_id(user.id, "read") |         prompts = Prompts.get_prompts_by_user_id(user.id, "read") | ||||||
|  | @ -31,7 +31,7 @@ async def get_prompts(user=Depends(get_verified_user)): | ||||||
| 
 | 
 | ||||||
| @router.get("/list", response_model=list[PromptUserResponse]) | @router.get("/list", response_model=list[PromptUserResponse]) | ||||||
| async def get_prompt_list(user=Depends(get_verified_user)): | async def get_prompt_list(user=Depends(get_verified_user)): | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         prompts = Prompts.get_prompts() |         prompts = Prompts.get_prompts() | ||||||
|     else: |     else: | ||||||
|         prompts = Prompts.get_prompts_by_user_id(user.id, "write") |         prompts = Prompts.get_prompts_by_user_id(user.id, "write") | ||||||
|  |  | ||||||
|  | @ -22,7 +22,7 @@ from open_webui.utils.access_control import has_access, has_permission | ||||||
| from open_webui.utils.tools import get_tool_servers | from open_webui.utils.tools import get_tool_servers | ||||||
| 
 | 
 | ||||||
| from open_webui.env import SRC_LOG_LEVELS | from open_webui.env import SRC_LOG_LEVELS | ||||||
| from open_webui.config import CACHE_DIR, ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS | from open_webui.config import CACHE_DIR, BYPASS_ADMIN_ACCESS_CONTROL | ||||||
| from open_webui.constants import ERROR_MESSAGES | from open_webui.constants import ERROR_MESSAGES | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
|  | @ -67,7 +67,7 @@ async def get_tools(request: Request, user=Depends(get_verified_user)): | ||||||
|             ) |             ) | ||||||
|         ) |         ) | ||||||
| 
 | 
 | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         # Admin can see all tools |         # Admin can see all tools | ||||||
|         return tools |         return tools | ||||||
|     else: |     else: | ||||||
|  | @ -87,7 +87,7 @@ async def get_tools(request: Request, user=Depends(get_verified_user)): | ||||||
| 
 | 
 | ||||||
| @router.get("/list", response_model=list[ToolUserResponse]) | @router.get("/list", response_model=list[ToolUserResponse]) | ||||||
| async def get_tool_list(user=Depends(get_verified_user)): | async def get_tool_list(user=Depends(get_verified_user)): | ||||||
|     if user.role == "admin" and ENABLE_ADMIN_WORKSPACE_CONTENT_ACCESS: |     if user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL: | ||||||
|         tools = Tools.get_tools() |         tools = Tools.get_tools() | ||||||
|     else: |     else: | ||||||
|         tools = Tools.get_tools_by_user_id(user.id, "write") |         tools = Tools.get_tools_by_user_id(user.id, "write") | ||||||
|  |  | ||||||
		Loading…
	
		Reference in New Issue