Archiver permissions policy

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

@@ -10,7 +10,10 @@
 
 from flask import current_app
 from flask_principal import RoleNeed, UserNeed
-from invenio_records_permissions.generators import Generator
+from invenio_records_permissions.generators import AuthenticatedUser, Generator
+from invenio_search.engine import dsl
+
+oais_archiver_role = RoleNeed("oais-archiver")
 
 
 class CERNEmailsGroups(Generator):
@@ -46,3 +49,28 @@ def needs(self, **kwargs):
     def query_filter(self, **kwargs):
         """Match all in search."""
         raise NotImplementedError
+
+
+class AuthenticatedRegularUser(AuthenticatedUser):
+    """Generator for regular users. Excludes robot accounts."""
+
+    def excludes(self, **kwargs):
+        """Exclude service/robot accounts."""
+        excludes = super().excludes(**kwargs)
+        return excludes + [oais_archiver_role]
+
+
+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,12 @@
 """Permission policy."""
 
 from invenio_communities.permissions import CommunityPermissionPolicy
+from invenio_rdm_records.services.generators import IfRecordDeleted
+from invenio_rdm_records.services.permissions import RDMRecordPermissionPolicy
 from invenio_records_permissions.generators import SystemProcess
+from invenio_users_resources.services.permissions import UserManager
 
-from .generators import CERNEmailsGroups
+from .generators import Archiver, AuthenticatedRegularUser, CERNEmailsGroups
 
 
 class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy):
@@ -25,3 +28,24 @@ class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy):
         ),
         SystemProcess(),
     ]
+
+
+class CDSRDMRecordPermissionPolicy(RDMRecordPermissionPolicy):
+    """Record permission policy."""
+
+    can_create = [AuthenticatedRegularUser(), SystemProcess()]
+    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()],
+        )
+    ]