dbconn: Add tests for search_path CVE

We have added tests to ensure that the search_path setting is taking effect. We
also are using a test database for the tests.
Co-authored-by: NJamie McAtamney <jmcatamney@pivotal.io>
Co-authored-by: NJacob Champion <pchampion@pivotal.io>
Co-authored-by: NNikolaos Kalampalikis <nkalampalikis@pivotal.io>

gpMgmt/bin/gppylib/db/test/unit/test_cluster_dbconn.py

@@ -8,9 +8,17 @@ from gppylib.db import dbconn
 class ConnectTestCase(unittest.TestCase):
     """A test case for dbconn.connect()."""
 
-    def setUp(self):
+    @classmethod
+    def setUpClass(cls):
         # Connect to the database pointed to by PGHOST et al.
-        self.url = dbconn.DbURL()
+        with dbconn.connect(dbconn.DbURL()) as conn:
+            # using the pg.DB connection so that each SQL is done as a single
+            # transaction
+            db = pg.DB(conn)
+            test_database_name = "gpdb_test_database"
+            db.query("DROP DATABASE IF EXISTS %s" % test_database_name)
+            db.query("CREATE DATABASE %s" % test_database_name)
+            cls.url = dbconn.DbURL(dbname=test_database_name)
 
     def _raise_warning(self, db, msg):
         """Raises a WARNING message on the given pg.DB."""
@@ -72,6 +80,18 @@ class ConnectTestCase(unittest.TestCase):
 
         self.assertEqual(result, '"$user",public')
 
+    def test_search_path_cve_2018_1058(self):
+        with dbconn.connect(self.url) as conn:
+            dbconn.execSQL(conn, "CREATE TABLE public.Names (name VARCHAR(255))")
+            dbconn.execSQL(conn, "INSERT INTO public.Names VALUES ('AAA')")
+
+            dbconn.execSQL(conn, "CREATE FUNCTION public.lower(VARCHAR) RETURNS text AS $$ "
+                                 "  SELECT 'Alice was here: ' || $1;"
+                                 "$$ LANGUAGE SQL IMMUTABLE;")
+
+            name = dbconn.execSQLForSingleton(conn, "SELECT lower(name) FROM public.Names")
+            self.assertEqual(name, 'aaa')
+
     def test_no_transaction_after_connect(self):
         with dbconn.connect(self.url) as conn:
             db = pg.DB(conn)