moderation: add self-action prevention

* The problem is that an admin could block his own account. With this change it is possible to prevent the admin from doing that.
* Prevent self-action for: block, deactivate, restore, activate and approve.
* Update tests for self-action prevention
* introduce PreventSelf generator
* introduce _check_manage_permissions in users service
* add "PreventSelf" into "can_manage" permission

Author: Samk13

invenio_users_resources/services/generators.py

@@ -3,6 +3,7 @@
 # Copyright (C) 2022 TU Wien.
 # Copyright (C) 2022 CERN.
 # Copyright (C) 2023 Graz University of Technology.
+# Copyright (C) 2024 KTH Royal Institute of Technology.
 #
 # Invenio-Users-Resources is free software; you can redistribute it and/or
 # modify it under the terms of the MIT License; see LICENSE file for more
@@ -94,6 +95,18 @@ def query_filter(self, identity=None, **kwargs):
         return []
 
 
+class PreventSelf(Generator):
+    """Prevents users from performing actions on themselves."""
+
+    def excludes(self, record=None, actor_id=None, **kwargs):
+        """Exclude needs to prevent self-actions."""
+        if record is None or actor_id is None:
+            return []
+        if actor_id == record.id:
+            return [UserNeed(record.id)]
+        return []
+
+
 class IfGroupNotManaged(ConditionalGenerator):
     """Generator for managed group access."""
 

invenio_users_resources/services/permissions.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2022 TU Wien.
+# Copyright (C) 2024 KTH Royal Institute of Technology.
 #
 # Invenio-Users-Resources is free software; you can redistribute it and/or
 # modify it under the terms of the MIT License; see LICENSE file for more
@@ -23,6 +24,7 @@
     IfGroupNotManaged,
     IfPublicEmail,
     IfPublicUser,
+    PreventSelf,
     Self,
 )
 
@@ -51,10 +53,10 @@ class UsersPermissionPolicy(BasePermissionPolicy):
     can_read_all = [UserManager, SystemProcess()]
 
     # Moderation permissions
-    can_manage = [UserManager, SystemProcess()]
+    can_manage = [UserManager, PreventSelf(), SystemProcess()]
     can_search_all = [UserManager, SystemProcess()]
     can_read_system_details = [UserManager, SystemProcess()]
-    can_impersonate = [UserManager, SystemProcess()]
+    can_impersonate = [UserManager, PreventSelf(), SystemProcess()]
 
 
 class GroupsPermissionPolicy(BasePermissionPolicy):

invenio_users_resources/services/users/service.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2022 KTH Royal Institute of Technology
+# Copyright (C) 2022-2024 KTH Royal Institute of Technology.
 # Copyright (C) 2022 TU Wien.
 # Copyright (C) 2022 European Union.
 # Copyright (C) 2022 CERN.
@@ -129,15 +129,20 @@ def rebuild_index(self, identity, uow=None):
         self.indexer.bulk_index([u.id for u in users])
         return True
 
+    def _check_permission(self, identity, permission_type, user):
+        """Checks if given identity has the specified permission type on the user."""
+        self.require_permission(
+            identity, permission_type, record=user, actor_id=identity.id
+        )
+
     @unit_of_work()
     def block(self, identity, id_, uow=None):
         """Blocks a user."""
         user = UserAggregate.get_record(id_)
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-
-        self.require_permission(identity, "manage", record=user)
+        self._check_permission(identity, "manage", user)
 
         if user.blocked:
             raise ValidationError("User is already blocked.")
@@ -160,8 +165,7 @@ def restore(self, identity, id_, uow=None):
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-
-        self.require_permission(identity, "manage", record=user)
+        self._check_permission(identity, "manage", user)
 
         if not user.blocked:
             raise ValidationError("User is not blocked.")
@@ -185,8 +189,7 @@ def approve(self, identity, id_, uow=None):
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-
-        self.require_permission(identity, "manage", record=user)
+        self._check_permission(identity, "manage", user)
 
         if user.verified:
             raise ValidationError("User is already verified.")
@@ -209,7 +212,7 @@ def deactivate(self, identity, id_, uow=None):
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-        self.require_permission(identity, "manage", record=user)
+        self._check_permission(identity, "manage", user)
 
         if not user.active:
             raise ValidationError("User is already inactive.")
@@ -225,7 +228,8 @@ def activate(self, identity, id_, uow=None):
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-        self.require_permission(identity, "manage", record=user)
+        self._check_permission(identity, "manage", user)
+
         if user.active and user.confirmed:
             raise ValidationError("User is already active.")
         user.activate()
@@ -238,5 +242,6 @@ def can_impersonate(self, identity, id_):
         if user is None:
             # return 403 even on empty resource due to security implications
             raise PermissionDeniedError()
-        self.require_permission(identity, "impersonate", record=user)
+        self._check_permission(identity, "impersonate", user)
+
         return user.model.model_obj

tests/resources/test_resources_users.py

@@ -127,6 +127,10 @@ def test_approve_user(client, headers, user_pub, user_moderator, db):
     assert res.status_code == 200
     assert res.json["verified_at"] is not None
 
+    # Test user tries to approve themselves
+    res = client.post(f"/users/{user_moderator.id}/approve", headers=headers)
+    assert res.status_code == 403
+
 
 def test_block_user(client, headers, user_pub, user_moderator, db):
     """Tests block user endpoint."""
@@ -138,6 +142,13 @@ def test_block_user(client, headers, user_pub, user_moderator, db):
     assert res.status_code == 200
     assert res.json["blocked_at"] is not None
 
+    # Test user tries to block themselves
+    res = client.post(f"/users/{user_moderator.id}/block", headers=headers)
+    assert res.status_code == 403
+
+    res = client.get(f"/users/{user_moderator.id}")
+    assert res.status_code == 200
+
 
 def test_deactivate_user(client, headers, user_pub, user_moderator, db):
     """Tests deactivate user endpoint."""
@@ -149,6 +160,13 @@ def test_deactivate_user(client, headers, user_pub, user_moderator, db):
     assert res.status_code == 200
     assert res.json["active"] == False
 
+    # Test user tries to deactivate themselves
+    res = client.post(f"/users/{user_moderator.id}/deactivate", headers=headers)
+    assert res.status_code == 403
+
+    res = client.get(f"/users/{user_moderator.id}")
+    assert res.status_code == 200
+
 
 def test_management_permissions(client, headers, user_pub, db):
     """Test permissions at the resource level."""