[FIX] mail: send notification mail with right access link when using mail.thread

PURPOSE

Notification may be sent using a generic mail.thread record, notably when
sending user notifications. In that case links to view document are
incorrect.

HOW TO REPRODUCE

Issue

    - Install "Approvals"
    - Submit new approval with you as "Request Owner"
    - Click on "View Approval Request" in your mailbox

    The link redirects to a 505 error

Cause

    The model is not the correct one and the res_id is undefined

Solution

    Specify the model and the res_id to _notify_get_action_link
    when creating the link with kwargs

SPECIFICATIONS

Propagate message value through various notification sub methods. That way
we can rely on them if model seems void.

Also limit values given as URL parameters to some white listed values.

LINKS

opw-2358846
Task ID-2379766
Followup of odoo/odoo#60998
Followup of odoo/odoo#61545
Closes odoo/odoo#63292
Closes odoo/enterprise#15585

closes odoo/odoo#64229

X-original-commit: 58bac5d242d6548d54f0163328fa64b319852e40
Related: odoo/enterprise#15634
Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>
Co-authored-by: Achraf Ben Azzouz <abz@odoo.com>
Co-authored-by: Thibault Delavallée <tde@odoo.com>

addons/crm/models/crm_lead.py

@@ -1346,24 +1346,29 @@ class Lead(models.Model):
             return self.env.ref('crm.mt_lead_lost')
         return super(Lead, self)._track_subtype(init_values)
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Handle salesman recipients that can convert leads into opportunities
         and set opportunities as won / lost. """
-        groups = super(Lead, self)._notify_get_groups()
+        groups = super(Lead, self)._notify_get_groups(msg_vals=msg_vals)
+        msg_vals = msg_vals or {}
 
         self.ensure_one()
         if self.type == 'lead':
-            convert_action = self._notify_get_action_link('controller', controller='/lead/convert')
+            convert_action = self._notify_get_action_link('controller', controller='/lead/convert', **msg_vals)
             salesman_actions = [{'url': convert_action, 'title': _('Convert to opportunity')}]
         else:
-            won_action = self._notify_get_action_link('controller', controller='/lead/case_mark_won')
-            lost_action = self._notify_get_action_link('controller', controller='/lead/case_mark_lost')
+            won_action = self._notify_get_action_link('controller', controller='/lead/case_mark_won', **msg_vals)
+            lost_action = self._notify_get_action_link('controller', controller='/lead/case_mark_lost', **msg_vals)
             salesman_actions = [
                 {'url': won_action, 'title': _('Won')},
                 {'url': lost_action, 'title': _('Lost')}]
 
         if self.team_id:
-            salesman_actions.append({'url': self._notify_get_action_link('view', res_id=self.team_id.id, model=self.team_id._name), 'title': _('Sales Team Settings')})
+            custom_params = dict(msg_vals, res_id=self.team_id.id, model=self.team_id._name)
+            salesman_actions.append({
+                'url': self._notify_get_action_link('view', **custom_params),
+                'title': _('Sales Team Settings')
+            })
 
         salesman_group_id = self.env.ref('sales_team.group_sale_salesman').id
         new_group = (

addons/hr_holidays/models/hr_leave.py

@@ -1212,18 +1212,19 @@ class HolidaysRequest(models.Model):
             return leave_notif_subtype or self.env.ref('hr_holidays.mt_leave')
         return super(HolidaysRequest, self)._track_subtype(init_values)
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Handle HR users and officers recipients that can validate or refuse holidays
         directly from email. """
-        groups = super(HolidaysRequest, self)._notify_get_groups()
+        groups = super(HolidaysRequest, self)._notify_get_groups(msg_vals=msg_vals)
+        msg_vals = msg_vals or {}
 
         self.ensure_one()
         hr_actions = []
         if self.state == 'confirm':
-            app_action = self._notify_get_action_link('controller', controller='/leave/validate')
+            app_action = self._notify_get_action_link('controller', controller='/leave/validate', **msg_vals)
             hr_actions += [{'url': app_action, 'title': _('Approve')}]
         if self.state in ['confirm', 'validate', 'validate1']:
-            ref_action = self._notify_get_action_link('controller', controller='/leave/refuse')
+            ref_action = self._notify_get_action_link('controller', controller='/leave/refuse', **msg_vals)
             hr_actions += [{'url': ref_action, 'title': _('Refuse')}]
 
         holiday_user_group_id = self.env.ref('hr_holidays.group_hr_holidays_user').id

addons/hr_holidays/models/hr_leave_allocation.py

@@ -666,18 +666,19 @@ class HolidaysAllocation(models.Model):
             return allocation_notif_subtype_id or self.env.ref('hr_holidays.mt_leave_allocation')
         return super(HolidaysAllocation, self)._track_subtype(init_values)
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Handle HR users and officers recipients that can validate or refuse holidays
         directly from email. """
-        groups = super(HolidaysAllocation, self)._notify_get_groups()
+        groups = super(HolidaysAllocation, self)._notify_get_groups(msg_vals=msg_vals)
+        msg_vals = msg_vals or {}
 
         self.ensure_one()
         hr_actions = []
         if self.state == 'confirm':
-            app_action = self._notify_get_action_link('controller', controller='/allocation/validate')
+            app_action = self._notify_get_action_link('controller', controller='/allocation/validate', **msg_vals)
             hr_actions += [{'url': app_action, 'title': _('Approve')}]
         if self.state in ['confirm', 'validate', 'validate1']:
-            ref_action = self._notify_get_action_link('controller', controller='/allocation/refuse')
+            ref_action = self._notify_get_action_link('controller', controller='/allocation/refuse', **msg_vals)
             hr_actions += [{'url': ref_action, 'title': _('Refuse')}]
 
         holiday_user_group_id = self.env.ref('hr_holidays.group_hr_holidays_user').id

addons/mail/models/mail_channel.py

@@ -364,12 +364,12 @@ class Channel(models.Model):
         ]).unlink()
         self.invalidate_cache(fnames=['channel_partner_ids', 'channel_last_seen_partner_ids'])
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ All recipients of a message on a channel are considered as partners.
         This means they will receive a minimal email, without a link to access
         in the backend. Mailing lists should indeed send minimal emails to avoid
         the noise. """
-        groups = super(Channel, self)._notify_get_groups()
+        groups = super(Channel, self)._notify_get_groups(msg_vals=msg_vals)
         for (index, (group_name, group_func, group_data)) in enumerate(groups):
             if group_name != 'customer':
                 groups[index] = (group_name, lambda partner: False, group_data)

addons/mail/models/mail_thread.py

@@ -2192,7 +2192,7 @@ class MailThread(models.AbstractModel):
 
         model = msg_vals.get('model') if msg_vals else message.model
         model_name = model_description or (self._fallback_lang().env['ir.model']._get(model).display_name if model else False) # one query for display name
-        recipients_groups_data = self._notify_classify_recipients(partners_data, model_name)
+        recipients_groups_data = self._notify_classify_recipients(partners_data, model_name, msg_vals=msg_vals)
 
         if not recipients_groups_data:
             return True
@@ -2466,22 +2466,23 @@ class MailThread(models.AbstractModel):
         return hm
 
     def _notify_get_action_link(self, link_type, **kwargs):
-        local_kwargs = dict(kwargs)  # do not modify in-place, modify copy instead
-        base_params = {
+        """ Prepare link to an action: view document, follow document, ... """
+        params = {
             'model': kwargs.get('model', self._name),
             'res_id': kwargs.get('res_id', self.ids and self.ids[0] or False),
         }
-
-        local_kwargs.pop('message_id', None)
-        local_kwargs.pop('model', None)
-        local_kwargs.pop('res_id', None)
+        # whitelist accepted parameters: action (deprecated), token (assign), access_token
+        # (view), auth_signup_token and auth_login (for auth_signup support)
+        params.update(dict(
+            (key, value)
+            for key, value in kwargs.items()
+            if key in ('action', 'token', 'access_token', 'auth_signup_token', 'auth_login')
+        ))
 
         if link_type in ['view', 'assign', 'follow', 'unfollow']:
-            params = dict(base_params, **local_kwargs)
             base_link = '/mail/%s' % link_type
         elif link_type == 'controller':
-            controller = local_kwargs.pop('controller')
-            params = dict(base_params, **local_kwargs)
+            controller = kwargs.get('controller')
             params.pop('model')
             base_link = '%s' % controller
         else:
@@ -2497,7 +2498,7 @@ class MailThread(models.AbstractModel):
 
         return link
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Return groups used to classify recipients of a notification email.
         Groups is a list of tuple containing of form (group_name, group_func,
         group_data) where
@@ -2536,7 +2537,7 @@ class MailThread(models.AbstractModel):
             )
         ]
 
-    def _notify_classify_recipients(self, recipient_data, model_name):
+    def _notify_classify_recipients(self, recipient_data, model_name, msg_vals=None):
         """ Classify recipients to be notified of a message in groups to have
         specific rendering depending on their group. For example users could
         have access to buttons customers should not have in their emails.
@@ -2567,10 +2568,10 @@ class MailThread(models.AbstractModel):
         }]
         only return groups with recipients
         """
-
-        groups = self._notify_get_groups()
-
-        access_link = self._notify_get_action_link('view')
+        # keep a local copy of msg_vals as it may be modified to include more information about groups or links
+        local_msg_vals = dict(msg_vals) if msg_vals else {}
+        groups = self._notify_get_groups(msg_vals=local_msg_vals)
+        access_link = self._notify_get_action_link('view', **local_msg_vals)
 
         if model_name:
             view_title = _('View %s', model_name)
@@ -2748,7 +2749,7 @@ class MailThread(models.AbstractModel):
             values = {
                 'object': record,
                 'model_description': model_description,
-                'access_link': self._notify_get_action_link('view'),
+                'access_link': record._notify_get_action_link('view'),
             }
             assignation_msg = view._render(values, engine='ir.qweb', minimal_qcontext=True)
             assignation_msg = self.env['mail.render.mixin']._replace_local_links(assignation_msg)

addons/mail/tests/common.py

@@ -422,6 +422,25 @@ class MailCase(MockEmail):
         cls.email_template = cls.env['mail.template'].create(create_values)
         return cls.email_template
 
+
+    def _generate_notify_recipients(self, partners):
+        """ Tool method to generate recipients data according to structure used
+        in notification methods. Purpose is to allow testing of internals of
+        some notification methods, notably testing links or group-based notification
+        details.
+
+        See notably ``MailThread._notify_compute_recipients()``.
+        """
+        return [
+            {'id': partner.id,
+             'active': True,
+             'share': partner.partner_share,
+             'groups': partner.user_ids.groups_id.ids,
+             'notif': partner.user_ids.notification_type or 'email',
+             'type': 'user' if partner.user_ids and not partner.partner_share else partner.user_ids and 'portal' or 'customer',
+            } for partner in partners
+        ]
+
     # ------------------------------------------------------------
     # MAIL ASSERTS WRAPPERS
     # ------------------------------------------------------------

addons/portal/models/portal_mixin.py

@@ -60,16 +60,16 @@ class PortalMixin(models.AbstractModel):
 
         return '%s?%s' % ('/mail/view' if redirect else self.access_url, url_encode(params))
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         access_token = self._portal_ensure_token()
-        groups = super(PortalMixin, self)._notify_get_groups()
+        groups = super(PortalMixin, self)._notify_get_groups(msg_vals=msg_vals)
+        msg_vals = msg_vals or {}
+
         if access_token and 'partner_id' in self._fields and self['partner_id']:
             customer = self['partner_id']
-            additional_params = {
-                'access_token': self.access_token,
-            }
-            additional_params.update(customer.signup_get_auth_param()[customer.id])
-            access_link = self._notify_get_action_link('view', **additional_params)
+            msg_vals['access_token'] = self.access_token
+            msg_vals.update(customer.signup_get_auth_param()[customer.id])
+            access_link = self._notify_get_action_link('view', **msg_vals)
 
             new_group = [
                 ('portal_customer', lambda pdata: pdata['id'] == customer.id, {

addons/project/models/project.py

@@ -1208,13 +1208,13 @@ class Task(models.Model):
             return self.env.ref('project.mt_task_stage')
         return super(Task, self)._track_subtype(init_values)
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Handle project users and managers recipients that can assign
         tasks and create new one directly from notification emails. Also give
         access button to portal users and portal customers. If they are notified
         they should probably have access to the document. """
-        groups = super(Task, self)._notify_get_groups()
-
+        groups = super(Task, self)._notify_get_groups(msg_vals=msg_vals)
+        msg_vals = msg_vals or {}
         self.ensure_one()
 
         project_user_group_id = self.env.ref('project.group_project_user').id
@@ -1226,7 +1226,7 @@ class Task(models.Model):
         new_group = ('group_project_user', group_func, {})
 
         if not self.user_id and not self.stage_id.fold:
-            take_action = self._notify_get_action_link('assign')
+            take_action = self._notify_get_action_link('assign', **msg_vals)
             project_actions = [{'url': take_action, 'title': _('I take it')}]
             new_group[2]['actions'] = project_actions
 

addons/sale/models/sale.py

@@ -963,11 +963,11 @@ Reason(s) of this behavior could be:
         transaction = self.get_portal_last_transaction()
         return (self.state == 'sent' or (self.state == 'draft' and include_draft)) and not self.is_expired and self.require_payment and transaction.state != 'done' and self.amount_total
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Give access button to users and portal customer as portal is integrated
         in sale. Customer and portal group have probably no right to see
         the document so they don't have the access button. """
-        groups = super(SaleOrder, self)._notify_get_groups()
+        groups = super(SaleOrder, self)._notify_get_groups(msg_vals=msg_vals)
 
         self.ensure_one()
         if self.state not in ('draft', 'cancel'):

addons/test_mail/tests/test_message_post.py

@@ -8,13 +8,13 @@ from unittest.mock import patch
 from odoo.addons.test_mail.data.test_mail_data import MAIL_TEMPLATE_PLAINTEXT
 from odoo.addons.test_mail.models.test_mail_models import MailTestSimple
 from odoo.addons.test_mail.tests.common import TestMailCommon, TestRecipients
-from odoo.exceptions import AccessError
-from odoo.tools import mute_logger, formataddr
 from odoo.api import call_kw
+from odoo.exceptions import AccessError
 from odoo.tests import tagged
+from odoo.tools import mute_logger, formataddr
 
-import urllib.parse
 
+@tagged('mail_post')
 class TestMessagePost(TestMailCommon, TestRecipients):
 
     @classmethod
@@ -25,6 +25,48 @@ class TestMessagePost(TestMailCommon, TestRecipients):
         cls._reset_mail_context(cls.test_record)
         cls.user_admin.write({'notification_type': 'email'})
 
+    # This method should be run inside a post_install class to ensure that all
+    # message_post overrides are tested.
+    def test_message_post_return(self):
+        test_channel = self.env['mail.channel'].create({
+            'name': 'Test',
+        })
+        # Use call_kw as shortcut to simulate a RPC call.
+        messageId = call_kw(self.env['mail.channel'], 'message_post', [test_channel.id], {'body': 'test'})
+        self.assertTrue(isinstance(messageId, int))
+
+    def test_notify_recipients_internals(self):
+        pdata = self._generate_notify_recipients(self.partner_1 | self.partner_employee)
+        msg_vals = {
+            'body': 'Message body',
+            'model': self.test_record._name,
+            'res_id': self.test_record.id,
+            'subject': 'Message subject',
+        }
+        link_vals = {
+            'token': 'token_val',
+            'access_token': 'access_token_val',
+            'auth_signup_token': 'auth_signup_token_val',
+            'auth_login': 'auth_login_val',
+        }
+        notify_msg_vals = dict(msg_vals, **link_vals)
+        classify_res = self.env[self.test_record._name]._notify_classify_recipients(pdata, 'My Custom Model Name', msg_vals=notify_msg_vals)
+        # find back information for each recipients
+        partner_info = next(item for item in classify_res if item['recipients'] == self.partner_1.ids)
+        emp_info = next(item for item in classify_res if item['recipients'] == self.partner_employee.ids)
+
+        # partner: no access button
+        self.assertFalse(partner_info['has_button_access'])
+
+        # employee: access button and link
+        self.assertTrue(emp_info['has_button_access'])
+        for param, value in link_vals.items():
+            self.assertIn('%s=%s' % (param, value), emp_info['button_access']['url'])
+        self.assertIn('model=%s' % self.test_record._name, emp_info['button_access']['url'])
+        self.assertIn('res_id=%s' % self.test_record.id, emp_info['button_access']['url'])
+        self.assertNotIn('body', emp_info['button_access']['url'])
+        self.assertNotIn('subject', emp_info['button_access']['url'])
+
     @mute_logger('odoo.addons.mail.models.mail_mail')
     def test_post_needaction(self):
         (self.user_employee | self.user_admin).write({'notification_type': 'inbox'})
@@ -240,19 +282,35 @@ class TestMessagePost(TestMailCommon, TestRecipients):
     @mute_logger('odoo.addons.mail.models.mail_mail')
     def test_post_notify(self):
         self.user_employee.write({'notification_type': 'inbox'})
-        new_notification = self.test_record.message_notify(
-            subject='This should be a subject',
-            body='<p>You have received a notification</p>',
-            partner_ids=[self.partner_1.id, self.user_employee.partner_id.id],
-        )
+
+        with self.mock_mail_gateway():
+            new_notification = self.test_record.message_notify(
+                subject='This should be a subject',
+                body='<p>You have received a notification</p>',
+                partner_ids=[self.partner_1.id, self.partner_admin.id, self.user_employee.partner_id.id],
+            )
 
         self.assertEqual(new_notification.subtype_id, self.env.ref('mail.mt_note'))
         self.assertEqual(new_notification.message_type, 'user_notification')
         self.assertEqual(new_notification.body, '<p>You have received a notification</p>')
         self.assertEqual(new_notification.author_id, self.env.user.partner_id)
         self.assertEqual(new_notification.email_from, formataddr((self.env.user.name, self.env.user.email)))
-        self.assertEqual(new_notification.notified_partner_ids, self.partner_1 | self.user_employee.partner_id)
+        self.assertEqual(new_notification.notified_partner_ids, self.partner_1 | self.user_employee.partner_id | self.partner_admin)
         self.assertNotIn(new_notification, self.test_record.message_ids)
+
+        admin_mails = [x for x in self._mails if self.partner_admin.name in x.get('email_to')[0]]
+        self.assertEqual(len(admin_mails), 1, 'There should be exactly one email sent to admin')
+        admin_mail = admin_mails[0].get('body')
+        admin_access_link = admin_mail[admin_mail.index('model='):admin_mail.index('/>') - 1] if 'model=' in admin_mail else None
+  
+        self.assertIsNotNone(admin_access_link, 'The email sent to admin should contain an access link')
+        self.assertIn('model=%s' % self.test_record._name, admin_access_link, 'The access link should contain a valid model argument')
+        self.assertIn('res_id=%d' % self.test_record.id, admin_access_link, 'The access link should contain a valid res_id argument')
+
+        partner_mails = [x for x in self._mails if self.partner_1.name in x.get('email_to')[0]]
+        self.assertEqual(len(partner_mails), 1, 'There should be exactly one email sent to partner')
+        partner_mail = partner_mails[0].get('body')
+        self.assertNotIn('/mail/view?model=', partner_mail, 'The email sent to admin should not contain an access link')
         # todo xdo add test message_notify on thread with followers and stuff
 
     @mute_logger('odoo.addons.mail.models.mail_mail')
@@ -297,13 +355,3 @@ class TestMessagePost(TestMailCommon, TestRecipients):
                 subject='About %s' % test_record.name,
                 body_content=test_record.name,
                 attachments=[('first.txt', b'My first attachment', 'text/plain'), ('second.txt', b'My second attachment', 'text/plain')])
-
-    # This method should be run inside a post_install class to ensure that all
-    # message_post overrides are tested.
-    def test_message_post_return(self):
-        test_channel = self.env['mail.channel'].create({
-            'name': 'Test',
-        })
-        # Use call_kw as shortcut to simulate a RPC call.
-        messageId = call_kw(self.env['mail.channel'], 'message_post', [test_channel.id], {'body': 'test'})
-        self.assertTrue(isinstance(messageId, int))

addons/website_blog/models/website_blog.py

@@ -234,9 +234,9 @@ class BlogPost(models.Model):
             'res_id': self.id,
         }
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Add access button to everyone if the document is published. """
-        groups = super(BlogPost, self)._notify_get_groups()
+        groups = super(BlogPost, self)._notify_get_groups(msg_vals=msg_vals)
 
         if self.website_published:
             for group_name, group_method, group_data in groups:

addons/website_forum/models/forum.py

@@ -877,9 +877,9 @@ class Post(models.Model):
             'res_id': self.id,
         }
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Add access button to everyone if the document is active. """
-        groups = super(Post, self)._notify_get_groups()
+        groups = super(Post, self)._notify_get_groups(msg_vals=msg_vals)
 
         if self.state == 'active':
             for group_name, group_method, group_data in groups:

addons/website_slides/models/slide_slide.py

@@ -509,9 +509,9 @@ class Slide(models.Model):
             }
         return super(Slide, self).get_access_action(access_uid)
 
-    def _notify_get_groups(self):
+    def _notify_get_groups(self, msg_vals=None):
         """ Add access button to everyone if the document is active. """
-        groups = super(Slide, self)._notify_get_groups()
+        groups = super(Slide, self)._notify_get_groups(msg_vals=msg_vals)
 
         if self.website_published:
             for group_name, group_method, group_data in groups: