[IMP] core: disable x2many manipulation as sudo for sensitive models

This revision adds a flag
`_allow_sudo_commands` to which models can opt-in
to protect themselves against malicious
manipulation of one2many or many2many fields
through an environment using `sudo` or a more priviledged user.

Flagging a model `_allow_sudo_commands = False`
disable `sudo` and `with_user(...)`
when manipulating a one2many or many2many
field targeting this model.

task-3695103

addons/auth_totp_mail/tests/test_notify_security_update_totp.py

@@ -42,7 +42,7 @@ class TestNotifySecurityUpdateTotp(TestNotifySecurityUpdate):
         # now remove the key using the user's relationship
         with self.mock_mail_gateway():
             self.env['auth_totp.device'].flush_model()
-            self.env.user.write({'totp_trusted_device_ids': [(5, 0)]})
+            self.env.user.sudo(False)._revoke_all_devices()
 
         self.assertMailMailWEmails(recipients, 'outgoing', fields_values={
             'subject': 'Security Update: Device Removed',

addons/portal/tests/__init__.py

@@ -3,5 +3,6 @@
 
 from . import test_login
 from . import test_message_format_portal
+from . import test_portal
 from . import test_tours
 from . import test_portal_wizard

addons/portal/tests/test_portal.py

@@ -0,0 +1,43 @@
+from odoo import Command
+from odoo.http import Request
+from odoo.tests.common import HttpCase, tagged
+
+
+@tagged('-at_install', 'post_install')
+class TestUsersHttp(HttpCase):
+
+    def test_deactivate_portal_user(self):
+        # Create a portal user with data which should be removed on deactivation
+        login = 'portal_user'
+        portal_user = self.env['res.users'].create({
+            'name': login,
+            'login': login,
+            'password': login,
+            'groups_id': [Command.set([self.env.ref('base.group_portal').id])],
+        })
+        self.env['res.users.apikeys'].with_user(portal_user)._generate(None, 'Portal API Key')
+        self.assertTrue(portal_user.api_key_ids)
+
+        # Request the deactivation of the portal account as portal through the route meant for this purpose
+        self.authenticate(login, login)
+        self.url_open('/my/deactivate_account', data={
+            'validation': login,
+            'password': login,
+            'csrf_token': Request.csrf_token(self)
+        })
+
+        # Assert the user is disabled, correctly renamed, the critical data is well removed.
+        self.assertFalse(portal_user.active)
+        self.assertTrue(portal_user.login.startswith('__deleted_user_'))
+        self.env.cr.execute('SELECT password FROM res_users WHERE id = %s', [portal_user.id])
+        [hashed] = self.env.cr.fetchone()
+        # Assert the password is an empty string. In `check_credentials`, an empty string as password is invalid
+        # so you are not able to login with an empty password.
+        self.assertTrue(self.env['res.users']._crypt_context().verify('', hashed))
+        self.assertFalse(portal_user.api_key_ids)
+        # Assert the deletion of the account is added to the deletion processing queue.
+        self.assertTrue(self.env['res.users.deletion'].search([('user_id', '=', portal_user.id)]))
+        # Run the cron processing the deletion queue
+        self.env.ref('base.ir_cron_res_users_deletion').method_direct_trigger()
+        # Assert the account is completely deleted
+        self.assertFalse(portal_user.exists())

odoo/addons/base/models/ir_actions.py

@@ -52,6 +52,7 @@ class IrActions(models.Model):
     _description = 'Actions'
     _table = 'ir_actions'
     _order = 'name'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Action Name', required=True, translate=True)
     type = fields.Char(string='Action Type', required=True)
@@ -222,6 +223,7 @@ class IrActionsActWindow(models.Model):
     _table = 'ir_act_window'
     _inherit = 'ir.actions.actions'
     _order = 'name'
+    _allow_sudo_commands = False
 
     @api.constrains('res_model', 'binding_model_id')
     def _check_model(self):
@@ -355,6 +357,7 @@ class IrActionsActWindowView(models.Model):
     _table = 'ir_act_window_view'
     _rec_name = 'view_id'
     _order = 'sequence,id'
+    _allow_sudo_commands = False
 
     sequence = fields.Integer()
     view_id = fields.Many2one('ir.ui.view', string='View')
@@ -374,6 +377,7 @@ class IrActionsActWindowclose(models.Model):
     _description = 'Action Window Close'
     _inherit = 'ir.actions.actions'
     _table = 'ir_actions'
+    _allow_sudo_commands = False
 
     type = fields.Char(default='ir.actions.act_window_close')
 
@@ -391,6 +395,7 @@ class IrActionsActUrl(models.Model):
     _table = 'ir_act_url'
     _inherit = 'ir.actions.actions'
     _order = 'name'
+    _allow_sudo_commands = False
 
     type = fields.Char(default='ir.actions.act_url')
     url = fields.Text(string='Action URL', required=True)
@@ -446,6 +451,7 @@ class IrActionsServer(models.Model):
     _table = 'ir_act_server'
     _inherit = 'ir.actions.actions'
     _order = 'sequence,name'
+    _allow_sudo_commands = False
 
     DEFAULT_PYTHON_CODE = """# Available variables:
 #  - env: environment on which the action is triggered
@@ -1018,6 +1024,7 @@ class IrActionsTodo(models.Model):
     _description = "Configuration Wizards"
     _rec_name = 'action_id'
     _order = "sequence, id"
+    _allow_sudo_commands = False
 
     action_id = fields.Many2one('ir.actions.actions', string='Action', required=True, index=True)
     sequence = fields.Integer(default=10)
@@ -1094,6 +1101,7 @@ class IrActionsActClient(models.Model):
     _inherit = 'ir.actions.actions'
     _table = 'ir_act_client'
     _order = 'name'
+    _allow_sudo_commands = False
 
     type = fields.Char(default='ir.actions.client')
 

odoo/addons/base/models/ir_actions_report.py

@@ -109,6 +109,7 @@ class IrActionsReport(models.Model):
     _inherit = 'ir.actions.actions'
     _table = 'ir_act_report_xml'
     _order = 'name'
+    _allow_sudo_commands = False
 
     type = fields.Char(default='ir.actions.report')
     binding_type = fields.Selection(default='report')

odoo/addons/base/models/ir_asset.py

@@ -61,6 +61,7 @@ class IrAsset(models.Model):
     _name = 'ir.asset'
     _description = 'Asset'
     _order = 'sequence, id'
+    _allow_sudo_commands = False
 
     @api.model_create_multi
     def create(self, vals_list):

odoo/addons/base/models/ir_config_parameter.py

@@ -32,6 +32,7 @@ class IrConfigParameter(models.Model):
     _description = 'System Parameter'
     _rec_name = 'key'
     _order = 'key'
+    _allow_sudo_commands = False
 
     key = fields.Char(required=True)
     value = fields.Text(required=True)

odoo/addons/base/models/ir_cron.py

@@ -50,6 +50,7 @@ class ir_cron(models.Model):
     _name = "ir.cron"
     _order = 'cron_name'
     _description = 'Scheduled Actions'
+    _allow_sudo_commands = False
 
     ir_actions_server_id = fields.Many2one(
         'ir.actions.server', 'Server action',
@@ -528,6 +529,7 @@ class ir_cron_trigger(models.Model):
     _name = 'ir.cron.trigger'
     _description = 'Triggered actions'
     _rec_name = 'cron_id'
+    _allow_sudo_commands = False
 
     cron_id = fields.Many2one("ir.cron", index=True)
     call_at = fields.Datetime()

odoo/addons/base/models/ir_default.py

@@ -12,6 +12,7 @@ class IrDefault(models.Model):
     _name = 'ir.default'
     _description = 'Default Values'
     _rec_name = 'field_id'
+    _allow_sudo_commands = False
 
     field_id = fields.Many2one('ir.model.fields', string="Field", required=True,
                                ondelete='cascade', index=True)

odoo/addons/base/models/ir_logging.py

@@ -6,6 +6,7 @@ class IrLogging(models.Model):
     _name = 'ir.logging'
     _description = 'Logging'
     _order = 'id DESC'
+    _allow_sudo_commands = False
 
     # The _log_access fields are defined manually for the following reasons:
     #

odoo/addons/base/models/ir_mail_server.py

@@ -89,6 +89,7 @@ class IrMailServer(models.Model):
     _name = "ir.mail_server"
     _description = 'Mail Server'
     _order = 'sequence, id'
+    _allow_sudo_commands = False
 
     NO_VALID_RECIPIENT = ("At least one valid recipient address should be "
                           "specified for outgoing emails (To/Cc/Bcc)")

odoo/addons/base/models/ir_model.py

@@ -182,6 +182,7 @@ class IrModel(models.Model):
     _description = "Models"
     _order = 'model'
     _rec_names_search = ['name', 'model']
+    _allow_sudo_commands = False
 
     def _default_field_id(self):
         if self.env.context.get('install_mode'):
@@ -499,6 +500,7 @@ class IrModelFields(models.Model):
     _description = "Fields"
     _order = "name"
     _rec_name = 'field_description'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Field Name', default='x_', required=True, index=True)
     complete_name = fields.Char(index=True)
@@ -1422,6 +1424,7 @@ class IrModelSelection(models.Model):
     _name = 'ir.model.fields.selection'
     _order = 'sequence, id'
     _description = "Fields Selection"
+    _allow_sudo_commands = False
 
     field_id = fields.Many2one("ir.model.fields",
         required=True, ondelete="cascade", index=True,
@@ -1729,6 +1732,7 @@ class IrModelConstraint(models.Model):
     """
     _name = 'ir.model.constraint'
     _description = 'Model Constraint'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Constraint', required=True, index=True,
                        help="PostgreSQL constraint or foreign key name.")
@@ -1883,6 +1887,7 @@ class IrModelRelation(models.Model):
     """
     _name = 'ir.model.relation'
     _description = 'Relation Model'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Relation Name', required=True, index=True,
                        help="PostgreSQL table name implementing a many2many relation.")
@@ -1945,6 +1950,7 @@ class IrModelAccess(models.Model):
     _name = 'ir.model.access'
     _description = 'Model Access'
     _order = 'model_id,group_id,name,id'
+    _allow_sudo_commands = False
 
     name = fields.Char(required=True, index=True)
     active = fields.Boolean(default=True, help='If you uncheck the active field, it will disable the ACL without deleting it (if you delete a native ACL, it will be re-created when you reload the module).')
@@ -2113,6 +2119,7 @@ class IrModelData(models.Model):
     _name = 'ir.model.data'
     _description = 'Model Data'
     _order = 'module, model, name'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='External Identifier', required=True,
                        help="External Key/Identifier that can be used for "

odoo/addons/base/models/ir_module.py

@@ -79,6 +79,7 @@ class ModuleCategory(models.Model):
     _name = "ir.module.category"
     _description = "Application"
     _order = 'name'
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Name', required=True, translate=True, index=True)
     parent_id = fields.Many2one('ir.module.category', string='Parent Application', index=True)
@@ -153,6 +154,7 @@ class Module(models.Model):
     _rec_names_search = ['name', 'shortdesc', 'summary']
     _description = "Module"
     _order = 'application desc,sequence,name'
+    _allow_sudo_commands = False
 
     @classmethod
     def get_module_info(cls, name):
@@ -935,6 +937,7 @@ class ModuleDependency(models.Model):
     _name = "ir.module.module.dependency"
     _description = "Module dependency"
     _log_access = False  # inserts are done manually, create and write uid, dates are always null
+    _allow_sudo_commands = False
 
     # the dependency name
     name = fields.Char(index=True)
@@ -977,6 +980,7 @@ class ModuleDependency(models.Model):
 class ModuleExclusion(models.Model):
     _name = "ir.module.module.exclusion"
     _description = "Module exclusion"
+    _allow_sudo_commands = False
 
     # the exclusion name
     name = fields.Char(index=True)

odoo/addons/base/models/ir_profile.py

@@ -22,6 +22,7 @@ class IrProfile(models.Model):
     _description = 'Profiling results'
     _log_access = False  # avoid useless foreign key on res_user
     _order = 'session desc, id desc'
+    _allow_sudo_commands = False
 
     create_date = fields.Datetime('Creation Date')
 

odoo/addons/base/models/ir_property.py

@@ -37,6 +37,7 @@ TYPE2CLEAN = {
 class Property(models.Model):
     _name = 'ir.property'
     _description = 'Company Property'
+    _allow_sudo_commands = False
 
     name = fields.Char(index=True)
     res_id = fields.Char(string='Resource', index=True, help="If not set, acts as a default value for new resources",)

odoo/addons/base/models/ir_rule.py

@@ -14,6 +14,7 @@ class IrRule(models.Model):
     _description = 'Record Rule'
     _order = 'model_id DESC,id'
     _MODES = ['read', 'write', 'create', 'unlink']
+    _allow_sudo_commands = False
 
     name = fields.Char(index=True)
     active = fields.Boolean(default=True, help="If you uncheck the active field, it will disable the record rule without deleting it (if you delete a native record rule, it may be re-created when you reload the module).")

odoo/addons/base/models/ir_sequence.py

@@ -95,6 +95,7 @@ class IrSequence(models.Model):
     _name = 'ir.sequence'
     _description = 'Sequence'
     _order = 'name'
+    _allow_sudo_commands = False
 
     def _get_number_next_actual(self):
         '''Return number from ir_sequence row when no_gap implementation,
@@ -319,6 +320,7 @@ class IrSequenceDateRange(models.Model):
     _name = 'ir.sequence.date_range'
     _description = 'Sequence Date Range'
     _rec_name = "sequence_id"
+    _allow_sudo_commands = False
 
     def _get_number_next_actual(self):
         '''Return number from ir_sequence row when no_gap implementation,

odoo/addons/base/models/ir_ui_menu.py

@@ -21,6 +21,7 @@ class IrUiMenu(models.Model):
     _description = 'Menu'
     _order = "sequence,id"
     _parent_store = True
+    _allow_sudo_commands = False
 
     name = fields.Char(string='Menu', required=True, translate=True)
     active = fields.Boolean(default=True)

odoo/addons/base/models/ir_ui_view.py

@@ -78,6 +78,7 @@ class ViewCustom(models.Model):
     _description = 'Custom View'
     _order = 'create_date desc'  # search(limit=1) should return the last customization
     _rec_name = 'user_id'
+    _allow_sudo_commands = False
 
     ref_id = fields.Many2one('ir.ui.view', string='Original View', index=True, required=True, ondelete='cascade')
     user_id = fields.Many2one('res.users', string='User', index=True, required=True, ondelete='cascade')
@@ -153,6 +154,7 @@ class View(models.Model):
     _name = 'ir.ui.view'
     _description = 'View'
     _order = "priority,name,id"
+    _allow_sudo_commands = False
 
     name = fields.Char(string='View Name', required=True)
     model = fields.Char(index=True)

odoo/addons/base/models/res_partner.py

@@ -166,6 +166,7 @@ class Partner(models.Model):
     _name = "res.partner"
     _order = "complete_name ASC, id DESC"
     _rec_names_search = ['complete_name', 'email', 'ref', 'vat', 'company_registry']  # TODO vat must be sanitized the same way for storing/searching
+    _allow_sudo_commands = False
 
     # the partner types that must be added to a partner's complete name, like "Delivery"
     _complete_name_displayed_types = ('invoice', 'delivery', 'other')

odoo/addons/base/models/res_users.py

@@ -131,6 +131,7 @@ class Groups(models.Model):
     _description = "Access Groups"
     _rec_name = 'full_name'
     _order = 'name'
+    _allow_sudo_commands = False
 
     name = fields.Char(required=True, translate=True)
     users = fields.Many2many('res.users', 'res_groups_users_rel', 'gid', 'uid')
@@ -277,6 +278,7 @@ class Users(models.Model):
     _description = 'User'
     _inherits = {'res.partner': 'partner_id'}
     _order = 'name, login'
+    _allow_sudo_commands = False
 
     def _check_company_domain(self, companies=None):
         return [('company_ids', 'in', models.to_company_ids(companies))] if companies else []
@@ -942,8 +944,8 @@ class Users(models.Model):
             user.write({
                 'login': f'__deleted_user_{user.id}_{time.time()}',
                 'password': '',
-                'api_key_ids': Command.clear(),
             })
+            user.api_key_ids._remove()
 
             res_users_deletion_values.append({
                 'user_id': user.id,
@@ -2051,6 +2053,7 @@ class APIKeys(models.Model):
     _name = 'res.users.apikeys'
     _description = 'Users API Keys'
     _auto = False # so we can have a secret column
+    _allow_sudo_commands = False
 
     name = fields.Char("Description", required=True, readonly=True)
     user_id = fields.Many2one('res.users', index=True, required=True, readonly=True, ondelete="cascade")

odoo/addons/base/tests/test_api.py

@@ -225,6 +225,15 @@ class TestAPI(SavepointCaseWithUserDemo):
         self.assertEqual(partner.with_context(lang='nl_NL').env.lang, None, 'Inactive lang context lang should have None env.lang')
         self.assertEqual(partner.with_context(lang='Dummy').env.lang, None, 'Ilegal lang context should have None env.lang')
 
+    def test_56_environment_uid_origin(self):
+        """Check the expected behavior of `env.uid_origin`"""
+        user_demo = self.env.ref('base.user_demo')
+        user_admin = self.env.ref('base.user_admin')
+        self.assertEqual(self.env.uid_origin, None)
+        self.assertEqual(self.env['base'].with_user(user_demo).env.uid_origin, user_demo.id)
+        self.assertEqual(self.env['base'].with_user(user_demo).with_user(user_admin).env.uid_origin, user_demo.id)
+        self.assertEqual(self.env['base'].with_user(user_admin).with_user(user_demo).env.uid_origin, user_admin.id)
+
     @mute_logger('odoo.models')
     def test_60_cache(self):
         """ Check the record cache behavior """

odoo/addons/test_new_api/models/test_new_api.py

@@ -1456,7 +1456,9 @@ class ComputeMember(models.Model):
 
 class User(models.Model):
     _name = _description = 'test_new_api.user'
+    _allow_sudo_commands = False
 
+    name = fields.Char()
     group_ids = fields.Many2many('test_new_api.group')
     group_count = fields.Integer(compute='_compute_group_count', store=True)
 
@@ -1468,7 +1470,9 @@ class User(models.Model):
 
 class Group(models.Model):
     _name = _description = 'test_new_api.group'
+    _allow_sudo_commands = False
 
+    name = fields.Char()
     user_ids = fields.Many2many('test_new_api.user')
 
 

odoo/addons/test_new_api/tests/test_new_fields.py

@@ -3179,6 +3179,130 @@ class TestX2many(common.TransactionCase):
         })
         self.assertTrue(field.unlink())
 
+    @mute_logger('odoo.addons.base.models.ir_model')
+    @common.users('portal')
+    def test_sudo_commands(self):
+        """Test manipulating a x2many field using Commands with `sudo` or with another user (`with_user`)
+        is not allowed when the destination model is flagged `_allow_sudo_commands = False` and the transaction user
+        does not have the required access rights.
+
+        This test asserts an AccessError is raised
+        when a user attempts to pass Commands to a One2many and Many2many field
+        targeting a model flagged with `_allow_sudo_commands = False`
+        while using an environment with `sudo()` or `with_user(admin_user)`.
+
+        The `with_user` are edge cases in some business codes, where a more-priviledged user is used temporary
+        to perform an action, such as:
+        - `Documents.with_user(share.create_uid)`
+        - `request.env['sign.request'].with_user(contract.hr_responsible_id).sudo()`
+        """
+
+        admin_user = self.env.ref('base.user_admin')
+        my_user = self.env.user.sudo(False)
+
+        # 1. one2many field `res.partner.user_ids`
+        # Sanity checks
+        # `res.partner` must be flagged as `_allow_sudo_commands = False` otherwise the test is pointless
+        self.assertEqual(self.env['res.users']._allow_sudo_commands, False)
+        # in case the type of `res.partner.user_ids` changes in a future release.
+        # if `res.partner.user_ids` is no longer a one2many, this test must be adapted.
+        self.assertEqual(self.env['res.partner']._fields['user_ids'].type, 'one2many')
+        my_partner = my_user.partner_id
+
+        for Partner, my_partner in [
+            (self.env['res.partner'].with_user(admin_user), my_partner.with_user(admin_user)),
+            (self.env['res.partner'].sudo(), my_partner.sudo()),
+        ]:
+            # 1.0 Command.CREATE
+            # Case: a public/portal user creating a new users with arbitrary values
+            with self.assertRaisesRegex(AccessError, "not allowed to create 'User'"):
+                Partner.create({
+                    'name': 'foo',
+                    'user_ids': [Command.create({
+                        'login': 'foo',
+                        'password': 'foo',
+                    })],
+                })
+            # 1.1 Command.UPDATE
+            # Case: a public/portal updating his user to add himself a group
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.update(my_partner.user_ids[0].id, {
+                        'groups_id': [self.env.ref('base.group_system').id],
+                    })],
+                })
+            # 1.2 Command.DELETE
+            # Case: a public user deleting the public user to mess with the database
+            with self.assertRaisesRegex(AccessError, "not allowed to delete 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.delete(my_partner.user_ids[0].id)],
+                })
+            # 1.3 Command.UNLINK
+            # Case: a public user unlinking the public partner and the public user to mess with the database
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.unlink(my_partner.user_ids[0].id)],
+                })
+            # 1.4 Command.LINK
+            # Case: a public/portal user changing the `partner_id` of an admin,
+            # to change the email address of the user and ask for a reset password.
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.link(admin_user.id)],
+                })
+            # 1.5 Command.CLEAR
+            # Case: a public user unlinking the public partner and the public user just to mess with the database
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.clear()],
+                })
+            # 1.6 Command.SET
+            # Case: a public/portal user changing the `partner_id` of an admin,
+            # to change the email address of the user and ask for a reset password.
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'User'"):
+                my_partner.write({
+                    'user_ids': [Command.set([admin_user.id])],
+                })
+
+        # 2. many2many field `test_new_api.discussion.participants`
+        # Sanity checks
+        # `test_new_api.user` must be flagged as `_allow_sudo_commands = False` otherwise the test is pointless
+        self.assertEqual(self.env['test_new_api.group']._allow_sudo_commands, False)
+        # in case the type of `test_new_api.discussion.participants` changes in a future release.
+        # if `test_new_api.discussion.participants` is no longer a many2many, this test must be adapted.
+        self.assertEqual(self.env['test_new_api.user']._fields['group_ids'].type, 'many2many')
+        public_group = self.env['test_new_api.group'].with_user(admin_user).create({
+            'name': 'public'
+        }).with_user(self.env.user)
+        my_user = self.env['test_new_api.user'].with_user(admin_user).create({
+            'name': 'foo',
+            'group_ids': [public_group.id],
+        }).with_user(self.env.user)
+
+        for User, my_user in [
+            (self.env['test_new_api.user'].with_user(admin_user), my_user.with_user(admin_user)),
+            (self.env['test_new_api.user'].sudo(), my_user.sudo()),
+        ]:
+            # 2.0 Command.CREATE
+            # Case: a public/portal user creating a new users with arbitrary values
+            with self.assertRaisesRegex(AccessError, "not allowed to create 'test_new_api.group'"):
+                User.create({
+                    'name': 'foo',
+                    'group_ids': [Command.create({})],
+                })
+            # 2.1 Command.UPDATE
+            # Case: a public/portal updating his user to add himself a group
+            with self.assertRaisesRegex(AccessError, "not allowed to modify 'test_new_api.group'"):
+                my_user.write({
+                    'group_ids': [Command.update(my_user.group_ids[0].id, {})],
+                })
+            # 2.2 Command.DELETE
+            # Case: a public user deleting the public user to mess with the database
+            with self.assertRaisesRegex(AccessError, "not allowed to delete 'test_new_api.group'"):
+                my_user.write({
+                    'group_ids': [Command.delete(my_user.group_ids[0].id)],
+                })
+
 
 class TestHtmlField(common.TransactionCase):
 

odoo/api.py

@@ -486,11 +486,16 @@ class Environment(Mapping):
         """ Reset the transaction, see :meth:`Transaction.reset`. """
         self.transaction.reset()
 
-    def __new__(cls, cr, uid, context, su=False):
+    def __new__(cls, cr, uid, context, su=False, uid_origin=None):
         if uid == SUPERUSER_ID:
             su = True
+
+        # isinstance(uid, int) is to handle `RequestUID`
+        uid_origin = uid_origin or (uid if isinstance(uid, int) else None)
+        if uid_origin == SUPERUSER_ID:
+            uid_origin = None
+
         assert context is not None
-        args = (cr, uid, context, su)
 
         # determine transaction object
         transaction = cr.transaction
@@ -499,13 +504,13 @@ class Environment(Mapping):
 
         # if env already exists, return it
         for env in transaction.envs:
-            if env.args == args:
+            if (env.cr, env.uid, env.context, env.su, env.uid_origin) == (cr, uid, context, su, uid_origin):
                 return env
 
         # otherwise create environment, and add it in the set
         self = object.__new__(cls)
-        args = (cr, uid, frozendict(context), su)
-        self.cr, self.uid, self.context, self.su = self.args = args
+        self.cr, self.uid, self.context, self.su = self.args = (cr, uid, frozendict(context), su)
+        self.uid_origin = uid_origin
 
         self.transaction = self.all = transaction
         self.registry = transaction.registry
@@ -561,7 +566,7 @@ class Environment(Mapping):
         if context is None:
             context = clean_context(self.context) if su and not self.su else self.context
         su = (user is None and self.su) if su is None else su
-        return Environment(cr, uid, context, su)
+        return Environment(cr, uid, context, su, self.uid_origin)
 
     def ref(self, xml_id, raise_if_not_found=True):
         """ Return the record corresponding to the given ``xml_id``.

odoo/fields.py

@@ -41,6 +41,7 @@ from .tools import DEFAULT_SERVER_DATETIME_FORMAT as DATETIME_FORMAT
 from .tools.translate import html_translate, _
 from .tools.mimetypes import guess_mimetype
 
+from odoo import SUPERUSER_ID
 from odoo.exceptions import CacheMiss
 from odoo.osv import expression
 
@@ -4341,6 +4342,13 @@ class _RelationalMulti(_Relational):
             assert not any(record_ids), f"{records_commands_list} contains a mix of real and new records. It is not supported."
             self.write_new(records_commands_list)
 
+    def _check_sudo_commands(self, comodel):
+        # if the model doesn't accept sudo commands
+        if not comodel._allow_sudo_commands:
+            # Then, disable sudo and reset the transaction origin user
+            return comodel.sudo(False).with_user(comodel.env.uid_origin)
+        return comodel
+
 
 class One2many(_RelationalMulti):
     """One2many field; the value of such a field is the recordset of all the
@@ -4450,6 +4458,7 @@ class One2many(_RelationalMulti):
 
         model = records_commands_list[0][0].browse()
         comodel = model.env[self.comodel_name].with_context(**self.context)
+        comodel = self._check_sudo_commands(comodel)
 
         if self.store:
             inverse = self.inverse_name
@@ -4558,6 +4567,7 @@ class One2many(_RelationalMulti):
         model = records_commands_list[0][0].browse()
         cache = model.env.cache
         comodel = model.env[self.comodel_name].with_context(**self.context)
+        comodel = self._check_sudo_commands(comodel)
 
         ids = {record.id for records, _ in records_commands_list for record in records}
         records = model.browse(ids)
@@ -4829,6 +4839,7 @@ class Many2many(_RelationalMulti):
 
         model = records_commands_list[0][0].browse()
         comodel = model.env[self.comodel_name].with_context(**self.context)
+        comodel = self._check_sudo_commands(comodel)
         cr = model.env.cr
 
         # determine old and new relation {x: ys}
@@ -4996,6 +5007,7 @@ class Many2many(_RelationalMulti):
 
         model = records_commands_list[0][0].browse()
         comodel = model.env[self.comodel_name].with_context(**self.context)
+        comodel = self._check_sudo_commands(comodel)
         new = lambda id_: id_ and NewId(id_)
 
         # determine old and new relation {x: ys}

odoo/models.py

@@ -616,6 +616,13 @@ class BaseModel(metaclass=MetaModel):
     as attribute.
     """
 
+    _allow_sudo_commands = True
+    """Allow One2many and Many2many Commands targeting this model in an environment using `sudo()` or `with_user()`.
+    By disabling this flag, security-sensitive models protect themselves
+    against malicious manipulation of One2many or Many2many fields
+    through an environment using `sudo` or a more priviledged user.
+    """
+
     _depends = frozendict()
     """dependencies of models backed up by SQL views
     ``{model_name: field_names}``, where ``field_names`` is an iterable.
@@ -5867,6 +5874,8 @@ class BaseModel(metaclass=MetaModel):
 
         """
         assert isinstance(flag, bool)
+        if flag == self.env.su:
+            return self
         return self.with_env(self.env(su=flag))
 
     def with_user(self, user):