Permissions: Add Community Creator Role for can_create

invenio_communities/config.py

@@ -2,8 +2,8 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2016-2024 CERN.
-# Copyright (C) 2023 Graz University of Technology.
-# Copyright (C) 2023 KTH Royal Institute of Technology.
+# Copyright (C) 2023      Graz University of Technology.
+# Copyright (C) 2023-2024 KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -334,3 +334,6 @@
 
 COMMUNITIES_DEFAULT_RECORD_SUBMISSION_POLICY = RecordSubmissionPolicyEnum.OPEN
 """Default value of record submission policy community access setting."""
+
+COMMUNITIES_CREATOR_ROLE = "community-creator"
+"""Depends on 'RDM_COMMUNITY_REQUIRED_TO_PUBLISH' set to True."""

invenio_communities/generators.py

@@ -5,6 +5,7 @@
 # Copyright (C) 2021 Graz University of Technology.
 # Copyright (C) 2021 TU Wien.
 # Copyright (C) 2022 Northwestern University.
+# Copyright (C) 2024 KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -16,7 +17,8 @@
 from functools import partial, reduce
 from itertools import chain
 
-from flask_principal import UserNeed
+from flask import current_app
+from flask_principal import RoleNeed, UserNeed
 from invenio_access.permissions import any_user, authenticated_user, system_process
 from invenio_records.dictutils import dict_lookup
 from invenio_records_permissions.generators import Generator
@@ -342,6 +344,17 @@ def roles(self, **kwargs):
         return [r.name for r in current_roles.can("manage")]
 
 
+class CommunityCreator(Generator):
+    """Allows users with the "trusted-user" role."""
+
+    def needs(self, **kwargs):
+        """Enabling Needs."""
+        role_name = current_app.config.get(
+            "COMMUNITIES_CREATOR_ROLE", "community-creator"
+        )
+        return [RoleNeed(role_name)]
+
+
 class CommunityManagersForRole(CommunityRoles):
     """Roles representing all managers of a community for a role update."""
 

invenio_communities/permissions.py

@@ -5,6 +5,7 @@
 # Copyright (C) 2021 Graz University of Technology.
 # Copyright (C) 2021 TU Wien.
 # Copyright (C) 2022 Northwestern University.
+# Copyright (C) 2024 KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -26,6 +27,7 @@
 from .generators import (
     AllowedMemberTypes,
     AuthenticatedButNotCommunityMembers,
+    CommunityCreator,
     CommunityCurators,
     CommunityManagers,
     CommunityManagersForRole,
@@ -45,7 +47,15 @@ class CommunityPermissionPolicy(BasePermissionPolicy):
     """Permissions for Community CRUD operations."""
 
     # Community
-    can_create = [AuthenticatedUser(), SystemProcess()]
+    _can_create = [AuthenticatedUser(), SystemProcess()]
+
+    can_create = [
+        IfConfig(
+            "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
+            then_=[CommunityCreator(), Administration(), SystemProcess()],
+            else_=_can_create,
+        )
+    ]
 
     can_read = [
         IfRestricted("visibility", then_=[CommunityMembers()], else_=[AnyUser()]),
@@ -97,14 +107,22 @@ class CommunityPermissionPolicy(BasePermissionPolicy):
     ]
 
     # who can include a record directly, without a review
-    can_include_directly = [
+    _can_include_directly = [
         ReviewPolicy(
             closed_=[Disable()],
             open_=[CommunityCurators()],
             members_=[CommunityMembers()],
         )
     ]
 
+    can_include_directly = [
+        IfConfig(
+            "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
+            then_=[Disable()],
+            else_=_can_include_directly,
+        )
+    ]
+
     can_members_add = [
         CommunityManagersForRole(),
         AllowedMemberTypes("group"),

invenio_communities/views/communities.py

@@ -3,6 +3,7 @@
 # This file is part of Invenio.
 # Copyright (C) 2016-2024 CERN.
 # Copyright (C) 2023 Graz University of Technology.
+# Copyright (C) 2024 KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -432,14 +433,23 @@ def communities_settings_submission_policy(pid_value, community, community_ui):
     if not permissions["can_update"]:
         raise PermissionDeniedError()
 
+    if current_app.config.get("RDM_COMMUNITY_REQUIRED_TO_PUBLISH", False):
+        # Restrict review policies when publishing with community is required
+        available_review_policies = [
+            policy for policy in REVIEW_POLICY_FIELDS if policy["value"] == "closed"
+        ]
+
+    else:
+        available_review_policies = REVIEW_POLICY_FIELDS
+
     return render_community_theme_template(
         "invenio_communities/details/settings/submission_policy.html",
         theme=community_ui.get("theme", {}),
         community=community_ui,
         permissions=permissions,
         form_config=dict(
             access=dict(
-                review_policy=REVIEW_POLICY_FIELDS,
+                review_policy=available_review_policies,
                 record_submission_policy=RECORDS_SUBMISSION_POLICY_FIELDS,
             ),
         ),

tests/communities/test_community_ui_serializer.py

@@ -7,11 +7,17 @@
 
 """Resources serializers tests."""
 
+from collections import namedtuple
+from functools import partial
+
 from flask import g
+from flask_principal import Identity, RoleNeed
+from invenio_access.permissions import system_identity
 
 from invenio_communities.communities.resources.serializer import (
     UICommunityJSONSerializer,
 )
+from invenio_communities.permissions import CommunityPermissionPolicy
 
 
 def test_ui_serializer(app, community, users, any_user):
@@ -75,3 +81,227 @@ def test_ui_serializer(app, community, users, any_user):
         serialized_record["ui"]["permissions"]
         == require_review_expected_data["permissions"]
     )
+
+
+def test_can_include_directly_closed(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'closed' review policy."""
+    # Set the community's review policy to 'closed'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "closed"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    superuser_id = superuser_identity
+    system_process_identity = system_identity
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Member (Owner)
+    member_identity = Identity(owner.id)
+    member_identity.provides.update(owner_identity.provides)
+    # Assume owner is a member
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Community Curator
+    curator_identity = Identity(any_user.id)
+    curator_identity.provides.update(authenticated_identity.provides)
+    curator_identity.provides.add(CommunityRoleNeed(community_id, "curator"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to True
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = True
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions: No one can include directly
+    identities = [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        member_identity,
+        curator_identity,
+        owner_identity,
+        superuser_id,
+        system_process_identity,
+    ]
+
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    # Assertions remain the same: No one can include directly
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+
+def test_can_include_directly_open(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'open' review policy."""
+    # Set the community's review policy to 'open'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "open"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    system_process_identity = system_identity
+
+    # Create identities with roles
+    from flask_principal import Identity, RoleNeed
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Curator
+    from collections import namedtuple
+    from functools import partial
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    curator_identity = Identity(any_user.id)
+    curator_identity.provides.update(authenticated_identity.provides)
+    curator_identity.provides.add(CommunityRoleNeed(community_id, "curator"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to False
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is False
+    # Only community curators can include directly
+    # Superuser should not be allowed
+    assert not policy(action="include_directly", record=community_record).allows(
+        anonymous_identity
+    )
+    assert not policy(action="include_directly", record=community_record).allows(
+        authenticated_identity
+    )
+    assert not policy(action="include_directly", record=community_record).allows(
+        community_creator_identity
+    )
+
+    # Community curator can include directly
+    assert policy(action="include_directly", record=community_record).allows(
+        curator_identity
+    )
+
+
+def test_can_include_directly_members(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'members' review policy."""
+    # Set the community's review policy to 'members'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "members"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    superuser_id = superuser_identity
+    system_process_identity = system_identity
+
+    # Create identities with roles
+    from flask_principal import Identity, RoleNeed
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Member
+    member_identity = Identity(any_user.id)
+    member_identity.provides.update(authenticated_identity.provides)
+    from collections import namedtuple
+    from functools import partial
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Assume owner is a member
+    owner_member_identity = Identity(owner.id)
+    owner_member_identity.provides.update(owner_identity.provides)
+    owner_member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to True
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = True
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is True
+    identities = [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        member_identity,
+    ]
+
+    # All users are denied
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to False
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is False
+    # Only community members can include directly
+    for identity in [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        system_process_identity,
+    ]:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )