BUG#34984850: Fix binary conversion with NO_BACKSLASH_ESCAPES mode

nmariz · nmariz · commit a46e60ef65ef · 2023-02-23T15:22:36.000Z
The default converter used by the pure Python implementation does not
escape the single quotes correctly when using NO_BACKSLASH_ESCAPES
SQL mode.

This patch is an extension of the initial fix for BUG#34984850.

Change-Id: I46be753bb34f21dc6eff1a559906f6e50257ce6b
diff --git a/lib/mysql/connector/connection_cext.py b/lib/mysql/connector/connection_cext.py
@@ -770,7 +770,9 @@ def prepare_for_mysql(
             if self.converter:
                 result = [
                     self.converter.quote(
-                        self.converter.escape(self.converter.to_mysql(value))
+                        self.converter.escape(
+                            self.converter.to_mysql(value), self._sql_mode
+                        )
                     )
                     for value in params
                 ]
@@ -781,7 +783,9 @@ def prepare_for_mysql(
             if self.converter:
                 for key, value in params.items():
                     result[key] = self.converter.quote(
-                        self.converter.escape(self.converter.to_mysql(value))
+                        self.converter.escape(
+                            self.converter.to_mysql(value), self._sql_mode
+                        )
                     )
             else:
                 for key, value in params.items():
diff --git a/lib/mysql/connector/conversion.py b/lib/mysql/connector/conversion.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2009, 2022, Oracle and/or its affiliates.
+# Copyright (c) 2009, 2023, Oracle and/or its affiliates.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License, version 2.0, as
@@ -132,7 +132,10 @@ def to_python(
             return value
 
     @staticmethod
-    def escape(value: Any) -> Any:
+    def escape(
+        value: Any,
+        sql_mode: Optional[str] = None,  # pylint: disable=unused-argument
+    ) -> Any:
         """Escape buffer for sending to MySQL"""
         return value
 
@@ -169,7 +172,7 @@ def __init__(
         ] = {}
 
     @staticmethod
-    def escape(value: Any) -> Any:
+    def escape(value: Any, sql_mode: Optional[str] = None) -> Any:
         """
         Escapes special characters as they are expected to by when MySQL
         receives them.
@@ -178,13 +181,17 @@ def escape(value: Any) -> Any:
         Returns the value if not a string, or the escaped string.
         """
         if isinstance(value, (bytes, bytearray)):
+            if sql_mode == "NO_BACKSLASH_ESCAPES":
+                return value.replace(b"'", b"''")
             value = value.replace(b"\\", b"\\\\")
             value = value.replace(b"\n", b"\\n")
             value = value.replace(b"\r", b"\\r")
             value = value.replace(b"\047", b"\134\047")  # single quotes
             value = value.replace(b"\042", b"\134\042")  # double quotes
             value = value.replace(b"\032", b"\134\032")  # for Win32
         elif isinstance(value, str) and not isinstance(value, HexLiteral):
+            if sql_mode == "NO_BACKSLASH_ESCAPES":
+                return value.replace("'", "''")
             value = value.replace("\\", "\\\\")
             value = value.replace("\n", "\\n")
             value = value.replace("\r", "\\r")
@@ -270,8 +277,8 @@ def _int_to_mysql(value: int) -> int:
     def _long_to_mysql(value: int) -> int:
         """Convert value to int
 
-        Note: there is not type "long" in Python 3 since integers (int) are of unlimited size.
-        Since Python 2 is no longer supported, this method should be deprecated.
+        Note: There is no type "long" in Python 3 since integers are of unlimited size.
+              Since Python 2 is no longer supported, this method should be deprecated.
         """
         return int(value)
 
diff --git a/lib/mysql/connector/cursor.py b/lib/mysql/connector/cursor.py
@@ -442,7 +442,7 @@ def _process_params_dict(
             for key, value in params.items():
                 conv = value
                 conv = to_mysql(conv)
-                conv = escape(conv)
+                conv = escape(conv, self._connection.sql_mode)
                 if not isinstance(value, Decimal):
                     conv = quote(conv)
                 res[key.encode()] = conv
@@ -462,7 +462,7 @@ def _process_params(
             escape = self._connection.converter.escape
             quote = self._connection.converter.quote
             res = [to_mysql(value) for value in res]
-            res = [escape(value) for value in res]
+            res = [escape(value, self._connection.sql_mode) for value in res]
             res = [
                 quote(value) if not isinstance(params[i], Decimal) else value
                 for i, value in enumerate(res)
diff --git a/tests/test_bugs.py b/tests/test_bugs.py
@@ -7390,7 +7390,8 @@ def setUp(self):
                     f"""
                     CREATE TABLE {self.table_name} (
                         id INT AUTO_INCREMENT PRIMARY KEY,
-                        data LONGBLOB
+                        data LONGBLOB,
+                        text VARCHAR(50)
                     )
                     """
                 )
@@ -7401,16 +7402,24 @@ def tearDown(self):
         with mysql.connector.connect(**config) as cnx:
             cnx.cmd_query(f"DROP TABLE IF EXISTS {self.table_name}")
 
-    @foreach_cnx()
-    def test_no_backslash_escapes(self):
-        data = b"\x01\x02\x00\x01"
+    def _run_test(self):
+        data = (b"\x01\x02\x00\x01", "abc'cba")
         for use_backslash_escapes in (True, False):
             if use_backslash_escapes:
                 self.cnx.sql_mode = [constants.SQLMode.NO_BACKSLASH_ESCAPES]
             with self.cnx.cursor() as cur:
                 cur.execute(
-                    f"INSERT INTO {self.table_name} (data) VALUES (%s)", (data,)
+                    f"INSERT INTO {self.table_name} (data, text) VALUES (%s, %s)", data
                 )
-                cur.execute(f"SELECT data FROM {self.table_name} LIMIT 1")
+                cur.execute(f"SELECT data, text FROM {self.table_name} LIMIT 1")
                 res = cur.fetchone()
-                self.assertEqual(data, res[0])
+                self.assertEqual(data, res)
+
+    @foreach_cnx()
+    def test_no_backslash_escapes(self):
+        self._run_test()
+
+    @cnx_config(converter_class=conversion.MySQLConverter)
+    @foreach_cnx()
+    def test_no_backslash_escapes_with_converter_class(self):
+        self._run_test()
