Small improvements and sql injection prevention.

ckanext/datastore/db.py

@@ -27,13 +27,19 @@
 _methods = [INSERT, UPSERT, UPDATE]
 
 
+def _strip(input):
+    if isinstance(input, basestring):
+        return input.strip('"')
+    return input
+
+
 def _get_list(input):
     """Transforms a string or list to a list"""
     if input == None:
         return
     if input == '':
         return []
-    return aslist(input, ',', True)
+    return [_strip(x) for x in aslist(input, ',', True)]
 
 
 def _get_bool(input, default=False):
@@ -457,7 +463,7 @@ def upsert_data(context, data_dict):
 
 
 def _get_unique_key(context, data_dict):
-    sql_get_uique_key = '''
+    sql_get_unique_key = '''
     select
         a.attname as column_names
     from
@@ -473,7 +479,7 @@ def _get_unique_key(context, data_dict):
         and idx.indisprimary = false
         and t.relname = '%s'
     '''
-    key_parts = context['connection'].execute(sql_get_uique_key, data_dict['resource_id'])
+    key_parts = context['connection'].execute(sql_get_unique_key, data_dict['resource_id'])
     return [x[0] for x in key_parts]
 
 

ckanext/datastore/logic/action.py

@@ -3,6 +3,7 @@
 import ckan.logic as logic
 import ckan.plugins as p
 import ckanext.datastore.db as db
+from sqlalchemy import text
 
 log = logging.getLogger(__name__)
 _get_or_bust = logic.get_or_bust
@@ -159,9 +160,8 @@ def datastore_search(context, data_dict):
     alias_exists = False
     if not res_exists:
         # assume id is an alias
-        alias_sql = ('select alias_of from "_table_metadata" '
-            "where name = '{}'").format(id)
-        result = db._get_engine(None, data_dict).execute(alias_sql).fetchone()
+        alias_sql = text('select alias_of from "_table_metadata" where name = :id')
+        result = db._get_engine(None, data_dict).execute(alias_sql, id=id).fetchone()
         if result:
             alias_exists = model.Resource.get(result[0].strip('"'))
 

ckanext/datastore/tests/test_datastore.py

@@ -20,6 +20,7 @@ def test_list(self):
         assert db._get_list('') == []
         assert db._get_list('foo') == ['foo']
         assert db._get_list('foo, bar') == ['foo', 'bar']
+        assert db._get_list('"foo", "bar"') == ['foo', 'bar']
         assert db._get_list(u'foo, bar') == ['foo', 'bar']
         assert db._get_list(['foo', 'bar']) == ['foo', 'bar']
         assert db._get_list([u'foo', u'bar']) == ['foo', 'bar']
@@ -443,7 +444,6 @@ def setup_class(cls):
         resource = model.Package.get('annakarenina').resources[0]
         cls.data = {
             'resource_id': resource.id,
-            'alias': 'books3',
             'fields': [{'id': u'b\xfck', 'type': 'text'},
                        {'id': 'author', 'type': 'text'},
                        {'id': 'published'}],