[IMP] mail: added _search override on mail_message, to filter the results of a search according to our custom rules. Added some tests.

bzr revid: tde@openerp.com-20121024141806-8iayf7cz2ndxyz2l

addons/mail/mail_message.py

@@ -433,7 +433,7 @@ class mail_message(osv.Model):
     #     return attachment_list
 
     #------------------------------------------------------
-    # Email api
+    # mail_message internals
     #------------------------------------------------------
 
     def init(self, cr):
@@ -441,6 +441,61 @@ class mail_message(osv.Model):
         if not cr.fetchone():
             cr.execute("""CREATE INDEX mail_message_model_res_id_idx ON mail_message (model, res_id)""")
 
+    def _search(self, cr, uid, args, offset=0, limit=None, order=None,
+        context=None, count=False, access_rights_uid=None):
+        """ Override that adds specific access rights of mail.message, to remove
+            ids uid could not see according to our custom rules. Please refer
+            to check_access_rule for more details about those rules.
+
+            After having received ids of a classic search, keep only:
+            - if author_id == pid, uid is the author, OR
+            - a notification (id, pid) exists, uid has been notified, OR
+            - uid have read access to the related document is model, res_id
+            - otherwise: remove the id
+        """
+        # Rules do not apply to administrator
+        if uid == SUPERUSER_ID:
+            return super(mail_message, self)._search(cr, uid, args, offset=offset, limit=limit, order=order,
+            context=context, count=count, access_rights_uid=access_rights_uid)
+        # Perform a super with count as False, to have the ids, not a counter
+        ids = super(mail_message, self)._search(cr, uid, args, offset=offset, limit=limit, order=order,
+            context=context, count=False, access_rights_uid=access_rights_uid)
+        if not ids and count:
+            return 0
+        elif not ids:
+            return ids
+
+        author_ids = set([])
+        partner_ids = set([])
+        allowed_ids = set([])
+        model_ids = {}
+
+        pid = self.pool.get('res.users').read(cr, uid, uid, ['partner_id'])['partner_id'][0]
+        messages = super(mail_message, self).read(cr, uid, ids, ['author_id', 'model', 'res_id', 'partner_ids'])
+        for message in messages:
+            if message.get('author_id') and message.get('author_id')[0] == pid:
+                author_ids.add(message.get('id'))
+            elif pid in message.get('partner_ids'):
+                partner_ids.add(message.get('id'))
+            elif message.get('model') and message.get('res_id'):
+                model_ids.setdefault(message.get('model'), {}).setdefault(message.get('res_id'), set()).add(message.get('id'))
+        # print '_search', ids, author_ids, partner_ids, model_ids
+
+        model_access_obj = self.pool.get('ir.model.access')
+        for doc_model, doc_dict in model_ids.iteritems():
+            # print '>>>>', doc_model, doc_dict
+            if not model_access_obj.check(cr, uid, doc_model, 'read', False):
+                continue
+            doc_ids = doc_dict.keys()
+            allowed_doc_ids = self.pool.get(doc_model).search(cr, uid, [('id', 'in', doc_ids)], context=context)
+            allowed_ids |= set([message_id for allowed_doc_id in allowed_doc_ids for message_id in doc_dict[allowed_doc_id]])
+
+        final_ids = author_ids | partner_ids | allowed_ids
+        if count:
+            return len(final_ids)
+        else:
+            return list(final_ids)
+
     def check_access_rule(self, cr, uid, ids, operation, context=None):
         """ Access rules of mail.message:
             - read: if
@@ -449,7 +504,7 @@ class mail_message(osv.Model):
                 - I can read the related document if res_model, res_id
                 - Otherwise: raise
             - create: if
-                - I am in the document message_follower_ids OR
+                - I am in the document message_follower_ids if res_model, res_id OR
                 - I can write on the related document if res_model, res_id OR
                 - I create a private message (no model, no res_id)
                 - Otherwise: raise
@@ -568,6 +623,10 @@ class mail_message(osv.Model):
             self.pool.get('ir.attachment').unlink(cr, uid, attachments_to_delete, context=context)
         return super(mail_message, self).unlink(cr, uid, ids, context=context)
 
+    #------------------------------------------------------
+    # Messaging API
+    #------------------------------------------------------
+
     def _notify_followers(self, cr, uid, newid, message, context=None):
         """ Add the related record followers to the destination partner_ids.
         """
@@ -606,7 +665,7 @@ class mail_message(osv.Model):
         self.pool.get('mail.notification')._notify(cr, uid, newid, context=context)
 
     def copy(self, cr, uid, id, default=None, context=None):
-        """Overridden to avoid duplicating fields that are unique to each email"""
+        """ Overridden to avoid duplicating fields that are unique to each email """
         if default is None:
             default = {}
         default.update(message_id=False, headers=False)

addons/mail/security/ir.model.access.csv

@@ -2,6 +2,7 @@ id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
 access_mail_message_all,mail.message.all,model_mail_message,,1,0,1,0
 access_mail_message_group_user,mail.message.group.user,model_mail_message,base.group_user,1,1,1,1
 access_mail_mail_all,mail.mail.all,model_mail_mail,,0,0,1,0
+access_mail_mail_user,mail.mail,model_mail_mail,base.group_user,1,1,1,0
 access_mail_mail_system,mail.mail.system,model_mail_mail,base.group_system,1,1,1,1
 access_mail_followers_all,mail.followers.all,model_mail_followers,,1,0,0,0
 access_mail_followers_system,mail.followers.system,model_mail_followers,base.group_system,1,1,1,1
@@ -12,6 +13,6 @@ access_mail_group_user,mail.group.user,model_mail_group,base.group_user,1,1,1,1
 access_mail_alias_all,mail.alias.all,model_mail_alias,,1,0,0,0
 access_mail_alias_user,mail.alias,model_mail_alias,base.group_user,1,1,1,0
 access_mail_alias_system,mail.alias,model_mail_alias,base.group_system,1,1,1,1
-access_mail_message_subtype,mail.message.subtype,model_mail_message_subtype,,1,1,1,1
-access_mail_mail_user,mail.mail,model_mail_mail,base.group_user,1,1,1,0
+access_mail_message_subtype_all,mail.message.subtype.all,model_mail_message_subtype,,1,0,0,0
 access_mail_vote_all,mail.vote.all,model_mail_vote,,1,1,1,1
+access_mail_favorite_all,mail.favorite.all,model_mail_favorite,,1,1,1,1

addons/mail/tests/test_mail_access_rights.py

@@ -51,7 +51,37 @@ class test_mail_access_rights(test_mail.TestMailMockups):
         self.user_raoul = self.res_users.browse(cr, uid, self.user_raoul_id)
         self.partner_raoul_id = self.user_raoul.partner_id.id
 
-    def test_00_mail_message_read_access_rights(self):
+    def test_00_mail_message_search_access_rights(self):
+        """ Test mail_message search override about access rights. """
+        cr, uid, group_pigs_id = self.cr, self.uid, self.group_pigs_id
+        partner_bert_id, partner_raoul_id = self.partner_bert_id, self.partner_raoul_id
+        user_bert_id, user_raoul_id = self.user_bert_id, self.user_raoul_id
+
+        # Data: Birds group, private
+        group_birds_id = self.mail_group.create(self.cr, self.uid, {'name': 'Birds', 'public': 'private'})
+        # Data: raoul is member of Pigs
+        self.mail_group.message_subscribe(cr, uid, [group_pigs_id], [partner_raoul_id])
+        # Data: various author_ids, partner_ids, documents
+        msg_id1 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A'})
+        msg_id2 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A+B', 'partner_ids': [(6, 0, [partner_bert_id])]})
+        msg_id3 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A Pigs', 'model': 'mail.group', 'res_id': group_pigs_id})
+        msg_id4 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A+B Pigs', 'model': 'mail.group', 'res_id': group_pigs_id, 'partner_ids': [(6, 0, [partner_bert_id])]})
+        msg_id5 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A+R Pigs', 'model': 'mail.group', 'res_id': group_pigs_id, 'partner_ids': [(6, 0, [partner_raoul_id])]})
+        msg_id6 = self.mail_message.create(cr, uid, {'subject': 'Test', 'body': 'A Birds', 'model': 'mail.group', 'res_id': group_birds_id})
+        msg_id7 = self.mail_message.create(cr, user_bert_id, {'subject': 'Test', 'body': 'B'})
+        msg_id8 = self.mail_message.create(cr, user_bert_id, {'subject': 'Test', 'body': 'B+R', 'partner_ids': [(6, 0, [partner_raoul_id])]})
+
+        # Test: Bert: 2 messages that have Bert in partner_ids + 2 messages as author
+        msg_ids = self.mail_message.search(cr, user_bert_id, [('subject', 'like', 'Test')])
+        self.assertEqual(set([msg_id2, msg_id4, msg_id7, msg_id8]), set(msg_ids), 'mail_message search failed')
+        # Test: Raoul: 3 messages on Pigs Raoul can read (employee can read group with default values), 0 on Birds (private group)
+        msg_ids = self.mail_message.search(cr, user_raoul_id, [('body', 'like', 'A')])
+        self.assertEqual(set([msg_id3, msg_id4, msg_id5]), set(msg_ids), 'mail_message search failed')
+        # Test: Admin: all messages
+        msg_ids = self.mail_message.search(cr, uid, [('subject', 'like', 'Test')])
+        self.assertEqual(set([msg_id1, msg_id2, msg_id3, msg_id4, msg_id5, msg_id6, msg_id7, msg_id8]), set(msg_ids), 'mail_message search failed')
+
+    def test_05_mail_message_read_access_rights(self):
         """ Test basic mail_message read access rights. """
         cr, uid = self.cr, self.uid
         partner_bert_id, partner_raoul_id = self.partner_bert_id, self.partner_raoul_id
@@ -98,10 +128,6 @@ class test_mail_access_rights(test_mail.TestMailMockups):
         self.assertRaises(except_orm, self.mail_message.read,
             cr, user_bert_id, message_id)
 
-    def test_05_mail_message_search_access_rights(self):
-        """ Test mail_message search override about access rights. """
-        self.assertTrue(1 == 1, 'Test not implemented, do not replace by return True')
-
     def test_10_mail_flow_access_rights(self):
         """ Test a Chatter-looks alike flow. """
         cr, uid = self.cr, self.uid