[IMP] mail: factorize code that handles alias security in mail gateway

Currently check of alias security is directly embedded in mail.thread
routing method. In this commit we move this check in the mail.alias.mixin
class itself. Doing this allows models to extend or improve the behavior
depending on how inheriting models use aliases.

addons/hr/models/__init__.py

@@ -4,6 +4,5 @@
 import hr
 import hr_config_settings
 import mail_alias
-import mail_thread
 import res_partner
 import res_users

addons/hr/models/mail_alias.py

@@ -1,10 +1,29 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-from odoo import fields, models
+from odoo import fields, models, tools
 
 
 class Alias(models.Model):
     _inherit = 'mail.alias'
 
     alias_contact = fields.Selection(selection_add=[('employees', 'Authenticated Employees')])
+
+
+class MailAlias(models.AbstractModel):
+    _inherit = 'mail.alias.mixin'
+
+    def _alias_check_contact(self, message, message_dict, alias):
+        if alias.alias_contact == 'employees' and self.ids:
+            email_from = tools.decode_message_header(message, 'From')
+            email_address = tools.email_split(email_from)[0]
+            employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
+            if not employee:
+                employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
+            if not employee:
+                return {
+                    'error_message': 'restricted to employees',
+                    'error_template': self.env.ref('hr.mail_template_data_unknown_employee_email_address').body_html,
+                }
+            return True
+        return super(MailAlias, self)._alias_check_contact(message, message_dict, alias)

addons/hr/models/mail_thread.py

@@ -1,44 +1,24 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-import logging
-
-from odoo import api, models, _
-from odoo.tools import decode_message_header, email_split
-
-_logger = logging.getLogger(__name__)
+from odoo import models, tools
+from odoo.tools import email_split
 
 
-class MailThread(models.AbstractModel):
-    _inherit = "mail.thread"
+class MailAlias(models.AbstractModel):
+    _inherit = 'mail.alias.mixin'
 
-    @api.model
-    def message_route_verify(self, message, message_dict, route,
-                             update_author=True, assert_model=True,
-                             create_fallback=True, allow_private=False,
-                             drop_alias=False):
-        res = super(MailThread, self).message_route_verify(
-            message, message_dict, route,
-            update_author=update_author,
-            assert_model=assert_model,
-            create_fallback=create_fallback,
-            allow_private=allow_private,
-            drop_alias=drop_alias)
-
-        if res:
-            alias = route[4]
-            email_from = decode_message_header(message, 'From')
-            message_id = message.get('Message-Id')
-
-            # Alias: check alias_contact settings for employees
-            if alias and alias.alias_contact == 'employees':
-                email_address = email_split(email_from)[0]
-                employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
-                if not employee:
-                    employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
-                if not employee:
-                    mail_template = self.env.ref('hr.mail_template_data_unknown_employee_email_address')
-                    self._routing_warn(_('alias %s does not accept unknown employees') % alias.alias_name, _('skipping'), message_id, route, False)
-                    self._routing_create_bounce_email(email_from, mail_template.body_html, message)
-                    return False
-        return res
+    def _alias_check_contact(self, message, message_dict, alias):
+        if alias.alias_contact == 'employees' and self.ids:
+            email_from = tools.decode_message_header(message, 'From')
+            email_address = email_split(email_from)[0]
+            employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
+            if not employee:
+                employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
+            if not employee:
+                return {
+                    'error_message': 'restricted to employees',
+                    'error_template': self.env.ref('hr.mail_template_data_unknown_employee_email_address').body_html,
+                }
+            return True
+        return super(MailAlias, self)._alias_check_contact(message, message_dict, alias)

addons/mail/models/mail_alias.py

@@ -262,3 +262,21 @@ class AliasMixin(models.AbstractModel):
             record.with_context({'mail_notrack': True}).alias_id = alias
             _logger.info('Mail alias created for %s %s (id %s)',
                          record._name, record.display_name, record.id)
+
+    def _alias_check_contact(self, message, message_dict, alias):
+        author = self.env['res.partner'].browse(message_dict.get('author_id', False))
+        if alias.alias_contact == 'followers' and self.ids:
+            if not hasattr(self, "message_partner_ids") or not hasattr(self, "message_channel_ids"):
+                return {
+                    'error_mesage': _('incorrectly configured alias'),
+                }
+            accepted_partner_ids = self.message_partner_ids | self.message_channel_ids.mapped('channel_partner_ids')
+            if not author or author not in accepted_partner_ids:
+                return {
+                    'error_mesage': _('restricted to followers'),
+                }
+        elif alias.alias_contact == 'partners' and not author:
+            return {
+                'error_message': _('restricted to known authors')
+            }
+        return True

addons/mail/models/mail_thread.py

@@ -944,27 +944,23 @@ class MailThread(models.AbstractModel):
         if not author_id and update_author:
             author_ids = self.env['mail.thread']._find_partner_from_emails([email_from], res_model=model, res_id=thread_id)
             if author_ids:
-                author_id = author_ids[0]
-                message_dict['author_id'] = author_id
+                message_dict['author_id'] = author_ids[0]
 
         # Alias: check alias_contact settings
-        if alias and alias.alias_contact == 'followers' and (thread_id or alias.alias_parent_thread_id):
+        if alias:
             if thread_id:
                 obj = record_set[0]
-            else:
+            elif alias.alias_parent_thread_id:
                 obj = self.env[alias.alias_parent_model_id.model].browse(alias.alias_parent_thread_id)
-            accepted_partner_ids = list(
-                set(partner.id for partner in obj.message_partner_ids) |
-                set(partner.id for channel in obj.message_channel_ids for partner in channel.channel_partner_ids)
-            )
-            if not author_id or author_id not in accepted_partner_ids:
-                self._routing_warn(_('alias %s restricted to internal followers') % alias.alias_name, _('skipping'), message_id, route, False)
-                self._routing_create_bounce_email(email_from, _generic_bounce_body_html, message)
+            elif model and hasattr(record_set, '_alias_check_contact'):
+                obj = self.env[model]
+            else:
+                obj = self.env['mail.alias.mixin']
+            check_result = obj._alias_check_contact(message, message_dict, alias)
+            if check_result is not True:
+                self._routing_warn(_('alias %s: %s') % (alias.alias_name, check_result.get('error_message', _('unknown error'))), _('skipping'), message_id, route, False)
+                self._routing_create_bounce_email(email_from, check_result.get('error_template', _generic_bounce_body_html), message)
                 return False
-        elif alias and alias.alias_contact == 'partners' and not author_id:
-            self._routing_warn(_('alias %s does not accept unknown author') % alias.alias_name, _('skipping'), message_id, route, False)
-            self._routing_create_bounce_email(email_from, _generic_bounce_body_html, message)
-            return False
 
         if not model and not thread_id and not alias and not allow_private:
             return False