From af51217b81328fa6d208f5dbcb2dfefd918f0914 Mon Sep 17 00:00:00 2001
From: DrMelone <27028174+Classic298@users.noreply.github.com>
Date: Sun, 12 Apr 2026 17:48:38 +0200
Subject: [PATCH] fix: enforce ownership and access grant checks on model
 import

The model import endpoint allowed any user with workspace.models_import permission to overwrite existing models without verifying ownership or write access, and without filtering access grants through filter_allowed_access_grants.

Now enforces the same ownership/write-access check used by update_model_by_id, and applies filter_allowed_access_grants to both new and updated models during import.
---
 backend/open_webui/routers/models.py | 49 +++++++++++++++++++++++++++-
 1 file changed, 48 insertions(+), 1 deletion(-)

diff --git a/backend/open_webui/routers/models.py b/backend/open_webui/routers/models.py
index 6f7b3d48df..2191e61ad2 100644
--- a/backend/open_webui/routers/models.py
+++ b/backend/open_webui/routers/models.py
@@ -281,24 +281,71 @@ async def import_models(
                 model.id: model for model in (Models.get_models_by_ids(model_ids, db=db) if model_ids else [])
             }
 
+            # Batch-resolve write permissions in one query instead of
+            # per-model has_access calls (N+1 avoidance).
+            existing_model_ids = list(existing_models.keys())
+            if user.role != 'admin' and existing_model_ids:
+                user_group_ids = {
+                    group.id for group in Groups.get_groups_by_member_id(user.id, db=db)
+                }
+                writable_model_ids = AccessGrants.get_accessible_resource_ids(
+                    user_id=user.id,
+                    resource_type='model',
+                    resource_ids=existing_model_ids,
+                    permission='write',
+                    user_group_ids=user_group_ids,
+                    db=db,
+                )
+            else:
+                writable_model_ids = set(existing_model_ids)
+
             for model_data in data:
-                # Here, you can add logic to validate model_data if needed
                 model_id = model_data.get('id')
 
                 if model_id and is_valid_model_id(model_id):
                     existing_model = existing_models.get(model_id)
                     if existing_model:
+                        # Enforce ownership/write-access before allowing
+                        # overwrite — same check as update_model_by_id.
+                        if (
+                            user.role != 'admin'
+                            and existing_model.user_id != user.id
+                            and model_id not in writable_model_ids
+                        ):
+                            log.warning(
+                                f'Import: user {user.id} denied write access to model {model_id}, skipping'
+                            )
+                            continue
+
                         # Update existing model
                         model_data['meta'] = model_data.get('meta', {})
                         model_data['params'] = model_data.get('params', {})
 
                         updated_model = ModelForm(**{**existing_model.model_dump(), **model_data})
+                        # Only filter access_grants when explicitly provided
+                        # in the payload to avoid altering existing ACLs on
+                        # metadata-only imports.
+                        if 'access_grants' in model_data:
+                            updated_model.access_grants = filter_allowed_access_grants(
+                                request.app.state.config.USER_PERMISSIONS,
+                                user.id,
+                                user.role,
+                                updated_model.access_grants,
+                                'sharing.public_models',
+                            )
                         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)
+                        new_model.access_grants = filter_allowed_access_grants(
+                            request.app.state.config.USER_PERMISSIONS,
+                            user.id,
+                            user.role,
+                            new_model.access_grants,
+                            'sharing.public_models',
+                        )
                         Models.insert_new_model(user_id=user.id, form_data=new_model, db=db)
             return True
         else: