diff --git a/site/cds_rdm/generators.py b/site/cds_rdm/generators.py index 77506d1..42eb9ee 100644 --- a/site/cds_rdm/generators.py +++ b/site/cds_rdm/generators.py @@ -10,7 +10,7 @@ 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") @@ -51,6 +51,15 @@ def query_filter(self, **kwargs): 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.""" diff --git a/site/cds_rdm/permissions.py b/site/cds_rdm/permissions.py index f97cdd9..4eea7c1 100644 --- a/site/cds_rdm/permissions.py +++ b/site/cds_rdm/permissions.py @@ -9,14 +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 .generators import CERNEmailsGroups, Archiver -from invenio_records_permissions.generators import ( - SystemProcess, -) +from invenio_records_permissions.generators import SystemProcess from invenio_users_resources.services.permissions import UserManager -from invenio_rdm_records.services.generators import IfRecordDeleted +from .generators import Archiver, AuthenticatedRegularUser, CERNEmailsGroups class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy): @@ -33,7 +31,9 @@ class CDSCommunitiesPermissionPolicy(CommunityPermissionPolicy): class CDSRDMRecordPermissionPolicy(RDMRecordPermissionPolicy): - can_view = RDMRecordPermissionPolicy.can_view + """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()] diff --git a/site/tests/conftest.py b/site/tests/conftest.py index 714851a..ec57acb 100644 --- a/site/tests/conftest.py +++ b/site/tests/conftest.py @@ -7,10 +7,26 @@ """Pytest fixtures.""" +from collections import namedtuple + import pytest -from invenio_app.factory import create_api +from invenio_access.models import ActionRoles +from invenio_access.permissions import superuser_access, system_identity +from invenio_accounts.models import Role +from invenio_administration.permissions import administration_access_action +from invenio_app import factory as app_factory +from invenio_rdm_records.cli import create_records_custom_field +from invenio_rdm_records.services.pids import providers +from invenio_records_resources.proxies import current_service_registry +from invenio_vocabularies.contrib.awards.api import Award +from invenio_vocabularies.contrib.funders.api import Funder +from invenio_vocabularies.proxies import current_service as vocabulary_service +from invenio_vocabularies.records.api import Vocabulary -from cds_rdm.permissions import CDSCommunitiesPermissionPolicy +from cds_rdm.permissions import ( + CDSCommunitiesPermissionPolicy, + CDSRDMRecordPermissionPolicy, +) @pytest.fixture(scope="module") @@ -26,6 +42,7 @@ def app_config(app_config): } app_config["CERN_LDAP_URL"] = "" # mock app_config["COMMUNITIES_PERMISSION_POLICY"] = CDSCommunitiesPermissionPolicy + app_config["RDM_PERMISSION_POLICY"] = CDSRDMRecordPermissionPolicy app_config["COMMUNITIES_ALLOW_RESTRICTED"] = True app_config["CDS_GROUPS_ALLOW_CREATE_COMMUNITIES"] = [ "group-allowed-create-communities" @@ -33,7 +50,730 @@ def app_config(app_config): return app_config +# @pytest.fixture(scope="module") +# def create_app(): +# """Create test app.""" +# return create_api + + +@pytest.fixture(scope="function") +def db_session_options(): + """Database session options.""" + # TODO: Look into having this be the default in ``pytest-invenio`` + # This helps with ``sqlalchemy.orm.exc.DetachedInstanceError`` when models are not + # bound to the session between transactions/requests/service-calls. + return {"expire_on_commit": False} + + +@pytest.fixture(scope="module") +def create_app(instance_path): + """Application factory fixture.""" + return app_factory.create_app + + +RunningApp = namedtuple( + "RunningApp", + [ + "app", + "superuser_identity", + "location", + "cache", + "resource_type_v", + "languages_type", + "funders_v", + "awards_v", + "licenses_v", + "contributors_role_v", + "description_type_v", + "relation_type_v", + "initialise_custom_fields", + ], +) + + +@pytest.fixture +def running_app( + app, + superuser_identity, + location, + cache, + resource_type_v, + languages_v, + funders_v, + awards_v, + licenses_v, + contributors_role_v, + description_type_v, + relation_type_v, + initialise_custom_fields, +): + """This fixture provides an app with the typically needed db data loaded. + + All of these fixtures are often needed together, so collecting them + under a semantic umbrella makes sense. + """ + return RunningApp( + app, + superuser_identity, + location, + cache, + resource_type_v, + languages_v, + funders_v, + awards_v, + licenses_v, + contributors_role_v, + description_type_v, + relation_type_v, + initialise_custom_fields, + ) + + +@pytest.fixture +def test_app(running_app): + """Get current app.""" + return running_app.app + + +@pytest.fixture(scope="session") +def headers(): + """Default headers for making requests.""" + return { + "content-type": "application/json", + "accept": "application/json", + } + + +@pytest.fixture() +def superuser_role_need(db): + """Store 1 role with 'superuser-access' ActionNeed. + + WHY: This is needed because expansion of ActionNeed is + done on the basis of a User/Role being associated with that Need. + If no User/Role is associated with that Need (in the DB), the + permission is expanded to an empty list. + """ + role = Role(name="superuser-access") + db.session.add(role) + + action_role = ActionRoles.create(action=superuser_access, role=role) + db.session.add(action_role) + db.session.commit() + + return action_role.need + + +@pytest.fixture() +def superuser(UserFixture, app, db, superuser_role_need): + """Superuser.""" + u = UserFixture( + email="superuser@inveniosoftware.org", + password="superuser", + ) + u.create(app, db) + + datastore = app.extensions["security"].datastore + _, role = datastore._prepare_role_modify_args(u.user, "superuser-access") + + datastore.add_role_to_user(u.user, role) + db.session.commit() + return u + + +@pytest.fixture() +def admin_role_need(db): + """Store 1 role with 'superuser-access' ActionNeed. + + WHY: This is needed because expansion of ActionNeed is + done on the basis of a User/Role being associated with that Need. + If no User/Role is associated with that Need (in the DB), the + permission is expanded to an empty list. + """ + role = Role(name="administration-access") + db.session.add(role) + + action_role = ActionRoles.create(action=administration_access_action, role=role) + db.session.add(action_role) + db.session.commit() + + return action_role.need + + +@pytest.fixture() +def admin(UserFixture, app, db, admin_role_need): + """Admin user for requests.""" + u = UserFixture( + email="admin@inveniosoftware.org", + password="admin", + ) + u.create(app, db) + + datastore = app.extensions["security"].datastore + _, role = datastore._prepare_role_modify_args(u.user, "administration-access") + + datastore.add_role_to_user(u.user, role) + db.session.commit() + return u + + +@pytest.fixture() +def superuser_identity(admin, superuser_role_need): + """Superuser identity fixture.""" + identity = admin.identity + identity.provides.add(superuser_role_need) + return identity + + +@pytest.fixture() +def uploader(UserFixture, app, db, test_app): + """Uploader.""" + u = UserFixture( + email="uploader@inveniosoftware.org", + password="uploader", + preferences={ + "visibility": "public", + "email_visibility": "restricted", + "notifications": { + "enabled": True, + }, + }, + active=True, + confirmed=True, + ) + u.create(app, db) + + return u + + +@pytest.fixture() +def archiver(UserFixture, app, db): + """Uploader.""" + ds = app.extensions["invenio-accounts"].datastore + user = UserFixture( + email="archiver@inveniosoftware.org", + password="archiver", + preferences={ + "visibility": "public", + "email_visibility": "restricted", + "notifications": { + "enabled": True, + }, + }, + active=True, + confirmed=True, + ) + user_obj = user.create(app, db) + r = ds.create_role(name="oais-archiver", description="1234") + ds.add_role_to_user(user.user, r) + + return user + + +@pytest.fixture(scope="module") +def resource_type_type(app): + """Resource type vocabulary type.""" + return vocabulary_service.create_type(system_identity, "resourcetypes", "rsrct") + + +@pytest.fixture(scope="module") +def resource_type_v(app, resource_type_type): + """Resource type vocabulary record.""" + vocabulary_service.create( + system_identity, + { + "id": "dataset", + "icon": "table", + "props": { + "csl": "dataset", + "datacite_general": "Dataset", + "datacite_type": "", + "openaire_resourceType": "21", + "openaire_type": "dataset", + "eurepo": "info:eu-repo/semantics/other", + "schema.org": "https://schema.org/Dataset", + "subtype": "", + "type": "dataset", + }, + "title": {"en": "Dataset"}, + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + }, + ) + + vocabulary_service.create( + system_identity, + { # create base resource type + "id": "image", + "props": { + "csl": "figure", + "datacite_general": "Image", + "datacite_type": "", + "openaire_resourceType": "25", + "openaire_type": "dataset", + "eurepo": "info:eu-repo/semantic/other", + "schema.org": "https://schema.org/ImageObject", + "subtype": "", + "type": "image", + }, + "icon": "chart bar outline", + "title": {"en": "Image"}, + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "publication-book", + "icon": "file alternate", + "props": { + "csl": "book", + "datacite_general": "Text", + "datacite_type": "Book", + "openaire_resourceType": "2", + "openaire_type": "publication", + "eurepo": "info:eu-repo/semantics/book", + "schema.org": "https://schema.org/Book", + "subtype": "publication-book", + "type": "publication", + }, + "title": {"en": "Book", "de": "Buch"}, + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "presentation", + "icon": "group", + "props": { + "csl": "speech", + "datacite_general": "Text", + "datacite_type": "Presentation", + "openaire_resourceType": "0004", + "openaire_type": "publication", + "eurepo": "info:eu-repo/semantics/lecture", + "schema.org": "https://schema.org/PresentationDigitalDocument", + "subtype": "", + "type": "presentation", + }, + "title": {"en": "Presentation", "de": "Präsentation"}, + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "publication", + "icon": "file alternate", + "props": { + "csl": "report", + "datacite_general": "Text", + "datacite_type": "", + "openaire_resourceType": "0017", + "openaire_type": "publication", + "eurepo": "info:eu-repo/semantics/other", + "schema.org": "https://schema.org/CreativeWork", + "subtype": "", + "type": "publication", + }, + "title": {"en": "Publication", "de": "Publikation"}, + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "image-photo", + "tags": ["depositable", "linkable"], + "type": "resourcetypes", + "title": {"en": "Image: Photo"}, + "props": { + "csl": "graphic", + "datacite_general": "Image", + "datacite_type": "Photo", + "openaire_resourceType": "0025", + "openaire_type": "dataset", + "eurepo": "info:eu-repo/semantics/other", + "schema.org": "https://schema.org/Photograph", + "subtype": "image-photo", + "type": "image", + }, + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "software", + "icon": "code", + "type": "resourcetypes", + "props": { + "csl": "software", + "datacite_general": "Software", + "datacite_type": "", + "openaire_resourceType": "0029", + "openaire_type": "software", + "eurepo": "info:eu-repo/semantics/other", + "schema.org": "https://schema.org/SoftwareSourceCode", + "subtype": "", + "type": "software", + }, + "title": {"en": "Software", "de": "Software"}, + "tags": ["depositable", "linkable"], + }, + ) + + vocab = vocabulary_service.create( + system_identity, + { + "id": "other", + "icon": "asterisk", + "type": "resourcetypes", + "props": { + "csl": "article", + "datacite_general": "Other", + "datacite_type": "", + "openaire_resourceType": "0020", + "openaire_type": "other", + "eurepo": "info:eu-repo/semantics/other", + "schema.org": "https://schema.org/CreativeWork", + "subtype": "", + "type": "other", + }, + "title": { + "en": "Other", + "de": "Sonstige", + }, + "tags": ["depositable", "linkable"], + }, + ) + + Vocabulary.index.refresh() + + return vocab + + +@pytest.fixture(scope="module") +def languages_type(app): + """Lanuage vocabulary type.""" + return vocabulary_service.create_type(system_identity, "languages", "lng") + + +@pytest.fixture(scope="module") +def languages_v(app, languages_type): + """Language vocabulary record.""" + vocabulary_service.create( + system_identity, + { + "id": "dan", + "title": { + "en": "Danish", + "da": "Dansk", + }, + "props": {"alpha_2": "da"}, + "tags": ["individual", "living"], + "type": "languages", + }, + ) + + vocab = vocabulary_service.create( + system_identity, + { + "id": "eng", + "title": { + "en": "English", + "da": "Engelsk", + }, + "tags": ["individual", "living"], + "type": "languages", + }, + ) + + Vocabulary.index.refresh() + + return vocab + + +@pytest.fixture(scope="module") +def funders_v(app, funder_data): + """Funder vocabulary record.""" + funders_service = current_service_registry.get("funders") + funder = funders_service.create( + system_identity, + funder_data, + ) + + Funder.index.refresh() + + return funder + + +@pytest.fixture(scope="module") +def awards_v(app, funders_v): + """Funder vocabulary record.""" + awards_service = current_service_registry.get("awards") + award = awards_service.create( + system_identity, + { + "id": "00rbzpz17::755021", + "identifiers": [ + { + "identifier": "https://cordis.europa.eu/project/id/755021", + "scheme": "url", + } + ], + "number": "755021", + "title": { + "en": ( + "Personalised Treatment For Cystic Fibrosis Patients With " + "Ultra-rare CFTR Mutations (and beyond)" + ), + }, + "funder": {"id": "00rbzpz17"}, + "acronym": "HIT-CF", + "program": "H2020", + }, + ) + + Award.index.refresh() + + return award + + +@pytest.fixture(scope="module") +def licenses(app): + """Licenses vocabulary type.""" + return vocabulary_service.create_type(system_identity, "licenses", "lic") + + +@pytest.fixture(scope="module") +def licenses_v(app, licenses): + """Licenses vocabulary record.""" + cc_zero = vocabulary_service.create( + system_identity, + { + "id": "cc0-1.0", + "title": { + "en": "Creative Commons Zero v1.0 Universal", + }, + "description": { + "en": ( + "CC0 waives copyright interest in a work you've created and " + "dedicates it to the world-wide public domain. Use CC0 to opt out " + "of copyright entirely and ensure your work has the widest reach." + ), + }, + "icon": "cc-cc0-icon", + "tags": ["recommended", "all", "data", "software"], + "props": { + "url": "https://creativecommons.org/publicdomain/zero/1.0/legalcode", + "scheme": "spdx", + "osi_approved": "", + }, + "type": "licenses", + }, + ) + cc_by = vocabulary_service.create( + system_identity, + { + "id": "cc-by-4.0", + "title": { + "en": "Creative Commons Attribution 4.0 International", + }, + "description": { + "en": ( + "The Creative Commons Attribution license allows re-distribution " + "and re-use of a licensed work on the condition that the creator " + "is appropriately credited." + ), + }, + "icon": "cc-by-icon", + "tags": ["recommended", "all", "data"], + "props": { + "url": "https://creativecommons.org/licenses/by/4.0/legalcode", + "scheme": "spdx", + "osi_approved": "", + }, + "type": "licenses", + }, + ) + + Vocabulary.index.refresh() + + return [cc_zero, cc_by] + + +@pytest.fixture(scope="module") +def contributors_role_type(app): + """Contributor role vocabulary type.""" + return vocabulary_service.create_type(system_identity, "contributorsroles", "cor") + + +@pytest.fixture(scope="module") +def contributors_role_v(app, contributors_role_type): + """Contributor role vocabulary record.""" + vocabulary_service.create( + system_identity, + { + "id": "other", + "props": {"datacite": "Other"}, + "title": {"en": "Other"}, + "type": "contributorsroles", + }, + ) + + vocab = vocabulary_service.create( + system_identity, + { + "id": "datacurator", + "props": {"datacite": "DataCurator"}, + "title": {"en": "Data curator", "de": "DatenkuratorIn"}, + "type": "contributorsroles", + }, + ) + + vocabulary_service.create( + system_identity, + { + "id": "supervisor", + "props": {"datacite": "Supervisor"}, + "title": {"en": "Supervisor"}, + "type": "contributorsroles", + }, + ) + + Vocabulary.index.refresh() + + return vocab + + +@pytest.fixture(scope="module") +def description_type(app): + """Title vocabulary type.""" + return vocabulary_service.create_type(system_identity, "descriptiontypes", "dty") + + +@pytest.fixture(scope="module") +def description_type_v(app, description_type): + """Title Type vocabulary record.""" + vocabulary_service.create( + system_identity, + { + "id": "methods", + "title": {"en": "Methods"}, + "props": {"datacite": "Methods"}, + "type": "descriptiontypes", + }, + ) + + vocab = vocabulary_service.create( + system_identity, + { + "id": "notes", + "title": {"en": "Notes"}, + "props": {"datacite": "Notes"}, + "type": "descriptiontypes", + }, + ) + + Vocabulary.index.refresh() + + return vocab + + @pytest.fixture(scope="module") -def create_app(): - """Create test app.""" - return create_api +def relation_type(app): + """Relation type vocabulary type.""" + return vocabulary_service.create_type(system_identity, "relationtypes", "rlt") + + +@pytest.fixture(scope="module") +def relation_type_v(app, relation_type): + """Relation type vocabulary record.""" + vocabulary_service.create( + system_identity, + { + "id": "iscitedby", + "props": {"datacite": "IsCitedBy"}, + "title": {"en": "Is cited by"}, + "type": "relationtypes", + }, + ) + + vocab = vocabulary_service.create( + system_identity, + { + "id": "cites", + "props": {"datacite": "Cites"}, + "title": {"en": "Cites", "de": "Zitiert"}, + "type": "relationtypes", + }, + ) + + Vocabulary.index.refresh() + + return vocab + + +@pytest.fixture(scope="function") +def initialise_custom_fields(app, db, location, cli_runner): + """Fixture initialises custom fields.""" + return cli_runner(create_records_custom_field) + + +@pytest.fixture(scope="module") +def funder_data(): + """Implements a funder's data.""" + return { + "id": "00rbzpz17", + "identifiers": [ + { + "identifier": "00rbzpz17", + "scheme": "ror", + }, + {"identifier": "10.13039/501100001665", "scheme": "doi"}, + ], + "name": "Agence Nationale de la Recherche", + "title": { + "fr": "National Agency for Research", + }, + "country": "FR", + } + + +@pytest.fixture() +def minimal_restricted_record(): + """Minimal record data as dict coming from the external world.""" + return { + "pids": {}, + "access": { + "record": "restricted", + "files": "restricted", + }, + "files": { + "enabled": False, # Most tests don't care about files + }, + "metadata": { + "creators": [ + { + "person_or_org": { + "family_name": "Brown", + "given_name": "Troy", + "type": "personal", + } + }, + ], + "publication_date": "2020-06-01", + "publisher": "Acme Inc", + "resource_type": {"id": "image-photo"}, + "title": "A Romans story", + }, + } diff --git a/site/tests/test_permissions.py b/site/tests/test_permissions.py new file mode 100644 index 0000000..8b8a7c3 --- /dev/null +++ b/site/tests/test_permissions.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2024 CERN. +# +# CDS-RDM is free software; you can redistribute it and/or modify it under +# the terms of the MIT License; see LICENSE file for more details. + +"""Permissions tests.""" +import pytest +from invenio_rdm_records.proxies import current_rdm_records +from invenio_rdm_records.records.api import RDMDraft, RDMParent, RDMRecord +from invenio_records_resources.services.errors import PermissionDeniedError + + +def test_archiver_permissions( + db, app, minimal_restricted_record, uploader, client, headers, archiver +): + """Check the permissions of the archiver.""" + service = current_rdm_records.records_service + draft = service.create(uploader.identity, minimal_restricted_record) + recid = draft.id + r = service.publish(uploader.identity, draft.id) + RDMRecord.index.refresh() + + with pytest.raises(PermissionDeniedError): + new_draft = service.edit(archiver.identity, recid) + + with pytest.raises(PermissionDeniedError): + deleted = service.delete(archiver.identity, recid) + + with pytest.raises(PermissionDeniedError): + create = service.create(archiver.identity, minimal_restricted_record) + + results = service.search(archiver.identity) + assert results.total == 1 + assert results.to_dict()["hits"]["hits"][0]["id"] == recid