details: https://code.tryton.org/tryton/commit/aa9ae9c04e8c
branch: default
user: Nicolas Évrard <[email protected]>
date: Fri Sep 11 17:53:07 2026 +0200
description:
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
diffstat:
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(-)
diffs (288 lines):
diff -r 6ebedcf8826e -r aa9ae9c04e8c trytond/trytond/ir/message.xml
--- a/trytond/trytond/ir/message.xml Sun Sep 13 17:28:32 2026 +0200
+++ b/trytond/trytond/ir/message.xml Fri Sep 11 17:53:07 2026 +0200
@@ -414,6 +414,9 @@
<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>
diff -r 6ebedcf8826e -r aa9ae9c04e8c trytond/trytond/ir/resource.py
--- a/trytond/trytond/ir/resource.py Sun Sep 13 17:28:32 2026 +0200
+++ b/trytond/trytond/ir/resource.py Fri Sep 11 17:53:07 2026 +0200
@@ -6,11 +6,12 @@
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']
@@ -80,58 +81,60 @@
(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 -r 6ebedcf8826e -r aa9ae9c04e8c trytond/trytond/tests/test_resource.py
--- a/trytond/trytond/tests/test_resource.py Sun Sep 13 17:28:32 2026 +0200
+++ b/trytond/trytond/tests/test_resource.py Fri Sep 11 17:53:07 2026 +0200
@@ -79,31 +79,6 @@
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"
pool = Pool()
@@ -251,27 +226,92 @@
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')
+
+ 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",
+ }])
- 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()
+ with Transaction().set_user(user.id):
+ notes = Note.search([
+ ('message', '=', 'Foo'),
+ ('resource', '=', str(record1)),
+ ])
+ self.assertEqual([note], notes)
+
+ notes = Note.search([
+ ('message', '=', 'Foo'),
+ ('resource', 'in', [str(record1), str(record2)]),
+ ])
+ self.assertEqual([note], notes)
+
+ notes = Note.search([
+ ('message', '=', 'Foo'),
+ ('resource', '=', str(record2)),
+ ])
+ self.assertEqual([], notes)
+
+ with self.assertRaises(AccessError):
+ Note.search([('message', '=', 'Foo')])
- with Transaction().set_context(_check_access=True):
- notes = Note.search([], limit=10, offset=0)
- self.assertEqual(len(notes), 10)
+ @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')
- notes = Note.search([], limit=200, offset=0)
- self.assertEqual(len(notes), 66)
+ 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",
+ }])
- notes = Note.search([])
- self.assertEqual(len(notes), 66)
+ 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')])
