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.
This commit is contained in:
DrMelone 2026-04-12 17:48:38 +02:00
parent 92dfa3f2f2
commit af51217b81

View file

@ -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: