fix(roles): block escalation to protected roles

* Validate role updates at the API level
* Prevent regular roles from becoming protected roles
* Enforce protected group permissions consistently
* Update group identifier references
* Add tests for protected roles and super admin access
* groups pagination for late-alphabet roles
  raise groups default_max_results to 50 to allow deeper paging
  update pagination test expectations to match real page/size handling
* fix(schemas): set proper dump only
  Neither ID nor domain should be allowed to be set from the API since
  they are automatically assigned either by the database or the email
  property setter.

Author: Samk13

invenio_users_resources/config.py

@@ -271,12 +271,19 @@
 """Invenio groups admin search configuration."""
 
 USERS_RESOURCES_PROTECTED_GROUP_NAMES = [
-    "superuser-access",
     "admin",
     "administration",
+    "superuser-access",
     "administration-moderation",
 ]
-"""Group identifiers that cannot be mutated via the public API (system process only)."""
+"""Group identifiers that cannot be mutated via API (system process only).
+
+References:
+- superuser-access: https://github.com/inveniosoftware/invenio-access/blob/master/invenio_access/permissions.py
+- administration: https://github.com/inveniosoftware/invenio-administration/blob/master/invenio_administration/permissions.py
+- admin: https://github.com/inveniosoftware/invenio-cli/blob/master/invenio_cli/commands/services.py (created during instance setup)
+- administration-moderation: https://github.com/inveniosoftware/invenio-users-resources/blob/master/invenio_users_resources/permissions.py
+"""
 
 
 class OrgPropsSchema(Schema):

invenio_users_resources/ext.py

@@ -3,6 +3,7 @@
 # Copyright (C) 2022 CERN.
 # Copyright (C) 2022 TU Wien.
 # Copyright (C) 2023-2024 Graz University of Technology.
+# Copyright (C) 2025 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

invenio_users_resources/records/api.py

@@ -27,7 +27,7 @@
 from invenio_records_resources.records.api import Record
 from invenio_records_resources.records.systemfields import IndexField
 from marshmallow import ValidationError
-from sqlalchemy import or_
+from sqlalchemy import select
 from sqlalchemy.exc import NoResultFound
 
 from .dumpers import EmailFieldDumperExt
@@ -142,29 +142,25 @@ def _validate_user_data(user_data):
         raise ValidationError(errors)
 
 
-def _validate_group_data(group_data):
-    """Validate data for group creation."""
-    group_id = group_data.get("id")
-    name = group_data.get("name")
-
-    if not (group_id or name):
+def _validate_group_data(data, exclude_id=None):
+    """Validate the group data."""
+    name = data.get("name")
+    if not name:
         return
 
-    role = current_datastore.role_model
-
-    stmt = db.select(role).where(or_(role.id == group_id, role.name == name)).limit(1)
+    errors = {}
+    stmt = select(current_datastore.role_model.id).where(
+        current_datastore.role_model.name == name
+    )
+    if exclude_id is not None:
+        stmt = stmt.where(current_datastore.role_model.id != exclude_id)
 
-    row = db.session.execute(stmt).scalars().first()
-    if not row:
-        return
+    existing_role_id = db.session.execute(stmt).scalar_one_or_none()
 
-    errors = {}
-    if group_id and row.id == group_id:
-        errors["id"] = [_("Role id already used by another group.")]
-    if name and row.name == name:
+    if existing_role_id:
         errors["name"] = [_("Role name already used by another group.")]
-
-    raise ValidationError(errors)
+    if errors:
+        raise ValidationError(errors)
 
 
 class UserAggregate(BaseAggregate):
@@ -270,9 +266,6 @@ def avatar_color(self):
     def create(cls, data, id_=None, validator=None, format_checker=None, **kwargs):
         """Create a new User and store it in the database."""
         try:
-            #  Admin group view passes in an empty string as id, which will be a valid Role id.
-            if "id" in data and data["id"] == "":
-                data.pop("id")
             # Check if email and  username already exists by another account.
             _validate_user_data(data)
             # Create User
@@ -437,6 +430,9 @@ def update(self, data, id_):
         if role is None:
             role = db.session.get(current_datastore.role_model, id_)
 
+        if "name" in data:
+            _validate_group_data({"name": data["name"]}, exclude_id=role.id)
+
         role.name = data.get("name", role.name)
         role.description = data.get("description", role.description)
         role = current_datastore.update_role(role)

invenio_users_resources/resources/groups/config.py

@@ -35,7 +35,7 @@ class GroupSearchRequestArgsSchema(SearchRequestArgsSchema):
 
 def handle_group_validation_error(err):
     """Handle groups errors."""
-    marshmallow_error = ValidationError(err.errors or {})
+    marshmallow_error = ValidationError(err.errors)
     response = HTTPJSONException(
         code=400,
         description=err.description,

invenio_users_resources/resources/groups/errors.py

@@ -23,6 +23,6 @@ class GroupValidationError(GroupsException):
 
     def __init__(self, errors):
         """Marshmallow validation errors."""
-        self.errors = errors or {}
+        self.errors = errors
         self.description = self.message
         super().__init__(self.message)

invenio_users_resources/resources/groups/resource.py

@@ -33,8 +33,8 @@ def create_url_rules(self):
         routes = self.config.routes
         return [
             route("GET", routes["list"], self.search),
-            route("POST", routes["list"], self.create),
             route("GET", routes["item"], self.read),
+            route("POST", routes["list"], self.create),
             route("PUT", routes["item"], self.update),
             route("DELETE", routes["item"], self.delete),
             route("GET", routes["item-avatar"], self.avatar),

invenio_users_resources/services/generators.py

@@ -3,7 +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.
+# Copyright (C) 2024-2025 KTH Royal Institute of Technology.
 # Copyright (C) 2025 Ubiquity Press.
 #
 # Invenio-Users-Resources is free software; you can redistribute it and/or
@@ -186,19 +186,28 @@ def excludes(self, **kwargs):
 class ProtectedGroupIdentifiers(Generator):
     """Exclude access to protected groups unless system process."""
 
-    def _is_protected(self, record):
-        protected = current_app.config.get("USERS_RESOURCES_PROTECTED_GROUP_NAMES", [])
+    def _is_protected(self, record, **kwargs):
+        """Check if the role is a protected."""
+        protected = set(
+            current_app.config.get("USERS_RESOURCES_PROTECTED_GROUP_NAMES", [])
+        )
         if not protected or record is None:
             return False
 
-        values = [getattr(record, "id", None), getattr(record, "name", None)]
+        # create new role
+        name_dict = record.get("name") if isinstance(record, dict) else None
+        # update protected role (object form)
+        name_obj = getattr(record, "name", None)
+        id_obj = getattr(record, "id", None)
+        # update role to protected
+        update_name = kwargs.get("update_grp", {}).get("name")
 
-        candidates = {str(val) for val in values if val is not None}
-        return bool(set(protected).intersection(candidates))
+        candidates = {str(v) for v in (name_dict, name_obj, id_obj, update_name) if v}
+        return bool(protected & candidates)
 
     def excludes(self, record=None, identity=None, **kwargs):
         """Exclude non-system identities from protected groups."""
-        if not self._is_protected(record):
+        if not self._is_protected(record, **kwargs):
             return []
         if identity and system_process in identity.provides:
             return []

invenio_users_resources/services/groups/config.py

@@ -21,14 +21,14 @@
 from invenio_records_resources.services.base.config import ConfiguratorMixin
 from invenio_records_resources.services.records.params import (
     FacetsParam,
+    PaginationParam,
     QueryStrParam,
     SortParam,
 )
 from invenio_records_resources.services.records.queryparser import QueryParser
 
 from ...records.api import GroupAggregate
 from ..common import EndpointLinkWithId
-from ..params import FixedPagination
 from ..permissions import GroupsPermissionPolicy
 from ..schemas import GroupSchema
 from . import facets as groups_facets
@@ -39,8 +39,8 @@ class GroupSearchOptions(SearchOptions):
     """Search options."""
 
     pagination_options = {
-        "default_results_per_page": 10,
-        "default_max_results": 10,
+        "default_results_per_page": 20,
+        "default_max_results": 50,
     }
 
     query_parser_cls = QueryParser.factory(fields=["id", "name"])
@@ -77,7 +77,7 @@ class GroupSearchOptions(SearchOptions):
     params_interpreters_cls = [
         QueryStrParam,
         SortParam,
-        FixedPagination,
+        PaginationParam,
         FacetsParam,
     ]
 

invenio_users_resources/services/groups/service.py

@@ -15,7 +15,6 @@
 from flask import current_app
 from invenio_accounts.models import Role
 from invenio_db import db
-from invenio_i18n import gettext as _
 from invenio_records_resources.resources.errors import PermissionDeniedError
 from invenio_records_resources.services import RecordService
 from invenio_records_resources.services.uow import (
@@ -24,7 +23,6 @@
     unit_of_work,
 )
 from marshmallow import ValidationError
-from sqlalchemy.exc import IntegrityError
 
 from ...records.api import GroupAggregate
 from ...resources.groups.errors import GroupValidationError
@@ -48,12 +46,10 @@ def create(self, identity, data, raise_errors=True, uow=None):
             raise GroupValidationError(err.messages)
 
         try:
+            # Create group using API
             group = self.record_cls.create(data)
-        except IntegrityError as err:
-            db.session.rollback()
-            raise GroupValidationError(
-                {"name": [_("Role name already used by another group.")]}
-            ) from err
+        except ValidationError as err:
+            raise GroupValidationError(err.messages) from err
 
         current_app.logger.debug(
             "Group created: '%s' by user %s", group.name, identity.id
@@ -90,7 +86,7 @@ def update(self, identity, id_, data, raise_errors=True, uow=None):
         group = GroupAggregate.get_record(id_)
         if group is None:
             raise PermissionDeniedError()
-        self.require_permission(identity, "update", record=group)
+        self.require_permission(identity, "update", record=group, update_grp=data)
 
         try:
             data, errors = self.schema.load(
@@ -103,12 +99,10 @@ def update(self, identity, id_, data, raise_errors=True, uow=None):
             raise GroupValidationError(err.messages) from err
 
         try:
+            # Update group using API
             group = group.update(data, id_)
-        except IntegrityError as err:
-            db.session.rollback()
-            raise GroupValidationError(
-                {"name": [_("Role name already used by another group.")]}
-            ) from err
+        except ValidationError as err:
+            raise GroupValidationError(err.messages) from err
         current_app.logger.debug(
             "Group updated: '%s' by user %s", group.name, identity.id
         )

invenio_users_resources/services/schemas.py

@@ -105,6 +105,7 @@ class UserSchema(BaseRecordSchema, FieldPermissionsMixin):
     }
 
     # NOTE: API should only deliver users that are active & confirmed
+    id = fields.Str(dump_only=True)
     active = fields.Boolean()
     confirmed = fields.Boolean(dump_only=True)
     blocked = fields.Boolean(dump_only=True)
@@ -114,7 +115,7 @@ class UserSchema(BaseRecordSchema, FieldPermissionsMixin):
     is_current_user = fields.Method("is_self", dump_only=True)
 
     email = fields.Email(required=True)
-    domain = fields.String()
+    domain = fields.String(dump_only=True)
     domaininfo = fields.Nested(DomainInfoSchema)
     identities = fields.Nested(IdentitiesSchema, dump_default={})
     username = fields.String(validate=validate_username)
@@ -144,6 +145,7 @@ def is_self(self, obj):
 class GroupSchema(BaseRecordSchema):
     """Schema for user groups."""
 
+    id = fields.Str(dump_only=True)
     name = fields.String(
         required=True,
         validate=[