permissions: add oais-archiver role

invenio.cfg

@@ -11,7 +11,7 @@ import os
 
 from datetime import datetime, timedelta
 from invenio_i18n import lazy_gettext as _
-from cds_rdm.permissions import CDSCommunitiesPermissionPolicy
+from cds_rdm.permissions import CDSCommunitiesPermissionPolicy, CDSRDMRecordPermissionPolicy
 from cds_rdm.files import storage_factory
 from invenio_app_rdm.config import CELERY_BEAT_SCHEDULE as APP_RDM_CELERY_BEAT_SCHEDULE
 from celery.schedules import crontab
@@ -331,3 +331,5 @@ CDS_EOS_OFFLOAD_X509_CERT_PATH = ""
 CDS_EOS_OFFLOAD_X509_KEY_PATH = ""
 # check nginx config for more details
 CDS_EOS_OFFLOAD_REDIRECT_BASE_PATH = ""
+
+RDM_PERMISSION_POLICY = CDSRDMRecordPermissionPolicy

site/cds_rdm/generators.py

@@ -11,6 +11,9 @@
 from flask import current_app
 from flask_principal import RoleNeed, UserNeed
 from invenio_records_permissions.generators import Generator
+from invenio_search.engine import dsl
+
+oais_archiver_role = RoleNeed("oais-archiver")
 
 
 class CERNEmailsGroups(Generator):
@@ -46,3 +49,19 @@ def needs(self, **kwargs):
     def query_filter(self, **kwargs):
         """Match all in search."""
         raise NotImplementedError
+
+
+class Archiver(Generator):
+    """Allows system_process role."""
+
+    def needs(self, **kwargs):
+        """Enabling Needs."""
+        return [oais_archiver_role]
+
+    def query_filter(self, identity=None, **kwargs):
+        """Filters for current identity as system process."""
+        for need in identity.provides:
+            if need == oais_archiver_role:
+                return dsl.Q("match_all")
+        else:
+            return []

site/cds_rdm/permissions.py

@@ -9,9 +9,14 @@
 """Permission policy."""
 
 from invenio_communities.permissions import CommunityPermissionPolicy
-from invenio_records_permissions.generators import SystemProcess
+from invenio_rdm_records.services.permissions import RDMRecordPermissionPolicy
+from .generators import CERNEmailsGroups, Archiver
+from invenio_records_permissions.generators import (
+    SystemProcess,
+)
+from invenio_users_resources.services.permissions import UserManager
 
-from .generators import CERNEmailsGroups
+from invenio_rdm_records.services.generators import IfRecordDeleted
 
 
 class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy):
@@ -25,3 +30,22 @@ class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy):
         ),
         SystemProcess(),
     ]
+
+
+class CDSRDMRecordPermissionPolicy(RDMRecordPermissionPolicy):
+    can_view = RDMRecordPermissionPolicy.can_view
+    can_read = RDMRecordPermissionPolicy.can_read + [Archiver()]
+    can_search = RDMRecordPermissionPolicy.can_search + [Archiver()]
+    can_read_files = RDMRecordPermissionPolicy.can_read_files + [Archiver()]
+    can_get_content_files = RDMRecordPermissionPolicy.can_get_content_files + [
+        Archiver()
+    ]
+    can_media_get_content_files = RDMRecordPermissionPolicy.can_get_content_files + [
+        Archiver()
+    ]
+    can_read_deleted = [
+        IfRecordDeleted(
+            then_=[UserManager, SystemProcess()],
+            else_=can_read + [Archiver()],
+        )
+    ]