Use the default search implementation on ResourceAccessMixin

The access to search is now guarded by a method allowing only administrator to
access it.

Closes #14817

trytond/trytond/ir/message.xml

@@ -414,6 +414,9 @@ this repository contains the full copyright notices and license terms. -->
         <record model="ir.message" id="msg_access_report_error">
             <field name="text">You are not allowed to execute report "%(report)s".</field>
         </record>
+        <record model="ir.message" id="msg_access_resource_search_error">
+            <field name="text">You are not allowed to search on the "%(resource)s" resource.</field>
+        </record>
         <record model="ir.message" id="msg_email_template_invalid_subject">
             <field name="text">Invalid subject in email template "%(template)s" with exception "%(exception)s".</field>
         </record>

trytond/trytond/ir/resource.py

@@ -6,11 +6,12 @@ from collections import defaultdict
 from sql.conditionals import Coalesce
 from sql.functions import DateTrunc
 
-from trytond.i18n import lazy_gettext
-from trytond.model import (
-    Index, Model, ModelSQL, ModelStorage, ModelView, fields)
+from trytond.i18n import gettext, lazy_gettext
+from trytond.model import Index, ModelSQL, ModelStorage, ModelView, fields
+from trytond.model.exceptions import AccessError
 from trytond.pool import Pool
 from trytond.pyson import Eval
+from trytond.tools.domain_inversion import is_leaf
 from trytond.transaction import Transaction, without_check_access
 
 __all__ = ['ResourceAccessMixin', 'ResourceMixin', 'resource_copy']
@@ -79,59 +80,61 @@ class ResourceAccessMixin(ModelStorage):
         return [
             (model, {'create': 'write', 'delete': 'write'}.get(mode, mode))]
 
+    @classmethod
+    def check_search_access(cls, domain):
+        pool = Pool()
+        ModelAccess = pool.get('ir.model.access')
+        Rule = pool.get('ir.rule')
+        User = pool.get('res.user')
+
+        def check_ressource_in_domain(domain):
+            if not domain:
+                return False
+            elif domain[0] == 'OR':
+                return False
+            elif domain[0] == 'AND':
+                return check_ressource_in_domain(domain[1:])
+            elif is_leaf(domain):
+                return False
+
+            for expr in domain:
+                match expr:
+                    case ('resource', '=', resource):
+                        model, id_ = resource.split(',', 1)
+                        id_ = int(id_)
+                        ModelAccess.check(model, mode='read')
+                        Rule.check(model, [id_], mode='read')
+                        return True
+                    case ('resource', 'in', resources):
+                        models = defaultdict(list)
+                        for resource in resources:
+                            model, id_ = resource.split(',', 1)
+                            id_ = int(id_)
+                            models[model].append(id_)
+                        for model, ids in models.items():
+                            ModelAccess.check(model, mode='read')
+                            Rule.check(model, ids, mode='read')
+                        return True
+                    case _:
+                        if check_ressource_in_domain(expr):
+                            return True
+            return False
+
+        if (Transaction().check_access
+                and not User.is_administrator()
+                and not check_ressource_in_domain(domain)):
+            raise AccessError(gettext(
+                    'ir.msg_access_resource_search_error',
+                    resource=cls.__name__))
+
     @classmethod
     def search(
             cls, domain, offset=0, limit=None, order=None, count=False,
             **kwargs):
-        transaction = Transaction()
-        enforce_access = (
-            not kwargs.get('query', False)
-            and transaction.user and transaction.check_access)
-        result = super().search(
-            domain, offset=offset, limit=limit, order=order,
-            count=False if enforce_access else count, **kwargs)
-        if not enforce_access:
-            return result
-
-        loop = 0
-        fetched = []
-        while True:
-            records = result
-            resources = defaultdict(set)
-            allowed = set()
-            with without_check_access():
-                records = cls.browse(records)
-            for record in records:
-                if isinstance(record.resource, Model):
-                    resources[record.resource.__class__].add(
-                        record.resource.id)
-
-            for RModel, ids in resources.items():
-                allowed.update(RModel.search([
-                            ('id', 'in', ids),
-                            ]))
-
-            fetched.extend([
-                r for r in records
-                if not r.resource or r.resource in allowed])
-
-            if limit is None or len(fetched) >= limit:
-                if limit is not None:
-                    fetched = fetched[:limit]
-                break
-
-            loop += 1
-            result = super().search(
-                domain, offset=offset + loop * limit, limit=limit, order=order,
-                count=False)
-            if not result:
-                break
-
-        if count:
-            return len(fetched)
-        else:
-            # re-browse to have same context
-            return cls.browse(fetched)
+        cls.check_search_access(domain)
+        return super().search(
+            domain, offset=offset, limit=limit, order=order, count=count,
+            **kwargs)
 
     @classmethod
     def read(cls, ids, fields_names):

trytond/trytond/tests/test_resource.py

@@ -78,31 +78,6 @@ class ResourceTestCase(DBTestCase):
         note = Note(note.id)
         self.assertEqual(user_note.write_date, write_date)
 
-    @with_transaction()
-    def test_resources_rule(self):
-        "Test resources rules are applied on search"
-        pool = Pool()
-        Note = pool.get('ir.note')
-        Warning = pool.get('res.user.warning')
-
-        warning1 = Warning(user=0, name="root")
-        warning1.save()
-        note1 = Note(resource=warning1)
-        note1.save()
-        warning2 = Warning(user=1, name="admin")
-        warning2.save()
-        note2 = Note(resource=warning2)
-        note2.save()
-
-        with Transaction().set_context(_check_access=True):
-            notes = Note.search([])
-            count = Note.search([], count=True)
-            query = Note.search([], query=True)
-
-            self.assertEqual(notes, [note2])
-            self.assertEqual(count, 1)
-            self.assertTrue(query)
-
     @with_transaction()
     def test_resource_with_access(self):
         "Test create/write/read/delete on resource with access"
@@ -251,27 +226,92 @@ class ResourceTestCase(DBTestCase):
             with self.assertRaises(AccessError):
                 Note.delete([note])
 
-    @with_transaction()
-    def test_resources_search_limit(self):
-        "Test resource search limit work as expected"
+    @with_transaction(context={'_check_access': True})
+    def test_resource_search_with_access(self):
+        "Test searching on the resource when the user has access"
         pool = Pool()
         Note = pool.get('ir.note')
-        Warning_ = pool.get('res.user.warning')
+        Resource = pool.get('test.resource')
+        ModelAccess = pool.get('ir.model.access')
+        User = pool.get('res.user')
 
-        warning1 = Warning_(user=0, name="root")
-        warning1.save()
-        warning2 = Warning_(user=1, name="admin")
-        warning2.save()
-        for i in range(100):
-            note = Note(resource=warning2 if i % 3 else warning1)
-            note.save()
+        user = User(login='foo')
+        user.save()
+        ModelAccess.create([{
+                    'model': Resource.__name__,
+                    'group': None,
+                    'perm_write': True,
+                    'perm_read': True,
+                    }])
+        record1, record2 = Resource.create([{}, {}])
+        note, = Note.create([{
+                    'resource': record1,
+                    'message': "Foo",
+                    }])
 
-        with Transaction().set_context(_check_access=True):
-            notes = Note.search([], limit=10, offset=0)
-            self.assertEqual(len(notes), 10)
+        with Transaction().set_user(user.id):
+            notes = Note.search([
+                    ('message', '=', 'Foo'),
+                    ('resource', '=', str(record1)),
+                    ])
+            self.assertEqual([note], notes)
 
-            notes = Note.search([], limit=200, offset=0)
-            self.assertEqual(len(notes), 66)
+            notes = Note.search([
+                    ('message', '=', 'Foo'),
+                    ('resource', 'in', [str(record1), str(record2)]),
+                    ])
+            self.assertEqual([note], notes)
 
-            notes = Note.search([])
-            self.assertEqual(len(notes), 66)
+            notes = Note.search([
+                    ('message', '=', 'Foo'),
+                    ('resource', '=', str(record2)),
+                    ])
+            self.assertEqual([], notes)
+
+            with self.assertRaises(AccessError):
+                Note.search([('message', '=', 'Foo')])
+
+    @with_transaction(context={'_check_access': True})
+    def test_resource_search_without_access(self):
+        "Test searching on the resource when the user doesn't have access"
+        pool = Pool()
+        Note = pool.get('ir.note')
+        Resource = pool.get('test.resource')
+        ModelAccess = pool.get('ir.model.access')
+        User = pool.get('res.user')
+
+        user = User(login='foo')
+        user.save()
+        ModelAccess.create([{
+                    'model': Resource.__name__,
+                    'group': None,
+                    'perm_write': False,
+                    'perm_read': False,
+                    }])
+        record1, record2 = Resource.create([{}, {}])
+        note, = Note.create([{
+                    'resource': record1,
+                    'message': "Foo",
+                    }])
+
+        with Transaction().set_user(user.id):
+            with self.assertRaises(AccessError):
+                Note.search([
+                        ('message', '=', 'Foo'),
+                        ('resource', '=', str(record1)),
+                        ])
+
+            with self.assertRaises(AccessError):
+                Note.search([
+                        ('message', '=', 'Foo'),
+                        ('resource', 'in', [str(record1), str(record2)]),
+                        ])
+
+            with self.assertRaises(AccessError):
+                Note.search([
+                        ('message', '=', 'Foo'),
+                        ('resource', '=', str(record2)),
+                        ])
+
+            with self.assertRaises(AccessError):
+                Note.search([('message', '=', 'Foo')])