diff --git a/addons/mail/models/mail_channel.py b/addons/mail/models/mail_channel.py index fbe2eada7c5..cd1a3b2cda1 100644 --- a/addons/mail/models/mail_channel.py +++ b/addons/mail/models/mail_channel.py @@ -10,6 +10,7 @@ from odoo import _, api, fields, models, modules, tools, Command from odoo.exceptions import UserError, ValidationError from odoo.osv import expression from odoo.tools import ormcache, formataddr +from odoo.exceptions import AccessError from odoo.addons.base.models.ir_model import MODULE_UNINSTALL_FLAG MODERATION_FIELDS = ['moderation', 'moderator_ids', 'moderation_ids', 'moderation_notify', 'moderation_notify_msg', 'moderation_guidelines', 'moderation_guidelines_msg'] @@ -32,6 +33,28 @@ class ChannelPartner(models.Model): is_minimized = fields.Boolean("Conversation is minimized") is_pinned = fields.Boolean("Is pinned on the interface", default=True) + @api.model_create_multi + def create(self, vals_list): + """Similar access rule as the access rule of the mail channel. + + It can not be implemented in XML, because when the record will be created, the + partner will be added in the channel and the security rule will always authorize + the creation. + """ + if not self.env.is_admin(): + for vals in vals_list: + if 'channel_id' in vals and not self.env.is_admin(): + channel_id = self.env['mail.channel'].browse(vals['channel_id']) + if not channel_id._can_invite(vals.get('partner_id')): + raise AccessError(_('This user can not be added in this channel')) + return super(ChannelPartner, self).create(vals_list) + + def write(self, vals): + if not self.env.is_admin(): + if {'channel_id', 'partner_id', 'partner_email'} & set(vals): + raise AccessError(_('You can not write on this field')) + return super(ChannelPartner, self).write(vals) + class Moderation(models.Model): _name = 'mail.moderation' @@ -67,9 +90,6 @@ class Channel(models.Model): res['alias_contact'] = 'everyone' if res.get('public', 'private') == 'public' else 'followers' return res - def _default_channel_last_seen_partner_ids(self): - return [Command.create({"partner_id": self.env.user.partner_id.id})] - def _get_default_image(self): image_path = modules.get_module_resource('mail', 'static/src/img', 'groupdefault.png') return base64.b64encode(open(image_path, 'rb').read()) @@ -86,7 +106,7 @@ class Channel(models.Model): email_send = fields.Boolean('Send messages by email', default=False) # multi users channel # depends=['...'] is for `test_mail/tests/common.py`, class Moderation, `setUpClass` - channel_last_seen_partner_ids = fields.One2many('mail.channel.partner', 'channel_id', string='Last Seen', depends=['channel_partner_ids'], default=_default_channel_last_seen_partner_ids) + channel_last_seen_partner_ids = fields.One2many('mail.channel.partner', 'channel_id', string='Last Seen', depends=['channel_partner_ids']) channel_partner_ids = fields.Many2many('res.partner', 'mail_channel_partner', 'channel_id', 'partner_id', string='Listeners', depends=['channel_last_seen_partner_ids']) channel_message_ids = fields.Many2many('mail.message', 'mail_message_mail_channel_rel') is_member = fields.Boolean('Is a member', compute='_compute_is_member') @@ -205,16 +225,39 @@ class Channel(models.Model): @api.model_create_multi def create(self, vals_list): defaults = self.default_get(['image_128']) + current_partner = self.env.user.partner_id.id + + visibilities = [] for vals in vals_list: # ensure image at quick create if not vals.get('image_128'): vals['image_128'] = defaults['image_128'] + # always add current user to new channel, go through + # channel_last_seen_partner_ids otherwise in v14 the channel is not + # visible for the user (because is_pinned is false and taken in account) + if 'channel_partner_ids' in vals: + vals['channel_partner_ids'] = [ + entry + for entry in vals['channel_partner_ids'] + if entry[0] != 4 or entry[1] != current_partner + ] + membership = vals.setdefault('channel_last_seen_partner_ids', []) + if all(entry[0] != 0 or entry[2].get('partner_id') != current_partner for entry in membership): + membership.append((0, False, {'partner_id': current_partner})) + + visibility_default = self._fields['public'].default(self) + visibilities.append(vals.pop('public', visibility_default)) + vals['public'] = 'public' # Create channel and alias channels = super(Channel, self.with_context( mail_create_nolog=True, mail_create_nosubscribe=True) ).create(vals_list) + for visibility, channel in zip(visibilities, channels): + if visibility != 'public': + channel.sudo().public = visibility + channels._subscribe_users() # make channel listen itself: posting on a channel notifies the channel @@ -398,7 +441,7 @@ class Channel(models.Model): if moderation_status == 'rejected': return self.env['mail.message'] - self.filtered(lambda channel: channel.is_chat).mapped('channel_last_seen_partner_ids').write({'is_pinned': True}) + self.filtered(lambda channel: channel.is_chat).mapped('channel_last_seen_partner_ids').sudo().write({'is_pinned': True}) message = super(Channel, self.with_context(mail_create_nosubscribe=True)).message_post(message_type=message_type, moderation_status=moderation_status, **kwargs) @@ -474,7 +517,7 @@ class Channel(models.Model): return True def _update_moderation_email(self, emails, status): - """ This method adds emails into either white or black of the channel list of emails + """ This method adds emails into either white or black of the channel list of emails according to status. If an email in emails is already moderated, the method updates the email status. :param emails: list of email addresses to put in white or black list of channel. :param status: value is 'allow' or 'ban'. Emails are put in white list if 'allow', in black list if 'ban'. @@ -883,6 +926,26 @@ class Channel(models.Model): ', '.join('%s (channel %s)' % (partner.name, channel.name) for channel, partner in failed) ) + def _can_invite(self, partner_id): + """Return True if the current user can invite the partner to the channel. + + * public: ok; + * private: must be member; + * group: both current user and target must have group; + + :return boolean: whether inviting is ok""" + partner = self.env['res.partner'].browse(partner_id) + + for channel in self.sudo(): + if channel.public == 'private' and not channel.is_member: + return False + if channel.public == 'groups': + if not partner.user_ids or channel.group_public_id not in partner.user_ids.groups_id: + return False + if channel.group_public_id not in self.env.user.groups_id: + return False + return True + @api.model def channel_set_custom_name(self, channel_id, name=False): domain = [('partner_id', '=', self.env.user.partner_id.id), ('channel_id.id', '=', channel_id)] @@ -977,7 +1040,6 @@ class Channel(models.Model): 'name': name, 'public': privacy, 'email_send': False, - 'channel_partner_ids': [Command.link(self.env.user.partner_id.id)] }) notification = _('
created #%s
') % (new_channel.id, new_channel.name,) new_channel.message_post(body=notification, message_type="notification", subtype_xmlid="mail.mt_comment") diff --git a/addons/mail/security/mail_security.xml b/addons/mail/security/mail_security.xml index 5f48516a4a1..d38c1cccae3 100644 --- a/addons/mail/security/mail_security.xml +++ b/addons/mail/security/mail_security.xml @@ -13,6 +13,20 @@ + + mail.channel.partner: write its own entries + + + ['|', '|', +('channel_id.public', '=', 'public'), +'&', ('channel_id.public', '=', 'private'), ('channel_id.channel_partner_ids', 'in', [user.partner_id.id]), +'&', ('channel_id.public', '=', 'groups'), ('channel_id.group_public_id', 'in', [g.id for g in user.groups_id])] + + + + + + mail.notifications: group_user: write its own entries diff --git a/addons/mail/tests/__init__.py b/addons/mail/tests/__init__.py index 5d739b895b2..f275a83b1b0 100644 --- a/addons/mail/tests/__init__.py +++ b/addons/mail/tests/__init__.py @@ -2,6 +2,7 @@ # Part of Odoo. See LICENSE file for full copyright and licensing details. from . import test_mail_channel +from . import test_mail_channel_partner from . import test_mail_full_composer from . import test_res_partner from . import test_update_notification diff --git a/addons/mail/tests/test_mail_channel.py b/addons/mail/tests/test_mail_channel.py index d9e13e9a4da..f4a9e02f369 100644 --- a/addons/mail/tests/test_mail_channel.py +++ b/addons/mail/tests/test_mail_channel.py @@ -4,7 +4,7 @@ from odoo.addons.mail.tests.common import mail_new_test_user from odoo.addons.mail.tests.common import MailCommon from odoo.exceptions import AccessError, ValidationError, UserError -from odoo.tests import tagged +from odoo.tests import tagged, Form from odoo.tests.common import users from odoo.tools import mute_logger, formataddr @@ -133,13 +133,11 @@ class TestChannelAccessRights(MailCommon): with self.assertRaises(AccessError): group_private.read(['name']) - self.env['mail.channel.partner'].create({ - 'partner_id': self.env.user.partner_id.id, - 'channel_id': group_private.id, - }) - group_private = self.env['mail.channel'].browse(self.group_private.id) - self.assertTrue(group_private.is_member) - group_private.read(['name', 'channel_last_seen_partner_ids']) + with self.assertRaises(AccessError): + self.env['mail.channel.partner'].create({ + 'partner_id': self.env.user.partner_id.id, + 'channel_id': group_private.id, + }) class TestChannelInternals(MailCommon): @@ -170,6 +168,15 @@ class TestChannelInternals(MailCommon): ) cls.partner_employee_nomail = cls.user_employee_nomail.partner_id + @users('employee') + def test_channel_form(self): + """A user that create a private channel should be able to read it.""" + channel_form = Form(self.env['mail.channel'].with_user(self.user_employee)) + channel_form.name = 'Test private channel' + channel_form.public = 'private' + channel = channel_form.save() + self.assertEqual(channel.name, 'Test private channel', 'Must be able to read the created channel') + def test_channel_members(self): self.assertEqual(self.test_channel.message_channel_ids, self.test_channel) self.assertEqual(self.test_channel.message_partner_ids, self.env['res.partner']) diff --git a/addons/mail/tests/test_mail_channel_partner.py b/addons/mail/tests/test_mail_channel_partner.py new file mode 100644 index 00000000000..ae16dd4195d --- /dev/null +++ b/addons/mail/tests/test_mail_channel_partner.py @@ -0,0 +1,223 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from odoo.addons.mail.tests.common import mail_new_test_user +from odoo.addons.mail.tests.common import MailCommon +from odoo.exceptions import AccessError, UserError + + +class TestMailChannelMembers(MailCommon): + + @classmethod + def setUpClass(cls): + super(TestMailChannelMembers, cls).setUpClass() + + cls.secret_group = cls.env['res.groups'].create({ + 'name': 'Secret User Group', + }) + cls.env['ir.model.data'].create({ + 'name': 'secret_group', + 'module': 'mail', + 'model': cls.secret_group._name, + 'res_id': cls.secret_group.id, + }) + + cls.user_1 = mail_new_test_user( + cls.env, login='user_1', + name='User 1', + groups='base.group_user,mail.secret_group') + cls.user_2 = mail_new_test_user( + cls.env, login='user_2', + name='User 2', + groups='base.group_user,mail.secret_group') + cls.user_portal = mail_new_test_user( + cls.env, login='user_portal', + name='User Portal', + groups='base.group_portal') + cls.user_public = mail_new_test_user( + cls.env, login='user_ublic', + name='User Public', + groups='base.group_public') + + cls.private_channel = cls.env['mail.channel'].create({ + 'name': 'Secret channel', + 'public': 'private', + 'channel_type': 'channel', + }) + cls.group_channel = cls.env['mail.channel'].create({ + 'name': 'Group channel', + 'public': 'groups', + 'channel_type': 'channel', + 'group_public_id': cls.secret_group.id, + }) + cls.public_channel = cls.env['mail.channel'].create({ + 'name': 'Public channel of user 1', + 'public': 'public', + 'channel_type': 'channel', + }) + (cls.private_channel | cls.group_channel | cls.public_channel).channel_last_seen_partner_ids.unlink() + + # ------------------------------------------------------------ + # PRIVATE CHANNELS + # ------------------------------------------------------------ + + def test_channel_private_01(self): + """Test access on private channel.""" + res = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertFalse(res) + + # User 1 can join private channel with SUDO + self.private_channel.with_user(self.user_1).sudo().action_follow() + res = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(res.partner_id, self.user_1.partner_id) + + # User 2 can not join private channel + with self.assertRaises(AccessError): + self.private_channel.with_user(self.user_2).action_follow() + + # User 2 can not create a `mail.channel.partner` to join the private channel + with self.assertRaises(AccessError): + self.env['mail.channel.partner'].with_user(self.user_2).create({ + 'partner_id': self.user_2.partner_id.id, + 'channel_id': self.private_channel.id, + }) + + # User 2 can not write on `mail.channel.partner` to join the private channel + channel_partner = self.env['mail.channel.partner'].with_user(self.user_2).search([('partner_id', '=', self.user_2.partner_id.id)])[0] + with self.assertRaises(AccessError): + channel_partner.channel_id = self.private_channel.id + with self.assertRaises(AccessError): + channel_partner.write({'channel_id': self.private_channel.id}) + + # But with SUDO, User 2 can + channel_partner.sudo().channel_id = self.private_channel.id + + # User 2 can not write on the `partner_id` of `mail.channel.partner` + # of an other partner to join a private channel + channel_partner_1 = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id), ('partner_id', '=', self.user_1.partner_id.id)]) + with self.assertRaises(AccessError): + channel_partner_1.with_user(self.user_2).partner_id = self.user_2.partner_id + self.assertEqual(channel_partner_1.partner_id, self.user_1.partner_id) + + # but with SUDO he can... + channel_partner_1.with_user(self.user_2).sudo().partner_id = self.user_2.partner_id + self.assertEqual(channel_partner_1.partner_id, self.user_2.partner_id) + + def test_channel_private_members(self): + """Test invitation in private channel part 1 (invite using crud methods).""" + self.private_channel.with_user(self.user_1).sudo().action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(len(channel_partners), 1) + + # User 2 is not in the private channel, he can not invite user 3 + with self.assertRaises(AccessError): + self.env['mail.channel.partner'].with_user(self.user_2).create({ + 'partner_id': self.user_portal.partner_id.id, + 'channel_id': self.private_channel.id, + }) + + # User 1 is in the private channel, he can invite other users + self.env['mail.channel.partner'].with_user(self.user_1).create({ + 'partner_id': self.user_portal.partner_id.id, + 'channel_id': self.private_channel.id, + }) + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_portal.partner_id) + + # But User 3 can not write on the `mail.channel.partner` of other user + channel_partner_1 = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id), ('partner_id', '=', self.user_1.partner_id.id)]) + channel_partner_3 = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id), ('partner_id', '=', self.user_portal.partner_id.id)]) + channel_partner_3.with_user(self.user_portal).custom_channel_name = 'Test' + with self.assertRaises(AccessError): + channel_partner_1.with_user(self.user_2).custom_channel_name = 'Blabla' + self.assertNotEqual(channel_partner_1.custom_channel_name, 'Blabla') + + def test_channel_private_invite(self): + """Test invitation in private channel part 2 (use `invite` action).""" + self.private_channel.with_user(self.user_1).sudo().action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + # User 2 is not in the channel, he can not invite user_portal + with self.assertRaises(AccessError): + self.private_channel.with_user(self.user_2).channel_invite([self.user_portal.partner_id.id]) + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + # User 1 is in the channel, he can invite user_portal + self.private_channel.with_user(self.user_1).channel_invite([self.user_portal.partner_id.id]) + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_portal.partner_id) + + def test_channel_private_leave(self): + """Test kick/leave channel.""" + self.private_channel.with_user(self.user_1).sudo().action_follow() + self.private_channel.with_user(self.user_portal).sudo().action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel.id)]) + self.assertEqual(len(channel_partners), 2) + + # User 2 is not in the channel, he can not kick user 1 + with self.assertRaises(AccessError): + channel_partners.with_user(self.user_2).unlink() + + # User 3 is in the channel, he can kick user 1 + channel_partners.with_user(self.user_portal).unlink() + + # ------------------------------------------------------------ + # GROUP BASED CHANNELS + # ------------------------------------------------------------ + + def test_channel_group(self): + """Test basics on group channel.""" + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + self.assertFalse(channel_partners) + + # user 1 is in the group, he can join the channel + self.group_channel.with_user(self.user_1).action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + # user 3 is not in the group, he can not join + with self.assertRaises(AccessError): + self.group_channel.with_user(self.user_portal).action_follow() + + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + with self.assertRaises(AccessError): + channel_partners.with_user(self.user_portal).partner_id = self.user_portal.partner_id + + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + # user 1 can not invite user 3 because he's not in the group + with self.assertRaises(UserError): + self.group_channel.with_user(self.user_1).channel_invite([self.user_portal.partner_id.id]) + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + # but user 2 is in the group and can be invited by user 1 + self.group_channel.with_user(self.user_1).channel_invite([self.user_2.partner_id.id]) + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_2.partner_id) + + # ------------------------------------------------------------ + # PUBLIC CHANNELS + # ------------------------------------------------------------ + + def test_channel_public(self): + """ Test access on public channels """ + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel.id)]) + self.assertFalse(channel_partners) + + self.public_channel.with_user(self.user_1).action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id) + + self.public_channel.with_user(self.user_2).action_follow() + channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel.id)]) + self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_2.partner_id) + + # portal/public users still cannot join a public channel, should go through dedicated controllers + with self.assertRaises(AccessError): + self.public_channel.with_user(self.user_portal).action_follow() + with self.assertRaises(AccessError): + self.public_channel.with_user(self.user_public).action_follow()