CVE-2020-36768: Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection') (CWE-89)
value: A vulnerability was found in rl-institut NESP2 Initial Release/1.0. It has been classified as critical. Affected is an unknown function of the file app/database.py. The manipulation leads to sql injection. It is possible to launch the attack remotely. The exploit has been disclosed to the public and may be used. The patch is identified as 07c0cdf36cf6a4345086d07b54423723a496af5e. It is recommended to apply a patch to fix this issue. VDB-246642 is the identifier assigned to this vulnerability.
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-12-03T11:15Z
Repository: https://github.com/rl-institut/NESP2
Language: Python
Files changed: app/database.py
Commit date: 2020-11-05 12:02:08 +0100
Commit message: Protect user input against sql injection

All fields which are provided by the user are now checked
converted to numerical values in python to prevent
SQL injection.

Best would be to use ORM, but SQLAlchemy does not handle it
great for materialized views, and I wanted to patch the
vulnerability asap

Diff: diff --git a/app/database.py b/app/database.py
index 65374af..2fcfb4b 100644
--- a/app/database.py
+++ b/app/database.py
@@ -1,11 +1,11 @@
 import os
 import json
 import pandas as pd
 from sqlalchemy import create_engine
 from sqlalchemy.orm import scoped_session, sessionmaker
-import geoalchemy2
+from sqlalchemy.sql import text
 from sqlalchemy import Table, MetaData
 from sqlalchemy.ext.declarative import declarative_base
 import random
 
 def get_env_variable(name):
@@ -73,68 +73,92 @@ def filter_materialized_view(
         building=None,
         buildingfp=None,
         limit=None,
         keys=None,
 ):
+    """
+
+    :param engine:
+    :param view_name: name of the view in the database (NOT A USER INPUT)
+    :param schema: name of the schema in the database (NOT A USER INPUT)
+    :param state_code: admin code of nigerian state (PROOFED USER INPUT)
+    :param area: boundaries for settlement's area filter (USER INPUT)
+    :param distance_grid: boundaries for settlement's distance to grid filter (USER INPUT)
+    :param building: boundaries for settlement's building count filter (USER INPUT)
+    :param buildingfp: boundaries for settlement's building percentage of area filter (USER INPUT)
+    :param limit: boundaries for settlement area filter (NOT A USER INPUT)
+    :param keys: list of columns to query values from (NOT A USER INPUT)
+    :return: returned data from the query
+    """
+
+    # to hold query parameters
+    values = {}
+
     if schema is not None:
         view_name = "{}.{}".format(schema, view_name)
+
     if limit is None:
         limit = ""
     else:
-        limit = " LIMIT {}".format(limit)
+        values["limit"] = int(limit)
+        limit = " LIMIT :limit"
 
-    filter_cond = ""
+    filter_cond = []
 
     if state_code is not None:
         key = "adm1_pcode"
-        filter_cond = f" WHERE {view_name}.{key}='{state_code}'"
+        filter_cond += [f"{view_name}.{key} = :{key}"]
+        values[key] = state_code
 
     if area is not None:
         key = "area_km2"
-        if "WHERE" in filter_cond:
-            filter_cond = filter_cond + f" AND {view_name}.{key} >= {area[0]} AND" \
-                                        f" {view_name}.{key} <= {area[1]}"
-        else:
-            filter_cond = f" WHERE {view_name}.{key} >= {area[0]} AND {view_name}.{key} <= {area[1]}"
+        val1 = key + "_1"
+        val2 = key + "_2"
+        filter_cond += [f"{view_name}.{key} >= :{val1}", f"{view_name}.{key} <= :{val2}"]
+        values[val1] = float(area[0])
+        values[val2] = float(area[1])
 
     if distance_grid is not None:
         key = "grid_dist_km"
-        if "WHERE" in filter_cond:
-            filter_cond = filter_cond + f" AND {view_name}.{key} >= {distance_grid[0]} AND" \
-                                        f" {view_name}.{key} <= {distance_grid[1]}"
-        else:
-            filter_cond = f" WHERE {view_name}.{key} >= {distance_grid[0]} AND" \
-                          f" {view_name}.{key} <= {distance_grid[1]}"
+        val1 = key + "_1"
+        val2 = key + "_2"
+        filter_cond += [f"{view_name}.{key} >= :{val1}", f"{view_name}.{key} <= :{val2}"]
+        values[val1] = float(distance_grid[0])
+        values[val2] = float(distance_grid[1])
 
     if building is not None:
         key = "building_count"
-        if "WHERE" in filter_cond:
-            filter_cond = filter_cond + f" AND {view_name}.{key} >= {building[0]} AND" \
-                                        f" {view_name}.{key} <= {building[1]}"
-        else:
-            filter_cond = f" WHERE {view_name}.{key} >= {building[0]} AND" \
-                          f" {view_name}.{key} <= {building[1]}"
+        val1 = key + "_1"
+        val2 = key + "_2"
+        filter_cond += [f"{view_name}.{key}>=:{val1}", f"{view_name}.{key}<=:{val2}"]
+        values[val1] = int(building[0])
+        values[val2] = int(building[1])
 
     if buildingfp is not None:
         key = "percentage_building_area"
-        if "WHERE" in filter_cond:
-            filter_cond = filter_cond + f" AND {view_name}.{key} >= {buildingfp[0]} AND" \
-                                        f" {view_name}.{key} <= {buildingfp[1]}"
-        else:
-            filter_cond = f" WHERE {view_name}.{key} >= {buildingfp[0]} AND" \
-                          f" {view_name}.{key} <= {buildingfp[1]}"
+        val1 = key + "_1"
+        val2 = key + "_2"
+        filter_cond += [f"{view_name}.{key}>=:{val1}", f"{view_name}.{key}<=:{val2}"]
+        values[val1] = float(buildingfp[0])
+        values[val2] = float(buildingfp[1])
 
     if keys is None:
         columns = "*"
     else:
         if not isinstance(keys, str):
             columns = ", ".join(keys)
         else:
             columns = "COUNT({})".format(keys)
+
+    if len(filter_cond) > 0:
+        filter_cond_str = " WHERE " + " AND ".join(filter_cond)
+    else:
+        filter_cond_str = ""
+
     with engine.connect() as con:
-        query = 'SELECT {} FROM {}{}{};'.format(columns, view_name, filter_cond, limit)
-        rs = con.execute(query)
+        query = 'SELECT {} FROM {}{}{};'.format(columns, view_name, filter_cond_str, limit)
+        rs = con.execute(text(query), **values)
         data = rs.fetchall()
     return data
 
 
 def convert_web_mat_view_to_light_json(records, cols):

Vulnerable code: import geoalchemy2
        limit = " LIMIT {}".format(limit)
    filter_cond = ""
        filter_cond = f" WHERE {view_name}.{key}='{state_code}'"
        if "WHERE" in filter_cond:
            filter_cond = filter_cond + f" AND {view_name}.{key} >= {area[0]} AND" \
                                        f" {view_name}.{key} <= {area[1]}"
        else:
            filter_cond = f" WHERE {view_name}.{key} >= {area[0]} AND {view_name}.{key} <= {area[1]}"
        if "WHERE" in filter_cond:
            filter_cond = filter_cond + f" AND {view_name}.{key} >= {distance_grid[0]} AND" \
                                        f" {view_name}.{key} <= {distance_grid[1]}"
        else:
            filter_cond = f" WHERE {view_name}.{key} >= {distance_grid[0]} AND" \
                          f" {view_name}.{key} <= {distance_grid[1]}"
        if "WHERE" in filter_cond:
            filter_cond = filter_cond + f" AND {view_name}.{key} >= {building[0]} AND" \
                                        f" {view_name}.{key} <= {building[1]}"
        else:
            filter_cond = f" WHERE {view_name}.{key} >= {building[0]} AND" \
                          f" {view_name}.{key} <= {building[1]}"
        if "WHERE" in filter_cond:
            filter_cond = filter_cond + f" AND {view_name}.{key} >= {buildingfp[0]} AND" \
                                        f" {view_name}.{key} <= {buildingfp[1]}"
        else:
            filter_cond = f" WHERE {view_name}.{key} >= {buildingfp[0]} AND" \
                          f" {view_name}.{key} <= {buildingfp[1]}"
        query = 'SELECT {} FROM {}{}{};'.format(columns, view_name, filter_cond, limit)
        rs = con.execute(query)
Fixed code: from sqlalchemy.sql import text
    """

    :param engine:
    :param view_name: name of the view in the database (NOT A USER INPUT)
    :param schema: name of the schema in the database (NOT A USER INPUT)
    :param state_code: admin code of nigerian state (PROOFED USER INPUT)
    :param area: boundaries for settlement's area filter (USER INPUT)
    :param distance_grid: boundaries for settlement's distance to grid filter (USER INPUT)
    :param building: boundaries for settlement's building count filter (USER INPUT)
    :param buildingfp: boundaries for settlement's building percentage of area filter (USER INPUT)
    :param limit: boundaries for settlement area filter (NOT A USER INPUT)
    :param keys: list of columns to query values from (NOT A USER INPUT)
    :return: returned data from the query
    """

    # to hold query parameters
    values = {}


        values["limit"] = int(limit)
        limit = " LIMIT :limit"
    filter_cond = []
        filter_cond += [f"{view_name}.{key} = :{key}"]
        values[key] = state_code
        val1 = key + "_1"
        val2 = key + "_2"
        filter_cond += [f"{view_name}.{key} >= :{val1}", f"{view_name}.{key} <= :{val2}"]
        values[val1] = float(area[0])
        values[val2] = float(area[1])
        val1 = key + "_1"
        val2 = key + "_2"
        filter_cond += [f"{view_name}.{key} >= :{val1}", f"{view_name}.{key} <= :{val2}"]
        values[val1] = float(distance_grid[0])
        values[val2] = float(distance_grid[1])
        val1 = key + "_1"
        val2 = key + "_2"
        filter_cond += [f"{view_name}.{key}>=:{val1}", f"{view_name}.{key}<=:{val2}"]
        values[val1] = int(building[0])
        values[val2] = int(building[1])
        val1 = key + "_1"
        val2 = key + "_2"
        filter_cond += [f"{view_name}.{key}>=:{val1}", f"{view_name}.{key}<=:{val2}"]
        values[val1] = float(buildingfp[0])
        values[val2] = float(buildingfp[1])

    if len(filter_cond) > 0:
        filter_cond_str = " WHERE " + " AND ".join(filter_cond)
    else:
        filter_cond_str = ""

        query = 'SELECT {} FROM {}{}{};'.format(columns, view_name, filter_cond_str, limit)
        rs = con.execute(text(query), **values)