services: add record access config flag

invenio_rdm_records/config.py

@@ -118,6 +118,12 @@ def always_valid(identifier):
 RDM_ALLOW_METADATA_ONLY_RECORDS = True
 """Allow users to publish metadata-only records."""
 
+#
+# Record access
+#
+RDM_ALLOW_RESTRICTED_RECORDS = True
+"""Allow users to set restricted/private records."""
+
 #
 # Search configuration
 #

invenio_rdm_records/services/components/access.py

@@ -9,6 +9,8 @@
 
 """RDM service component for access integration."""
 
+from flask_babelex import gettext as _
+
 from invenio_access.permissions import system_process
 from invenio_drafts_resources.services.records.components import ServiceComponent
 from marshmallow import ValidationError
@@ -19,13 +21,20 @@ class AccessComponent(ServiceComponent):
 
     def _populate_access_and_validate(self, identity, data, record, **kwargs):
         """Populate and validate the record's access field."""
+        errors = []
         if record is not None and "access" in data:
-            # populate the record's access field with the data already
-            # validated by marshmallow
-            record.update({"access": data.get("access")})
-            record.access.refresh_from_dict(record.get("access"))
+            access = data.get("access")
+            record_access = (access or {}).get("record")
+            can_manage = self.service.check_permission(identity, "manage_record_access")
+            if record_access and can_manage:
+                # populate the record's access field with the data already
+                # validated by marshmallow
+                record.update({"access": access})
+                record.access.refresh_from_dict(record.get("access"))
+            else:
+                errors.append(_("You don't have permissions to manage record access."))
 
-        errors = record.access.errors
+        errors.extend(record.access.errors)
         if errors:
             # filter out duplicate error messages
             messages = list({str(e) for e in errors})

invenio_rdm_records/services/permissions.py

@@ -116,6 +116,10 @@ class RDMRecordPermissionPolicy(RecordPermissionPolicy):
     can_manage_files = [
         IfConfig("RDM_ALLOW_METADATA_ONLY_RECORDS", then_=can_review, else_=[]),
     ]
+    # Allow managing record access
+    can_manage_record_access = [
+        IfConfig("RDM_ALLOW_RESTRICTED_RECORDS", then_=can_review, else_=[]),
+    ]
 
     #
     # PIDs

tests/resources/test_resources.py

@@ -457,6 +457,42 @@ def test_multiple_files_record(
     assert response.status_code == 202
 
 
+@pytest.fixture()
+def restricted_records_disabled(app):
+    old_value = app.config.get("RDM_ALLOW_RESTRICTED_RECORDS", True)
+    app.config["RDM_ALLOW_RESTRICTED_RECORDS"] = False
+    yield
+    app.config["RDM_ALLOW_RESTRICTED_RECORDS"] = old_value
+
+
+def test_restricted_records_disabled(
+    running_app, client_with_login, headers, minimal_record, search_clear,
+    superuser,
+    restricted_records_disabled,
+):
+    client = client_with_login
+    response = client.post("/records", json=minimal_record, headers=headers)
+    recid = response.json["id"]
+
+    assert response.status_code == 201
+    assert response.json["access"]["record"] == "public"
+    assert response.json["errors"][0]["field"] == "access"
+    assert response.json["errors"][0]["message"] == [
+        "You don't have permissions to manage record access.",
+    ]
+
+    superuser.api_login(client, logout_first=True)
+    minimal_record["access"]["record"] = "restricted"
+    response = client.put(
+        f"/records/{recid}/draft",
+        json=minimal_record,
+        headers=headers,
+    )
+    assert response.status_code == 200
+    assert response.json["access"]["record"] == "restricted"
+    assert "errors" not in response.json
+
+
 # TODO
 @pytest.mark.skip()
 def test_create_publish_new_revision(