fix: enforce model access control on /responses endpoint (#23481)

The /responses proxy endpoint only required authentication via
get_verified_user but did not check per-model access grants. This
allowed any authenticated user to access any model through this
endpoint, bypassing the access control system.

Extract a shared check_model_access helper into utils/access_control
and replace all inline access control blocks across openai.py and
ollama.py (7 locations) with calls to this helper. This eliminates
code duplication and prevents future policy drift between endpoints.

CWE-862: Missing Authorization
CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:L/I:N/A:H (6.5 Medium)
This commit is contained in:
Classic298
2026-04-12 11:06:33 -05:00
committed by GitHub
parent b0df527224
commit e790e7be7a
3 changed files with 64 additions and 113 deletions
+10 -89
View File
@@ -47,6 +47,7 @@ from open_webui.internal.db import get_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.misc import (
calculate_sha256,
cleanup_response,
@@ -1097,29 +1098,9 @@ async def generate_chat_completion(
if not bypass_system_prompt:
payload = apply_system_prompt_to_body(system, payload, metadata, user)
# Check if user has access to the model
if not bypass_filter and user.role == 'user':
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not (
user.id == model_info.user_id
or AccessGrants.has_access(
user_id=user.id,
resource_type='model',
resource_id=model_info.id,
permission='read',
user_group_ids=user_group_ids,
)
):
raise HTTPException(
status_code=403,
detail='Model not found',
)
elif not bypass_filter:
if user.role != 'admin':
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, model_info, bypass_filter)
else:
check_model_access(user, None, bypass_filter)
url, url_idx = await get_ollama_url(request, payload['model'], url_idx)
api_config = request.app.state.config.OLLAMA_API_CONFIGS.get(
@@ -1206,29 +1187,9 @@ async def generate_openai_completion(
if params:
payload = apply_model_params_to_body_openai(params, payload)
# Check if user has access to the model
if user.role == 'user':
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not (
user.id == model_info.user_id
or AccessGrants.has_access(
user_id=user.id,
resource_type='model',
resource_id=model_info.id,
permission='read',
user_group_ids=user_group_ids,
)
):
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, model_info)
else:
if user.role != 'admin':
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, None)
url, url_idx = await get_ollama_url(request, payload['model'], url_idx)
api_config = request.app.state.config.OLLAMA_API_CONFIGS.get(
@@ -1292,29 +1253,9 @@ async def generate_openai_chat_completion(
payload = apply_model_params_to_body_openai(params, payload)
payload = apply_system_prompt_to_body(system, payload, metadata, user)
# Check if user has access to the model
if user.role == 'user':
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not (
user.id == model_info.user_id
or AccessGrants.has_access(
user_id=user.id,
resource_type='model',
resource_id=model_info.id,
permission='read',
user_group_ids=user_group_ids,
)
):
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, model_info)
else:
if user.role != 'admin':
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, None)
url, url_idx = await get_ollama_url(request, payload['model'], url_idx)
api_config = request.app.state.config.OLLAMA_API_CONFIGS.get(
@@ -1364,29 +1305,9 @@ async def generate_anthropic_messages(
if model_info.base_model_id:
payload['model'] = model_info.base_model_id
# Check if user has access to the model
if user.role == 'user':
user_group_ids = {group.id for group in Groups.get_groups_by_member_id(user.id)}
if not (
user.id == model_info.user_id
or AccessGrants.has_access(
user_id=user.id,
resource_type='model',
resource_id=model_info.id,
permission='read',
user_group_ids=user_group_ids,
)
):
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, model_info)
else:
if user.role != 'admin':
raise HTTPException(
status_code=403,
detail='Model not found',
)
check_model_access(user, None)
url, url_idx = await get_ollama_url(request, payload['model'], url_idx)
api_config = request.app.state.config.OLLAMA_API_CONFIGS.get(