[IMP] mail: allow to force the FROM headers of the outgoing mail server

Purpose
=======
We want to be able to force the FROM headers when we sent email in
SMTP. So we can avoid the emails to be considered as spam.

Specifications
==============
This is done with 2 system parameters.

If the system parameter `mail.force.smtp.from` is set we encapsulate all
outgoing email from with the given value.

If the previous system parameter is not set and if both
`mail.dynamic.smtp.from` and `mail.catchall.domain` are set, we
encapsulate the FROM only if the domain of the email is not the same as
the domain of the catchall parameter.

Otherwise we do not encapsulate the email (same behavior as before this
commit).

Task 2367946
See odoo/odoo/pull/61853

closes odoo/odoo#70854

X-original-commit: 044532aed6f796cd93e37556b4a8ce47d2c23a7c
Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>
Signed-off-by: Olivier Dony (odo) <odo@openerp.com>

3 files changed, 146 insertions, 1 deletions

diff --git a/odoo/addons/base/models/ir_mail_server.py b/odoo/addons/base/models/ir_mail_server.py
index 15a2faf19ea..b9d22531629 100644
--- a/odoo/addons/base/models/ir_mail_server.py
+++ b/odoo/addons/base/models/ir_mail_server.py
@@ -19,7 +19,8 @@ import idna
 
 from odoo import api, fields, models, tools, _
 from odoo.exceptions import UserError
-from odoo.tools import ustr, pycompat, formataddr
+from odoo.tools import ustr, pycompat, formataddr, encapsulate_email, email_domain_extract
+
 
 _logger = logging.getLogger(__name__)
 _test_logger = logging.getLogger('odoo.tests')
@@ -305,7 +306,12 @@ class IrMailServer(models.Model):
         if references:
             msg['references'] = references
         msg['Subject'] = subject
+
+        email_from, return_path = self._get_email_from(email_from)
         msg['From'] = email_from
+        if return_path:
+            headers.setdefault('Return-Path', return_path)
+
         del msg['Reply-To']
         msg['Reply-To'] = reply_to or email_from
         msg['To'] = email_to
@@ -333,6 +339,35 @@ class IrMailServer(models.Model):
                 msg.add_attachment(fcontent, maintype, subtype, filename=fname)
         return msg
 
+    def _get_email_from(self, email_from):
+        """Logic which determines which email to use when sending the email.
+
+        - If the system parameter `mail.force.smtp.from` is set we encapsulate all
+          outgoing email from
+        - If the previous system parameter is not set and if both `mail.dynamic.smtp.from`
+          and `mail.catchall.domain` are set, we encapsulate the FROM only if the domain
+          of the email is not the same as the domain of the catchall parameter
+        - Otherwise we do not encapsulate the email and given email_from is used as is
+
+        :param email_from: The initial FROM headers
+        :return: The FROM to used in the headers and optionally the Return-Path
+        """
+        force_smtp_from = self.env['ir.config_parameter'].sudo().get_param('mail.force.smtp.from')
+        dynamic_smtp_from = self.env['ir.config_parameter'].sudo().get_param('mail.dynamic.smtp.from')
+        catchall_domain = self.env['ir.config_parameter'].sudo().get_param('mail.catchall.domain')
+
+        if force_smtp_from:
+            rfc2822_force_smtp_from = extract_rfc2822_addresses(force_smtp_from)
+            rfc2822_force_smtp_from = rfc2822_force_smtp_from[0] if rfc2822_force_smtp_from else None
+            return encapsulate_email(email_from, force_smtp_from), rfc2822_force_smtp_from
+
+        elif dynamic_smtp_from and catchall_domain and email_domain_extract(email_from) != catchall_domain:
+            rfc2822_dynamic_smtp_from = extract_rfc2822_addresses(dynamic_smtp_from)
+            rfc2822_dynamic_smtp_from = rfc2822_dynamic_smtp_from[0] if rfc2822_dynamic_smtp_from else None
+            return encapsulate_email(email_from, dynamic_smtp_from), rfc2822_dynamic_smtp_from
+
+        return email_from, None
+
     @api.model
     def _get_default_bounce_address(self):
         '''Compute the default bounce address.
diff --git a/odoo/addons/base/tests/test_mail.py b/odoo/addons/base/tests/test_mail.py
index 9f85bad64f4..b74ff23ed11 100644
--- a/odoo/addons/base/tests/test_mail.py
+++ b/odoo/addons/base/tests/test_mail.py
@@ -447,6 +447,76 @@ class EmailConfigCase(SavepointCase):
         )
         self.assertEqual(message["From"], "settings@example.com")
 
+    def test_email_from_rewrite(self):
+        get_email_from = self.env['ir.mail_server']._get_email_from
+        set_param = self.env['ir.config_parameter'].set_param
+
+        # Standard case, no setting
+        set_param('mail.force.smtp.from', False)
+        set_param('mail.dynamic.smtp.from', False)
+        set_param('mail.catchall.domain', 'odoo.example.com')
+
+        email_from, return_path = get_email_from('admin@test.example.com')
+        self.assertEqual(email_from, 'admin@test.example.com')
+        self.assertFalse(return_path)
+
+        email_from, return_path = get_email_from('"Admin" <admin@test.example.com>')
+        self.assertEqual(email_from, '"Admin" <admin@test.example.com>')
+        self.assertFalse(return_path)
+
+        # We always force the email FROM
+        set_param('mail.force.smtp.from', 'email_force@domain.com')
+        set_param('mail.dynamic.smtp.from', False)
+        set_param('mail.catchall.domain', 'odoo.example.com')
+
+        email_from, return_path = get_email_from('admin@test.example.com')
+        self.assertEqual(email_from, '"admin@test.example.com" <email_force@domain.com>')
+        self.assertEqual(return_path, 'email_force@domain.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@test.example.com>')
+        self.assertEqual(email_from, '"Admin (admin@test.example.com)" <email_force@domain.com>')
+        self.assertEqual(return_path, 'email_force@domain.com')
+
+        # We always force the email FROM (notification email contains a name part)
+        set_param('mail.force.smtp.from', '"Your notification bot" <email_force@domain.com>')
+        set_param('mail.dynamic.smtp.from', False)
+        set_param('mail.catchall.domain', 'odoo.example.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@test.example.com>')
+        self.assertEqual(email_from, '"Admin (admin@test.example.com)" <email_force@domain.com>',
+                         msg='Should drop the name part of the forced email')
+        self.assertEqual(return_path, 'email_force@domain.com')
+
+        # We dynamically force the email FROM
+        set_param('mail.force.smtp.from', False)
+        set_param('mail.dynamic.smtp.from', 'notification@odoo.example.com')
+        set_param('mail.catchall.domain', 'odoo.example.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@test.example.com>')
+        self.assertEqual(email_from, '"Admin (admin@test.example.com)" <notification@odoo.example.com>',
+                         msg='Domain is not the same as the catchall domain, we should force the email FROM')
+        self.assertEqual(return_path, 'notification@odoo.example.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@odoo.example.com>')
+        self.assertEqual(email_from, '"Admin" <admin@odoo.example.com>',
+                         msg='Domain is the same as the catchall domain, we should not force the email FROM')
+        self.assertFalse(return_path)
+
+        # We dynamically force the email FROM (notification email contains a name part)
+        set_param('mail.force.smtp.from', False)
+        set_param('mail.dynamic.smtp.from', '"Your notification bot" <notification@odoo.example.com>')
+        set_param('mail.catchall.domain', 'odoo.example.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@test.example.com>')
+        self.assertEqual(email_from, '"Admin (admin@test.example.com)" <notification@odoo.example.com>',
+                         msg='Domain is not the same as the catchall domain, we should force the email FROM')
+        self.assertEqual(return_path, 'notification@odoo.example.com')
+
+        email_from, return_path = get_email_from('"Admin" <admin@odoo.example.com>')
+        self.assertEqual(email_from, '"Admin" <admin@odoo.example.com>',
+                         msg='Domain is the same as the catchall domain, we should not force the email FROM')
+        self.assertFalse(return_path)
+
 
 class TestEmailMessage(TransactionCase):
     def test_as_string(self):
diff --git a/odoo/tools/mail.py b/odoo/tools/mail.py
index 0bfc51cc2ad..a52e555c52a 100644
--- a/odoo/tools/mail.py
+++ b/odoo/tools/mail.py
@@ -554,6 +554,19 @@ def email_escape_char(email_address):
     """ Escape problematic characters in the given email address string"""
     return email_address.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
 
+
+def email_domain_extract(email):
+    """Return the domain of the given email."""
+    if not email:
+        return
+
+    email_split = getaddresses([email])
+    if not email_split or not email_split[0]:
+        return
+
+    _, _, domain = email_split[0][1].rpartition('@')
+    return domain
+
 # was mail_thread.decode_header()
 def decode_message_header(message, header, separator=' '):
     return separator.join(h for h in message.get_all(header, []) if h)
@@ -600,3 +613,30 @@ def formataddr(pair, charset='utf-8'):
             name = email_addr_escapes_re.sub(r'\\\g<0>', name)
             return f'"{name}" <{local}@{domain}>'
     return f"{local}@{domain}"
+
+
+def encapsulate_email(old_email, new_email):
+    """Change the FROM of the message and use the old one as name.
+
+    e.g.
+    * Old From: "Admin" <admin@gmail.com>
+    * New From: notifications@odoo.com
+    * Output:   "Admin (admin@gmail.com)" <notifications@odoo.com>
+    """
+    old_email_split = getaddresses([old_email])
+    if not old_email_split or not old_email_split[0]:
+        return old_email
+
+    new_email_split = getaddresses([new_email])
+    if not new_email_split or not new_email_split[0]:
+        return
+
+    if old_email_split[0][0]:
+        name_part = '%s (%s)' % old_email_split[0]
+    else:
+        name_part = old_email_split[0][1]
+
+    return formataddr((
+        name_part,
+        new_email_split[0][1],
+    ))