CVE-2018-25088: 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 Blue Yonder postgraas_server up to 2.0.0b2. Affected is the function _create_pg_connection/create_postgres_db of the file postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py of the component PostgreSQL Backend Handler. The manipulation leads to sql injection. Upgrading to version 2.0.0 is able to address this issue. The patch is identified as 7cd8d016edc74a78af0d81c948bfafbcc93c937c. It is recommended to upgrade the affected component. VDB-234246 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-07-18T13:15Z
Repository: https://github.com/blue-yonder/postgraas_server
Language: Python
Files changed: postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py, tests/test_integration/backends/postgres_cluster/test_postgres_cluster_driver.py, tests/test_integration/test_backend_behaviour.py
Commit date: 2018-02-01 16:07:25 +0100
Commit message: Harden the database creation against SQL injections

If using the PostgreSQL backend, database creation was previously vulnerable to SQL injections. This commit fixes these.

There are two tests which no longer behave as expected and have been disabled for now. They use usernames that previously broke the SQL statement but have now become valid. We might want to think about restricting the possible username somewhat again.

Diff: diff --git a/postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py b/postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py
index 1337af8..5b8995b 100644
--- a/postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py
+++ b/postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py
@@ -1,7 +1,8 @@
 import psycopg2
 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
+from psycopg2.sql import SQL, Identifier
 
 
 def _create_pg_connection(config):
     if 'server' in config:
         username = '@'.join([config['username'], config['server']])
@@ -17,13 +18,13 @@ def _create_pg_connection(config):
 
 
 def check_db_or_user_exists(db_name, db_user, config):
     with _create_pg_connection(config) as con:
         with con.cursor() as cur:
-            cur.execute("SELECT 1 FROM pg_database WHERE datname='{}';".format(db_name))
+            cur.execute("SELECT 1 FROM pg_database WHERE datname=%s;", (db_name, ))
             db_exists = cur.fetchone() is not None
-            cur.execute("SELECT 1 FROM pg_roles WHERE rolname='{}';".format(db_user))
+            cur.execute("SELECT 1 FROM pg_roles WHERE rolname=%s;", (db_user, ))
             user = cur.fetchone()
             user_exists = user is not None
             return db_exists or user_exists
 
 
@@ -31,27 +32,33 @@ def create_postgres_db(connection_dict, config):
     if check_db_or_user_exists(connection_dict["db_name"], connection_dict["db_username"], config):
         raise ValueError("db or user already exists")
     with _create_pg_connection(config) as con:
         con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
         with con.cursor() as cur:
-            create_role = "CREATE USER {db_username} WITH PASSWORD '{db_pwd}';".format(**connection_dict)
-            drop_role = "DROP ROLE {db_username};".format(**connection_dict)
-            grant_role = 'GRANT {db_username} TO "{postgraas_user}";'.format(
-                db_username=connection_dict['db_username'], postgraas_user=get_normalized_username(config['username'])
-            )
-            create_database = "CREATE DATABASE {db_name} OWNER {db_username};".format(**connection_dict)
             try:
-                cur.execute(create_role)
-                cur.execute(grant_role)
+                cur.execute(SQL("CREATE USER {} WITH PASSWORD %s;").format(
+                    Identifier(connection_dict['db_username']),
+                ), (
+                    connection_dict['db_pwd'],
+                ))
+                cur.execute(SQL("GRANT {} TO {};").format(
+                    Identifier(connection_dict['db_username']),
+                    Identifier(get_normalized_username(config['username'])),
+                ))
             except psycopg2.ProgrammingError as e:
                 raise ValueError(e.args[0])
             # cleanup role in case database creation fails
-            # saidly 'CREATE DATABASE' cannot run inside a transaction block
+            # sadly 'CREATE DATABASE' cannot run inside a transaction block
             try:
-                cur.execute(create_database)
+                cur.execute(SQL("CREATE DATABASE {} OWNER {};").format(
+                    Identifier(connection_dict['db_name']),
+                    Identifier(connection_dict['db_username']),
+                ))
             except psycopg2.ProgrammingError as e:
-                cur.execute(drop_role)
+                cur.execute(SQL("DROP ROLE {};").format(
+                    Identifier(connection_dict['db_username']),
+                ))
                 raise ValueError(e.args[0])
 
 
 def get_normalized_username(username):
     return username.split('@')[0]
diff --git a/tests/test_integration/backends/postgres_cluster/test_postgres_cluster_driver.py b/tests/test_integration/backends/postgres_cluster/test_postgres_cluster_driver.py
index 0617a41..e9d8cb2 100644
--- a/tests/test_integration/backends/postgres_cluster/test_postgres_cluster_driver.py
+++ b/tests/test_integration/backends/postgres_cluster/test_postgres_cluster_driver.py
@@ -191,10 +191,11 @@ class TestPostgraasApi(PostgraasApiTestBase):
         delete_test_database_and_user(db_credentials['db_name'], db_credentials['db_username'], backend_config)
         delete_test_database_and_user(db_credentials_same_user['db_name'], db_credentials_same_user['db_username'],
                                       backend_config)
         assert ("database or user already exists" in json.loads(response.get_data(as_text=True))['description']) is True
 
+    @pytest.mark.xfail(reason='Username now valid due to hardening against SQL injections.')
     def test_create_postgres_instance_bad_username(self):
         backend_config = CONFIGS[self.backend]['backend']
         db_credentials = {
             "postgraas_instance_name": "tests_postgraas_test_create_postgres_instance_api",
             "db_name": 'test_create_postgres_instance_exists',
diff --git a/tests/test_integration/test_backend_behaviour.py b/tests/test_integration/test_backend_behaviour.py
index 40f3166..7658967 100644
--- a/tests/test_integration/test_backend_behaviour.py
+++ b/tests/test_integration/test_backend_behaviour.py
@@ -165,10 +165,11 @@ class TestPostgraasApi():
         elif self.backend == "docker":
             assert excinfo.value.message == 'Container exists already'
         self.this_app.postgraas_backend.delete(db_entry)
         assert True
 
+    @pytest.mark.xfail(reason='Username now valid due to hardening against SQL injections.')
     def test_create_postgraas_bad_username(self):
         db_credentials = {
             "db_name": 'tests_postgraas_instance_name',
             "db_username": 'tests_postgraas_db-bad username',
             "db_pwd": 'test_db_pwd',

Vulnerable code:             cur.execute("SELECT 1 FROM pg_database WHERE datname='{}';".format(db_name))
            cur.execute("SELECT 1 FROM pg_roles WHERE rolname='{}';".format(db_user))
            create_role = "CREATE USER {db_username} WITH PASSWORD '{db_pwd}';".format(**connection_dict)
            drop_role = "DROP ROLE {db_username};".format(**connection_dict)
            grant_role = 'GRANT {db_username} TO "{postgraas_user}";'.format(
                db_username=connection_dict['db_username'], postgraas_user=get_normalized_username(config['username'])
            )
            create_database = "CREATE DATABASE {db_name} OWNER {db_username};".format(**connection_dict)
                cur.execute(create_role)
                cur.execute(grant_role)
            # saidly 'CREATE DATABASE' cannot run inside a transaction block
                cur.execute(create_database)
                cur.execute(drop_role)
Fixed code: from psycopg2.sql import SQL, Identifier
            cur.execute("SELECT 1 FROM pg_database WHERE datname=%s;", (db_name, ))
            cur.execute("SELECT 1 FROM pg_roles WHERE rolname=%s;", (db_user, ))
                cur.execute(SQL("CREATE USER {} WITH PASSWORD %s;").format(
                    Identifier(connection_dict['db_username']),
                ), (
                    connection_dict['db_pwd'],
                ))
                cur.execute(SQL("GRANT {} TO {};").format(
                    Identifier(connection_dict['db_username']),
                    Identifier(get_normalized_username(config['username'])),
                ))
            # sadly 'CREATE DATABASE' cannot run inside a transaction block
                cur.execute(SQL("CREATE DATABASE {} OWNER {};").format(
                    Identifier(connection_dict['db_name']),
                    Identifier(connection_dict['db_username']),
                ))
                cur.execute(SQL("DROP ROLE {};").format(
                    Identifier(connection_dict['db_username']),
                ))
    @pytest.mark.xfail(reason='Username now valid due to hardening against SQL injections.')
    @pytest.mark.xfail(reason='Username now valid due to hardening against SQL injections.')