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

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.

Task 1963414

5 files changed, 296 insertions, 42 deletions

diff --git a/addons/mail/models/mail_channel.py b/addons/mail/models/mail_channel.py
index 74e1eac1332..e0bfd771de4 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
 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,26 @@ class ChannelPartner(models.Model):
     is_minimized = fields.Boolean("Conversation is minimized")
     is_pinned = fields.Boolean("Is pinned on the interface", default=True)
 
+    @api.model
+    def create(self, vals):
+        """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 '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)
+
+    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 +88,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 [(0, 0, {"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 +104,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')
@@ -211,10 +229,29 @@ class Channel(models.Model):
             defaults = self.default_get(['image_128'])
             vals['image_128'] = defaults['image_128']
 
+        current_partner = self.env.user.partner_id.id
+        # 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)
+        visibility = vals.pop('public', visibility_default)
+        vals['public'] = 'public'
         # Create channel and alias
         channel = super(Channel, self.with_context(
             mail_create_nolog=True, mail_create_nosubscribe=True)
         ).create(vals)
+        if visibility != 'public':
+            channel.sudo().public = visibility
 
         if vals.get('group_ids'):
             channel._subscribe_users()
@@ -374,7 +411,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)
 
@@ -450,7 +487,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'.
@@ -843,6 +880,22 @@ class Channel(models.Model):
         # broadcast the channel header to the added partner
         self._broadcast(partner_ids)
 
+    def _can_invite(self, partner_id):
+        """Return True if the current user can invite the partner to the channel."""
+        self.ensure_one()
+        sudo_self = self.sudo()
+        if sudo_self.public == 'public':
+            return True
+        if sudo_self.public == 'private':
+            return self.is_member
+
+        # get the user related to the invited partner
+        partner = self.env['res.partner'].browse(partner_id).exists()
+        invited_user_id = partner.user_ids[:1]
+        if invited_user_id:
+            return (self.env.user | invited_user_id) <= sudo_self.group_public_id.users
+        return False
+
     @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)]
@@ -937,7 +990,6 @@ class Channel(models.Model):
             'name': name,
             'public': privacy,
             'email_send': False,
-            'channel_partner_ids': [(4, 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")
diff --git a/addons/mail/security/mail_security.xml b/addons/mail/security/mail_security.xml
index 565af06b20a..6aef7841d6d 100644
--- a/addons/mail/security/mail_security.xml
+++ b/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"/>
diff --git a/addons/test_mail/tests/__init__.py b/addons/test_mail/tests/__init__.py
index e328b555861..3c362ee53b9 100644
--- a/addons/test_mail/tests/__init__.py
+++ b/addons/test_mail/tests/__init__.py
@@ -8,6 +8,7 @@ from . import test_mail_followers
 from . import test_mail_message
 from . import test_mail_mail
 from . import test_mail_channel
+from . import test_mail_channel_partner
 from . import test_mail_gateway
 from . import test_mail_template
 from . import test_mail_thread_internals
diff --git a/addons/test_mail/tests/test_mail_channel.py b/addons/test_mail/tests/test_mail_channel.py
index 168bf51c760..d81fb161bf4 100644
--- a/addons/test_mail/tests/test_mail_channel.py
+++ b/addons/test_mail/tests/test_mail_channel.py
@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-from odoo.tests import tagged
+from odoo.tests import tagged, Form
 from odoo.addons.mail.tests.common import mail_new_test_user
 from odoo.addons.test_mail.tests.common import TestMailCommon
 from odoo.exceptions import AccessError, ValidationError, UserError
@@ -191,40 +191,13 @@ class TestChannelFeatures(TestMailCommon):
             self.test_channel.message_post(body="Test", message_type='comment', subtype_xmlid='mail.mt_comment')
         self.assertSentEmail(self.test_channel.env.user.partner_id, [self.test_partner])
 
-    @mute_logger('odoo.models.unlink')
-    def test_channel_auto_unsubscribe_archived_or_deleted_users(self):
-        """Archiving / deleting a user should automatically unsubscribe related partner from private channels"""
-        test_channel_private = self.env['mail.channel'].with_context(self._test_context).create({
-            'name': 'Winden caves',
-            'description': 'Channel to travel through time',
-            'public': 'private',
-        })
-        test_channel_group = self.env['mail.channel'].with_context(self._test_context).create({
-            'name': 'Sic Mundus',
-            'public': 'groups',
-            'group_public_id': self.env.ref('base.group_user').id})
-
-        test_user = self.env['res.users'].create({
-            "login": "adam",
-            "name": "Jonas",
-        })
-        test_partner = test_user.partner_id
-
-        self._join_channel(self.test_channel, self.user_employee.partner_id | test_partner)
-        self._join_channel(test_channel_private, self.user_employee.partner_id | test_partner)
-        self._join_channel(test_channel_group, self.user_employee.partner_id | test_partner)
-
-        # Unsubscribe archived user from the private channels, but not from public channels
-        self.user_employee.active = False
-        self.assertEqual(test_channel_private.channel_partner_ids, test_partner)
-        self.assertEqual(test_channel_group.channel_partner_ids, test_partner)
-        self.assertEqual(self.test_channel.channel_partner_ids, self.user_employee.partner_id | test_partner)
-
-        # Unsubscribe deleted user from the private channels, but not from public channels
-        test_user.unlink()
-        self.assertEqual(test_channel_private.channel_partner_ids, self.env['res.partner'])
-        self.assertEqual(test_channel_group.channel_partner_ids, self.env['res.partner'])
-        self.assertEqual(self.test_channel.channel_partner_ids, self.user_employee.partner_id | test_partner)
+    def test_channel_creation(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_get(self):
         current_user = self.env['res.users'].create({
diff --git a/addons/test_mail/tests/test_mail_channel_partner.py b/addons/test_mail/tests/test_mail_channel_partner.py
new file mode 100644
index 00000000000..d2629b64551
--- /dev/null
+++ b/addons/test_mail/tests/test_mail_channel_partner.py
@@ -0,0 +1,214 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+
+from odoo.tests.common import SavepointCase
+from odoo.exceptions import AccessError
+
+
+class TestMailSecurity(SavepointCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.secret_group = cls.env['res.groups'].create({
+            'name': 'secret group',
+        })
+        cls.user_1 = cls.env['res.users'].create({
+            'name': 'User 1',
+            'login': 'user_1',
+            'email': '---',
+            'groups_id': [(6, 0, [cls.secret_group.id, cls.env.ref('base.group_user').id])],
+        })
+        cls.user_2 = cls.env['res.users'].create({
+            'name': 'User 2',
+            'login': 'user_2',
+            'email': '---',
+            'groups_id': [(6, 0, [cls.secret_group.id, cls.env.ref('base.group_user').id])],
+        })
+        cls.user_3 = cls.env['res.users'].create({
+            'name': 'User 3',
+            'login': 'user_3',
+            'email': '---',
+        })
+
+        cls.private_channel_1 = cls.env['mail.channel'].create({
+            'name': 'Secret channel',
+            'public': 'private',
+            'channel_type': 'channel',
+        })
+        cls.group_channel_1 = cls.env['mail.channel'].create({
+            'name': 'Group channel',
+            'public': 'groups',
+            'channel_type': 'channel',
+            'group_public_id': cls.secret_group.id,
+        })
+        cls.public_channel_1 = cls.env['mail.channel'].create({
+            'name': 'Public channel of user 1',
+            'public': 'public',
+            'channel_type': 'channel',
+        })
+        cls.private_channel_1.channel_last_seen_partner_ids.unlink()
+        cls.group_channel_1.channel_last_seen_partner_ids.unlink()
+        cls.public_channel_1.channel_last_seen_partner_ids.unlink()
+
+    ###########################
+    # PRIVATE CHANNEL & BASIC #
+    ###########################
+
+    def test_channel_acls_01(self):
+        """Test access on private channel."""
+        res = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.id)])
+        self.assertFalse(res)
+
+        # User 1 can join private channel with SUDO
+        self.private_channel_1.with_user(self.user_1).sudo().action_follow()
+        res = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.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_1.with_user(self.user_2).action_follow()
+
+        # But user 2 can join public channel
+        self.public_channel_1.with_user(self.user_2).action_follow()
+        res = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel_1.id)])
+        self.assertEqual(res.partner_id, self.user_2.partner_id)
+
+        # 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_1.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_1.id
+        with self.assertRaises(AccessError):
+            channel_partner.write({'channel_id': self.private_channel_1.id})
+
+        # But with SUDO, User 2 can
+        channel_partner.sudo().channel_id = self.private_channel_1.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_1.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_acls_03(self):
+        """Test invitation in private channel part 1 (invite using crud methods)."""
+        self.private_channel_1.with_user(self.user_1).sudo().action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.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_3.partner_id.id,
+                'channel_id': self.private_channel_1.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_3.partner_id.id,
+            'channel_id': self.private_channel_1.id,
+        })
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_3.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_1.id), ('partner_id', '=', self.user_1.partner_id.id)])
+        channel_partner_3 = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.id), ('partner_id', '=', self.user_3.partner_id.id)])
+        channel_partner_3.with_user(self.user_3).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_acls_04(self):
+        """Test invitation in private channel part 2 (use `invite` action)."""
+        self.private_channel_1.with_user(self.user_1).sudo().action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.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 3
+        with self.assertRaises(AccessError):
+            self.private_channel_1.with_user(self.user_2).channel_invite([self.user_3.partner_id.id])
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id)
+
+        # User 1 is in the channel, he can not invite user 3
+        self.private_channel_1.with_user(self.user_1).channel_invite([self.user_3.partner_id.id])
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_3.partner_id)
+
+    def test_channel_acls_05(self):
+        """Test kick/leave channel."""
+        self.private_channel_1.with_user(self.user_1).sudo().action_follow()
+        self.private_channel_1.with_user(self.user_3).sudo().action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.private_channel_1.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_3).unlink()
+
+    #################
+    # GROUP CHANNEL #
+    #################
+    def test_channel_acls_06(self):
+        """Test basics on group channel."""
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel_1.id)])
+        self.assertFalse(channel_partners)
+
+        # user 1 is in the group, he can join the channel
+        self.group_channel_1.with_user(self.user_1).action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel_1.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_1.with_user(self.user_3).action_follow()
+
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel_1.id)])
+        with self.assertRaises(AccessError):
+            channel_partners.with_user(self.user_3).partner_id = self.user_3.partner_id
+
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel_1.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(AccessError):
+            self.group_channel_1.with_user(self.user_1).channel_invite([self.user_3.partner_id.id])
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.group_channel_1.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_1.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_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_2.partner_id)
+
+    ##################
+    # PUBLIC CHANNEL #
+    ##################
+    def test_channel_acls_07(self):
+        """Test basics on public channel."""
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel_1.id)])
+        self.assertFalse(channel_partners)
+
+        self.public_channel_1.with_user(self.user_1).action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id)
+
+        self.public_channel_1.with_user(self.user_2).action_follow()
+        channel_partners = self.env['mail.channel.partner'].search([('channel_id', '=', self.public_channel_1.id)])
+        self.assertEqual(channel_partners.mapped('partner_id'), self.user_1.partner_id | self.user_2.partner_id)