services: configure record access via config

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

@@ -11,6 +11,7 @@
 
 from invenio_access.permissions import system_process
 from invenio_drafts_resources.services.records.components import ServiceComponent
+from invenio_i18n import gettext as _
 from marshmallow import ValidationError
 
 
@@ -19,13 +20,31 @@ 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:
+            access = data.get("access", {})
+
+            # Explicit permission check for modifying record access. This is inflexible,
+            # but generalizing management permissions per-field is difficult.
+            new_record_access = access.get("record")
+            if record.access.protection.record != new_record_access:
+                can_manage = self.service.check_permission(
+                    identity, "manage_record_access"
+                )
+                if not can_manage:
+                    # Set the data to what it was before
+                    if "record" in access:
+                        access["record"] = record.access.protection.record
+                    errors.append(
+                        _("You don't have permissions to manage record access.")
+                    )
+
             # populate the record's access field with the data already
             # validated by marshmallow
-            record.update({"access": data.get("access")})
+            record.update({"access": access})
             record.access.refresh_from_dict(record.get("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/conftest.py

@@ -322,6 +322,11 @@ def app_config(app_config):
     # Communities
     app_config["COMMUNITY_SERVICE_COMPONENTS"] = CommunityServiceComponents
 
+    # TODO: Remove when https://github.com/inveniosoftware/pytest-invenio/pull/95 is
+    #       merged and released.
+    # Disable rate-limiting
+    app_config["RATELIMIT_ENABLED"] = False
+
     return app_config
 
 

tests/resources/test_resources.py

@@ -457,6 +457,59 @@ 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"
+
+    # Trying to change record access to restricted will fail
+    minimal_record["access"]["record"] = "restricted"
+    response = client.put(
+        f"/records/{recid}/draft",
+        json=minimal_record,
+        headers=headers,
+    )
+    assert response.status_code == 400
+    assert response.json["errors"][0]["field"] == "access"
+    assert response.json["errors"][0]["messages"] == [
+        "You don't have permissions to manage record access.",
+    ]
+    # Record access should still be "public"
+    response = client.get(f"/records/{recid}/draft", headers=headers)
+    assert response.json["access"]["record"] == "public"
+
+    # Superuser can change record access
+    superuser.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(