services: Allow only admin to bypass feature flag

invenio_rdm_records/config.py

@@ -134,7 +134,7 @@ def always_valid(identifier):
 #
 # Record communities
 #
-RDM_RECORD_ALWAYS_IN_COMMUNITY = False
+RDM_COMMUNITY_REQUIRED_TO_PUBLISH = False
 """Enforces at least one community per record."""
 
 #

invenio_rdm_records/resources/config.py

@@ -36,7 +36,7 @@
 
 from ..services.errors import (
     AccessRequestExistsError,
-    CommunityNotSelectedError,
+    CommunityRequiredError,
     GrantExistsError,
     InvalidAccessRestrictions,
     RecordDeletedException,
@@ -257,7 +257,7 @@ class RDMRecordResourceConfig(RecordResourceConfig, ConfiguratorMixin):
                 description=e.description,
             )
         ),
-        CommunityNotSelectedError: create_error_handler(
+        CommunityRequiredError: create_error_handler(
             HTTPJSONException(
                 code=400,
                 description="Cannot publish without selecting a community.",

invenio_rdm_records/services/communities/service.py

@@ -203,26 +203,20 @@ def _remove(self, identity, community_id, record):
         if community_id not in record.parent.communities.ids:
             raise RecordCommunityMissing(record.id, community_id)
 
-        # Check for permission to remove a community from a record
-        self.require_permission(
-            identity, "remove_community", record=record, community_id=community_id
-        )
-        is_community_required = current_app.config["RDM_RECORD_ALWAYS_IN_COMMUNITY"]
-        is_last_community = len(record.parent.communities.ids) == 1
-        # If community is required for a record and if it is the last community to remove
-        # Then, only users with special permissions can remove
-        can_remove_last_community = self.check_permission(
-            identity,
-            "remove_community_elevated",
-            record=record,
-            community_id=community_id,
-        )
-        if (
-            is_community_required
-            and is_last_community
-            and not can_remove_last_community
-        ):
-            raise CannotRemoveCommunityError()
+        try:
+            self.require_permission(
+                identity, "remove_community", record=record, community_id=community_id
+            )
+            # By default, admin/superuser has permission to do everything, so PermissionDeniedError won't be raised for admin in any case
+        except PermissionDeniedError as exc:
+            # If user doesn't have permission, then check if community is required
+            if current_app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"]:
+                # If community is required for a record and if it is the last community to remove, raise error
+                if len(record.parent.communities.ids) == 1:
+                    raise CannotRemoveCommunityError()
+            else:
+                # If the config wasn't enabled, then raise the PermissionDeniedError
+                raise exc
 
         # Default community is deleted when the exact same community is removed from the record
         record.parent.communities.remove(community_id)

invenio_rdm_records/services/errors.py

@@ -210,13 +210,13 @@ class RecordSubmissionClosedCommunityError(PermissionDenied):
     )
 
 
-class CommunityNotSelectedError(Exception):
+class CommunityRequiredError(Exception):
     """Error thrown when a record is being created/updated with less than 1 community."""
 
-    description = _("Cannot publish without selecting a community.")
+    description = _("Cannot publish without a community.")
 
 
 class CannotRemoveCommunityError(Exception):
     """Error thrown when the last community is being removed from the record."""
 
-    description = _("Cannot remove. A record should be part of at least 1 community.")
+    description = _("A record should be part of at least 1 community.")

invenio_rdm_records/services/generators.py

@@ -414,3 +414,25 @@ def needs(self, request=None, **kwargs):
             return [AccessRequestTokenNeed(request["payload"]["token"])]
 
         return []
+
+
+class IfOneCommunity(ConditionalGenerator):
+    """Conditional generator for records always in communities case."""
+
+    def _condition(self, record=None, **kwargs):
+        """Check if the record is associated with zero or one community."""
+        if record is None:
+            return True
+        rec_communities = record.parent.communities.ids
+        return len(rec_communities) == 1
+
+
+class IfAtLeastOneCommunity(ConditionalGenerator):
+    """Conditional generator for records always in communities case."""
+
+    def _condition(self, record=None, **kwargs):
+        """Check if the record is associated with zero or one community."""
+        if record is None:
+            return True
+        rec_communities = record.parent.communities.ids
+        return len(rec_communities) > 0

invenio_rdm_records/services/permissions.py

@@ -30,11 +30,13 @@
     AccessGrant,
     CommunityInclusionReviewers,
     GuestAccessRequestToken,
+    IfAtLeastOneCommunity,
     IfCreate,
     IfDeleted,
     IfExternalDOIRecord,
     IfFileIsLocal,
     IfNewRecord,
+    IfOneCommunity,
     IfRecordDeleted,
     IfRequestType,
     IfRestricted,
@@ -66,7 +68,6 @@ class RDMRecordPermissionPolicy(RecordPermissionPolicy):
         RecordOwners(),
         RecordCommunitiesAction("curate"),
         AccessGrant("manage"),
-        Administration(),
         SystemProcess(),
     ]
     can_curate = can_manage + [AccessGrant("edit"), SecretLinks("edit")]
@@ -200,9 +201,18 @@ class RDMRecordPermissionPolicy(RecordPermissionPolicy):
         ),
     ]
     # Allow publishing a new record or changes to an existing record.
-    can_publish = can_review
-    # Permission to allow special users to publish a record in special cases
-    can_publish_elevated = [Administration(), SystemProcess()]
+    can_publish = [
+        IfConfig(
+            "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
+            then_=[
+                IfAtLeastOneCommunity(
+                    then_=can_review,
+                    else_=[Administration(), SystemProcess()],
+                ),
+            ],
+            else_=can_review,
+        )
+    ]
     # Allow lifting a record or draft.
     can_lift_embargo = can_manage
 
@@ -212,16 +222,25 @@ class RDMRecordPermissionPolicy(RecordPermissionPolicy):
     # Who can add record to a community
     can_add_community = can_manage
     # Who can remove a community from a record
-    can_remove_community = [
+    can_remove_community_ = [
         RecordOwners(),
         CommunityCurators(),
-        Administration(),
         SystemProcess(),
     ]
-    # Permission to allow special users to remove community in special cases
-    can_remove_community_elevated = [Administration(), SystemProcess()]
+    can_remove_community = [
+        IfConfig(
+            "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
+            then_=[
+                IfOneCommunity(
+                    then_=[Administration(), SystemProcess()],
+                    else_=can_remove_community_,
+                ),
+            ],
+            else_=can_remove_community_,
+        ),
+    ]
     # Who can remove records from a community
-    can_remove_record = [CommunityCurators(), Administration()]
+    can_remove_record = [CommunityCurators(), Administration(), SystemProcess()]
     # Who can add records to a community in bulk
     can_bulk_add = [SystemProcess()]
 

invenio_rdm_records/services/services.py

@@ -38,7 +38,7 @@
 
 from ..records.systemfields.deletion_status import RecordDeletionStatusEnum
 from .errors import (
-    CommunityNotSelectedError,
+    CommunityRequiredError,
     DeletionStatusException,
     EmbargoNotLiftedError,
     RecordDeletedException,
@@ -412,22 +412,19 @@ def publish(self, identity, id_, uow=None, expand=False):
 
         Check for permissions to publish a draft and then call invenio_drafts_resourcs.services.records.services.publish()
         """
-        # Get the draft
-        draft = self.draft_cls.pid.resolve(id_, registered_only=False)
-
-        is_community_required = current_app.config["RDM_RECORD_ALWAYS_IN_COMMUNITY"]
-        is_community_missing = len(draft.parent.communities.ids) == 0
-        # If community is required for a record and there are no communities selected
-        # Then, check for permissions to publish without community
-        can_publish_without_community = self.check_permission(
-            identity, "publish_elevated", record=draft
-        )
-        if (
-            is_community_required
-            and is_community_missing
-            and not can_publish_without_community
-        ):
-            raise CommunityNotSelectedError()
+        try:
+            draft = self.draft_cls.pid.resolve(id_, registered_only=False)
+            self.require_permission(identity, "publish", record=draft)
+            # By default, admin/superuser has permission to do everything, so PermissionDeniedError won't be raised for admin in any case
+        except PermissionDeniedError as exc:
+            # If user doesn't have permission, then check if community is required
+            if current_app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"]:
+                # If there are no communities selected, raise custom error to inform user
+                if len(draft.parent.communities.ids) == 0:
+                    raise CommunityRequiredError()
+            else:
+                # If the config wasn't enabled, then raise the PermissionDeniedError
+                raise exc
 
         return super().publish(identity, id_, uow=uow, expand=expand)
 

tests/conftest.py

@@ -2014,20 +2014,22 @@ def create_record(
             """Creates new record that belongs to the same community."""
             # create draft
             draft = current_rdm_records_service.create(uploader.identity, record_dict)
-            # publish and get record
-            result_item = current_rdm_records_service.publish(
-                uploader.identity, draft.id
-            )
-            record = result_item._record
+            record = draft._record
             if community:
                 # add the record to the community
                 community_record = community._record
                 record.parent.communities.add(community_record, default=False)
                 record.parent.commit()
                 db.session.commit()
-                current_rdm_records_service.indexer.index(
-                    record, arguments={"refresh": True}
-                )
+
+            # publish and get record
+            result_item = current_rdm_records_service.publish(
+                uploader.identity, draft.id
+            )
+            record = result_item._record
+            current_rdm_records_service.indexer.index(
+                record, arguments={"refresh": True}
+            )
 
             return record
 
@@ -2081,41 +2083,6 @@ def create_record(
     return RecordFactory()
 
 
-@pytest.fixture()
-def record_required_community(db, uploader, minimal_record, community):
-    """Creates a record that belongs to a community before publishing."""
-
-    class Record:
-        """Test record class."""
-
-        def create_record(
-            self,
-            record_dict=minimal_record,
-            uploader=uploader,
-            community=community,
-        ):
-            """Creates new record that belongs to the same community."""
-            # create draft
-            draft = current_rdm_records_service.create(uploader.identity, record_dict)
-            record = draft._record
-            # add the record to the community
-            community_record = community._record
-            record.parent.communities.add(community_record, default=False)
-            record.parent.commit()
-            db.session.commit()
-            current_rdm_records_service.indexer.index(
-                record, arguments={"refresh": True}
-            )
-
-            # publish and get record
-            community_record = current_rdm_records_service.publish(
-                uploader.identity, draft.id
-            )
-            return community_record
-
-    return Record()
-
-
 @pytest.fixture(scope="session")
 def headers():
     """Default headers for making requests."""