[FIX] core: fix inconsistencies between _apply_ir_rule and check_access_rule
Issues ====== - `_apply_ir_rule` applies `ir.rule` of the current model and also `ir.rule` from the inherited model (via inherits). But `check_access_rule` doesn't check the later one. - `_flush_search` doesn't flush fields coming from the `ir.rule` of the inherited model (via inherits). Then the filtering done by `_apply_ir_rule` may be inconsistent with cached values. Changes ======= Because of https://github.com/odoo/odoo/blob/6ddcb448612f5d784c8e9ebb90f19077e65be3e1/odoo/osv/expression.py#L1073-L1073, and https://github.com/odoo/odoo/blob/00e86b1552d1e5541a8dbf9411de5cfdb8990cc4/odoo/fields.py#L2895 leaf like `('<many2one_delegate>', 'any', [<sub-domain>])`, will be translated in the same way as `_inherits_join_add` does. We can remove `_inherits_join_add` and its usage in `_apply_ir_rule` and change `ir.rule._compute_domain` to also return the inherited (via inherits) `ir.rule` domain (with the new 'any' operator). Since `_compute_domain` is used by `_apply_ir_rule` and `_filter_access_rules_python`, everything is consistent. Also fix `BaseModel._flush_search` to take in account 'any'/'not any' operators (compulsory in order to flush correctly new domain from `ir.rule._compute_domain` generated). Part-of: odoo/odoo#125916
This commit is contained in:
@@ -1,9 +1,8 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
import logging
|
||||
import warnings
|
||||
|
||||
from odoo import api, fields, models, tools, SUPERUSER_ID, _
|
||||
from odoo import api, fields, models, tools, _
|
||||
from odoo.exceptions import AccessError, ValidationError
|
||||
from odoo.osv import expression
|
||||
from odoo.tools import config
|
||||
@@ -126,14 +125,20 @@ class IrRule(models.Model):
|
||||
'tuple(self._compute_domain_context_values())'),
|
||||
)
|
||||
def _compute_domain(self, model_name, mode="read"):
|
||||
global_domains = [] # list of domains
|
||||
|
||||
# add rules for parent models
|
||||
for parent_model_name, parent_field_name in self.env[model_name]._inherits.items():
|
||||
if domain := self._compute_domain(parent_model_name, mode):
|
||||
global_domains.append([(parent_field_name, 'any', domain)])
|
||||
|
||||
rules = self._get_rules(model_name, mode=mode)
|
||||
if not rules:
|
||||
return
|
||||
return expression.AND(global_domains) if global_domains else []
|
||||
|
||||
# browse user and rules as SUPERUSER_ID to avoid access errors!
|
||||
# browse user and rules with sudo to avoid access errors!
|
||||
eval_context = self._eval_context()
|
||||
user_groups = self.env.user.groups_id
|
||||
global_domains = [] # list of domains
|
||||
group_domains = [] # list of domains
|
||||
for rule in rules.sudo():
|
||||
# evaluate the domain for the current user
|
||||
|
||||
@@ -1275,8 +1275,7 @@ class TestQueries(TransactionCase):
|
||||
LEFT JOIN "res_partner" AS "res_users__partner_id" ON
|
||||
("res_users"."partner_id" = "res_users__partner_id"."id")
|
||||
WHERE ("res_users"."active" = %s)
|
||||
AND ("res_users"."id" = %s)
|
||||
AND ("res_users__partner_id"."id" = %s)
|
||||
AND (("res_users"."id" = %s) AND ("res_users__partner_id"."id" = %s))
|
||||
ORDER BY "res_users__partner_id"."name", "res_users"."login"
|
||||
''']):
|
||||
Model.search([])
|
||||
|
||||
@@ -3,8 +3,10 @@ access_test_access_right_some_obj_employee,access_test_access_right_some_obj,mod
|
||||
access_test_access_right_some_obj_public,access_test_access_right_some_obj,model_test_access_right_some_obj,base.group_public,1,1,1,1
|
||||
access_test_access_right_container_employee,access_test_access_right_container,model_test_access_right_container,base.group_user,1,1,1,1
|
||||
access_test_access_right_container_public,access_test_access_right_container,model_test_access_right_container,base.group_public,1,1,1,1
|
||||
access_test_access_right_parent,access_test_access_right_parent,model_test_access_right_parent,base.group_user,1,1,1,1
|
||||
access_test_access_right_child,access_test_access_right_child,model_test_access_right_child,base.group_user,1,1,1,1
|
||||
access_test_access_right_inherits_public,access_test_access_right_inherits,model_test_access_right_inherits,base.group_public,1,1,1,1
|
||||
access_test_access_right_inherits_employee,access_test_access_right_inherits,model_test_access_right_inherits,base.group_user,1,1,1,1
|
||||
access_test_access_right_child_public,access_test_access_right_child,model_test_access_right_child,base.group_public,1,1,1,1
|
||||
access_test_access_right_child_employee,access_test_access_right_child,model_test_access_right_child,base.group_user,1,1,1,1
|
||||
access_test_access_right_obj_categ,access_test_access_right_obj_categ,model_test_access_right_obj_categ,base.group_user,1,1,1,1
|
||||
access_test_ticket_portal,access_test_ticket_portal,model_test_access_right_ticket,base.group_portal,1,0,0,0
|
||||
access_test_ticket_user,access_test_ticket_user,model_test_access_right_ticket,base.group_user,1,1,1,1
|
||||
|
||||
|
@@ -23,13 +23,13 @@ class Container(models.Model):
|
||||
|
||||
some_ids = fields.Many2many('test_access_right.some_obj', 'test_access_right_rel', 'container_id', 'some_id')
|
||||
|
||||
class Parent(models.Model):
|
||||
_name = 'test_access_right.parent'
|
||||
class Inherits(models.Model):
|
||||
_name = 'test_access_right.inherits'
|
||||
_description = 'Object for testing related access rights'
|
||||
|
||||
_inherits = {'test_access_right.some_obj': 'obj_id'}
|
||||
_inherits = {'test_access_right.some_obj': 'some_id'}
|
||||
|
||||
obj_id = fields.Many2one('test_access_right.some_obj', required=True, ondelete='restrict')
|
||||
some_id = fields.Many2one('test_access_right.some_obj', required=True, ondelete='restrict')
|
||||
|
||||
class Child(models.Model):
|
||||
_name = 'test_access_right.child'
|
||||
|
||||
@@ -206,14 +206,21 @@ This restriction is due to the following rules:
|
||||
Contact your administrator to request access if necessary.""" % (self.record.display_name, self.record.id, self.user.name, self.user.id)
|
||||
)
|
||||
|
||||
ChildModel = self.env['test_access_right.inherits']
|
||||
with self.assertRaises(AccessError) as ctx:
|
||||
ChildModel.with_user(self.user).create({'some_id': self.record.id, 'val': 2})
|
||||
self.assertEqual(
|
||||
ctx.exception.args[0],
|
||||
"""Due to security restrictions, you are not allowed to modify 'Object For Test Access Right' (test_access_right.some_obj) records.
|
||||
|
||||
Records: %s (id=%s)
|
||||
User: %s (id=%s)
|
||||
|
||||
p = self.env['test_access_right.parent'].create({'obj_id': self.record.id})
|
||||
with self.assertRaisesRegex(
|
||||
AccessError,
|
||||
r"Implicitly accessed through 'Object for testing related access rights' \(test_access_right.parent\)\.",
|
||||
):
|
||||
p.with_user(self.user).write({'val': 1})
|
||||
This restriction is due to the following rules:
|
||||
- rule 0
|
||||
|
||||
Contact your administrator to request access if necessary.""" % (self.record.display_name, self.record.id, self.user.name, self.user.id)
|
||||
)
|
||||
|
||||
def test_locals(self):
|
||||
self.env.ref('base.group_no_one').write({'users': [Command.link(self.user.id)]})
|
||||
@@ -390,12 +397,12 @@ Note: this might be a multi-company issue.
|
||||
|
||||
Contact your administrator to request access if necessary.""" % (self.record.display_name, self.record.id, self.record.sudo().company_id.display_name, self.user.name, self.user.id)
|
||||
)
|
||||
p = self.env['test_access_right.parent'].create({'obj_id': self.record.id})
|
||||
p = self.env['test_access_right.inherits'].create({'some_id': self.record.id})
|
||||
self.env.flush_all()
|
||||
self.env.invalidate_all()
|
||||
with self.assertRaisesRegex(
|
||||
AccessError,
|
||||
r"Implicitly accessed through 'Object for testing related access rights' \(test_access_right.parent\)\.",
|
||||
r"Implicitly accessed through 'Object for testing related access rights' \(test_access_right.inherits\)\.",
|
||||
):
|
||||
p.with_user(self.user).val
|
||||
|
||||
|
||||
@@ -8,85 +8,82 @@ from odoo import Command
|
||||
|
||||
|
||||
class TestRules(TransactionCase):
|
||||
def setUp(self):
|
||||
super(TestRules, self).setUp()
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
|
||||
ObjCateg = self.env['test_access_right.obj_categ']
|
||||
SomeObj = self.env['test_access_right.some_obj']
|
||||
self.categ1 = ObjCateg.create({'name': 'Food'}).id
|
||||
self.id1 = SomeObj.create({'val': 1, 'categ_id': self.categ1}).id
|
||||
self.id2 = SomeObj.create({'val': -1, 'categ_id': self.categ1}).id
|
||||
ObjCateg = cls.env['test_access_right.obj_categ']
|
||||
SomeObj = cls.env['test_access_right.some_obj']
|
||||
cls.categ = ObjCateg.create({'name': 'Food'})
|
||||
cls.allowed = SomeObj.create({'val': 1, 'categ_id': cls.categ.id})
|
||||
cls.forbidden = SomeObj.create({'val': -1, 'categ_id': cls.categ.id})
|
||||
# create a global rule forbidding access to records with a negative
|
||||
# (or zero) val
|
||||
self.env['ir.rule'].create({
|
||||
cls.env['ir.rule'].create({
|
||||
'name': 'Forbid negatives',
|
||||
'model_id': self.browse_ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'model_id': cls.env.ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'domain_force': "[('val', '>', 0)]"
|
||||
})
|
||||
# create a global rule that forbid access to records without
|
||||
# categories, the search is part of the test
|
||||
self.env['ir.rule'].create({
|
||||
cls.env['ir.rule'].create({
|
||||
'name': 'See all categories',
|
||||
'model_id': self.browse_ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'model_id': cls.env.ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'domain_force': "[('categ_id', 'in', user.env['test_access_right.obj_categ'].search([]).ids)]"
|
||||
})
|
||||
|
||||
@mute_logger('odoo.addons.base.models.ir_rule')
|
||||
def test_basic_access(self):
|
||||
env = self.env(user=self.browse_ref('base.public_user'))
|
||||
env = self.env(user=self.env.ref('base.public_user'))
|
||||
allowed = self.allowed.with_env(env)
|
||||
forbidden = self.forbidden.with_env(env)
|
||||
|
||||
# put forbidden record in cache
|
||||
browse2 = env['test_access_right.some_obj'].browse(self.id2)
|
||||
# this is the one we want
|
||||
browse1 = env['test_access_right.some_obj'].browse(self.id1)
|
||||
# this one should not blow up
|
||||
self.assertEqual(allowed.val, 1)
|
||||
|
||||
# this should not blow up
|
||||
self.assertEqual(browse1.val, 1)
|
||||
|
||||
# but this should
|
||||
browse1.invalidate_model(['val'])
|
||||
# but this one should
|
||||
allowed.invalidate_model(['val'])
|
||||
with self.assertRaises(AccessError):
|
||||
self.assertEqual(browse2.val, -1)
|
||||
self.assertEqual(forbidden.val, -1)
|
||||
|
||||
@mute_logger('odoo.addons.base.models.ir_rule')
|
||||
def test_group_rule(self):
|
||||
env = self.env(user=self.browse_ref('base.public_user'))
|
||||
env = self.env(user=self.env.ref('base.public_user'))
|
||||
allowed = self.allowed.with_env(env)
|
||||
forbidden = self.forbidden.with_env(env)
|
||||
|
||||
# we forbid access to the public group, to which the public user belongs
|
||||
self.env['ir.rule'].create({
|
||||
'name': 'Forbid public group',
|
||||
'model_id': self.browse_ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'groups': [Command.set([self.browse_ref('base.group_public').id])],
|
||||
'model_id': self.env.ref('test_access_rights.model_test_access_right_some_obj').id,
|
||||
'groups': [Command.set([self.env.ref('base.group_public').id])],
|
||||
'domain_force': "[(0, '=', 1)]"
|
||||
})
|
||||
|
||||
browse2 = env['test_access_right.some_obj'].browse(self.id2)
|
||||
browse1 = env['test_access_right.some_obj'].browse(self.id1)
|
||||
|
||||
# everything should blow up
|
||||
(browse1 + browse2).invalidate_model(['val'])
|
||||
(allowed + forbidden).invalidate_model(['val'])
|
||||
with self.assertRaises(AccessError):
|
||||
self.assertEqual(browse2.val, -1)
|
||||
self.assertEqual(forbidden.val, -1)
|
||||
with self.assertRaises(AccessError):
|
||||
self.assertEqual(browse1.val, 1)
|
||||
self.assertEqual(allowed.val, 1)
|
||||
|
||||
def test_many2many(self):
|
||||
""" Test assignment of many2many field where rules apply. """
|
||||
ids = [self.id1, self.id2]
|
||||
ids = [self.allowed.id, self.forbidden.id]
|
||||
|
||||
# create container as superuser, connected to all some_objs
|
||||
container_admin = self.env['test_access_right.container'].create({'some_ids': [Command.set(ids)]})
|
||||
self.assertItemsEqual(container_admin.some_ids.ids, ids)
|
||||
|
||||
# check the container as the public user
|
||||
container_user = container_admin.with_user(self.browse_ref('base.public_user'))
|
||||
container_user = container_admin.with_user(self.env.ref('base.public_user'))
|
||||
container_user.invalidate_model(['some_ids'])
|
||||
self.assertItemsEqual(container_user.some_ids.ids, [self.id1])
|
||||
self.assertItemsEqual(container_user.some_ids.ids, [self.allowed.id])
|
||||
|
||||
# this should not fail
|
||||
container_user.write({'some_ids': [Command.set(ids)]})
|
||||
container_user.invalidate_model(['some_ids'])
|
||||
self.assertItemsEqual(container_user.some_ids.ids, [self.id1])
|
||||
self.assertItemsEqual(container_user.some_ids.ids, [self.allowed.id])
|
||||
container_admin.invalidate_model(['some_ids'])
|
||||
self.assertItemsEqual(container_admin.some_ids.ids, ids)
|
||||
|
||||
@@ -98,14 +95,13 @@ class TestRules(TransactionCase):
|
||||
self.assertItemsEqual(container_admin.some_ids.ids, [])
|
||||
|
||||
def test_access_rule_performance(self):
|
||||
env = self.env(user=self.browse_ref('base.public_user'))
|
||||
env = self.env(user=self.env.ref('base.public_user'))
|
||||
Model = env['test_access_right.some_obj']
|
||||
with self.assertQueryCount(0):
|
||||
Model._filter_access_rules('read')
|
||||
|
||||
def test_no_context_in_ir_rules(self):
|
||||
""" The context should not impact the ir rules. """
|
||||
env = self.env(user=self.browse_ref('base.public_user'))
|
||||
ObjCateg = self.env['test_access_right.obj_categ']
|
||||
SomeObj = self.env['test_access_right.some_obj']
|
||||
|
||||
@@ -116,11 +112,55 @@ class TestRules(TransactionCase):
|
||||
|
||||
# record1 is food and is accessible with an empy context
|
||||
self.env.registry.clear_cache()
|
||||
records = SomeObj.search([('id', '=', self.id1)])
|
||||
records = SomeObj.search([('id', '=', self.allowed.id)])
|
||||
self.assertTrue(records)
|
||||
|
||||
# it should also be accessible as the context is not used when
|
||||
# searching for SomeObjs
|
||||
self.env.registry.clear_cache()
|
||||
records = SomeObj.with_context(only_media=True).search([('id', '=', self.id1)])
|
||||
records = SomeObj.with_context(only_media=True).search([('id', '=', self.allowed.id)])
|
||||
self.assertTrue(records)
|
||||
|
||||
def test_check_access_rule_with_inherits(self):
|
||||
"""
|
||||
For models in `_inherits`, verify that both methods `check_access_rule`
|
||||
and `_apply_ir_rules` check the rules from parent models.
|
||||
"""
|
||||
ChildModel = self.env['test_access_right.inherits']
|
||||
allowed_child, __ = children = ChildModel.create([
|
||||
{'some_id': self.allowed.id}, {'some_id': self.forbidden.id},
|
||||
])
|
||||
|
||||
user = self.env.ref('base.public_user')
|
||||
search_result = children.with_user(user).search([('id', 'in', children.ids)], order='id')
|
||||
filter_result = children.with_user(user)._filter_access_rules_python('read')
|
||||
|
||||
self.assertEqual(search_result, allowed_child)
|
||||
self.assertEqual(filter_result, allowed_child)
|
||||
|
||||
def test_flush_search_with_inherits(self):
|
||||
"""
|
||||
For models with `_inherits`, verify that method `_flush_search` takes in
|
||||
account the rules from inherited models, as method `_search` does.
|
||||
"""
|
||||
ChildModel = self.env['test_access_right.inherits']
|
||||
child = ChildModel.create([{'some_id': self.allowed.id}])
|
||||
self.env.flush_all()
|
||||
|
||||
self.env['ir.rule'].create({
|
||||
'name': 'Forbid 0 value',
|
||||
'model_id': self.env['ir.model']._get('test_access_right.some_obj').id,
|
||||
'domain_force': str([('val', '!=', 0)]),
|
||||
})
|
||||
|
||||
user = self.env.ref('base.public_user')
|
||||
|
||||
# the parent record is accessible, so is the child record
|
||||
search_result = ChildModel.with_user(user).search([('id', '=', child.id)], order='id')
|
||||
self.assertEqual(search_result, child)
|
||||
|
||||
# make the parent record inaccessible, and verify that the child record
|
||||
# becomes inaccessible, too
|
||||
self.allowed.val = 0
|
||||
search_result = ChildModel.with_user(user).search([('id', '=', child.id)], order='id')
|
||||
self.assertEqual(search_result, ChildModel)
|
||||
|
||||
+3
-23
@@ -2587,20 +2587,6 @@ class BaseModel(metaclass=MetaModel):
|
||||
|
||||
return rows_dict
|
||||
|
||||
def _inherits_join_add(self, current_model, parent_model_name, query):
|
||||
"""
|
||||
Add missing table SELECT and JOIN clause to ``query`` for reaching the parent table (no duplicates)
|
||||
:param current_model: current model object
|
||||
:param parent_model_name: name of the parent model for which the clauses should be added
|
||||
:param query: query object on which the JOIN should be added
|
||||
"""
|
||||
inherits_field = current_model._inherits[parent_model_name]
|
||||
parent_model = self.env[parent_model_name]
|
||||
parent_alias = query.left_join(
|
||||
current_model._table, inherits_field, parent_model._table, 'id', inherits_field,
|
||||
)
|
||||
return parent_alias
|
||||
|
||||
@api.model
|
||||
def _inherits_join_calc(self, alias, fname, query):
|
||||
"""
|
||||
@@ -4740,14 +4726,6 @@ class BaseModel(metaclass=MetaModel):
|
||||
if domain:
|
||||
expression.expression(domain, self.sudo(), self._table, query)
|
||||
|
||||
# apply ir.rules from the parents (through _inherits)
|
||||
for parent_model_name in self._inherits:
|
||||
domain = Rule._compute_domain(parent_model_name, mode)
|
||||
if domain:
|
||||
parent_model = self.env[parent_model_name]
|
||||
parent_alias = self._inherits_join_add(self, parent_model_name, query)
|
||||
expression.expression(domain, parent_model.sudo(), parent_alias, query)
|
||||
|
||||
@api.model
|
||||
def _generate_m2o_order_by(self, alias, order_field, query, reverse_direction, seen):
|
||||
"""
|
||||
@@ -4868,7 +4846,7 @@ class BaseModel(metaclass=MetaModel):
|
||||
return
|
||||
seen.add(self._name)
|
||||
|
||||
to_flush = defaultdict(set) # {model_name: field_names}
|
||||
to_flush = defaultdict(OrderedSet) # {model_name: field_names}
|
||||
if fields:
|
||||
to_flush[self._name].update(fields)
|
||||
|
||||
@@ -4882,6 +4860,8 @@ class BaseModel(metaclass=MetaModel):
|
||||
if arg[1] in ('child_of', 'parent_of') and comodel._parent_store:
|
||||
# hierarchy operators need the parent field
|
||||
collect_from_path(comodel, comodel._parent_name)
|
||||
if arg[1] in ('any', 'not any'):
|
||||
collect_from_domain(comodel, arg[2])
|
||||
|
||||
def collect_from_path(model, path):
|
||||
# path is a dot-separated sequence of field names
|
||||
|
||||
Reference in New Issue
Block a user