From 50363ba66b19613a2fc0cab6a3f7f724a825135e Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Fri, 17 Apr 2026 13:52:11 +0900 Subject: [PATCH] refac --- .../utils/access_control/__init__.py | 47 +++++++++++++++++++ backend/open_webui/utils/models.py | 7 ++- 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/backend/open_webui/utils/access_control/__init__.py b/backend/open_webui/utils/access_control/__init__.py index 41c888d44..628e05276 100644 --- a/backend/open_webui/utils/access_control/__init__.py +++ b/backend/open_webui/utils/access_control/__init__.py @@ -257,6 +257,47 @@ async def filter_allowed_access_grants( return access_grants +async def has_base_model_access( + user_id: str, + model_info, + *, + user_group_ids: set[str] | None = None, + db=None, +) -> bool: + """ + Walk the ``base_model_id`` chain and verify the caller has read access + at every hop. + + Returns ``True`` when access is granted (or the chain ends at a raw + provider model that has no per-model ACL). Returns ``False`` the + moment a registered base model denies access. + """ + from open_webui.models.models import Models + from open_webui.models.access_grants import AccessGrants + + base_model_id = getattr(model_info, 'base_model_id', None) + seen = {model_info.id} + while base_model_id and base_model_id not in seen: + seen.add(base_model_id) + base_model_info = await Models.get_model_by_id(base_model_id, db=db) + if base_model_info is None: + break # Raw provider model — no per-model ACL + if not ( + user_id == base_model_info.user_id + or await AccessGrants.has_access( + user_id=user_id, + resource_type='model', + resource_id=base_model_info.id, + permission='read', + user_group_ids=user_group_ids, + db=db, + ) + ): + return False + base_model_id = getattr(base_model_info, 'base_model_id', None) + return True + + async def check_model_access( user: UserModel, model_info, @@ -296,6 +337,12 @@ async def check_model_access( ) ): raise HTTPException(status_code=403, detail='Model not found') + + # Enforce access on chained base models + if not await has_base_model_chain_access( + user.id, model_info, user_group_ids=user_group_ids + ): + raise HTTPException(status_code=403, detail='Model not found') else: if user.role != 'admin': raise HTTPException(status_code=403, detail='Model not found') diff --git a/backend/open_webui/utils/models.py b/backend/open_webui/utils/models.py index c8ebc190e..768f7588d 100644 --- a/backend/open_webui/utils/models.py +++ b/backend/open_webui/utils/models.py @@ -22,7 +22,7 @@ from open_webui.utils.plugin import ( load_function_module_by_id, get_function_module_from_cache, ) -from open_webui.utils.access_control import has_access +from open_webui.utils.access_control import has_access, has_base_model_access from open_webui.config import ( @@ -404,6 +404,11 @@ async def check_model_access(user, model, db=None): ): raise Exception('Model not found') + # Enforce access on chained base models + if not await has_base_model_access(user.id, model_info, db=db): + raise Exception('Model not found') + + async def get_filtered_models(models, user, db=None): # Filter out models that the user does not have access to