Security advisory GHSA-vghm-8cjp-hjw6: postgraas-server vulnerable to SQL injection
Package: postgraas-server (pypi), vulnerable version: 0.1.11, fixed in: 2.0.0
CWE: CWE-89
Files changed by the fix:
postgraas_server/backends/postgres_cluster/postgres_cluster_driver.py

Fix 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,5 +1,6 @@
 import psycopg2
 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
+from psycopg2.sql import SQL, Identifier
 
 
 def _create_pg_connection(config):
@@ -19,9 +20,9 @@ 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
@@ -33,23 +34,29 @@ def create_postgres_db(connection_dict, config):
     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])
 
 
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
@@ -193,6 +193,7 @@ def test_create_postgres_instance_username_exists(self):
                                       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 = {
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
@@ -167,6 +167,7 @@ def test_create_postgraas_twice(self):
         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',