feat: SameAs generator

* Changed permission policy's context to
  propagate policy instance to generators
* Added CompositeGenerator class
* Added implementation of SameAs with tests

Co-authored-by: Ronald Krist <ronald.krist@cesnet.cz>

Author: mesemus

invenio_records_permissions/generators.py

@@ -3,6 +3,7 @@
 # Copyright (C) 2019-2023 CERN.
 # Copyright (C) 2019-2020 Northwestern University.
 # Copyright (C) 2024 Ubiquity Press.
+# Copyright (C) 2026 CESNET z.s.p.o.
 #
 # Invenio-Records-Permissions is free software; you can redistribute it
 # and/or modify it under the terms of the MIT License; see LICENSE file for
@@ -11,12 +12,12 @@
 """Invenio Records Permissions Generators."""
 
 import operator
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from functools import reduce
 from itertools import chain
 
 from flask import current_app
-from flask_principal import ActionNeed, UserNeed
+from flask_principal import UserNeed
 from invenio_access import ActionRoles, ActionUsers, Permission
 from invenio_access.permissions import (
     any_user,
@@ -225,7 +226,7 @@ def query_filter(self, identity=None, **kwargs):
                         "id": id_need.value,
                         # TODO: Implement other schemes
                     }
-                }
+                },
             )
             for access_level in read_levels
         ]
@@ -339,3 +340,132 @@ def _condition(self, **_):
 # |       False     |       False      |     Open     |  True  |
 # |-----------------|------------------|--------------|--------|
 #
+
+
+class CompositeGenerator(Generator, ABC):
+    """Base class for generators that compose multiple generators.
+
+    This generator implements the Composite design pattern, allowing you to combine
+    multiple generators and treat them as a single generator. Subclasses must implement
+    the ``_generators(**context)`` method to return the list of generators to compose.
+
+    Behavior:
+        - ``needs``: Combines (flattens) the needs from all composed generators
+        - ``excludes``: Combines (flattens) the excludes from all composed generators
+        - ``query_filter``: Combines query filters using OR logic (any filter matches)
+    """
+
+    @abstractmethod
+    def _generators(self, **context):
+        """Return the list of generators to compose.
+
+        Must be implemented by subclasses.
+
+        :param context: Context dictionary that may contain record, etc.
+        :returns: List of Generator instances to compose
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def needs(self, **context):
+        """Get enabling needs from all composed generators.
+
+        Combines needs from all generators into a single flattened list.
+        """
+        needs = [
+            generator.needs(**context) for generator in self._generators(**context)
+        ]
+        return list(chain.from_iterable(needs))
+
+    def excludes(self, **context):
+        """Get preventing needs from all composed generators.
+
+        Combines excludes from all generators into a single flattened list.
+        """
+        excludes = [
+            generator.excludes(**context) for generator in self._generators(**context)
+        ]
+        return list(chain.from_iterable(excludes))
+
+    def query_filter(self, **context):
+        """Get search filters from all composed generators.
+
+        Combines query filters from all generators using OR logic. This means a record
+        matches if it satisfies ANY of the composed generator's filters.
+
+        :returns: Combined query using OR, or match_none if no generators provide filters
+        """
+        generators = self._generators(**context)
+
+        queries = [g.query_filter(**context) for g in generators]
+        queries = [q for q in queries if q]
+        if not queries:
+            return dsl.Q("match_none")
+
+        return reduce(operator.or_, queries) if queries else None
+
+
+class SameAs(CompositeGenerator):
+    """Generator that delegates permissions to another permission on the same policy.
+
+    This generator allows you to reuse the permission configuration from one action
+    for another action, promoting DRY (Don't Repeat Yourself) principles. It dynamically
+    retrieves the generators from the specified permission attribute on the policy.
+
+    This is particularly useful when:
+        - Multiple actions should have identical permission requirements
+        - You want permission inheritance when subclassing policies
+        - You need to maintain a single source of truth for related permissions
+
+    Example:
+        .. code-block:: python
+
+            class RecordPermissionPolicy(BasePermissionPolicy):
+                can_edit = [RecordOwners(), AdminAction("admin-access")]
+                can_delete = [SameAs("can_edit")]  # Delegates to can_edit
+                can_create_files = [SameAs("can_edit")]  # Also delegates to can_edit
+
+        In this example, ``can_delete`` and ``can_create_files`` will have the exact
+        same permissions as ``can_edit``. If you later modify ``can_edit`` or override
+        it in a subclass, the delegating permissions automatically inherit those changes.
+
+    Note:
+        The permission name must be an attribute on the policy instance and must
+        contain a list of generators.
+    """
+
+    def __init__(self, permission_name: str) -> None:
+        """Initialize the generator.
+
+        :param permission_name: Name of the permission attribute to delegate to.
+            Must be in the format "can_<action>" (e.g., "can_edit", "can_read").
+            This attribute must exist on the permission policy and contain a list of generators.
+        """
+        self._delegated_permission_name = permission_name
+
+    def _generators(self, **context):
+        """Get the generators from the delegated permission on the policy.
+
+        :param context: Must contain 'permission_policy' key with the policy instance
+        :returns: List of generators from the delegated permission
+        """
+        policy = context["permission_policy"]
+        return getattr(policy, self._delegated_permission_name)
+
+    def __add__(self, other):
+        """Allow adding another generator or list of generators to this SameAs generator.
+
+        Example:
+            can_delete = SameAs("can_edit") + [RecordOwners()]
+        """
+        if isinstance(other, (tuple, list, set)):
+            return [self] + list(other)
+        else:
+            return [self, other]
+
+    def __iter__(self):
+        """Let the SameAs be used as the sole element inside the can_abc properties.
+
+        Example:
+            can_delete = SameAs("can_edit")
+        """
+        return iter([self])

invenio_records_permissions/policies/base.py

@@ -58,7 +58,12 @@ def __init__(self, action, **over):
         """Constructor."""
         super().__init__()
         self.action = action
-        self.over = over
+
+        # adding self to the permission policy context for generators to use
+        self.over = {
+            **over,
+            "permission_policy": self,
+        }
 
     @property
     def generators(self):

invenio_records_permissions/policies/records.py

@@ -14,7 +14,14 @@
 from werkzeug.utils import import_string
 
 from ..errors import UnknownGeneratorError
-from ..generators import AnyUser, AnyUserIfPublic, Disable, RecordOwners, SystemProcess
+from ..generators import (
+    AnyUser,
+    AnyUserIfPublic,
+    Disable,
+    RecordOwners,
+    SameAs,
+    SystemProcess,
+)
 from .base import BasePermissionPolicy
 
 
@@ -43,7 +50,14 @@ def obj_or_import_string(value, default=None):
 
 
 class RecordPermissionPolicy(BasePermissionPolicy):
-    """Access control configuration for records."""
+    """Access control configuration for records.
+
+    can_read and can_update are the main permissions.
+    can_delete, can_update_files are delegating to can_update,
+    but can be overridden if needed.
+    can_read_files is delegating to can_read, but can be overridden if needed.
+    )
+    """
 
     NEED_LABEL_TO_ACTION = {
         "bucket-update": "update_files",
@@ -57,15 +71,15 @@ class RecordPermissionPolicy(BasePermissionPolicy):
     # be used.
     can_create = [Disable()]
     # Read access given to everyone if public record/files and owners always.
-    can_read = [AnyUserIfPublic(), RecordOwners()]
+    can_read = [AnyUserIfPublic(), SameAs("can_update")]
     # Update access given to record owners.
     can_update = [RecordOwners()]
     # Delete access given to superuser-access action only
     # (superuser-access is added by default by base policy)
     can_delete = []
     # Associated files permissions (which are really bucket permissions)
-    can_read_files = [AnyUserIfPublic(), RecordOwners()]
-    can_update_files = [RecordOwners()]
+    can_read_files = [SameAs("can_read")]
+    can_update_files = [SameAs("can_update")]
     can_read_deleted_files = []
     can_create_or_update_many = [SystemProcess()]
 

tests/test_composite_generator.py

@@ -0,0 +1,169 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2026 CESNET z.s.p.o.
+#
+# Invenio-Records-Permissions 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 CompositeGenerator."""
+
+import json
+
+from flask_principal import Need
+from invenio_access.permissions import any_user, authenticated_user
+
+from invenio_records_permissions.generators import (
+    AnyUser,
+    AnyUserIfPublic,
+    AuthenticatedUser,
+    CompositeGenerator,
+    Disable,
+    RecordOwners,
+)
+
+
+class StaticCompositeGenerator(CompositeGenerator):
+    """Test implementation that returns a static list of generators."""
+
+    def __init__(self, generators):
+        """Constructor."""
+        self._generator_list = generators
+
+    def _generators(self, **context):
+        """Return the static list of generators."""
+        return self._generator_list
+
+
+def stable_dict_sort(data):
+    """Recursively sort dictionaries and lists for stable comparison.
+
+    Dictionaries are re-created with sorted keys, and lists are sorted by their JSON representation.
+    This ensures that we can compare the output of query_filter regardless
+    of the order of keys in dictionaries or items in lists.
+
+    Note: this is intended just for tests as can be slow due to the multiple
+    json dumps for each comparison.
+    """
+    if isinstance(data, dict):
+        return {k: stable_dict_sort(v) for k, v in sorted(data.items())}
+    elif isinstance(data, list):
+        deep_sorted = [stable_dict_sort(x) for x in data]
+        return sorted(deep_sorted, key=lambda x: json.dumps(x, sort_keys=True))
+    else:
+        return data
+
+
+def test_composite_generator_empty_list():
+    """Test CompositeGenerator with empty generator list."""
+    generator = StaticCompositeGenerator([])
+
+    assert generator.needs() == []
+    assert generator.excludes() == []
+    assert generator.query_filter().to_dict() == {"match_none": {}}
+
+
+def test_composite_generator_single_generator():
+    """Test CompositeGenerator with a single generator."""
+    generator = StaticCompositeGenerator([AnyUser()])
+
+    assert generator.needs() == [any_user]
+    assert generator.excludes() == []
+    assert generator.query_filter().to_dict() == {"match_all": {}}
+
+
+def test_composite_generator_multiple_generators_needs(create_record):
+    """Test CompositeGenerator combines needs from multiple generators."""
+    from flask_principal import UserNeed
+
+    record = create_record()
+    generator = StaticCompositeGenerator([AnyUser(), RecordOwners()])
+
+    needs = generator.needs(record=record)
+    assert any_user in needs
+    assert UserNeed(1) in needs
+    assert UserNeed(2) in needs
+    assert UserNeed(3) in needs
+    assert len(needs) == 4
+
+
+def test_composite_generator_multiple_generators_excludes():
+    """Test CompositeGenerator combines excludes from multiple generators."""
+    generator = StaticCompositeGenerator([Disable()])
+
+    excludes = generator.excludes()
+    assert excludes == [any_user]
+
+
+def test_composite_generator_query_filter_or_logic(mocker):
+    """Test CompositeGenerator combines query filters with OR logic."""
+    generator = StaticCompositeGenerator([AnyUserIfPublic(), RecordOwners()])
+
+    # Test with identity
+    identity = mocker.Mock(provides=[Need(method="id", value=1)])
+    query_filter = generator.query_filter(identity=identity)
+
+    query_dict = query_filter.to_dict()
+    assert stable_dict_sort(query_dict) == {
+        "bool": {
+            "should": [
+                {"term": {"_access.metadata_restricted": False}},
+                {"term": {"owners": 1}},
+            ]
+        }
+    }
+
+
+def test_composite_generator_query_filter_single_result():
+    """Test CompositeGenerator with only one generator providing query filter."""
+    generator = StaticCompositeGenerator([AnyUser()])
+
+    query_filter = generator.query_filter()
+    assert query_filter.to_dict() == {"match_all": {}}
+
+
+def test_composite_generator_query_filter_all_empty(mocker):
+    """Test CompositeGenerator when no generator provides query filter."""
+
+    class EmptyFilterGenerator(AnyUser):
+        def query_filter(self, **kwargs):
+            return []
+
+    generator = StaticCompositeGenerator([EmptyFilterGenerator()])
+    query_filter = generator.query_filter()
+    assert query_filter.to_dict() == {"match_none": {}}
+
+
+def test_composite_generator_with_context(create_record):
+    """Test CompositeGenerator passes context to all generators."""
+    from flask_principal import UserNeed
+
+    record = create_record()
+    generator = StaticCompositeGenerator([RecordOwners(), AnyUserIfPublic()])
+
+    needs = generator.needs(record=record)
+    assert UserNeed(1) in needs
+    assert UserNeed(2) in needs
+    assert UserNeed(3) in needs
+    assert any_user in needs
+
+
+def test_composite_generator_mixed_generators(create_record, mocker):
+    """Test CompositeGenerator with various generator types."""
+    from flask_principal import UserNeed
+
+    record = create_record()
+    generator = StaticCompositeGenerator(
+        [AnyUser(), RecordOwners(), AuthenticatedUser()]
+    )
+
+    # Test needs
+    needs = generator.needs(record=record)
+    assert any_user in needs
+    assert authenticated_user in needs
+    assert UserNeed(1) in needs
+    assert UserNeed(2) in needs
+    assert UserNeed(3) in needs
+
+    # Test excludes
+    excludes = generator.excludes(record=record)
+    assert excludes == []