From 9d2afcae3fa570e1ac97697cdf2f75872a81fbb2 Mon Sep 17 00:00:00 2001
From: Vo Minh Thu <vmt@openerp.com>
Date: Wed, 12 Dec 2012 12:36:47 +0100
Subject: [PATCH] [IMP] orm: check groups-based access rights on model fields
 in read() and write().

The commented-out tests present in test_acl.py now pass.
Other tests now fail :-(.

bzr revid: vmt@openerp.com-20121212113647-11y3buulifg6tyhj
---
 openerp/osv/orm.py        | 27 +++++++++++++++++++++++++--
 openerp/tests/test_acl.py | 16 +++++++++-------
 2 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/openerp/osv/orm.py b/openerp/osv/orm.py
index 060d6da0277..1c0afa7131a 100644
--- a/openerp/osv/orm.py
+++ b/openerp/osv/orm.py
@@ -3536,6 +3536,29 @@ class BaseModel(object):
 
         return res
 
+    def check_field_access_rights(self, cr, user, operation, fields, context=None):
+        """
+        Check the user access rights on the given fields. This raises Access
+        Denied if the user does not have the rights. Otherwise it returns the
+        fields (as is if the fields is not falsy, or the readable/writable
+        fields if fields is falsy).
+        """
+        def p(field_name):
+            """Predicate to test if the user has access to the given field name."""
+            field = self._all_columns[field_name].column
+            if field.groups:
+                return self.user_has_groups(cr, user, groups=field.groups, context=context)
+            else:
+                return True
+        if not fields:
+            fields = filter(p, self._all_columns.keys())
+        else:
+            filtered_fields = filter(lambda a: not p(a), fields)
+            if filtered_fields:
+                _logger.warning('Access Denied by ACLs for operation: %s, uid: %s, model: %s, fields: %s', operation, user, self._name, ', '.join(filtered_fields))
+                raise except_orm(_('Access Denied'), 'TODO')
+        return fields
+
     def read(self, cr, user, ids, fields=None, context=None, load='_classic_read'):
         """ Read records with given ids with the given fields
 
@@ -3561,8 +3584,7 @@ class BaseModel(object):
         if not context:
             context = {}
         self.check_access_rights(cr, user, 'read')
-        if not fields:
-            fields = list(set(self._columns.keys() + self._inherit_fields.keys()))
+        fields = self.check_field_access_rights(cr, user, 'read', fields)
         if isinstance(ids, (int, long)):
             select = [ids]
         else:
@@ -4018,6 +4040,7 @@ class BaseModel(object):
 
         """
         readonly = None
+        self.check_field_access_rights(cr, user, 'write', vals.keys())
         for field in vals.copy():
             fobj = None
             if field in self._columns:
diff --git a/openerp/tests/test_acl.py b/openerp/tests/test_acl.py
index a10a705092c..1d8d6bfb22a 100644
--- a/openerp/tests/test_acl.py
+++ b/openerp/tests/test_acl.py
@@ -1,6 +1,9 @@
 import unittest2
 from lxml import etree
 
+import openerp
+from openerp.tools.misc import mute_logger
+
 import common
 
 # test group that demo user should not have
@@ -55,6 +58,7 @@ class TestACL(common.TransactionCase):
         self.tech_group.write({'users': [(3, self.demo_uid)]})
         self.res_currency._columns['rate'].groups = False
 
+    @mute_logger('openerp.osv.orm')
     def test_field_crud_restriction(self):
         "Read/Write RPC access to restricted field should be forbidden"
         # Verify the test environment first
@@ -65,12 +69,10 @@ class TestACL(common.TransactionCase):
 
         # Now restrict access to the field and check it's forbidden
         self.res_partner._columns['bank_ids'].groups = GROUP_TECHNICAL_FEATURES
-        # FIXME TODO: enable next tests when access rights checks per field are implemented
-        # from openerp.osv.orm import except_orm
-        # with self.assertRaises(except_orm):
-        #     self.res_partner.read(self.cr, self.demo_uid, [1], ['bank_ids'])
-        # with self.assertRaises(except_orm):
-        #     self.res_partner.write(self.cr, self.demo_uid, [1], {'bank_ids': []})
+        with self.assertRaises(openerp.osv.orm.except_orm):
+            self.res_partner.read(self.cr, self.demo_uid, [1], ['bank_ids'])
+        with self.assertRaises(openerp.osv.orm.except_orm):
+            self.res_partner.write(self.cr, self.demo_uid, [1], {'bank_ids': []})
 
         # Add the restricted group, and check that it works again
         self.tech_group.write({'users': [(4, self.demo_uid)]})
@@ -86,4 +88,4 @@ class TestACL(common.TransactionCase):
 if __name__ == '__main__':
     unittest2.main()
 
-# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
\ No newline at end of file
+# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: