[#2733] Check that limit and offset are integers.

ckan · Jul 31, 2012 · b264dcb · b264dcb
1 parent 67e6f4c
commit b264dcb
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 0 deletions.
diff --git a/ckanext/datastore/db.py b/ckanext/datastore/db.py
@@ -23,6 +23,15 @@ def _is_valid_field_name(name):
     return True
 
 
+def _validate_int(i, field_name):
+    try:
+        int(i)
+    except ValueError:
+        raise p.toolkit.ValidationError({
+            'field_name': '{} is not an integer'.format(i)
+        })
+
+
 def _get_engine(context, data_dict):
     'Get either read or write engine.'
     connection_url = data_dict['connection_url']
@@ -341,6 +350,9 @@ def search_data(context, data_dict):
     limit = data_dict.get('limit', 100)
     offset = data_dict.get('offset', 0)
 
+    _validate_int(limit, 'limit')
+    _validate_int(offset, 'offset')
+
     if data_dict.get('sort'):
         sort = 'order by {}'.format(data_dict['sort'])
     else:

diff --git a/ckanext/datastore/tests/test_datastore.py b/ckanext/datastore/tests/test_datastore.py
@@ -578,6 +578,16 @@ def test_search_limit(self):
         assert result['records'] == [{'book': 'annakarenina',
                                       'author': 'tolstoy'}]
 
+    def test_search_invalid_limit(self):
+        data = {'resource_id': self.data['resource_id'],
+                'limit': 'bad'}
+        postparams = '%s=1' % json.dumps(data)
+        auth = {'Authorization': str(self.sysadmin_user.apikey)}
+        res = self.app.post('/api/action/datastore_search', params=postparams,
+                            extra_environ=auth, status=409)
+        res_dict = json.loads(res.body)
+        assert res_dict['success'] is False
+
     def test_search_offset(self):
         data = {'resource_id': self.data['resource_id'],
                 'limit': 1,
@@ -593,6 +603,16 @@ def test_search_offset(self):
         assert result['records'] == [{'book': 'warandpeace',
                                       'author': 'tolstoy'}]
 
+    def test_search_invalid_offset(self):
+        data = {'resource_id': self.data['resource_id'],
+                'offset': 'bad'}
+        postparams = '%s=1' % json.dumps(data)
+        auth = {'Authorization': str(self.sysadmin_user.apikey)}
+        res = self.app.post('/api/action/datastore_search', params=postparams,
+                            extra_environ=auth, status=409)
+        res_dict = json.loads(res.body)
+        assert res_dict['success'] is False
+
     def test_search_full_text(self):
         data = {'resource_id': self.data['resource_id'],
                 'q': 'annakarenina'}