[MERGE] mail: allow the admin to open any emails

When the admin opens an email, they might get an error message if they
can not access one of the attachment of the email or because of the ACL on
mail message.

In this merge we avoid that issue by fetching only accessible attachments
and allow checking related email_from, reply-to and subject as sudo.

Task-2464212

closes odoo/odoo#78734

Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>

addons/mail/models/mail_mail.py

@@ -40,9 +40,13 @@ class MailMail(models.Model):
 
     # content
     mail_message_id = fields.Many2one('mail.message', 'Message', required=True, ondelete='cascade', index=True, auto_join=True)
+    mail_message_id_int = fields.Integer(compute='_compute_mail_message_id_int', compute_sudo=True)
     body_html = fields.Text('Rich-text Contents', help="Rich-text/HTML message")
     references = fields.Text('References', help='Message references, such as identifiers of previous messages', readonly=1)
     headers = fields.Text('Headers', copy=False)
+    restricted_attachment_count = fields.Integer('Restricted attachments', compute='_compute_restricted_attachments')
+    unrestricted_attachment_ids = fields.Many2many('ir.attachment', string='Unrestricted Attachments',
+        compute='_compute_restricted_attachments', inverse='_inverse_unrestricted_attachment_ids')
     # Auto-detected based on create() - if 'mail_message_id' was passed then this mail is a notification
     # and during unlink() we will not cascade delete the parent and its attachments
     is_notification = fields.Boolean('Notification Email', help='Mail has been created to notify people of an existing mail.message')
@@ -81,6 +85,28 @@ class MailMail(models.Model):
     scheduled_date = fields.Char('Scheduled Send Date',
         help="If set, the queue manager will send the email after the date. If not set, the email will be send as soon as possible.")
 
+    def _compute_mail_message_id_int(self):
+        for mail in self:
+            mail.mail_message_id_int = mail.mail_message_id.id
+
+    @api.depends('attachment_ids')
+    def _compute_restricted_attachments(self):
+        """We might not have access to all the attachments of the emails.
+        Compute the attachments we have access to,
+        and the number of attachments we do not have access to.
+        """
+        IrAttachment = self.env['ir.attachment']
+        for mail_sudo, mail in zip(self.sudo(), self):
+            mail.unrestricted_attachment_ids = IrAttachment._filter_attachment_access(mail_sudo.attachment_ids.ids)
+            mail.restricted_attachment_count = len(mail_sudo.attachment_ids) - len(mail.unrestricted_attachment_ids)
+
+    def _inverse_unrestricted_attachment_ids(self):
+        """We can only remove the attachments we have access to."""
+        IrAttachment = self.env['ir.attachment']
+        for mail_sudo, mail in zip(self.sudo(), self):
+            restricted_attaments = mail_sudo.attachment_ids - IrAttachment._filter_attachment_access(mail_sudo.attachment_ids.ids)
+            mail_sudo.attachment_ids = restricted_attaments | mail.unrestricted_attachment_ids
+
     @api.model_create_multi
     def create(self, values_list):
         # notification field: if not set, set if mail comes from an existing mail.message
@@ -114,6 +140,21 @@ class MailMail(models.Model):
             self.env['mail.message'].browse(mail_msg_cascade_ids).unlink()
         return res
 
+    @api.model
+    def _add_inherited_fields(self):
+        """Allow to bypass ACLs for some mail message fields.
+
+        This trick add a related_sudo on the inherits fields, it can't be done with
+        >>> subject = fields.Char(related='mail_message_id.subject', related_sudo=True)
+        because the field of <mail.message> will be fetched two times (one time before of
+        the inherits, and a second time because of the related), and so it will add extra
+        SQL queries.
+        """
+        super()._add_inherited_fields()
+        cls = type(self)
+        for field in ('email_from', 'reply_to', 'subject'):
+            cls._fields[field].related_sudo = True
+
     def action_retry(self):
         self.filtered(lambda mail: mail.state == 'exception').mark_outgoing()
 

addons/mail/views/mail_mail_views.xml

@@ -13,13 +13,13 @@
                         <field name="state" widget="statusbar" statusbar_visible="outgoing,sent,received,exception,cancel"/>
                     </header>
                     <sheet>
-                        <field name="mail_message_id" required="0" invisible="1"/>
+                        <field name="mail_message_id_int" required="0" invisible="1"/>
                         <label for="subject" class="oe_edit_only"/>
                         <h2><field name="subject"/></h2>
                         <div style="vertical-align: top;">
                             by <field name="author_id" class="oe_inline" string="User"/> on <field name="date" readonly="1" class="oe_inline"/>
                             <button name="%(action_email_compose_message_wizard)d" string="Reply" type="action" icon="fa-reply text-warning"
-                                context="{'default_composition_mode':'comment', 'default_parent_id': mail_message_id}" states='received,sent,exception,cancel'/>
+                                context="{'default_composition_mode':'comment', 'default_parent_id': mail_message_id_int}" states='received,sent,exception,cancel'/>
                         </div>
                         <group>
                             <field name="email_from"/>
@@ -51,7 +51,13 @@
                                 </group>
                             </page>
                             <page string="Attachments" name="attachments">
-                                <field name="attachment_ids"/>
+                                <div class="alert alert-warning" role="alert"
+                                    attrs="{'invisible': [('restricted_attachment_count', '=', 0)]}">
+                                    You do not have access to <field name="restricted_attachment_count"/>
+                                    attachment(s) of this email.
+                                </div>
+                                <field name="unrestricted_attachment_ids"
+                                    domain="[('res_field','=', False)]"/>
                             </page>
                             <page string="Failure Reason" name="failure_reason" attrs="{'invisible': [('state', '!=', 'exception')]}">
                                 <field name="failure_reason"/>

addons/test_mail/tests/test_mail_mail.py

@@ -6,12 +6,13 @@ import smtplib
 
 from OpenSSL.SSL import Error as SSLError
 from socket import gaierror, timeout
-from unittest.mock import call
+from unittest.mock import call, patch
 
-from odoo import api
+from odoo import api, Command
 from odoo.addons.base.models.ir_mail_server import MailDeliveryException
 from odoo.addons.test_mail.tests.common import TestMailCommon
-from odoo.tests import common, tagged
+from odoo.exceptions import AccessError
+from odoo.tests import common, tagged, users
 from odoo.tools import mute_logger
 
 
@@ -60,6 +61,67 @@ class TestMailMail(TestMailCommon):
         self.test_mail.write({'failure_reason': False, 'failure_type': False, 'state': 'outgoing'})
         self.test_notification.write({'failure_type': False, 'notification_status': 'ready'})
 
+    @users('admin')
+    def test_mail_mail_attachment_access(self):
+        mail = self.env['mail.mail'].create({
+            'body_html': 'Test',
+            'email_to': 'test@example.com',
+            'partner_ids': [(4, self.user_employee.partner_id.id)],
+            'attachment_ids': [
+                (0, 0, {'name': 'file 1', 'datas': 'c2VjcmV0'}),
+                (0, 0, {'name': 'file 2', 'datas': 'c2VjcmV0'}),
+                (0, 0, {'name': 'file 3', 'datas': 'c2VjcmV0'}),
+                (0, 0, {'name': 'file 4', 'datas': 'c2VjcmV0'}),
+            ],
+        })
+
+        def _patched_check(self, *args, **kwargs):
+            if self.env.is_superuser():
+                return
+            if any(attachment.name in ('file 2', 'file 4') for attachment in self):
+                raise AccessError('No access')
+
+        mail.invalidate_cache(ids=mail.ids)
+
+        new_attachment = self.env['ir.attachment'].create({
+            'name': 'new file',
+            'datas': 'c2VjcmV0',
+        })
+
+        with patch.object(type(self.env['ir.attachment']), 'check', _patched_check):
+            # Sanity check
+            self.assertEqual(mail.restricted_attachment_count, 2)
+            self.assertEqual(len(mail.unrestricted_attachment_ids), 2)
+            self.assertEqual(mail.unrestricted_attachment_ids.mapped('name'), ['file 1', 'file 3'])
+
+            # Add a new attachment
+            mail.write({
+                'unrestricted_attachment_ids': [Command.link(new_attachment.id)],
+            })
+            self.assertEqual(mail.restricted_attachment_count, 2)
+            self.assertEqual(len(mail.unrestricted_attachment_ids), 3)
+            self.assertEqual(mail.unrestricted_attachment_ids.mapped('name'), ['file 1', 'file 3', 'new file'])
+            self.assertEqual(len(mail.attachment_ids), 5)
+
+            # Remove an attachment
+            mail.write({
+                'unrestricted_attachment_ids': [Command.unlink(new_attachment.id)],
+            })
+            self.assertEqual(mail.restricted_attachment_count, 2)
+            self.assertEqual(len(mail.unrestricted_attachment_ids), 2)
+            self.assertEqual(mail.unrestricted_attachment_ids.mapped('name'), ['file 1', 'file 3'])
+            self.assertEqual(len(mail.attachment_ids), 4)
+
+            # Reset command
+            mail.invalidate_cache(ids=mail.ids)
+            mail.write({'unrestricted_attachment_ids': [Command.clear()]})
+            self.assertEqual(len(mail.unrestricted_attachment_ids), 0)
+            self.assertEqual(len(mail.attachment_ids), 2)
+
+            # Read in SUDO
+            mail.invalidate_cache(ids=mail.ids)
+            self.assertEqual(mail.sudo().restricted_attachment_count, 2)
+            self.assertEqual(len(mail.sudo().unrestricted_attachment_ids), 0)
 
     @mute_logger('odoo.addons.mail.models.mail_mail')
     def test_mail_mail_recipients(self):

odoo/addons/base/models/ir_attachment.py

@@ -466,6 +466,27 @@ class IrAttachment(models.Model):
             records.check_access_rights(access_mode)
             records.check_access_rule(access_mode)
 
+    @api.model
+    def _filter_attachment_access(self, attachment_ids):
+        """Filter the given attachment to return only the records the current user have access to.
+
+        :param attachment_ids: List of attachment ids we want to filter
+        :return: <ir.attachment> the current user have access to
+        """
+        ret_attachments = self.env['ir.attachment']
+        attachments = self.browse(attachment_ids)
+        if not attachments.check_access_rights('read', raise_exception=False):
+            return ret_attachments
+
+        for attachment in attachments.sudo():
+            # Use SUDO here to not raise an error during the prefetch
+            # And then drop SUDO right to check if we can access it
+            try:
+                attachment.sudo(False).check('read')
+                ret_attachments |= attachment
+            except AccessError:
+                continue
+        return ret_attachments
 
     def _read_group_allowed_fields(self):
         return ['type', 'company_id', 'res_id', 'create_date', 'create_uid', 'name', 'mimetype', 'id', 'url', 'res_field', 'res_model']