CVE-2014-125065: Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection') (CWE-89)
value: A vulnerability, which was classified as critical, was found in john5223 bottle-auth. Affected is an unknown function. The manipulation leads to sql injection. The name of the patch is 99cfbcc0c1429096e3479744223ffb4fda276875. It is recommended to apply a patch to fix this issue. The identifier of this vulnerability is VDB-217632.
CWE description: The product constructs all or part of an SQL command using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify the intended SQL command when it is sent to a downstream component. Without sufficient removal or quoting of SQL syntax in user-controllable inputs, the generated SQL query can cause those inputs to be interpreted as SQL instead of ordinary user data.
CVSS v3 base score: 9.8
Published: 2023-01-07T20:15Z
Repository: https://github.com/john5223/bottle-auth
Language: Python
Files changed: auth/controllers/group_controller.py, auth/controllers/user_controller.py, tests/test_integration.py, tests/test_units.py
Commit date: 2014-12-15 10:45:53 -0600
Commit message: no more sql injection and better tests

Diff: diff --git a/auth/controllers/group_controller.py b/auth/controllers/group_controller.py
index 541bb7d..062f7a2 100644
--- a/auth/controllers/group_controller.py
+++ b/auth/controllers/group_controller.py
@@ -30,13 +30,21 @@ def get_group(db, group_name):
     group = groups_table.find(name=group_name)
     rows = [x for x in group]
     if not rows:
         return error(404, {'error': 'Not a valid group'})
 
-    userids = ["'%s'" % x['userid'] for x in rows if x['userid']]
-    users = db.query("SELECT * FROM users WHERE userid IN (%s) " % ','.join(userids))
-    ret = {group_name: list(users) }
+    userids = [x['userid'] for x in rows if x['userid']]
+    if not userids:
+        return {group_name: []}
+
+    params = {}
+    for i, userid in enumerate(userids,1):
+        params['userid_' + str(i)] = str(userid)
+    where_clause = 'userid IN(:' + ",:".join(params.keys()) + ')' # b/c sqlalchemy can't use a list!?
+    q = "SELECT * FROM users WHERE " + where_clause
+    users = db.executable.execute(q, params).fetchall()
+    ret = {group_name: [dict(x.items()) for x in users] }
     return ret
 
 
 @route('/groups/<group_name>', method=['POST', 'PUT'])
 def post_group(db, group_name):
diff --git a/auth/controllers/user_controller.py b/auth/controllers/user_controller.py
index 05640a7..b528bb0 100644
--- a/auth/controllers/user_controller.py
+++ b/auth/controllers/user_controller.py
@@ -36,11 +36,11 @@ def delete_user(db, userid):
         user_table.delete(userid=userid)
         return {'status': 200}
 
 
 @route('/users/<userid>', method=['POST', 'PUT'])
-def create__update_user(db, userid):
+def create_update_user(db, userid):
     data = request.json
     data_keys = data.keys()
     required_fields = ['first_name', 'last_name', 'userid', 'groups']
     missing_fields = [x for x in required_fields if x not in data_keys]
     extra_fields = [x for x in data_keys if x not in required_fields]
@@ -69,12 +69,15 @@ def create__update_user(db, userid):
     for name in groups:
         groups_table.upsert(dict(name=name, userid=userid), ['name','userid'])
 
     if request.method == 'PUT':
         #get rid of any old groups for this user
-        q = '''DELETE FROM groups 
-                WHERE userid='{userid}'
-                            AND name not in ({group_names})
-              '''.format(userid=userid, group_names=','.join(["'%s'" % x for x in groups]))
-        db.query(q)
+        params = {}
+        for counter, group in enumerate(groups,1):
+            params["group_name" + str(counter)] = group
+            counter += 1
+        where_clause = 'name NOT IN(:' + ",:".join(params.keys()) + ')' # b/c sqlalchemy can't use a list!?
+        params['userid'] = userid
+        q = '''DELETE FROM groups WHERE userid=:userid AND ''' + where_clause
+        db.executable.execute(q, params)
 
     return {'status': 200, 'user': get_user(db, userid)}
diff --git a/tests/test_auth.py b/tests/test_integration.py
similarity index 98%
rename from tests/test_auth.py
rename to tests/test_integration.py
index 5bca4e2..dc78a66 100644
--- a/tests/test_auth.py
+++ b/tests/test_integration.py
@@ -6,11 +6,11 @@ from webtest import TestApp #Docs: http://webtest.pythonpaste.org/en/latest/
 from auth import simpleauth
 
 
 app = TestApp(simpleauth.app)
 
-def test_users():
+def test_user_integration():
     headers = [('Content-type', 'application/json')]
     payload = { "first_name": "Joe",
 	    "last_name": "Smith",
 	    "userid": "jsmith",
 	    "groups": ["admins", "users"] }
@@ -53,11 +53,11 @@ def test_users():
     #DELETE
     delete_user = app.delete('/users/%s' % payload.get('userid'))
     assert delete_user.status == '200 OK'
 
 
-def test_groups():
+def test_groups_integration():
     headers = [('Content-type', 'application/json')]
     payload = { "first_name": "Joe",
         "last_name": "Smith",
         "userid": "jsmith2",
         "groups": ["admins", "users"] }
diff --git a/tests/test_units.py b/tests/test_units.py
new file mode 100644
index 0000000..79a03ac
--- /dev/null
+++ b/tests/test_units.py
@@ -0,0 +1,35 @@
+import logging
+logging.basicConfig(level=logging.DEBUG)
+import json
+
+from webtest import TestApp #Docs: http://webtest.pythonpaste.org/en/latest/
+from auth import simpleauth
+
+
+app = TestApp(simpleauth.app)
+
+def test_sql_injection():
+	
+	pass
+
+
+def test_create_user():
+	pass
+
+def test_update_user():
+	pass
+
+def test_delete_user():
+	pass
+
+
+
+def test_create_group():
+	pass
+
+def test_update_group():
+	pass
+
+def test_delete_group():
+	pass
+

Vulnerable code:     userids = ["'%s'" % x['userid'] for x in rows if x['userid']]
    users = db.query("SELECT * FROM users WHERE userid IN (%s) " % ','.join(userids))
    ret = {group_name: list(users) }
def create__update_user(db, userid):
        q = '''DELETE FROM groups 
                WHERE userid='{userid}'
                            AND name not in ({group_names})
              '''.format(userid=userid, group_names=','.join(["'%s'" % x for x in groups]))
        db.query(q)
def test_users():
def test_groups():
Fixed code:     userids = [x['userid'] for x in rows if x['userid']]
    if not userids:
        return {group_name: []}

    params = {}
    for i, userid in enumerate(userids,1):
        params['userid_' + str(i)] = str(userid)
    where_clause = 'userid IN(:' + ",:".join(params.keys()) + ')' # b/c sqlalchemy can't use a list!?
    q = "SELECT * FROM users WHERE " + where_clause
    users = db.executable.execute(q, params).fetchall()
    ret = {group_name: [dict(x.items()) for x in users] }
def create_update_user(db, userid):
        params = {}
        for counter, group in enumerate(groups,1):
            params["group_name" + str(counter)] = group
            counter += 1
        where_clause = 'name NOT IN(:' + ",:".join(params.keys()) + ')' # b/c sqlalchemy can't use a list!?
        params['userid'] = userid
        q = '''DELETE FROM groups WHERE userid=:userid AND ''' + where_clause
        db.executable.execute(q, params)
def test_user_integration():
def test_groups_integration():
import logging
logging.basicConfig(level=logging.DEBUG)
import json

from webtest import TestApp #Docs: http://webtest.pythonpaste.org/en/latest/
from auth import simpleauth


app = TestApp(simpleauth.app)

def test_sql_injection():
	
	pass


def test_create_user():
	pass

def test_update_user():
	pass

def test_delete_user():
	pass



def test_create_group():
	pass

def test_update_group():
	pass

def test_delete_group():
	pass