diff --git a/backend/open_webui/routers/audio.py b/backend/open_webui/routers/audio.py index 27634cec199f..4f310f598522 100644 --- a/backend/open_webui/routers/audio.py +++ b/backend/open_webui/routers/audio.py @@ -49,6 +49,7 @@ DEVICE_TYPE, ENABLE_FORWARD_USER_INFO_HEADERS, ) +from open_webui.utils.content_types import get_stt_supported_content_types router = APIRouter() @@ -238,9 +239,11 @@ async def update_audio_config( request.app.state.config.STT_OPENAI_API_KEY = form_data.stt.OPENAI_API_KEY request.app.state.config.STT_ENGINE = form_data.stt.ENGINE request.app.state.config.STT_MODEL = form_data.stt.MODEL - request.app.state.config.STT_SUPPORTED_CONTENT_TYPES = ( - form_data.stt.SUPPORTED_CONTENT_TYPES - ) + request.app.state.config.STT_SUPPORTED_CONTENT_TYPES = [ + content_type.strip() + for content_type in form_data.stt.SUPPORTED_CONTENT_TYPES + if content_type.strip() + ] request.app.state.config.WHISPER_MODEL = form_data.stt.WHISPER_MODEL request.app.state.config.DEEPGRAM_API_KEY = form_data.stt.DEEPGRAM_API_KEY @@ -919,10 +922,9 @@ def transcription( ): log.info(f"file.content_type: {file.content_type}") - supported_content_types = request.app.state.config.STT_SUPPORTED_CONTENT_TYPES or [ - "audio/*", - "video/webm", - ] + supported_content_types = get_stt_supported_content_types( + request.app.state.config.STT_SUPPORTED_CONTENT_TYPES + ) if not any( fnmatch(file.content_type, content_type) diff --git a/backend/open_webui/routers/files.py b/backend/open_webui/routers/files.py index b9bb15c7b400..7b76cb5bbd32 100644 --- a/backend/open_webui/routers/files.py +++ b/backend/open_webui/routers/files.py @@ -36,6 +36,7 @@ from open_webui.routers.audio import transcribe from open_webui.storage.provider import Storage from open_webui.utils.auth import get_admin_user, get_verified_user +from open_webui.utils.content_types import get_stt_supported_content_types from pydantic import BaseModel log = logging.getLogger(__name__) @@ -155,12 +156,8 @@ def upload_file( if process: try: if file.content_type: - stt_supported_content_types = ( + stt_supported_content_types = get_stt_supported_content_types( request.app.state.config.STT_SUPPORTED_CONTENT_TYPES - or [ - "audio/*", - "video/webm", - ] ) if any( diff --git a/backend/open_webui/routers/ollama.py b/backend/open_webui/routers/ollama.py index 135359937447..2c7d4119b258 100644 --- a/backend/open_webui/routers/ollama.py +++ b/backend/open_webui/routers/ollama.py @@ -48,8 +48,9 @@ apply_model_params_to_body_openai, apply_model_system_prompt_to_body, ) +from open_webui.utils.messages import remove_user_system_messages_from_body from open_webui.utils.auth import get_admin_user, get_verified_user -from open_webui.utils.access_control import has_access +from open_webui.utils.access_control import has_access, has_permission from open_webui.config import ( @@ -69,6 +70,17 @@ log.setLevel(SRC_LOG_LEVELS["OLLAMA"]) +def user_can_use_system_prompt(request: Request, user: UserModel) -> bool: + if user.role == "admin": + return True + + return has_permission( + user.id, + "chat.system_prompt", + request.app.state.config.USER_PERMISSIONS, + ) + + ########################################## # # Utility functions @@ -1289,6 +1301,9 @@ async def generate_chat_completion( if "metadata" in payload: del payload["metadata"] + if not user_can_use_system_prompt(request, user): + payload = remove_user_system_messages_from_body(payload) + model_id = payload["model"] model_info = Models.get_model_by_id(model_id) @@ -1472,6 +1487,9 @@ async def generate_openai_chat_completion( if "metadata" in payload: del payload["metadata"] + if not user_can_use_system_prompt(request, user): + payload = remove_user_system_messages_from_body(payload) + model_id = completion_form.model if ":" not in model_id: model_id = f"{model_id}:latest" diff --git a/backend/open_webui/routers/openai.py b/backend/open_webui/routers/openai.py index 7649271feebb..2a29b5cb7ef4 100644 --- a/backend/open_webui/routers/openai.py +++ b/backend/open_webui/routers/openai.py @@ -37,18 +37,30 @@ apply_model_params_to_body_openai, apply_model_system_prompt_to_body, ) +from open_webui.utils.messages import remove_user_system_messages_from_body from open_webui.utils.misc import ( convert_logit_bias_input_to_json, ) from open_webui.utils.auth import get_admin_user, get_verified_user -from open_webui.utils.access_control import has_access +from open_webui.utils.access_control import has_access, has_permission log = logging.getLogger(__name__) log.setLevel(SRC_LOG_LEVELS["OPENAI"]) +def user_can_use_system_prompt(request: Request, user: UserModel) -> bool: + if user.role == "admin": + return True + + return has_permission( + user.id, + "chat.system_prompt", + request.app.state.config.USER_PERMISSIONS, + ) + + ########################################## # # Utility functions @@ -705,6 +717,9 @@ async def generate_chat_completion( payload = {**form_data} metadata = payload.pop("metadata", None) + if not user_can_use_system_prompt(request, user): + payload = remove_user_system_messages_from_body(payload) + model_id = form_data.get("model") model_info = Models.get_model_by_id(model_id) diff --git a/backend/open_webui/routers/users.py b/backend/open_webui/routers/users.py index 16cc2c375f95..58fc328c11c3 100644 --- a/backend/open_webui/routers/users.py +++ b/backend/open_webui/routers/users.py @@ -215,9 +215,10 @@ async def update_user_settings_by_session_user( request: Request, form_data: UserSettings, user=Depends(get_verified_user) ): updated_user_settings = form_data.model_dump() + ui_settings = updated_user_settings.get("ui") or {} if ( user.role != "admin" - and "toolServers" in updated_user_settings.get("ui").keys() + and "toolServers" in ui_settings.keys() and not has_permission( user.id, "features.direct_tool_servers", @@ -227,6 +228,15 @@ async def update_user_settings_by_session_user( # If the user is not an admin and does not have permission to use tool servers, remove the key updated_user_settings["ui"].pop("toolServers", None) + if user.role != "admin" and not has_permission( + user.id, + "chat.system_prompt", + request.app.state.config.USER_PERMISSIONS, + ): + updated_user_settings.pop("system", None) + if isinstance(updated_user_settings.get("params"), dict): + updated_user_settings["params"].pop("system", None) + user = Users.update_user_settings_by_id(user.id, updated_user_settings) if user: return user.settings diff --git a/backend/open_webui/utils/chat.py b/backend/open_webui/utils/chat.py index 268c910e3ec2..5a4bdd72f863 100644 --- a/backend/open_webui/utils/chat.py +++ b/backend/open_webui/utils/chat.py @@ -45,7 +45,11 @@ get_function_module_from_cache, ) from open_webui.utils.models import get_all_models, check_model_access -from open_webui.utils.payload import convert_payload_openai_to_ollama +from open_webui.utils.payload import ( + convert_payload_openai_to_ollama, +) +from open_webui.utils.messages import remove_user_system_messages_from_body +from open_webui.utils.access_control import has_permission from open_webui.utils.response import ( convert_response_ollama_to_openai, convert_streaming_response_ollama_to_openai, @@ -63,6 +67,17 @@ log.setLevel(SRC_LOG_LEVELS["MAIN"]) +def user_can_use_system_prompt(request: Request, user: Any) -> bool: + if user.role == "admin": + return True + + return has_permission( + user.id, + "chat.system_prompt", + request.app.state.config.USER_PERMISSIONS, + ) + + async def generate_direct_chat_completion( request: Request, form_data: dict, @@ -189,6 +204,9 @@ async def generate_chat_completion( if model_id not in models: raise Exception("Model not found") + if not user_can_use_system_prompt(request, user): + form_data = remove_user_system_messages_from_body(form_data) + model = models[model_id] if getattr(request.state, "direct", False): diff --git a/backend/open_webui/utils/content_types.py b/backend/open_webui/utils/content_types.py new file mode 100644 index 000000000000..42f1f9d6869f --- /dev/null +++ b/backend/open_webui/utils/content_types.py @@ -0,0 +1,14 @@ +from typing import Optional + + +DEFAULT_STT_SUPPORTED_CONTENT_TYPES = ["audio/*", "video/webm"] + + +def get_stt_supported_content_types(configured_content_types: Optional[list[str]]): + content_types = [ + content_type.strip() + for content_type in configured_content_types or [] + if isinstance(content_type, str) and content_type.strip() + ] + + return content_types or DEFAULT_STT_SUPPORTED_CONTENT_TYPES diff --git a/backend/open_webui/utils/messages.py b/backend/open_webui/utils/messages.py new file mode 100644 index 000000000000..e3d36a73c827 --- /dev/null +++ b/backend/open_webui/utils/messages.py @@ -0,0 +1,13 @@ +def remove_user_system_messages_from_body(form_data: dict) -> dict: + messages = form_data.get("messages") + if isinstance(messages, list): + form_data["messages"] = [ + message + for message in messages + if not ( + isinstance(message, dict) + and str(message.get("role", "")).lower() == "system" + ) + ] + + return form_data diff --git a/tests/test_critical_permission_regressions.py b/tests/test_critical_permission_regressions.py new file mode 100644 index 000000000000..a80c65ed24d0 --- /dev/null +++ b/tests/test_critical_permission_regressions.py @@ -0,0 +1,49 @@ +import importlib.util +from pathlib import Path + + +def load_module(module_name: str, relative_path: str): + module_path = Path(__file__).parents[1] / relative_path + spec = importlib.util.spec_from_file_location(module_name, module_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +content_types = load_module( + "content_types", "backend/open_webui/utils/content_types.py" +) +messages = load_module("messages", "backend/open_webui/utils/messages.py") + +DEFAULT_STT_SUPPORTED_CONTENT_TYPES = content_types.DEFAULT_STT_SUPPORTED_CONTENT_TYPES +get_stt_supported_content_types = content_types.get_stt_supported_content_types +remove_user_system_messages_from_body = messages.remove_user_system_messages_from_body + + +def test_blank_stt_content_types_fall_back_to_defaults(): + assert get_stt_supported_content_types(["", " "]) == DEFAULT_STT_SUPPORTED_CONTENT_TYPES + + +def test_stt_content_types_are_trimmed_and_empty_entries_removed(): + assert get_stt_supported_content_types([" audio/wav ", "", "video/mp4"]) == [ + "audio/wav", + "video/mp4", + ] + + +def test_user_system_messages_are_removed_from_payload(): + payload = { + "messages": [ + {"role": "system", "content": "user controlled"}, + {"role": "user", "content": "hello"}, + {"role": "SYSTEM", "content": "also user controlled"}, + {"role": "assistant", "content": "hi"}, + ] + } + + assert remove_user_system_messages_from_body(payload) == { + "messages": [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "hi"}, + ] + }