mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-09-16 06:38:00 +00:00
Fix: enforce set_permissions shape via a nested DRF serializer (#14119)
SetPermissionsSerializer was a bare DictField, and the legacy bulk_edit set_permissions parameter bypassed even that by hand-calling validate_set_permissions() on an unchecked dict. A bool or a non-list in place of users/groups crashed with a raw TypeError instead of a 400. It is now a nested serializer (view/change, each with users/groups as lists of integers), used for owned-object create/update, the legacy bulk_edit set_permissions parameter, and bulk_edit_objects permissions. Unknown action keys are rejected: previously a typo like "veiw" was silently dropped, leaving an empty permission set that could clear existing grants. An explicit set_permissions null (an owner-only change) remains a no-op, and an empty bulk_edit_objects permissions dict is still rejected.
This commit is contained in:
@@ -177,13 +177,10 @@ def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permis
|
||||
"""
|
||||
Resolves `codenames` to Permission rows, raising like the single-object
|
||||
assign_perm() this bulk path replaces does (via a `.get()` internally)
|
||||
if any codename doesn't exist -- e.g. a client-supplied action name that
|
||||
was never validated (BulkEditObjectsSerializer._validate_permissions
|
||||
calls validate_set_permissions() only for its side-effecting id checks
|
||||
and discards the filtered dict it returns, so an unrecognized action key
|
||||
reaches this function as-is). A plain `.filter()` with no existence
|
||||
check would otherwise silently build zero rows and no-op instead of
|
||||
reporting the bad input.
|
||||
if any codename doesn't exist. SetPermissionsSerializer rejects unknown
|
||||
action names at the API, but a caller passing one directly would
|
||||
otherwise get a plain `.filter()` that silently builds zero rows and
|
||||
no-ops instead of reporting the bad input.
|
||||
"""
|
||||
permission_objs = list(
|
||||
Permission.objects.filter(content_type=ctype, codename__in=codenames),
|
||||
@@ -298,10 +295,9 @@ def set_permissions_for_objects(
|
||||
|
||||
# Every action is resolved up front, before anything is written, so an
|
||||
# unrecognized action name (see _resolve_permissions) aborts the whole
|
||||
# call instead of leaving the actions ahead of it already applied --
|
||||
# BulkEditObjectsSerializer lets unknown keys through and its view turns
|
||||
# the exception into a 400, so a half-applied change would otherwise be
|
||||
# reported to the client as a failure.
|
||||
# call instead of leaving the actions ahead of it already applied.
|
||||
# SetPermissionsSerializer rejects unknown actions at the API, so this
|
||||
# guards any other caller.
|
||||
permissions_by_action: dict[str, list[Permission]] = {}
|
||||
for action, entry in permissions.items():
|
||||
if "users" not in entry and "groups" not in entry:
|
||||
|
||||
@@ -210,6 +210,9 @@ class MatchingModelSerializer(serializers.ModelSerializer[Any]):
|
||||
return match
|
||||
|
||||
|
||||
PERMISSION_ACTIONS = ("view", "change")
|
||||
|
||||
|
||||
class SetPermissionsMixin:
|
||||
def _validate_user_ids(self, user_ids):
|
||||
users = User.objects.none()
|
||||
@@ -232,12 +235,9 @@ class SetPermissionsMixin:
|
||||
return groups
|
||||
|
||||
def validate_set_permissions(self, set_permissions=None):
|
||||
permissions_dict = {
|
||||
"view": {},
|
||||
"change": {},
|
||||
}
|
||||
permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
|
||||
if set_permissions is not None:
|
||||
for action in ["view", "change"]:
|
||||
for action in PERMISSION_ACTIONS:
|
||||
if action in set_permissions:
|
||||
if "users" in set_permissions[action]:
|
||||
users = set_permissions[action]["users"]
|
||||
@@ -265,41 +265,31 @@ class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
@extend_schema_field(
|
||||
field={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"view": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"users": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
"groups": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
},
|
||||
},
|
||||
"change": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"users": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
"groups": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
class SetPermissionsSerializer(serializers.DictField):
|
||||
pass
|
||||
class PermissionSetSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
users = serializers.ListField(
|
||||
child=serializers.IntegerField(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
groups = serializers.ListField(
|
||||
child=serializers.IntegerField(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
)
|
||||
|
||||
|
||||
class SetPermissionsSerializer(serializers.Serializer[dict[str, Any]]):
|
||||
view = PermissionSetSerializer(required=False)
|
||||
change = PermissionSetSerializer(required=False)
|
||||
|
||||
def to_internal_value(self, data):
|
||||
if isinstance(data, dict):
|
||||
unknown_keys = set(data) - set(PERMISSION_ACTIONS)
|
||||
if unknown_keys:
|
||||
raise serializers.ValidationError(
|
||||
{key: "Unknown permission action." for key in sorted(unknown_keys)},
|
||||
)
|
||||
return super().to_internal_value(data)
|
||||
|
||||
|
||||
class OwnedObjectSerializer(
|
||||
@@ -470,7 +460,6 @@ class OwnedObjectSerializer(
|
||||
|
||||
set_permissions = SetPermissionsSerializer(
|
||||
label="Set permissions",
|
||||
allow_empty=True,
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
@@ -2051,8 +2040,13 @@ class BulkEditSerializer(
|
||||
def _validate_parameters_set_permissions(self, parameters) -> None:
|
||||
if "set_permissions" not in parameters:
|
||||
raise serializers.ValidationError("set_permissions not specified")
|
||||
set_permissions = parameters["set_permissions"]
|
||||
if set_permissions is not None:
|
||||
set_permissions = SetPermissionsSerializer().run_validation(
|
||||
set_permissions,
|
||||
)
|
||||
parameters["set_permissions"] = self.validate_set_permissions(
|
||||
parameters["set_permissions"],
|
||||
set_permissions,
|
||||
)
|
||||
if "owner" in parameters and parameters["owner"] is not None:
|
||||
self._validate_owner(parameters["owner"])
|
||||
@@ -3011,9 +3005,8 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
allow_null=True,
|
||||
)
|
||||
|
||||
permissions = serializers.DictField(
|
||||
permissions = SetPermissionsSerializer(
|
||||
label="Set permissions",
|
||||
allow_empty=False,
|
||||
required=False,
|
||||
write_only=True,
|
||||
)
|
||||
@@ -3049,8 +3042,8 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
)
|
||||
return objects
|
||||
|
||||
def _validate_permissions(self, permissions) -> None:
|
||||
self.validate_set_permissions(
|
||||
def _validate_permissions(self, permissions) -> dict:
|
||||
return self.validate_set_permissions(
|
||||
permissions,
|
||||
)
|
||||
|
||||
@@ -3074,7 +3067,11 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
if operation == "set_permissions":
|
||||
permissions = attrs.get("permissions")
|
||||
if permissions is not None:
|
||||
self._validate_permissions(permissions)
|
||||
if not permissions:
|
||||
raise serializers.ValidationError(
|
||||
"permissions must not be empty",
|
||||
)
|
||||
attrs["permissions"] = self._validate_permissions(permissions)
|
||||
|
||||
return attrs
|
||||
|
||||
|
||||
@@ -1165,6 +1165,77 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
|
||||
self.assertIn(b"set_permissions not specified", response.content)
|
||||
m.assert_not_called()
|
||||
|
||||
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
|
||||
def test_set_permissions_rejects_malformed_set_permissions(self, m) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A set_permissions bulk edit where set_permissions is not an
|
||||
object, has a non-list users value, or has an unknown action
|
||||
WHEN:
|
||||
- API to bulk edit is called
|
||||
THEN:
|
||||
- API returns HTTP 400 describing the problem
|
||||
- set_permissions is not called
|
||||
"""
|
||||
self.setup_mock(m, "set_permissions")
|
||||
|
||||
for set_permissions, expected_message in (
|
||||
(False, b"Expected a dictionary"),
|
||||
({"view": {"users": False}}, b"Expected a list"),
|
||||
({"not_a_real_action": {"users": [1]}}, b"Unknown permission action"),
|
||||
):
|
||||
with self.subTest(set_permissions=set_permissions):
|
||||
response = self.client.post(
|
||||
"/api/documents/bulk_edit/",
|
||||
json.dumps(
|
||||
{
|
||||
"documents": [self.doc2.id],
|
||||
"method": "set_permissions",
|
||||
"parameters": {"set_permissions": set_permissions},
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn(expected_message, response.content)
|
||||
m.assert_not_called()
|
||||
|
||||
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
|
||||
def test_set_permissions_null_is_a_noop(self, m) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- A set_permissions bulk edit with set_permissions null and an
|
||||
owner, i.e. a request that only changes the owner
|
||||
WHEN:
|
||||
- API to bulk edit is called
|
||||
THEN:
|
||||
- Request succeeds
|
||||
- set_permissions receives no users or groups for any action
|
||||
"""
|
||||
self.setup_mock(m, "set_permissions")
|
||||
|
||||
response = self.client.post(
|
||||
"/api/documents/bulk_edit/",
|
||||
json.dumps(
|
||||
{
|
||||
"documents": [self.doc2.id],
|
||||
"method": "set_permissions",
|
||||
"parameters": {
|
||||
"set_permissions": None,
|
||||
"owner": self.user.id,
|
||||
},
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
m.assert_called_once()
|
||||
self.assertEqual(
|
||||
m.call_args.kwargs["set_permissions"],
|
||||
{"view": {}, "change": {}},
|
||||
)
|
||||
|
||||
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
|
||||
def test_set_permissions_merge(self, m) -> None:
|
||||
self.setup_mock(m, "set_permissions")
|
||||
@@ -1300,7 +1371,7 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
|
||||
self.client.force_authenticate(user=user1)
|
||||
|
||||
permissions = {
|
||||
"owner": user1.id,
|
||||
"view": {"users": [user1.id]},
|
||||
}
|
||||
|
||||
response = self.client.post(
|
||||
|
||||
@@ -1499,6 +1499,110 @@ class TestBulkEditObjectPermissions(APITestCase):
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
self.assertTrue(Tag.objects.filter(pk=self.t1.id).exists())
|
||||
|
||||
def test_bulk_object_set_permissions_rejects_empty_permissions(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Existing objects
|
||||
WHEN:
|
||||
- bulk_edit_objects API endpoint is called with set_permissions
|
||||
operation and an empty permissions dict
|
||||
THEN:
|
||||
- Validation fails rather than silently applying a no-op
|
||||
"""
|
||||
response = self.client.post(
|
||||
"/api/bulk_edit_objects/",
|
||||
json.dumps(
|
||||
{
|
||||
"objects": [self.t1.id],
|
||||
"object_type": "tags",
|
||||
"operation": "set_permissions",
|
||||
"permissions": {},
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_bulk_object_set_permissions_rejects_non_dict_permissions(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Existing objects
|
||||
WHEN:
|
||||
- bulk_edit_objects API endpoint is called with set_permissions
|
||||
operation and a non-dict permissions value
|
||||
THEN:
|
||||
- Validation fails rather than crashing
|
||||
"""
|
||||
response = self.client.post(
|
||||
"/api/bulk_edit_objects/",
|
||||
json.dumps(
|
||||
{
|
||||
"objects": [self.t1.id],
|
||||
"object_type": "tags",
|
||||
"operation": "set_permissions",
|
||||
"permissions": False,
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_bulk_object_set_permissions_rejects_unknown_action(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- Existing objects
|
||||
WHEN:
|
||||
- bulk_edit_objects API endpoint is called with set_permissions
|
||||
operation and an unrecognized permission action name
|
||||
THEN:
|
||||
- Validation fails rather than silently no-oping
|
||||
"""
|
||||
response = self.client.post(
|
||||
"/api/bulk_edit_objects/",
|
||||
json.dumps(
|
||||
{
|
||||
"objects": [self.t1.id],
|
||||
"object_type": "tags",
|
||||
"operation": "set_permissions",
|
||||
"permissions": {"not_a_real_action": {"users": [self.user1.id]}},
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_bulk_object_set_permissions_null_users_clears_users(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- An object a user has view permission on
|
||||
WHEN:
|
||||
- bulk_edit_objects API endpoint is called with set_permissions
|
||||
operation, merge off, and an explicit null for the view users
|
||||
THEN:
|
||||
- Request succeeds and null is treated as an empty user list,
|
||||
so the existing view permission is removed
|
||||
"""
|
||||
assign_perm("view_tag", self.user1, self.t1)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/bulk_edit_objects/",
|
||||
json.dumps(
|
||||
{
|
||||
"objects": [self.t1.id],
|
||||
"object_type": "tags",
|
||||
"operation": "set_permissions",
|
||||
"permissions": {"view": {"users": None}},
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertNotIn(self.user1, get_users_with_perms(self.t1))
|
||||
|
||||
def test_bulk_edit_object_permissions_validation(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
|
||||
@@ -636,14 +636,9 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
THEN:
|
||||
- Permission.DoesNotExist is raised, not a silent no-op
|
||||
|
||||
Regression test: the endpoint that calls this
|
||||
(BulkEditObjectPermissionsView) never actually validates action
|
||||
names against the raw client-supplied permissions dict --
|
||||
BulkEditObjectsSerializer._validate_permissions calls
|
||||
validate_set_permissions() only for its side-effecting user/group id
|
||||
checks and discards the filtered dict it returns -- so a bogus
|
||||
action key reaches this function as-is. Resolving the Permission via
|
||||
a bare `.filter()` (which returns empty instead of raising) would
|
||||
The API rejects unknown action names before they get here, but a
|
||||
direct caller could still pass one. Resolving the Permission via a
|
||||
bare `.filter()` (which returns empty instead of raising) would
|
||||
silently drop the grant and report success.
|
||||
"""
|
||||
with self.assertRaises(Permission.DoesNotExist):
|
||||
|
||||
Reference in New Issue
Block a user