From 6fafcaca825aed7f75dfb09af60f875554a585ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicolas=20=C3=89vrard?= Date: Fri, 11 Sep 2026 17:53:07 +0200 Subject: [PATCH] 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 | 3 + trytond/trytond/ir/resource.py | 107 +++++++++++---------- trytond/trytond/tests/test_resource.py | 126 ++++++++++++++++--------- 3 files changed, 141 insertions(+), 95 deletions(-) diff --git a/trytond/trytond/ir/message.xml b/trytond/trytond/ir/message.xml index 4c43d3f037..d1ea423a59 100644 --- a/trytond/trytond/ir/message.xml +++ b/trytond/trytond/ir/message.xml @@ -414,6 +414,9 @@ this repository contains the full copyright notices and license terms. --> You are not allowed to execute report "%(report)s". + + You are not allowed to search on the "%(resource)s" resource. + Invalid subject in email template "%(template)s" with exception "%(exception)s". diff --git a/trytond/trytond/ir/resource.py b/trytond/trytond/ir/resource.py index d0d092783f..28fc88d793 100644 --- a/trytond/trytond/ir/resource.py +++ b/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): diff --git a/trytond/trytond/tests/test_resource.py b/trytond/trytond/tests/test_resource.py index c423edfc2c..6ee045dc54 100644 --- a/trytond/trytond/tests/test_resource.py +++ b/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')])