pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[MERGE] mail: allow SSL authentication and stop spoofing

Browse Source 
Purpose
=======

The global purpose of this PR is to improve the score of the emails sent by
Odoo as well as allowing to authenticate the Odoo server with SSL certificate.

SSL authentication
==================

We want to be able to authenticate our servers with a certificate for the
entire domain name instead of using an username and a password. That way
we will be able to configure an SMTP relay (on Office 365 e.g.) based on an
SSL certificate instead of an hard-coded IP address.

The host name of the server is defined in the MX DNS record on your domain.
The certificate must be valid for your domain name.

e.g.
>  Host: openerp-org.mail.protection.outlook.com
    Port: 25
    Username: \<keep it blank>
    Password: \<keep it blank>
    Security: STARTTLS
    Email: admin@odoobe.com

No spoofing
===========

Before, we were spoofing the FROM headers all the times. E.g., if someone has
an email address outside of the current email server (e.g. `test@gmail.com`)
we sent it using the default mail server and we spoofed the FROM. This causes
some issues, because emails may be marked as spam by the email clients. A
small fix for stable was merged at odoo/odoo#69218 and will be reverted as this merge
improves that behavior.

Now, we try to *not* spoof the mail from in the SMTP protocol. For that purpose
we define a default email address which is used when no mail server is
available for the sender email address. In that case, the email address
of the sender will be "encapsulated" in the default email address
(e.g. `"Admin (admin@gmail.com)" <notifications@odoo.com>`).

To know for which email address an outgoing mail server can be used, we use a
field "`from_filter`" which can be either an complete email address or a
domain name.

The headers are changed following this heuristic

1. We first look for a mail server which match the entire mail FROM
   in that case, we do not change the email header (not needed)
2. If not found, we search a mail server which match the domain name of
   the mail from (do not need to change the headers in that case)
3. If not found, find the mail server linked to the "notifications"
   email (defined in the system parameter). Then change the FROM header
   to the notifications email, and put the old one in the name part of
   this header.
   e.g.
      Initial mail from: "Admin" < admin@example.com >
      Final mail from:   "Admin (admin@odoo.com)" < notifications@odoo.com >
4. If no notifications email is configured or if no mail server are
   found for the notifications email, fallback to the old system and
   spoof the FROM header. In that case we do not have the choice if we
   want to send the email, he will probably be marked as spam.

LINKS

Task-2367946

closes odoo/odoo#61853

Related: odoo/upgrade#1903
Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>
This commit is contained in:
Odoo's Mergebot
2021-08-13 17:33:50 +02:00
committed by GitHub
parent 347111c7b8 5c44c5438d
commit 0f4dfb4459
23 changed files with 1117 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/mail/__manifest__.py
-1
View File
@@ -27,7 +27,6 @@
'views/mail_shortcode_views.xml',
'views/mail_activity_views.xml',
'views/res_config_settings_views.xml',
'data/ir_config_parameter_data.xml',
'data/res_partner_data.xml',
'data/mail_message_subtype_data.xml',
'data/mail_templates.xml',
addons/mail/data/ir_config_parameter_data.xml
-16
View File
@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="utf-8"?>
<odoo><data noupdate="1">
<!-- Catchall Email Alias -->
<record id="icp_mail_catchall_alias" model="ir.config_parameter">
<field name="key">mail.catchall.alias</field>
<field name="value">catchall</field>
</record>
<!-- Bounce Email Alias -->
<record id="icp_mail_bounce_alias" model="ir.config_parameter">
<field name="key">mail.bounce.alias</field>
<field name="value">bounce</field>
</record>
</data></odoo>
addons/mail/models/mail_mail.py
+40 -15
View File
@@ -217,22 +217,47 @@ class MailMail(models.Model):
}
return res
def _split_by_server(self):
"""Returns an iterator of pairs `(mail_server_id, record_ids)` for current recordset.
def _split_by_mail_configuration(self):
"""Group the <mail.mail> based on their "email_from" and their "mail_server_id".
The same `mail_server_id` may repeat in order to limit batch size according to
the `mail.session.batch.size` system parameter.
The <mail.mail> will have the "same sending configuration" if they have the same
mail server or the same mail from. For performance purpose, we can use an SMTP
session in batch and therefore we need to group them by the parameter that will
influence the mail server used.
The same "sending configuration" may repeat in order to limit batch size
according to the `mail.session.batch.size` system parameter.
Return iterators over
mail_server_id, email_from, Records<mail.mail>.ids
"""
groups = defaultdict(list)
# Turn prefetch OFF to avoid MemoryError on very large mail queues, we only care
# about the mail server ids in this case.
for mail in self.with_context(prefetch_fields=False):
groups[mail.mail_server_id.id].append(mail.id)
mail_values = self.read(['id', 'email_from', 'mail_server_id'])
# First group the <mail.mail> per mail_server_id and per email_from
group_per_email_from = defaultdict(list)
for values in mail_values:
mail_server_id = values['mail_server_id'][0] if values['mail_server_id'] else False
group_per_email_from[(mail_server_id, values['email_from'])].append(values['id'])
# Then find the mail server for each email_from and group the <mail.mail>
# per mail_server_id and smtp_from
mail_servers = self.env['ir.mail_server'].sudo().search([], order='sequence')
group_per_smtp_from = defaultdict(list)
for (mail_server_id, email_from), mail_ids in group_per_email_from.items():
if not mail_server_id:
mail_server, smtp_from = self.env['ir.mail_server']._find_mail_server(email_from, mail_servers)
mail_server_id = mail_server.id if mail_server else False
else:
smtp_from = email_from
group_per_smtp_from[(mail_server_id, smtp_from)].extend(mail_ids)
sys_params = self.env['ir.config_parameter'].sudo()
batch_size = int(sys_params.get_param('mail.session.batch.size', 1000))
for server_id, record_ids in groups.items():
for mail_batch in tools.split_every(batch_size, record_ids):
yield server_id, mail_batch
for (mail_server_id, smtp_from), record_ids in group_per_smtp_from.items():
for batch_ids in tools.split_every(batch_size, record_ids):
yield mail_server_id, smtp_from, batch_ids
def send(self, auto_commit=False, raise_exception=False):
""" Sends the selected emails immediately, ignoring their current
@@ -249,10 +274,10 @@ class MailMail(models.Model):
email sending process has failed
:return: True
"""
for server_id, batch_ids in self._split_by_server():
for mail_server_id, smtp_from, batch_ids in self._split_by_mail_configuration():
smtp_session = None
try:
smtp_session = self.env['ir.mail_server'].connect(mail_server_id=server_id)
smtp_session = self.env['ir.mail_server'].connect(mail_server_id=mail_server_id, smtp_from=smtp_from)
except Exception as exc:
if raise_exception:
# To be consistent and backward compatible with mail_mail.send() raised
@@ -269,7 +294,7 @@ class MailMail(models.Model):
smtp_session=smtp_session)
_logger.info(
'Sent batch %s emails via mail server ID #%s',
len(batch_ids), server_id)
len(batch_ids), mail_server_id)
finally:
if smtp_session:
smtp_session.quit()
addons/mail/tests/common.py
+2
View File
@@ -100,9 +100,11 @@ class MockEmail(common.BaseCase):
cls.alias_domain = 'test.com'
cls.alias_catchall = 'catchall.test'
cls.alias_bounce = 'bounce.test'
cls.default_from = 'notifications'
cls.env['ir.config_parameter'].set_param('mail.bounce.alias', cls.alias_bounce)
cls.env['ir.config_parameter'].set_param('mail.catchall.domain', cls.alias_domain)
cls.env['ir.config_parameter'].set_param('mail.catchall.alias', cls.alias_catchall)
cls.env['ir.config_parameter'].set_param('mail.default.from', cls.default_from)
cls.mailer_daemon_email = formataddr(('MAILER-DAEMON', '%s@%s' % (cls.alias_bounce, cls.alias_domain)))
def format(self, template, to='groups@example.com, other@gmail.com', subject='Frogs',
addons/mass_mailing/models/mailing.py
+36 -2
View File
@@ -71,8 +71,8 @@ class MassMailing(models.Model):
help='Catchy preview sentence that encourages recipients to open this email.\n'
'In most inboxes, this is displayed next to the subject.\n'
'Keep it empty if you prefer the first characters of your email content to appear instead.')
email_from = fields.Char(string='Send From', required=True,
default=lambda self: self.env.user.email_formatted)
email_from = fields.Char(string='Send From', required=True, store=True, readonly=False, compute='_compute_email_from',
default=lambda self: self.env.user.email_formatted)
sent_date = fields.Datetime(string='Sent Date', copy=False)
schedule_type = fields.Selection([('now', 'Send now'), ('scheduled', 'Send on')], string='Schedule',
@@ -157,6 +157,28 @@ class MassMailing(models.Model):
bounced_ratio = fields.Integer(compute="_compute_statistics", string='Bounced Ratio')
clicks_ratio = fields.Integer(compute="_compute_clicks_ratio", string="Number of Clicks")
next_departure = fields.Datetime(compute="_compute_next_departure", string='Scheduled date')
# UX
warning_message = fields.Char(
'Warning Message', compute='_compute_warning_message',
help='Warning message displayed in the mailing form view')
@api.depends('mail_server_id')
def _compute_email_from(self):
user_email = self.env.user.email_formatted
notification_email = self.env['ir.mail_server']._get_default_from_address()
for mailing in self:
server = mailing.mail_server_id
if not server:
mailing.email_from = mailing.email_from or user_email
elif mailing.email_from and server._match_from_filter(mailing.email_from, server.from_filter):
mailing.email_from = mailing.email_from
elif server._match_from_filter(user_email, server.from_filter):
mailing.email_from = user_email
elif server._match_from_filter(notification_email, server.from_filter):
mailing.email_from = notification_email
else:
mailing.email_from = mailing.email_from or user_email
def _compute_total(self):
for mass_mailing in self:
@@ -232,6 +254,18 @@ class MassMailing(models.Model):
else:
mass_mailing.next_departure = cron_time
@api.depends('email_from', 'mail_server_id')
def _compute_warning_message(self):
for mailing in self:
mail_server = mailing.mail_server_id
if mail_server and not mail_server._match_from_filter(mailing.email_from, mail_server.from_filter):
mailing.warning_message = _(
'This email from can not be used with this mail server.\n'
'Your emails might be marked as spam on the mail clients.'
)
else:
mailing.warning_message = False
@api.depends('mailing_type')
def _compute_medium_id(self):
for mailing in self:
addons/mass_mailing/views/mailing_mailing_views.xml
+5 -1
View File
@@ -65,7 +65,8 @@
<field name="state" readonly="1" widget="statusbar"/>
</header>
<div class="alert alert-info text-center" role="alert" attrs="{'invisible': ['&amp;','&amp;','&amp;','&amp;',('state', '!=', 'in_queue'),('sent', '=', 0),('ignored', '=', 0),('scheduled', '=', 0),('failed', '=', 0)]}">
<div class="alert alert-info text-center" role="alert"
attrs="{'invisible': ['&amp;','&amp;','&amp;','&amp;','&amp;',('state', '!=', 'in_queue'),('sent', '=', 0),('ignored', '=', 0),('scheduled', '=', 0),('failed', '=', 0),('warning_message', '=', False)]}">
<div class="o_mails_ignored" attrs="{'invisible': [('ignored', '=', 0)]}">
<button class="btn-link py-0"
name="action_view_traces_ignored"
@@ -112,6 +113,9 @@
<field name="next_departure" class="oe_inline"/>.
</strong>
</div>
<div attrs="{'invisible': [('warning_message', '=', False)]}">
<strong><field name="warning_message"/></strong>
</div>
</div>
<sheet>
addons/test_mail/tests/test_mail_mail.py
+77 -1
View File
@@ -2,20 +2,29 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import psycopg2
from unittest.mock import call
from odoo import api
from odoo.addons.base.tests.common import MockSmtplibCase
from odoo.addons.test_mail.tests.common import TestMailCommon
from odoo.tests import common, tagged
from odoo.tools import mute_logger
@tagged('mail_mail')
class TestMailMail(TestMailCommon):
class TestMailMail(TestMailCommon, MockSmtplibCase):
@classmethod
def setUpClass(cls):
super(TestMailMail, cls).setUpClass()
cls._init_mail_gateway()
cls._init_mail_servers()
cls.server_domain_2 = cls.env['ir.mail_server'].create({
'name': 'Server 2',
'smtp_host': 'test_2.com',
'from_filter': 'test_2.com',
})
cls.test_record = cls.env['mail.test.gateway'].with_context(cls._test_context).create({
'name': 'Test',
@@ -56,6 +65,73 @@ class TestMailMail(TestMailCommon):
mail.send()
self.assertEqual(self._mails[0]['headers']['Return-Path'], '%s@%s' % (self.alias_bounce, self.alias_domain))
@mute_logger('odoo.addons.mail.models.mail_mail')
def test_mail_mail_send_server(self):
"""Test that the mails are send in batch.
Batch are defined by the mail server and the email from field.
"""
self.assertEqual(self.env['ir.mail_server']._get_default_from_address(), 'notifications@test.com')
mail_values = {
'body_html': '<p>Test</p>',
'email_to': 'user@example.com',
}
# Should be encapsulated in the notification email
mails = self.env['mail.mail'].create([{
**mail_values,
'email_from': 'test@unknown_domain.com',
} for _ in range(5)]) | self.env['mail.mail'].create([{
**mail_values,
'email_from': 'test_2@unknown_domain.com',
} for _ in range(5)])
# Should use the test_2 mail server
# Once with "user_1@test_2.com" as login
# Once with "user_2@test_2.com" as login
mails |= self.env['mail.mail'].create([{
**mail_values,
'email_from': 'user_1@test_2.com',
} for _ in range(5)]) | self.env['mail.mail'].create([{
**mail_values,
'email_from': 'user_2@test_2.com',
} for _ in range(5)])
# Mail server is forced
mails |= self.env['mail.mail'].create([{
**mail_values,
'email_from': 'user_1@test_2.com',
'mail_server_id': self.server_domain.id,
} for _ in range(5)])
with self.mock_smtplib_connection():
mails.send()
self.assertEqual(self.find_mail_server_mocked.call_count, 4, 'Must be called only once per "mail from" when the mail server is not forced')
self.assertEqual(len(self.emails), 25)
# Check call to the connect method to ensure that we authenticate
# to the right mail server with the right login
self.assertEqual(self.connect_mocked.call_count, 4, 'Must be called once per batch which share the same mail server and the same smtp from')
self.connect_mocked.assert_has_calls(
calls=[
call(smtp_from='notifications@test.com', mail_server_id=self.server_notification.id),
call(smtp_from='user_1@test_2.com', mail_server_id=self.server_domain_2.id),
call(smtp_from='user_2@test_2.com', mail_server_id=self.server_domain_2.id),
call(smtp_from='user_1@test_2.com', mail_server_id=self.server_domain.id),
],
any_order=True,
)
self.assert_email_sent_smtp(message_from='"test@unknown_domain.com" <notifications@test.com>',
emails_count=5, from_filter=self.server_notification.from_filter)
self.assert_email_sent_smtp(message_from='"test_2@unknown_domain.com" <notifications@test.com>',
emails_count=5, from_filter=self.server_notification.from_filter)
self.assert_email_sent_smtp(message_from='user_1@test_2.com', emails_count=5, from_filter=self.server_domain_2.from_filter)
self.assert_email_sent_smtp(message_from='user_2@test_2.com', emails_count=5, from_filter=self.server_domain_2.from_filter)
self.assert_email_sent_smtp(message_from='user_1@test_2.com', emails_count=5, from_filter=self.server_domain.from_filter)
class TestMailMailRace(common.TransactionCase):
addons/test_mail/tests/test_performance.py
+14 -14
View File
@@ -271,7 +271,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
'partner_ids': [(4, customer_id)],
})
with self.assertQueryCount(__system__=30, emp=36):
with self.assertQueryCount(__system__=32, emp=38):
composer.send_mail()
@users('__system__', 'emp')
@@ -292,7 +292,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
'partner_ids': [(4, customer_id)],
})
with self.assertQueryCount(__system__=23, emp=29):
with self.assertQueryCount(__system__=25, emp=31):
composer.send_mail()
@users('__system__', 'emp')
@@ -311,7 +311,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
}).create({})
composer.onchange_template_id_wrapper()
with self.assertQueryCount(__system__=31, emp=37):
with self.assertQueryCount(__system__=33, emp=39):
composer.send_mail()
# remove created partner to ensure tests are the same each run
@@ -323,7 +323,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
def test_message_assignation_email(self):
self.user_test.write({'notification_type': 'email'})
record = self.env['mail.test.track'].create({'name': 'Test'})
with self.assertQueryCount(__system__=36, emp=37):
with self.assertQueryCount(__system__=38, emp=39):
record.write({
'user_id': self.user_test.id,
})
@@ -376,7 +376,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
def test_message_post_one_email_notification(self):
record = self.env['mail.test.simple'].create({'name': 'Test'})
with self.assertQueryCount(__system__=27, emp=30):
with self.assertQueryCount(__system__=29, emp=32):
record.message_post(
body='<p>Test Post Performances with an email ping</p>',
partner_ids=self.customer.ids,
@@ -498,7 +498,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
'recipient_ids': [(4, pid) for pid in self.partners.ids],
})
mail_ids = mail.ids
with self.assertQueryCount(__system__=6, emp=6):
with self.assertQueryCount(__system__=8, emp=8):
self.env['mail.mail'].sudo().browse(mail_ids).send()
self.assertEqual(mail.body_html, '<p>Test</p>')
@@ -511,7 +511,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
self.container.message_subscribe(self.user_portal.partner_id.ids)
record = self.container.with_user(self.env.user)
with self.assertQueryCount(__system__=61, emp=62):
with self.assertQueryCount(__system__=63, emp=64):
record.message_post(
body='<p>Test Post Performances</p>',
message_type='comment',
@@ -528,7 +528,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
record = self.container.with_user(self.env.user)
template_id = self.env.ref('test_mail.mail_test_container_tpl').id
with self.assertQueryCount(__system__=70, emp=71):
with self.assertQueryCount(__system__=72, emp=73):
record.message_post_with_template(template_id, message_type='comment', composition_mode='comment')
self.assertEqual(record.message_ids[0].body, '<p>Adding stuff on %s</p>' % record.name)
@@ -591,7 +591,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
})
rec1 = rec.with_context(active_test=False) # to see inactive records
self.assertEqual(rec1.message_partner_ids, self.partners | self.env.user.partner_id)
with self.assertQueryCount(__system__=36, emp=37):
with self.assertQueryCount(__system__=38, emp=39):
rec.write({'user_id': self.user_portal.id})
self.assertEqual(rec1.message_partner_ids, self.partners | self.env.user.partner_id | self.user_portal.partner_id)
# write tracking message
@@ -611,7 +611,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
customer_id = self.customer.id
user_id = self.user_portal.id
with self.assertQueryCount(__system__=105, emp=106):
with self.assertQueryCount(__system__=109, emp=110):
rec = self.env['mail.test.ticket'].create({
'name': 'Test',
'container_id': container_id,
@@ -640,7 +640,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
rec1 = rec.with_context(active_test=False) # to see inactive records
self.assertEqual(rec1.message_partner_ids, self.user_portal.partner_id | self.env.user.partner_id)
self.assertEqual(len(rec1.message_ids), 1)
with self.assertQueryCount(__system__=75, emp=75):
with self.assertQueryCount(__system__=77, emp=77):
rec.write({
'name': 'Test2',
'container_id': self.container.id,
@@ -677,7 +677,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
rec1 = rec.with_context(active_test=False) # to see inactive records
self.assertEqual(rec1.message_partner_ids, self.user_portal.partner_id | self.env.user.partner_id)
with self.assertQueryCount(__system__=83, emp=83):
with self.assertQueryCount(__system__=85, emp=85):
rec.write({
'name': 'Test2',
'container_id': container_id,
@@ -710,7 +710,7 @@ class TestMailComplexPerformance(BaseMailPerformance):
rec1 = rec.with_context(active_test=False) # to see inactive records
self.assertEqual(rec1.message_partner_ids, self.partners | self.env.user.partner_id | self.user_portal.partner_id)
with self.assertQueryCount(__system__=28, emp=29):
with self.assertQueryCount(__system__=29, emp=30):
rec.write({
'name': 'Test2',
'customer_id': customer_id,
@@ -952,7 +952,7 @@ class TestMailHeavyPerformancePost(BaseMailPerformance):
]
self.attachements = self.env['ir.attachment'].with_user(self.env.user).create(self.vals)
attachement_ids = self.attachements.ids
with self.assertQueryCount(emp=66):
with self.assertQueryCount(emp=68):
self.cr.sql_log = self.warm and self.cr.sql_log_count
record.with_context({}).message_post(
body='<p>Test body <img src="cid:cid1"> <img src="cid:cid2"></p>',
addons/test_mail_full/tests/test_mass_mailing.py
-5
View File
@@ -11,11 +11,6 @@ from odoo.tests import tagged
@tagged('mass_mailing')
class TestMassMailing(TestMailFullCommon):
@classmethod
def setUpClass(cls):
super(TestMassMailing, cls).setUpClass()
@users('user_marketing')
@mute_logger('odoo.addons.mail.models.mail_mail')
def test_mailing_w_blacklist_opt_out(self):
addons/test_mass_mailing/tests/__init__.py
+1
View File
@@ -7,4 +7,5 @@ from . import test_blacklist_mixin
from . import test_link_tracker
from . import test_mailing_test
from . import test_mailing
from . import test_mass_mailing_server
from . import test_performance
addons/test_mass_mailing/tests/test_mass_mailing_server.py
+118
View File
@@ -0,0 +1,118 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.addons.test_mass_mailing.tests.common import TestMassMailCommon
from odoo.addons.base.tests.common import MockSmtplibCase
from odoo.tests import tagged
from odoo.tests.common import users
from odoo.tools import mute_logger
@tagged('mass_mailing')
class TestMassMailingServer(TestMassMailCommon, MockSmtplibCase):
@classmethod
def setUpClass(cls):
super(TestMassMailingServer, cls).setUpClass()
cls._init_mail_gateway()
cls.env['ir.mail_server'].search([]).unlink()
cls._init_mail_servers()
cls.recipients = cls._create_mailing_test_records(model='mailing.test.optout', count=8)
@users('user_marketing')
@mute_logger('odoo.addons.mail.models.mail_mail', 'odoo.models.unlink', 'odoo.addons.mass_mailing.models.mailing')
def test_mass_mailing_server_batch(self):
"""Test that the right mail server is chosen to send the mailing.
Test also the envelop and the SMTP headers.
"""
# Test sending mailing in batch
mailings = self.env['mailing.mailing'].create([{
'subject': 'Mailing',
'body_html': 'Body for ${object.name}',
'email_from': 'specific_user@test.com',
'mailing_model_id': self.env['ir.model']._get('mailing.test.optout').id,
}, {
'subject': 'Mailing',
'body_html': 'Body for ${object.name}',
'email_from': 'unknown_name@test.com',
'mailing_model_id': self.env['ir.model']._get('mailing.test.optout').id,
}])
with self.mock_smtplib_connection():
mailings.action_send_mail()
self.assertEqual(self.find_mail_server_mocked.call_count, 2, 'Must be called only once per mail from')
self.assert_email_sent_smtp(
smtp_from='specific_user@test.com',
message_from='specific_user@test.com',
from_filter=self.server_user.from_filter,
emails_count=8,
)
self.assert_email_sent_smtp(
# Must use the bounce address here because the mail server
# is configured for the entire domain "test.com"
smtp_from=lambda x: 'bounce' in x,
message_from='unknown_name@test.com',
from_filter=self.server_domain.from_filter,
emails_count=8,
)
@users('user_marketing')
@mute_logger('odoo.addons.mail.models.mail_mail', 'odoo.models.unlink', 'odoo.addons.mass_mailing.models.mailing')
def test_mass_mailing_server_default(self):
# We do not have a mail server for this address email, so fall back to the
# "notifications@domain" email.
mailings = self.env['mailing.mailing'].create([{
'subject': 'Mailing',
'body_html': 'Body for ${object.name}',
'email_from': '"Testing" <unknow_email@unknow_domain.com>',
'mailing_model_id': self.env['ir.model']._get('mailing.test.optout').id,
}])
with self.mock_smtplib_connection():
mailings.action_send_mail()
self.assertEqual(self.find_mail_server_mocked.call_count, 1)
self.assert_email_sent_smtp(
smtp_from='notifications@test.com',
message_from='"Testing (unknow_email@unknow_domain.com)" <notifications@test.com>',
from_filter=self.server_notification.from_filter,
emails_count=8,
)
self.assertEqual(self.find_mail_server_mocked.call_count, 1, 'Must be called only once')
@users('user_marketing')
@mute_logger('odoo.addons.mail.models.mail_mail', 'odoo.models.unlink', 'odoo.addons.mass_mailing.models.mailing')
def test_mass_mailing_server_forced(self):
# We force a mail server on one mailing
mailings = self.env['mailing.mailing'].create([{
'subject': 'Mailing',
'body_html': 'Body for ${object.name}',
'email_from': self.server_user.from_filter,
'mailing_model_id': self.env['ir.model']._get('mailing.test.optout').id,
}, {
'subject': 'Mailing',
'body_html': 'Body for ${object.name}',
'email_from': 'unknow_email@unknow_domain.com',
'mailing_model_id': self.env['ir.model']._get('mailing.test.optout').id,
'mail_server_id': self.server_notification.id,
}])
with self.mock_smtplib_connection():
mailings.action_send_mail()
self.assertEqual(self.find_mail_server_mocked.call_count, 1, 'Must not be called when mail server is forced')
self.assert_email_sent_smtp(
smtp_from='specific_user@test.com',
message_from='specific_user@test.com',
from_filter=self.server_user.from_filter,
emails_count=8,
)
self.assert_email_sent_smtp(
smtp_from='unknow_email@unknow_domain.com',
message_from='unknow_email@unknow_domain.com',
from_filter=self.server_notification.from_filter,
emails_count=8,
)
addons/test_mass_mailing/tests/test_performance.py
+2 -2
View File
@@ -51,7 +51,7 @@ class TestMassMailPerformance(TestMassMailPerformanceBase):
})
# runbot needs +50 compared to local
with self.assertQueryCount(__system__=1715, marketing=1715):
with self.assertQueryCount(__system__=1718, marketing=1718):
mailing.action_send_mail()
self.assertEqual(mailing.sent, 50)
@@ -91,7 +91,7 @@ class TestMassMailBlPerformance(TestMassMailPerformanceBase):
})
# runbot needs +62 compared to local
with self.assertQueryCount(__system__=2005, marketing=2005):
with self.assertQueryCount(__system__=2008, marketing=2008):
mailing.action_send_mail()
self.assertEqual(mailing.sent, 50)
odoo/addons/base/__manifest__.py
+1
View File
@@ -22,6 +22,7 @@ The kernel of Odoo, needed for all installation.
'data/res_currency_data.xml',
'data/res_country_data.xml',
'data/ir_demo_data.xml',
'data/ir_config_parameter_data.xml',
'security/base_groups.xml',
'security/base_security.xml',
'views/base_menus.xml',
odoo/addons/base/data/ir_config_parameter_data.xml
+23
View File
@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<data noupdate="1">
<!-- Catchall Email Alias -->
<record id="icp_mail_catchall_alias" model="ir.config_parameter">
<field name="key">mail.catchall.alias</field>
<field name="value">catchall</field>
</record>
<!-- Bounce Email Alias -->
<record id="icp_mail_bounce_alias" model="ir.config_parameter">
<field name="key">mail.bounce.alias</field>
<field name="value">bounce</field>
</record>
<!-- Notifications -->
<record id="icp_mail_default_from" model="ir.config_parameter">
<field name="key">mail.default.from</field>
<field name="value">notifications</field>
</record>
</data>
</odoo>
odoo/addons/base/models/ir_mail_server.py
+248 -83
View File
@@ -3,23 +3,28 @@
from email.message import EmailMessage
from email.utils import make_msgid
import base64
import datetime
import email
import email.policy
import html2text
import idna
import logging
import re
import smtplib
from socket import gaierror, timeout
from ssl import SSLError
import ssl
import sys
import threading
import html2text
import idna
from socket import gaierror, timeout
from OpenSSL import crypto as SSLCrypto
from OpenSSL.crypto import Error as SSLCryptoError, FILETYPE_PEM
from OpenSSL.SSL import Error as SSLError
from urllib3.contrib.pyopenssl import PyOpenSSLContext
from odoo import api, fields, models, tools, _
from odoo.exceptions import UserError
from odoo.tools import ustr, pycompat, formataddr, encapsulate_email, email_domain_extract
from odoo.tools import ustr, pycompat, formataddr, email_normalize, encapsulate_email, email_domain_extract, email_domain_normalize
_logger = logging.getLogger(__name__)
@@ -84,8 +89,13 @@ class IrMailServer(models.Model):
"specified for outgoing emails (To/Cc/Bcc)")
name = fields.Char(string='Description', required=True, index=True)
from_filter = fields.Char(
"From Filter",
help='Define for which email address or domain this server can be used.\n'
'e.g.: "notification@odoo.com" or "odoo.com"')
smtp_host = fields.Char(string='SMTP Server', required=True, help="Hostname or IP of SMTP server")
smtp_port = fields.Integer(string='SMTP Port', required=True, default=25, help="SMTP Port. Usually 465 for SSL, and 25 or 587 for other cases.")
smtp_authentication = fields.Selection([('login', 'Username'), ('certificate', 'SSL Certificate')], string='Authenticate with', required=True, default='login')
smtp_user = fields.Char(string='Username', help="Optional username for SMTP authentication", groups='base.group_system')
smtp_pass = fields.Char(string='Password', help="Optional password for SMTP authentication", groups='base.group_system')
smtp_encryption = fields.Selection([('none', 'None'),
@@ -96,6 +106,12 @@ class IrMailServer(models.Model):
"- None: SMTP sessions are done in cleartext.\n"
"- TLS (STARTTLS): TLS encryption is requested at start of SMTP session (Recommended)\n"
"- SSL/TLS: SMTP sessions are encrypted with SSL/TLS through a dedicated port (default: 465)")
smtp_ssl_certificate = fields.Binary(
'SSL Certificate', groups='base.group_system', attachment=False,
help='SSL certificate used for authentication')
smtp_ssl_private_key = fields.Binary(
'SSL Private Key', groups='base.group_system', attachment=False,
help='SSL private key used for authentication')
smtp_debug = fields.Boolean(string='Debugging', help="If enabled, the full output of SMTP sessions will "
"be written to the server log at DEBUG level "
"(this is very verbose and may include confidential info!)")
@@ -103,6 +119,15 @@ class IrMailServer(models.Model):
"is used. Default priority is 10 (smaller number = higher priority)")
active = fields.Boolean(default=True)
@api.constrains('smtp_ssl_certificate', 'smtp_ssl_private_key')
def _check_smtp_ssl_files(self):
"""We must provided both files or none."""
for mail_server in self:
if mail_server.smtp_ssl_certificate and not mail_server.smtp_ssl_private_key:
raise UserError(_('SSL private key is missing for %s.', mail_server.name))
elif mail_server.smtp_ssl_private_key and not mail_server.smtp_ssl_certificate:
raise UserError(_('SSL certificate is missing for %s.', mail_server.name))
def _get_test_email_addresses(self):
self.ensure_one()
email_from = self.env.user.email
@@ -146,6 +171,8 @@ class IrMailServer(models.Model):
raise UserError(_("The server has closed the connection unexpectedly. Check configuration served on this port number.\n %s", ustr(e.strerror)))
except smtplib.SMTPResponseException as e:
raise UserError(_("Server replied with following exception:\n %s", ustr(e.smtp_error)))
except smtplib.SMTPNotSupportedError as e:
raise UserError(_("An option is not supported by the server:\n %s", e.strerror))
except smtplib.SMTPException as e:
raise UserError(_("An SMTP exception occurred. Check port number and connection security type.\n %s", ustr(e)))
except SSLError as e:
@@ -172,7 +199,7 @@ class IrMailServer(models.Model):
}
def connect(self, host=None, port=None, user=None, password=None, encryption=None,
smtp_debug=False, mail_server_id=None):
smtp_from=None, ssl_certificate=None, ssl_private_key=None, smtp_debug=False, mail_server_id=None):
"""Returns a new SMTP connection to the given SMTP server.
When running in test mode, this method does nothing and returns `None`.
@@ -181,36 +208,79 @@ class IrMailServer(models.Model):
:param user: optional username to authenticate with
:param password: optional password to authenticate with
:param string encryption: optional, ``'ssl'`` | ``'starttls'``
:param smtp_from: FROM SMTP envelop, used to find the best mail server
:param ssl_certificate: filename of the SSL certificate used for authentication
Used when no mail server is given and overwrite the odoo-bin argument "smtp_ssl_certificate"
:param ssl_private_key: filename of the SSL private key used for authentication
Used when no mail server is given and overwrite the odoo-bin argument "smtp_ssl_private_key"
:param bool smtp_debug: toggle debugging of SMTP sessions (all i/o
will be output in logs)
:param mail_server_id: ID of specific mail server to use (overrides other parameters)
"""
# Do not actually connect while running in test mode
if getattr(threading.currentThread(), 'testing', False):
return None
if self._is_test_mode():
return
mail_server = smtp_encryption = None
if mail_server_id:
mail_server = self.sudo().browse(mail_server_id)
elif not host:
mail_server = self.sudo().search([], order='sequence', limit=1)
mail_server, smtp_from = self.sudo()._find_mail_server(smtp_from)
ssl_context = None
if mail_server:
smtp_server = mail_server.smtp_host
smtp_port = mail_server.smtp_port
smtp_user = mail_server.smtp_user
smtp_password = mail_server.smtp_pass
if mail_server.smtp_authentication == "login":
smtp_user = mail_server.smtp_user
smtp_password = mail_server.smtp_pass
else:
smtp_user = None
smtp_password = None
smtp_encryption = mail_server.smtp_encryption
smtp_debug = smtp_debug or mail_server.smtp_debug
from_filter = mail_server.from_filter
if (mail_server.smtp_authentication == "certificate"
and mail_server.smtp_ssl_certificate
and mail_server.smtp_ssl_private_key):
try:
ssl_context = PyOpenSSLContext(ssl.PROTOCOL_TLS)
smtp_ssl_certificate = base64.b64decode(mail_server.smtp_ssl_certificate)
certificate = SSLCrypto.load_certificate(FILETYPE_PEM, smtp_ssl_certificate)
smtp_ssl_private_key = base64.b64decode(mail_server.smtp_ssl_private_key)
private_key = SSLCrypto.load_privatekey(FILETYPE_PEM, smtp_ssl_private_key)
ssl_context._ctx.use_certificate(certificate)
ssl_context._ctx.use_privatekey(private_key)
# Check that the private key match the certificate
ssl_context._ctx.check_privatekey()
except SSLCryptoError as e:
raise UserError(_('The private key or the certificate is not a valid file. \n%s', str(e)))
except SSLError as e:
raise UserError(_('Could not load your certificate / private key. \n%s', str(e)))
else:
# we were passed individual smtp parameters or nothing and there is no default server
smtp_server = host or tools.config.get('smtp_server')
smtp_port = tools.config.get('smtp_port', 25) if port is None else port
smtp_user = user or tools.config.get('smtp_user')
smtp_password = password or tools.config.get('smtp_password')
from_filter = tools.config.get('from_filter')
smtp_encryption = encryption
if smtp_encryption is None and tools.config.get('smtp_ssl'):
smtp_encryption = 'starttls' # smtp_ssl => STARTTLS as of v7
smtp_ssl_certificate_filename = ssl_certificate or tools.config.get('smtp_ssl_certificate_filename')
smtp_ssl_private_key_filename = ssl_private_key or tools.config.get('smtp_ssl_private_key_filename')
if smtp_ssl_certificate_filename and smtp_ssl_private_key_filename:
try:
ssl_context = PyOpenSSLContext(ssl.PROTOCOL_TLS)
ssl_context.load_cert_chain(smtp_ssl_certificate_filename, keyfile=smtp_ssl_private_key_filename)
# Check that the private key match the certificate
ssl_context._ctx.check_privatekey()
except SSLCryptoError as e:
raise UserError(_('The private key or the certificate is not a valid file. \n%s', str(e)))
except SSLError as e:
raise UserError(_('Could not load your certificate / private key. \n%s', str(e)))
if not smtp_server:
raise UserError(
@@ -228,6 +298,7 @@ class IrMailServer(models.Model):
connection = smtplib.SMTP_SSL(smtp_server, smtp_port, timeout=SMTP_TIMEOUT)
else:
connection = smtplib.SMTP(smtp_server, smtp_port, timeout=SMTP_TIMEOUT)
connection.set_debuglevel(smtp_debug)
if smtp_encryption == 'starttls':
# starttls() will perform ehlo() if needed first
@@ -236,7 +307,7 @@ class IrMailServer(models.Model):
# (as per RFC 3207) so for example any AUTH
# capability that appears only on encrypted channels
# will be correctly detected for next step
connection.starttls()
connection.starttls(context=ssl_context)
if smtp_user:
# Attempt authentication - will raise if AUTH service not supported
@@ -249,6 +320,11 @@ class IrMailServer(models.Model):
# Anyway, as it may have been sent by login(), all subsequent usages should consider this command as sent.
connection.ehlo_or_helo_if_needed()
# Store the "from_filter" of the mail server / odoo-bin argument to know if we
# need to change the FROM headers or not when we will prepare the mail message
connection.from_filter = from_filter
connection.smtp_from = smtp_from
return connection
def build_email(self, email_from, email_to, subject, body, email_cc=None, email_bcc=None, reply_to=False,
@@ -305,12 +381,7 @@ 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
@@ -338,35 +409,6 @@ 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.
@@ -407,10 +449,77 @@ class IrMailServer(models.Model):
return "%s@%s" % (email_from, domain)
return tools.config.get("email_from")
def _prepare_email_message(self, message, smtp_session):
"""Prepare the SMTP information (from, to, message) before sending.
:param message: the email.message.Message to send, information like the
Return-Path, the From, etc... will be used to find the smtp_from and to smtp_to
:param smtp_session: the opened SMTP session to use to authenticate the sender
:return: smtp_from, smtp_to_list, message
smtp_from: email to used during the authentication to the mail server
smtp_to_list: list of email address which will receive the email
message: the email.message.Message to send
"""
# Use the default bounce address **only if** no Return-Path was
# provided by caller. Caller may be using Variable Envelope Return
# Path (VERP) to detect no-longer valid email addresses.
bounce_address = message['Return-Path'] or self._get_default_bounce_address() or message['From']
smtp_from = message['From'] or bounce_address
assert smtp_from, "The Return-Path or From header is required for any outbound email"
email_to = message['To']
email_cc = message['Cc']
email_bcc = message['Bcc']
del message['Bcc']
# All recipient addresses must only contain ASCII characters
smtp_to_list = [
address
for base in [email_to, email_cc, email_bcc]
for address in extract_rfc2822_addresses(base)
if address
]
assert smtp_to_list, self.NO_VALID_RECIPIENT
x_forge_to = message['X-Forge-To']
if x_forge_to:
# `To:` header forged, e.g. for posting on mail.channels, to avoid confusion
del message['X-Forge-To']
del message['To'] # avoid multiple To: headers!
message['To'] = x_forge_to
# Try to not spoof the mail from headers
from_filter = getattr(smtp_session, 'from_filter', False)
smtp_from = getattr(smtp_session, 'smtp_from', False) or smtp_from
notifications_email = email_normalize(self._get_default_from_address())
if notifications_email and smtp_from == notifications_email and message['From'] != notifications_email:
smtp_from = encapsulate_email(message['From'], notifications_email)
if message['From'] != smtp_from:
del message['From']
message['From'] = smtp_from
# Check if it's still possible to put the bounce address as smtp_from
if self._match_from_filter(bounce_address, from_filter):
# Mail headers FROM will be spoofed to be able to receive bounce notifications
# Because the mail server support the domain of the bounce address
smtp_from = bounce_address
# The email's "Envelope From" (Return-Path) must only contain ASCII characters.
smtp_from_rfc2822 = extract_rfc2822_addresses(smtp_from)
assert smtp_from_rfc2822, (
f"Malformed 'Return-Path' or 'From' address: {smtp_from} - "
"It should contain one valid plain ASCII email")
smtp_from = smtp_from_rfc2822[-1]
return smtp_from, smtp_to_list, message
@api.model
def send_email(self, message, mail_server_id=None, smtp_server=None, smtp_port=None,
smtp_user=None, smtp_password=None, smtp_encryption=None, smtp_debug=False,
smtp_session=None):
smtp_user=None, smtp_password=None, smtp_encryption=None,
smtp_ssl_certificate=None, smtp_ssl_private_key=None,
smtp_debug=False, smtp_session=None):
"""Sends an email directly (no queuing).
No retries are done, the caller should handle MailDeliveryException in order to ensure that
@@ -436,53 +545,28 @@ class IrMailServer(models.Model):
:param smtp_port: optional SMTP port, if mail_server_id is not passed
:param smtp_user: optional SMTP user, if mail_server_id is not passed
:param smtp_password: optional SMTP password to use, if mail_server_id is not passed
:param smtp_ssl_certificate: filename of the SSL certificate used for authentication
:param smtp_ssl_private_key: filename of the SSL private key used for authentication
:param smtp_debug: optional SMTP debug flag, if mail_server_id is not passed
:return: the Message-ID of the message that was just sent, if successfully sent, otherwise raises
MailDeliveryException and logs root cause.
"""
# Use the default bounce address **only if** no Return-Path was
# provided by caller. Caller may be using Variable Envelope Return
# Path (VERP) to detect no-longer valid email addresses.
smtp_from = message['Return-Path'] or self._get_default_bounce_address() or message['From']
assert smtp_from, "The Return-Path or From header is required for any outbound email"
smtp = smtp_session
if not smtp:
smtp = self.connect(
smtp_server, smtp_port, smtp_user, smtp_password, smtp_encryption,
smtp_from=message['From'], ssl_certificate=smtp_ssl_certificate, ssl_private_key=smtp_ssl_private_key,
smtp_debug=smtp_debug, mail_server_id=mail_server_id,)
# The email's "Envelope From" (Return-Path), and all recipient addresses must only contain ASCII characters.
from_rfc2822 = extract_rfc2822_addresses(smtp_from)
assert from_rfc2822, ("Malformed 'Return-Path' or 'From' address: %r - "
"It should contain one valid plain ASCII email") % smtp_from
# use last extracted email, to support rarities like 'Support@MyComp <support@mycompany.com>'
smtp_from = from_rfc2822[-1]
email_to = message['To']
email_cc = message['Cc']
email_bcc = message['Bcc']
del message['Bcc']
smtp_to_list = [
address
for base in [email_to, email_cc, email_bcc]
for address in extract_rfc2822_addresses(base)
if address
]
assert smtp_to_list, self.NO_VALID_RECIPIENT
x_forge_to = message['X-Forge-To']
if x_forge_to:
# `To:` header forged, e.g. for posting on mail.channels, to avoid confusion
del message['X-Forge-To']
del message['To'] # avoid multiple To: headers!
message['To'] = x_forge_to
smtp_from, smtp_to_list, message = self._prepare_email_message(message, smtp)
# Do not actually send emails in testing mode!
if getattr(threading.currentThread(), 'testing', False) or self.env.registry.in_test_mode():
if self._is_test_mode():
_test_logger.info("skip sending email in test mode")
return message['Message-Id']
try:
message_id = message['Message-Id']
smtp = smtp_session
smtp = smtp or self.connect(
smtp_server, smtp_port, smtp_user, smtp_password,
smtp_encryption, smtp_debug, mail_server_id=mail_server_id)
if sys.version_info < (3, 7, 4):
# header folding code is buggy and adds redundant carriage
@@ -511,6 +595,79 @@ class IrMailServer(models.Model):
raise MailDeliveryException(_("Mail Delivery Failed"), msg)
return message_id
def _find_mail_server(self, email_from, mail_servers=None):
"""Find the appropriate mail server for the given email address.
Returns: Record<ir.mail_server>, email_from
- Mail server to use to send the email (None if we use the odoo-bin arguments)
- Email FROM to use to send the email (in some case, it might be impossible
to use the given email address directly if no mail server is configured for)
"""
email_from_normalized = email_normalize(email_from)
email_from_domain = email_domain_extract(email_from_normalized)
notifications_email = email_normalize(self._get_default_from_address())
notifications_domain = email_domain_extract(notifications_email)
if mail_servers is None:
mail_servers = self.sudo().search([], order='sequence')
# 1. Try to find a mail server for the right mail from
mail_server = mail_servers.filtered(lambda m: email_normalize(m.from_filter) == email_from_normalized)
if mail_server:
return mail_server[0], email_from
mail_server = mail_servers.filtered(lambda m: email_domain_normalize(m.from_filter) == email_from_domain)
if mail_server:
return mail_server[0], email_from
# 2. Try to find a mail server for <notifications@domain.com>
if notifications_email:
mail_server = mail_servers.filtered(lambda m: email_normalize(m.from_filter) == notifications_email)
if mail_server:
return mail_server[0], notifications_email
mail_server = mail_servers.filtered(lambda m: email_domain_normalize(m.from_filter) == notifications_domain)
if mail_server:
return mail_server[0], notifications_email
# 3. Take the first mail server without "from_filter" because
# nothing else has been found... Will spoof the FROM because
# we have no other choices
mail_server = mail_servers.filtered(lambda m: not m.from_filter)
if mail_server:
return mail_server[0], email_from
# 4. Return the first mail server even if it was configured for another domain
if mail_servers:
return mail_servers[0], email_from
# 5: SMTP config in odoo-bin arguments
from_filter = tools.config.get('from_filter')
if self._match_from_filter(email_from, from_filter):
return None, email_from
if notifications_email and self._match_from_filter(notifications_email, from_filter):
return None, notifications_email
return None, email_from
@api.model
def _match_from_filter(self, email_from, from_filter):
"""Return True is the given email address match the "from_filter" field.
The from filter can be Falsy (always match),
a domain name or an full email address.
"""
if not from_filter:
return True
normalized_mail_from = email_normalize(email_from)
if '@' in from_filter:
return email_normalize(from_filter) == normalized_mail_from
return email_domain_extract(normalized_mail_from) == email_domain_normalize(from_filter)
@api.onchange('smtp_encryption')
def _onchange_encryption(self):
result = {}
@@ -524,3 +681,11 @@ class IrMailServer(models.Model):
else:
self.smtp_port = 25
return result
def _is_test_mode(self):
"""Return True if we are running the tests, so we do not send real emails.
Can be overridden in tests after mocking the SMTP lib to test in depth the
outgoing mail server.
"""
return getattr(threading.currentThread(), 'testing', False) or self.env.registry.in_test_mode()
odoo/addons/base/tests/__init__.py
+1
View File
@@ -47,3 +47,4 @@ from . import test_form_create
from . import test_cloc
from . import test_profiler
from . import test_pdf
from . import test_ir_mail_server
odoo/addons/base/tests/common.py
+139
View File
@@ -1,6 +1,9 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from contextlib import contextmanager
from unittest.mock import patch
from odoo.tests.common import TransactionCase, HttpCase
from odoo import Command
@@ -197,3 +200,139 @@ class HttpCaseWithUserPortal(HttpCase):
'partner_id': self.partner_portal.id,
'groups_id': [Command.set([self.env.ref('base.group_portal').id])],
})
class MockSmtplibCase:
"""Class which allows you to mock the smtplib feature, to be able to test in depth the
sending of emails. Unlike "MockEmail" which mocks mainly the <ir.mail_server> methods,
here we mainly mock the smtplib to be able to test the <ir.mail_server> model.
"""
@contextmanager
def mock_smtplib_connection(self):
self.emails = []
origin = self
class TestingSMTPSession:
"""SMTP session object returned during the testing.
So we do not connect to real SMTP server. Store the mail
server id used for the SMTP connection and other information.
Can be mocked for testing to know which with arguments the email was sent.
"""
def quit(self):
pass
def send_message(self, message, smtp_from, smtp_to_list):
origin.emails.append({
'smtp_from': smtp_from,
'smtp_to_list': smtp_to_list,
'message': message.as_string(),
'from_filter': self.from_filter,
})
def sendmail(self, smtp_from, smtp_to_list, message_str, mail_options):
origin.emails.append({
'smtp_from': smtp_from,
'smtp_to_list': smtp_to_list,
'message': message_str,
'from_filter': self.from_filter,
})
def set_debuglevel(self, smtp_debug):
pass
def ehlo_or_helo_if_needed(self):
pass
def login(self, user, password):
pass
self.testing_smtp_session = TestingSMTPSession()
IrMailServer = self.env['ir.mail_server']
connect = IrMailServer.connect
find_mail_server = IrMailServer._find_mail_server
with patch.object(type(IrMailServer), '_is_test_mode', lambda self: False), \
patch('smtplib.SMTP_SSL', side_effect=lambda *args, **kwargs: self.testing_smtp_session), \
patch('smtplib.SMTP', side_effect=lambda *args, **kwargs: self.testing_smtp_session), \
patch.object(type(IrMailServer), 'connect', side_effect=connect) as connect_mocked, \
patch.object(type(IrMailServer), '_find_mail_server', side_effect=find_mail_server) as find_mail_server_mocked:
self.connect_mocked = connect_mocked
self.find_mail_server_mocked = find_mail_server_mocked
yield
def assert_email_sent_smtp(self, smtp_from=None, smtp_to_list=None, message_from=None, from_filter=None, emails_count=1):
"""Check that the given email has been sent.
If one of the parameter is None, it's just ignored and not used to retrieve the email.
:param smtp_from: FROM used for the authentication to the mail server
:param smtp_to_list: List of destination email address
:param message_from: FROM used in the SMTP headers
:param from_filter: from_filter of the <ir.mail_server> used to send the email
Can use a lambda to check the value
:param emails_count: the number of emails which should match the condition
:return: True if at least one email has been found with those parameters
"""
matching_emails = filter(
lambda email:
(smtp_from is None or (
smtp_from(email['smtp_from'])
if callable(smtp_from)
else smtp_from == email['smtp_from'])
)
and (smtp_to_list is None or smtp_to_list == email['smtp_to_list'])
and (message_from is None or 'From: %s' % message_from in email['message'])
and (from_filter is None or from_filter == email['from_filter']),
self.emails,
)
matching_emails_count = len(list(matching_emails))
self.assertTrue(
matching_emails_count == emails_count,
msg='Emails not sent, %i emails match the condition but %i are expected' % (matching_emails_count, emails_count),
)
@classmethod
def _init_mail_servers(cls):
cls.env['ir.config_parameter'].sudo().set_param('mail.catchall.domain', 'test.com')
cls.env['ir.config_parameter'].sudo().set_param('mail.default.from', 'notifications')
cls.env['ir.config_parameter'].sudo().set_param('mail.bounce.alias', 'bounce')
cls.alias_bounce = 'bounce'
cls.alias_domain = 'test.com'
cls.env['ir.mail_server'].search([]).unlink()
ir_mail_server_values = {
'smtp_host': 'smtp_host',
'smtp_encryption': 'none',
}
(
cls.server_domain,
cls.server_user,
cls.server_notification,
cls.server_default,
) = cls.env['ir.mail_server'].create([
{
'name': 'Domain based server',
'from_filter': 'test.com',
** ir_mail_server_values,
}, {
'name': 'User specific server',
'from_filter': 'specific_user@test.com',
** ir_mail_server_values,
}, {
'name': 'Server Notifications',
'from_filter': 'notifications@test.com',
** ir_mail_server_values,
}, {
'name': 'Server No From Filter',
'from_filter': False,
** ir_mail_server_values,
},
])
odoo/addons/base/tests/test_ir_mail_server.py
+362
View File
@@ -0,0 +1,362 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from unittest.mock import patch
from odoo.addons.base.tests.common import MockSmtplibCase
from odoo.tests.common import TransactionCase
from odoo.tools import mute_logger
class TestIrMailServer(TransactionCase, MockSmtplibCase):
def setUp(self):
self._init_mail_servers()
def _build_email(self, mail_from, return_path=None):
return self.env['ir.mail_server'].build_email(
email_from=mail_from,
email_to='dest@example-é.com',
subject='subject', body='body',
headers={'Return-Path': return_path} if return_path else None
)
def test_match_from_filter(self):
"""Test the from_filter field on the "ir.mail_server"."""
match_from_filter = self.env['ir.mail_server']._match_from_filter
# Should match
tests = [
('admin@mail.example.com', 'mail.example.com'),
('admin@mail.example.com', 'mail.EXAMPLE.com'),
('admin@mail.example.com', 'admin@mail.example.com'),
('admin@mail.example.com', False),
('"fake@test.com" <admin@mail.example.com>', 'mail.example.com'),
('"fake@test.com" <ADMIN@mail.example.com>', 'mail.example.com'),
]
for email, from_filter in tests:
self.assertTrue(match_from_filter(email, from_filter))
# Should not match
tests = [
('admin@mail.example.com', 'test@mail.example.com'),
('admin@mail.example.com', 'test.com'),
('admin@mail.example.com', 'mail.éxample.com'),
('admin@mmail.example.com', 'mail.example.com'),
('admin@mail.example.com', 'mmail.example.com'),
('"admin@mail.example.com" <fake@test.com>', 'mail.example.com'),
]
for email, from_filter in tests:
self.assertFalse(match_from_filter(email, from_filter))
@mute_logger('odoo.models.unlink')
def test_mail_server_priorities(self):
"""Test if we choose the right mail server to send an email.
Priorities are
1. Forced mail server (e.g.: in mass mailing)
- If the "from_filter" of the mail server match the notification email
use the notifications email in the "From header"
- Otherwise spoof the "From" (because we force the mail server but we don't
know which email use to send it)
2. A mail server for which the "from_filter" match the "From" header
3. A mail server for which the "from_filter" match the domain of the "From" header
4. The mail server used for notifications
5. A mail server without "from_filter" (and so spoof the "From" header because we
do not know for which email address it can be used)
"""
# sanity checks
self.assertTrue(self.env['ir.mail_server']._get_default_from_address(), 'Notifications email must be set for testing')
self.assertTrue(self.env['ir.mail_server']._get_default_bounce_address(), 'Bounce email must be set for testing')
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='specific_user@test.com')
self.assertEqual(mail_server, self.server_user)
self.assertEqual(mail_from, 'specific_user@test.com')
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='"Name name@strange.name" <specific_user@test.com>')
self.assertEqual(mail_server, self.server_user, 'Must extract email from full name')
self.assertEqual(mail_from, '"Name name@strange.name" <specific_user@test.com>', 'Must keep the given mail from')
# Should not be case sensitive
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='specific_user@test.com')
self.assertEqual(mail_server, self.server_user, 'Mail from is case insensitive')
self.assertEqual(mail_from, 'specific_user@test.com', 'Should not change the mail from')
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='unknown_email@test.com')
self.assertEqual(mail_server, self.server_domain)
self.assertEqual(mail_from, 'unknown_email@test.com')
# Cover a different condition that the "email case insensitive" test
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='unknown_email@TEST.COM')
self.assertEqual(mail_server, self.server_domain, 'Domain is case insensitive')
self.assertEqual(mail_from, 'unknown_email@TEST.COM', 'Domain is case insensitive')
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='"Test" <test@unknown_domain.com>')
self.assertEqual(mail_server, self.server_notification, 'Should take the notification email')
self.assertEqual(mail_from, 'notifications@test.com')
# remove the notifications email to simulate a mis-configured Odoo database
# so we do not have the choice, we have to spoof the FROM
# (otherwise we can not send the email)
self.env['ir.config_parameter'].sudo().set_param('mail.catchall.domain', False)
with mute_logger('odoo.addons.base.models.ir_mail_server'):
mail_server, mail_from = self.env['ir.mail_server']._find_mail_server(email_from='test@unknown_domain.com')
self.assertEqual(mail_server.from_filter, False, 'No notifications email set, must be forced to spoof the FROM')
self.assertEqual(mail_from, 'test@unknown_domain.com')
@mute_logger('odoo.models.unlink')
def test_mail_server_send_email(self):
IrMailServer = self.env['ir.mail_server']
default_bounce_adress = self.env['ir.mail_server']._get_default_bounce_address()
# A mail server is configured for the email
with self.mock_smtplib_connection():
message = self._build_email(mail_from='specific_user@test.com')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from='specific_user@test.com',
message_from='specific_user@test.com',
from_filter='specific_user@test.com',
)
# No mail server are configured for the email address,
# so it will use the notifications email instead and encapsulate the old email
with self.mock_smtplib_connection():
message = self._build_email(mail_from='"Name" <test@unknown_domain.com>')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from='notifications@test.com',
message_from='"Name (test@unknown_domain.com)" <notifications@test.com>',
from_filter='notifications@test.com',
)
# Same situation, but the original email has no name part
with self.mock_smtplib_connection():
message = self._build_email(mail_from='test@unknown_domain.com')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from='notifications@test.com',
message_from='"test@unknown_domain.com" <notifications@test.com>',
from_filter='notifications@test.com',
)
# A mail server is configured for the entire domain name, so we can use the bounce
# email address because the mail server supports it
with self.mock_smtplib_connection():
message = self._build_email(mail_from='unknown_name@test.com')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='unknown_name@test.com',
from_filter='test.com',
)
# remove the notification server
# so <notifications@test.com> will use the <test.com> mail server
self.server_notification.unlink()
# The mail server configured for the notifications email has been removed
# but we can still use the mail server configured for test.com
# and so we will be able to use the bounce address
# because we use the mail server for "test.com"
with self.mock_smtplib_connection():
message = self._build_email(mail_from='"Name" <test@unknown_domain.com>')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='"Name (test@unknown_domain.com)" <notifications@test.com>',
from_filter='test.com',
)
# Test that the mail from / recipient envelop are encoded using IDNA
self.env['ir.config_parameter'].sudo().set_param('mail.catchall.domain', 'ééééééé.com')
with self.mock_smtplib_connection():
message = self._build_email(mail_from='test@ééééééé.com')
IrMailServer.send_email(message)
self.assertEqual(len(self.emails), 1)
self.assert_email_sent_smtp(
smtp_from='bounce@xn--9caaaaaaa.com',
smtp_to_list=['dest@xn--example--i1a.com'],
message_from='test@=?utf-8?b?w6nDqcOpw6nDqcOpw6k=?=.com',
from_filter=False,
)
@mute_logger('odoo.models.unlink')
def test_mail_server_send_email_smtp_session(self):
"""Test all the cases when we provide the SMTP session.
The results must be the same as passing directly the parameter to "send_email".
"""
IrMailServer = self.env['ir.mail_server']
default_bounce_adress = self.env['ir.mail_server']._get_default_bounce_address()
# A mail server is configured for the email
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='specific_user@test.com')
message = self._build_email(mail_from='specific_user@test.com')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from='specific_user@test.com',
message_from='specific_user@test.com',
from_filter='specific_user@test.com',
)
# No mail server are configured for the email address,
# so it will use the notifications email instead and encapsulate the old email
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='"Name" <test@unknown_domain.com>')
message = self._build_email(mail_from='"Name" <test@unknown_domain.com>')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from='notifications@test.com',
message_from='"Name (test@unknown_domain.com)" <notifications@test.com>',
from_filter='notifications@test.com',
)
# A mail server is configured for the entire domain name, so we can use the bounce
# email address because the mail server supports it
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='unknown_name@test.com')
message = self._build_email(mail_from='unknown_name@test.com')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='unknown_name@test.com',
from_filter='test.com',
)
# remove the notification server
# so <notifications@test.com> will use the <test.com> mail server
self.server_notification.unlink()
# The mail server configured for the notifications email has been removed
# but we can still use the mail server configured for test.com
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='"Name" <test@unknown_domain.com>')
message = self._build_email(mail_from='"Name" <test@unknown_domain.com>')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='"Name (test@unknown_domain.com)" <notifications@test.com>',
from_filter='test.com',
)
@mute_logger('odoo.models.unlink')
@patch.dict("odoo.tools.config.options", {"from_filter": "test.com"})
def test_mail_server_binary_arguments_domain(self):
"""Test the configuration provided in the odoo-bin arguments.
This config is used when no mail server exists.
"""
IrMailServer = self.env['ir.mail_server']
default_bounce_adress = self.env['ir.mail_server']._get_default_bounce_address()
# Remove all mail server so we will use the odoo-bin arguments
self.env['ir.mail_server'].search([]).unlink()
self.assertFalse(self.env['ir.mail_server'].search([]))
# Use an email in the domain of the "from_filter"
with self.mock_smtplib_connection():
message = self._build_email(mail_from='specific_user@test.com')
IrMailServer.send_email(message)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='specific_user@test.com',
from_filter='test.com',
)
# Test if the domain name is normalized before comparison
with self.mock_smtplib_connection():
message = self._build_email(mail_from='specific_user@test.com')
IrMailServer.send_email(message)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='specific_user@test.com',
from_filter='test.com',
)
# Use an email outside of the domain of the "from_filter"
# So we will use the notifications email in the headers and the bounce address
# in the envelop because the "from_filter" allows to use the entire domain
with self.mock_smtplib_connection():
message = self._build_email(mail_from='test@unknown_domain.com')
IrMailServer.send_email(message)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='"test@unknown_domain.com" <notifications@test.com>',
from_filter='test.com',
)
@mute_logger('odoo.models.unlink')
@patch.dict("odoo.tools.config.options", {"from_filter": "test.com"})
def test_mail_server_binary_arguments_domain_smtp_session(self):
"""Test the configuration provided in the odoo-bin arguments.
This config is used when no mail server exists.
Use a pre-configured SMTP session.
"""
IrMailServer = self.env['ir.mail_server']
default_bounce_adress = self.env['ir.mail_server']._get_default_bounce_address()
# Remove all mail server so we will use the odoo-bin arguments
self.env['ir.mail_server'].search([]).unlink()
self.assertFalse(self.env['ir.mail_server'].search([]))
# Use an email in the domain of the "from_filter"
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='specific_user@test.com')
message = self._build_email(mail_from='specific_user@test.com')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='specific_user@test.com',
from_filter='test.com',
)
# Use an email outside of the domain of the "from_filter"
# So we will use the notifications email in the headers and the bounce address
# in the envelop because the "from_filter" allows to use the entire domain
with self.mock_smtplib_connection():
smtp_session = IrMailServer.connect(smtp_from='test@unknown_domain.com')
message = self._build_email(mail_from='test@unknown_domain.com')
IrMailServer.send_email(message, smtp_session=smtp_session)
self.connect_mocked.assert_called_once()
self.assert_email_sent_smtp(
smtp_from=default_bounce_adress,
message_from='"test@unknown_domain.com" <notifications@test.com>',
from_filter='test.com',
)
odoo/addons/base/tests/test_mail.py
+20 -71
View File
@@ -8,10 +8,11 @@ import email.message
import re
import threading
from odoo.addons.base.models.ir_mail_server import extract_rfc2822_addresses
from odoo.tests.common import BaseCase, TransactionCase
from odoo.tools import (
is_html_empty, html_sanitize, append_content_to_html, plaintext2html,
email_split,
email_split, email_domain_normalize,
misc, formataddr,
prepend_html_content,
)
@@ -435,6 +436,23 @@ class TestEmailTools(BaseCase):
with self.subTest(pair=pair, charset=charset):
self.assertEqual(formataddr(pair, charset), expected)
def test_extract_rfc2822_addresses(self):
tests = [
('"Admin" <admin@example.com>', ['admin@example.com']),
('"Admin" <admin@example.com>, Demo <demo@test.com>', ['admin@example.com', 'demo@test.com']),
('admin@example.com', ['admin@example.com']),
('"Admin" <admin@example.com>, Demo <malformed email>', ['admin@example.com']),
('admin@éxample.com', ['admin@xn--xample-9ua.com']),
]
for (rfc2822_email, expected) in tests:
self.assertEqual(extract_rfc2822_addresses(rfc2822_email), expected)
def test_email_domain_normalize(self):
self.assertEqual(email_domain_normalize("Test.Com"), "test.com", "Should have normalized the domain")
self.assertEqual(email_domain_normalize("email@test.com"), False, "The domain is not valid, should return False")
self.assertEqual(email_domain_normalize(False), False, "The domain is not valid, should return False")
class EmailConfigCase(TransactionCase):
@patch.dict("odoo.tools.config.options", {"email_from": "settings@example.com"})
@@ -457,76 +475,6 @@ class EmailConfigCase(TransactionCase):
)
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):
@@ -544,6 +492,7 @@ class TestEmailMessage(TransactionCase):
"""SMTP stub"""
def __init__(this):
this.email_sent = False
this.from_filter = 'example.com'
# Python 3 before 3.7.4
def sendmail(this, smtp_from, smtp_to_list, message_str,
odoo/addons/base/views/ir_mail_server_views.xml
+12 -6
View File
@@ -8,6 +8,7 @@
<sheet>
<group col="4">
<field name="name"/>
<field name="from_filter"/>
<field name="sequence"/>
<field name="active" widget="boolean_toggle"/>
</group>
@@ -16,13 +17,18 @@
<field name="smtp_port" options="{'format': false}"/>
<field name="smtp_debug" groups="base.group_no_one"/>
</group>
<group string="Security and Authentication" colspan="4">
<field name="smtp_encryption"/>
<field name="smtp_user"/>
<field name="smtp_pass" password="True"/>
<button name="test_smtp_connection" type="object" string="Test Connection" icon="fa-television"/>
<group>
<group string="Security and Authentication">
<field name="smtp_authentication"/>
<field name="smtp_encryption"/>
<field name="smtp_user" attrs="{'invisible': [('smtp_authentication', '!=', 'login')]}" force_save="1"/>
<field name="smtp_pass" attrs="{'invisible': [('smtp_authentication', '!=', 'login')]}" password="True" force_save="1"/>
<field name="smtp_ssl_certificate" attrs="{'invisible': [('smtp_authentication', '!=', 'certificate')]}" force_save="1"/>
<field name="smtp_ssl_private_key" attrs="{'invisible': [('smtp_authentication', '!=', 'certificate')]}" force_save="1"/>
<button name="test_smtp_connection" type="object" string="Test Connection" icon="fa-television"/>
</group>
</group>
</sheet>
</sheet>
</form>
</field>
</record>
odoo/tools/config.py
+8 -1
View File
@@ -216,6 +216,8 @@ class configmanager(object):
group = optparse.OptionGroup(parser, "SMTP Configuration")
group.add_option('--email-from', dest='email_from', my_default=False,
help='specify the SMTP email address for sending email')
group.add_option('--from-filter', dest='from_filter', my_default=False,
help='specify for which email address the SMTP configuration can be used')
group.add_option('--smtp', dest='smtp_server', my_default='localhost',
help='specify the SMTP server for sending email')
group.add_option('--smtp-port', dest='smtp_port', my_default=25,
@@ -226,6 +228,10 @@ class configmanager(object):
help='specify the SMTP username for sending email')
group.add_option('--smtp-password', dest='smtp_password', my_default=False,
help='specify the SMTP password for sending email')
group.add_option('--smtp-ssl-certificate-filename', dest='smtp_ssl_certificate_filename', my_default=False,
help='specify the SSL certificate used for authentication')
group.add_option('--smtp-ssl-private-key-filename', dest='smtp_ssl_private_key_filename', my_default=False,
help='specify the SSL private key used for authentication')
parser.add_option_group(group)
group = optparse.OptionGroup(parser, "Database related options")
@@ -437,7 +443,8 @@ class configmanager(object):
keys = ['http_interface', 'http_port', 'longpolling_port', 'http_enable',
'db_name', 'db_user', 'db_password', 'db_host', 'db_sslmode',
'db_port', 'db_template', 'logfile', 'pidfile', 'smtp_port',
'email_from', 'smtp_server', 'smtp_user', 'smtp_password',
'email_from', 'smtp_server', 'smtp_user', 'smtp_password', 'from_filter',
'smtp_ssl_certificate_filename', 'smtp_ssl_private_key_filename',
'db_maxconn', 'import_partial', 'addons_path', 'upgrade_path',
'syslog', 'without_demo', 'screencasts', 'screenshots',
'dbfilter', 'log_level', 'log_db',
odoo/tools/mail.py
+7
View File
@@ -560,6 +560,13 @@ def email_domain_extract(email):
return normalized_email.split('@')[1]
return False
def email_domain_normalize(domain):
"""Return the domain normalized or False if the domain is invalid."""
if not domain or '@' in domain:
return False
return domain.lower()
def url_domain_extract(url):
""" Extract the company domain to be used by IAP services notably. Domain
is extracted from an URL e.g:
requirements.txt
+1
View File
@@ -55,3 +55,4 @@ xlwt==1.3.*
xlrd==1.1.0; python_version < '3.8'
xlrd==1.2.0; python_version >= '3.8'
pypiwin32 ; sys_platform == 'win32'
pyopenssl==19.0.0