Merge pull request #337 from Eficent/11.0-imp-mail_activity_board

[11.0][IMP] mail_activity_board: filter activities by user read permissions

mail_activity_board/__manifest__.py

@@ -3,7 +3,7 @@
 {
     'name': 'Activities board',
     'summary': 'Add Activity Boards',
-    'version': '11.0.1.0.0',
+    'version': '11.0.1.0.1',
     'development_status': 'Beta',
     'category': 'Social Network',
     'website': 'https://github.com/OCA/social',

mail_activity_board/models/mail_activity.py

@@ -1,6 +1,6 @@
 # Copyright 2018 David Juaneda - <djuaneda@sdi.es>
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
-from odoo import api, models, fields
+from odoo import api, models, fields, SUPERUSER_ID
 
 
 class MailActivity(models.Model):
@@ -43,3 +43,64 @@ class MailActivity(models.Model):
         action = self.env.ref(
             'mail_activity_board.open_boards_activities').read()[0]
         return action
+
+    @api.model
+    def _find_allowed_model_wise(self, doc_model, doc_dict):
+        doc_ids = list(doc_dict)
+        allowed_doc_ids = self.env[doc_model].with_context(
+            active_test=False).search([('id', 'in', doc_ids)]).ids
+        return set([message_id for allowed_doc_id in allowed_doc_ids
+                    for message_id in doc_dict[allowed_doc_id]])
+
+    @api.model
+    def _find_allowed_doc_ids(self, model_ids):
+        IrModelAccess = self.env['ir.model.access']
+        allowed_ids = set()
+        for doc_model, doc_dict in model_ids.items():
+            if not IrModelAccess.check(doc_model, 'read', False):
+                continue
+            allowed_ids |= self._find_allowed_model_wise(doc_model, doc_dict)
+        return allowed_ids
+
+    @api.model
+    def _search(self, args, offset=0, limit=None, order=None, count=False,
+                access_rights_uid=None):
+        # Rules do not apply to administrator
+        if self._uid == SUPERUSER_ID:
+            return super(MailActivity, self)._search(
+                args, offset=offset, limit=limit, order=order,
+                count=count, access_rights_uid=access_rights_uid)
+
+        ids = super(MailActivity, self)._search(
+            args, offset=offset, limit=limit, order=order,
+            count=False, access_rights_uid=access_rights_uid)
+        if not ids and count:
+            return 0
+        elif not ids:
+            return ids
+
+        # check read access rights before checking the actual rules
+        super(MailActivity, self.sudo(access_rights_uid or self._uid)).\
+            check_access_rights('read')
+
+        model_ids = {}
+
+        self._cr.execute("""
+            SELECT DISTINCT a.id, im.id, im.model, a.res_id
+            FROM "%s" a
+            LEFT JOIN ir_model im ON im.id = a.res_model_id
+            WHERE a.id = ANY (%%(ids)s)""" % self._table, dict(ids=ids))
+        for a_id, ir_model_id, model, model_id in self._cr.fetchall():
+            model_ids.setdefault(model, {}).setdefault(
+                model_id, set()).add(a_id)
+
+        allowed_ids = self._find_allowed_doc_ids(model_ids)
+
+        final_ids = allowed_ids
+
+        if count:
+            return len(final_ids)
+        else:
+            # re-construct a list based on ids, because set didn't keep order
+            id_list = [a_id for a_id in ids if a_id in final_ids]
+            return id_list

mail_activity_board/tests/test_mail_activity_board.py

@@ -18,6 +18,16 @@ class TestMailActivityBoardMethods(TransactionCase):
             'groups_id': [(6, 0, [self.env.ref('base.group_user').id])]
         })
 
+        # Create a user who doesn't have access to anything except activities
+        mail_activity_group = self.create_mail_activity_group()
+        self.employee2 = self.env['res.users'].create({
+            'company_id': self.env.ref("base.main_company").id,
+            'name': "Employee2",
+            'login': "alien",
+            'email': "alien@yourcompany.com",
+            'groups_id': [(6, 0, [mail_activity_group.id])],
+        })
+
         # lead_model_id = self.env['ir.model']._get('crm.lead').id
         partner_model_id = self.env['ir.model']._get('res.partner').id
 
@@ -45,6 +55,9 @@ class TestMailActivityBoardMethods(TransactionCase):
         # I create an opportunity, as employee
         self.partner_client = self.env.ref("base.res_partner_1")
 
+        # assure there isn't any mail activity yet
+        self.env['mail.activity'].sudo().search([]).unlink()
+
         self.act1 = self.env['mail.activity'].sudo().create({
             'activity_type_id': self.activity3.id,
             'note': 'Partner activity 1.',
@@ -67,6 +80,23 @@ class TestMailActivityBoardMethods(TransactionCase):
             'user_id': self.employee.id
         })
 
+    def create_mail_activity_group(self):
+        manager_mail_activity_test_group = self.env['res.groups'].create({
+            'name': 'group_manager_mail_activity_test',
+        })
+        mail_activity_model_id = self.env['ir.model'].sudo().search(
+            [('model', '=', 'mail.activity')], limit=1)
+        access = self.env['ir.model.access'].create({
+            'name': 'full_access_mail_activity',
+            'model_id': mail_activity_model_id.id,
+            'perm_read': True,
+            'perm_write': True,
+            'perm_create': True,
+            'perm_unlink': True,
+        })
+        access.group_id = manager_mail_activity_test_group
+        return manager_mail_activity_test_group
+
     def get_view(self, activity):
         action = activity.open_origin()
         result = self.env[action.get('res_model')]\
@@ -114,7 +144,7 @@ class TestMailActivityBoardMethods(TransactionCase):
         kwargs = {
             'groupby': [
                 "activity_type_id"
-            ]
+            ],
         }
         kwargs['domain'] = action.get('domain')
 
@@ -130,7 +160,13 @@ class TestMailActivityBoardMethods(TransactionCase):
             records = self.env['mail.activity'].search_read(
                 domain=group.get('__domain'), fields=kwargs['fields']
             )
-            acts += [id.get('id') for id in records]
+            acts += [record_id.get('id') for record_id in records]
 
         for act in acts:
             self.assertIn(act, self.partner_client.activity_ids.ids)
+
+    def test_read_permissions(self):
+        search1 = self.env['mail.activity'].sudo(self.employee).search([])
+        self.assertEqual(len(search1), 3)
+        search2 = self.env['mail.activity'].sudo(self.employee2).search([])
+        self.assertEqual(len(search2), 0)