BUG32496788: Prepared statements accepts any type of parameters

For prepared statements any type or argument were possible to be used,
which could produce undesired results. This patch enforces the use of
list or type objects for the argument and If the parameter for the
prepared statement is not of type list or tuple an error will be raised.

Author: isrgomez

CHANGES.txt

@@ -17,12 +17,14 @@ v8.0.24
 - WL#14027: Add support for Python 3.9
 - BUG#32532744: MSI destination folder page lacks details of what installs
 - BUG#32497631: Prepared statements fail when parameters are not given
+- BUG#32496788: Prepared statements accepts any type of parameters
 - BUG#32435181: Add support for Django 3.2
 - BUG#32162928: Change user command fails on pure python implementation
 - BUG#32120659: Prepared statements w/o parameters violates MySQL protocol
 - BUG#32039427: Remove python 3.4 support in MSI packaging
 - BUG#32029891: Add context manager support for pooled connections
 - BUG#31490101: Fix wrong cast of Python unicode to std::string
+- BUG#31315173: Added documentation for multi-host and failover
 - BUG#30416704: Binary columns returned as strings
 
 v8.0.23

lib/mysql/connector/cursor.py

@@ -1211,11 +1211,17 @@ def execute(self, operation, params=None, multi=False):  # multi is unused
 
         if self._prepared['parameters'] and not params:
             return
-        elif params and len(self._prepared['parameters']) != len(params):
-            raise errors.ProgrammingError(
-                errno=1210,
-                msg="Incorrect number of arguments " \
-                    "executing prepared statement")
+        elif params:
+            if not isinstance(params, (tuple, list)):
+                raise errors.ProgrammingError(
+                    errno=1210,
+                    msg="Incorrect type of argument, it must be of type tuple "
+                        "or list the argument given to the prepared statement")
+            if len(self._prepared['parameters']) != len(params):
+                raise errors.ProgrammingError(
+                    errno=1210,
+                    msg="Incorrect number of arguments " \
+                        "executing prepared statement")
 
         if params is None:
             params = ()

lib/mysql/connector/cursor_cext.py

@@ -958,11 +958,17 @@ def execute(self, operation, params=None, multi=False):  # multi is unused
 
         if self._stmt.param_count > 0 and not params:
             return
-        elif params and self._stmt.param_count != len(params):
-            raise errors.ProgrammingError(
-                errno=1210,
-                msg="Incorrect number of arguments executing prepared "
-                    "statement")
+        elif params:
+            if not isinstance(params, (tuple, list)):
+                raise errors.ProgrammingError(
+                    errno=1210,
+                    msg="Incorrect type of argument, it must be of type tuple "
+                        "or list the argument given to the prepared statement")
+            if self._stmt.param_count != len(params):
+                raise errors.ProgrammingError(
+                    errno=1210,
+                    msg="Incorrect number of arguments executing prepared "
+                        "statement")
 
         if params is None:
             params = ()

tests/test_bugs.py

@@ -5945,6 +5945,85 @@ def test_without_use_unicode_cursor_prepared(self):
         )
 
 
+class Bug32496788(tests.MySQLConnectorTests):
+    """BUG#32496788: PREPARED STATEMETS ACCEPTS ANY TYPE OF PARAMETERS."""
+
+    table_name = "Bug32496788"
+    test_values = (
+        ("John", "", "Doe", 21, 77),
+        ["Jane", "", "Doe", 19, 7]
+    )
+    exp_values = (
+        [("John", "", "Doe", 21, 77)],
+        [("Jane", "", "Doe", 19, 7)]
+    )
+
+    def setUp(self):
+        config = tests.get_mysql_config()
+        with mysql.connector.connection.MySQLConnection(**config) as cnx:
+            cnx.cmd_query(f"DROP TABLE IF EXISTS {self.table_name}")
+            cnx.cmd_query(
+                f"""
+                CREATE TABLE {self.table_name} (
+                    column_first_name VARCHAR(30) DEFAULT '' NOT NULL,
+                    column_midle_name VARCHAR(30) DEFAULT '' NOT NULL,
+                    column_last_name VARCHAR(30) DEFAULT '' NOT NULL,
+                    age INT DEFAULT 0 NOT NULL,
+                    lucky_number INT DEFAULT 0 NOT NULL
+                ) CHARACTER SET utf8 COLLATE utf8_general_ci
+                """
+            )
+            cnx.commit()
+
+    def tearDown(self):
+        config = tests.get_mysql_config()
+        with mysql.connector.connection.MySQLConnection(**config) as cnx:
+            cnx.cmd_query(f"DROP TABLE IF EXISTS {self.table_name}")
+
+    @foreach_cnx()
+    def test_parameters_type(self):
+        stmt = f"SELECT * FROM  {self.table_name} WHERE age = ? and lucky_number = ?"
+        with self.cnx.cursor(prepared=True) as cur:
+            # Test incorrect types must raise error
+            for param in ["12", 12, 1.3, lambda : "1"+ "2"]:
+                self.assertRaises(errors.ProgrammingError, cur.execute, stmt, (param,))
+                with self.assertRaises(errors.ProgrammingError) as contex:
+                    cur.execute(stmt, param)
+                self.assertIn("Incorrect type of argument, it must be of type tuple or list",
+                              contex.exception.msg)
+
+        with self.cnx.cursor(prepared=True) as cur:
+            # Correct form with tuple
+            cur.execute("SELECT ?, ?", ("1", "2"))
+            self.assertEqual(cur.fetchall(),  [('1', '2')])
+
+            # Correct form with list
+            cur.execute("SELECT ?, ?", ["1", "2"])
+            self.assertEqual(cur.fetchall(),  [('1', '2')])
+
+        with self.cnx.cursor(prepared=True) as cur:
+            insert_stmt = f"INSERT INTO {self.table_name} VALUES (?, ?, ?, ?, ?)"
+            cur.execute(insert_stmt)
+            with self.assertRaises(errors.ProgrammingError) as contex:
+                cur.execute(insert_stmt, "JD")
+                self.assertIn("Incorrect type of argument, it must be of type tuple or list",
+                              contex.exception.msg)
+            self.assertRaises(errors.ProgrammingError, cur.execute, insert_stmt, ("JohnDo"))
+            cur.execute(insert_stmt, self.test_values[0])
+            cur.execute(insert_stmt, self.test_values[1])
+
+            select_stmt = f"SELECT * FROM {self.table_name} WHERE column_last_name=? and column_first_name=?"
+            self.assertRaises(errors.ProgrammingError, cur.execute, select_stmt, ("JD"))
+            with self.assertRaises(errors.ProgrammingError) as contex:
+                cur.execute(select_stmt, "JD")
+                self.assertIn("Incorrect type of argument, it must be of type tuple or list",
+                              contex.exception.msg)
+            cur.execute(select_stmt, ("Doe", "John"))
+            self.assertEqual(cur.fetchall(),  self.exp_values[0])
+            cur.execute(select_stmt, ("Doe", "Jane"))
+            self.assertEqual(cur.fetchall(),  self.exp_values[1])
+
+
 class Bug32162928(tests.MySQLConnectorTests):
     """BUG#32162928: change user command fails with pure python implementation.