permissions: restrict non searchable values

* CERNDocumentServer/cds-rdm#194

invenio_vocabularies/contrib/affiliations/affiliations.py

@@ -16,7 +16,8 @@
 from invenio_records_resources.records.systemfields import ModelPIDField
 from invenio_records_resources.resources.records.headers import etag_headers
 
-from ...services.permissions import PermissionPolicy
+from invenio_vocabularies.services.permissions import PermissionPolicy
+
 from .config import AffiliationsSearchOptions, service_components
 from .schema import AffiliationSchema
 

invenio_vocabularies/contrib/awards/awards.py

@@ -27,7 +27,8 @@
 )
 from invenio_records_resources.resources.records.headers import etag_headers
 
-from ...services.permissions import PermissionPolicy
+from invenio_vocabularies.services.permissions import PermissionPolicy
+
 from ..funders.api import Funder
 from ..subjects.api import Subject
 from .config import AwardsSearchOptions, service_components

invenio_vocabularies/contrib/funders/funders.py

@@ -21,7 +21,8 @@
 from invenio_records_resources.records.systemfields import ModelPIDField
 from invenio_records_resources.resources.records.headers import etag_headers
 
-from ...services.permissions import PermissionPolicy
+from invenio_vocabularies.services.permissions import PermissionPolicy
+
 from .config import FundersSearchOptions, service_components
 from .schema import FunderSchema
 from .serializer import FunderL10NItemSchema

invenio_vocabularies/contrib/names/permissions.py

@@ -10,11 +10,23 @@
 
 from invenio_records_permissions.generators import AuthenticatedUser, SystemProcess
 
-from ...services.permissions import PermissionPolicy
+from invenio_vocabularies.services.generators import IfTags
+from invenio_vocabularies.services.permissions import PermissionPolicy
 
 
 class NamesPermissionPolicy(PermissionPolicy):
-    """Permission policy."""
+    """Names permission policy.
 
-    can_search = [SystemProcess(), AuthenticatedUser()]
-    can_read = [SystemProcess(), AuthenticatedUser()]
+    Names endpoints are protected, only authenticated users can access them.
+    """
+
+    can_search = [
+        SystemProcess(),
+        IfTags(exclude=["unlisted"], only_authenticated=True),
+    ]
+    can_read = [SystemProcess(), IfTags(exclude=["unlisted"], only_authenticated=True)]
+    # this permission is needed for the /api/vocabularies/ endpoint
+    can_list_vocabularies = [
+        SystemProcess(),
+        IfTags(exclude=["unlisted"], only_authenticated=True),
+    ]

invenio_vocabularies/contrib/names/schema.py

@@ -44,7 +44,6 @@ class NameSchema(BaseVocabularySchema, ModePIDFieldVocabularyMixin):
     affiliations = fields.List(fields.Nested(AffiliationRelationSchema))
     props = fields.Dict(keys=fields.Str(), values=fields.Raw())
 
-
     @validates_schema
     def validate_names(self, data, **kwargs):
         """Validate names."""

invenio_vocabularies/services/generators.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2024 CERN.
+#
+# Invenio-Vocabularies is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+#
+
+"""Vocabulary generators."""
+
+from invenio_access import any_user, authenticated_user
+from invenio_records_permissions.generators import Generator
+from invenio_search.engine import dsl
+
+
+class IfTags(Generator):
+    """Generator to filter based on tags.
+
+    This generator will filter records based on the tags field.
+    Optionally, it can be configured to only allow authenticated users.
+    """
+
+    def __init__(self, include=None, exclude=None, only_authenticated=False):
+        """Constructor."""
+        self.include = include or []
+        self.exclude = exclude or []
+        self.only_authenticated = only_authenticated
+
+    def needs(self, **kwargs):
+        """Enabling Needs."""
+        return [authenticated_user] if self.only_authenticated else [any_user]
+
+    def query_filter(self, **kwargs):
+        """Search based on configured tags."""
+        must_clauses = []
+        must_not_clauses = []
+
+        if self.include:
+            must_clauses.append(dsl.Q("terms", tags=self.include))
+
+        if self.exclude:
+            must_not_clauses.append(dsl.Q("terms", tags=self.exclude))
+
+        return dsl.Q(
+            "bool",
+            must=must_clauses,
+            must_not=must_not_clauses,
+        )

invenio_vocabularies/services/permissions.py

@@ -9,17 +9,19 @@
 """Vocabulary permissions."""
 
 from invenio_records_permissions import RecordPermissionPolicy
-from invenio_records_permissions.generators import AnyUser, SystemProcess
+from invenio_records_permissions.generators import SystemProcess
+
+from invenio_vocabularies.services.generators import IfTags
 
 
 class PermissionPolicy(RecordPermissionPolicy):
     """Permission policy."""
 
-    can_search = [SystemProcess(), AnyUser()]
-    can_read = [SystemProcess(), AnyUser()]
+    can_search = [SystemProcess(), IfTags(exclude=["unlisted"])]
+    can_read = [SystemProcess(), IfTags(exclude=["unlisted"])]
     can_create = [SystemProcess()]
     can_update = [SystemProcess()]
     can_delete = [SystemProcess()]
     can_manage = [SystemProcess()]
     # this permission is needed for the /api/vocabularies/ endpoint
-    can_list_vocabularies = [SystemProcess(), AnyUser()]
+    can_list_vocabularies = [SystemProcess(), IfTags(exclude=["unlisted"])]

tests/conftest.py

@@ -37,7 +37,12 @@
 from flask_principal import Identity, Need, UserNeed
 from flask_security import login_user
 from flask_security.utils import hash_password
-from invenio_access.permissions import ActionUsers, any_user, system_process
+from invenio_access.permissions import (
+    ActionUsers,
+    any_user,
+    superuser_access,
+    system_process,
+)
 from invenio_access.proxies import current_access
 from invenio_accounts.proxies import current_datastore
 from invenio_accounts.testutils import login_user_via_session
@@ -113,6 +118,17 @@ def identity():
     return i
 
 
+@pytest.fixture(scope="module")
+def superuser_identity():
+    """Super user identity to interact with the services."""
+    i = Identity(2)
+    i.provides.add(UserNeed(2))
+    i.provides.add(any_user)
+    i.provides.add(system_process)
+    i.provides.add(superuser_access)
+    return i
+
+
 @pytest.fixture(scope="module")
 def service(app):
     """Vocabularies service object."""
@@ -151,6 +167,14 @@ def lang_data2(lang_data):
     return data
 
 
+@pytest.fixture()
+def non_searchable_lang_data(lang_data):
+    """Example data for testing unlisted cases."""
+    data = dict(lang_data)
+    data["tags"] = ["unlisted", "recommended"]
+    return data
+
+
 @pytest.fixture()
 def example_record(db, identity, service, example_data):
     """Example record."""

tests/contrib/names/conftest.py

@@ -13,8 +13,11 @@
 """
 
 import pytest
+from invenio_indexer.api import RecordIndexer
 from invenio_records_resources.proxies import current_service_registry
 
+from invenio_vocabularies.contrib.names.api import Name
+
 
 @pytest.fixture(scope="module")
 def service():
@@ -56,3 +59,29 @@ def name_full_data():
         ],
         "affiliations": [{"id": "cern"}, {"name": "CustomORG"}],
     }
+
+
+@pytest.fixture(scope="function")
+def non_searchable_name_data():
+    """Full name data."""
+    return {
+        "id": "0000-0001-8135-3489",
+        "name": "Doe, John",
+        "given_name": "John",
+        "family_name": "Doe",
+        "identifiers": [
+            {"identifier": "0000-0001-8135-3489", "scheme": "orcid"},
+            {"identifier": "gnd:4079154-3", "scheme": "gnd"},
+        ],
+        "affiliations": [{"id": "cern"}, {"name": "CustomORG"}],
+        "tags": ["unlisted"],
+    }
+
+
+@pytest.fixture(scope="module")
+def indexer():
+    """Indexer instance with correct Record class."""
+    return RecordIndexer(
+        record_cls=Name,
+        record_to_index=lambda r: (r.__class__.index._name, "_doc"),
+    )

tests/contrib/names/test_name_permissions.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2024 CERN.
+#
+# Invenio-Vocabularies is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Test the names vocabulary permissions."""
+
+import pytest
+from flask_principal import Identity
+from invenio_access.permissions import any_user, authenticated_user
+from invenio_records_resources.services.errors import PermissionDeniedError
+
+
+#
+# Fixtures
+#
+@pytest.fixture()
+def anyuser_idty():
+    """Simple identity to interact with the service."""
+    identity = Identity(1)
+    identity.provides.add(any_user)
+    return identity
+
+
+@pytest.fixture()
+def authenticated_user_idty():
+    """Authenticated identity to interact with the service."""
+    identity = Identity(2)
+    identity.provides.add(authenticated_user)
+    return identity
+
+
+def test_non_searchable_tag(
+    app,
+    service,
+    identity,
+    non_searchable_name_data,
+    anyuser_idty,
+    authenticated_user_idty,
+    example_affiliation,
+    superuser_identity,
+    indexer,
+):
+    """Test that unlisted tags are not returned in search results."""
+    # Service
+    assert service.id == "names"
+    assert service.config.indexer_queue_name == "names"
+    # Create it
+    item = service.create(identity, non_searchable_name_data)
+    id_ = item.id
+
+    # Index document in ES
+    assert indexer.refresh()
+
+    with pytest.raises(PermissionDeniedError):
+        # Read - only searchable values should be returned
+        res = service.search(anyuser_idty, type="names", q=f"id:{id_}", size=25, page=1)
+
+    # Search - only searchable values should be returned
+    res = service.search(
+        authenticated_user_idty, type="names", q=f"id:{id_}", size=25, page=1
+    )
+    assert res.total == 0
+
+    # Admins should be able to see the unlisted tags
+    res = service.search(
+        superuser_identity, type="names", q=f"id:{id_}", size=25, page=1
+    )
+    assert res.total == 1

tests/contrib/names/test_names_api.py

@@ -12,7 +12,6 @@
 from functools import partial
 
 import pytest
-from invenio_indexer.api import RecordIndexer
 from invenio_records.systemfields.relations.errors import InvalidRelationValue
 from invenio_search import current_search_client
 from jsonschema import ValidationError as SchemaValidationError
@@ -26,15 +25,6 @@ def search_get():
     return partial(current_search_client.get, Name.index._name)
 
 
-@pytest.fixture()
-def indexer():
-    """Indexer instance with correct Record class."""
-    return RecordIndexer(
-        record_cls=Name,
-        record_to_index=lambda r: (r.__class__.index._name, "_doc"),
-    )
-
-
 @pytest.fixture()
 def example_name(db, name_full_data, example_affiliation):
     """Example name."""

tests/services/test_permissions.py

@@ -68,3 +68,23 @@ def test_permissions_readonly(anyuser_idty, lang_type, lang_data, service):
     with pytest.raises(PermissionDenied):
         service.delete(anyuser_idty, ("languages", id_))
     service.delete(system_identity, ("languages", id_))
+
+
+def test_non_searchable_tag(
+    anyuser_idty, lang_type, non_searchable_lang_data, service, superuser_identity
+):
+    """Test that unlisted tags are not returned in search results."""
+    item = service.create(system_identity, non_searchable_lang_data)
+    id_ = item.id
+    # Refresh index to make changes live.
+    Vocabulary.index.refresh()
+
+    # Search - only searchable values should be returned
+    res = service.search(anyuser_idty, type="languages", q=f"id:{id_}", size=25, page=1)
+    assert res.total == 0
+
+    # Admins should be able to see the unlisted tags
+    res = service.search(
+        superuser_identity, type="languages", q=f"id:{id_}", size=25, page=1
+    )
+    assert res.total == 1