authentication: add read permission for records

Author: dianarand

cernopendata/config.py

@@ -341,54 +341,65 @@
     ),
 )
 
+_RECORD_PERMISSION_FACTORY = (
+    "cernopendata.modules.records.permissions:record_read_permission_factory"
+)
+
 RECORDS_UI_ENDPOINTS = dict(
     recid=dict(
         pid_type="recid",
         route="/record/<pid_value>",
-        permission_factory_imp=None,
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         record_class="cernopendata.api:RecordFilesWithIndex",
         view_imp="cernopendata.modules.records.utils:record_metadata_view",
     ),
     recid_files=dict(
         pid_type="recid",
         route="/record/<pid_value>/files/<path:filename>",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:file_download_ui",
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_file_index=dict(
         pid_type="recid",
         route="/record/<pid_value>/file_index/<path:file_index>",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:get_file_index",
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_files_assets=dict(
         pid_type="recid",
         route="/record/<pid_value>/files/assets/<path:filepath>",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:eos_file_download_ui",
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_files_page=dict(
         pid_type="recid",
         route="/record/<pid_value>/filepage/<int:page>",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:record_file_page",
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_export=dict(
         pid_type="recid",
         route="/record/<pid_value>/export/<format>",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:export_json_view",
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_stage=dict(
         pid_type="recid",
         route="/record/<pid_value>/stage",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:stage",
         methods=["POST"],
         record_class="cernopendata.api:RecordFilesWithIndex",
     ),
     recid_subscribe=dict(
         pid_type="recid",
         route="/record/<pid_value>/subscribe",
+        permission_factory_imp=_RECORD_PERMISSION_FACTORY,
         view_imp="cernopendata.modules.records.utils:subscribe",
         methods=["POST"],
         record_class="cernopendata.api:RecordFilesWithIndex",
@@ -460,6 +471,7 @@ def _query_parser_and(qstr=None):
             ),
         },
         "search_factory_imp": "cernopendata.modules.records.queries:search_factory",
+        "read_permission_factory_imp": _RECORD_PERMISSION_FACTORY,
     }
 )
 

cernopendata/modules/records/permissions.py

@@ -0,0 +1,40 @@
+"""CERN Open Data record permissions."""
+
+from flask_login import current_user
+
+from cernopendata.modules.releases.models import ReleaseMetadata, ReleaseStatus
+from cernopendata.modules.releases.utils import curator_experiments
+
+
+def record_read_permission_factory(record, *args, **kwargs):
+    """Return a read permission for a record.
+
+    By default, records are publicly accessible. If the record appears
+    in a release that has not yet been published, access is restricted
+    to curators of the corresponding experiment.
+    """
+
+    def can(self):
+        prerelease = record.get("prerelease")
+        if not prerelease:
+            return True
+
+        try:
+            experiment, release_id = prerelease.split("/", 1)
+            release_id = int(release_id)
+        except (ValueError, AttributeError):
+            return True
+
+        release = ReleaseMetadata.query.filter_by(
+            id=release_id, experiment=experiment
+        ).first()
+
+        if release is None or release.status == ReleaseStatus.PUBLISHED.value:
+            return True
+
+        if not current_user.is_authenticated:
+            return False
+
+        return experiment in curator_experiments()["curator_experiments"]
+
+    return type("RecordReadPermission", (), {"can": can})()