scriptoriumadmin/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-09-07 08:27:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: enforce per-model access on chained base models

Browse source 
When a custom model declares base_model_id, the access check on the
user-facing wrapper did not extend to the base it forwards to. A caller
with read access to the wrapper (ownership or grant) could reach an
upstream model they were not authorized for.

Add a helper that re-runs the model access check against the resolved
base, and wire it in at two layers: upfront when persisting a chain
(create/import/update) so callers cannot plant bases they cannot read,
and at dispatch time in every chat completion router path so access
revocations after chain creation are still honored.
This commit is contained in:
DrMelone 2026-04-13 01:19:31 +02:00
parent 67023037f8
commit ad6ea3622e
6 changed files with 100 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/open_webui/functions.py
View file

@ -34,6 +34,7 @@ from open_webui.utils.plugin import (
load_function_module_by_id,
get_function_module_from_cache,
)
from open_webui.utils.access_control import check_base_model_access
from open_webui.env import GLOBAL_LOG_LEVEL
@ -259,6 +260,7 @@ async def generate_function_chat_completion(request, form_data, user, models: di
if model_info:
if model_info.base_model_id:
form_data['model'] = model_info.base_model_id
await check_base_model_access(user, model_info.base_model_id)
params = model_info.params.model_dump()

9
backend/open_webui/main.py
View file

@ -1672,10 +1672,19 @@ async def chat_completion(
# Update model and form_data so routing uses the fallback model's type
model = request.app.state.MODELS[fallback_model_id]
form_data['model'] = fallback_model_id
resolved_base_model_id = fallback_model_id
else:
raise Exception('Model not found')
else:
raise Exception('Model not found')
else:
resolved_base_model_id = base_model_id
# Re-run the access check against the resolved base model. Access on
# the user-facing wrapper does not extend to the chain target, so
# each link must be authorized independently.
if not BYPASS_MODEL_ACCESS_CONTROL and (user.role != 'admin' or not BYPASS_ADMIN_ACCESS_CONTROL):
await check_model_access(user, request.app.state.MODELS[resolved_base_model_id])
# Chat Params
stream_delta_chunk_size = form_data.get('params', {}).get('stream_delta_chunk_size')

43
backend/open_webui/routers/models.py
View file

@ -33,8 +33,13 @@ from fastapi.responses import FileResponse, StreamingResponse
from open_webui.utils.auth import get_admin_user, get_verified_user
from open_webui.utils.access_control import has_permission, filter_allowed_access_grants
from open_webui.utils.access_control import (
has_permission,
filter_allowed_access_grants,
check_base_model_access,
)
from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL, STATIC_DIR
from open_webui.env import BYPASS_MODEL_ACCESS_CONTROL
from open_webui.internal.db import get_async_session
from sqlalchemy.ext.asyncio import AsyncSession
@ -196,6 +201,17 @@ async def create_new_model(
)
else:
# Reject planting a chain into a base model the caller cannot read.
# Ownership on the created wrapper would otherwise extend implicit
# read access to the base at dispatch time.
await check_base_model_access(
user,
form_data.base_model_id,
bypass_filter=BYPASS_MODEL_ACCESS_CONTROL
or (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL),
db=db,
)
form_data.access_grants = await filter_allowed_access_grants(
request.app.state.config.USER_PERMISSIONS,
user.id,
@ -281,6 +297,10 @@ async def import_models(
model.id: model for model in (await Models.get_models_by_ids(model_ids, db=db) if model_ids else [])
}
bypass = BYPASS_MODEL_ACCESS_CONTROL or (
user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL
)
for model_data in data:
# Here, you can add logic to validate model_data if needed
model_id = model_data.get('id')
@ -293,12 +313,21 @@ async def import_models(
model_data['params'] = model_data.get('params', {})
updated_model = ModelForm(**{**existing_model.model_dump(), **model_data})
# Only re-validate the base if the import actually changes it,
# so imports that preserve a pre-existing chain are not broken.
if updated_model.base_model_id != existing_model.base_model_id:
await check_base_model_access(
user, updated_model.base_model_id, bypass_filter=bypass, db=db
)
await Models.update_model_by_id(model_id, updated_model, db=db)
else:
# Insert new model
model_data['meta'] = model_data.get('meta', {})
model_data['params'] = model_data.get('params', {})
new_model = ModelForm(**model_data)
await check_base_model_access(
user, new_model.base_model_id, bypass_filter=bypass, db=db
)
await Models.insert_new_model(user_id=user.id, form_data=new_model, db=db)
return True
else:
@ -496,6 +525,18 @@ async def update_model_by_id(
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
)
# Re-validate the chain when the caller is changing base_model_id, so a
# user with write access to the wrapper cannot repoint it at a base they
# lack read access to.
if form_data.base_model_id != model.base_model_id:
await check_base_model_access(
user,
form_data.base_model_id,
bypass_filter=BYPASS_MODEL_ACCESS_CONTROL
or (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL),
db=db,
)
form_data.access_grants = await filter_allowed_access_grants(
request.app.state.config.USER_PERMISSIONS,
user.id,

11
backend/open_webui/routers/ollama.py
View file

@ -47,7 +47,7 @@ from open_webui.internal.db import get_async_session
from open_webui.models.models import Models
from open_webui.models.access_grants import AccessGrants
from open_webui.models.groups import Groups
from open_webui.utils.access_control import check_model_access
from open_webui.utils.access_control import check_model_access, check_base_model_access
from open_webui.utils.misc import (
calculate_sha256,
)
@ -1103,6 +1103,11 @@ async def generate_chat_completion(
) # Use request's base_model_id if available
payload['model'] = base_model_id
# Re-run the access check against the resolved base model. The check
# on the user-facing wrapper does not authorize the chain target:
# grants or ownership on the wrapper must not leak into the base.
await check_base_model_access(user, base_model_id, bypass_filter)
params = model_info.params.model_dump()
if params:
@ -1196,6 +1201,7 @@ async def generate_openai_completion(
if model_info:
if model_info.base_model_id:
payload['model'] = model_info.base_model_id
await check_base_model_access(user, model_info.base_model_id)
params = model_info.params.model_dump()
if params:
@ -1258,6 +1264,7 @@ async def generate_openai_chat_completion(
if model_info:
if model_info.base_model_id:
payload['model'] = model_info.base_model_id
await check_base_model_access(user, model_info.base_model_id)
params = model_info.params.model_dump()
@ -1318,6 +1325,7 @@ async def generate_anthropic_messages(
if model_info:
if model_info.base_model_id:
payload['model'] = model_info.base_model_id
await check_base_model_access(user, model_info.base_model_id)
await check_model_access(user, model_info)
else:
@ -1376,6 +1384,7 @@ async def generate_responses(
if model_info:
if model_info.base_model_id:
payload['model'] = model_info.base_model_id
await check_base_model_access(user, model_info.base_model_id)
# Check if user has access to the model
if user.role == 'user':

11
backend/open_webui/routers/openai.py
View file

@ -28,7 +28,11 @@ from open_webui.internal.db import get_async_session
from open_webui.models.models import Models
from open_webui.models.access_grants import AccessGrants
from open_webui.models.groups import Groups
from open_webui.utils.access_control import has_connection_access, check_model_access
from open_webui.utils.access_control import (
has_connection_access,
check_model_access,
check_base_model_access,
)
from open_webui.config import (
CACHE_DIR,
)
@ -1058,6 +1062,11 @@ async def generate_chat_completion(
payload['model'] = base_model_id
model_id = base_model_id
# Re-run the access check against the resolved base model. The check
# on the user-facing wrapper does not authorize the chain target:
# grants or ownership on the wrapper must not leak into the base.
await check_base_model_access(user, base_model_id, bypass_filter)
params = model_info.params.model_dump()
if params:

27
backend/open_webui/utils/access_control/__init__.py
View file

@ -299,3 +299,30 @@ async def check_model_access(
else:
if user.role != 'admin':
raise HTTPException(status_code=403, detail='Model not found')
async def check_base_model_access(
user: UserModel,
base_model_id: "str | None",
bypass_filter: bool = False,
db: AsyncSession | None = None,
) -> None:
"""
Enforce per-model read access for a chained base model.
A custom model that declares base_model_id forwards requests to that base at
dispatch time. Read access on the user-facing wrapper (ownership or a grant)
does not extend to the base model — each link in the chain is an independent
access decision. Call this before any code path that resolves or dispatches
to the base, and before persisting a base_model_id chosen by the caller.
Raises HTTPException(403) if not authorized. Does nothing if bypass_filter
is True or base_model_id is falsy.
"""
if bypass_filter or not base_model_id:
return
from open_webui.models.models import Models
base_model_info = await Models.get_model_by_id(base_model_id, db=db)
await check_model_access(user, base_model_info, bypass_filter)