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.

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])
 
 

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 = {

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',