Add ModelAccessProxy

#14907
(grafted from dafef28a1d65032628809acbfa130a2f2e74474f)

--HG--
branch : 8.0

modules/company/res.py

@@ -163,6 +163,10 @@ class User(metaclass=PoolMeta):
         Return an ordered tuple of company ids for the user
         '''
         transaction = Transaction()
+
+        if '_companies' in transaction.context:
+            return transaction.context['_companies']
+
         user_id = transaction.user
         companies = cls._get_companies_cache.get(user_id)
         if companies is not None:

trytond/CHANGELOG

@@ -1,3 +1,4 @@
+* Add ModelAccessProxy (issue14907)
 * Restrict Genshi evaluation (issue14869, issue5160)
 
 Version 8.0.5 - 2026-06-18

trytond/doc/ref/models.rst

@@ -882,6 +882,15 @@ Additional options are available:
      the string
 
 
+ModelAccessProxy
+================
+
+.. class:: ModelAccessProxy(record[, context])
+
+   A class proxying instance of :class:`ModelStorage` with check access using
+   ``context``.
+
+
 convert_from
 ------------
 

trytond/trytond/ir/model.py

@@ -588,10 +588,6 @@ class ModelAccess(
     @classmethod
     def get_access(cls, models):
         'Return access for models'
-        # root user above constraint
-        if Transaction().user == 0:
-            return defaultdict(lambda: defaultdict(lambda: True))
-
         pool = Pool()
         User = pool.get('res.user')
         cursor = Transaction().connection.cursor()
@@ -681,9 +677,8 @@ class ModelAccess(
         Model = pool.get(model_name)
         assert mode in ['read', 'write', 'create', 'delete'], \
             'Invalid access mode for security'
-        transaction = Transaction()
-        if (transaction.user == 0
-                or (raise_exception and not transaction.check_access)):
+
+        if not Transaction().check_access:
             return True
 
         User = pool.get('res.user')
@@ -828,11 +823,6 @@ class ModelFieldAccess(
     @classmethod
     def get_access(cls, models):
         'Return fields access for models'
-        # root user above constraint
-        if Transaction().user == 0:
-            return defaultdict(lambda: defaultdict(
-                    lambda: defaultdict(lambda: True)))
-
         pool = Pool()
         User = pool.get('res.user')
         field_access = cls.__table__()
@@ -878,8 +868,7 @@ class ModelFieldAccess(
         return accesses
 
     @classmethod
-    def check(cls, model_name, fields, mode='read', raise_exception=True,
-            access=False):
+    def check(cls, model_name, fields, mode='read', raise_exception=True):
         '''
         Check access for fields on model_name.
         '''
@@ -887,11 +876,8 @@ class ModelFieldAccess(
         Model = pool.get(model_name)
         assert mode in ('read', 'write', 'create', 'delete'), \
             'Invalid access mode'
-        transaction = Transaction()
-        if (transaction.user == 0
-                or (raise_exception and not transaction.check_access)):
-            if access:
-                return dict((x, True) for x in fields)
+
+        if not Transaction().check_access:
             return True
 
         User = pool.get('res.user')
@@ -899,8 +885,6 @@ class ModelFieldAccess(
 
         accesses = dict((f, a[mode])
             for f, a in cls.get_access([model_name])[model_name].items())
-        if access:
-            return accesses
         for field in fields:
             if not accesses.get(field, True):
                 if raise_exception:

trytond/trytond/ir/rule.py

@@ -10,7 +10,8 @@ from trytond.model import Check, Index, ModelSQL, ModelView, fields
 from trytond.model.exceptions import AccessError, ValidationError
 from trytond.pool import Pool
 from trytond.pyson import Eval, If, PYSONDecoder
-from trytond.transaction import Transaction, inactive_records
+from trytond.transaction import (
+    Transaction, inactive_records, without_check_access)
 
 
 class DomainError(ValidationError):
@@ -239,9 +240,6 @@ class Rule(ModelSQL, ModelView):
         model_names = list(model_names)
 
         cursor = transaction.connection.cursor()
-        # root user above constraint
-        if transaction.user == 0:
-            return {}, {}
         cursor.execute(*rule_table.join(rule_group,
                 condition=rule_group.id == rule_table.rule_group
                 ).join(rule_group_group, 'LEFT',
@@ -266,8 +264,8 @@ class Rule(ModelSQL, ModelView):
 
         clause = defaultdict(lambda: ['OR'])
         clause_global = defaultdict(lambda: ['OR'])
-        # Use root user without context to prevent recursion
-        with transaction.set_user(0), transaction.set_context(user=0):
+        # Without check access to prevent recursion
+        with without_check_access():
             rules = cls.browse(ids)
         for rule in rules:
             decoder = PYSONDecoder(
@@ -295,9 +293,8 @@ class Rule(ModelSQL, ModelView):
     @classmethod
     def domain_get(cls, model_name, mode='read'):
         pool = Pool()
-        transaction = Transaction()
-        # root user above constraint
-        if transaction.user == 0 or not transaction.check_access:
+
+        if not Transaction().check_access:
             return []
 
         assert mode in cls.modes
@@ -346,7 +343,9 @@ class Rule(ModelSQL, ModelView):
 
         def test_domain(ids, domain):
             # Use root to prevent infinite recursion
-            with transaction.set_user(0, set_context=True), inactive_records():
+            with transaction.set_user(0, set_context=True), \
+                    inactive_records(), \
+                    without_check_access():
                 records = Model.search([
                         ('id', 'in', ids),
                         domain,
@@ -354,6 +353,8 @@ class Rule(ModelSQL, ModelView):
             return list(set(ids).difference(map(int, records)))
 
         domain = cls.domain_get(model_name, mode=mode)
+        if not domain:
+            return
         forbidden = test_domain(ids, domain)
         if forbidden:
             ids = ', '.join(map(str, forbidden[:5]))

trytond/trytond/model/__init__.py

@@ -10,7 +10,7 @@ from .match import MatchMixin
 from .model import Model
 from .modelsingleton import ModelSingleton
 from .modelsql import Check, Exclude, Index, ModelSQL, Unique, convert_from
-from .modelstorage import EvalEnvironment, ModelStorage
+from .modelstorage import EvalEnvironment, ModelAccessProxy, ModelStorage
 from .modelview import ModelView
 from .multivalue import MultiValueMixin, ValueMixin
 from .order import sequence_ordered, sequence_reorder, sort
@@ -30,6 +30,7 @@ __all__ = [
     Index,
     MatchMixin,
     Model,
+    ModelAccessProxy,
     ModelSQL,
     ModelSingleton,
     ModelStorage,

trytond/trytond/model/modelstorage.py

@@ -30,7 +30,8 @@ from trytond.tools import (
 from trytond.tools.domain_inversion import domain_inversion, eval_domain
 from trytond.tools.domain_inversion import parse as domain_parse
 from trytond.transaction import (
-    Transaction, inactive_records, record_cache_size, without_check_access)
+    Transaction, check_access, inactive_records, record_cache_size,
+    without_check_access)
 
 from . import fields
 from .descriptors import dualmethod
@@ -1914,11 +1915,11 @@ class ModelStorage(Model):
 
         if load_eager or multiple_getter:
             FieldAccess = pool.get('ir.model.field.access')
-            fread_accesses = {}
-            fread_accesses.update(FieldAccess.check(self.__name__,
-                list(self._fields.keys()), 'read', access=True))
-            to_remove = set(x for x, y in fread_accesses.items()
-                    if not y and x != name)
+            fields_access = FieldAccess.get_access(
+                [self.__name__])[self.__name__]
+            to_remove = {
+                f for f, a in fields_access.items() if not a['read']}
+            to_remove.discard(name)
 
             def not_cached(item):
                 fname, field = item
@@ -2315,3 +2316,42 @@ def _record_eval_pyson(record, source, encoded=False):
 
 
 _pyson_encoder = PYSONEncoder()
+
+
+def ModelAccessProxy(record, context=None):
+    pool = Pool()
+    ModelAccess = pool.get('ir.model.access')
+    FieldAccess = pool.get('ir.model.field.access')
+    Rule = pool.get('ir.rule')
+
+    model = record.__class__
+    with Transaction().set_context(context), check_access():
+        ModelAccess.check(model.__name__)
+        Rule.check(model.__name__, [record.id])
+
+    class _ModelAccessProxy:
+        __class__ = model
+
+        def __init__(self, id):
+            self.id = id
+
+        def __getattr__(self, name):
+            if name in model._fields:
+                with Transaction().set_context(context), check_access():
+                    FieldAccess.check(model.__name__, [name])
+            value = getattr(record, name)
+            if isinstance(value, Model):
+                value = ModelAccessProxy(value, context)
+            elif isinstance(value, (list, tuple)):
+                value = [
+                    ModelAccessProxy(r, context)
+                    if isinstance(r, Model) else r for r in value]
+            return value
+
+        def __int__(self):
+            return int(self.id)
+
+        def __str__(self):
+            return f'{model.__name__},{self.id}'
+
+    return _ModelAccessProxy(record.id)

trytond/trytond/model/modelview.py

@@ -436,10 +436,9 @@ class ModelView(Model):
         tree_root = tree.getroottree().getroot()
 
         # Find field without read access
-        fread_accesses = FieldAccess.check(cls.__name__,
-                list(cls._fields.keys()), 'read', access=True)
-        fields_to_remove = set(
-            x for x, y in fread_accesses.items() if not y)
+        fields_access = FieldAccess.get_access([cls.__name__])[cls.__name__]
+        fields_to_remove = {
+            f for f, a in fields_access.items() if not a['read']}
 
         # Find relation field without read access
         for name, field in cls._fields.items():
@@ -666,8 +665,8 @@ class ModelView(Model):
             button_groups = Button.get_groups(cls.__name__, button_name)
             if ((button_groups and not groups & button_groups)
                     or (not button_groups
-                        and not ModelAccess.check(
-                            cls.__name__, 'write', raise_exception=False))):
+                        and not ModelAccess.get_access(
+                            [cls.__name__])[cls.__name__]['write'])):
                 states = states.copy()
                 states['readonly'] = True
             element.set('states', encoder.encode(states))
@@ -702,8 +701,8 @@ class ModelView(Model):
                 action = None
             if (not action
                     or not action.res_model
-                    or not ModelAccess.check(
-                        action.res_model, 'read', raise_exception=False)):
+                    or not ModelAccess.get_access(
+                        [action.res_model])[action.res_model]['read']):
                 element.tag = 'label'
                 colspan = element.attrib.get('colspan')
                 link_name = element.attrib['name']

trytond/trytond/res/user.py

@@ -617,6 +617,10 @@ class User(avatar_mixin(100, 'login'), DeactivableMixin, ModelSQL, ModelView):
         Group = pool.get('res.group')
 
         transaction = Transaction()
+
+        if '_groups' in transaction.context:
+            return transaction.context['_groups']
+
         user = transaction.user
         groups = cls._get_groups_cache.get(user)
         if groups is not None:

trytond/trytond/tests/test_access.py

@@ -2,6 +2,7 @@
 # This file is part of Tryton.  The COPYRIGHT file at the top level of
 # this repository contains the full copyright notices and license terms.
 
+from trytond.model import ModelAccessProxy
 from trytond.model.exceptions import AccessError
 from trytond.pool import Pool
 from trytond.tests.test_tryton import (
@@ -394,6 +395,35 @@ class ModelAccessReadTestCase(_ModelAccessTestCase):
         with self.assertRaises(AccessError):
             TestAccess.search([], order=[('relate.value', 'ASC')])
 
+    @with_transaction()
+    def test_model_access_proxy(self):
+        "Test model access proxy"
+        pool = Pool()
+        ModelAccess = pool.get('ir.model.access')
+        TestAccess = pool.get(self.model_name)
+        record, = TestAccess.create([{}])
+        ModelAccess.create([{
+                    'model': self.model_name,
+                    'perm_read': True,
+                    }])
+
+        ModelAccessProxy(record, {})
+
+    @with_transaction()
+    def test_model_access_proxy_no_access(self):
+        "Test model access proxy without access"
+        pool = Pool()
+        ModelAccess = pool.get('ir.model.access')
+        TestAccess = pool.get(self.model_name)
+        record, = TestAccess.create([{}])
+        ModelAccess.create([{
+                    'model': self.model_name,
+                    'perm_read': False,
+                    }])
+
+        with self.assertRaises(AccessError):
+            ModelAccessProxy(record, {})
+
 
 class ModelAccessWriteTestCase(_ModelAccessTestCase):
     _perm = 'perm_write'
@@ -1101,6 +1131,41 @@ class ModelFieldAccessReadTestCase(_ModelFieldAccessTestCase):
         with self.assertRaises(AccessError):
             TestAccess.search([('relate', 'child_of', 42, 'parent')])
 
+    @with_transaction()
+    def test_model_access_proxy(self):
+        "Test model access proxy"
+        pool = Pool()
+        FieldAccess = pool.get('ir.model.field.access')
+        TestAccess = pool.get('test.access')
+        record, = TestAccess.create([{'field1': "foo"}])
+        FieldAccess.create([{
+                    'model': 'test.access',
+                    'field': 'field1',
+                    'perm_read': True,
+                    }])
+
+        proxy = ModelAccessProxy(record, {})
+
+        self.assertEqual(proxy.field1, 'foo')
+
+    @with_transaction()
+    def test_model_access_proxy_no_access(self):
+        "Test model access proxy without access"
+        pool = Pool()
+        FieldAccess = pool.get('ir.model.field.access')
+        TestAccess = pool.get('test.access')
+        record, = TestAccess.create([{'field1': "foo"}])
+        FieldAccess.create([{
+                    'model': 'test.access',
+                    'field': 'field1',
+                    'perm_read': False,
+                    }])
+
+        proxy = ModelAccessProxy(record, {})
+
+        with self.assertRaises(AccessError):
+            ModelAccessProxy(proxy.field1)
+
 
 class ModelFieldAccessWriteTestCase(_ModelFieldAccessTestCase):
     _perm = 'perm_write'

trytond/trytond/tests/test_rule.py

@@ -2,6 +2,7 @@
 # this repository contains the full copyright notices and license terms.
 import json
 
+from trytond.model import ModelAccessProxy
 from trytond.model.exceptions import AccessError
 from trytond.pool import Pool
 from trytond.tests.test_tryton import (
@@ -663,3 +664,46 @@ class ModelRuleTestCase(TestCase):
 
         with self.assertRaisesRegex(AccessError, "Field different from foo"):
             TestRuleModel.read([test.id], ['name'])
+
+    @with_transaction()
+    def test_model_access_proxy(self):
+        "Test model access proxy"
+        pool = Pool()
+        TestRule = pool.get('test.rule')
+        RuleGroup = pool.get('ir.rule.group')
+        rule_group, = RuleGroup.create([{
+                    'name': "Field different from foo",
+                    'model': TestRule.__name__,
+                    'global_p': True,
+                    'perm_read': False,
+                    'rules': [('create', [{
+                                    'domain': json.dumps(
+                                        [('field', '!=', 'foo')]),
+                                    }])],
+                    }])
+        record, = TestRule.create([{'field': 'foo'}])
+
+        proxy = ModelAccessProxy(record, {})
+
+        self.assertEqual(proxy.field, 'foo')
+
+    @with_transaction()
+    def test_model_access_proxy_no_access(self):
+        "Test model access proxy without access"
+        pool = Pool()
+        TestRule = pool.get('test.rule')
+        RuleGroup = pool.get('ir.rule.group')
+        rule_group, = RuleGroup.create([{
+                    'name': "Field different from foo",
+                    'model': TestRule.__name__,
+                    'global_p': True,
+                    'perm_read': True,
+                    'rules': [('create', [{
+                                    'domain': json.dumps(
+                                        [('field', '!=', 'foo')]),
+                                    }])],
+                    }])
+        record, = TestRule.create([{'field': 'foo'}])
+
+        with self.assertRaises(AccessError):
+            ModelAccessProxy(record, {})

trytond/trytond/transaction.py

@@ -450,7 +450,9 @@ class Transaction(object):
 
     @property
     def check_access(self):
-        return self.context.get('_check_access', False)
+        return (
+            self.context.get('_check_access', False)
+            and self.user != 0)
 
     @property
     def active_records(self):