[MERGE] mail: improve alias security and alias check on channel model

This commit links the alias security and the channel configuration.
Indeed private or group-based channels should always have a restricted
alias. Moreover this alias should check channel members instead of
channel followers, as a channel has members and should not have followers.

This implies a small refactoring of the alias check in the mailgateway.
Code is extracted and moved on alias.mixin class, allowing to tune the
behavior through inheritance.
This commit is contained in:
Thibault Delavallée
2017-03-23 13:12:05 +01:00
7 changed files with 101 additions and 67 deletions
-1
View File
@@ -4,6 +4,5 @@
import hr
import hr_config_settings
import mail_alias
import mail_thread
import res_partner
import res_users
+20 -1
View File
@@ -1,10 +1,29 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import fields, models
from odoo import fields, models, tools
class Alias(models.Model):
_inherit = 'mail.alias'
alias_contact = fields.Selection(selection_add=[('employees', 'Authenticated Employees')])
class MailAlias(models.AbstractModel):
_inherit = 'mail.alias.mixin'
def _alias_check_contact(self, message, message_dict, alias):
if alias.alias_contact == 'employees' and self.ids:
email_from = tools.decode_message_header(message, 'From')
email_address = tools.email_split(email_from)[0]
employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
if not employee:
employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
if not employee:
return {
'error_message': 'restricted to employees',
'error_template': self.env.ref('hr.mail_template_data_unknown_employee_email_address').body_html,
}
return True
return super(MailAlias, self)._alias_check_contact(message, message_dict, alias)
+18 -38
View File
@@ -1,44 +1,24 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
from odoo import api, models, _
from odoo.tools import decode_message_header, email_split
_logger = logging.getLogger(__name__)
from odoo import models, tools
from odoo.tools import email_split
class MailThread(models.AbstractModel):
_inherit = "mail.thread"
class MailAlias(models.AbstractModel):
_inherit = 'mail.alias.mixin'
@api.model
def message_route_verify(self, message, message_dict, route,
update_author=True, assert_model=True,
create_fallback=True, allow_private=False,
drop_alias=False):
res = super(MailThread, self).message_route_verify(
message, message_dict, route,
update_author=update_author,
assert_model=assert_model,
create_fallback=create_fallback,
allow_private=allow_private,
drop_alias=drop_alias)
if res:
alias = route[4]
email_from = decode_message_header(message, 'From')
message_id = message.get('Message-Id')
# Alias: check alias_contact settings for employees
if alias and alias.alias_contact == 'employees':
email_address = email_split(email_from)[0]
employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
if not employee:
employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
if not employee:
mail_template = self.env.ref('hr.mail_template_data_unknown_employee_email_address')
self._routing_warn(_('alias %s does not accept unknown employees') % alias.alias_name, _('skipping'), message_id, route, False)
self._routing_create_bounce_email(email_from, mail_template.body_html, message)
return False
return res
def _alias_check_contact(self, message, message_dict, alias):
if alias.alias_contact == 'employees' and self.ids:
email_from = tools.decode_message_header(message, 'From')
email_address = email_split(email_from)[0]
employee = self.env['hr.employee'].search([('work_email', 'ilike', email_address)], limit=1)
if not employee:
employee = self.env['hr.employee'].search([('user_id.email', 'ilike', email_address)], limit=1)
if not employee:
return {
'error_message': 'restricted to employees',
'error_template': self.env.ref('hr.mail_template_data_unknown_employee_email_address').body_html,
}
return True
return super(MailAlias, self)._alias_check_contact(message, message_dict, alias)
+18
View File
@@ -262,3 +262,21 @@ class AliasMixin(models.AbstractModel):
record.with_context({'mail_notrack': True}).alias_id = alias
_logger.info('Mail alias created for %s %s (id %s)',
record._name, record.display_name, record.id)
def _alias_check_contact(self, message, message_dict, alias):
author = self.env['res.partner'].browse(message_dict.get('author_id', False))
if alias.alias_contact == 'followers' and self.ids:
if not hasattr(self, "message_partner_ids") or not hasattr(self, "message_channel_ids"):
return {
'error_mesage': _('incorrectly configured alias'),
}
accepted_partner_ids = self.message_partner_ids | self.message_channel_ids.mapped('channel_partner_ids')
if not author or author not in accepted_partner_ids:
return {
'error_mesage': _('restricted to followers'),
}
elif alias.alias_contact == 'partners' and not author:
return {
'error_message': _('restricted to known authors')
}
return True
+29 -5
View File
@@ -20,6 +20,7 @@ class ChannelPartner(models.Model):
_rec_name = 'partner_id'
partner_id = fields.Many2one('res.partner', string='Recipient', ondelete='cascade')
partner_email = fields.Char('Email', related='partner_id.email')
channel_id = fields.Many2one('mail.channel', string='Channel', ondelete='cascade')
seen_message_id = fields.Many2one('mail.message', string='Last Seen')
fold_state = fields.Selection([('open', 'Open'), ('folded', 'Folded'), ('closed', 'Closed')], string='Conversation Fold State', default='open')
@@ -34,8 +35,7 @@ class Channel(models.Model):
_name = 'mail.channel'
_mail_flat_thread = False
_mail_post_access = 'read'
_inherit = ['mail.thread']
_inherits = {'mail.alias': 'alias_id'}
_inherit = ['mail.thread', 'mail.alias.mixin']
MAX_BOUNCE_LIMIT = 10
@@ -43,6 +43,13 @@ class Channel(models.Model):
image_path = modules.get_module_resource('mail', 'static/src/img', 'groupdefault.png')
return tools.image_resize_image_big(open(image_path, 'rb').read().encode('base64'))
@api.model
def default_get(self, fields):
res = super(Channel, self).default_get(fields)
if not res.get('alias_contact') and (not fields or 'alias_contact' in fields):
res['alias_contact'] = 'everyone' if res.get('public', 'private') == 'public' else 'followers'
return res
name = fields.Char('Name', required=True, translate=True)
channel_type = fields.Selection([
('chat', 'Chat Discussion'),
@@ -81,9 +88,6 @@ class Channel(models.Model):
help="Small-sized photo of the group. It is automatically "
"resized as a 64x64px image, with aspect ratio preserved. "
"Use this field anywhere a small image is required.")
alias_id = fields.Many2one(
'mail.alias', 'Alias', ondelete="restrict", required=True,
help="The email address associated with this group. New emails received will automatically create new topics.")
is_subscribed = fields.Boolean(
'Is Subscribed', compute='_compute_is_subscribed')
@@ -102,6 +106,13 @@ class Channel(models.Model):
for record in self:
record.is_member = record in membership_ids
@api.onchange('public')
def _onchange_public(self):
if self.public == 'public':
self.alias_contact = 'everyone'
else:
self.alias_contact = 'followers'
@api.model
def create(self, vals):
tools.image_resize_images(vals)
@@ -144,6 +155,9 @@ class Channel(models.Model):
self._subscribe_users()
return result
def get_alias_model_name(self, vals):
return vals.get('alias_model', 'mail.channel')
def _subscribe_users(self):
for mail_channel in self:
mail_channel.write({'channel_partner_ids': [(4, pid) for pid in mail_channel.mapped('group_ids').mapped('users').mapped('partner_id').ids]})
@@ -232,6 +246,16 @@ class Channel(models.Model):
message = super(Channel, self.with_context(mail_create_nosubscribe=True)).message_post(body=body, subject=subject, message_type=message_type, subtype=subtype, parent_id=parent_id, attachments=attachments, content_subtype=content_subtype, **kwargs)
return message
def _alias_check_contact(self, message, message_dict, alias):
if alias.alias_contact == 'followers' and self.ids:
author = self.env['res.partner'].browse(message_dict.get('author_id', False))
if not author or author not in self.channel_partner_ids:
return {
'error_mesage': _('restricted to channel members'),
}
return True
return super(Channel, self)._alias_check_contact(message, message_dict, alias)
@api.model_cr
def init(self):
self._cr.execute('SELECT indexname FROM pg_indexes WHERE indexname = %s', ('mail_channel_partner_seen_message_id_idx',))
+11 -15
View File
@@ -944,27 +944,23 @@ class MailThread(models.AbstractModel):
if not author_id and update_author:
author_ids = self.env['mail.thread']._find_partner_from_emails([email_from], res_model=model, res_id=thread_id)
if author_ids:
author_id = author_ids[0]
message_dict['author_id'] = author_id
message_dict['author_id'] = author_ids[0]
# Alias: check alias_contact settings
if alias and alias.alias_contact == 'followers' and (thread_id or alias.alias_parent_thread_id):
if alias:
if thread_id:
obj = record_set[0]
else:
elif alias.alias_parent_thread_id:
obj = self.env[alias.alias_parent_model_id.model].browse(alias.alias_parent_thread_id)
accepted_partner_ids = list(
set(partner.id for partner in obj.message_partner_ids) |
set(partner.id for channel in obj.message_channel_ids for partner in channel.channel_partner_ids)
)
if not author_id or author_id not in accepted_partner_ids:
self._routing_warn(_('alias %s restricted to internal followers') % alias.alias_name, _('skipping'), message_id, route, False)
self._routing_create_bounce_email(email_from, _generic_bounce_body_html, message)
elif model and hasattr(record_set, '_alias_check_contact'):
obj = self.env[model]
else:
obj = self.env['mail.alias.mixin']
check_result = obj._alias_check_contact(message, message_dict, alias)
if check_result is not True:
self._routing_warn(_('alias %s: %s') % (alias.alias_name, check_result.get('error_message', _('unknown error'))), _('skipping'), message_id, route, False)
self._routing_create_bounce_email(email_from, check_result.get('error_template', _generic_bounce_body_html), message)
return False
elif alias and alias.alias_contact == 'partners' and not author_id:
self._routing_warn(_('alias %s does not accept unknown author') % alias.alias_name, _('skipping'), message_id, route, False)
self._routing_create_bounce_email(email_from, _generic_bounce_body_html, message)
return False
if not model and not thread_id and not alias and not allow_private:
return False
+5 -7
View File
@@ -106,8 +106,7 @@
<field name="alias_name" class="oe_inline"/>@<field name="alias_domain" class="oe_inline" readonly="1"/>
</div>
</div>
<label for="alias_contact" string=" " class="fa fa-lock" style="min-width: 20px;"/>
<field name="alias_contact" class="oe_inline" nolabel="1"/>
<field name="alias_contact" class="oe_inline" invisible="1"/>
</group>
<notebook>
<page string="Privacy">
@@ -120,11 +119,10 @@
</group>
</page>
<page string="Members">
<field name="channel_partner_ids" mode="tree">
<tree string="members">
<field name="name"/>
<field name="email"/>
<field name="im_status"/>
<field name="channel_last_seen_partner_ids" mode="tree" context="{'active_test': False}">
<tree string="Members" editable="bottom">
<field name="partner_id"/>
<field name="partner_email"/>
</tree>
</field>
</page>