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

[FIX] mail: fix blacklist mechanism

Browse Source 
Several quick fixes:

- repair malformed SQL queries and injection problems
- proper token comparisons (consteq)
- better translatable terms (less context-lessplaceholders)
- basic testing of case-(in)sensitivity of blacklist
- safety check for non-implemented cases in search methods for computed
  field

More testing and review is needed for case-sensitivity and email
extraction from generic "addressing fields".
This commit is contained in:
Olivier Dony
2018-09-29 04:47:44 +02:00
parent 2cee34fea2
commit 2ff9b379ef
5 changed files with 72 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/mail/models/mail_blacklist.py
+33 -22
View File
@@ -3,7 +3,6 @@
from odoo import api, fields, models, _
from odoo.exceptions import UserError
from odoo.tools.pycompat import izip
class MailBlackList(models.Model):
@@ -24,21 +23,16 @@ class MailBlackList(models.Model):
@api.model_create_multi
def create(self, values):
""" To avoid crash during import due to unique email, return the existing records if any """
sql = '''SELECT id, email FROM mail_blacklist
WHERE LOWER(email) = any (array[%s])
''' % (', '.join(['%s'] * len(values)))
params = [value['email'].lower() for value in values]
self._cr.execute(sql, params)
records = self._cr.fetchall()
bl_ids = bl_emails = []
if records:
bl_ids, bl_emails = list(izip(*records))
non_blacklisted_records = [value for value in values if value['email'] not in bl_emails]
sql = '''SELECT LOWER(email), id FROM mail_blacklist WHERE LOWER(email) = ANY(%s)'''
emails = [(v['email'] or '').lower() for v in values]
self._cr.execute(sql, (emails,))
bl_entries = dict(self._cr.fetchall())
to_create = [v for v in values
if (v['email'] or '').lower() not in bl_entries]
# TODO DBE Fixme : reorder ids according to incoming ids.
results = super(MailBlackList, self).create(non_blacklisted_records)
return self.env['mail.blacklist'].browse(bl_ids) | results
results = super(MailBlackList, self).create(to_create)
return self.env['mail.blacklist'].browse(bl_entries.values()) | results
def _add(self, email):
record = self.env["mail.blacklist"].with_context(active_test=False).search([('email', '=', email)])
@@ -74,15 +68,28 @@ class MailBlackListMixin(models.AbstractModel):
@api.model
def _search_is_blacklisted(self, operator, value):
# Assumes operator is '=' and value is True or False
if not hasattr(self.env[self._name], "_primary_email"):
raise UserError(_('Invalid primary email field on model %s') % self._name)
if operator != '=':
raise NotImplementedError()
[email_field] = self._primary_email
join_condition = 'INNER' if value else 'RIGHT'
where_clause = '' if value else ' where b.id is null'
blacklisted_sql = '''SELECT a.id FROM mail_blacklist b
%s JOIN %s a
ON b.email = a.%s AND b.active = True%s''' % (join_condition, self._table, email_field, where_clause)
self._cr.execute(blacklisted_sql)
if value:
query = """
SELECT m.id
FROM mail_blacklist bl
JOIN %s m on (LOWER(m.email) = LOWER(bl.email) AND bl.active)
"""
else:
query = """
SELECT m.id
FROM %s m
LEFT JOIN mail_blacklist bl
ON (LOWER(m.email) = LOWER(bl.email) AND bl.active)
WHERE bl.id IS NULL
"""
self._cr.execute(query % self._table)
res = self._cr.fetchall()
if not res:
return [(0, '=', 1)]
@@ -93,6 +100,10 @@ class MailBlackListMixin(models.AbstractModel):
[email_field] = self._primary_email
# TODO : Should remove the sudo as compute_sudo defined on methods.
# But if user doesn't have access to mail.blacklist, doen't work without sudo().
blacklist = self.env['mail.blacklist'].sudo().search([('email', 'in', self.mapped(email_field))]).mapped('email')
BL_sudo = self.env['mail.blacklist'].sudo()
emails_lower = [(email or '').lower() for email in self.mapped(email_field)]
blacklist = set(e.lower()
for e in BL_sudo.search([('email', 'in', emails_lower)]).mapped('email'))
for record in self:
record.is_blacklisted = record[email_field] in blacklist
email_lower = (record[email_field] or '').lower()
record.is_blacklisted = email_lower in blacklist
addons/mail/wizard/mail_compose_message.py
+7 -6
View File
@@ -281,14 +281,15 @@ class MailComposer(models.TransientModel):
blacklisted_rec_ids = []
if mass_mail_mode and hasattr(self.env[self.model], "_primary_email"):
blacklist = self.env['mail.blacklist'].sudo().search([]).mapped('email')
BL_sudo = self.env['mail.blacklist'].sudo()
blacklist = set(email.lower() for email in BL_sudo.search([]).mapped('email'))
if blacklist:
[email_field] = self.env[self.model]._primary_email
sql = """ SELECT id from %s WHERE LOWER(%s) = any (array[%s]) AND id in (%s)""" % \
(self.env[self.model]._table, email_field, ', '.join("'" + rec + "'" for rec in blacklist),
', '.join(str(res_id) for res_id in res_ids))
self._cr.execute(sql)
blacklisted_rec_ids = [rec[0] for rec in self._cr.fetchall()]
targets = self.env[self.model].browse(res_ids).read([email_field])
blacklisted_rec_ids = [r['id']
for r in targets
if r[email_field]
if r[email_field].lower() in blacklist]
for res_id in res_ids:
# static wizard (mail.message) values
mail_values = {
addons/mass_mailing/controllers/main.py
+30 -20
View File
@@ -11,6 +11,17 @@ from odoo.tools import consteq
class MassMailController(http.Controller):
def _valid_unsubscribe_token(self, mailing_id, res_id, email, token):
if not (mailing_id and res_id and email and token):
return False
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
return consteq(mailing._unsubscribe_token(res_id, email), token)
def _log_blacklist_action(self, blacklist_entry, mailing_id, description):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
model_display = mailing.mailing_model_id.display_name
blacklist_entry._message_log(description + " ({})".format(model_display))
@http.route(['/unsubscribe_from_list'], type='http', website=True, multilang=False, auth='public')
def unsubscribe_placeholder_link(self, **post):
"""Dummy route so placeholder is not prefixed by language, MUST have multilang=False"""
@@ -21,8 +32,7 @@ class MassMailController(http.Controller):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing.exists():
res_id = res_id and int(res_id)
right_token = mailing._unsubscribe_token(res_id, email)
if not consteq(str(token), right_token):
if not self._valid_unsubscribe_token(mailing_id, res_id, email, str(token)):
raise exceptions.AccessDenied()
if mailing.mailing_model_real == 'mail.mass_mailing.contact':
@@ -50,26 +60,27 @@ class MassMailController(http.Controller):
'mailing_id': mailing_id,
'res_id': res_id,
'show_blacklist_button': request.env['ir.config_parameter'].sudo().get_param('mass_mailing.show_blacklist_buttons'),
})
})
else:
blacklist_rec = request.env['mail.blacklist'].sudo()._add(email)
blacklist_rec._message_log(_("""The %s asked to not be contacted anymore
using an unsubscribe link.""" % request.env['ir.model']._get(mailing.mailing_model_real).display_name))
self._log_blacklist_action(
blacklist_rec, mailing_id,
_("""Requested blacklisting via unsubscribe link."""))
return request.render('mass_mailing.page_unsubscribed', {
'email': email,
'mailing_id': mailing_id,
'res_id': res_id,
'show_blacklist_button': request.env['ir.config_parameter'].sudo().get_param(
'mass_mailing.show_blacklist_buttons'),
})
})
return request.redirect('/web')
@http.route('/mail/mailing/unsubscribe', type='json', auth='none')
def unsubscribe(self, mailing_id, opt_in_ids, opt_out_ids, email, res_id, token):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing._unsubscribe_token(res_id, email) != token:
return 'unauthorized'
if mailing.exists():
if not self._valid_unsubscribe_token(mailing_id, res_id, email, token):
return 'unauthorized'
mailing.update_opt_out(email, opt_in_ids, False)
mailing.update_opt_out(email, opt_out_ids, True)
return True
@@ -96,8 +107,7 @@ class MassMailController(http.Controller):
@http.route('/mailing/blacklist/check', type='json', auth='none')
def blacklist_check(self, mailing_id, res_id, email, token):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing._unsubscribe_token(res_id, email) != token:
if not self._valid_unsubscribe_token(mailing_id, res_id, email, token):
return 'unauthorized'
if email:
record = request.env['mail.blacklist'].sudo().with_context(active_test=False).search([('email', '=ilike', email)])
@@ -108,34 +118,34 @@ class MassMailController(http.Controller):
@http.route('/mailing/blacklist/add', type='json', auth='none')
def blacklist_add(self, mailing_id, res_id, email, token):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing._unsubscribe_token(res_id, email) != token:
if not self._valid_unsubscribe_token(mailing_id, res_id, email, token):
return 'unauthorized'
if email:
blacklist_rec = request.env['mail.blacklist'].sudo()._add(email)
blacklist_rec._message_log(_("""The %s asked to not be contacted anymore
using the unsubscription page.""" % request.env['ir.model']._get(mailing.mailing_model_real).display_name))
self._log_blacklist_action(
blacklist_rec, mailing_id,
_("""Requested blacklisting via unsubscription page."""))
return True
return 'error'
@http.route('/mailing/blacklist/remove', type='json', auth='none')
def blacklist_remove(self, mailing_id, res_id, email, token):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing._unsubscribe_token(res_id, email) != token:
if not self._valid_unsubscribe_token(mailing_id, res_id, email, token):
return 'unauthorized'
if email:
blacklist_rec = request.env['mail.blacklist'].sudo()._remove(email)
blacklist_rec._message_log(_("""The %s asked to be removed from the blacklist
using the unsubscription page.""" % request.env['ir.model']._get(mailing.mailing_model_real).display_name))
self._log_blacklist_action(
blacklist_rec, mailing_id,
_("""Requested de-blacklisting via unsubscription page."""))
return True
return 'error'
@http.route('/mailing/feedback', type='json', auth='none')
def send_feedback(self, mailing_id, res_id, email, feedback, token):
mailing = request.env['mail.mass_mailing'].sudo().browse(mailing_id)
if mailing._unsubscribe_token(res_id, email) != token:
return 'unauthorized'
if mailing.exists() and email:
if not self._valid_unsubscribe_token(mailing_id, res_id, email, token):
return 'unauthorized'
model = request.env[mailing.mailing_model_real]
[email_field] = model._primary_email
records = model.sudo().search([(email_field, '=ilike', email)])
addons/mass_mailing/tests/test_mail_auto_blacklist.py
+1 -1
View File
@@ -13,7 +13,7 @@ class TestAutoBlacklist(common.TransactionCase):
mail_thread = self.env['mail.thread']
# create mailing contact record
self.mailing_contact_1 = mass_mailing_contacts.create({'name': 'test email 1', 'email': 'test1@email.com'})
self.mailing_contact_1 = mass_mailing_contacts.create({'name': 'test email 1', 'email': 'Test1@email.com'})
# create bounced history
mail_statistics.create({
addons/mass_mailing/tests/test_mass_mailing.py
+1 -1
View File
@@ -222,7 +222,7 @@ class TestOnResPartner(common.TransactionCase):
# Set Blacklist
self.blacklist_contact_entry = self.env['mail.blacklist'].create({
'email': 'test2@email.com',
'email': 'Test2@email.com',
})
# create mass mailing record