Merge pull request #90 from postatum/101161516_hidden_fields_update

Apply privacy to request data

nefertari/utils/utils.py

@@ -167,3 +167,21 @@ def str2dict(dotted_str, value=None, separator='.'):
         if value is not None:
             prev[part] = value
     return dict_
+
+
+def validate_data_privacy(request, data):
+    """ Validate :data: contains only data allowed by privacy settings.
+
+    :param request: Pyramid Request instance
+    :param data: Dict containing request/response data which should be
+        validated
+    """
+    from nefertari import wrappers
+    wrapper = wrappers.apply_privacy(request)
+    allowed_fields = wrapper(result=data).keys()
+    data = data.copy()
+    data.pop('_type', None)
+    not_allowed_fields = set(data.keys()) - set(allowed_fields)
+
+    if not_allowed_fields:
+        raise wrappers.ValidationError(', '.join(not_allowed_fields))

nefertari/view.py

@@ -281,6 +281,11 @@ def setup_default_wrappers(self):
                 self._after_calls[meth] += [
                     wrappers.apply_privacy(self.request),
                 ]
+            for meth in ('update', 'replace', 'update_many'):
+                self._before_calls[meth] += [
+                    wrappers.apply_request_privacy(
+                        self.Model, self._json_params),
+                ]
 
     def __getattr__(self, attr):
         if attr in ACTIONS:

nefertari/view_helpers.py

@@ -1,6 +1,6 @@
 import six
 
-from nefertari.utils import dictset
+from nefertari.utils import dictset, validate_data_privacy
 from nefertari import wrappers
 from nefertari.json_httpexceptions import JHTTPForbidden
 
@@ -185,14 +185,12 @@ def check_aggregations_privacy(self, aggregations_params):
         fields_dict = dictset.fromkeys(fields)
         fields_dict['_type'] = self.view.Model.__name__
 
-        wrapper = wrappers.apply_privacy(self.view.request)
-        allowed_fields = set(wrapper(result=fields_dict).keys())
-        not_allowed_fields = set(fields) - set(allowed_fields)
-
-        if not_allowed_fields:
-            err = 'Not enough permissions to aggregate on fields: {}'.format(
-                ','.join(not_allowed_fields))
-            raise JHTTPForbidden(err)
+        try:
+            validate_data_privacy(self.view.request, fields_dict)
+        except wrappers.ValidationError as ex:
+            raise JHTTPForbidden(
+                'Not enough permissions to aggregate on '
+                'fields: {}'.format(ex))
 
     def aggregate(self):
         """ Perform aggregation and return response. """

nefertari/wrappers.py

@@ -97,8 +97,36 @@ def __call__(self, **kwargs):
         return result
 
 
+class apply_request_privacy(object):
+    """ Apply privacy rules to request data.
+
+    If request data contains fields user does not have access to,
+    JHTTPForbidden exception is raised listing all forbidden fields.
+    """
+    def __init__(self, model_cls, request_data):
+        """
+        :param model_cls: Model class affected by request.
+        :param request_data: Request data.
+        """
+        self.model_cls = model_cls
+        self.request_data = request_data
+
+    def __call__(self, **kwargs):
+        from nefertari.utils import validate_data_privacy, dictset
+        from nefertari.json_httpexceptions import JHTTPForbidden
+        request = kwargs.pop('request')
+        request_data = dictset(self.request_data)
+        request_data['_type'] = self.model_cls.__name__
+
+        try:
+            validate_data_privacy(request, request_data)
+        except ValidationError as ex:
+            raise JHTTPForbidden(
+                'Not enough permissions to update fields: {}'.format(ex))
+
+
 class apply_privacy(object):
-    """ Apply privacy rules to a JSON output.
+    """ Apply privacy rules to a JSON response.
 
     Passed 'result' kwarg's value may be a dictset or a collection JSON
     output which contains objects' data under 'data' key as a sequence of
@@ -142,7 +170,6 @@ def _filter_fields(self, data):
             else:
                 fields &= public_fields
 
-
         fields.update(['_type', '_pk', '_self'])
         return data.subset(fields)
 

tests/test_utils/test_utils.py

@@ -1,5 +1,5 @@
 import pytest
-from mock import patch, call
+from mock import patch, call, Mock
 
 from nefertari.utils import utils
 
@@ -151,3 +151,31 @@ def test_str2dict_value(self):
     def test_str2dict_separator(self):
         assert utils.str2dict('foo:bar', value=2, separator=':') == {
             'foo': {'bar': 2}}
+
+    @patch('nefertari.wrappers.apply_privacy')
+    def test_validate_data_privacy_valid(self, mock_wrapper):
+        from nefertari import wrappers
+        wrapper = Mock()
+        wrapper.return_value = {'foo': 1, 'bar': 2}
+        mock_wrapper.return_value = wrapper
+        data = {'foo': None, '_type': 'ASD'}
+        try:
+            utils.validate_data_privacy(None, data)
+        except wrappers.ValidationError:
+            raise Exception('Unexpected error')
+        mock_wrapper.assert_called_once_with(None)
+        wrapper.assert_called_once_with(result=data)
+
+    @patch('nefertari.wrappers.apply_privacy')
+    def test_validate_data_privacy_invalid(self, mock_wrapper):
+        from nefertari import wrappers
+        wrapper = Mock()
+        wrapper.return_value = {'foo': 1, 'bar': 2}
+        mock_wrapper.return_value = wrapper
+        data = {'qoo': None, '_type': 'ASD'}
+        with pytest.raises(wrappers.ValidationError) as ex:
+            utils.validate_data_privacy(None, data)
+
+        assert str(ex.value) == 'qoo'
+        mock_wrapper.assert_called_once_with(None)
+        wrapper.assert_called_once_with(result=data)

tests/test_view_helpers.py

@@ -167,7 +167,7 @@ def test_get_aggregations_fields(self):
         result = sorted(ESAggregator.get_aggregations_fields(params))
         assert result == sorted(['foo', 'bar', 'baz'])
 
-    @patch('nefertari.view.wrappers.apply_privacy')
+    @patch('nefertari.wrappers.apply_privacy')
     def test_check_aggregations_privacy_all_allowed(self, mock_privacy):
         view = self.DemoView()
         view.request = 1
@@ -186,7 +186,7 @@ def test_check_aggregations_privacy_all_allowed(self, mock_privacy):
         wrapper.assert_called_once_with(
             result={'_type': 'Zoo', 'foo': None, 'bar': None})
 
-    @patch('nefertari.view.wrappers.apply_privacy')
+    @patch('nefertari.wrappers.apply_privacy')
     def test_check_aggregations_privacy_not_allowed(self, mock_privacy):
         view = self.DemoView()
         view.request = 1

tests/test_wrappers.py

@@ -9,6 +9,7 @@
 
 from nefertari import wrappers
 from nefertari.utils import dictset
+from nefertari.json_httpexceptions import JHTTPForbidden
 
 
 class TestWrappers(unittest.TestCase):
@@ -171,6 +172,29 @@ def test_add_object_url_is_singular_property(self):
         assert wrapper.is_singular
         assert wrapper._is_singular
 
+    @patch('nefertari.utils.validate_data_privacy')
+    def test_apply_request_privacy_valid(self, mock_validate):
+        wrapper = wrappers.apply_request_privacy(
+            Mock(__name__='Foo'), {'zoo': 1})
+        try:
+            wrapper(request=4)
+        except Exception:
+            raise Exception('Unexpected error')
+        mock_validate.assert_called_once_with(
+            4, {'zoo': 1, '_type': 'Foo'})
+
+    @patch('nefertari.utils.validate_data_privacy')
+    def test_apply_request_privacy_invalid(self, mock_validate):
+        mock_validate.side_effect = wrappers.ValidationError('boo')
+        wrapper = wrappers.apply_request_privacy(
+            Mock(__name__='Foo'), {'zoo': 1})
+        with pytest.raises(JHTTPForbidden) as ex:
+            wrapper(request=4)
+        expected = 'Not enough permissions to update fields: boo'
+        assert str(ex.value) == expected
+        mock_validate.assert_called_once_with(
+            4, {'zoo': 1, '_type': 'Foo'})
+
     def test_apply_privacy_no_data(self):
         assert wrappers.apply_privacy(None)(result={}) == {}