mirror of
https://github.com/inventree/InvenTree.git
synced 2026-09-09 22:30:17 +00:00
[bug] API permission fixes: (#12766)
- User can only change their own profile - Users can only access notes against models they are scoped to - Users can only access parameters against models they are scoped to Co-authored-by: Matthias Mair <code@mjmair.com>
This commit is contained in:
@@ -486,6 +486,11 @@ class ContentTypePermission(OASTokenMixin, permissions.BasePermission):
|
||||
|
||||
def has_object_permission(self, request, view, obj):
|
||||
"""Check if the user has permission to access the object."""
|
||||
permission = 'view' if request.method in permissions.SAFE_METHODS else 'change'
|
||||
|
||||
if hasattr(obj, 'check_permission'):
|
||||
return obj.check_permission(permission, request.user)
|
||||
|
||||
if model_class := obj.__class__:
|
||||
return users.permissions.check_user_permission(
|
||||
request.user, model_class, 'change'
|
||||
|
||||
@@ -499,8 +499,37 @@ class NotesImageList(ListCreateAPI):
|
||||
|
||||
filter_backends = SEARCH_ORDER_FILTER
|
||||
|
||||
def get_queryset(self):
|
||||
"""Filter notes images to those linked to a note the requesting user can view."""
|
||||
import common.validators
|
||||
from users.permissions import check_user_permission, prefetch_rule_sets
|
||||
|
||||
qs = super().get_queryset()
|
||||
user = self.request.user
|
||||
|
||||
if user.is_superuser:
|
||||
return qs
|
||||
|
||||
groups = prefetch_rule_sets(user)
|
||||
|
||||
allowed_ct_ids = [
|
||||
ContentType.objects.get_for_model(model_class).pk
|
||||
for model_class in common.validators.note_model_types()
|
||||
if check_user_permission(user, model_class, 'view', groups=groups)
|
||||
]
|
||||
|
||||
return qs.filter(
|
||||
Q(note__template=True) | Q(note__model_type__in=allowed_ct_ids)
|
||||
)
|
||||
|
||||
def perform_create(self, serializer):
|
||||
"""Create (upload) a new notes image."""
|
||||
note = serializer.validated_data['note']
|
||||
|
||||
common.serializers.check_note_change_permission(
|
||||
self.request.user, template=note.template, model_type=note.model_type
|
||||
)
|
||||
|
||||
serializer.save(user=self.request.user)
|
||||
|
||||
|
||||
@@ -1315,6 +1344,35 @@ class ParameterMixin:
|
||||
serializer_class = common.serializers.ParameterSerializer
|
||||
permission_classes = [IsAuthenticatedOrReadScope]
|
||||
|
||||
def get_queryset(self):
|
||||
"""Filter parameters to those the requesting user has view permission for.
|
||||
|
||||
Parameter has no RuleSet permissions of its own (see
|
||||
users.ruleset.get_ruleset_ignore()) - access is instead scoped by the
|
||||
'view' permission of the model type the parameter is linked to.
|
||||
"""
|
||||
import common.validators
|
||||
from users.permissions import check_user_permission, prefetch_rule_sets
|
||||
|
||||
qs = super().get_queryset()
|
||||
user = self.request.user
|
||||
|
||||
if user.is_superuser:
|
||||
return qs
|
||||
|
||||
# Fetch the user's groups (with prefetched rule sets) once, and reuse it
|
||||
# for every model type below - otherwise each check_user_permission()
|
||||
# call re-fetches the same groups/rule-sets from scratch.
|
||||
groups = prefetch_rule_sets(user)
|
||||
|
||||
allowed_ct_ids = [
|
||||
ContentType.objects.get_for_model(model_class).pk
|
||||
for model_class in common.validators.parameter_model_types()
|
||||
if check_user_permission(user, model_class, 'view', groups=groups)
|
||||
]
|
||||
|
||||
return qs.filter(model_type__in=allowed_ct_ids)
|
||||
|
||||
|
||||
class ParameterList(
|
||||
OutputOptionsMixin,
|
||||
@@ -1346,10 +1404,44 @@ class ParameterList(
|
||||
|
||||
unique_create_fields = ['model_type', 'model_id', 'template']
|
||||
|
||||
def validate_delete(self, queryset, request) -> None:
|
||||
"""Ensure that the user has correct permissions for a bulk-delete.
|
||||
|
||||
- Extract all model types from the provided queryset
|
||||
- Ensure that the user has correct 'delete' permissions for each linked model
|
||||
"""
|
||||
from users.permissions import check_user_permission
|
||||
|
||||
content_type_ids = queryset.values_list('model_type', flat=True).distinct()
|
||||
|
||||
for content_type in ContentType.objects.filter(pk__in=content_type_ids):
|
||||
model_class = content_type.model_class()
|
||||
|
||||
if not model_class or not check_user_permission(
|
||||
request.user, model_class, 'delete'
|
||||
):
|
||||
raise ValidationError(
|
||||
_('User does not have permission to delete these parameters')
|
||||
)
|
||||
|
||||
|
||||
class ParameterDetail(ParameterMixin, RetrieveUpdateDestroyAPI):
|
||||
"""Detail API endpoint for Parameter objects."""
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
"""Enforce a delete permission check on the linked model before deleting.
|
||||
|
||||
DRF's default destroy() calls instance.delete() directly, bypassing
|
||||
ParameterSerializer.save() (and the permission checks it performs)
|
||||
entirely. Without this, get_queryset()'s 'view' permission gate is
|
||||
all that stands between a user and deleting the parameter.
|
||||
"""
|
||||
if not instance.check_permission('delete', self.request.user):
|
||||
raise PermissionDenied(
|
||||
_('User does not have permission to delete this parameter')
|
||||
)
|
||||
super().perform_destroy(instance)
|
||||
|
||||
|
||||
class InstanceInfoView(APIView):
|
||||
"""Return aggregated attachment/note/parameter counts for a single model instance.
|
||||
|
||||
@@ -170,6 +170,10 @@ class ProjectCode(InvenTree.models.InvenTreeMetadataModel):
|
||||
"""String representation of a ProjectCode."""
|
||||
return self.code
|
||||
|
||||
def check_permission(self, permission, user):
|
||||
"""Check if the user has the required permission for this project code."""
|
||||
return permission == 'view' or user.is_staff
|
||||
|
||||
code = models.CharField(
|
||||
max_length=50,
|
||||
unique=True,
|
||||
@@ -3299,6 +3303,20 @@ class Note(
|
||||
self.check_delete()
|
||||
super().delete(*args, **kwargs)
|
||||
|
||||
def check_permission(self, permission, user):
|
||||
"""Check if the user has the required permission for this note."""
|
||||
from InvenTree.models import InvenTreeNoteMixin
|
||||
|
||||
if self.template:
|
||||
return user.is_staff
|
||||
|
||||
model_class = self.model_type.model_class() if self.model_type else None
|
||||
|
||||
if not model_class or not issubclass(model_class, InvenTreeNoteMixin):
|
||||
return False
|
||||
|
||||
return model_class.check_related_permission(permission, user)
|
||||
|
||||
def cleanup_images(self):
|
||||
"""Remove any images which are no longer referenced in the note content."""
|
||||
for image in self.images.all():
|
||||
|
||||
@@ -595,6 +595,12 @@ class ParameterAPITests(InvenTreeAPITestCase):
|
||||
self.assertEqual(data['data'], '-2 inches')
|
||||
self.assertAlmostEqual(data['data_numeric'], -50.8, places=2)
|
||||
|
||||
# Deleting requires 'delete' permission against the linked model - the
|
||||
# 'add' permission granted above is not sufficient
|
||||
response = self.delete(url, expected_code=403)
|
||||
|
||||
self.assignRole('part.delete')
|
||||
|
||||
# Finally, delete the Parameter via the API
|
||||
response = self.delete(url, expected_code=204)
|
||||
|
||||
@@ -2847,6 +2853,10 @@ class NotesImageAPITests(InvenTreeAPITestCase):
|
||||
|
||||
super().setUp()
|
||||
|
||||
# 'change' also grants 'view' (see RuleSet.save()), covering both the
|
||||
# upload permission check and the get_queryset() view-permission filter
|
||||
self.assignRole('part.change')
|
||||
|
||||
self.part = Part.objects.create(name='Notes Image Test Part', description='x')
|
||||
ct = ContentType.objects.get_for_model(Part)
|
||||
self.note = common.models.Note.objects.create(
|
||||
@@ -2929,3 +2939,307 @@ class NotesImageAPITests(InvenTreeAPITestCase):
|
||||
buf = io.BytesIO()
|
||||
Image.new('RGB', (16, 16), color='red').save(buf, format='PNG')
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
class ParameterPermissionAPITests(InvenTreeAPITestCase):
|
||||
"""Tests for Parameter API permission enforcement."""
|
||||
|
||||
# No roles by default - each test assigns only what it needs
|
||||
roles = []
|
||||
|
||||
def setUp(self):
|
||||
"""Create a Part, ParameterTemplate and pre-existing Parameter."""
|
||||
from part.models import Part
|
||||
|
||||
super().setUp()
|
||||
|
||||
self.part = Part.objects.create(
|
||||
name='Perm Test Part', description='Part for permission testing'
|
||||
)
|
||||
|
||||
self.template = common.models.ParameterTemplate.objects.create(
|
||||
name='Perm Test Template', model_type=self.part.get_content_type()
|
||||
)
|
||||
|
||||
# Create a parameter directly via ORM (bypasses API permission checks)
|
||||
self.parameter = common.models.Parameter.objects.create(
|
||||
model_type=self.part.get_content_type(),
|
||||
model_id=self.part.pk,
|
||||
template=self.template,
|
||||
data='1',
|
||||
)
|
||||
|
||||
def _parameter_url(self, pk=None):
|
||||
if pk:
|
||||
return reverse('api-parameter-detail', kwargs={'pk': pk})
|
||||
return reverse('api-parameter-list')
|
||||
|
||||
def test_list_parameters_no_role_returns_empty(self):
|
||||
"""A user with no roles cannot see parameters attached to a Part."""
|
||||
response = self.get(
|
||||
self._parameter_url(),
|
||||
data={'model_type': 'part', 'model_id': self.part.pk},
|
||||
expected_code=200,
|
||||
)
|
||||
self.assertEqual(len(response.data), 0)
|
||||
|
||||
def test_list_parameters_with_view_role_returns_parameters(self):
|
||||
"""A user with part.view can see parameters attached to a Part."""
|
||||
self.assignRole('part.view')
|
||||
response = self.get(
|
||||
self._parameter_url(),
|
||||
data={'model_type': 'part', 'model_id': self.part.pk},
|
||||
expected_code=200,
|
||||
)
|
||||
pks = [p['pk'] for p in response.data]
|
||||
self.assertIn(self.parameter.pk, pks)
|
||||
|
||||
def test_detail_parameter_no_role_returns_404(self):
|
||||
"""A user with no roles gets 404 for a parameter attached to a Part."""
|
||||
self.get(self._parameter_url(self.parameter.pk), expected_code=404)
|
||||
|
||||
def test_delete_parameter_view_only_role_is_denied(self):
|
||||
"""A user with only 'view' permission must not be able to delete a parameter.
|
||||
|
||||
Regression test: ParameterDetail (RetrieveUpdateDestroyAPI) used DRF's
|
||||
default destroy()/perform_destroy(), which calls instance.delete()
|
||||
directly - bypassing any permission check entirely. A 'view'-only user
|
||||
(visible via get_queryset(), but without 'delete') must not be able to
|
||||
delete the parameter.
|
||||
"""
|
||||
self.assignRole('part.view')
|
||||
self.delete(self._parameter_url(self.parameter.pk), expected_code=403)
|
||||
self.assertTrue(
|
||||
common.models.Parameter.objects.filter(pk=self.parameter.pk).exists()
|
||||
)
|
||||
|
||||
def test_delete_parameter_with_delete_role_is_allowed(self):
|
||||
"""A user with part.delete can delete a parameter for a Part."""
|
||||
self.assignRole('part.delete')
|
||||
self.delete(self._parameter_url(self.parameter.pk), expected_code=204)
|
||||
self.assertFalse(
|
||||
common.models.Parameter.objects.filter(pk=self.parameter.pk).exists()
|
||||
)
|
||||
|
||||
def test_bulk_delete_parameter_view_only_role_is_denied(self):
|
||||
"""A user with only 'view' permission must not be able to bulk-delete parameters.
|
||||
|
||||
Regression test: ParameterList (BulkDeleteMixin) had no validate_delete()
|
||||
override, so the bulk-delete endpoint bypassed permission checks entirely.
|
||||
"""
|
||||
self.assignRole('part.view')
|
||||
self.delete(
|
||||
self._parameter_url(),
|
||||
data={'items': [self.parameter.pk]},
|
||||
expected_code=400,
|
||||
)
|
||||
self.assertTrue(
|
||||
common.models.Parameter.objects.filter(pk=self.parameter.pk).exists()
|
||||
)
|
||||
|
||||
def test_bulk_delete_parameter_with_delete_role_is_allowed(self):
|
||||
"""A user with part.delete can bulk-delete parameters for a Part."""
|
||||
self.assignRole('part.delete')
|
||||
self.delete(
|
||||
self._parameter_url(),
|
||||
data={'items': [self.parameter.pk]},
|
||||
expected_code=200,
|
||||
)
|
||||
self.assertFalse(
|
||||
common.models.Parameter.objects.filter(pk=self.parameter.pk).exists()
|
||||
)
|
||||
|
||||
|
||||
class NotesImagePermissionAPITests(InvenTreeAPITestCase):
|
||||
"""Tests for NotesImage API permission enforcement.
|
||||
|
||||
Regression coverage for a permission gap: NotesImageList had no queryset
|
||||
scoping and no permission check in perform_create(), so a user with no
|
||||
(or insufficient) permission against the model a note was attached to
|
||||
could still see, and attach images to, that note.
|
||||
"""
|
||||
|
||||
# No roles by default - each test assigns only what it needs
|
||||
roles = []
|
||||
|
||||
def setUp(self):
|
||||
"""Create a Part and a pre-existing Note to attach images to."""
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
||||
from part.models import Part
|
||||
|
||||
super().setUp()
|
||||
|
||||
self.part = Part.objects.create(
|
||||
name='Perm Test Part', description='Part for permission testing'
|
||||
)
|
||||
ct = ContentType.objects.get_for_model(Part)
|
||||
self.note = common.models.Note.objects.create(
|
||||
model_type=ct, model_id=self.part.pk, title='N', content='<p>c</p>'
|
||||
)
|
||||
self.image = common.models.NotesImage.objects.create(note=self.note)
|
||||
self.image.image.save('a.png', ContentFile(self._image_bytes()))
|
||||
|
||||
def _image_bytes(self):
|
||||
buf = io.BytesIO()
|
||||
Image.new('RGB', (16, 16), color='blue').save(buf, format='PNG')
|
||||
return buf.getvalue()
|
||||
|
||||
def _generate_upload(self, name='test.png'):
|
||||
buf = io.BytesIO()
|
||||
Image.new('RGB', (16, 16), color='blue').save(buf, format='PNG')
|
||||
buf.seek(0)
|
||||
return SimpleUploadedFile(name, buf.read(), content_type='image/png')
|
||||
|
||||
def test_list_images_no_role_returns_empty(self):
|
||||
"""A user with no roles cannot see images attached to the note."""
|
||||
response = self.get(reverse('api-notes-image-list'), expected_code=200)
|
||||
pks = [i['pk'] for i in response.data]
|
||||
self.assertNotIn(self.image.pk, pks)
|
||||
|
||||
def test_list_images_with_view_role_returns_images(self):
|
||||
"""A user with part.view can see images attached to the note."""
|
||||
self.assignRole('part.view')
|
||||
response = self.get(reverse('api-notes-image-list'), expected_code=200)
|
||||
pks = [i['pk'] for i in response.data]
|
||||
self.assertIn(self.image.pk, pks)
|
||||
|
||||
def test_upload_no_role_is_denied(self):
|
||||
"""A user with no roles cannot attach an image to the note."""
|
||||
self.post(
|
||||
reverse('api-notes-image-list'),
|
||||
data={'image': self._generate_upload(), 'note': self.note.pk},
|
||||
format='multipart',
|
||||
expected_code=403,
|
||||
)
|
||||
|
||||
def test_upload_view_only_role_is_denied(self):
|
||||
"""A user with only part.view cannot attach an image to the note.
|
||||
|
||||
'view' does not imply 'change' in the InvenTree ruleset hierarchy.
|
||||
"""
|
||||
self.assignRole('part.view')
|
||||
self.post(
|
||||
reverse('api-notes-image-list'),
|
||||
data={'image': self._generate_upload(), 'note': self.note.pk},
|
||||
format='multipart',
|
||||
expected_code=403,
|
||||
)
|
||||
|
||||
def test_upload_with_change_role_is_allowed(self):
|
||||
"""A user with part.change can attach an image to the note."""
|
||||
self.assignRole('part.change')
|
||||
self.post(
|
||||
reverse('api-notes-image-list'),
|
||||
data={'image': self._generate_upload(), 'note': self.note.pk},
|
||||
format='multipart',
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
|
||||
class GenericMetadataAuthorizationTests(InvenTreeAPITestCase):
|
||||
"""Tests for the generic '/api/metadata/<model>/pk/<pk>/' endpoint.
|
||||
|
||||
Regression coverage for a permission gap: several models (Attachment,
|
||||
Parameter, Note, ProjectCode) have no RuleSet permissions of their own
|
||||
(see users.ruleset.get_ruleset_ignore()) - access to them is meant to be
|
||||
scoped by a different rule instead (typically the RuleSet permission of
|
||||
another model they're linked to). ContentTypePermission previously
|
||||
treated that ignore-listing as blanket permission-exemption for *any*
|
||||
request against the object's own (ignore-listed) model, regardless of
|
||||
what its dedicated endpoint enforces - so this generic, catch-all
|
||||
metadata endpoint could read/write metadata on any Attachment/Parameter/
|
||||
Note/ProjectCode, bypassing whatever permission its own endpoint required.
|
||||
"""
|
||||
|
||||
roles = []
|
||||
|
||||
def setUp(self):
|
||||
"""Create a Part plus one linked Attachment, Parameter and Note."""
|
||||
from part.models import Part
|
||||
|
||||
super().setUp()
|
||||
|
||||
self.part = Part.objects.create(name='Metadata Perm Test Part', description='x')
|
||||
|
||||
self.attachment = common.models.Attachment.objects.create(
|
||||
model_type='part', model_id=self.part.pk, link='https://example.com'
|
||||
)
|
||||
|
||||
template = common.models.ParameterTemplate.objects.create(
|
||||
name='Metadata Perm Test Template', model_type=self.part.get_content_type()
|
||||
)
|
||||
self.parameter = common.models.Parameter.objects.create(
|
||||
model_type=self.part.get_content_type(),
|
||||
model_id=self.part.pk,
|
||||
template=template,
|
||||
data='1',
|
||||
)
|
||||
|
||||
self.note = common.models.Note.objects.create(
|
||||
model_type=self.part.get_content_type(),
|
||||
model_id=self.part.pk,
|
||||
title='N',
|
||||
content='<p>c</p>',
|
||||
)
|
||||
|
||||
self.project_code = common.models.ProjectCode.objects.create(code='PERM-001')
|
||||
|
||||
def _metadata_url(self, model, pk):
|
||||
return reverse('api-generic-metadata', kwargs={'model': model, 'pk': pk})
|
||||
|
||||
def test_attachment_metadata_requires_linked_permission(self):
|
||||
"""Attachment metadata is scoped by 'view'/'change' on the linked Part."""
|
||||
url = self._metadata_url('attachment', self.attachment.pk)
|
||||
|
||||
self.get(url, expected_code=403)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.view')
|
||||
self.get(url, expected_code=200)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.change')
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=200)
|
||||
|
||||
def test_parameter_metadata_requires_linked_permission(self):
|
||||
"""Parameter metadata is scoped by 'view'/'change' on the linked Part."""
|
||||
url = self._metadata_url('parameter', self.parameter.pk)
|
||||
|
||||
self.get(url, expected_code=403)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.view')
|
||||
self.get(url, expected_code=200)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.change')
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=200)
|
||||
|
||||
def test_note_metadata_requires_linked_permission(self):
|
||||
"""Note metadata is scoped by 'view'/'change' on the linked Part."""
|
||||
url = self._metadata_url('note', self.note.pk)
|
||||
|
||||
self.get(url, expected_code=403)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.view')
|
||||
self.get(url, expected_code=200)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.assignRole('part.change')
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=200)
|
||||
|
||||
def test_project_code_metadata_requires_staff_for_write(self):
|
||||
"""ProjectCode metadata mirrors the staff-only write restriction of its own endpoint."""
|
||||
url = self._metadata_url('projectcode', self.project_code.pk)
|
||||
|
||||
# Reading metadata does not require staff (matches IsStaffOrReadOnlyScope)
|
||||
self.user.is_staff = False
|
||||
self.user.save()
|
||||
self.get(url, expected_code=200)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
self.user.is_staff = True
|
||||
self.user.save()
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=200)
|
||||
|
||||
@@ -496,6 +496,15 @@ class UserProfile(InvenTree.models.MetadataMixin):
|
||||
user = models.OneToOneField(
|
||||
User, on_delete=models.CASCADE, related_name='profile', verbose_name=_('User')
|
||||
)
|
||||
|
||||
def check_permission(self, permission, user):
|
||||
"""Check if the user has the required permission for this profile.
|
||||
|
||||
UserProfile has no RuleSet permissions of its own,
|
||||
so we manually check if the user is the owner of this profile.
|
||||
"""
|
||||
return self.user_id == user.pk
|
||||
|
||||
language = models.CharField(
|
||||
max_length=10,
|
||||
blank=True,
|
||||
|
||||
@@ -500,3 +500,34 @@ class UserProfileTest(InvenTreeAPITestCase):
|
||||
# Ensure primary_group is set to None
|
||||
profile.refresh_from_db()
|
||||
self.assertIsNone(profile.primary_group)
|
||||
|
||||
|
||||
class UserProfileMetadataPermissionTests(InvenTreeAPITestCase):
|
||||
"""Tests for the generic metadata endpoint against the UserProfile model."""
|
||||
|
||||
def setUp(self):
|
||||
"""Create a second user with their own profile."""
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
super().setUp()
|
||||
|
||||
self.other_user = get_user_model().objects.create_user(
|
||||
username='other_metadata_user', password='password'
|
||||
)
|
||||
|
||||
def _metadata_url(self, pk):
|
||||
return reverse(
|
||||
'api-generic-metadata', kwargs={'model': 'userprofile', 'pk': pk}
|
||||
)
|
||||
|
||||
def test_own_profile_metadata_is_accessible(self):
|
||||
"""A user can read/write their own profile metadata via the generic endpoint."""
|
||||
url = self._metadata_url(self.user.profile.pk)
|
||||
self.get(url, expected_code=200)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=200)
|
||||
|
||||
def test_other_users_profile_metadata_is_denied(self):
|
||||
"""A user cannot read/write another user's profile metadata via the generic endpoint."""
|
||||
url = self._metadata_url(self.other_user.profile.pk)
|
||||
self.get(url, expected_code=403)
|
||||
self.patch(url, {'metadata': {'x': 1}}, expected_code=403)
|
||||
|
||||
Reference in New Issue
Block a user