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

@@ -23,7 +23,8 @@
 from invenio_records_resources.records.systemfields import ModelPIDField, PIDRelation
 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 .config import AwardsSearchOptions, service_components
 from .schema import AwardSchema

invenio_vocabularies/contrib/funders/funders.py

@@ -21,7 +21,9 @@
 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/names.py

@@ -21,7 +21,8 @@
 )
 from invenio_records_resources.resources.records.headers import etag_headers
 
-from ...services.permissions import PermissionPolicy
+from invenio_vocabularies.services.permissions import PermissionPolicy
+
 from ..affiliations.api import Affiliation
 from .config import NamesSearchOptions, service_components
 from .schema import NameSchema

invenio_vocabularies/contrib/names/permissions.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2020-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 permissions."""
+
+from invenio_records_permissions.generators import SystemProcess, AuthenticatedUser
+
+from invenio_vocabularies.services.generators import Tags
+from invenio_vocabularies.services.permissions import PermissionPolicy
+
+
+class NamesPermissionPolicy(PermissionPolicy):
+ """Names permission policy."""
+
+ can_search = [
+ SystemProcess(),
+ Tags(exclude=["non-searchable"], only_authenticated=True),
+ ]
+ can_read = [SystemProcess(), AuthenticatedUser()]
+ # this permission is needed for the /api/vocabularies/ endpoint
+ can_list_vocabularies = [
+ SystemProcess(),
+ Tags(exclude=["non-searchable"], 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,60 @@
+# -*- 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 AnyUser(Generator):
+ """Allows any user."""
+
+ def needs(self, **kwargs):
+ """Enabling Needs."""
+ return [any_user]
+
+ def query_filter(self, **kwargs):
+ """Match only searchable values in search."""
+ return dsl.Q(
+ "bool",
+ must_not=[dsl.Q("term", tags="non-searchable")],
+ )
+
+
+class Tags(Generator):
+ """Allows any user."""
+
+ 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 Tags
 
 
 class PermissionPolicy(RecordPermissionPolicy):
  """Permission policy."""
 
- can_search = [SystemProcess(), AnyUser()]
- can_read = [SystemProcess(), AnyUser()]
+ can_search = [SystemProcess(), Tags(exclude=["non-searchable"])]
+ can_read = [SystemProcess(), Tags(exclude=["non-searchable"])]
  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(), Tags(exclude=["non-searchable"])]

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,
+ system_process,
+ superuser_access,
+)
 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 non-searchable cases."""
+ data = dict(lang_data)
+ data["tags"] = ["non-searchable", "recommended"]
+ return data
+
+
 @pytest.fixture()
 def example_record(db, identity, service, example_data):
  """Example record."""

tests/contrib/names/conftest.py

@@ -56,3 +56,20 @@ 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": ["non-searchable"],
+ }

tests/contrib/names/test_name_permissions.py

@@ -0,0 +1,59 @@
+# -*- 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
+
+from invenio_vocabularies.records.api import Vocabulary
+
+
+#
+# Fixtures
+#
+@pytest.fixture()
+def anyuser_idty():
+ """Simple identity to interact with the service."""
+ identity = Identity(1)
+ identity.provides.add(any_user)
+ return identity
+
+
+def test_non_searchable_tag(
+ app,
+ service,
+ identity,
+ non_searchable_name_data,
+ anyuser_idty,
+ example_affiliation,
+ superuser_identity,
+):
+ """Test that non-searchable 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
+
+ # Refresh index to make changes live.
+ Vocabulary.index.refresh()
+
+ # Search - only searchable values should be returned
+ res = service.search(anyuser_idty, type="names", q=f"id:{id_}", size=25, page=1)
+ assert res.total == 0
+
+ # Admins should be able to see the non-searchable tags
+ res = service.search(
+ superuser_identity, type="names", q=f"id:{id_}", size=25, page=1
+ )
+ assert res.total == 1

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 non-searchable 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 non-searchable tags
+ res = service.search(
+ superuser_identity, type="languages", q=f"id:{id_}", size=25, page=1
+ )
+ assert res.total == 1