[MERGE] trunk-acl-thu (check read/write access restrictions on fields with a groups attribute)

bzr revid: rco@openerp.com-20121217101528-nt5hsjvvodaap57m

doc/04_security.rst

@@ -141,10 +141,6 @@ for users who do not belong to the authorized groups:
 
 .. note:: The tests related to this feature are in ``openerp/tests/test_acl.py``.
  
-.. warning:: At the time of writing the implementation of this feature is partial
-             and does not yet restrict read/write RPC access to the field.
-             The corresponding test is written already but currently disabled.
-
 Workflow transition rules
 +++++++++++++++++++++++++
 

openerp/addons/base/base_demo.xml

@@ -82,7 +82,7 @@ BZrmED0AAAAASUVORK5CYII=
             <field name="signature">--
 Mr Demo</field>
             <field name="company_id" ref="main_company"/>
-            <field name="groups_id" eval="[(6,0,[ref('base.group_user')])]"/>
+            <field name="groups_id" eval="[(6,0,[ref('base.group_user'), ref('base.group_partner_manager')])]"/>
             <field name="image">/9j/4AAQSkZJRgABAQEASABIAAD/2wBDAAUDBAQEAwUEBAQFBQUGBwwIBwcHBw8LCwkMEQ8SEhEP
 ERETFhwXExQaFRERGCEYGh0dHx8fExciJCIeJBweHx7/2wBDAQUFBQcGBw4ICA4eFBEUHh4eHh4e
 Hh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh4eHh7/wAARCACEAIQDASIA

openerp/osv/fields.py

@@ -1560,7 +1560,7 @@ class column_info(object):
 
     def __str__(self):
         return '%s(%s, %s, %s, %s, %s)' % (
-            self.__name__, self.name, self.column,
+            self.__class__.__name__, self.name, self.column,
             self.parent_model, self.parent_column, self.original_parent)
 
 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:

openerp/osv/orm.py

@@ -3537,6 +3537,37 @@ 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."""
+            # Ignore requested field if it doesn't exist. This is ugly but
+            # it seems to happen at least with 'name_alias' on res.partner.
+            if field_name not in self._all_columns:
+                return True
+            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'),
+                    _('The requested operation cannot be completed due to security restrictions. '
+                    'Please contact your system administrator.\n\n(Document type: %s, Operation: %s)') % \
+                    (self._description, operation))
+        return fields
+
     def read(self, cr, user, ids, fields=None, context=None, load='_classic_read'):
         """ Read records with given ids with the given fields
 
@@ -3562,8 +3593,7 @@ class BaseModel(object):
         if not context:
             context = {}
         self.check_access_rights(cr, user, 'read')
-        if not fields:
+        fields = self.check_field_access_rights(cr, user, 'read', fields)
-            fields = list(set(self._columns.keys() + self._inherit_fields.keys()))
         if isinstance(ids, (int, long)):
             select = [ids]
         else:
@@ -4020,6 +4050,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:

openerp/tests/__init__.py

@@ -8,6 +8,7 @@ Tests can be explicitely added to the `fast_suite` or `checks` lists or not.
 See the :ref:`test-framework` section in the :ref:`features` list.
 """
 
+from . import test_acl
 from . import test_expression, test_mail, test_ir_sequence, test_orm, \
               test_fields, test_basecase, \
               test_view_validation, test_uninstall, test_misc, test_db_cursor, \
@@ -20,6 +21,7 @@ fast_suite = [
 ]
 
 checks = [
+    test_acl,
     test_expression,
     test_mail,
     test_db_cursor,

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
+        with self.assertRaises(openerp.osv.orm.except_orm):
-        # from openerp.osv.orm import except_orm
+            self.res_partner.read(self.cr, self.demo_uid, [1], ['bank_ids'])
-        # with self.assertRaises(except_orm):
+        with self.assertRaises(openerp.osv.orm.except_orm):
-        #     self.res_partner.read(self.cr, self.demo_uid, [1], ['bank_ids'])
+            self.res_partner.write(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': []})
 
         # 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:
+# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:

openerp/tests/test_fields.py

@@ -53,31 +53,46 @@ class TestRelatedField(common.TransactionCase):
     def test_1_single_related(self):
         """ test a related field with a single indirection like fields.related('foo') """
         # add a related field test_related_company_id on res.partner
+        # and simulate a _inherits_reload() to populate _all_columns.
         old_columns = self.partner._columns
+        old_all_columns = self.partner._all_columns
         self.partner._columns = dict(old_columns)
+        self.partner._all_columns = dict(old_all_columns)
         self.partner._columns.update({
             'single_related_company_id': fields.related('company_id', type='many2one', obj='res.company'),
         })
+        self.partner._all_columns.update({
+            'single_related_company_id': fields.column_info('single_related_company_id', self.partner._columns['single_related_company_id'], None, None, None)
+        })
 
         self.do_test_company_field('single_related_company_id')
 
         # restore res.partner fields
         self.partner._columns = old_columns
+        self.partner._all_columns = old_all_columns
 
     def test_2_related_related(self):
         """ test a related field referring to a related field """
         # add a related field on a related field on res.partner
+        # and simulate a _inherits_reload() to populate _all_columns.
         old_columns = self.partner._columns
+        old_all_columns = self.partner._all_columns
         self.partner._columns = dict(old_columns)
+        self.partner._all_columns = dict(old_all_columns)
         self.partner._columns.update({
             'single_related_company_id': fields.related('company_id', type='many2one', obj='res.company'),
             'related_related_company_id': fields.related('single_related_company_id', type='many2one', obj='res.company'),
         })
+        self.partner._all_columns.update({
+            'single_related_company_id': fields.column_info('single_related_company_id', self.partner._columns['single_related_company_id'], None, None, None),
+            'related_related_company_id': fields.column_info('related_related_company_id', self.partner._columns['related_related_company_id'], None, None, None)
+        })
 
         self.do_test_company_field('related_related_company_id')
 
         # restore res.partner fields
         self.partner._columns = old_columns
+        self.partner._all_columns = old_all_columns
 
     def test_3_read_write(self):
         """ write on a related field """