[FIX] mail: make the current user member of the created channel

Forward port fixes done in stable versions and not correclty forward ported
into master at merge time.

ORIGINAL COMMIT

Purpose
=======
Before, by default, if a user create a channel, he will not be member
of this channel.

After, the current user will always be member of the new channel.

LINKS

Task ID-2421795
COM PR odoo/odoo#63677
X-Original-commit odoo/odoo@eda542c82f
X-Original-Task ID-1963414

X-original-commit: 1d5d7871d72c8b747ab2e1b3597f6b8b8e371b26

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 = _('<div class="o_mail_notification">created <a href="#" class="o_channel_redirect" data-oe-id="%s">#%s</a></div>') % (new_channel.id, new_channel.name,)
         new_channel.message_post(body=notification, message_type="notification", subtype_xmlid="mail.mt_comment")

addons/mail/security/mail_security.xml

@@ -13,6 +13,20 @@
             <field name="perm_create" eval="False"/>
         </record>
 
+        <record id="ir_rule_mail_channel_partner_group_user" model="ir.rule">
+            <field name="name">mail.channel.partner: write its own entries</field>
+            <field name="model_id" ref="model_mail_channel_partner"/>
+            <field name="groups" eval="[(4, ref('base.group_user')), (4, ref('base.group_portal'))]"/>
+            <field name="domain_force">['|', '|',
+('channel_id.public', '=', 'public'),
+'&amp;', ('channel_id.public', '=', 'private'), ('channel_id.channel_partner_ids', 'in', [user.partner_id.id]),
+'&amp;', ('channel_id.public', '=', 'groups'), ('channel_id.group_public_id', 'in', [g.id for g in user.groups_id])]</field>
+            <field name="perm_read" eval="False"/>
+            <field name="perm_write" eval="True"/>
+            <field name="perm_create" eval="False"/>
+            <field name="perm_unlink" eval="True"/>
+        </record>
+
         <record id="ir_rule_mail_notifications_group_user" model="ir.rule">
             <field name="name">mail.notifications: group_user: write its own entries</field>
             <field name="model_id" ref="model_mail_notification"/>

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

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'])

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()